wshobson/temporal-python-testing

O que temporal-python-testing faz?

Teste fluxos de trabalho do Temporal com pytest, avanço de tempo e estratégias de simulação. Abrange testes unitários, de integração, de reprodução e configuração de desenvolvimento local. Use ao implementar testes de fluxo de trabalho do Temporal ou depurar falhas em testes.

Visão Geral

Este guia cobre como testar fluxos de trabalho do Temporal em Python, usando as bibliotecas pytest, time-skipping e estratégias de simulação para isolar componentes e garantir a corretude.

Tipos de Teste

Teste Unitário

• Testa a lógica do workflow sem o servidor Temporal.
• Usa temporalio.testing.WorkflowEnvironment.start_local() ou WorkflowEnvironment.time_skipping() para criar um ambiente leve.
• Simula atividades com unittest.mock ou pytest.monkeypatch.

Teste de Integração

• Conecta a um servidor Temporal real (local ou de desenvolvimento).
• Executa o fluxo completo, incluindo atividades reais (se configuradas) e o servidor.
• Útil para validar a interação entre workflow e atividades.

Teste de Reprodução (Replay)

• Verifica que um workflow pode ser reproduzido a partir de um histórico salvo.
• Garante que mudanças no código não quebrem a compatibilidade com eventos passados.
• Usa temporalio.worker.UnsandboxedWorkflowRunner ou Replayer.

Configuração Local

• Inicia um servidor Temporal local com Docker ou docker-compose.
• Configura TemporalClient e Worker apontando para o servidor.
• Permite testes manuais ou scripts de automação.

Como Usar

1. Instale as dependências:

   pip install temporalio pytest pytest-asyncio
   

2. Crie um arquivo de teste, ex.: test_workflows.py, e importe as ferramentas:

   import pytest
   from temporalio.testing import WorkflowEnvironment
   from your_workflow import YourWorkflow
   
   @pytest.mark.asyncio
   async def test_unitário_workflow():
       async with await WorkflowEnvironment.start_local() as env:
           # Executa o workflow
           handle = await env.client.start_workflow(
               YourWorkflow.run,
               id="test-id",
               task_queue="test-task-queue",
           )
           resultado = await handle.result()
           assert resultado == esperado
   

3. Para avanço de tempo, use env.sleep() ou env.time_skipping().

Boas Práticas

• Utilize time-skipping em testes de integração para acelerar temporizadores (timers).
• Simule atividades pesadas ou externas com mocks.
• Sempre execute pytest com -v para verbosidade.

Referências

• Documentação oficial de teste do Temporal Python
• Exemplos de teste no repositório Temporal Python SDK

---

Nota: Este guia é um resumo; consulte a documentação oficial para detalhes completos.