在 python 3.5 中模拟异步调用

Question

如何使用 unittest.mock.patch 模拟从一个本机协程到另一个本机协程的异步调用？

我目前有一个很尴尬的解决方案：

class CoroutineMock(MagicMock):
    def __await__(self, *args, **kwargs):
        future = Future()
        future.set_result(self)
        result = yield from future
        return result

然后

class TestCoroutines(TestCase):
    @patch('some.path', new_callable=CoroutineMock)
    def test(self, mock):
        some_action()
        mock.assert_called_with(1,2,3)

这有效但看起来很丑。有没有更多的 pythonic 方法来做到这一点？

Answer 1

另一种模拟协程的方法是制作协程，returns 模拟。通过这种方式，您可以模拟将传递给 asyncio.wait 或 asyncio.wait_for.

的协程

这使得更通用的协程虽然使测试设置更麻烦：

def make_coroutine(mock)
    async def coroutine(*args, **kwargs):
        return mock(*args, **kwargs)
    return coroutine


class Test(TestCase):
    def setUp(self):
        self.coroutine_mock = Mock()
        self.patcher = patch('some.coroutine',
                             new=make_coroutine(self.coroutine_mock))
        self.patcher.start()

    def tearDown(self):
        self.patcher.stop()

Answer 2

解决方法其实很简单：我只需要将 mock 的 __call__ 方法转换为协程：

class AsyncMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)

这很完美，当调用 mock 时，代码接收本机协程

用法示例：

@mock.patch('my.path.asyncio.sleep', new_callable=AsyncMock)
def test_stuff(sleep):
    # code

Answer 3

Subclassing MagicMock 将为您的协程模拟生成的所有模拟传播您的自定义 class。例如，AsyncMock().__str__ 也会变成 AsyncMock，这可能不是您要查找的内容。

相反，您可能想要定义一个工厂来创建带有自定义参数的 Mock（或 MagicMock），例如 side_effect=coroutine(coro)。此外，将协程函数与协程分开可能是个好主意（如 documentation 中所述）。

这是我想出的：

from asyncio import coroutine

def CoroMock():
    coro = Mock(name="CoroutineResult")
    corofunc = Mock(name="CoroutineFunction", side_effect=coroutine(coro))
    corofunc.coro = coro
    return corofunc

不同对象的解释：

corofunc: 协程函数 mock
corofunc.side_effect()：协程，为每次调用生成
corofunc.coro：协程获取结果使用的mock
corofunc.coro.return_value：协程返回的值
corofunc.coro.side_effect：可能用于引发异常

示例：

async def coro(a, b):
    return await sleep(1, result=a+b)

def some_action(a, b):
    return get_event_loop().run_until_complete(coro(a, b))

@patch('__main__.coro', new_callable=CoroMock)
def test(corofunc):
    a, b, c = 1, 2, 3
    corofunc.coro.return_value = c
    result = some_action(a, b)
    corofunc.assert_called_with(a, b)
    assert result == c

Answer 4

每个人都错过了可能是最简单明了的解决方案：

@patch('some.path')
def test(self, mock):
    f = asyncio.Future()
    f.set_result('whatever result you want')
    process_smtp_message.return_value = f
    mock.assert_called_with(1, 2, 3)

记住协程可以被认为只是一个函数，它保证 return 一个可以等待的未来。

Answer 5

基于@scolvin 的回答，我创建了这个 (imo) 更简洁的方法：

import asyncio

def async_return(result):
    f = asyncio.Future()
    f.set_result(result)
    return f

就是这样，只要围绕 return 你想要异步的任何东西使用它，如

mock = MagicMock(return_value=async_return("Example return"))
await mock()

Answer 6

模拟异步对象的 "simplest" 解决方案的另一种变体，它只是一个衬里。

来源：

class Yo:
    async def foo(self):
        await self.bar()
    async def bar(self):
        # Some code

测试中：

from asyncio import coroutine

yo = Yo()
# Here bounded method bar is mocked and will return a customised result.
yo.bar = Mock(side_effect=coroutine(lambda:'the awaitable should return this'))
event_loop.run_until_complete(yo.foo())

Answer 7

您可以像这样设置异步方法的return_value：

mock = unittest.mock.MagicMock()
mock.your_async_method.return_value = task_from_result(your_return_value)

async def task_from_result(result):
    return result

调用者必须await your_async_method(..)就像该方法没有被模拟一样。

Answer 8

我不知道为什么没有人提到可用的默认选项。 python 提供 MagicMock 的异步版本。

您可以在此处阅读更多相关信息。 https://docs.python.org/3/library/unittest.mock.html#unittest.mock.AsyncMock

如果您使用的是补丁，那么您也不需要进行任何其他更改。如果需要，它将自动用异步模拟函数替换它。在这里阅读更多 https://docs.python.org/3/library/unittest.mock.html#patch

Answer 9

我喜欢这种方法，它也使 AsyncMock 的行为与 Mock 完全一样：

class AsyncMock:
    def __init__(self, *args, **kwargs):
        self.mock = Mock(*args, **kwargs)

    async def __call__(self, *args, **kwargs):
        return self.mock(*args, **kwargs)

    def __getattr__(self, item):
        return getattr(self.mock, item)

然后你可以像使用 Mock 一样使用它，即：

@pytest.mark.asyncio
async def test_async_mock_example(monkeypatch):
    fn = AsyncMock(side_effect=ValueError)
    with pytest.raises(ValueError):
        await fn()
    assert fn.call_count == 1

在 python 3.5 中模拟异步调用

Mocking async call in python 3.5

python

python-asyncio

python-mock