FastAPI: contrato de espera para emails de prueba

Probar un registro con confirmación por email suele empezar de forma simple: llamar al endpoint, abrir un buzón y esperar a que llegue un mensaje. El problema aparece cuando el test falla. ¿La aplicación no creó el email, el proveedor tardó, la fixture fue compartida o el enlace era incorrecto?

En proyectos con FastAPI, una solución práctica es definir un contrato de eventos antes de hacer polling. Así el test no espera “un email cualquiera”, sino una señal concreta que puede correlacionarse, expirar y explicarse en CI. La idea también funciona cuando usas un correo temporal o una dirección de correo desechable para aislar pruebas.

El problema de esperar un email sin contrato

Un sleep(10) no describe qué está ocurriendo. Solo aplaza el diagnóstico y hace que cada ejecución lenta parezca un fallo de entrega. En una suite con varios workers, además, una fixture puede quedar disponible demasiado pronto y otro test la toma. Eso hace los logs se vuelva dificil de leer cuando dos problemas ocurren juntos.

Antes de consultar el buzón, conviene registrar al menos:

• run_id: identifica esta ejecución y su worker.
• fixture_id: identifica el buzón reservado.
• expected_event: el evento que debería crear el email.
• expected_subject: una comprobación rápida de intención.
• deadline: el momento máximo de espera.

Este contexto convierte un timeout en una pregunta concreta. También ayuda a separar un error de negocio de un problema en la entrega. En una prueba de rotación de secretos por correo, por ejemplo, no basta con saber que llegó un mensaje: hay que demostrar que pertenece al secreto y a la ejecución correctos.

Define el evento antes de llamar al endpoint

El contrato debe existir antes de enviar el formulario o crear el usuario. Un ejemplo pequeño podría ser:

from datetime import datetime, timedelta, timezone
from uuid import uuid4


def new_email_expectation(fixture_id: str) -> dict[str, str]:
    now = datetime.now(timezone.utc)
    return {
        "run_id": str(uuid4()),
        "fixture_id": fixture_id,
        "expected_event": "signup.verification.created",
        "expected_subject": "Confirma tu cuenta",
        "deadline": (now + timedelta(seconds=45)).isoformat(),
    }

Después, el endpoint puede recibir el run_id como un identificador de correlación. No hace falta guardar el token completo en el contrato. Un ID de evento o un digest del token es suficiente para comprobar que el mensaje observado corresponde al flujo, sin convertir los logs en una fuente de datos sensibles.

Este patrón es parecido a trabajar con fixtures de email que expiran en FastAPI, pero añade una frontera visible entre la reserva del buzón y la observación del mensaje.

Un contrato pequeño para FastAPI

En la aplicación, el evento puede modelarse con Pydantic. La respuesta no necesita exponer contenido privado; solo el estado que la automatización necesita:

from pydantic import BaseModel


class VerificationCreated(BaseModel):
    event_id: str
    run_id: str
    fixture_id: str
    subject: str


@app.post("/internal/test-events/verification", status_code=202)
async def record_verification(event: VerificationCreated) -> dict[str, str]:
    await event_store.append(event.model_dump())
    return {"status": "accepted", "event_id": event.event_id}

En producción, este endpoint debe estar protegido y separado de las rutas públicas. Para pruebas de integración, lo importante es que el test pueda verificar la primera mitad del recorrido: la aplicación aceptó el evento. Luego puede consultar el buzón para validar la segunda mitad.

Haz que el polling explique su espera

El poller debe tener una pausa limitada, un cursor y estados explícitos. Por ejemplo, puede devolver pending, matched, wrong_message o expired. No es lo mismo un buzón vacío que un mensaje de otra ejecución.

async def wait_for_message(mailbox, expected, poll_every=2):
    while datetime.now(timezone.utc) < expected.deadline:
        messages = await mailbox.messages(after=expected.cursor)
        for message in messages:
            if message.run_id != expected.run_id:
                continue
            if message.subject == expected.expected_subject:
                return {"status": "matched", "message_id": message.id}
            return {"status": "wrong_message", "message_id": message.id}
        await asyncio.sleep(poll_every)
    return {"status": "expired", "run_id": expected.run_id}

En código real, convierte deadline a datetime y valida el origen de cada mensaje. El ejemplo se centra en la decisión: el test sabe que hacer con cada estado. Si el proveedor está temporalmente lento, un reintento acotado puede tener sentido. Si el mensaje pertenece a otro run_id, reintentar no arregla la propiedad de la fixture.

Aísla la fixture y clasifica el fallo

Reserva el buzón con un propietario y libera la fixture al terminar, incluso cuando el test falla. Las fixtures cruzadas puede crear resultados que parecen aleatorios, aunque el problema sea determinista. Un nombre descriptivo como temp org mail puede aparecer en una prueba de entrada inválida, pero nunca debe aceptarse como dominio o identificador real.

También conviene imprimir un resumen seguro:

status=wrong_message run_id=ci-1842 fixture_id=mailbox-7f2
expected=signup.verification.created observed=password.reset

Oculta direcciones completas, tokens y cuerpos de email. Para investigar, guarda un ID de mensaje o una referencia controlada. Una rutina de observabilidad como esta se complementa bien con una prueba de rotación de secretos por correo, donde la identidad del evento importa tanto como su llegada.

Checklist para CI

Antes de dar por estable la prueba, comprueba que:

• la fixture tiene un propietario y una fecha de expiración;
• el contrato se crea antes de llamar al endpoint;
• cada evento lleva run_id y, si es posible, event_id;
• el polling usa cursor, deadline y estados claros;
• wrong_message no se trata como un timeout transitorio;
• los logs omiten tokens y contenido sensible;
• la limpieza ocurre en un bloque finally.

El objetivo no es esperar más tiempo. Es saber por que estamos esperando y qué límite tiene esa espera. Cuando la señal de email es explícita, FastAPI, el buzón y CI pueden fallar de una manera mucho más útil.

Preguntas rápidas

¿Un correo temporal hace fiable el test por sí solo?

No. Aísla los datos, pero la fiabilidad viene del propietario, la correlación, el deadline y la limpieza. El buzón es una dependencia, no la estrategia completa.

¿Cuándo conviene reintentar?

Solo ante estados posiblemente transitorios, como un timeout del proveedor. Un mensaje equivocado, una fixture ocupada o un evento ausente requieren diagnóstico, no más intentos ciegos.

¿Cuál es el cambio más pequeño?

Añade run_id, fixture_id y failure_class al resultado actual. Incluso ese cambio sencillo vuelve mucho más claro qué pasó y que debería revisar el siguiente desarrollador.