24. Probar envío de correos, notificaciones y colas de mensajes

24.1 Objetivo del tema

Enviar correos, notificaciones push, SMS o mensajes a una cola son efectos externos. En una prueba unitaria no queremos que esos efectos ocurran realmente. Queremos verificar que el sistema intentó producirlos con los datos correctos.

En este tema veremos cómo probar este tipo de colaboraciones usando mocks, spies y fakes.

24.2 Caso de correo de bienvenida

Supongamos este servicio:

class ServicioUsuarios:
    def __init__(self, repositorio, email):
        self.repositorio = repositorio
        self.email = email

    def registrar(self, datos):
        usuario_id = self.repositorio.guardar(datos)
        self.email.enviar(
            destino=datos["email"],
            asunto="Bienvenido",
            cuerpo="Tu cuenta fue creada",
        )
        return usuario_id

El comportamiento importante es que el usuario se guarda y se solicita enviar el correo. No queremos enviar un email real.

24.3 Probar con Mock

Podemos usar un fake para el repositorio y un mock para el email:

from unittest.mock import Mock


class RepositorioUsuariosFake:
    def __init__(self):
        self.usuarios = {}
        self.proximo_id = 1

    def guardar(self, usuario):
        usuario_id = self.proximo_id
        self.proximo_id += 1
        self.usuarios[usuario_id] = usuario.copy()
        return usuario_id


def test_registrar_usuario_envia_bienvenida():
    repositorio = RepositorioUsuariosFake()
    email = Mock()
    servicio = ServicioUsuarios(repositorio, email)

    usuario_id = servicio.registrar({
        "email": "ana@example.com",
        "nombre": "Ana",
    })

    assert repositorio.usuarios[usuario_id]["email"] == "ana@example.com"
    email.enviar.assert_called_once_with(
        destino="ana@example.com",
        asunto="Bienvenido",
        cuerpo="Tu cuenta fue creada",
    )

El mock verifica la interacción con el servicio de email.

24.4 Spy manual para email

También podemos usar un spy manual:

class EmailSpy:
    def __init__(self):
        self.enviados = []

    def enviar(self, destino, asunto, cuerpo):
        self.enviados.append({
            "destino": destino,
            "asunto": asunto,
            "cuerpo": cuerpo,
        })

Prueba:

def test_registrar_usuario_con_email_spy():
    repositorio = RepositorioUsuariosFake()
    email = EmailSpy()
    servicio = ServicioUsuarios(repositorio, email)

    servicio.registrar({"email": "ana@example.com", "nombre": "Ana"})

    assert email.enviados == [
        {
            "destino": "ana@example.com",
            "asunto": "Bienvenido",
            "cuerpo": "Tu cuenta fue creada",
        }
    ]

El spy puede ser más expresivo si queremos guardar varios mensajes y analizarlos como datos.

24.5 Verificar que no se envía correo

También es importante probar casos donde no debe enviarse nada:

class EmailDuplicadoError(Exception):
    pass


class ServicioUsuarios:
    def __init__(self, repositorio, email):
        self.repositorio = repositorio
        self.email = email

    def registrar(self, datos):
        if self.repositorio.existe_email(datos["email"]):
            raise EmailDuplicadoError("El email ya existe")

        usuario_id = self.repositorio.guardar(datos)
        self.email.enviar(datos["email"], "Bienvenido", "Tu cuenta fue creada")
        return usuario_id

Prueba:

import pytest


def test_no_envia_email_si_el_usuario_ya_existe():
    repositorio = Mock()
    repositorio.existe_email.return_value = True
    email = Mock()
    servicio = ServicioUsuarios(repositorio, email)

    with pytest.raises(EmailDuplicadoError):
        servicio.registrar({"email": "ana@example.com"})

    email.enviar.assert_not_called()

24.6 Notificaciones push

Una notificación push se prueba de forma similar:

def notificar_cambio_estado(pedido, notificador):
    if pedido["estado"] == "enviado":
        notificador.enviar_push(
            usuario_id=pedido["usuario_id"],
            titulo="Pedido enviado",
            mensaje=f"Tu pedido {pedido['id']} fue enviado",
        )

Prueba:

def test_notificar_cambio_estado_enviado():
    notificador = Mock()
    pedido = {
        "id": 10,
        "usuario_id": "USR-1",
        "estado": "enviado",
    }

    notificar_cambio_estado(pedido, notificador)

    notificador.enviar_push.assert_called_once_with(
        usuario_id="USR-1",
        titulo="Pedido enviado",
        mensaje="Tu pedido 10 fue enviado",
    )

24.7 Notificación no enviada

Si el estado no corresponde, no debe notificarse:

def test_no_notifica_si_el_pedido_no_fue_enviado():
    notificador = Mock()
    pedido = {
        "id": 10,
        "usuario_id": "USR-1",
        "estado": "pendiente",
    }

    notificar_cambio_estado(pedido, notificador)

    notificador.enviar_push.assert_not_called()

Las pruebas negativas evitan efectos externos incorrectos.

24.8 Colas de mensajes

Una cola permite publicar mensajes para que otro proceso los consuma. En una prueba unitaria no queremos publicar en RabbitMQ, Kafka, SQS u otro sistema real.

Ejemplo:

def publicar_evento_pedido_creado(pedido, cola):
    evento = {
        "tipo": "pedido_creado",
        "pedido_id": pedido["id"],
        "usuario_id": pedido["usuario_id"],
        "total": pedido["total"],
    }
    cola.publicar("pedidos", evento)

24.9 Probar publicación en cola

Prueba con mock:

def test_publicar_evento_pedido_creado():
    cola = Mock()
    pedido = {
        "id": 10,
        "usuario_id": "USR-1",
        "total": 2500,
    }

    publicar_evento_pedido_creado(pedido, cola)

    cola.publicar.assert_called_once_with(
        "pedidos",
        {
            "tipo": "pedido_creado",
            "pedido_id": 10,
            "usuario_id": "USR-1",
            "total": 2500,
        },
    )

Verificamos el tópico o cola y el contenido del mensaje.

24.10 Fake de cola

Si varias pruebas publican mensajes, un fake puede ser más cómodo:

class ColaFake:
    def __init__(self):
        self.mensajes = []

    def publicar(self, destino, mensaje):
        self.mensajes.append({
            "destino": destino,
            "mensaje": mensaje,
        })

Uso:

def test_publicar_evento_con_cola_fake():
    cola = ColaFake()
    pedido = {"id": 10, "usuario_id": "USR-1", "total": 2500}

    publicar_evento_pedido_creado(pedido, cola)

    assert cola.mensajes == [
        {
            "destino": "pedidos",
            "mensaje": {
                "tipo": "pedido_creado",
                "pedido_id": 10,
                "usuario_id": "USR-1",
                "total": 2500,
            },
        }
    ]

24.11 Simular fallos de envío

Los servicios externos pueden fallar. Podemos simularlo con side_effect:

class NotificacionError(Exception):
    pass


def enviar_alerta(usuario, notificador):
    try:
        notificador.enviar(usuario["email"], "Alerta")
    except TimeoutError as error:
        raise NotificacionError("No se pudo enviar la alerta") from error

Prueba:

def test_enviar_alerta_traduce_timeout():
    notificador = Mock()
    notificador.enviar.side_effect = TimeoutError("Tiempo agotado")

    with pytest.raises(NotificacionError):
        enviar_alerta({"email": "ana@example.com"}, notificador)

24.12 Reintentos

Si el código reintenta, podemos usar side_effect con una secuencia:

def enviar_con_reintento(notificador, destino, mensaje):
    try:
        notificador.enviar(destino, mensaje)
    except TimeoutError:
        notificador.enviar(destino, mensaje)

Prueba:

def test_enviar_con_reintento_funciona_en_segundo_intento():
    notificador = Mock()
    notificador.enviar.side_effect = [
        TimeoutError("fallo"),
        None,
    ]

    enviar_con_reintento(notificador, "ana@example.com", "Hola")

    assert notificador.enviar.call_count == 2

24.13 Verificar contenido sin acoplarse demasiado

En correos y mensajes, a veces no conviene verificar todo el cuerpo exacto si es largo o cambia con facilidad. Podemos verificar los campos importantes:

def test_email_contiene_datos_importantes():
    email = Mock()

    email.enviar(
        destino="ana@example.com",
        asunto="Factura",
        cuerpo="Hola Ana, tu factura 10 está disponible",
    )

    args, kwargs = email.enviar.call_args
    assert kwargs["destino"] == "ana@example.com"
    assert kwargs["asunto"] == "Factura"
    assert "factura 10" in kwargs["cuerpo"]

Esto reduce fragilidad cuando el texto completo no es el foco de la prueba.

24.14 Eventos como datos

Para colas, suele ser recomendable construir eventos como diccionarios o dataclasses antes de publicarlos. Eso facilita probar el contenido.

def crear_evento_pedido_creado(pedido):
    return {
        "tipo": "pedido_creado",
        "pedido_id": pedido["id"],
        "total": pedido["total"],
    }

La creación del evento puede probarse como función pura, y la publicación puede probarse por separado.

24.15 Separar construcción y envío

Ejemplo:

def publicar_pedido_creado(pedido, cola):
    evento = crear_evento_pedido_creado(pedido)
    cola.publicar("pedidos", evento)

Esto permite una prueba simple para el evento y otra para la colaboración con la cola.

24.16 Buenas prácticas

• No envíes correos, SMS ni mensajes reales desde pruebas unitarias.
• Usa mocks para verificar interacciones puntuales.
• Usa spies o fakes si quieres inspeccionar varios mensajes como datos.
• Prueba también los casos donde no debe enviarse nada.
• Evita verificar textos largos completos si solo importan algunos datos.

24.17 Ejercicio práctico

Prueba esta función:

def confirmar_reserva(reserva, email, cola):
    email.enviar(
        destino=reserva["email"],
        asunto="Reserva confirmada",
        cuerpo=f"Tu reserva {reserva['id']} fue confirmada",
    )
    cola.publicar("reservas", {
        "tipo": "reserva_confirmada",
        "reserva_id": reserva["id"],
    })
    return True

Verifica que se envía el correo y se publica el evento correcto.

24.18 Solución posible del ejercicio

Una solución con mocks:

from unittest.mock import Mock

from reservas import confirmar_reserva


def test_confirmar_reserva_envia_email_y_publica_evento():
    email = Mock()
    cola = Mock()
    reserva = {
        "id": 7,
        "email": "ana@example.com",
    }

    resultado = confirmar_reserva(reserva, email, cola)

    assert resultado is True
    email.enviar.assert_called_once_with(
        destino="ana@example.com",
        asunto="Reserva confirmada",
        cuerpo="Tu reserva 7 fue confirmada",
    )
    cola.publicar.assert_called_once_with(
        "reservas",
        {
            "tipo": "reserva_confirmada",
            "reserva_id": 7,
        },
    )

24.19 Conclusión

Correos, notificaciones y colas son efectos externos que normalmente deben reemplazarse en pruebas unitarias. Con mocks, spies y fakes podemos verificar que el sistema solicita esos efectos con los datos correctos sin ejecutarlos realmente.

En el próximo tema veremos mocking en código asíncrono con AsyncMock.