FastAPI: un reloj para emails de prueba

Los flujos de registro y recuperación de contraseña suelen fallar por una razón poco visible: nadie ha definido cuanto tiempo debe esperar el sistema por un email. En local, un correo llega enseguida. En CI, la misma prueba puede quedarse esperando, reintentando y dejando datos dificiles de interpretar.

Un correo electrónico desechable o un generador de cuentas de correo temporal puede ayudar a aislar cada ejecución, pero no resuelve por si solo el problema del tiempo. En este artículo propongo un patrón pequeño con FastAPI: tratar el reloj como parte del contrato de la automatización.

Por qué el tiempo es parte del contrato del email

Una prueba de verificación no solo necesita una dirección. Necesita saber:

• cuándo se creó el buzón temporal;
• cuánto tiempo puede esperar el primer mensaje;
• cuántos reintentos están permitidos;
• qué evidencia se guarda cuando no llega nada.

Sin estos límites, un sleep(10) termina siendo una política escondida. A veces es demasiado corto y produce un falso fallo; otras veces es demasiado largo y hace lenta toda la suite. El resultado parece aleatorio, aunque el problema real es que la prueba no tiene un deadline explicito.

La automatización gana claridad cuando cada ejecución tiene un run_id, un límite de tiempo y un estado final. Así es mas fácil distinguir entre “el proveedor tardó”, “el webhook llegó tarde” y “la aplicación nunca solicitó el mensaje”.

Un modelo pequeño de deadline en FastAPI

Podemos representar el contrato con un modelo de Pydantic y calcular el tiempo restante antes de cada espera. El ejemplo no depende de un proveedor concreto; la función read_inbox puede ser un cliente HTTP interno o un adaptador de pruebas.

from datetime import datetime, timedelta, timezone
from pydantic import BaseModel, Field


class EmailLease(BaseModel):
    run_id: str
    address: str
    created_at: datetime
    deadline_seconds: int = Field(default=45, ge=1, le=300)

    def expires_at(self) -> datetime:
        return self.created_at + timedelta(seconds=self.deadline_seconds)

    def remaining_seconds(self) -> float:
        now = datetime.now(timezone.utc)
        return max(0.0, (self.expires_at() - now).total_seconds())

El endpoint puede recibir una solicitud de lectura y devolver un estado explícito. No conviene bloquear una petición web durante toda la espera; para una suite de integración, un worker o una tarea de fondo suele ser mejor. Aun así, el mismo contrato sirve para ambos caminos.

from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/runs/{run_id}/email-status")
async def email_status(run_id: str):
    lease = await load_lease(run_id)
    if lease is None:
        raise HTTPException(status_code=404, detail="run_id desconocido")

    remaining = lease.remaining_seconds()
    if remaining <= 0:
        return {"run_id": run_id, "status": "expired", "remaining": 0}

    message = await read_inbox(lease.address, timeout=min(remaining, 5))
    if message is None:
        return {"run_id": run_id, "status": "waiting", "remaining": round(remaining, 2)}

    return {"run_id": run_id, "status": "received", "message_id": message.id}

En producción, load_lease y read_inbox necesitan límites propios y manejo de errores. El punto importante es que nunca deben esperar más que el tiempo que queda en el contrato.

Separar espera, reintentos y evidencia

Un timeout no debe decidir por accidente cuánto reintentar. Son tres decisiones distintas:

1. Espera: cuánto dura una lectura individual, por ejemplo cinco segundos.
2. Reintento: cuántas lecturas se permiten antes del fallo, por ejemplo nueve.
3. Evidencia: qué se registra en cada intento: run_id, número de intento, tiempo restante y resultado.

Para depurar, guardo un evento compacto por intento. No guardo el contenido completo del correo si no hace falta. Esta separación hace mas simple comparar dos ejecuciones y evita que un log lleno de ruido esconda el fallo.

También conviene usar un idempotency key al pedir el mensaje de prueba. Si el worker se reinicia, no debería crear tres buzones por el mismo escenario. Un nombre como signup:{run_id} permite localizar los datos y limpiarlos después.

Si el equipo busca una implementación de colas, este enfoque combina bien con unas colas de email con trazas útiles. Y cuando una ejecución sí termina en incidente, los correos de incidentes sin ruido operativo ofrecen una buena referencia para decidir qué contexto conservar.

Cómo probar el reloj sin pruebas lentas

No hace falta esperar 45 segundos en cada test. Inyecta un reloj y un lector de bandeja falsos. Para una prueba, el reloj puede empezar en 12:00:00; el primer intento devuelve None, el segundo también y el tercero devuelve un mensaje. Otra prueba avanza el reloj más allá del deadline y verifica expired.

El test debe comprobar el comportamiento, no los segundos exactos de una máquina compartida. Un esquema útil es:

• crea una lease con un deadline de 10 segundos;
• simula dos lecturas sin mensaje;
• devuelve un mensaje con el tercer intento;
• verifica que el estado final es received;
• verifica que cada lectura recibió un timeout menor o igual al tiempo restante.

Para documentación y datos de ejemplo, algunas personas escriben nombres como dummy e mail o temp gamil com. Pueden servir para probar búsqueda y validación, pero mantenlos como texto de prueba, nunca como direcciones reales ni como una regla de negocio.

Checklist para automatización estable

• [ ] Cada ejecución tiene un run_id único.
• [ ] La lease guarda creación y expiración.
• [ ] El timeout de lectura respeta el tiempo restante.
• [ ] Los reintentos tienen un máximo explícito.
• [ ] Los eventos registran resultado y contexto mínimo.
• [ ] Los tests usan reloj y lector inyectables.
• [ ] Un proceso de limpieza elimina buzones y mensajes expirados.

Este patrón es pequeño, pero cambia la conversación del equipo. En lugar de preguntar “¿por qué a veces falla el email?”, podemos preguntar “¿en qué estado terminó la lease y cuánto tiempo quedaba?”. Esa pregunta suele llevar mas rápido a la causa.

Preguntas frecuentes

¿Debo hacer polling o usar webhooks?

Usa webhooks si el proveedor y tu entorno los soportan de forma fiable. Mantén un polling acotado como respaldo para pruebas locales o servicios que no tienen eventos. En ambos casos, el deadline debe ser único y visible.

¿Un buzón temporal sustituye las pruebas de email reales?

No. Es útil para flujos de integración, aislamiento y datos efímeros. Las pruebas de entrega, reputación y compatibilidad de plantillas necesitan otros entornos y destinatarios controlados.

¿Qué métrica empezaría a medir?

Mide el tiempo desde la solicitud hasta la recepción, la tasa de expiración y el número de reintentos. Separar esos valores revela si el cuello de botella está en FastAPI, en la cola o en el proveedor de correo.