Naming Test Topics So Tests Never Collide

MQTT tests that pass alone and fail together almost always share topics. Two CI jobs publish to test/temp, a retained message from yesterday answers today's assertion, or a subscriber on # swallows someone else's traffic. The fix is a naming convention: every run gets a unique prefix like testmqtt/<suite>/<run-id>/, tests subscribe only below it, and teardown deletes whatever was retained.

This guide shows a naming scheme that scales, how to build run IDs, helpers for Python and JavaScript, and how to clean up retained test data — including leftovers from crashed runs.

Why MQTT tests collide

A broker is shared state. Unlike an in-memory mock, it remembers and fans out, so tests leak into each other in a few predictable ways:

• Parallel runs — two pipelines, two developers or two pytest workers publish to the same topic and receive each other's messages.
• Retained leftovers — a retained message survives the test that created it and is delivered instantly to the next subscriber, making a broken feature look like it works.
• Broad wildcards — a test subscribing to sensors/# receives every sensor message from every other test on the broker.
• Reused client IDs — two runs with the same client ID kick each other off the broker, and persistent sessions deliver queued messages from a previous run.
• Shared public brokers — on a public sandbox, strangers may use the same "obvious" topic names you do.

A naming scheme that scales

Build test topics from fixed levels, most general first:

The result is testmqtt/orders/20261017T101500-gh812-a3f9/test-retained-status/device/42/status. It looks long, but the logical part at the end is unchanged — your code under test just receives a prefix, which is the same mechanism used to separate environments (see setting up dev, staging and production).

A few rules keep these names safe:

• Lowercase ASCII, digits, - and _ only; no spaces, no leading /, never + or # inside a name.
• Do not start with $ — topics beginning with $ are reserved for the broker ($SYS) and are excluded from # subscriptions.
• Keep each level short. Topics travel in every PUBLISH packet; see topic design best practices for the general rules.

Choosing a run ID

The best choice combines them: a UTC timestamp first (so leftovers can be cleaned up by age), the CI run number and attempt if available, and a few random hex characters to separate runs that start in the same second.

Python: a prefix fixture for 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 arrive right after 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()

A test then builds every topic from prefix and never hard-codes one. Because the test name is part of the prefix, parametrized tests and parallel workers (pytest-xdist) stay separate automatically. For more fixture patterns, see pytest fixtures for MQTT.

JavaScript: the same idea for Jest or 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, "-")}`;

With MQTT.js 5, a test that must publish retained data can also attach an expiry, so the broker discards it even if teardown never runs:

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

Message expiry requires an MQTT 5 connection (protocolVersion: 5) and a broker that supports it; details are in MQTT message expiry.

Subscribe narrowly

A test should subscribe to <its prefix>/# or to exact topics below it — nothing broader. Before relying on a filter, check it against the topics you expect with the topic matcher; a misplaced + is the classic way a test starts receiving another test's messages.

On a broker you control, enforce this with ACLs: give the CI user read and write access to testmqtt/# only. A test that accidentally targets a production topic then fails with "not authorized" instead of publishing.

Clean up leftovers from crashed runs

Teardown does not run when a CI job is cancelled or a laptop goes to sleep. Because every run ID starts with a timestamp, a scheduled janitor can delete retained messages from runs older than a day:

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

Run it nightly, for example as a scheduled job next to your test pipeline in GitHub Actions. If a retained message refuses to go away, retained message won't clear lists the usual reasons.

Client IDs and sessions

Topic isolation is wasted if two runs share a client ID. Include the run ID in every client ID (as the fixture does), connect with a clean session (clean_start=True in MQTT 5, clean_session=True in 3.1.1) unless the test is specifically about persistence, and keep IDs within 23 characters if you must support strict MQTT 3.1.1 brokers. The client ID generator produces IDs in a consistent format.

Checklist

1. Every topic in a test is built from a prefix; no hard-coded topic strings.
2. The prefix contains a run ID that starts with a UTC timestamp.
3. Subscriptions never go above the test's own prefix.
4. Teardown clears retained messages under the prefix; a nightly janitor catches the rest.
5. Retained test messages carry a message expiry where MQTT 5 is available.
6. Client IDs include the run ID; sessions are clean by default.

Inspect your test topics in the browser

Naming test topics is a five-minute decision that removes a whole class of flaky tests. Put a unique, timestamped prefix in front of everything, keep subscriptions inside it, and clean up after yourself — then you can run as many suites in parallel against one broker as you like.

Frequently asked questions