Test-Topics so benennen, dass Tests nie kollidieren

MQTT-Tests, die einzeln bestehen und gemeinsam scheitern, teilen sich fast immer Topics. Zwei CI-Jobs publizieren auf test/temp, eine Retained Message von gestern beantwortet die Assertion von heute, oder ein Subscriber auf # schluckt den Traffic von jemand anderem. Die Lösung ist eine Namenskonvention: Jeder Lauf bekommt ein eindeutiges Präfix wie testmqtt/<suite>/<run-id>/, Tests abonnieren nur darunter, und der Teardown löscht alles, was retained wurde.

Dieser Guide zeigt ein skalierbares Namensschema, wie du Run-IDs erzeugst, Helfer für Python und JavaScript und wie du Retained Testdaten aufräumst – auch Überreste abgestürzter Läufe.

Warum MQTT-Tests kollidieren

Ein Broker ist geteilter Zustand. Anders als ein In-Memory-Mock merkt er sich Dinge und verteilt sie weiter, und so beeinflussen sich Tests auf einige vorhersehbare Arten:

• Parallele Läufe – zwei Pipelines, zwei Entwickler oder zwei pytest-Worker publizieren auf dasselbe Topic und empfangen gegenseitig ihre Nachrichten.
• Retained Überreste – eine Retained Message überlebt den Test, der sie erzeugt hat, und wird sofort an den nächsten Subscriber ausgeliefert, sodass ein kaputtes Feature zu funktionieren scheint.
• Zu breite Wildcards – ein Test, der sensors/# abonniert, erhält jede Sensornachricht aller anderen Tests auf dem Broker.
• Wiederverwendete Client-IDs – zwei Läufe mit derselben Client-ID werfen sich gegenseitig vom Broker, und persistente Sessions liefern Nachrichten aus der Warteschlange eines früheren Laufs aus.
• Geteilte öffentliche Broker – auf einer öffentlichen Sandbox nutzen Fremde womöglich dieselben „naheliegenden“ Topic-Namen wie du.

Ein Namensschema, das skaliert

Baue Test-Topics aus festen Ebenen auf, vom Allgemeinen zum Speziellen:

Das Ergebnis ist testmqtt/orders/20261017T101500-gh812-a3f9/test-retained-status/device/42/status. Das wirkt lang, aber der logische Teil am Ende bleibt unverändert – der getestete Code bekommt nur ein Präfix, derselbe Mechanismus, mit dem man Umgebungen trennt (siehe Dev, Staging und Produktion einrichten).

Ein paar Regeln halten diese Namen sicher:

• Nur ASCII-Kleinbuchstaben, Ziffern, - und _; keine Leerzeichen, kein führendes /, niemals + oder # in einem Namen.
• Nicht mit $ beginnen – Topics, die mit $ anfangen, sind für den Broker reserviert ($SYS) und von #-Subscriptions ausgenommen.
• Halte jede Ebene kurz. Topics reisen in jedem PUBLISH-Paket mit; die allgemeinen Regeln stehen in Best Practices für Topic-Design.

Eine Run-ID wählen

Am besten kombinierst du sie: zuerst ein UTC-Zeitstempel (damit Überreste nach Alter aufgeräumt werden können), dann CI-Laufnummer und -Versuch, falls vorhanden, und ein paar zufällige Hex-Zeichen, um Läufe zu trennen, die in derselben Sekunde starten.

Python: eine Präfix-Fixture für 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)                     # Retained Messages kommen direkt nach dem SUBACK an
    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()

Ein Test baut dann jedes Topic aus prefix und codiert nie eines fest. Weil der Testname Teil des Präfixes ist, bleiben parametrisierte Tests und parallele Worker (pytest-xdist) automatisch getrennt. Weitere Fixture-Muster findest du unter pytest-Fixtures für MQTT.

JavaScript: dieselbe Idee für Jest oder 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, "-")}`;

Mit MQTT.js 5 kann ein Test, der Retained-Daten publizieren muss, zusätzlich ein Ablaufdatum setzen, sodass der Broker sie verwirft, selbst wenn der Teardown nie läuft:

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

Message Expiry erfordert eine MQTT-5-Verbindung (protocolVersion: 5) und einen Broker, der es unterstützt; Details unter MQTT Message Expiry.

Eng abonnieren

Ein Test sollte <sein Präfix>/# oder exakte Topics darunter abonnieren – nichts Breiteres. Bevor du dich auf einen Filter verlässt, prüfe ihn mit dem Topic-Matcher gegen die erwarteten Topics; ein falsch gesetztes + ist der klassische Weg, auf dem ein Test plötzlich die Nachrichten eines anderen empfängt.

Auf einem Broker, den du kontrollierst, setzt du das per ACL durch: Gib dem CI-Benutzer Lese- und Schreibrechte nur auf testmqtt/#. Ein Test, der versehentlich ein Produktions-Topic anspricht, scheitert dann mit „not authorized“, statt zu publizieren.

Überreste abgebrochener Läufe aufräumen

Der Teardown läuft nicht, wenn ein CI-Job abgebrochen wird oder ein Laptop in den Ruhezustand geht. Weil jede Run-ID mit einem Zeitstempel beginnt, kann ein geplanter Aufräumjob Retained Messages von Läufen löschen, die älter als einen Tag sind:

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

Lass ihn jede Nacht laufen, zum Beispiel als geplanten Job neben deiner Test-Pipeline in GitHub Actions. Wenn eine Retained Message sich nicht löschen lässt, listet Retained Message lässt sich nicht löschen die üblichen Gründe auf.

Client-IDs und Sessions

Topic-Isolation ist verschenkt, wenn zwei Läufe sich eine Client-ID teilen. Nimm die Run-ID in jede Client-ID auf (wie die Fixture), verbinde dich mit einer sauberen Session (clean_start=True in MQTT 5, clean_session=True in 3.1.1), sofern der Test nicht gerade Persistenz prüft, und halte IDs bei höchstens 23 Zeichen, wenn du strikte MQTT-3.1.1-Broker unterstützen musst. Der Client-ID-Generator erzeugt IDs in einem einheitlichen Format.

Checkliste

1. Jedes Topic in einem Test wird aus einem Präfix gebaut; keine fest codierten Topic-Strings.
2. Das Präfix enthält eine Run-ID, die mit einem UTC-Zeitstempel beginnt.
3. Subscriptions reichen nie über das eigene Präfix des Tests hinaus.
4. Der Teardown löscht Retained Messages unter dem Präfix; ein nächtlicher Aufräumjob erledigt den Rest.
5. Retained Test-Nachrichten haben ein Ablaufdatum, wo MQTT 5 verfügbar ist.
6. Client-IDs enthalten die Run-ID; Sessions sind standardmäßig sauber.

Deine Test-Topics im Browser untersuchen

Test-Topics zu benennen ist eine Fünf-Minuten-Entscheidung, die eine ganze Klasse von flaky Tests beseitigt. Setze ein eindeutiges Präfix mit Zeitstempel vor alles, halte Subscriptions darin und räume hinter dir auf – dann kannst du beliebig viele Suites parallel gegen einen Broker laufen lassen.

Häufig gestellte Fragen