在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

子类化MagicMock将传播您的自定义类，用于从您的协同模拟生成的所有模拟。例如，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：coroutine函数mock
• corofunc.side_effect()：为每次通话生成的协程
• corofunc.coro：协程用来获得结果的模拟
• 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 2

每个人都错过了最简单，最清晰的解决方案：

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

记住一个协程可以被认为是一个保证返回未来的功能，而这个功能可以反过来等待。

Answer 3

解决方案实际上非常简单：我只需要将模拟的__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 4

另一种模仿协程的方法是制作coroutine，返回mock。这样你就可以模拟将传递给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 5

根据@scolvin的回答，我创建了这个（imo）更清洁的方式：

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

就是这样，只需在你想要异步的任何回报中使用它，就像在

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

Answer 6

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

来源：

class Yo:
    async foo(self):
        await self.bar()
    async 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())