FastAPI: contratos de tiempo para emails de prueba

Cuando una prueba de registro depende de un email, el fallo no siempre está en FastAPI. Puede que el mensaje tarde, que el buzón de prueba todavía no esté listo o que el worker de CI haya perdido un evento. El problema aparece cuando la prueba solo sabe decir “espera un poco más”.

En mis automatizaciones de backend me resulta más útil tratar la llegada del email como un pequeño contrato: estados claros, un presupuesto de tiempo y una evidencia que permita repetir el diagnóstico. Así el flujo deja de ser un sleep(10) misterioso y se convierte en una pieza que podemos observar.

Este patrón sirve para pruebas de verificación, upgrades y webhooks. También combina bien con una cuenta aislada —por ejemplo, un burner email generator— siempre que el buzón se use solo para datos de prueba y no para información real.

El problema: esperar no es una estrategia

Un polling ingenuo suele verse así:

await asyncio.sleep(10)
message = await mailbox.find(subject="Verify")

Tiene tres defectos. No distingue entre “todavía no llegó” y “la API ya falló”, tarda lo mismo cuando el mensaje llega en un segundo y oculta cuántos intentos se hicieron. En un portátil puede parecer suficiente; bajo carga, ese sleep se multiplica por cada worker y el diagnóstico se vuelve confuso.

Además, un correo temporal o una bandeja efímera no debe ser la única fuente de verdad. La aplicación debería exponer un request_id o un identificador de evento que conecte el signup, el envío y la lectura del mensaje. Es un detalle chico, pero ahorra bastante tiempo.

Define un contrato de estados

Antes de escribir el endpoint, define estados que una prueba pueda interpretar:

• queued: se aceptó la solicitud y el email está en cola.
• sent: el proveedor de correo confirmó el envío.
• received: el fixture encontró el mensaje esperado.
• expired: se terminó el presupuesto de espera.
• failed: ocurrió un error no recuperable.

El cliente de pruebas no debería inferir un estado mirando texto libre. Un esquema pequeño con status, request_id, attempts y last_checked_at es suficiente para comenzar. Para evitar carreras, haz que received sea idempotente: leer dos veces el mismo email no debe crear dos verificaciones.

Si ya estás trabajando con SaaS, vale la pena probar upgrades por email sin ruido. La misma separación entre evento, buzón y resultado evita que una prueba de pago dependa accidentalmente de un mensaje viejo.

Implementa polling con un presupuesto de tiempo

Un helper sencillo puede recibir un deadline, no solo una cantidad fija de intentos:

import asyncio
import time
from collections.abc import Awaitable, Callable


async def wait_for_message(
    lookup: Callable[[], Awaitable[dict | None]],
    timeout_seconds: float = 30.0,
    interval_seconds: float = 1.0,
) -> dict:
    deadline = time.monotonic() + timeout_seconds
    attempts = 0

    while time.monotonic() < deadline:
        attempts += 1
        message = await lookup()
        if message is not None:
            return {"message": message, "attempts": attempts}
        remaining = deadline - time.monotonic()
        await asyncio.sleep(min(interval_seconds, max(remaining, 0)))

    raise TimeoutError(f"email not received after {attempts} attempts")

time.monotonic() es importante: un cambio en el reloj del sistema no debe alargar la prueba. En un sistema real también limitaría el número máximo de mensajes revisados y filtraría por request_id, destinatario y una marca de tiempo reciente. Buscar solo por asunto es una receta para falsos positivos, aunque el asunto se vea correcto.

Un backoff corto puede reducir llamadas cuando la bandeja tarda, pero no lo uses para tapar una cola saturada. Registra el intervalo, el número de intentos y el motivo del último resultado. Esas métricas cuentan más que otro retry ciego.

Haz que FastAPI y CI dejen evidencia

El endpoint puede devolver el estado actual sin bloquear el worker:

from fastapi import FastAPI

app = FastAPI()


@app.get("/test-emails/{request_id}")
async def email_status(request_id: str) -> dict:
    state = await load_state(request_id)
    return {
        "request_id": request_id,
        "status": state.status,
        "attempts": state.attempts,
        "last_checked_at": state.last_checked_at,
    }

El job de CI puede consultar este endpoint y guardar el JSON de la última respuesta como artefacto. Si el test falla, conviene conservar el request_id, pero nunca el contenido completo del email si podría contener tokens. Redacta enlaces de verificación y limita los logs; privacidad importa incluso en fixtures.

Para evitar que una prueba antigua contamine la siguiente, crea un identificador único por ejecución y elimina el mensaje después de una prueba exitosa o vencida. La limpieza debe ocurrir en un finally, y debe tolerar que el recurso ya no exista. Esta parte suele olvidarse en el primer borrador, luego el buzón crece y las pruebas empiezan a leer resultados equivocados.

Una guía relacionada sobre emails transaccionales sin mezclar bandejas ayuda a pensar en esta frontera como una decisión de arquitectura, no como un truco de QA.

Q&A: dudas habituales

¿Debo hacer polling o esperar un webhook?

Usa webhook cuando el entorno sea estable y puedas verificar la firma del evento. Mantén polling como fallback para pruebas locales y para diagnosticar entregas tardías. Ambos caminos deben actualizar el mismo contrato de estados.

¿Qué timeout es correcto?

Mide primero el tiempo normal de tu proveedor y añade margen pequeño. Un timeout de 30 segundos puede ser razonable para CI, pero no es una constante universal. Si cada ejecución necesita 5 minutos, revisa la cola antes de subir el número.

¿Puedo reutilizar un buzón?

Sí, si cada mensaje tiene un identificador único y hay limpieza verificable. Reutilizarlo sin aislamiento crea una prueba que pasa por accidente. Incluso tem email o tempail que aparecen en una búsqueda manual no deben entrar como criterio de coincidencia.

Checklist final

1. Genera un request_id único por prueba.
2. Modela queued, sent, received, expired y failed.
3. Usa un deadline monotónico y limita los intentos.
4. Filtra por identificador, destinatario y fecha, no solo por asunto.
5. Guarda estados y métricas, pero redacta tokens del email.
6. Limpia fixtures en finally y acepta recursos ausentes.
7. Conserva un JSON pequeño cuando CI falle.

Con este contrato, FastAPI no necesita adivinar si el correo llegará. La prueba tiene un límite, el equipo tiene evidencia y una demora real deja de parecer un fallo aleatorio.