loop_factory en IsolatedAsyncioTestCase permite controlar cómo se crea el event loop durante las pruebas asíncronas de Python. En lugar de depender únicamente de una política global, una clase de prueba puede proporcionar su propia fábrica y mantener cada ejecución aislada, predecible y más fácil de depurar. Es una función útil para bibliotecas de red, clientes HTTP, colas, workers, servicios y aplicaciones cuyo comportamiento depende de asyncio.
Qué problema resuelve loop_factory
Toda prueba asíncrona necesita un event loop. Cuando la creación depende de estado global, un archivo de pruebas puede afectar a otro. Plugins, frameworks, configuraciones locales y diferencias entre sistemas operativos también pueden cambiar el loop seleccionado. Esto produce fallos intermitentes, tareas pendientes y resultados distintos entre desarrollo y CI.
Una fábrica explícita convierte la creación del loop en parte del contrato de la prueba. Cada método recibe un entorno limpio y la clase documenta qué implementación espera. Esta idea complementa el aislamiento de unittest.IsolatedAsyncioTestCase.
Estructura básica
import asyncio
import unittest
class PruebasServicio(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def test_respuesta(self):
await asyncio.sleep(0)
self.assertTrue(True)
La fábrica debe ser invocable y devolver un loop nuevo. No conviene devolver una instancia global compartida, porque callbacks, tareas, handlers y otros estados podrían pasar de una prueba a otra.
Fábricas personalizadas
Una fábrica propia es útil cuando necesitas modo debug, instrumentación, comprobaciones específicas o comportamiento consistente entre máquinas.
def crear_loop_debug():
loop = asyncio.new_event_loop()
loop.set_debug(True)
return loop
class PruebasConcurrencia(unittest.IsolatedAsyncioTestCase):
loop_factory = staticmethod(crear_loop_debug)
El modo debug ayuda a detectar callbacks lentos, corutinas olvidadas y problemas de planificación. No sustituye las assertions, pero aporta información valiosa durante el desarrollo.
Preparación y limpieza asíncronas
IsolatedAsyncioTestCase ofrece asyncSetUp y asyncTearDown. Úsalos para crear y cerrar recursos asociados al loop actual, como servidores, clientes, tareas, workers, colas y conexiones.
class PruebasCliente(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def asyncSetUp(self):
self.cola = asyncio.Queue()
self.worker = asyncio.create_task(self.consumir())
async def asyncTearDown(self):
self.worker.cancel()
with self.assertRaises(asyncio.CancelledError):
await self.worker
async def consumir(self):
while True:
await self.cola.get()
La limpieza explícita evita avisos de tareas destruidas mientras aún estaban pendientes. También comprueba que el código de producción tenga una ruta de cierre clara. Puedes ampliar la base con los artículos de Academify sobre asyncio en Python y pruebas unitarias con unittest.
Pruebas de timeout
Los timeouts deben ser cortos y deterministas. Es mejor usar eventos, futures y colas que pausas reales largas.
async def esperar_siempre():
await asyncio.Event().wait()
class PruebasTimeout(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def test_timeout(self):
with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(esperar_siempre(), timeout=0.01)
No uses márgenes excesivamente ajustados. Una máquina de CI con carga puede introducir retrasos. La prueba debe validar el contrato lógico del timeout y no intentar medir rendimiento.
Pruebas de cancelación
La cancelación forma parte del contrato de muchas APIs asíncronas. Una buena prueba confirma que la tarea recibe CancelledError, ejecuta la limpieza y no deja recursos abiertos.
class PruebasCancelacion(unittest.IsolatedAsyncioTestCase):
async def test_cancela_tarea(self):
tarea = asyncio.create_task(asyncio.sleep(60))
tarea.cancel()
with self.assertRaises(asyncio.CancelledError):
await tarea
Mantén referencias a todas las tareas creadas. El cierre del loop puede cancelar elementos restantes, pero depender de esa limpieza automática puede ocultar defectos en la aplicación.
Uso de AsyncMock
unittest.mock.AsyncMock combina muy bien con las pruebas asíncronas aisladas. Permite validar awaits, argumentos, valores de retorno y excepciones sin acceder a servicios externos.
from unittest.mock import AsyncMock
class PruebasRepositorio(unittest.IsolatedAsyncioTestCase):
async def test_guarda_item(self):
guardar = AsyncMock(return_value=42)
resultado = await guardar({"nombre": "Ada"})
self.assertEqual(resultado, 42)
guardar.assert_awaited_once_with({"nombre": "Ada"})
También son útiles las guías sobre mock en Python y pytest en Python. Incluso si el proyecto usa pytest, entender unittest ayuda a dominar el ciclo de vida de la biblioteca estándar.
Compatibilidad entre versiones
Confirma la versión mínima de Python antes de depender de loop_factory. Las bibliotecas compatibles con varias versiones pueden necesitar una pequeña clase base condicional.
import sys
if sys.version_info >= (3, 13):
class AsyncCase(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
else:
class AsyncCase(unittest.IsolatedAsyncioTestCase):
pass
Documenta la rama de compatibilidad y pruébala en CI. La documentación oficial de unittest es la referencia para la versión objetivo. Consulta también la documentación de event loops de asyncio.
Evitar estado global
Una ventaja importante de la fábrica por clase es reducir la necesidad de usar asyncio.set_event_loop_policy. Los cambios globales afectan a pruebas posteriores y pueden entrar en conflicto con plugins. Si una modificación global es inevitable, guarda la política anterior y restáurala de forma segura.
No mezcles objetos ligados a loops distintos. Tasks, Futures, Locks, Events y Queues deben permanecer en el contexto donde fueron creados. Compartirlos entre pruebas aisladas suele provocar errores o condiciones de carrera.
Detección de tareas pendientes
Una función defensiva puede identificar y cancelar tareas abandonadas.
async def cancelar_pendientes():
actual = asyncio.current_task()
pendientes = [
tarea for tarea in asyncio.all_tasks()
if tarea is not actual and not tarea.done()
]
for tarea in pendientes:
tarea.cancel()
await asyncio.gather(*pendientes, return_exceptions=True)
Este patrón es una red de seguridad, no un sustituto de la propiedad explícita. Cada componente debe ofrecer un método de cierre y cada prueba debe cerrar lo que crea.
Organización de la suite
Crea una clase base pequeña con la fábrica y helpers comunes. Separa pruebas de transporte, persistencia, colas y reglas de negocio. Usa nombres que describan comportamiento y limita cada prueba a una garantía clara.
Evita compartir clientes, conexiones y workers activos entre métodos. Los recursos por prueba pueden costar algo más, pero reducen acoplamiento. Para acelerar la suite, reemplaza dependencias lentas por doubles en lugar de debilitar el aislamiento del loop.
Ejemplo completo
import asyncio
import unittest
async def procesar(cola, salida):
while True:
item = await cola.get()
try:
if item is None:
return
salida.append(item * 2)
finally:
cola.task_done()
class PruebasPipeline(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def asyncSetUp(self):
self.cola = asyncio.Queue()
self.salida = []
self.worker = asyncio.create_task(
procesar(self.cola, self.salida)
)
async def asyncTearDown(self):
if not self.worker.done():
self.worker.cancel()
await asyncio.gather(self.worker, return_exceptions=True)
async def test_procesa_items(self):
await self.cola.put(2)
await self.cola.put(5)
await self.cola.join()
self.assertEqual(self.salida, [4, 10])
async def test_cierre(self):
await self.cola.put(None)
await self.worker
self.assertTrue(self.worker.done())
El ejemplo combina creación aislada, setup, teardown, procesamiento de cola, propiedad de la tarea y cierre ordenado. No depende de una política externa del event loop.
Errores frecuentes
Los errores habituales incluyen devolver el mismo loop desde la fábrica, crear tareas fuera del ciclo de vida de la prueba, usar sleeps largos, ocultar cancelaciones y dejar generadores asíncronos abiertos. También es un problema cambiar la política global solo para satisfacer una clase.
Usa propiedad explícita: quien crea una tarea debe saber cómo termina. Prefiere primitivas de sincronización deterministas, recoge excepciones de workers y garantiza que el teardown sea seguro incluso cuando el setup falla a mitad.
Conclusión
loop_factory vuelve IsolatedAsyncioTestCase más configurable y transparente. Reduce estado global, permite activar debug y ayuda a que las pruebas se comporten igual en diferentes máquinas. Sus mejores resultados aparecen al combinarlo con limpieza explícita, timeouts deterministas, pruebas de cancelación, AsyncMock y controles de tareas pendientes. Así, las pruebas asíncronas son más confiables y fáciles de mantener.
