FastAPI: contratos de tiempo para emails de prueba

Cuando una prueba espera un email de verificación, el timeout suele ser el primer sospechoso. Pero subirlo de 30 a 90 segundos no arregla una prueba que no sabe distinguir entre un mensaje nuevo y uno viejo. En una API de FastAPI, el problema real suele ser la falta de un contrato de tiempo claro.

Un contrato sencillo define cuándo empieza la espera, qué señales cuentan como éxito, cuántos reintentos se permiten y qué evidencia queda si la prueba falla. Esta idea hace que los emails de prueba sean más previsibles y que la automatización de CI tenga algo útil que explicar.

El problema no es solo el timeout

Imagina un endpoint que crea una cuenta y dispara un correo. El test hace una petición, consulta el buzón y busca el mensaje más reciente. Si el proveedor tarda, la consulta puede devolver cero mensajes. Si otro test comparte el buzón, puede devolver el mensaje equivocado. La prueba falla en la linea de lectura, aunque el bug esté en el aislamiento.

También hay una carrera menos visible: la API responde antes de que el proveedor acepte el mensaje, pero el test empieza a buscar demasiado pronto. Un sleep(5) parece ayudar, aunque solo mueve el problema. En el siguiente runner, el retraso puede ser diferente.

La respuesta no es llamar a cualquier correo temporal y esperar más. Es registrar una referencia propia para cada ejecución: run_id, instante de inicio y destinatario. Así la prueba sabe que debe aceptar un mensaje posterior a su inicio y dirigido a su identidad.

Define un contrato de tiempo

Antes de hacer polling, crea un objeto pequeño con las reglas de la espera:

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

@dataclass(frozen=True)
class EmailWait:
    started_at: datetime
    deadline: datetime
    recipient: str
    run_id: str

    @classmethod
    def start(cls, recipient: str, run_id: str, seconds: int = 30):
        now = datetime.now(timezone.utc)
        return cls(now, now + timedelta(seconds=seconds), recipient, run_id)

El contrato hace explícito algo que normalmente queda repartido entre fixtures y variables de entorno. started_at filtra mensajes antiguos, deadline limita el coste y run_id permite unir el resultado con el log de CI. Si la prueba utiliza una dirección compartida, el destinatario solo no será suficiente: añade también un asunto o una marca única.

Conviene medir el tiempo desde una señal estable. Si el endpoint devuelve 202 Accepted, empieza la espera cuando la API confirma que el trabajo fue encolado, no varios pasos antes. Para procesos largos, un SLO para emails de CI ayuda a convertir esa expectativa en una métrica operable.

Implementa el polling en FastAPI

La ruta no debería bloquearse durante toda la espera. Una opción práctica es que el endpoint de prueba cree un trabajo y que un worker consulte el proveedor. El código del test puede usar un cliente asíncrono con pausas cortas y un límite absoluto:

import asyncio
from datetime import datetime, timezone

async def wait_for_message(mailbox, wait: EmailWait):
    while datetime.now(timezone.utc) < wait.deadline:
        messages = await mailbox.list_messages(recipient=wait.recipient)
        for message in messages:
            received = message.received_at
            if received >= wait.started_at and wait.run_id in message.subject:
                return message
        await asyncio.sleep(1.5)
    raise TimeoutError(f"No llegó un email para {wait.run_id} antes del límite")

En producción, el adaptador de mailbox puede llamar a una API externa; en tests unitarios, puede ser un fake. Esa separación vuelve la prueba más rápida y permite comprobar los casos difíciles: mensaje tardío, respuesta vacía, duplicados y error temporal del proveedor. Un detalle pequeño pero importante es que el sleep debe tener una pausa razonable, no un bucle ocupado que consuma CPU.

No escondas todos los errores detrás de TimeoutError. Si el proveedor devuelve 401, 429 o 500, guarda esa diferencia en el resultado. Un retry para un 429 tiene sentido; repetir una credencial inválida solo hace el diagnóstico más confuso.

Guarda evidencia sin guardar todo el correo

Cuando falla un test, conserva un resumen seguro: run_id, destinatario anonimizado, número de mensajes vistos, IDs de mensajes, timestamps y causa final. No hace falta guardar el cuerpo completo, porque puede contener tokens o datos personales.

En los logs, normaliza los errores de entrada. Una búsqueda mal escrita como temp gamil com no debe transformarse en una regla de producción ni en un filtro de aceptación. Parece obvio, pero este tipo de texto acaba a veces en variables de configuración copiadas de una incidencia.

Para interfaces que muestran el estado del trabajo, los mensajes de error también deben ser estables. Las ideas de formularios con errores accesibles aplican aquí: una persona debe entender si el proceso está esperando, terminó o necesita reintento, sin depender solo del color.

Checklist para CI

Antes de dar por estable el flujo, comprueba lo siguiente:

• Cada ejecución tiene un run_id único y un instante de inicio.
• El filtro usa destinatario, timestamp y una señal del flujo.
• Hay un límite absoluto; el polling no es infinito.
• Los códigos 429 y 5xx tienen reintentos acotados.
• El reporte conserva IDs y tiempos, pero no el contenido sensible.
• La limpieza ocurre en un bloque finally, incluso tras una aserción fallida.
• El adaptador externo tiene un fake para tests unitarios.
• Un fallo de entrega se distingue de un fallo de parsing.

Esto tambien facilita comparar ejecuciones. Si el tiempo de entrega crece durante varios días, tienes una señal para investigar antes de que todos los tests se vuelvan lentos.

Preguntas frecuentes

¿Debo esperar en el endpoint de FastAPI?

Normalmente no. Devuelve una respuesta que represente el trabajo encolado y deja el polling a un worker o al cliente de pruebas. Bloquear el request puede agotar los workers de la aplicación cuando el proveedor de correo está lento.

¿Cuántos reintentos son suficientes?

Depende del proveedor, pero el límite debe estar relacionado con el presupuesto de la prueba. Tres reintentos rápidos no son una garantía; son una política que debe quedar visible y medirse.

¿Qué hago con un mensaje duplicado?

Trátalo como un resultado ambiguo y registra sus IDs. Elegir silenciosamente el primero puede ocultar una entrega repetida o una fuga entre pruebas.

El objetivo no es hacer que el email llegue siempre rápido. Es convertir una dependencia externa en una operación con identidad, tiempo y evidencia. Con ese contrato, FastAPI queda más fácil de probar, y la automatización deja de depender de pausas mágicas.