¿Cómo probar el código asyncio Python 3.4?

¿Cuál es la mejor manera de escribir pruebas unitarias para el código usando la biblioteca de Python 3.4 asyncio ? Supongamos que quiero probar un cliente TCP ( SocketConnection ):

 import asyncio import unittest class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @asyncio.coroutine def test_sends_handshake_after_connect(self): yield from self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake()) 

Cuando se ejecuta este caso de prueba con el corredor de prueba predeterminado, la prueba siempre tendrá éxito ya que el método se ejecuta solo hasta el primer yield from instrucción, después de lo cual regresa antes de ejecutar cualquier aseveración. Esto hace que las pruebas siempre tengan éxito.

¿Existe un corredor de prueba precomstackdo que pueda manejar código asíncrono como este?

Resolví temporalmente el problema usando un decorador inspirado en el gen_test de Tornado:

 def async_test(f): def wrapper(*args, **kwargs): coro = asyncio.coroutine(f) future = coro(*args, **kwargs) loop = asyncio.get_event_loop() loop.run_until_complete(future) return wrapper 

Como sugirió JF Sebastian, este decorador bloqueará hasta que el método de prueba coroutine haya terminado. Esto me permite escribir casos de prueba como este:

 class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @async_test def test_sends_handshake_after_connect(self): yield from self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake()) 

Esta solución probablemente pierde algunos casos de borde.

Creo que una instalación como esta debería agregarse a la biblioteca estándar de Python para hacer que la interacción de asyncio y unittest más conveniente fuera de la caja.

async_test , sugerido por Marvin Killing, definitivamente puede ayudar, así como también dirigir el loop.run_until_complete() llamadas.run_until_complete loop.run_until_complete()

Pero también recomiendo encarecidamente recrear un nuevo bucle de evento para cada prueba y pasar directamente el bucle a las llamadas a la API (al menos asyncio acepta el parámetro de solo palabra clave del loop para cada llamada que lo necesite).

Me gusta

 class Test(unittest.TestCase): def setUp(self): self.loop = asyncio.new_event_loop() asyncio.set_event_loop(None) def test_xxx(self): @asyncio.coroutine def go(): reader, writer = yield from asyncio.open_connection( '127.0.0.1', 8888, loop=self.loop) yield from asyncio.sleep(0.01, loop=self.loop) self.loop.run_until_complete(go()) 

que aísla las pruebas en el caso de prueba y evita errores extraños como la coroutine de larga data que se creó en test_a pero que terminó solo en el tiempo de ejecución de test_b .

pytest-asyncio parece prometedor:

 @pytest.mark.asyncio async def test_some_asyncio_code(): res = await library.do_something() assert b'expected result' == res 

Realmente me gusta el envoltorio async_test mencionado en https://stackoverflow.com/a/23036785/350195 , aquí hay una versión actualizada para Python 3.5+

 def async_test(coro): def wrapper(*args, **kwargs): loop = asyncio.new_event_loop() return loop.run_until_complete(coro(*args, **kwargs)) return wrapper class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @async_test async def test_sends_handshake_after_connect(self): await self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake()) 

Use esta clase en lugar de la clase base unittest.TestCase :

 import asyncio import unittest class AioTestCase(unittest.TestCase): # noinspection PyPep8Naming def __init__(self, methodName='runTest', loop=None): self.loop = loop or asyncio.get_event_loop() self._function_cache = {} super(AioTestCase, self).__init__(methodName=methodName) def coroutine_function_decorator(self, func): def wrapper(*args, **kw): return self.loop.run_until_complete(func(*args, **kw)) return wrapper def __getattribute__(self, item): attr = object.__getattribute__(self, item) if asyncio.iscoroutinefunction(attr): if item not in self._function_cache: self._function_cache[item] = self.coroutine_function_decorator(attr) return self._function_cache[item] return attr class TestMyCase(AioTestCase): async def test_dispatch(self): self.assertEqual(1, 1) 

También puede usar aiounittest que adopta un enfoque similar al de @Andrew Svetlov, @Marvin Killing responde y envuélvalo en una clase fácil de usar de AsyncTestCase :

 import asyncio import aiounittest async def add(x, y): await asyncio.sleep(0.1) return x + y class MyTest(aiounittest.AsyncTestCase): async def test_async_add(self): ret = await add(5, 6) self.assertEqual(ret, 11) # or 3.4 way @asyncio.coroutine def test_sleep(self): ret = yield from add(5, 6) self.assertEqual(ret, 11) # some regular test code def test_something(self): self.assertTrue(true) 

Como puede ver, el caso async es manejado por AsyncTestCase . Soporta también la prueba síncrona. Existe la posibilidad de proporcionar un bucle de eventos personalizado, simplemente anule AsyncTestCase.get_event_loop .

Si prefiere (por alguna razón) la otra clase TestCase (por ejemplo, unittest.TestCase ), puede usar el decorador async_test :

 import asyncio import unittest from aiounittest import async_test async def add(x, y): await asyncio.sleep(0.1) return x + y class MyTest(unittest.TestCase): @async_test async def test_async_add(self): ret = await add(5, 6) self.assertEqual(ret, 11) 

Normalmente defino mis pruebas asíncronas como corrutinas y uso un decorador para “sincronizarlas”:

 import asyncio import unittest def sync(coro): def wrapper(*args, **kwargs): loop = asyncio.get_event_loop() loop.run_until_complete(coro(*args, **kwargs)) return wrapper class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @sync async def test_sends_handshake_after_connect(self): await self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake())