Pruebas de integración MQTT: prueba tu app contra un broker real

Las pruebas de integración MQTT ejecutan tu aplicación contra un broker real: tu código se conecta, se suscribe y publica, y el test comprueba los mensajes que realmente cruzan la red. Detectan lo que las pruebas unitarias con mocks no pueden: filtros de topic incorrectos, acuses de QoS que faltan, estado retained, denegaciones de ACL y fallos de reconexión.

Esta guía muestra patrones que funcionan con pytest (paho-mqtt 2.x) y Jest (MQTT.js 5), y las cinco reglas que mantienen estos tests rápidos y sin inestabilidad.

¿Unitarias, de integración o end-to-end?

Mantén los tests con mocks para la lógica pura; brokers mock para pruebas unitarias cubre esa capa. Todo lo que depende de cómo se comporta un broker (wildcards, retained messages, reentrega con QoS 1, expiración de sesión) pertenece a las pruebas de integración contra uno real.

Elegir el broker de pruebas

Tienes tres opciones prácticas:

• Mosquitto local (instalado o vía Docker): rápido y privado, pero cada desarrollador y cada runner de CI tiene que arrancarlo.
• Contenedor efímero por ejecución: totalmente aislado y desechable; consulta tests MQTT con Testcontainers.
• Broker alojado: sin configuración, con TLS y autenticación reales, y el mismo endpoint desde portátiles, CI y dispositivos de staging.

A tu código de pruebas no debería importarle con cuál habla. Lee el host, el puerto y las credenciales de variables de entorno y usa por defecto algo que funcione de entrada:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
# MQTT_PORT=8883 for TLS; set MQTT_USER / MQTT_PASS for a private broker

Las cinco reglas para tests MQTT estables

1. Prefijo de topic único por test. Construye cada topic a partir de testmqtt/it/<run-id>/<random-id>. Así los workers paralelos, los jobs de CI y los desarrolladores nunca ven los mensajes de los demás.
2. Client ID único por conexión. Dos conexiones con el mismo ID se desconectan entre sí: una fuente clásica de tests que pierden mensajes "al azar".
3. Espera eventos, nunca duermas. Espera el CONNACK antes de suscribirte, el SUBACK antes de publicar y el mensaje con un timeout. Un sleep(1) es o demasiado largo (suite lenta) o demasiado corto (suite inestable).
4. Limpia lo que sobrevive al test. Los retained messages y las sesiones persistentes sobreviven a las desconexiones. Borra los topics retained con una publicación retained vacía y usa sesiones limpias salvo que estés probando sesiones.
5. Comprueba el contenido, no los tiempos ni el orden global. MQTT solo garantiza el orden por topic y QoS desde un mismo publicador. Correlaciona los mensajes con un ID en el payload.

pytest: un fixture reutilizable de cliente

El fixture siguiente da a cada test un helper conectado que espera los acuses y recoge los mensajes en una cola thread-safe. Ponlo en conftest.py:

import os, queue, threading, time, uuid
import pytest
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN_ID = os.getenv("GITHUB_RUN_ID", uuid.uuid4().hex[:8])


class MqttProbe:
    def __init__(self):
        self.messages = queue.Queue()
        self.retained = set()
        self._connected = threading.Event()
        self._subacks = {}
        self._cond = threading.Condition()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"it-{uuid.uuid4().hex[:10]}")
        self.client.on_connect = lambda c, u, f, rc, p: self._connected.set()
        self.client.on_subscribe = self._on_subscribe
        self.client.on_message = lambda c, u, msg: self.messages.put(msg)
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        if PORT == 8883:
            self.client.tls_set()

    def _on_subscribe(self, client, userdata, mid, reason_codes, properties):
        with self._cond:
            self._subacks[mid] = reason_codes
            self._cond.notify_all()

    def connect(self, timeout=5):
        self.client.connect(HOST, PORT, keepalive=30)
        self.client.loop_start()
        assert self._connected.wait(timeout), "no CONNACK from broker"

    def subscribe(self, topic, qos=1, timeout=5):
        _, mid = self.client.subscribe(topic, qos)
        with self._cond:
            assert self._cond.wait_for(lambda: mid in self._subacks, timeout), f"no SUBACK for {topic}"
            code = self._subacks.pop(mid)[0]
        assert not code.is_failure, f"subscribe to {topic} refused: {code}"

    def publish(self, topic, payload, qos=1, retain=False, timeout=5):
        info = self.client.publish(topic, payload, qos=qos, retain=retain)
        info.wait_for_publish(timeout)
        assert info.is_published(), f"publish to {topic} not acknowledged"
        if retain:
            self.retained.add(topic)

    def expect(self, topic, timeout=5):
        deadline = time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                msg = self.messages.get(timeout=left)
            except queue.Empty:
                break
            if msg.topic == topic:
                return msg
        raise AssertionError(f"no message on {topic} within {timeout}s")

    def close(self):
        for topic in self.retained:  # clear retained state left by the test
            self.client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
        self.client.disconnect()
        self.client.loop_stop()


@pytest.fixture
def prefix():
    return f"testmqtt/it/{RUN_ID}/{uuid.uuid4().hex[:8]}"


@pytest.fixture
def mqtt_probe():
    probe = MqttProbe()
    probe.connect()
    yield probe
    probe.close()

Un test para un servicio de la aplicación —aquí, un termostato que escucha comandos e informa de su estado— se lee entonces como una especificación:

import json
from conftest import HOST, PORT
from myapp.thermostat import ThermostatService


def test_setpoint_command_updates_state(mqtt_probe, prefix):
    service = ThermostatService(host=HOST, port=PORT, prefix=prefix)
    service.start()  # returns once the service has its own SUBACK
    try:
        mqtt_probe.subscribe(f"{prefix}/state")
        mqtt_probe.publish(f"{prefix}/cmd/setpoint", json.dumps({"value": 21.5}))

        msg = mqtt_probe.expect(f"{prefix}/state")
        assert json.loads(msg.payload)["setpoint"] == 21.5
    finally:
        service.stop()

El prefix se pasa a la aplicación, y esa es la decisión de diseño más útil para la testabilidad: si tu servicio construye los topics a partir de una raíz configurable, cada test puede ejecutarse en su propio espacio de nombres. Fíjate en que service.start() bloquea hasta que el servicio está suscrito; si tu aplicación no expone una señal de disponibilidad así, añádela (un evento, un callback o un mensaje de estado online retained). De lo contrario, el comando del test puede llegar antes de que el servicio escuche. Fixtures de pytest para MQTT muestra más variantes de fixtures, incluidos brokers con scope de sesión.

Jest: helpers con promesas en MQTT.js 5

MQTT.js 5 incluye connectAsync, subscribeAsync, publishAsync y endAsync, que hacen los tests lineales. El único helper que necesitas es "esperar el siguiente mensaje en un topic":

const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

const URL = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const opts = () => ({
  clientId: `it-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  reconnectPeriod: 0,
});

function nextMessage(client, topic, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const onMessage = (t, payload, packet) => {
      if (t !== topic) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ payload: payload.toString(), retain: packet.retain });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on ${topic} within ${timeoutMs} ms`));
    }, timeoutMs);
    client.on('message', onMessage);
  });
}

let publisher;
let prefix;

beforeAll(async () => {
  publisher = await mqtt.connectAsync(URL, opts());
});
afterAll(() => publisher.endAsync());
beforeEach(() => {
  prefix = `testmqtt/it/${process.env.GITHUB_RUN_ID ?? 'local'}/${randomUUID().slice(0, 8)}`;
});

test('late subscribers receive the retained device config', async () => {
  const topic = `${prefix}/config`;
  await publisher.publishAsync(topic, JSON.stringify({ interval: 30 }), { qos: 1, retain: true });

  const device = await mqtt.connectAsync(URL, opts());
  try {
    const received = nextMessage(device, topic); // listen before subscribing
    await device.subscribeAsync(topic, { qos: 1 });
    const msg = await received;
    expect(msg.retain).toBe(true);
    expect(JSON.parse(msg.payload)).toEqual({ interval: 30 });
  } finally {
    await device.endAsync();
    await publisher.publishAsync(topic, '', { qos: 1, retain: true }); // clear retained
  }
}, 15000);

Dos detalles evitan los problemas más comunes con Jest. El listener se registra antes de subscribeAsync, porque el broker envía los retained messages justo después del SUBACK, a veces antes de que se resuelva la promesa esperada. Y cada cliente se cierra con endAsync(), para que Jest no se quede colgado con "Jest did not exit one second after the test run has completed". Tienes más patrones de Jest en probar MQTT con Jest.

Prueba también los caminos de error

El camino feliz es la parte menos interesante de un cliente MQTT. Las pruebas de integración son donde puedes demostrar a bajo coste el comportamiento que importa en el campo:

• Payloads mal formados. Publica JSON inválido en tu topic de comandos y comprueba que el servicio registra un error y sigue funcionando, en lugar de tumbar su bucle de red.
• Autorización. Conéctate con un usuario que no tiene permiso para publicar en un topic y comprueba que tu código informa del rechazo. Con MQTT 5, una publicación QoS 1 devuelve un reason code en el PUBACK; con MQTT 3.1.1 el broker normalmente descarta el mensaje o cierra la conexión.
• Reconexiones. Desconecta el cliente de la aplicación (o reinicia un broker local) y comprueba que vuelve a suscribirse y procesa el siguiente comando. Pruebas de reconexión MQTT muestra cómo automatizarlo.

Cada uno de estos casos ocupa unas pocas líneas con los fixtures anteriores y detecta fallos que, de otro modo, solo aparecen en los logs de producción.

Checklist de tests inestables

Si se rechaza una suscripción, el reason code del SUBACK te dice por qué: 0x87 Not authorized suele significar que la ACL del usuario de pruebas no cubre el prefijo. Decodifica cualquier código con la búsqueda de reason codes MQTT.

Probar los tests

Antes de confiar en una suite nueva, obsérvala desde fuera. Suscríbete a testmqtt/it/# en el cliente MQTT online y ejecuta los tests: deberías ver aparecer el prefijo de cada test, fluir sus mensajes y, al final, cómo se borran los topics retained (un mensaje retained vacío). Después ejecuta la suite dos veces en paralelo —por ejemplo pytest -n 4 con pytest-xdist, o dos jobs de CI— y confirma que nada cambia.

Ver el tráfico de las pruebas de integración

Las pruebas de integración contra un broker real son la forma más barata de demostrar que tu código MQTT se comporta como lo hará en producción. Aísla cada test por topic y client ID, espera los acuses en lugar de dormir, limpia el estado retained y la suite seguirá en verde por las razones correctas. Cuando quieras ejecutarla en cada push, continúa con probar MQTT en CI con GitHub Actions.

Preguntas frecuentes