loop_factory: isole event loops em testes asyncio

loop_factory em IsolatedAsyncioTestCase permite controlar como o loop de eventos é criado durante testes assíncronos no Python. Em vez de depender apenas da política global de eventos, uma classe de teste pode fornecer uma fábrica própria, tornando a suíte mais previsível, isolada e adequada para ambientes diferentes. O recurso é especialmente útil em bibliotecas assíncronas, aplicações de rede, clientes HTTP, filas, serviços e projetos que precisam testar comportamento ligado ao event loop.

O problema que loop_factory resolve

Testes assíncronos precisam de um loop de eventos. Quando esse loop depende de configuração global, diferentes arquivos de teste podem interferir entre si. Plugins, frameworks ou configurações da máquina também podem alterar a política padrão. Isso gera falhas difíceis de reproduzir, recursos não fechados e diferenças entre sistemas operacionais.

Com uma fábrica explícita, cada caso de teste pode criar o loop de forma controlada. A intenção fica documentada no próprio código e o ciclo de vida permanece associado à classe. Essa abordagem complementa o isolamento oferecido por unittest.IsolatedAsyncioTestCase.

Estrutura básica

import asyncio
import unittest

class TestServico(unittest.IsolatedAsyncioTestCase):
    loop_factory = asyncio.new_event_loop

    async def test_resposta(self):
        await asyncio.sleep(0)
        self.assertTrue(True)

A classe executa cada método assíncrono em um contexto isolado. A fábrica deve ser chamável e devolver um novo loop de eventos. Evite reutilizar uma instância global, pois isso elimina parte do isolamento e pode carregar tarefas pendentes de um teste para outro.

Quando personalizar a fábrica

A personalização faz sentido quando o projeto precisa de um tipo específico de loop, deseja instrumentar a criação, validar configurações ou garantir comportamento consistente. Em sistemas Unix, por exemplo, pode haver diferenças entre implementações. Em testes de biblioteca, também é útil criar uma subclasse que registra chamadas ou ajusta o modo de depuração.

def criar_loop_debug():
    loop = asyncio.new_event_loop()
    loop.set_debug(True)
    return loop

class TestConcorrencia(unittest.IsolatedAsyncioTestCase):
    loop_factory = staticmethod(criar_loop_debug)

O modo de depuração ajuda a detectar callbacks lentos, corrotinas esquecidas e operações suspeitas. Ele não substitui assertions, mas adiciona sinais importantes durante o desenvolvimento.

setUp e tearDown assíncronos

IsolatedAsyncioTestCase oferece asyncSetUp e asyncTearDown. Use esses métodos para abrir e fechar recursos ligados ao loop, como servidores, conexões, workers e tarefas.

class TestCliente(unittest.IsolatedAsyncioTestCase):
    loop_factory = asyncio.new_event_loop

    async def asyncSetUp(self):
        self.fila = 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.fila.get()

O encerramento explícito evita avisos de tarefas destruídas enquanto ainda estavam pendentes. Para aprofundar esse tema, veja asyncio no Python e testes unitários com unittest.

Testando timeouts

Timeouts devem ser curtos e determinísticos. Evite depender de longas pausas reais. Prefira asyncio.wait_for, eventos e filas para controlar exatamente quando uma operação deve concluir.

async def esperar_para_sempre():
    await asyncio.Event().wait()

class TestTimeout(unittest.IsolatedAsyncioTestCase):
    loop_factory = asyncio.new_event_loop

    async def test_timeout(self):
        with self.assertRaises(asyncio.TimeoutError):
            await asyncio.wait_for(esperar_para_sempre(), timeout=0.01)

Mesmo com um loop novo por teste, tempos muito apertados podem falhar em máquinas carregadas. Escolha margens pequenas, mas realistas, e teste o resultado lógico em vez de medir desempenho.

Testando cancelamento

Cancelamento faz parte do contrato de muitas APIs assíncronas. Um bom teste verifica se a tarefa recebe CancelledError, executa a limpeza e não deixa recursos abertos.

class TestCancelamento(unittest.IsolatedAsyncioTestCase):
    async def test_cancela_tarefa(self):
        tarefa = asyncio.create_task(asyncio.sleep(60))
        tarefa.cancel()
        with self.assertRaises(asyncio.CancelledError):
            await tarefa

Ao criar várias tarefas, mantenha referências e finalize todas no teardown. Não dependa apenas do encerramento automático do loop, pois isso pode esconder falhas de limpeza na aplicação real.

Mocks assíncronos

unittest.mock.AsyncMock combina bem com IsolatedAsyncioTestCase. Ele permite validar awaits, argumentos e quantidade de chamadas sem acessar serviços externos.

from unittest.mock import AsyncMock

class TestRepositorio(unittest.IsolatedAsyncioTestCase):
    async def test_salva_item(self):
        salvar = AsyncMock(return_value=42)
        resultado = await salvar({"nome": "Ada"})
        self.assertEqual(resultado, 42)
        salvar.assert_awaited_once_with({"nome": "Ada"})

Para mais práticas, consulte mock no Python e pytest no Python. Mesmo quando o projeto usa pytest, compreender unittest ajuda a entender o ciclo de vida dos testes da biblioteca padrão.

Compatibilidade entre versões

Antes de definir loop_factory, confirme a versão mínima suportada pelo projeto. Bibliotecas distribuídas para várias versões podem precisar de uma classe base condicional ou manter a configuração padrão nas versões antigas. Documente essa decisão no arquivo de configuração e na matriz de CI.

import sys

if sys.version_info >= (3, 13):
    class AsyncCase(unittest.IsolatedAsyncioTestCase):
        loop_factory = asyncio.new_event_loop
else:
    class AsyncCase(unittest.IsolatedAsyncioTestCase):
        pass

A verificação evita usar atributos não disponíveis. A documentação oficial de unittest deve ser a referência para a versão utilizada. Consulte também a documentação de event loops do asyncio.

Evite alterar estado global

Uma vantagem da fábrica por classe é reduzir a necessidade de chamar asyncio.set_event_loop_policy. Alterações globais podem afetar testes executados depois, inclusive em paralelo. Quando uma alteração global for inevitável, salve o valor anterior e restaure no teardown da classe.

Também evite misturar loops manualmente. Futures, Locks, Queues e Tasks pertencem ao contexto em que foram criados. Passar objetos ligados a um loop para outro costuma produzir erros ou comportamento indefinido.

Detecção de tarefas pendentes

Uma suíte robusta pode verificar se o teste deixou tarefas ativas. Faça essa checagem com cuidado para não contar a própria tarefa do método.

async def cancelar_pendentes():
    atual = asyncio.current_task()
    pendentes = [
        t for t in asyncio.all_tasks()
        if t is not atual and not t.done()
    ]
    for tarefa in pendentes:
        tarefa.cancel()
    await asyncio.gather(*pendentes, return_exceptions=True)

Esse padrão é útil como defesa, mas o ideal é que cada teste encerre os recursos que criou. A limpeza automática não deve substituir uma API bem projetada.

Organização da suíte

Crie uma classe base pequena com a fábrica e utilitários comuns. Separe testes de rede, banco, filas e regras de negócio. Use nomes que expressem o comportamento esperado e mantenha cada teste focado em uma única garantia.

Evite compartilhar clientes, conexões ou tarefas entre métodos. Fixtures por teste são mais lentas em alguns casos, porém reduzem acoplamento e tornam falhas independentes. Para acelerar a suíte, otimize as dependências e use doubles, não um loop compartilhado sem controle.

Exemplo completo

import asyncio
import unittest

async def processar(fila, saida):
    while True:
        item = await fila.get()
        try:
            if item is None:
                return
            saida.append(item * 2)
        finally:
            fila.task_done()

class TestPipeline(unittest.IsolatedAsyncioTestCase):
    loop_factory = asyncio.new_event_loop

    async def asyncSetUp(self):
        self.fila = asyncio.Queue()
        self.saida = []
        self.worker = asyncio.create_task(
            processar(self.fila, self.saida)
        )

    async def asyncTearDown(self):
        if not self.worker.done():
            self.worker.cancel()
        await asyncio.gather(self.worker, return_exceptions=True)

    async def test_processa_itens(self):
        await self.fila.put(2)
        await self.fila.put(5)
        await self.fila.join()
        self.assertEqual(self.saida, [4, 10])

    async def test_encerramento(self):
        await self.fila.put(None)
        await self.worker
        self.assertTrue(self.worker.done())

O exemplo demonstra criação isolada, setup, teardown, fila, worker e encerramento. A classe não depende de uma política externa e mantém o ciclo de vida explícito.

Conclusão

loop_factory torna os testes com IsolatedAsyncioTestCase mais controláveis. O recurso ajuda a evitar estado global, facilita instrumentação e documenta qual loop deve executar a suíte. O benefício real aparece quando ele é combinado com limpeza explícita, timeouts determinísticos, mocks assíncronos e verificação de tarefas pendentes. Assim, testes de concorrência ficam mais confiáveis, portáveis e próximos do comportamento esperado em produção.