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

Question

Следующий код не работает с TypeError: 'Mock' object is not iterable в ImBeingTested.i_call_other_coroutines, потому что я заменил ImGoingToBeMocked объектом 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) 

     ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines()) 

Answer 1

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

Ваш пример может выглядеть следующим образом:

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

Answer 2

Выскочив прочь Эндрю Светлова 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) 

Answer 3

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

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

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) 

этих двух непосредственных минусов я вижу в этом подходе:

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

Answer 4

Я накладываю обертку на 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 ожидают ed для правильной работы (patch() и т. д.).

Answer 5

Вы можете создать асинхронный дразнит себя:

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