首页
学习
活动
专区
圈层
工具
发布
社区首页 >问答首页 >Python asyncio:如何模拟__aiter__()方法?

Python asyncio:如何模拟__aiter__()方法?
EN

Stack Overflow用户
提问于 2016-04-18 21:22:31
回答 4查看 8.9K关注 0票数 8

我有一个使用aiohttp在WebSocket上监听消息的代码。

看起来是这样的:

代码语言:javascript
复制
async for msg in ws:
    await self._ws_msg_handler.handle_message(ws, msg, _services)

其中wsaiohttp.web.WebSocketResponse() (original code)的实例

在我的测试中,我模拟了WebSocketResponse()及其__aiter__方法:

代码语言:javascript
复制
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

代码语言:javascript
复制
=================================================================================== 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来使代码可测试,但如果有人告诉我如何处理原始问题中描述的问题,我会非常感激。

EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/36695256

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档