在 python 3.5 中模拟异步调用

Posted 2023-03-30

【中文标题】在 python 3.5 中模拟异步调用

【英文标题】：Mocking async call in python 3.5

【发布时间】：2015-09-09 12:54:11

【问题描述】：

如何使用 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方式来做到这一点？

【问题讨论】：

此外，由于 asyncio.tasks.ensure_future，此模拟不适用于 asyncio.await

【参考方案1】：

解决方案其实很简单： 我只需要将 mock 的 __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

【讨论】：

这很好，但它不能很好地与 autospec 配合使用，这在使用 MagicMock 时基本上是强制性的。关于如何让它工作的任何想法？我对内部结构不够熟悉...

它非常适合我。我这样使用它：``` @mock.patch( 'my.path.asyncio.sleep', new_callable=AsyncMock, ) def test_stuff(sleep): # code ```

这行得通。我最初喜欢 Ivan Castellanos 下面的另一个解决方案。但未来永远不会被执行，我拼命尝试，但没有成功。

这非常方便和优雅。至少用于基本用途。

注意：AsyncMock 从 python 3.8 开始在 unittest.mock 中可用，mock 还会自动检测它应该在哪里使用（参见 docs for examples）。

【参考方案2】：

每个人都错过了可能是最简单和最清晰的解决方案：

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

请记住，协程可以被认为只是一个函数，它保证返回一个可以反过来等待的未来。

【讨论】：

process_smtp_message.return_value = f 是什么？另外，被测函数的调用在哪里？

@Skorpeo - 我认为他的意思是 mock.return_value = f

我绝对不会。 Mock 是测试夹具。 process_smtp_message 显然是您要模拟的任何内容。

该解决方案与 Python 3.8 不兼容，其中引入了原生 AsyncMock。因此，由于 Future 类问题，带有解决方案的代码将因错误而失败。但是具有简单AsyncMock 实现的Zozz 解决方案可以在Python 3.7（例如）和Python 3.8（如果您将有条件地导入本机AsyncMock）中工作

对于python3.8 及以上，我最终使用：patch.object(your_object, 'async_method_name', return_value=your_return_value)

【参考方案3】：

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

import asyncio

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

就是这样，只需将它用于您想要异步的任何返回，如

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

【参考方案4】：

子类化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：协程函数mock corofunc.side_effect()：协程，每次调用都会生成 corofunc.coro：协程用来获取结果的mock 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

【讨论】：

这个不行，side_effect=coroutine(coro)，协程没有定义

我实际上更喜欢原来的解决方案，因为它不需要对测试函数进行特殊的重写。与问题中显示的方法相比，这种方法有什么优势？

【参考方案5】：

模拟协程的另一种方式是制作协程，它返回模拟。这样你就可以模拟将被传递到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()

【参考方案6】：

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

来源：

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

【参考方案7】：

您可以像这样设置异步方法的return_value：

mock = unittest.mock.MagicMock()
mock.your_async_method.return_value = task_from_result(your_return_value)

async def task_from_result(result):
    return result

调用者必须做await your_async_method(..)，就像方法没有被模拟一样。

【参考方案8】：

我不明白为什么没有人提到可用的默认选项。 python 提供了一个 Async 版本的 MagicMock。

您可以在此处阅读有关此内容的更多信息。 https://docs.python.org/3/library/unittest.mock.html#unittest.mock.AsyncMock

如果您使用的是补丁，那么您也不需要进行任何其他更改。如果需要，它将自动用异步模拟函数替换它。 在这里阅读更多 https://docs.python.org/3/library/unittest.mock.html#patch