run - import async python

¿Cómo burlarse de asyncio coroutines?

Question

El siguiente código falla con TypeError: ''Mock'' object is not iterable en ImBeingTested.i_call_other_coroutines porque he reemplazado ImGoingToBeMocked por un objeto Mock.

¿Cómo puedo burlarme de las coroutinas?

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

Bueno, ya hay un montón de respuestas aquí, pero contribuiré con mi versión ampliada de la respuesta de e-satis . Esta clase simula una función asíncrona y hace un seguimiento del conteo de llamadas y de los argumentos de llamada, al igual que la clase de simulacros para las funciones de sincronización.

Probado en 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

Ejemplo de uso:

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

Answer 2

Como la biblioteca de mock no es compatible con las rutinas, creo las corutinas simuladas de forma manual y las asigno a objetos simulados. Un poco más detallado pero funciona.

Tu ejemplo puede verse así:

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 3

Estoy escribiendo un envoltorio para unittest que apunta a cortar la placa de repetición al escribir pruebas para asyncio.

El código vive aquí: https://github.com/Martiusweb/asynctest

Puedes simular una coroutine con asynctest.CoroutineMock :

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

También funciona con el atributo side_effect , y un asynctest.Mock con una spec puede devolver 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''>

Se espera que todas las características de unittest.Mock funcionen correctamente (parche (), etc.).

Answer 4

La respuesta de Dustin es probablemente la correcta en la gran mayoría de los casos. Tuve un problema diferente en el que el coroutine necesitaba devolver más de un valor, por ejemplo, simulando una operación de read() , como se describe brevemente en mi comment .

Después de algunas pruebas más, el siguiente código funcionó para mí, definiendo un iterador fuera de la función de simulación, recordando efectivamente el último valor devuelto para enviar el siguiente:

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.

Entonces, expandiendo la respuesta de Dustin, se vería como:

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)

Los dos inconvenientes inmediatos que puedo ver en este enfoque son:

1. No permite generar excepciones fácilmente (por ejemplo, primero devuelve algunos datos y luego genera un error en la segunda operación de lectura).
2. No he encontrado una manera de usar los atributos estándar de Mock .side_effect o .return_value para hacerlo más obvio y legible.

Answer 5

Partiendo de la answer de Andrew Svetlov, solo quería compartir esta función de ayuda:

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

Esto le permite utilizar el assert_called_with estándar assert_called_with , call_count y otros métodos y atributos que una assert_called_with call_count regular le ofrece.

Puedes usar esto con código en la pregunta como:

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 6

Puede usar asynctest e importar CoroutineMock o usar asynctest.mock.patch

Answer 7

Puedes crear simulacros asíncronos tu mismo:

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