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

python-asyncio python-mock

Question

¿Cómo me burlo de una llamada asíncrona de una coroutina 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 pitónica de hacer esto?

Answer 1

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

Answer 2

La solución era 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

Answer 3

Otra forma de burlarse de la coroutina es hacer la coroutina, que se vuelve burlona. De esta manera puedes simular que las corutinas 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()

Answer 4

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.

Answer 5

Una variante más de la solución "más simple" para simular un objeto asíncrono, que es solo un trazador de líneas.

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

Answer 6

MagicMock propagará su clase personalizada para todas las simulaciones generadas a partir de su simulacro de rutina. 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 documentation ).

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 : la función coroutina simulacro.
• corofunc.side_effect() : la coroutina, generada para cada llamada
• corofunc.coro : el simulacro usado 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