代码之家 › 专栏 › 技术社区 › Eugene Naydenov

Python异步:如何模拟aiter()方法?

python-unittest.mock aiohttp python-asyncio pytest python-3.x

Eugene Naydenov · 技术社区 · 8 年前

我有一个代码正在使用WebSocket上侦听消息 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”需要具有 __升降机__ 方法,获得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 回复 | 直到 8 年前

Pavel Martin Richard 4 年前

您可以使模拟类返回实现预期接口的对象:

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 bug async for 拒绝 MagicMock 即使 hasattr(the_magic_mock, '__aiter__') 是 True .

编辑(2017年12月13日) :库asynctest自0.11起支持异步迭代器和上下文管理器asyncttest。MagicMock免费提供此功能。

Lex Scarisbrick 6 年前

对于后代来说,我也有同样的问题,需要测试 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

golkedj 3 年前

我有一个支持 AsyncMock 我还利用 pytest_mock .我结合使用 异步模拟 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]

Limon 4 年前

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)

Python异步:如何模拟__aiter__()方法?

Python异步:如何模拟aiter()方法?