在Python3.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)

这很管用，但看起来很难看。还有什么比这更像Python的方法吗？

Answer 1

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

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

解决方法其实很简单： 我只需要将mock的__call__方法转换为coroutine：

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 3

子类化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：协程函数模拟
• 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