Mocking async-вызов в python 3.5

Как мне высмеять асинхронный вызов из одной встроенной сопрограммы в другую с помощью 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)

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

Ответ 1

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

Вместо этого вы можете захотеть определить фабрику, которая создает Mock (или 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

Ответ 2

Все упускают то, что, вероятно, самое простое и ясное решение:

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

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

Ответ 3

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

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

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

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

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

Ответ 4

Другой способ издевательства над сопрограммой состоит в том, чтобы сделать сопрограмму, которая возвращает насмешку. Таким образом вы можете высмеять сопрограммы, которые будут переданы в 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()

Ответ 5

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

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

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

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

Ответ 6

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

В источнике:

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())