Enregistrer et rejouer du trafic MQTT pour vos tests

Enregistrer et rejouer du trafic MQTT, c'est capturer de vrais messages d'un broker dans un fichier — topic, payload, QoS, flag retain et l'instant d'arrivée de chacun — puis republier ce fichier plus tard avec le même timing. Vous obtenez des tests de régression construits à partir de ce que vos appareils envoient réellement, et non de ce que quelqu'un a supposé.

Ce guide construit un petit enregistreur et un rejoueur au timing précis en Python (paho-mqtt 2.x), puis montre comment transformer un enregistrement en test de régression automatisé.

Pourquoi le rejeu bat les données synthétiques

Les données de test générées sont propres. Les flottes réelles, non : différentes versions de firmware envoient un JSON légèrement différent, des capteurs envoient dix mesures en 50 ms après une reconnexion, une passerelle republie chaque minute le même statut retenu. Les bugs se cachent dans ces détails.

Les données synthétiques restent utiles — utilisez un générateur de données de test pour les cas limites que personne n'a encore enregistrés — mais un enregistrement est ce qui se rapproche le plus de la production sur un portable.

Choisir quoi enregistrer

Un enregistreur est un client MQTT ordinaire avec un abonnement. Choisissez un filtre de topic qui couvre le flux à tester, pas # sur tout le broker : des captures plus petites sont plus faciles à relire et exposent moins de données. Vérifiez le filtre au préalable avec le topic matcher.

À décider en amont :

• Filtre de topic — par exemple testmqtt/replay/site1/# pour la télémétrie d'un site.
• QoS de l'abonnement — abonnez-vous en QoS 2. Le QoS de livraison est le minimum entre le QoS de publication et celui de l'abonnement ; un abonnement QoS 2 préserve donc le niveau d'origine.
• Encodage du payload — stockez les payloads en base64 pour que les formats binaires (Protobuf, CBOR) survivent à l'aller-retour.
• Confidentialité — les vrais payloads peuvent contenir des positions, des noms ou des tokens. Nettoyez-les avant de committer une capture dans un dépôt.

Un enregistreur en 30 lignes

L'enregistreur écrit un objet JSON par ligne (JSONL). Le champ t correspond aux secondes écoulées depuis le démarrage, mesurées avec une horloge monotone pour que les ajustements de l'horloge système ne faussent pas le timing.

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("connecté :", reason_code)
    client.subscribe(FILTER, qos=2)        # se réabonne après chaque reconnexion

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()

Lancez-le avec python record.py 'testmqtt/replay/#' capture.jsonl, générez un peu de trafic puis appuyez sur Ctrl+C. Le buffering par ligne (buffering=1) rend le fichier exploitable même si le processus est tué.

Messages retenus dans une capture

Au moment de l'abonnement, le broker envoie immédiatement tous les messages retenus correspondants avec retain: true. Ils représentent un instantané de l'état au moment de l'enregistrement, pas du trafic en direct, et arrivent tous à t ≈ 0. Les messages transmis en direct arrivent avec retain: false, même si le publisher a positionné le flag (les clients MQTT 5 peuvent changer cela avec l'option d'abonnement Retain As Published). Le rejoueur ci-dessous ignore cet instantané par défaut ; publiez-le séparément si votre test a besoin d'un état initial.

Un rejoueur qui respecte le timing

L'approche naïve — sleep(gap) entre les messages — dérive, car chaque appel à publish prend aussi du temps. Calculez plutôt l'heure prévue de chaque message depuis le début du rejeu et dormez seulement jusque-là. Un facteur de vitesse permet de compresser une heure de trafic en quelques minutes.

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 = deux fois plus vite
SRC, DST = "testmqtt/replay/", "testmqtt/replay-run/"          # réécriture de topic

records = [json.loads(line) for line in open(FILE) if line.strip()]
records = [r for r in records if not r["retain"]]              # ignore l'instantané retenu
if not records:
    sys.exit("rien à rejouer")
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"{len(records)} messages rejoués, pire retard {late * 1000:.1f} ms")
client.loop_stop()
client.disconnect()

La réécriture de topic est le filet de sécurité : l'enregistrement provient de testmqtt/replay/..., le rejeu part vers testmqtt/replay-run/..., donc rien de ce qui écoute les topics d'origine ne réagit. Pointez le système testé vers le préfixe réécrit.

La ligne pire retard indique si le rejeu a été fidèle. Si elle augmente avec des facteurs de vitesse élevés, c'est le rejoueur lui-même le goulot d'étranglement : la fenêtre inflight du client est pleine de messages QoS 1/2 non acquittés. Consultez l'optimisation des performances client pour l'agrandir.

Transformer un rejeu en test de régression

Un rejeu devient un test quand vous comparez ce que le système testé a produit avec un résultat connu comme correct. Le schéma :

1. Enregistrez une fois les entrées (télémétrie des appareils) et committez la capture nettoyée comme fixture.
2. Lancez l'enregistreur sur les topics de sortie du système : alertes, agrégats, commandes.
3. Rejouez les entrées.
4. Comparez la capture de sortie avec un fichier golden enregistré à partir d'un build de confiance.

Le timing n'est jamais identique d'une exécution à l'autre : comparez le contenu, pas les horodatages. Regroupez par topic et comparez la séquence de payloads de chaque topic ; MQTT ne garantit l'ordre que par topic et par publisher, et une comparaison d'ordre globale produit des tests instables.

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}: attendu {len(golden[topic])} msgs, reçu {len(actual[topic])}")
print("PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)

Si les payloads contiennent des horodatages ou des ID générés, normalisez-les avant la comparaison (parsez le JSON et supprimez ces champs). Intégrez les trois scripts dans une fixture pytest ou un job de CI — le guide des tests d'intégration et les tests MQTT dans GitHub Actions montrent comment démarrer un broker à chaque exécution.

Pièges à surveiller

• Rejouer des commandes. Une capture de topics .../cmd/... actionnera de vrais relais si elle atteint un vrai appareil. Excluez les topics de commande des captures d'entrée, sauf si le test porte justement sur eux.
• Messages dupliqués. QoS 1 peut livrer un message deux fois après une reconnexion, et l'enregistrement conserve le doublon. C'est réaliste — vos consommateurs doivent le gérer — mais sachez-le avant de compter les messages.
• Métadonnées MQTT 5 manquantes. Les scripts ci-dessus stockent topic, payload, QoS et retain. Si votre système dépend de user properties, du content type ou de response topics, ajoutez les champs de msg.properties à l'enregistrement et passez un objet Properties lors de la publication.
• Consommateurs dépendants de l'horloge. Une logique basée sur l'heure réelle (par exemple « aucune mesure depuis 60 s ») se comporte différemment à 10×. Rejouez ces tests à 1×.
• Limites du broker. Des rejeux rapides de grosses captures peuvent atteindre des limites de débit de messages ou de files d'attente sur des brokers partagés. Utilisez votre propre instance pour tout ce qui dépasse une vérification rapide.

Vérifier le rejeu à l'œil

Avant d'automatiser quoi que ce soit, regardez un rejeu arriver. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com en WebSockets sécurisés et abonnez-vous au préfixe réécrit, puis lancez replay.py. Les messages doivent arriver par rafales qui suivent le profil d'origine.

Suivre le rejeu dans votre navigateur

Un enregistreur, un rejoueur et un script de comparaison tiennent en moins de cent lignes de Python et transforment du trafic réel en tests reproductibles. Enregistrez une fois sur le terrain, nettoyez la capture, gardez-la à côté de votre code, et chaque modification future sera testée contre ce que vos appareils envoient vraiment.

Questions fréquentes