Python asyncio:如何模拟__aiter__()方法？
Python asyncio:如何模拟__aiter__()方法？
提问于 2016-04-18 21:22:31
我有一个使用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”需要具有
__aiter__方法的对象，已获取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来使代码可测试，但如果有人告诉我如何处理原始问题中描述的问题，我会非常感激。
回答 4
Stack Overflow用户
回答已采纳
发布于 2016-04-20 00:27:28
您可以让模拟的类返回一个实现预期接口的对象：
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。
EDIT (13/12/2017)：asynctest库自0.11起支持异步迭代器和上下文管理器，asynctest.MagicMock免费提供此功能。
Stack Overflow用户
发布于 2018-08-17 03:21:19
对于后人来说，我遇到了同样的问题，需要测试一个async for循环，但是被接受的解决方案似乎不适用于Python3.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()有了上面的内容，mocking看起来像这样：
with mock.patch('some.async.iter', return_value=AsyncIter([1, 2, 3])):
# do test requiring mocked iterStack Overflow用户
发布于 2020-11-28 01:56:34
适用于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)页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/36695256
复制相关文章
相似问题
腾讯云开发者
