如何模拟异步协程？

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

我正在为 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 属性，并且带有spec 的asynctest.Mock 可以返回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 3

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

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 4

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

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 5

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

Answer 6

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

Answer 7

在绝大多数情况下，达斯汀的答案可能是正确的。我有一个不同的问题，协程需要返回多个值，例如模拟read() 操作，如我的comment 中所述。

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

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.

因此，扩展达斯汀的答案，它看起来像：

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)

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

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

Answer 8

好吧，这里已经有很多答案了，但我会贡献我的扩展版e-satis's answer。这个类模拟一个异步函数并跟踪调用计数和调用参数，就像 Mock 类对同步函数所做的一样。

在 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 9

您可以将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')