Python: pruebas de email con un contrato de evento

Las pruebas de verificación por email suelen fallar por una razón poco visible: el equipo trata el correo como un mensaje, cuando en realidad es un evento distribuido. La aplicación dispara una acción, un proveedor entrega el mensaje y la prueba debe encontrarlo dentro de una ventana concreta.

Cuando esa ventana no está escrita en ninguna parte, aparecen síntomas conocidos: el test pasa en local pero no en CI, el polling abre un mensaje viejo o dos workers consumen el mismo correo. No siempre es un bug de Python ni del proveedor. Muchas veces falta un contrato pequeño que diga qué ocurrió, cuándo ocurrió y qué mensaje pertenece a esa ejecución.

En este artículo propongo ese contrato para pruebas con Python y APIs. La idea sirve tanto si usas un generador de correo temporal como si tienes un buzón de pruebas interno. El servicio solo debe entregar eventos identificables; la prueba decide si el evento es suyo.

El problema: un email no es solo un mensaje

Imagina una prueba que crea una cuenta, solicita un código y espera el último correo recibido. El enfoque parece suficiente hasta que una ejecución anterior deja un mensaje en el buzón. Si el asunto coincide, la prueba puede recoger el token equivocado y fallar mucho después, cuando intenta usarlo.

También sucede al reintentar. El primer intento quizá creó el usuario, pero perdió la respuesta por un timeout. El segundo intenta crear otra cuenta y recibe dos eventos. Sin una identidad de ejecución, el diagnóstico se vuelve una adivinanza un poco cara.

Por eso me gusta modelar la entrega como un evento con cuatro datos mínimos:

• run_id: identifica la ejecución de la prueba.
• triggered_at: marca el instante en que se pidió el email.
• message_id: identifica el mensaje observado.
• consumed_at: indica cuándo la prueba aceptó el evento.

Un frontend puede seguir esta misma idea al probar estados de email sin perder señales. La interfaz muestra estados; el backend necesita además una prueba de pertenencia.

Qué debe contener el contrato de evento

No hace falta guardar el contenido completo del email en todos los casos. Para una prueba normal, el contrato puede ser un diccionario serializable:

from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)
class EmailEvent:
    run_id: str
    message_id: str
    subject: str
    received_at: datetime
    verification_url: str | None = None

El run_id debe viajar desde la prueba hasta el registro de la cuenta o el alias usado. Si el proveedor no permite filtros por metadatos, puedes incluir un fragmento seguro del identificador en el asunto o usar un buzón aislado por ejecución.

Hay un detalle que se suele olvidar: received_at no es lo mismo que triggered_at. Un mensaje puede aparecer con retraso, por eso la regla correcta es aceptar received_at >= triggered_at con una tolerancia explícita. Si no se documenta esa tolerancia, cada equipo la interpreta de una forma distinta y el test se vuelve inconsistantemente verde.

Polling con límites e idempotencia

El polling debe tener una fecha de inicio, un intervalo y un límite total. Un bucle sin stop rule consume recursos y además oculta que el sistema no entregó el email. Este helper es deliberadamente simple:

import time
from collections.abc import Callable

def wait_for_event(
    fetch: Callable[[], list[EmailEvent]],
    run_id: str,
    timeout_s: float = 30,
    interval_s: float = 1,
) -> EmailEvent:
    deadline = time.monotonic() + timeout_s
    seen: set[str] = set()

    while time.monotonic() < deadline:
        for event in fetch():
            if event.message_id in seen:
                continue
            seen.add(event.message_id)
            if event.run_id == run_id:
                return event
        time.sleep(interval_s)

    raise TimeoutError(f"No email event for run_id={run_id}")

La colección seen hace que leer la misma página dos veces no cambie el resultado. Eso es idempotencia en una escala muy práctica. En producción de pruebas, también conviene registrar los message_id rechazados y el número de intentos; esa información ayuda a ver si el filtro es demasiado estricto.

Si un compañero busca “tamp mail com” en los apuntes, no es una estrategia de aislamiento: es una señal de que falta una fixture clara. Vale la pena corregir la fixture, no solo repetir la prueba.

Cómo exponerlo en una API FastAPI

Una API de pruebas puede guardar el contrato antes de iniciar una acción externa. El endpoint no debería esperar al email: devuelve el run_id y otro proceso consulta el buzón.

from uuid import uuid4
from fastapi import FastAPI

app = FastAPI()

@app.post("/verification-runs")
def create_verification_run() -> dict[str, str]:
    run_id = str(uuid4())
    # Persistir run_id y triggered_at antes de enviar el email.
    return {"run_id": run_id, "status": "waiting"}

Este patrón separa el envío de la lectura y hace más fácil observar cada etapa. Si tu flujo usa webhooks, también puedes comparar este diseño con webhooks de email en FastAPI sin carreras: en ambos casos, la clave es que una entrega tardía no cambie de dueño.

Checklist para una prueba mantenible

Antes de dar por estable una prueba, compruebo lo siguiente:

1. Cada ejecución tiene un run_id único.
2. El instante de disparo se guarda antes de la llamada externa.
3. El filtro no usa únicamente el asunto del mensaje.
4. El polling tiene timeout e intervalo visibles.
5. Los mensajes observados quedan en el log de la ejecución.
6. El test consume un evento una sola vez.
7. El buzón o alias se elimina al terminar, incluso si la prueba falla.

No necesitas un sistema enorme para conseguirlo. Un contrato de evento bien nombrado suele quitar más flakiness que añadir reintentos al azar. Y cuando algo falla, el equipo puede responder qué pasó con datos, no con intuición.