Burlándose de una llamada asíncrona en Python 3.5

¿Cómo me burlo de una llamada asíncrona de una coroutine nativa a otra usando unittest.mock.patch ?

Actualmente tengo una solución bastante incómoda:

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

Entonces

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

Esto funciona pero se ve feo. ¿Hay alguna forma más pythonica de hacer esto?

    La subclasificación de MagicMock propagará su clase personalizada para todas las simulaciones generadas a partir de su simulación de coroutine. Por ejemplo, AsyncMock().__str__ también se convertirá en un AsyncMock que probablemente no sea lo que está buscando.

    En su lugar, es posible que desee definir una fábrica que cree un Mock (o un MagicMock ) con argumentos personalizados, por ejemplo side_effect=coroutine(coro) . Además, podría ser una buena idea separar la función de coroutine de la coroutine (como se explica en la documentación ).

    Aquí está lo que se me ocurrió:

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

    Una explicación de los diferentes objetos:

    • corofunc : el simulacro de la función coroutina.
    • corofunc.side_effect() : la coroutina, generada para cada llamada
    • corofunc.coro : el simulacro utilizado por la coroutine para obtener el resultado
    • corofunc.coro.return_value : el valor devuelto por la coroutine
    • corofunc.coro.side_effect : podría usarse para generar una excepción

    Ejemplo:

     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 

    La solución fue bastante simple: solo necesitaba convertir el método __call__ de simulacro en coroutine:

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

    Esto funciona perfectamente, cuando se llama simulacro, el código recibe coroutine nativo

    Todos se pierden, lo que probablemente sea la solución más simple y clara:

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

    recuerde que una coroutina puede considerarse simplemente como una función que garantiza devolver un futuro que, a su vez, puede ser esperado.

    Otra forma de burlarse de la coroutina es hacer la coroutina, que se vuelve burlona. De esta manera, puedes simular que los coroutines se pasarán a asyncio.wait o asyncio.wait_for .

    Esto hace que las corrutinas más universales hagan que la configuración de las pruebas sea más complicada:

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

    Basado en la respuesta de @scolvin, creé esta (imo) forma más limpia:

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

    Eso es todo, solo úselo alrededor de cualquier retorno que quiera ser asíncrono, como en

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

    Una variante más de la solución “más simple” para simular un objeto asíncrono, que es solo una línea.

    En fuente:

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

    En prueba:

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