Блог

Как мне имитировать асинхронный вызов из одной встроенной сопрограммы в другую с помощью unittest.mock.patch ?

В настоящее время у меня довольно неудобное решение:

 class CoroutineMock(MagicMock):
    def __await__(self, *args, **kwargs):
        future = Future()
        future.set_result(self)
        result = yield from future
        return result
  

Затем

 class TestCoroutines(TestCase):
    @patch('some.path', new_callable=CoroutineMock)
    def test(self, mock):
        some_action()
        mock.assert_called_with(1,2,3)
  

Это работает, но выглядит уродливо. Есть ли более питонический способ сделать это?

Решение было на самом деле довольно простым: мне просто нужно было преобразовать __call__ метод mock в сопрограмму:

 class AsyncMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)
  

Это работает отлично, при вызове mock код получает встроенную сопрограмму

Пример использования:

 @mock.patch('my.path.asyncio.sleep', new_callable=AsyncMock)
def test_stuff(sleep):
    # code
  

Всем не хватает того, что, вероятно, является самым простым и ясным решением:

 @patch('some.path')
def test(self, mock):
    f = asyncio.Future()
    f.set_result('whatever result you want')
    process_smtp_message.return_value = f
    mock.assert_called_with(1, 2, 3)
  

помните, что сопрограмму можно рассматривать как просто функцию, которая гарантированно возвращает будущее, которого, в свою очередь, можно ожидать.

Основываясь на ответе @scolvin, я создал этот (imo) более чистый способ:

 import asyncio

def async_return(result):
    f = asyncio.Future()
    f.set_result(result)
    return f
  

Вот и все, просто используйте его для любого возврата, который вы хотите сделать асинхронным, как в

 mock = MagicMock(return_value=async_return("Example return"))
await mock()
  

Создание подклассов MagicMock распространит ваш пользовательский класс на все макеты, созданные из вашего макета сопрограммы. Например, AsyncMock().__str__ также станет AsyncMock , что, вероятно, не то, что вы ищете.

Вместо этого вы можете захотеть определить фабрику, которая создает, например, Mock (или a MagicMock ) с пользовательскими аргументами side_effect=coroutine(coro) . Кроме того, было бы неплохо отделить функцию сопрограммы от сопрограммы (как описано в документации).

Вот что я придумал:

 from asyncio import coroutine

def CoroMock():
    coro = Mock(name="CoroutineResult")
    corofunc = Mock(name="CoroutineFunction", side_effect=coroutine(coro))
    corofunc.coro = coro
    return corofunc
  

Объяснение различных объектов:

• corofunc : макет функции сопрограммы
• corofunc.side_effect() : сопрограмма, генерируемая для каждого вызова
• corofunc.coro : макет, используемый сопрограммой для получения результата
• corofunc.coro.return_value : значение, возвращаемое сопрограммой
• corofunc.coro.side_effect : может использоваться для создания исключения

Пример:

 async def coro(a, b):
    return await sleep(1, result=a b)

def some_action(a, b):
    return get_event_loop().run_until_complete(coro(a, b))

@patch('__main__.coro', new_callable=CoroMock)
def test(corofunc):
    a, b, c = 1, 2, 3
    corofunc.coro.return_value = c
    result = some_action(a, b)
    corofunc.assert_called_with(a, b)
    assert result == c
  

Другой способ имитировать сопрограмму — создать сопрограмму, которая возвращает макет. Таким образом, вы можете имитировать сопрограммы, которые будут переданы в asyncio.wait или asyncio.wait_for .

Это делает более универсальные сопрограммы, хотя и делает настройку тестов более громоздкой:

 def make_coroutine(mock)
    async def coroutine(*args, **kwargs):
        return mock(*args, **kwargs)
    return coroutine


class Test(TestCase):
    def setUp(self):
        self.coroutine_mock = Mock()
        self.patcher = patch('some.coroutine',
                             new=make_coroutine(self.coroutine_mock))
        self.patcher.start()

    def tearDown(self):
        self.patcher.stop()
  

Еще один вариант «простейшего» решения для моделирования асинхронного объекта, который является всего лишь однострочным.

В исходном коде:

 class Yo:
    async def foo(self):
        await self.bar()
    async def bar(self):
        # Some code
  

В тесте:

 from asyncio import coroutine

yo = Yo()
# Here bounded method bar is mocked and will return a customised result.
yo.bar = Mock(side_effect=coroutine(lambda:'the awaitable should return this'))
event_loop.run_until_complete(yo.foo())
  

Я не знаю, почему никто не упомянул доступный параметр по умолчанию. python предоставляет асинхронную версию MagicMock.

Вы можете прочитать больше об этом здесь. https://docs.python.org/3/library/unittest.mock.html#unittest.mock .AsyncMock

В случае, если вы используете patch, вам также не нужно вносить никаких других изменений. При необходимости он автоматически заменит его асинхронной макетной функцией. Подробнее читайте здесь https://docs.python.org/3/library/unittest.mock.html#patch

Вы можете установить return_value асинхронный метод следующим образом:

 mock = unittest.mock.MagicMock()
mock.your_async_method.return_value = task_from_result(your_return_value)

async def task_from_result(result):
    return result
  

Вызывающий должен будет сделать await your_async_method(..) точно так же, как если бы метод не был издевательским.

Мне нравится этот подход, который также заставляет AsyncMock вести себя точно так же, как Mock:

 class AsyncMock:
    def __init__(self, *args, **kwargs):
        self.mock = Mock(*args, **kwargs)

    async def __call__(self, *args, **kwargs):
        return self.mock(*args, **kwargs)

    def __getattr__(self, item):
        return getattr(self.mock, item)
  

Затем вы можете работать с ним так же, как с Mock, т. Е:

 @pytest.mark.asyncio
async def test_async_mock_example(monkeypatch):
    fn = AsyncMock(side_effect=ValueError)
    with pytest.raises(ValueError):
        await fn()
    assert fn.call_count == 1