Pruebas de clientes MQTT: unitarias, integración y E2E

Q: ¿Debo simular la biblioteca MQTT en los tests unitarios?

Simula en tu propio límite, como una función publish que inyectas, en lugar de simular las interioridades de la biblioteca. Verifica el comportamiento real del protocolo con un test de integración contra un broker en contenedor.

Las pruebas de clientes MQTT son incómodas porque el comportamiento interesante ocurre a través de una red, dentro de callbacks y en el hilo de otro. Esta guía muestra un enfoque por capas que funciona: tests unitarios rápidos con un cliente simulado, tests de integración contra un broker desechable en contenedor en CI y un breve smoke test end-to-end contra un broker real en la nube antes de cada versión.

Por qué es difícil probar el código de clientes MQTT

El código MQTT mezcla tres aspectos que fallan de maneras distintas:

Lógica de negocio: analizar payloads, decidir qué publicar, asignar topics a dispositivos.
Comportamiento del protocolo: QoS, retained messages, reconexiones, resuscripciones, last will.
Entorno: certificados TLS, credenciales, puertos, firewalls, cuotas del broker.

La mayoría de las suites de tests MQTT inestables intentan cubrir los tres en un único test que se conecta a un broker y espera con sleep. Separarlos permite que cada capa use la herramienta más barata capaz de detectar sus errores.

Capa	Qué detecta	Broker necesario	Velocidad	Cuándo se ejecuta
Unitaria	Análisis, asignación de topics, lógica de decisión	No (simulado)	Milisegundos	En cada guardado
Integración	Conexión de subscribe/publish, QoS, retenidos, reconexión	Contenedor local	Segundos	En cada pull request
Smoke test end-to-end	TLS, credenciales, puertos, red real	Broker real en la nube	Segundos	Antes de cada versión y de forma programada

Los ejemplos usan Python con paho-mqtt 2.x y pytest, pero la estructura se traslada a cualquier lenguaje.

Paso 1: diseñar código que se pueda probar

El cambio más útil con diferencia es mantener el cliente MQTT en el borde. Pon las decisiones en funciones o clases normales que reciban una función de publicación como dependencia, y deja que un adaptador fino las conecte a paho.

thermostat.py:

import json
from dataclasses import dataclass
from typing import Callable

Publish = Callable[[str, str, int, bool], None]


@dataclass
class Thermostat:
    prefix: str
    publish: Publish
    setpoint: float = 21.0
    hysteresis: float = 0.5
    heating: bool = False

    def on_temperature(self, topic: str, payload: bytes) -> None:
        try:
            value = float(json.loads(payload)["value"])
        except (ValueError, KeyError, TypeError, json.JSONDecodeError):
            return  # ignorar mensajes mal formados

        room = topic.split("/")[-2]
        if not self.heating and value < self.setpoint - self.hysteresis:
            self.heating = True
        elif self.heating and value > self.setpoint + self.hysteresis:
            self.heating = False
        else:
            return

        self.publish(f"{self.prefix}/{room}/heater/set",
                     "ON" if self.heating else "OFF", 1, True)

adapter.py:

import threading

import paho.mqtt.client as mqtt
from thermostat import Thermostat


def build_client(host, port, client_id, prefix, username=None, password=None, tls=False):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if username:
        client.username_pw_set(username, password)
    if tls:
        client.tls_set()

    thermostat = Thermostat(
        prefix=prefix,
        publish=lambda t, p, q, r: client.publish(t, p, qos=q, retain=r),
    )

    def on_connect(client, userdata, flags, reason_code, properties):
        if not reason_code.is_failure:
            client.subscribe(f"{prefix}/+/temperature", qos=1)

    ready = threading.Event()

    def on_subscribe(client, userdata, mid, reason_codes, properties):
        ready.set()

    def on_message(client, userdata, msg):
        thermostat.on_temperature(msg.topic, msg.payload)

    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    return client, thermostat, ready

Suscribirse dentro de on_connect es deliberado: restaura las suscripciones tras cada reconexión, y el evento ready devuelto permite a los tests esperar hasta que la suscripción esté activa en lugar de adivinar con sleep(). Ese es justo el tipo de conexión entre piezas que verificarán los tests de integración.

Paso 2: tests unitarios con un cliente simulado

Como Thermostat recibe publish como parámetro, un Mock sustituye por completo a la red.

test_thermostat.py:

import json
from unittest.mock import Mock

from thermostat import Thermostat

PREFIX = "home/test"


def make():
    pub = Mock()
    return Thermostat(prefix=PREFIX, publish=pub), pub


def msg(value):
    return json.dumps({"value": value}).encode()


def test_turns_heater_on_below_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    pub.assert_called_once_with(f"{PREFIX}/kitchen/heater/set", "ON", 1, True)


def test_no_publish_inside_hysteresis_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.2))
    pub.assert_not_called()


def test_turns_heater_off_above_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.6))
    assert pub.call_args.args[1] == "OFF"


def test_ignores_garbage():
    t, pub = make()
    for bad in [b"", b"not json", b'{"temp": 3}', b'{"value": "hot"}']:
        t.on_temperature(f"{PREFIX}/kitchen/temperature", bad)
    pub.assert_not_called()

Se ejecutan en milisegundos y cubren la lógica que con más frecuencia tiene errores. Resiste la tentación de simular paho en detalle (comprobar que se llamó a client.subscribe con argumentos exactos): esos tests reflejan la implementación, se rompen con cada refactorización y aun así no demuestran que el broker acepte lo que envías. Ese es el trabajo de la capa de integración.

Paso 3: tests de integración contra un broker en contenedor

Un broker real en un contenedor ofrece un comportamiento de protocolo fiel, sin estado compartido y sin depender de internet. La imagen oficial de Eclipse Mosquitto incluye un archivo de configuración que permite conexiones anónimas, ideal para pruebas:

docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf

Ahora escribe tests que ejerciten el adaptador real. Una función auxiliar recoge los mensajes en un segundo cliente y espera a que se cumpla una condición en lugar de dormir un tiempo fijo.

test_integration.py:

import os
import queue
import uuid

import paho.mqtt.client as mqtt
import pytest

from adapter import build_client

HOST = os.getenv("MQTT_TEST_HOST", "localhost")
PORT = int(os.getenv("MQTT_TEST_PORT", "1883"))


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


@pytest.fixture
def probe(prefix):
    received = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: received.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/#", qos=1)
    c.loop_start()
    yield c, received
    c.loop_stop()
    c.disconnect()


def wait_for(q, predicate, timeout=5.0):
    while True:
        topic, payload, retain = q.get(timeout=timeout)  # lanza queue.Empty al agotar el tiempo
        if predicate(topic, payload):
            return topic, payload, retain


def test_temperature_drives_heater(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/office/temperature", '{"value": 18.5}', qos=1)
        topic, payload, _ = wait_for(received, lambda t, p: t.endswith("/heater/set"))
        assert topic == f"{prefix}/office/heater/set"
        assert payload == "ON"
    finally:
        client.loop_stop()
        client.disconnect()


def test_heater_command_is_retained(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/hall/temperature", '{"value": 15}', qos=1)
        wait_for(received, lambda t, p: t.endswith("/heater/set"))
    finally:
        client.loop_stop()
        client.disconnect()

    # Un suscriptor tardío debe recibir el comando retenido de inmediato
    late = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"late-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: late.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/hall/heater/set", qos=1)
    c.loop_start()
    try:
        _, payload, retain = late.get(timeout=5)
        assert payload == "ON" and retain is True
    finally:
        c.loop_stop()
        c.disconnect()

Tres hábitos mantienen estables los tests de integración:

Un prefijo de topic único por test. Los retained messages de un test no pueden filtrarse al siguiente.
Espera a condiciones, no uses sleep(). Una cola con timeout falla rápido y nunca espera más de lo necesario.
Detén siempre los bucles y desconecta en finally. Los hilos de red huérfanos hacen que los tests posteriores se queden colgados.

Para probar la lógica de reconexión, detén y arranca el contenedor a mitad del test (docker restart test-broker) y comprueba que el cliente vuelve a suscribirse y recibe un mensaje publicado tras el reinicio.

Ejecutar el broker en CI

GitHub Actions puede arrancar el contenedor como un paso previo a los tests:

name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Start MQTT broker
        run: |
          docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
            mosquitto -c /mosquitto-no-auth.conf
      - run: pip install "paho-mqtt>=2,<3" pytest
      - run: pytest -q

Los tests unitarios y de integración pueden compartir una misma ejecución de pytest; marca los lentos con @pytest.mark.integration si quieres omitirlos en local.

Paso 4: smoke tests end-to-end contra un broker real

Un contenedor no puede decirte si tu configuración TLS, tu paquete de CA, tus reglas de firewall o tus credenciales de producción funcionan. Un breve smoke test contra el endpoint real sí. Para proyectos de código abierto o comprobaciones rápidas, basta con el servidor MQTT público gratuito en broker.connectmq.com; acepta clientes anónimos por TLS en el puerto 8883. Usa payloads sintéticos, porque el área pública es legible por todo el mundo.

smoke_test.py:

import os
import sys
import threading
import uuid

import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
USER = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"smoke-{uuid.uuid4().hex[:8]}"
TOPIC = f"{os.getenv('MQTT_PREFIX', 'connectmq/demo')}/smoke-{uuid.uuid4().hex[:8]}/ping"

got = threading.Event()
nonce = uuid.uuid4().hex
errors = []


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        errors.append(f"connect failed: {reason_code}")
        got.set()  # despertar al hilo principal; los callbacks no pueden terminar el proceso
        return
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    client.publish(TOPIC, nonce, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == nonce:
        got.set()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USER:
    client.username_pw_set(USER, PASSWORD)
client.tls_set()
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 8883, keepalive=30)
client.loop_start()
ok = got.wait(timeout=15) and not errors
for e in errors:
    print(e)
client.loop_stop()
client.disconnect()
print("smoke test passed" if ok else "smoke test FAILED: no echo within 15 s")
sys.exit(0 if ok else 1)

En producción, apunta el mismo script a tu broker privado definiendo MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID y MQTT_PREFIX desde los secretos de CI. En un broker MQTT gestionado como ConnectMQ, los client ID están vinculados a las credenciales emitidas, así que crea un conjunto de credenciales dedicado para CI en lugar de tomar prestada la identidad de un dispositivo; de lo contrario, el smoke test desconectará ese dispositivo.

Ejecuta los smoke tests antes de cada versión y de forma programada (por ejemplo, cada hora), no en cada commit. En los brokers compartidos se aplican cuotas: el servidor público permite 10 mensajes por segundo por cliente y 10 suscripciones por cliente, y el plan gratuito Serverless de ConnectMQ limita las conexiones simultáneas a 25.

Antipatrones comunes

Algunos hábitos hacen que las suites de tests MQTT sean lentas e inestables. Vigílalos en las revisiones de código:

Esperas fijas. Un time.sleep(2) después de publicar es o demasiado largo, y desperdicia minutos en toda la suite, o demasiado corto en un runner de CI cargado. Espera en su lugar a una cola o un evento con timeout.
Topics compartidos entre tests. Los retained messages y las suscripciones residuales de un test se filtran al siguiente. Genera un prefijo único por test.
Errores de callback silenciados. Las excepciones dentro de los callbacks MQTT se ejecutan en el hilo de red, así que no hacen fallar el test directamente. Captúralas en una lista y comprueba que está vacía en el teardown.
Credenciales de dispositivos en CI. Tomar prestada la identidad de un dispositivo real para los tests desconecta ese dispositivo. Crea credenciales separadas para la automatización.
Probar la biblioteca en lugar de tu código. No necesitas demostrar que paho entrega mensajes QoS 1; necesitas demostrar que tu código se suscribe a los topics correctos y reacciona bien.
Un único test end-to-end gigante. Cuando falla, nadie sabe qué capa se ha roto. Mantén pocos tests end-to-end y acotados, y lleva el detalle a los tests unitarios.

Lista de comprobación de pruebas

La lógica de negocio vive fuera de los callbacks MQTT y se prueba con tests unitarios con un publish simulado.
Los payloads mal formados, vacíos y demasiado grandes se prueban y se ignoran de forma segura.
Los tests de integración usan un contenedor de broker nuevo y un prefijo de topic único por test.
El comportamiento de reconexión y resuscripción se prueba reiniciando el broker.
El comportamiento de los retained messages y del last will se verifica con un suscriptor tardío.
Antes de cada versión se ejecuta un smoke test TLS con credenciales reales.
Las credenciales de CI están separadas de las de los dispositivos.

Para detalles específicos de cada lenguaje, consulta el tutorial de cliente MQTT en PHP, el tutorial del cliente Paho MQTT C y la guía de bibliotecas cliente MQTT, más general.

Empieza con ConnectMQ

Dale a tu pipeline de pruebas un broker real con el que hablar. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, añade un conjunto de credenciales dedicado para CI y ejecuta contra él el smoke test en cada versión. El explorador de topics en vivo y el cliente en el navegador del portal facilitan ver exactamente qué han publicado tus tests. ConnectMQ gestiona el broker; tú te quedas con los tests.

Preguntas frecuentes

¿Debo simular la biblioteca MQTT en los tests unitarios?

Simula en tu propio límite, como una función publish que inyectas, en lugar de simular las interioridades de la biblioteca. Verifica el comportamiento real del protocolo con un test de integración contra un broker en contenedor.

¿Qué broker debo usar para los tests de integración MQTT en CI?

Un contenedor desechable de Eclipse Mosquitto es la opción más sencilla: arranca en un segundo aproximadamente, no necesita credenciales con /mosquitto-no-auth.conf y no deja estado entre ejecuciones.

¿Puedo usar broker.connectmq.com en tests automatizados?

Sí, para smoke tests ocasionales con datos sintéticos y topics únicos, dentro de sus límites de 10 mensajes por segundo y 10 suscripciones por cliente. Para suites de tests intensivas, usa un contenedor local o tu propio broker privado.

¿Cómo pruebo la lógica de reconexión MQTT?

Reinicia el contenedor del broker durante el test y comprueba que el cliente reconecta, vuelve a suscribirse y recibe un mensaje publicado tras el reinicio.

Pruebas de clientes MQTT: tests unitarios, de integración y end-to-end