Как насмехаться с асинхронными сопрограммами?

Ответ 1

Так как библиотека mock не поддерживает сопрограммы, я создаю mocked coroutines вручную и назначаю их для имитации объекта. Немного более подробный, но он работает.

Ваш пример может выглядеть так:

import asyncio
import unittest
from unittest.mock import Mock


class ImGoingToBeMocked:
    @asyncio.coroutine
    def yeah_im_not_going_to_run(self):
        yield from asyncio.sleep(1)
        return "sup"


class ImBeingTested:
    def __init__(self, hidude):
        self.hidude = hidude

    @asyncio.coroutine
    def i_call_other_coroutines(self):
        return (yield from self.hidude.yeah_im_not_going_to_run())


class TestImBeingTested(unittest.TestCase):

    def test_i_call_other_coroutines(self):
        mocked = Mock(ImGoingToBeMocked)
        ibt = ImBeingTested(mocked)

        @asyncio.coroutine
        def mock_coro():
            return "sup"
        mocked.yeah_im_not_going_to_run = mock_coro

        ret = asyncio.get_event_loop().run_until_complete(
            ibt.i_call_other_coroutines())
        self.assertEqual("sup", ret)


if __name__ == '__main__':
    unittest.main()

Ответ 2

Спринг из Andrew Svetlov answer, я просто хотел поделиться этой вспомогательной функцией:

def get_mock_coro(return_value):
    @asyncio.coroutine
    def mock_coro(*args, **kwargs):
        return return_value

    return Mock(wraps=mock_coro)

Это позволяет использовать стандартные assert_called_with, call_count и другие методы и атрибуты для обычного unittest.Mock дает вам.

Вы можете использовать это с кодом в вопросе типа:

class ImGoingToBeMocked:
    @asyncio.coroutine
    def yeah_im_not_going_to_run(self):
        yield from asyncio.sleep(1)
        return "sup"

class ImBeingTested:
    def __init__(self, hidude):
        self.hidude = hidude

    @asyncio.coroutine
    def i_call_other_coroutines(self):
        return (yield from self.hidude.yeah_im_not_going_to_run())

class TestImBeingTested(unittest.TestCase):

    def test_i_call_other_coroutines(self):
        mocked = Mock(ImGoingToBeMocked)
        mocked.yeah_im_not_going_to_run = get_mock_coro()
        ibt = ImBeingTested(mocked)

        ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines())
        self.assertEqual(mocked.yeah_im_not_going_to_run.call_count, 1)

Ответ 3

Я накладываю обертку на unittest, целью которой является разрезание шаблона при написании тестов для асинхронности.

Код живет здесь: https://github.com/Martiusweb/asynctest

Вы можете высмеять сопрограмму с помощью asynctest.CoroutineMock:

>>> mock = CoroutineMock(return_value='a result')
>>> asyncio.iscoroutinefunction(mock)
True
>>> asyncio.iscoroutine(mock())
True
>>> asyncio.run_until_complete(mock())
'a result'

Он также работает с атрибутом side_effect, а asynctest.Mock с spec может возвращать CoroutineMock:

>>> asyncio.iscoroutinefunction(Foo().coroutine)
True
>>> asyncio.iscoroutinefunction(Foo().function)
False
>>> asynctest.Mock(spec=Foo()).coroutine
<class 'asynctest.mock.CoroutineMock'>
>>> asynctest.Mock(spec=Foo()).function
<class 'asynctest.mock.Mock'>

Ожидается, что все функции unittest.Mock будут работать корректно (patch() и т.д.).

Ответ 4

Вы можете сами создавать асинхронные издевки:

import asyncio
from unittest.mock import Mock


class AsyncMock(Mock):

    def __call__(self, *args, **kwargs):
        sup = super(AsyncMock, self)
        async def coro():
            return sup.__call__(*args, **kwargs)
        return coro()

    def __await__(self):
        return self().__await__()

Ответ 5

Ответ Дастина, вероятно, правильный в подавляющем большинстве случаев. У меня была другая проблема, когда сопрограмма должна была вернуть более одного значения, например. имитируя операцию read(), как это кратко описано в comment.

После еще нескольких тестов код ниже работал у меня, определяя итератор вне функции mocking, эффективно запоминая последнее значение, возвращаемое для отправки следующего:

def test_some_read_operation(self):
    #...
    data = iter([b'data', b''])
    @asyncio.coroutine
    def read(*args):
        return next(data)
    mocked.read = Mock(wraps=read)
    # Here, the business class would use its .read() method which
    # would first read 4 bytes of data, and then no data
    # on its second read.

Итак, расширяя ответ Дастина, он будет выглядеть так:

def get_mock_coro(return_values):
    values = iter(return_values)
    @asyncio.coroutine
    def mock_coro(*args, **kwargs):
        return next(values)

    return Mock(wraps=mock_coro)

Два непосредственных недостатка, которые я вижу в этом подходе, заключаются в следующем:

• Это не позволяет легко извлекать исключения (например, сначала возвращать некоторые данные, а затем поднимать ошибку во второй операции чтения).
• Я не нашел способ использовать стандартные атрибуты Mock .side_effect или .return_value, чтобы сделать его более понятным и понятным.

Ответ 6

Ну, здесь уже есть куча ответов, но я внесу свою расширенную версию ответа e-sat. Этот класс моделирует асинхронную функцию и отслеживает количество вызовов и аргументы вызовов, как класс Mock для функций синхронизации.

Протестировано на Python 3.7.0.

class AsyncMock:
    ''' A mock that acts like an async def function. '''
    def __init__(self, return_value=None, return_values=None):
        if return_values is not None:
            self._return_value = return_values
            self._index = 0
        else:
            self._return_value = return_value
            self._index = None
        self._call_count = 0
        self._call_args = None
        self._call_kwargs = None

    @property
    def call_args(self):
        return self._call_args

    @property
    def call_kwargs(self):
        return self._call_kwargs

    @property
    def called(self):
        return self._call_count > 0

    @property
    def call_count(self):
        return self._call_count

    async def __call__(self, *args, **kwargs):
        self._call_args = args
        self._call_kwargs = kwargs
        self._call_count += 1
        if self._index is not None:
            return_index = self._index
            self._index += 1
            return self._return_value[return_index]
        else:
            return self._return_value

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

async def test_async_mock():
    foo = AsyncMock(return_values=(1,2,3))
    assert await foo() == 1
    assert await foo() == 2
    assert await foo() == 3

Ответ 7

Вы можете использовать asynctest и импортировать CoroutineMock или использовать asynctest.mock.patch

Ответ 8

Вы можете подкласс Mock действовать как функция сопрограммы:

class CoroMock(Mock):
    async def __call__(self, *args, **kwargs):
        return super(CoroMock, self).__call__(*args, **kwargs)

    def _get_child_mock(self, **kw):
        return Mock(**kw)

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

Если у вас есть фиктивный объект и вы хотите сделать определенный метод сопрограммой, вы можете использовать Mock.attach_mock следующим образом:

mock.attach_mock(CoroMock(), 'method_name')

Ответ 9

Немного упрощенный пример для Python 3. 6+ адаптирован из нескольких ответов здесь:

import unittest

class MyUnittest()

  # your standard unittest function
  def test_myunittest(self):

    # define a local mock async function that does what you want, such as throw an exception. The signature should match the function you're mocking.
    async def mock_myasync_function():
      raise Exception('I am testing an exception within a coroutine here, do what you want')

    # patch the original function 'myasync_function' with the one you just defined above, note the usage of 'wrap', which hasn't been used in other answers.
    with unittest.mock.patch('mymodule.MyClass.myasync_function', wraps=mock_myasync_function) as mock:
      with self.assertRaises(Exception):
        # call some complicated code that ultimately schedules your asyncio corotine mymodule.MyClass.myasync_function
        do_something_to_call_myasync_function()