Probar un flujo de registro suele parecer sencillo: crear un usuario, esperar un email y abrir el enlace. En CI, sin embargo, “esperar” es una operación ambigua. El mensaje puede llegar tarde, otro worker puede leerlo primero o una prueba anterior puede dejar basura en el inbox.
La solución que más me funciona es tratar cada envío como un evento con contrato propio. No hace falta construir una plataforma enorme. Con FastAPI, una clave de correlación y un pequeño registro de estados podemos hacer que un generador de correos falsos sea útil para probar el backend, no solo para mirar si apareció un mensaje.
El problema: esperar un email no es un contrato
Un test que hace sleep(5) solo expresa una esperanza. No sabe cuándo se solicitó el email, cuál era el destinatario esperado ni si el mensaje encontrado pertenece a la ejecución actual. Cuando falla, el equipo recibe un timeout y poco mas.
Conviene separar tres momentos:
- La aplicación acepta la solicitud de envío.
- El proveedor o adaptador registra el evento de entrega.
- El test encuentra el mensaje correcto y consume el enlace.
Cada momento debe tener un identificador común. Así se puede distinguir “el email todavía no llegó” de “el email llegó, pero el enlace ya fue consumido”. Esa diferencia parece pequeña, pero cambia por completo el diagnóstico.
También ayuda a que la interfaz muestre estados de interfaz que explican el resultado, en vez de esconder un estado pendiente detrás de un spinner infinito.
Diseñar un evento pequeño y trazable
Un contrato mínimo puede tener esta forma:
from datetime import datetime from pydantic import BaseModel, EmailStr class EmailEvent(BaseModel): event_id: str recipient: EmailStr kind: str created_at: datetime status: str = "queued" El event_id no debe generarse de nuevo cuando se reintenta la misma operación. El cliente puede enviarlo como clave de idempotencia y el servidor lo guarda junto con el estado. Si llega la misma petición, la API devuelve el evento existente en lugar de crear dos emails.
En pruebas, kind también es importante. “welcome” y “verify-account” pueden terminar en el mismo inbox, pero no tienen el mismo asunto ni la misma acción. Un filtro por tipo es mas resistente que buscar solo la palabra “verify”.
Implementación de un endpoint FastAPI
El endpoint puede aceptar el evento y entregar un estado claro al worker que enviará el mensaje:
from fastapi import FastAPI, status app = FastAPI() @app.post("/email-events", status_code=status.HTTP_202_ACCEPTED) async def create_email_event(event: EmailEvent): stored = await event_store.get(event.event_id) if stored: return stored await event_store.insert(event) await email_queue.publish(event.event_id) return event El ejemplo omite la base de datos y la cola para mantener la idea enfocada. En producción, la inserción y la publicación necesitan una estrategia consistente; de lo contrario puede guardarse el evento y perderse la tarea. Para muchos equipos, un outbox transaccional es un buen siguiente paso.
El endpoint esperan una respuesta rápida: 202 Accepted significa que la solicitud fue aceptada, no que el email ya está en el inbox. Documentar esta diferencia evita que un test intente leer el mensaje inmediatamente y declare una falla que no existe.
Polling con identidad y límite
El test debe buscar por event_id, destinatario, tipo y una ventana de tiempo. Un bucle pequeño es más confiable que un sleep fijo:
for attempt in range(8): message = await inbox.find( recipient=event.recipient, kind=event.kind, event_id=event.event_id, ) if message: assert message.link break await asyncio.sleep(1) else: raise AssertionError(f"Email no encontrado: {event.event_id}") Ocho intentos de un segundo son solo un ejemplo; el límite debe salir del tiempo normal de entrega de tu entorno. Lo importante es guardar cada respuesta observada. Si una prueba escribe “temp mailid” por un error de configuración, ese texto puede aparecer en los logs como pista, pero no debe convertirse en una palabra clave del contrato.
Para evitar carreras entre workers, el adaptador del inbox puede marcar el mensaje como reservado usando event_id. Si dos pruebas consultan a la vez, solo una consume el enlace. La limpieza ocurre despues de capturar el resultado, no antes.
Cómo automatizar la evidencia en CI
Cuando el flujo falla, guarda un recibo pequeño:
-
event_id, commit y nombre del job; - estado del evento y timestamps;
- asunto, destinatario enmascarado y message ID;
- cantidad de intentos y última respuesta del inbox;
- motivo de expiración o consumo duplicado.
No guardes el cuerpo completo si no es necesario. Una evidencia útil debe ayudar a reproducir el problema sin convertir datos de prueba en un archivo permanente. Un SLO útil para emails de CI también puede separar latencia de entrega y disponibilidad del adaptador.
La automatización queda mas simple cuando cada job crea su propio evento y elimina sus fixtures al terminar. Si el worker muere, el estado queda pendiente; un proceso de reconciliación puede marcarlo como expirado y dejar una razón. No conviene reintentar sin mirar el estado anterior: eso produce duplicados y hace el fallo menos visible.
Preguntas frecuentes
¿Necesito una cola para empezar?
No siempre. Para un servicio pequeño, un BackgroundTasks puede servir como primer adaptador. Cuando haya varios workers o reintentos, una cola con una clave de idempotencia será mas segura.
¿Por qué no basta con buscar por asunto?
Porque dos ejecuciones pueden usar el mismo asunto. El asunto es una señal humana; event_id es la identidad técnica que permite automatizar.
¿Qué debe probar el test?
Primero que la API acepta y registra el evento. Después, en una prueba de integración, que llega el mensaje correcto, el enlace funciona una vez y un reintento no crea otra identidad. Ese reparto hace los tests mas rápidos y los fallos mucho más explicables.
Un contrato de eventos no elimina la latencia de email. Lo que hace es convertir una espera opaca en una secuencia observable: solicitado, entregado, encontrado, consumido o expirado. Para un backend FastAPI, esa claridad suele valer más que añadir otro retry al azar.
