Python asyncio：如何模拟 __aiter__() 方法？
Python asyncio: how to mock __aiter__() method?
我有一个代码正在使用 aiohttp.
监听 WebSocket 上的消息
看起来像：
async for msg in ws:
await self._ws_msg_handler.handle_message(ws, msg, _services)
其中 ws 是 aiohttp.web.WebSocketResponse() (original code)
的实例
在我的测试中，我模拟了 WebSocketResponse() 及其 __aiter__ 方法：
def coro_mock(**kwargs):
return asyncio.coroutine(mock.Mock(**kwargs))
@pytest.mark.asyncio
@mock.patch('aiojsonrpc.request_handler.WebSocketMessageHandler')
async def test_rpc_websocket_handler(
MockWebSocketMessageHandler,
rpc_websocket_handler
):
ws_response = 'aiojsonrpc.request_handler.WebSocketResponse'
with mock.patch(ws_response) as MockWebSocketResponse:
MockRequest = mock.MagicMock()
req = MockRequest()
ws_instance = MockWebSocketResponse.return_value
ws_instance.prepare = coro_mock()
ws_instance.__aiter__ = coro_mock(return_value=iter(range(5)))
ws_instance.__anext__ = coro_mock()
handle_msg_result = 'Message processed'
MockWebSocketMessageHandler.handle_message.side_effect = Exception(
handle_msg_result)
msg_handler = MockWebSocketMessageHandler()
with pytest.raises(Exception) as e:
await request_handler.RpcWebsocketHandler(msg_handler)(req)
assert str(e.value) == handle_msg_result
虽然当我 运行 test 它失败并显示错误消息：
'async for' requires an object with __aiter__ method, got MagicMock
=================================================================================== FAILURES ===================================================================================
__________________________________________________________________________ test_rpc_websocket_handler __________________________________________________________________________
MockWebSocketMessageHandler = <MagicMock name='WebSocketMessageHandler' id='140687969989632'>
rpc_websocket_handler = <aiojsonrpc.request_handler.RpcWebsocketHandler object at 0x7ff47879b0f0>
@pytest.mark.asyncio
@mock.patch('aiojsonrpc.request_handler.WebSocketMessageHandler')
async def test_rpc_websocket_handler(
MockWebSocketMessageHandler,
rpc_websocket_handler
):
ws_response = 'aiojsonrpc.request_handler.WebSocketResponse'
with mock.patch(ws_response) as MockWebSocketResponse:
# MockRequest = mock.create_autospec(aiohttp.web_reqrep.Request)
# req = MockRequest(*[None] * 6)
MockRequest = mock.MagicMock()
req = MockRequest()
ws_instance = MockWebSocketResponse.return_value
ret = mock.Mock()
ws_instance.prepare = coro_mock()
ws_instance.__aiter__ = coro_mock(return_value=iter(range(5)))
ws_instance.__anext__ = coro_mock()
handle_msg_result = 'Message processed'
MockWebSocketMessageHandler.handle_message.side_effect = Exception(
handle_msg_result)
msg_handler = MockWebSocketMessageHandler()
with pytest.raises(Exception) as e:
await request_handler.RpcWebsocketHandler(msg_handler)(req)
> assert str(e.value) == handle_msg_result
E assert "'async for' ...got MagicMock" == 'Message processed'
E - 'async for' requires an object with __aiter__ method, got MagicMock
E + Message processed
tests/test_request_handler.py:252: AssertionError
所以它的行为就像 __aiter__() 从未被嘲笑过。
在这种情况下我应该如何完成正确的模拟？
更新：
目前我已经找到 workaround 来使代码可测试，但如果有人告诉我如何处理原始问题中描述的问题，我将不胜感激。
您可以使模拟的 class return 成为实现预期接口的对象：
class AsyncIterator:
def __init__(self, seq):
self.iter = iter(seq)
def __aiter__(self):
return self
async def __anext__(self):
try:
return next(self.iter)
except StopIteration:
raise StopAsyncIteration
MockWebSocketResponse.return_value = AsyncIterator(range(5))
我认为（还）没有办法正确地模拟实现 __aiter__ 的对象，它可能是一个 python 错误，因为 async for 拒绝 MagicMock，即使 hasattr(the_magic_mock, '__aiter__') 是 True.
编辑（13/12/2017）：库 asynctest 从 0.11 开始支持异步迭代器和上下文管理器，asynctest.MagicMock 免费提供此功能。
为了后代，我遇到了同样的问题，需要测试一个 async for 循环，但公认的解决方案似乎不适用于 Python 3.7。下面的示例适用于 3.6.x 和 3.7.0，但 不适用于 3.5.x:
import asyncio
class AsyncIter:
def __init__(self, items):
self.items = items
async def __aiter__(self):
for item in self.items:
yield item
async def print_iter(items):
async for item in items:
print(item)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
things = AsyncIter([1, 2, 3])
loop.run_until_complete(print_iter(things))
loop.close()
根据以上内容，模拟它看起来像这样：
with mock.patch('some.async.iter', return_value=AsyncIter([1, 2, 3])):
# do test requiring mocked iter
适用于 py38
from unittest.mock import MagicMock
async def test_iterable(self):
loop_iterations = 0
mock = MagicMock()
mock.__aiter__.return_value = range(5)
async for _ in mock:
loop_iterations += 1
self.assertEqual(5, loop_iterations)
我有一个支持 AsyncMock 的 python 版本，我还利用了 pytest_mock。我想出了这个问题的解决方案结合使用 AsyncMock side_effect:
from typing import List
import pytest
import asyncio
from pytest_mock.plugin import MockerFixture
pytestmark = pytest.mark.asyncio
async def async_generator(numbers: List[int]):
for number in numbers:
yield number
await asyncio.sleep(0.1)
async def function_to_test(numbers: List[int]):
async for thing in async_generator(numbers):
yield thing * 3
await asyncio.sleep(0.1)
async def test_async_generator(mocker: MockerFixture):
mock_numbers = [1, 2, 3, 4, 5]
async def async_generator_side_effect(numbers: List[int]):
for number in numbers:
yield number
mock_async_generator = mocker.patch("tests.test_async_generator.async_generator")
mock_async_generator.side_effect = async_generator_side_effect
actual = []
async for result in function_to_test(mock_numbers):
actual.append(result)
assert actual == [3, 6, 9, 12, 15]
我有一个代码正在使用 aiohttp.
看起来像：
async for msg in ws:
await self._ws_msg_handler.handle_message(ws, msg, _services)
其中 ws 是 aiohttp.web.WebSocketResponse() (original code)
在我的测试中，我模拟了 WebSocketResponse() 及其 __aiter__ 方法：
def coro_mock(**kwargs):
return asyncio.coroutine(mock.Mock(**kwargs))
@pytest.mark.asyncio
@mock.patch('aiojsonrpc.request_handler.WebSocketMessageHandler')
async def test_rpc_websocket_handler(
MockWebSocketMessageHandler,
rpc_websocket_handler
):
ws_response = 'aiojsonrpc.request_handler.WebSocketResponse'
with mock.patch(ws_response) as MockWebSocketResponse:
MockRequest = mock.MagicMock()
req = MockRequest()
ws_instance = MockWebSocketResponse.return_value
ws_instance.prepare = coro_mock()
ws_instance.__aiter__ = coro_mock(return_value=iter(range(5)))
ws_instance.__anext__ = coro_mock()
handle_msg_result = 'Message processed'
MockWebSocketMessageHandler.handle_message.side_effect = Exception(
handle_msg_result)
msg_handler = MockWebSocketMessageHandler()
with pytest.raises(Exception) as e:
await request_handler.RpcWebsocketHandler(msg_handler)(req)
assert str(e.value) == handle_msg_result
虽然当我 运行 test 它失败并显示错误消息：
'async for' requires an object with
__aiter__method, got MagicMock
=================================================================================== FAILURES ===================================================================================
__________________________________________________________________________ test_rpc_websocket_handler __________________________________________________________________________
MockWebSocketMessageHandler = <MagicMock name='WebSocketMessageHandler' id='140687969989632'>
rpc_websocket_handler = <aiojsonrpc.request_handler.RpcWebsocketHandler object at 0x7ff47879b0f0>
@pytest.mark.asyncio
@mock.patch('aiojsonrpc.request_handler.WebSocketMessageHandler')
async def test_rpc_websocket_handler(
MockWebSocketMessageHandler,
rpc_websocket_handler
):
ws_response = 'aiojsonrpc.request_handler.WebSocketResponse'
with mock.patch(ws_response) as MockWebSocketResponse:
# MockRequest = mock.create_autospec(aiohttp.web_reqrep.Request)
# req = MockRequest(*[None] * 6)
MockRequest = mock.MagicMock()
req = MockRequest()
ws_instance = MockWebSocketResponse.return_value
ret = mock.Mock()
ws_instance.prepare = coro_mock()
ws_instance.__aiter__ = coro_mock(return_value=iter(range(5)))
ws_instance.__anext__ = coro_mock()
handle_msg_result = 'Message processed'
MockWebSocketMessageHandler.handle_message.side_effect = Exception(
handle_msg_result)
msg_handler = MockWebSocketMessageHandler()
with pytest.raises(Exception) as e:
await request_handler.RpcWebsocketHandler(msg_handler)(req)
> assert str(e.value) == handle_msg_result
E assert "'async for' ...got MagicMock" == 'Message processed'
E - 'async for' requires an object with __aiter__ method, got MagicMock
E + Message processed
tests/test_request_handler.py:252: AssertionError
所以它的行为就像 __aiter__() 从未被嘲笑过。
在这种情况下我应该如何完成正确的模拟？
更新：
目前我已经找到 workaround 来使代码可测试，但如果有人告诉我如何处理原始问题中描述的问题，我将不胜感激。
您可以使模拟的 class return 成为实现预期接口的对象：
class AsyncIterator:
def __init__(self, seq):
self.iter = iter(seq)
def __aiter__(self):
return self
async def __anext__(self):
try:
return next(self.iter)
except StopIteration:
raise StopAsyncIteration
MockWebSocketResponse.return_value = AsyncIterator(range(5))
我认为（还）没有办法正确地模拟实现 __aiter__ 的对象，它可能是一个 python 错误，因为 async for 拒绝 MagicMock，即使 hasattr(the_magic_mock, '__aiter__') 是 True.
编辑（13/12/2017）：库 asynctest 从 0.11 开始支持异步迭代器和上下文管理器，asynctest.MagicMock 免费提供此功能。
为了后代，我遇到了同样的问题，需要测试一个 async for 循环，但公认的解决方案似乎不适用于 Python 3.7。下面的示例适用于 3.6.x 和 3.7.0，但 不适用于 3.5.x:
import asyncio
class AsyncIter:
def __init__(self, items):
self.items = items
async def __aiter__(self):
for item in self.items:
yield item
async def print_iter(items):
async for item in items:
print(item)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
things = AsyncIter([1, 2, 3])
loop.run_until_complete(print_iter(things))
loop.close()
根据以上内容，模拟它看起来像这样：
with mock.patch('some.async.iter', return_value=AsyncIter([1, 2, 3])):
# do test requiring mocked iter
适用于 py38
from unittest.mock import MagicMock
async def test_iterable(self):
loop_iterations = 0
mock = MagicMock()
mock.__aiter__.return_value = range(5)
async for _ in mock:
loop_iterations += 1
self.assertEqual(5, loop_iterations)
我有一个支持 AsyncMock 的 python 版本，我还利用了 pytest_mock。我想出了这个问题的解决方案结合使用 AsyncMock side_effect:
from typing import List
import pytest
import asyncio
from pytest_mock.plugin import MockerFixture
pytestmark = pytest.mark.asyncio
async def async_generator(numbers: List[int]):
for number in numbers:
yield number
await asyncio.sleep(0.1)
async def function_to_test(numbers: List[int]):
async for thing in async_generator(numbers):
yield thing * 3
await asyncio.sleep(0.1)
async def test_async_generator(mocker: MockerFixture):
mock_numbers = [1, 2, 3, 4, 5]
async def async_generator_side_effect(numbers: List[int]):
for number in numbers:
yield number
mock_async_generator = mocker.patch("tests.test_async_generator.async_generator")
mock_async_generator.side_effect = async_generator_side_effect
actual = []
async for result in function_to_test(mock_numbers):
actual.append(result)
assert actual == [3, 6, 9, 12, 15]