wshobson/temporal-python-testing

¿Qué hace temporal-python-testing?

Prueba flujos de trabajo de Temporal con pytest, salto temporal y estrategias de simulación. Cubre pruebas unitarias, de integración, de reproducción y configuración de desarrollo local. Útil al implementar pruebas de flujos de trabajo de Temporal o depurar fallos en pruebas.

Visión General

Esta habilidad proporciona guías y ejemplos para probar flujos de trabajo de Temporal en Python de manera efectiva. Incluye:

• Pruebas unitarias: Prueba la lógica del flujo de trabajo de forma aislada.
• Pruebas de integración: Verifica la interacción con actividades reales o simuladas.
• Pruebas de reproducción (replay): Asegura la compatibilidad con versiones anteriores del flujo de trabajo.
• Configuración de desarrollo local: Usa el servidor de desarrollo de Temporal para pruebas locales.

Requisitos Previos

• Python 3.10+
• temporalio paquete instalado (pip install temporalio)
• pytest instalado (pip install pytest)
• Acceso a un servidor de Temporal (opcional para pruebas locales)

Instalación

Instala las dependencias necesarias:

pip install temporalio pytest

Estructura del Proyecto

project/
├── workflows/
│   └── my_workflow.py
├── activities/
│   └── my_activity.py
└── tests/
    └── test_my_workflow.py

Pruebas Unitarias con Time-Skipping

Usa el decorador @workflow_test y WorkflowEnvironment para saltar el tiempo:

import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from workflows.my_workflow import MyWorkflow

@pytest.mark.asyncio
async def test_my_workflow_with_time_skip():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue="my-task-queue",
            workflows=[MyWorkflow],
        ):
            result = await env.client.execute_workflow(
                MyWorkflow.run,
                arg="test",
                id="my-workflow-id",
                task_queue="my-task-queue",
            )
            assert result == "expected_output"

Pruebas de Integración

Para probar con actividades reales:

from activities.my_activity import my_activity

# En la prueba, registrar la actividad en el Worker
async with Worker(
    env.client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    activities=[my_activity],
):
    # ... ejecutar workflow

Estrategias de Simulación (Mocking)

Simula actividades usando unittest.mock:

from unittest.mock import AsyncMock, patch

@patch("activities.my_activity.my_activity", new_callable=AsyncMock)
async def test_with_mocked_activity(mock_activity):
    mock_activity.return_value = "mocked_result"
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(...):
            result = await env.client.execute_workflow(...)
            assert result == "expected_with_mocked_result"

Pruebas de Reproducción (Replay)

Usa Replayer para verificar que un flujo de trabajo se reproduce correctamente:

from temporalio.worker import Replayer

async def test_replay():
    replayer = Replayer(workflows=[MyWorkflow])
    # Cargar historial de eventos desde un archivo
    with open("history.json", "r") as f:
        history = json.load(f)
    await replayer.replay_workflow(history)

Configuración de Desarrollo Local

Inicia el servidor de desarrollo de Temporal:

temporal server start-dev

Luego, en las pruebas, usa WorkflowEnvironment.start_local():

async with await WorkflowEnvironment.start_local() as env:
    # ... pruebas con servidor local

Buenas Prácticas

• Usa time-skipping para pruebas rápidas y deterministas.
• Simula actividades externas para evitar dependencias de red.
• Escribe pruebas de reproducción después de cambios en el flujo de trabajo.
• Mantén las pruebas en archivos separados por flujo de trabajo.
• Ejecuta pruebas con pytest -v para obtener salida detallada.

Solución de Problemas

• Error de conexión: Asegúrate de que el servidor de Temporal esté en ejecución.
• Timeouts: Verifica la configuración de timeouts del flujo de trabajo.
• Fallos en reproducción: Compara el historial con la definición actual del flujo de trabajo.

Referencias

• Documentación de Temporal Python Testing
• Repositorio temporalio/python