如何模拟异步协程？

Question

以下代码因 ImBeingTested.i_call_other_coroutines 中的 TypeError: 'Mock' object is not iterable 而失败，因为我已将 ImGoingToBeMocked 替换为 Mock 对象。

如何模拟协程？

class ImGoingToBeMocked:
    @asyncio.coroutine
    def yeah_im_not_going_to_run(self):
        yield from asyncio.sleep(1)
        return "sup"

class ImBeingTested:
    def __init__(self, hidude):
        self.hidude = hidude

    @asyncio.coroutine
    def i_call_other_coroutines(self):
        return (yield from self.hidude.yeah_im_not_going_to_run())

class TestImBeingTested(unittest.TestCase):

    def test_i_call_other_coroutines(self):
        mocked = Mock(ImGoingToBeMocked)
        ibt = ImBeingTested(mocked)

        ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines())

Answer 1

由于 mock 库不支持协程，我手动创建模拟协程并将它们分配给模拟对象。有点冗长，但它有效。

您的示例可能如下所示：

import asyncio
import unittest
from unittest.mock import Mock


class ImGoingToBeMocked:
    @asyncio.coroutine
    def yeah_im_not_going_to_run(self):
        yield from asyncio.sleep(1)
        return "sup"


class ImBeingTested:
    def __init__(self, hidude):
        self.hidude = hidude

    @asyncio.coroutine
    def i_call_other_coroutines(self):
        return (yield from self.hidude.yeah_im_not_going_to_run())


class TestImBeingTested(unittest.TestCase):

    def test_i_call_other_coroutines(self):
        mocked = Mock(ImGoingToBeMocked)
        ibt = ImBeingTested(mocked)

        @asyncio.coroutine
        def mock_coro():
            return "sup"
        mocked.yeah_im_not_going_to_run = mock_coro

        ret = asyncio.get_event_loop().run_until_complete(
            ibt.i_call_other_coroutines())
        self.assertEqual("sup", ret)


if __name__ == '__main__':
    unittest.main()

Answer 2

源自 Andrew Svetlov 的，我只是想分享这个辅助函数：

def get_mock_coro(return_value):
    @asyncio.coroutine
    def mock_coro(*args, **kwargs):
        return return_value

    return Mock(wraps=mock_coro)

这让您可以使用标准 assert_called_with、call_count 和常规 unittest.Mock 为您提供的其他方法和属性。

您可以将其与问题中的代码一起使用，例如：

class ImGoingToBeMocked:
    @asyncio.coroutine
    def yeah_im_not_going_to_run(self):
        yield from asyncio.sleep(1)
        return "sup"

class ImBeingTested:
    def __init__(self, hidude):
        self.hidude = hidude

    @asyncio.coroutine
    def i_call_other_coroutines(self):
        return (yield from self.hidude.yeah_im_not_going_to_run())

class TestImBeingTested(unittest.TestCase):

    def test_i_call_other_coroutines(self):
        mocked = Mock(ImGoingToBeMocked)
        mocked.yeah_im_not_going_to_run = get_mock_coro()
        ibt = ImBeingTested(mocked)

        ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines())
        self.assertEqual(mocked.yeah_im_not_going_to_run.call_count, 1)

Answer 3

在绝大多数情况下，Dustin 的回答可能是正确的。我有一个不同的问题，协程需要 return 多个值，例如模拟 read() 操作，如我的 .

中简要描述的

经过更多测试，下面的代码对我有用，通过在模拟函数外定义迭代器，有效地记住最后一个值 returned 发送下一个：

def test_some_read_operation(self):
    #...
    data = iter([b'data', b''])
    @asyncio.coroutine
    def read(*args):
        return next(data)
    mocked.read = Mock(wraps=read)
    # Here, the business class would use its .read() method which
    # would first read 4 bytes of data, and then no data
    # on its second read.

因此，扩展 Dustin 的答案，它看起来像：

def get_mock_coro(return_values):
    values = iter(return_values)
    @asyncio.coroutine
    def mock_coro(*args, **kwargs):
        return next(values)

    return Mock(wraps=mock_coro)

我在这种方法中看到的两个直接缺点是：

不允许轻易引发异常（例如，首先 return 一些数据，然后在第二次读取操作时引发错误）。
我还没有找到使用标准 Mock .side_effect 或 .return_value 属性使其更明显和可读的方法。

Answer 4

我正在为 unittest 写一个包装器，目的是在为 asyncio 编写测试时减少样板文件。

代码在这里：https://github.com/Martiusweb/asynctest

你可以模拟协程 asynctest.CoroutineMock:

>>> mock = CoroutineMock(return_value='a result')
>>> asyncio.iscoroutinefunction(mock)
True
>>> asyncio.iscoroutine(mock())
True
>>> asyncio.run_until_complete(mock())
'a result'

它也适用于 side_effect 属性，asynctest.Mock 和 spec 可以 return CoroutineMock:

>>> asyncio.iscoroutinefunction(Foo().coroutine)
True
>>> asyncio.iscoroutinefunction(Foo().function)
False
>>> asynctest.Mock(spec=Foo()).coroutine
<class 'asynctest.mock.CoroutineMock'>
>>> asynctest.Mock(spec=Foo()).function
<class 'asynctest.mock.Mock'>

unittest.Mock 的所有功能都有望正常工作（patch() 等）。

Answer 5

您可以自己创建异步模拟：

import asyncio
from unittest.mock import Mock


class AsyncMock(Mock):

    def __call__(self, *args, **kwargs):
        sup = super(AsyncMock, self)
        async def coro():
            return sup.__call__(*args, **kwargs)
        return coro()

    def __await__(self):
        return self().__await__()

Answer 6

好吧，这里已经有很多答案了，但我会贡献我的的扩展版本。 class 模拟异步函数并跟踪调用计数和调用参数，就像 Mock class 对同步函数所做的那样。

在 Python 3.7.0.

上测试

class AsyncMock:
    ''' A mock that acts like an async def function. '''
    def __init__(self, return_value=None, return_values=None):
        if return_values is not None:
            self._return_value = return_values
            self._index = 0
        else:
            self._return_value = return_value
            self._index = None
        self._call_count = 0
        self._call_args = None
        self._call_kwargs = None

    @property
    def call_args(self):
        return self._call_args

    @property
    def call_kwargs(self):
        return self._call_kwargs

    @property
    def called(self):
        return self._call_count > 0

    @property
    def call_count(self):
        return self._call_count

    async def __call__(self, *args, **kwargs):
        self._call_args = args
        self._call_kwargs = kwargs
        self._call_count += 1
        if self._index is not None:
            return_index = self._index
            self._index += 1
            return self._return_value[return_index]
        else:
            return self._return_value

用法示例：

async def test_async_mock():
    foo = AsyncMock(return_values=(1,2,3))
    assert await foo() == 1
    assert await foo() == 2
    assert await foo() == 3

Answer 7

您可以使用 asynctest 并导入 CoroutineMock 或使用 asynctest.mock.patch

Answer 8

您可以将 Mock 子类化为协程函数：

class CoroMock(Mock):
    async def __call__(self, *args, **kwargs):
        return super(CoroMock, self).__call__(*args, **kwargs)

    def _get_child_mock(self, **kw):
        return Mock(**kw)

您可以像使用普通模拟一样使用 CoroMock，但需要注意的是，在事件循环执行协程之前不会记录调用。

如果你有一个模拟对象并且你想使一个特定的方法成为协程，你可以像这样使用Mock.attach_mock：

mock.attach_mock(CoroMock(), 'method_name')

Answer 9

python 3.6+ 的稍微简化的示例改编自此处的一些答案：

import unittest

class MyUnittest()

  # your standard unittest function
  def test_myunittest(self):

    # define a local mock async function that does what you want, such as throw an exception. The signature should match the function you're mocking.
    async def mock_myasync_function():
      raise Exception('I am testing an exception within a coroutine here, do what you want')

    # patch the original function `myasync_function` with the one you just defined above, note the usage of `wrap`, which hasn't been used in other answers.
    with unittest.mock.patch('mymodule.MyClass.myasync_function', wraps=mock_myasync_function) as mock:
      with self.assertRaises(Exception):
        # call some complicated code that ultimately schedules your asyncio corotine mymodule.MyClass.myasync_function
        do_something_to_call_myasync_function()

如何模拟异步协程？

How to mock asyncio coroutines?

python

unit-testing

mocking

python-3.x

python-asyncio