Grabar y reproducir tráfico MQTT para pruebas

Grabar y reproducir tráfico MQTT significa capturar mensajes reales de un broker en un archivo — topic, payload, QoS, flag retain y el instante en que llegó cada uno — para publicarlos después con la misma temporización. Así obtienes tests de regresión construidos con lo que tus dispositivos envían de verdad, no con lo que alguien supuso que envían.

Esta guía construye un pequeño grabador y un reproductor con temporización precisa en Python (paho-mqtt 2.x), y después muestra cómo convertir una grabación en un test de regresión automatizado.

Por qué la reproducción supera a los datos sintéticos

Los datos de prueba generados son limpios. Las flotas reales no: distintas versiones de firmware envían JSON ligeramente distinto, los sensores sueltan diez lecturas en 50 ms tras una reconexión, un gateway vuelve a publicar el mismo estado retenido cada minuto. Los bugs viven en esos detalles.

Los datos sintéticos siguen siendo útiles — usa un generador de datos de prueba para casos límite que nadie ha grabado aún — pero una grabación es lo más parecido a producción que puedes ejecutar en tu portátil.

Elige qué grabar

Un grabador es un cliente MQTT normal con una suscripción. Elige un filtro de topic que cubra el flujo que quieres probar, no # sobre todo el broker: las capturas pequeñas son más fáciles de revisar y filtran menos datos. Comprueba el filtro antes de empezar con el topic matcher.

Decide de antemano:

• Filtro de topic — por ejemplo testmqtt/replay/site1/# para la telemetría de un sitio.
• QoS de la suscripción — suscríbete con QoS 2. El QoS de entrega es el menor entre el QoS de publicación y el de suscripción, así que una suscripción QoS 2 conserva el nivel original.
• Codificación del payload — guarda los payloads en base64 para que los formatos binarios (Protobuf, CBOR) sobrevivan al viaje de ida y vuelta.
• Privacidad — los payloads reales pueden contener ubicaciones, nombres o tokens. Límpialos antes de subir una captura a un repositorio.

Un grabador en 30 líneas

El grabador escribe un objeto JSON por línea (JSONL). El campo t son los segundos desde que arrancó el grabador, medidos con un reloj monótono para que los ajustes del reloj del sistema no distorsionen la temporización.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = sys.argv[1] if len(sys.argv) > 1 else "testmqtt/replay/#"
out = open(sys.argv[2] if len(sys.argv) > 2 else "capture.jsonl", "w", buffering=1)
start = time.monotonic()

def on_connect(client, userdata, flags, reason_code, properties):
    print("conectado:", reason_code)
    client.subscribe(FILTER, qos=2)        # se vuelve a suscribir tras cada reconexión

def on_message(client, userdata, msg):
    out.write(json.dumps({
        "t": round(time.monotonic() - start, 6),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": msg.retain,
        "payload": base64.b64encode(msg.payload).decode(),
    }) + "\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-recorder-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, PORT, keepalive=30)
try:
    client.loop_forever()
except KeyboardInterrupt:
    client.disconnect()
    out.close()

Ejecútalo con python record.py 'testmqtt/replay/#' capture.jsonl, genera algo de tráfico y pulsa Ctrl+C. El buffering por líneas (buffering=1) hace que el archivo sea utilizable aunque el proceso muera.

Mensajes retenidos en una captura

Cuando el grabador se suscribe, el broker le envía de inmediato todos los mensajes retenidos que coinciden, con retain: true. Son una foto del estado en el momento de la grabación, no tráfico en vivo, y llegan todos en t ≈ 0. Los mensajes reenviados en vivo llegan con retain: false, aunque el publicador activara el flag (los clientes MQTT 5 pueden cambiarlo con la opción de suscripción Retain As Published). El reproductor de abajo omite esa foto por defecto; publícala aparte si tu test necesita un estado inicial.

Un reproductor que respeta la temporización

El enfoque ingenuo — sleep(gap) entre mensajes — se desvía, porque cada llamada a publish también consume tiempo. En su lugar, calcula el momento previsto de cada mensaje desde el inicio de la reproducción y duerme solo hasta entonces. Un factor de velocidad te permite comprimir una hora de tráfico en minutos.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

FILE = sys.argv[1]
SPEED = float(sys.argv[2]) if len(sys.argv) > 2 else 1.0     # 2.0 = el doble de rápido
SRC, DST = "testmqtt/replay/", "testmqtt/replay-run/"          # reescritura de topic

records = [json.loads(line) for line in open(FILE) if line.strip()]
records = [r for r in records if not r["retain"]]              # omitir la foto retenida
if not records:
    sys.exit("nada que reproducir")
offset = records[0]["t"]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-replayer-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.connect("broker.connectmq.com", 1883)
client.loop_start()

t0 = time.monotonic()
late = 0.0
for r in records:
    due = t0 + (r["t"] - offset) / SPEED
    delay = due - time.monotonic()
    if delay > 0:
        time.sleep(delay)
    else:
        late = max(late, -delay)
    topic = r["topic"].replace(SRC, DST, 1)
    info = client.publish(topic, base64.b64decode(r["payload"]), qos=r["qos"])

info.wait_for_publish(timeout=30)
print(f"reproducidos {len(records)} mensajes, peor retraso {late * 1000:.1f} ms")
client.loop_stop()
client.disconnect()

La reescritura de topic es la red de seguridad: la grabación vino de testmqtt/replay/... y la reproducción va a testmqtt/replay-run/..., así que nada que escuche los topics originales reacciona. Apunta el sistema bajo prueba al prefijo reescrito.

La línea de peor retraso te dice si la reproducción fue fiel. Si crece con factores de velocidad altos, el cuello de botella es el propio reproductor: la ventana inflight del cliente está llena de mensajes QoS 1/2 sin confirmar. Consulta el ajuste de rendimiento del cliente para ampliarla.

Convierte una reproducción en un test de regresión

Una reproducción se convierte en test cuando comparas lo que produjo el sistema bajo prueba con un resultado conocido como correcto. El patrón:

1. Graba una vez las entradas (telemetría de dispositivos) y sube la captura, ya limpia, como fixture.
2. Ejecuta el grabador sobre los topics de salida del sistema: alertas, agregados, comandos.
3. Reproduce las entradas.
4. Compara la captura de salida con un archivo golden grabado con una build de confianza.

La temporización nunca es idéntica entre ejecuciones, así que compara contenido, no marcas de tiempo. Agrupa por topic y compara la secuencia de payloads de cada topic; MQTT solo garantiza el orden por topic y por publicador, y una comparación de orden global produce tests inestables.

import base64, json, sys
from collections import defaultdict

def load(path):
    per_topic = defaultdict(list)
    for line in open(path):
        if line.strip():
            r = json.loads(line)
            per_topic[r["topic"]].append(base64.b64decode(r["payload"]))
    return per_topic

golden, actual = load(sys.argv[1]), load(sys.argv[2])
ok = True
for topic in sorted(set(golden) | set(actual)):
    if golden[topic] != actual[topic]:
        ok = False
        print(f"DIFF {topic}: esperados {len(golden[topic])} msgs, recibidos {len(actual[topic])}")
print("PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)

Si los payloads contienen marcas de tiempo o IDs generados, normalízalos antes de comparar (parsea el JSON y elimina esos campos). Integra los tres scripts en un fixture de pytest o en un job de CI: la guía de pruebas de integración y pruebas MQTT en GitHub Actions muestran cómo levantar un broker en cada ejecución.

Trampas a vigilar

• Reproducir comandos. Una captura de topics .../cmd/... encenderá relés reales si llega a un dispositivo real. Excluye los topics de comandos de las capturas de entrada salvo que el test trate de ellos.
• Mensajes duplicados. QoS 1 puede entregar un mensaje dos veces tras una reconexión, y la grabación conserva el duplicado. Es realista — tus consumidores deberían soportarlo — pero tenlo en cuenta antes de contar mensajes.
• Metadatos de MQTT 5 ausentes. Los scripts de arriba guardan topic, payload, QoS y retain. Si tu sistema depende de user properties, content type o response topics, añade campos de msg.properties al registro y pasa un objeto Properties al publicar.
• Consumidores dependientes del reloj. La lógica que usa la hora real (por ejemplo «ninguna lectura en 60 s») se comporta distinto a 10×. Reproduce esos tests a 1×.
• Límites del broker. Reproducciones rápidas de capturas grandes pueden chocar con límites de tasa de mensajes o de colas en brokers compartidos. Usa tu propia instancia para todo lo que vaya más allá de una comprobación rápida.

Comprueba la reproducción a simple vista

Antes de automatizar nada, observa llegar una reproducción. Abre el cliente MQTT online, conéctate a broker.connectmq.com por WebSockets seguros y suscríbete al prefijo reescrito; luego lanza replay.py. Deberías ver los mensajes llegar en ráfagas que siguen el patrón original.

Mira la reproducción en tu navegador

Un grabador, un reproductor y un script de comparación suman menos de cien líneas de Python y convierten tráfico real en tests repetibles. Graba una vez en campo, limpia la captura, guárdala junto a tu código, y cada cambio futuro se probará contra lo que tus dispositivos envían de verdad.

Preguntas frecuentes