在 python 3.5 中模拟异步调用

Mocking async call in python 3.5

提问by Zozz

How do I mock async call from one native coroutine to other one using unittest.mock.patch?

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

I currently have quite an awkward solution:

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

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

Then

然后

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

This works but looks ugly. Is there more pythonic way to do this?

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

采纳答案by Vincent

Subclassing MagicMockwill propagate your custom class for all the mocks generated from your coroutine mock. For instance, AsyncMock().__str__will also become an AsyncMockwhich is probably not what you're looking for.

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

Instead, you might want to define a factory that creates a Mock(or a MagicMock) with custom arguments, for instance side_effect=coroutine(coro). Also, it might be a good idea to separate the coroutine function from the coroutine (as explained in the documentation).

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

Here is what I came up with:

这是我想出的：

from asyncio import coroutine

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

An explanation of the different objects:

不同对象的解释：

• corofunc: the coroutine function mock
• corofunc.side_effect(): the coroutine, generated for each call
• corofunc.coro: the mock used by the coroutine to get the result
• corofunc.coro.return_value: the value returned by the coroutine
• corofunc.coro.side_effect: might be used to raise an exception

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

Example:

例子：

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

回答by Zozz

Another way of mocking coroutine is to make coroutine, that returns mock. This way you can mock coroutines that will be passed into asyncio.waitor asyncio.wait_for.

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

This makes more universal coroutines though makes setup of tests more cumbersome:

这使得更通用的协程虽然使测试的设置更加繁琐：

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()

回答by Zozz

The solution was actually quite simple: I just needed to convert __call__method of mock into coroutine:

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

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

This works perfectly, when mock is called, code receives native coroutine

这完美地工作，当调用模拟时，代码接收本机协程

Example usage:

用法示例：

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

回答by SColvin

Everyone's missing what's probably the simplest and clearest solution:

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

@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)

remember a coroutine can be thought of as just a function which is guaranteed to return a future which can, in turn be awaited.

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

回答by Ivan Castellanos

Based on @scolvin answer I created this (imo) cleaner way:

基于@scolvin 的回答，我创建了这个（imo）更简洁的方式：

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

That's it, just use it around whatever return you want to be async, as in

就是这样，只需在您想要异步的任何返回值周围使用它，如

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

回答by Murphy Meng

One more variant of "simplest" solution to mock a async object, which is just a one liner.

模拟异步对象的“最简单”解决方案的另一种变体，它只是一个衬垫。

In source:

在来源：

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

In test:

测试中：

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())