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

task in python

Question

¿Cuál es la mejor manera de escribir pruebas unitarias para el código utilizando la biblioteca 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())

Al ejecutar 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 aserciones. Esto hace que las pruebas siempre tengan éxito.

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

Answer 1

Normalmente defino mis pruebas asíncronas como corutinas 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())

Answer 2

Realmente me gusta el contenedor async_test mencionado en https://.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())

Answer 3

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

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, AsyncTestCase maneja el caso asíncrono. También es compatible con la prueba síncrona. Existe la posibilidad de proporcionar un bucle de evento personalizado, simplemente anule AsyncTestCase.get_event_loop .

Si prefiere (por algún motivo) la otra clase TestCase (p. Ej. unittest.TestCase ), puede usar async_test decorator:

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)

Answer 4

Utilice esta clase en lugar de unittest.TestCase base class:

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)

Answer 5

gen_test temporalmente el problema usando un decorador inspirado en 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 se 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 omita algunos casos extremos.

Creo que una instalación como esta debe agregarse a la biblioteca estándar de Python para que la interacción asyncio y unittest más conveniente de manera asyncio .

Answer 6

pytest-asyncio parece prometedor:

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

Answer 7

async_test , sugerido por Marvin Killing, definitivamente puede ayudar, así como las llamadas directas loop.run_until_complete()

Pero también recomiendo encarecidamente recrear un nuevo ciclo de eventos para cada prueba y pasar directamente el ciclo a las llamadas API (al menos asyncio acepta el parámetro de palabra clave solo 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 previene errores extraños como la corutina de larga data que se ha creado en test_a pero finalizó solo en el tiempo de ejecución de test_b .