当我尝试使用 MagicMock 在单元测试中模拟异步函数时,出现以下异常:
TypeError:对象 MagicMock 不能在“await”表达式中使用
示例代码如下:
# source code
class Service:
async def compute(self, x):
return x
class App:
def __init__(self):
self.service = Service()
async def handle(self, x):
return await self.service.compute(x)
# test code
import asyncio
import unittest
from unittest.mock import patch
class TestApp(unittest.TestCase):
@patch('__main__.Service')
def test_handle(self, mock):
loop = asyncio.get_event_loop()
app = App()
res = loop.run_until_complete(app.handle('foo'))
app.service.compute.assert_called_with("foo")
if __name__ == '__main__':
unittest.main()
我应该如何使用内置的 python3 库修复它?
在 python 3.8+ 中,您可以使用
AsyncMock
async def test_that_mock_can_be_awaited():
mock = AsyncMock()
mock.x.return_value = 123
result = await mock.x()
assert result == 123
类
AsyncMock
对象的行为使得该对象被识别为异步函数,并且调用的结果是可等待的。
>>> mock = AsyncMock()
>>> asyncio.iscoroutinefunction(mock)
True
>>> inspect.isawaitable(mock())
True
我最终得到了这个黑客。
# monkey patch MagicMock
async def async_magic():
pass
MagicMock.__await__ = lambda x: async_magic().__await__()
仅适用于MagicMock,不适用于其他预定义的return_value
shaun shia 提供了非常好的通用解决方案,但我发现在 python 3.8 中你可以只使用
@patch('__main__.Service', new=AsyncMock)
您可以使用 Future 获取模拟来返回可以等待的对象。以下是一个 pytest 测试用例,但使用 unittest 应该可以实现类似的功能。
async def test_that_mock_can_be_awaited():
mock = MagicMock(return_value=Future())
mock.return_value.set_result(123)
result = await mock()
assert result == 123
在你的情况下,由于你正在修补
Service
(它被传递为mock
),mock.return_value = Future()
应该可以解决问题。
我发现这个注释在尝试
await
Python中的模拟对象< 3.8. You simply create a child class AsyncMock
继承自MagicMock
并覆盖__call__
方法成为协程时非常有用:
class AsyncMock(MagicMock):
async def __call__(self, *args, **kwargs):
return super(AsyncMock, self).__call__(*args, **kwargs)
然后,在测试中执行:
@pytest.mark.asyncio
async def test_my_method():
# Test "my_method" coroutine by injecting an async mock
my_mock = AsyncMock()
assert await my_method(my_mock)
pytest-asyncio
如果您想要与
兼容的
pytest-mock
< py3.8, I did something like this. 解决方案
class AsyncMock(MagicMock):
async def __call__(self, *args, **kwargs):
return super().__call__(*args, **kwargs)
def test_my_method(mocker):
my_mock = mocker.patch("path.to.mocked.thing", AsyncMock())
my_mock.return_value = [1, 2, 3]
assert my_method()
绝对借鉴了 Tomasz 的解决方案!
要重写异步类,需要告诉
patch
,return_value
需要是 AsyncMock
。所以使用
@patch('__main__.Service', return_value=AsyncMock(Service))
def test_handle(self, mock):
loop = asyncio.get_event_loop()
app = App()
res = loop.run_until_complete(app.handle('foo'))
app.service.compute.assert_called_with("foo")
这样
Service
将是 MagicMock
,但 Service()
将返回 AsyncMock
的 Service
实例。
当我在自己的代码/测试代码中遇到如下错误时,我遇到了这个SO讨论:
TypeError: object MagicMock can't be used in 'await' expression
TypeError: object NonCallableMagicMock can't be used in 'await' expression
TypeError: object AsyncMock can't be used in 'await' expression
我的测试代码是异步 Django ORM 查询,我想模拟某些数据库查找的最终返回值,例如:
try:
user = await AllauthUser.objects.aget(email=email)
except AllauthUser.DoesNotExist:
# handle user not found
最终,我需要简单地将以下 kwarg 添加到我的模拟对象初始化中:
new_callable=AsyncMock
。可以在此处查看文档。
我分享这个答案是因为我花了比我愿意承认的更多的时间来解决这些类型错误!