Para probar con pruebas unitarias código que habla MQTT, no hagas mock del broker: haz mock de la frontera que controla tu código. Envuelve el cliente detrás de una interfaz pequeña, sustitúyelo por un mock o un fake en memoria en las pruebas unitarias y reserva un broker real para una capa más fina de pruebas de integración. Así tus pruebas unitarias se ejecutan en milisegundos, nunca fallan por la red y aun así demuestran que sale el topic y el payload correctos y que los mensajes entrantes se gestionan bien.
Esta guía muestra los tres dobles de prueba que funcionan en la práctica —un cliente mock, un fake escrito a mano y un broker embebido en memoria— con ejemplos en Python (paho-mqtt 2.x) y JavaScript (MQTT.js 5), y explica dónde deja de ser útil cada uno.
Elige el doble de prueba adecuado
| Doble | Qué es | Útil para | Puntos ciegos |
|---|---|---|---|
| Cliente mock / stub | unittest.mock, jest.fn() en lugar del cliente de la librería | Verificar llamadas: topic, payload, QoS, flag retain | Wildcards, QoS, mensajes retenidos, reconexiones: nada del protocolo es real |
| Fake escrito a mano | Una pequeña clase pub/sub en memoria con la misma interfaz | Probar flujos entre varios componentes, enrutado por topics | Solo es tan correcto como tu fake; sin red ni estado de sesión |
| Broker embebido | aedes (Node.js) arrancado dentro del proceso de pruebas | Paquetes MQTT reales sin Docker, muy rápido | No es tu broker de producción; ACL, límites y clustering difieren |
| Broker real | Mosquitto/EMQX en un contenedor o una instancia de prueba alojada | Pruebas de integración y end-to-end | Más lento, requiere preparación y limpieza |
Los dos primeros son pruebas unitarias. Los dos últimos son pruebas de integración, aunque parezcan rápidas. Mantener clara esta distinción evita escribir "pruebas unitarias" que dependen en secreto de un broker en marcha, y también evita hacer mock justo del comportamiento que querías verificar.
Diseña primero la costura
Hacer mock es sencillo cuando tu código de negocio nunca llama directamente a paho.mqtt.client ni a mqtt.connect. Inyecta el cliente (o un wrapper fino):
import json
class TelemetryPublisher:
def __init__(self, client, device_id: str):
self.client = client
self.topic = f"testmqtt/devices/{device_id}/telemetry"
def report(self, temperature: float) -> None:
if not -40 <= temperature <= 125:
raise ValueError("sensor out of range")
payload = json.dumps({"t": round(temperature, 2)})
self.client.publish(self.topic, payload, qos=1, retain=False)
def handle_command(client, userdata, msg):
"""paho-mqtt on_message callback: toggle a relay on 'on'/'off' commands."""
command = msg.payload.decode().strip().lower()
if command not in ("on", "off"):
return
userdata["relay"].set(command == "on")
client.publish(msg.topic.replace("/cmd", "/state"), command, qos=1, retain=True)Aquí nada abre un socket, así que nada necesita un broker para probarse.
Python: mock de paho-mqtt con unittest.mock
Usa create_autospec para que el mock tenga la firma real de Client: un error tipográfico como client.publsh(...) hará fallar la prueba en lugar de pasar en silencio.
import json
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, create_autospec
import paho.mqtt.client as mqtt
from telemetry import TelemetryPublisher, handle_command
class TelemetryTests(unittest.TestCase):
def setUp(self):
self.client = create_autospec(mqtt.Client, instance=True)
def test_report_publishes_json_at_qos1(self):
TelemetryPublisher(self.client, "dev-42").report(21.456)
self.client.publish.assert_called_once_with(
"testmqtt/devices/dev-42/telemetry", json.dumps({"t": 21.46}), qos=1, retain=False
)
def test_out_of_range_is_not_published(self):
with self.assertRaises(ValueError):
TelemetryPublisher(self.client, "dev-42").report(500)
self.client.publish.assert_not_called()
def test_command_sets_relay_and_reports_state(self):
relay = MagicMock()
msg = SimpleNamespace(topic="testmqtt/devices/dev-42/cmd", payload=b" ON ", qos=1, retain=False)
handle_command(self.client, {"relay": relay}, msg)
relay.set.assert_called_once_with(True)
self.client.publish.assert_called_once_with(
"testmqtt/devices/dev-42/state", "on", qos=1, retain=True
)
if __name__ == "__main__":
unittest.main()Dos técnicas hacen casi todo el trabajo: verificar las llamadas a publish para el tráfico saliente y llamar tú mismo al callback con un mensaje falso para el tráfico entrante. Basta un SimpleNamespace con topic, payload, qos y retain, porque tu manejador solo lee esos atributos. Si tu código llama a client.connect() al importarse, usa unittest.mock.patch("paho.mqtt.client.Client"), pero tómalo como una señal para mover la conexión a una factoría inyectable.
JavaScript: mock de MQTT.js con Jest
Un cliente de MQTT.js es un EventEmitter, así que el mock más fiel es un EventEmitter real con métodos jest.fn(). Tu código registra los manejadores on('message') como siempre y la prueba emite los eventos.
import { EventEmitter } from 'node:events';
import { connect } from 'mqtt';
import { startAlarm } from './alarm.js'; // subscribes to sensors, publishes alarms
jest.mock('mqtt', () => ({ connect: jest.fn() }));
function fakeClient() {
const client = new EventEmitter();
client.subscribe = jest.fn((topic, opts, cb) => cb?.(null, [{ topic, qos: opts?.qos ?? 0 }]));
client.publish = jest.fn((topic, payload, opts, cb) => cb?.());
client.end = jest.fn();
return client;
}
test('publishes an alarm when temperature exceeds the threshold', () => {
const client = fakeClient();
connect.mockReturnValue(client);
startAlarm({ url: 'mqtts://broker.connectmq.com:8883', threshold: 30 });
client.emit('connect');
expect(client.subscribe).toHaveBeenCalledWith('testmqtt/sensors/+/temp', { qos: 1 }, expect.any(Function));
client.emit('message', 'testmqtt/sensors/kitchen/temp', Buffer.from('31.5'), { retain: false });
expect(client.publish).toHaveBeenCalledWith(
'testmqtt/alarms/kitchen', expect.stringContaining('"temp":31.5'), { qos: 1 }, expect.any(Function)
);
});El evento message entrega (topic, payload, packet) con payload como Buffer: conserva esa forma en el mock, o tu prueba pasará con strings mientras producción recibe Buffers. Para más patrones (temporizadores falsos para la lógica de reconexión, helpers asíncronos) consulta pruebas de MQTT con Jest.
Un fake en memoria escrito a mano
Cuando varios componentes se comunican a través de topics, verificar llamadas individuales a publish se vuelve frágil. Un fake pequeño que enruta mensajes de verdad suele ser más claro. paho-mqtt incluye topic_matches_sub, así que el matching de wildcards es correcto sin esfuerzo:
from types import SimpleNamespace
from paho.mqtt.client import topic_matches_sub
class FakeMqtt:
"""In-memory pub/sub with the subset of the paho API our code uses."""
def __init__(self):
self.subscriptions, self.retained, self.published = [], {}, []
self.on_message = None
def subscribe(self, topic, qos=0):
self.subscriptions.append(topic)
for t, payload in self.retained.items():
if topic_matches_sub(topic, t):
self._deliver(t, payload, retain=True)
def publish(self, topic, payload, qos=0, retain=False):
data = payload.encode() if isinstance(payload, str) else payload
self.published.append((topic, data, qos, retain))
if retain and data:
self.retained[topic] = data
elif retain:
self.retained.pop(topic, None) # un payload retenido vacío lo borra
if any(topic_matches_sub(s, topic) for s in self.subscriptions):
self._deliver(topic, data, retain=False)
def _deliver(self, topic, payload, retain):
if self.on_message:
self.on_message(self, None, SimpleNamespace(topic=topic, payload=payload, qos=0, retain=retain))Mantén el fake diminuto y pruébalo una vez contra un broker real (el mismo escenario por ambos caminos); si no, acabarás probando el fake. Puedes comprobar lo que esperas de los wildcards con el topic matcher.
Node.js: aedes como broker en memoria
Aedes es un broker MQTT real que puedes arrancar dentro de un proceso de Jest o Mocha. Habla el protocolo de verdad sobre un puerto TCP local, así que tu código MQTT.js se ejecuta sin cambios y sin ningún mock:
import { createServer } from 'node:net';
import { Aedes } from 'aedes';
import mqtt from 'mqtt';
let broker, server, url;
beforeAll(async () => {
broker = await Aedes.createBroker(); // aedes 0.x: require('aedes')()
server = createServer(broker.handle);
await new Promise((resolve) => server.listen(0, resolve)); // random free port
url = `mqtt://127.0.0.1:${server.address().port}`;
});
afterAll(async () => {
await new Promise((resolve) => server.close(resolve));
await new Promise((resolve) => broker.close(resolve));
});
test('retained state is delivered to late subscribers', async () => {
const pub = await mqtt.connectAsync(url);
await pub.publishAsync('testmqtt/unit/state', 'ready', { qos: 1, retain: true });
const sub = await mqtt.connectAsync(url);
const received = new Promise((resolve) => sub.on('message', (t, p, pkt) => resolve([t, p.toString(), pkt.retain])));
await sub.subscribeAsync('testmqtt/unit/#', { qos: 1 });
expect(await received).toEqual(['testmqtt/unit/state', 'ready', true]);
await Promise.all([pub.endAsync(), sub.endAsync()]);
});Escuchar en el puerto 0 da a cada archivo de pruebas su propio puerto libre, de modo que los workers paralelos de Jest no chocan. Python no tiene un broker embebido igual de extendido; la mayoría de equipos usa un contenedor de Mosquitto: consulta Testcontainers para MQTT y fixtures de pytest.
Dónde terminan las pruebas unitarias
Los mocks son excelentes para verificar tus decisiones. No pueden verificar las del broker. Pasa una prueba a la capa de integración cuando dependa de:
- QoS y sesiones: reentrega tras una desconexión, sesiones persistentes, gestión de duplicados.
- Mensajes retenidos y Last Will: son funciones del broker; un fake solo reproduce tus suposiciones.
- Autenticación y ACL: un mock nunca devuelve
Not authorized; un broker real sí, a veces en silencio. - Límites de payload, topic aliases, propiedades de MQTT 5: los límites los fija la configuración del broker.
- TLS, WebSockets, tiempos de reconexión: el comportamiento de red necesita una red.
Una pirámide sana tiene muchas pruebas rápidas con mocks, unas pocas decenas de pruebas de integración contra un broker real (consulta pruebas de integración MQTT) y un puñado de comprobaciones end-to-end contra el entorno en el que despliegas.
Contrastar el mock con la realidad
Antes de confiar en una prueba con mocks, confirma una vez que sus suposiciones coinciden con un broker real. Publica el mismo mensaje desde el cliente MQTT online en broker.connectmq.com y compara: el tipo del payload, el flag retain para suscriptores tardíos y los strings exactos de los topics. Si difieren de lo que devuelve tu mock, corrige primero el mock.
Compara tu mock con un broker real
Haz mock del cliente para probar tu código rápido, usa un fake pequeño o un broker embebido cuando interactúan componentes y deja las garantías del protocolo a las pruebas de integración contra un broker real. Esa división mantiene la suite rápida sin que las pruebas mientan sobre lo que MQTT hace en realidad.
Preguntas frecuentes
¿Debo hacer mock del cliente MQTT o usar un broker real en las pruebas?
Ambas cosas, en capas distintas. Haz mock del cliente para probar tu propia lógica (qué topic, qué payload, cómo reaccionas a un mensaje) en milisegundos. Usa un broker real en las pruebas de integración para verificar el comportamiento del protocolo: QoS, mensajes retenidos, ACL y reconexiones.
¿Aedes es un mock o un broker MQTT real?
Aedes es un broker MQTT real escrito en Node.js. Embebido en el proceso de pruebas con persistencia en memoria se comporta como un broker rápido y desechable, así que las pruebas que lo usan son pequeñas pruebas de integración, no pruebas unitarias puras.
¿Cómo simulo un mensaje MQTT entrante en una prueba unitaria?
Llama directamente al manejador de mensajes con un objeto de mensaje falso que tenga los campos que lee tu código (topic, payload, qos, retain). Con paho-mqtt eso significa llamar a on_message(client, userdata, msg); con MQTT.js, emitir un evento 'message' en un cliente EventEmitter simulado.
