8

を使用して WebSocket でメッセージをリッスンするコードがありますaiohttp。

次のようになります。

async for msg in ws:
    await self._ws_msg_handler.handle_message(ws, msg, _services)

(元のコード)wsのインスタンスはどこにありますかaiohttp.web.WebSocketResponse()

私のテストでは、モック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

テストを実行すると、次のようなエラー メッセージが表示されて失敗します。

「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__()そのため、嘲笑されたことがないかのように動作します。この場合、どのように正しいモッキングを達成することになっていますか?

アップデート：

今のところ、コードをテスト可能にするための回避策を見つけましたが、元の質問で説明されている問題に対処する方法を誰かが教えてくれれば幸いです。

4

4 に答える 4

10

モック化されたクラスが、期待されるインターフェースを実装するオブジェクトを返すようにすることができます。

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__を実装するオブジェクトを正しくasync forモックする方法は(まだ) ないと思います。MagicMockhasattr(the_magic_mock, '__aiter__')True

編集 (2017 年 13 月 12 日) : ライブラリ asynctest は 0.11 以降、非同期イテレーターとコンテキスト マネージャーをサポートし、asynctest.MagicMock はこの機能を無料で提供します。

于 2016-04-19T16:27:28.463 に答える