如何模拟异步协程？

Question

以下代码在 TypeError: 'Mock' object is not iterable 中失败并显示 ImBeingTested.i_call_other_coroutines 因为我已经将 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())

原文由 Dustin Wyatt 发布，翻译遵循 CC BY-SA 4.0 许可协议

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

原文由 Andrew Svetlov 发布，翻译遵循 CC BY-SA 3.0 许可协议

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 属性，以及 asynctest.Mock 和 spec 可以返回 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() 等）。

原文由 Martin Richard 发布，翻译遵循 CC BY-SA 3.0 许可协议