es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

La clase IsolatedAsyncioTestCase de unitest no funciona.

04 Aug, 2023 Programación

Necesito usar IsolatedAsyncioTestCase para realizar una verificación asíncrona. ¿Por qué mi código no se ejecuta en paralelo o de manera asíncrona? Necesito que test2, test3 y test4 se ejecuten de manera asíncrona.

async def check_status(name:str):
    print(f"antes {name}")
    await asyncio.sleep(5)
    print(f"después {name}")
    return True

class TestAsync(IsolatedAsyncioTestCase):

async def test_002(self):
    print("test_002")
    res = await check_status("test_002")
    self.assertTrue(res)

async def test_003(self):
    print("test_003")
    res = await check_status("test_003")
    self.assertTrue(res)

async def test_004(self):
    print("test_004")
    res = await check_status("test_004")
    self.assertTrue(res)

if __name__ == "main":
    unittest.main()

Aquí está la salida, todavía está sincronizada:

test_002
antes test_002
después test_002
test_003
antes test_003
después test_003
test_004
antes test_004
después test_004
Tags:  , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    Para ejecutar las pruebas en paralelo y de manera asíncrona, puedes agregar el método asyncSetUp a la clase TestAsync para crear el bucle de eventos y asignarlo al atributo self.loop. Luego, en cada método de prueba, puedes usar la función asyncio.gather() para ejecutar la corrutina check_status() de forma concurrente con otras pruebas. Aquí tienes un código actualizado que realiza la verificación asíncrona usando IsolatedAsyncioTestCase:

    class TestAsync(IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.loop = asyncio.get_event_loop()

    async def test_002(self):
        print("test_002")
        await asyncio.gather(
            check_status("test_002"),
            self.loop.create_future(),
        )

    async def test_003(self):
        print("test_003")
        await asyncio.gather(
            check_status("test_003"),
            self.loop.create_future(),
        )

    async def test_004(self):
        print("test_004")
        await asyncio.gather(
            check_status("test_004"),
            self.loop.create_future(),
        )

if __name__ == "__main__":
    unittest.main()

    Aquí, agregamos self.loop.create_future() a cada llamada a asyncio.gather() para mantener el bucle de eventos en ejecución hasta que todas las corrutinas hayan terminado. Este método permite que las pruebas se ejecuten de forma concurrente y asíncrona, asegurando que cada prueba se ejecute de forma independiente de las demás.

Comments are closed.