Nommer les topics de test pour que les tests n'entrent jamais en collision

Les tests MQTT qui passent seuls et échouent ensemble partagent presque toujours des topics. Deux jobs de CI publient sur test/temp, un message retenu d'hier répond à l'assertion d'aujourd'hui, ou un subscriber sur # avale le trafic de quelqu'un d'autre. La solution est une convention de nommage : chaque exécution reçoit un préfixe unique comme testmqtt/<suite>/<run-id>/, les tests ne s'abonnent qu'en dessous, et le teardown supprime tout ce qui a été retenu.

Ce guide présente un schéma de nommage qui passe à l'échelle, la construction des IDs d'exécution, des helpers pour Python et JavaScript, et le nettoyage des données de test retenues, y compris les restes d'exécutions plantées.

Pourquoi les tests MQTT entrent en collision

Un broker est un état partagé. Contrairement à un mock en mémoire, il retient et redistribue ; les tests déteignent donc les uns sur les autres de quelques façons prévisibles :

• Exécutions en parallèle : deux pipelines, deux développeurs ou deux workers pytest publient sur le même topic et reçoivent les messages de l'autre.
• Restes retenus : un message retenu survit au test qui l'a créé et est livré immédiatement au subscriber suivant, si bien qu'une fonctionnalité cassée semble marcher.
• Wildcards trop larges : un test abonné à sensors/# reçoit tous les messages de capteurs de tous les autres tests du broker.
• Client IDs réutilisés : deux exécutions avec le même client ID s'éjectent mutuellement du broker, et les sessions persistantes livrent des messages en file d'attente d'une exécution précédente.
• Brokers publics partagés : sur un sandbox public, des inconnus peuvent utiliser les mêmes noms de topic « évidents » que vous.

Un schéma de nommage qui passe à l'échelle

Construisez les topics de test à partir de niveaux fixes, du plus général au plus précis :

Le résultat est testmqtt/orders/20261017T101500-gh812-a3f9/test-retained-status/device/42/status. Cela paraît long, mais la partie logique à la fin ne change pas : le code testé reçoit simplement un préfixe, le même mécanisme que celui qui sépare les environnements (voir mettre en place dev, staging et production).

Quelques règles rendent ces noms sûrs :

• Uniquement de l'ASCII en minuscules, des chiffres, - et _ ; pas d'espaces, pas de / initial, jamais de + ni de # dans un nom.
• Ne commencez pas par $ : les topics qui commencent par $ sont réservés au broker ($SYS) et exclus des abonnements à #.
• Gardez chaque niveau court. Les topics voyagent dans chaque paquet PUBLISH ; voir les bonnes pratiques de conception des topics pour les règles générales.

Choisir un ID d'exécution

Le meilleur choix combine les trois : d'abord un horodatage UTC (pour nettoyer les restes selon leur âge), puis le numéro et la tentative d'exécution de la CI s'ils existent, et quelques caractères hexadécimaux aléatoires pour distinguer des exécutions lancées dans la même seconde.

Python : une fixture de préfixe pour pytest

import os, re, secrets, time
import pytest
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
SUITE = "orders"

def make_run_id() -> str:
    rid = time.strftime("%Y%m%dT%H%M%S", time.gmtime())
    if os.getenv("GITHUB_RUN_ID"):
        rid += f"-gh{os.environ['GITHUB_RUN_ID']}-{os.getenv('GITHUB_RUN_ATTEMPT', '1')}"
    return f"{rid}-{secrets.token_hex(2)}"

RUN_ID = make_run_id()

def clear_retained(client, prefix, settle=1.5):
    found = set()
    def collect(c, userdata, msg):
        if msg.retain and msg.payload:
            found.add(msg.topic)
    client.message_callback_add(f"{prefix}/#", collect)
    client.subscribe(f"{prefix}/#", qos=1)
    time.sleep(settle)                     # les messages retenus arrivent juste après le SUBACK
    client.unsubscribe(f"{prefix}/#")
    client.message_callback_remove(f"{prefix}/#")
    for topic in found:
        client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
    return found

@pytest.fixture
def prefix(request):
    case = re.sub(r"[^a-z0-9_-]+", "-", request.node.name.lower()).strip("-")
    return f"testmqtt/{SUITE}/{RUN_ID}/{case}"

@pytest.fixture
def mqttc(prefix):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"t-{RUN_ID}-{secrets.token_hex(3)}")
    if os.getenv("MQTT_USER"):
        c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.connect(HOST, 1883)
    c.loop_start()
    yield c
    clear_retained(c, prefix)
    c.loop_stop()
    c.disconnect()

Un test construit ensuite chaque topic à partir de prefix et n'en code jamais un en dur. Comme le nom du test fait partie du préfixe, les tests paramétrés et les workers parallèles (pytest-xdist) restent séparés automatiquement. Pour d'autres modèles de fixtures, voir fixtures pytest pour MQTT.

JavaScript : la même idée pour Jest ou Vitest

import { randomBytes } from "node:crypto";

const ts = new Date().toISOString().replace(/[-:]/g, "").slice(0, 15); // 20261017T101500
const ci = process.env.GITHUB_RUN_ID ? `-gh${process.env.GITHUB_RUN_ID}` : "";
export const RUN_ID = `${ts}${ci}-${randomBytes(2).toString("hex")}`;

export const testPrefix = (suite, name) =>
  `testmqtt/${suite}/${RUN_ID}/${name.toLowerCase().replace(/[^a-z0-9_-]+/g, "-")}`;

Avec MQTT.js 5, un test qui doit publier des données retenues peut aussi leur fixer une expiration, afin que le broker les supprime même si le teardown ne s'exécute jamais :

client.publish(`${prefix}/device/42/status`, "online", {
  qos: 1,
  retain: true,
  properties: { messageExpiryInterval: 3600 }, // MQTT 5 uniquement
});

L'expiration des messages nécessite une connexion MQTT 5 (protocolVersion: 5) et un broker qui la prend en charge ; les détails sont dans expiration des messages MQTT.

S'abonner de façon restreinte

Un test doit s'abonner à <son préfixe>/# ou à des topics exacts en dessous, rien de plus large. Avant de vous fier à un filtre, vérifiez-le contre les topics attendus avec le topic matcher ; un + mal placé est la façon classique dont un test se met à recevoir les messages d'un autre.

Sur un broker que vous contrôlez, imposez-le par des ACL : donnez à l'utilisateur de la CI un accès en lecture et en écriture à testmqtt/# uniquement. Un test qui vise par erreur un topic de production échoue alors avec « not authorized » au lieu de publier.

Nettoyer les restes d'exécutions interrompues

Le teardown ne s'exécute pas quand un job de CI est annulé ou qu'un portable se met en veille. Comme chaque ID d'exécution commence par un horodatage, une tâche de nettoyage planifiée peut supprimer les messages retenus des exécutions de plus d'un jour :

import os, time
import paho.mqtt.client as mqtt

cutoff = time.strftime("%Y%m%dT%H%M%S", time.gmtime(time.time() - 86400))
stale = set()

def on_message(client, userdata, msg):
    parts = msg.topic.split("/")          # testmqtt/<suite>/<run-id>/...
    if msg.retain and msg.payload and len(parts) > 3 and parts[2][:15] < cutoff:
        stale.add(msg.topic)

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-janitor")
if os.getenv("MQTT_USER"):
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
c.on_message = on_message
c.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883)
c.loop_start()
c.subscribe("testmqtt/#", qos=1)
time.sleep(5)
for topic in stale:
    c.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
print(f"cleared {len(stale)} stale retained messages")
c.loop_stop(); c.disconnect()

Lancez-la chaque nuit, par exemple comme job planifié à côté de votre pipeline de tests dans GitHub Actions. Si un message retenu refuse de disparaître, le message retenu ne s'efface pas recense les causes habituelles.

Client IDs et sessions

L'isolation des topics ne sert à rien si deux exécutions partagent un client ID. Incluez l'ID d'exécution dans chaque client ID (comme le fait la fixture), connectez-vous avec une session propre (clean_start=True en MQTT 5, clean_session=True en 3.1.1) sauf si le test porte justement sur la persistance, et limitez les IDs à 23 caractères si vous devez prendre en charge des brokers MQTT 3.1.1 stricts. Le générateur de client ID produit des IDs dans un format cohérent.

Checklist

1. Chaque topic d'un test est construit à partir d'un préfixe ; aucun topic codé en dur.
2. Le préfixe contient un ID d'exécution qui commence par un horodatage UTC.
3. Les abonnements ne remontent jamais au-dessus du préfixe du test.
4. Le teardown efface les messages retenus sous le préfixe ; une tâche nocturne s'occupe du reste.
5. Les messages de test retenus portent une expiration lorsque MQTT 5 est disponible.
6. Les client IDs incluent l'ID d'exécution ; les sessions sont propres par défaut.

Inspecter vos topics de test dans le navigateur

Nommer les topics de test est une décision de cinq minutes qui élimine toute une catégorie de tests instables. Placez un préfixe unique et horodaté devant tout, gardez les abonnements à l'intérieur et nettoyez derrière vous : vous pourrez alors exécuter en parallèle autant de suites que vous le souhaitez sur un même broker.

Questions fréquentes