pytest Fixtures for MQTT: Clean, Isolated Broker Tests

Most MQTT test suites start the same way: a test that connects, sleeps two seconds, publishes, sleeps again and checks a global list. It works on the first day and turns flaky by the tenth. pytest fixtures fix this by moving connection setup, topic naming, waiting and cleanup into a few reusable building blocks, so each test body is three lines of intent.

This guide builds a conftest.py for paho-mqtt 2.x with an environment-selected broker, a connected-client factory with automatic teardown, a unique topic per test, a message collector with timeouts and a retained-message cleaner.

What a good MQTT fixture set provides

Fixture	Scope	Responsibility
`broker`	session	Host, port, credentials and TLS from environment variables; skip if unreachable
`run_id`	session	Random ID that separates this run from any other on the same broker
`topic`	function	A unique topic prefix per test, derived from the test name
`mqtt_client`	function	Factory that returns connected clients and disconnects them afterwards
`subscriber`	function	Factory: subscribe to any filter, get a collector back
`collector`	function	A subscriber on the test's own topic, with `wait_for()` and a timeout
`retained`	function	Records retained topics and clears them on teardown

Three properties matter: isolation (no test sees another test's messages), no fixed sleeps (wait for events, with a deadline) and guaranteed cleanup (teardown runs even when the test fails).

The conftest.py

Install the dependencies: pip install pytest paho-mqtt. Then put this in tests/conftest.py:

tests/conftest.py

import os, re, socket, threading, uuid
from dataclasses import dataclass

import paho.mqtt.client as mqtt
import pytest


@dataclass(frozen=True)
class BrokerConfig:
    host: str
    port: int
    username: str | None
    password: str | None
    tls: bool


@pytest.fixture(scope="session")
def broker():
    cfg = BrokerConfig(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        username=os.getenv("MQTT_USER"),
        password=os.getenv("MQTT_PASS"),
        tls=os.getenv("MQTT_TLS") == "1",
    )
    try:
        socket.create_connection((cfg.host, cfg.port), timeout=3).close()
    except OSError as exc:
        pytest.skip(f"MQTT broker {cfg.host}:{cfg.port} unreachable: {exc}")
    return cfg


@pytest.fixture(scope="session")
def run_id():
    return uuid.uuid4().hex[:8]


@pytest.fixture
def topic(request, run_id):
    name = re.sub(r"[^A-Za-z0-9_-]", "_", request.node.name)
    return f"testmqtt/pytest/{run_id}/{name}"


@pytest.fixture
def mqtt_client(broker):
    clients = []

    def make(client_id=None, protocol=mqtt.MQTTv311, **connect_kwargs):
        client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=client_id or f"pytest-{uuid.uuid4().hex[:10]}",
            protocol=protocol,
        )
        if broker.username:
            client.username_pw_set(broker.username, broker.password)
        if broker.tls:
            client.tls_set()
        clients.append(client)

        connected, result = threading.Event(), {}

        def on_connect(c, userdata, flags, reason_code, properties):
            result["rc"] = reason_code
            connected.set()

        client.on_connect = on_connect
        client.connect(broker.host, broker.port, keepalive=30, **connect_kwargs)
        client.loop_start()
        assert connected.wait(10), f"no CONNACK from {broker.host}:{broker.port}"
        assert not result["rc"].is_failure, f"connection refused: {result['rc']}"
        return client

    yield make

    for client in clients:
        client.disconnect()
        client.loop_stop()

A few deliberate choices:

The broker comes from the environment. Locally, MQTT_HOST=localhost points at a Mosquitto container; in CI the same tests run against a Testcontainers broker or a staging instance. The default is the shared broker.connectmq.com endpoint, which is fine for quick runs.
Unreachable broker → skip, not fail. A developer without Docker running gets a clear skip reason instead of thirty connection errors. If you want CI to fail hard, set a flag that turns the skip into pytest.fail.
The factory waits for CONNACK. connect() only sends the CONNECT packet; the Event waits for the broker's answer and the assertion shows the reason code if the broker refuses.
Teardown disconnects before stopping the loop, so the DISCONNECT packet is actually sent and the broker does not publish a Last Will for a test client.
Random client IDs. Two tests with the same ID would kick each other off the broker — see client ID best practices or generate readable ones with the client ID generator.

Waiting for messages without sleep

The collector is the piece that removes time.sleep() from your tests. It appends incoming messages under a condition variable and lets the test wait until a given number has arrived:

tests/conftest.py (continued)

class MessageCollector:
    def __init__(self):
        self.messages = []
        self._cond = threading.Condition()

    def __call__(self, client, userdata, msg):  # used as on_message
        with self._cond:
            self.messages.append(msg)
            self._cond.notify_all()

    def wait_for(self, count=1, timeout=5.0):
        with self._cond:
            if not self._cond.wait_for(lambda: len(self.messages) >= count, timeout):
                got = [m.topic for m in self.messages]
                raise AssertionError(f"expected {count} message(s) in {timeout}s, got {len(got)}: {got}")
            return list(self.messages[:count])

    def assert_silent(self, duration=1.0):
        with self._cond:
            self._cond.wait_for(lambda: len(self.messages) > 0, duration)
            assert not self.messages, f"unexpected messages: {[m.topic for m in self.messages]}"


def subscribe(client, topic_filter, qos=1, timeout=5.0):
    done, result = threading.Event(), {}

    def on_subscribe(c, userdata, mid, reason_code_list, properties):
        result["codes"] = reason_code_list
        done.set()

    client.on_subscribe = on_subscribe
    client.subscribe(topic_filter, qos=qos)
    assert done.wait(timeout), f"no SUBACK for {topic_filter}"
    assert not result["codes"][0].is_failure, f"subscription refused: {result['codes'][0]}"


@pytest.fixture
def subscriber(mqtt_client):
    def make(topic_filter, qos=1):
        messages = MessageCollector()
        sub = mqtt_client()
        sub.on_message = messages
        subscribe(sub, topic_filter, qos)
        return messages
    return make


@pytest.fixture
def collector(subscriber, topic):
    return subscriber(f"{topic}/#")


@pytest.fixture
def retained(mqtt_client):
    topics = []
    yield topics.append
    if topics:
        cleaner = mqtt_client()
        for t in topics:
            cleaner.publish(t, b"", qos=1, retain=True).wait_for_publish(5)

Waiting for the SUBACK matters as much as waiting for messages: if a test publishes before the subscription is active, the broker drops the message and the test fails randomly. The subscriber and collector fixtures return only after the broker has confirmed the subscription.

assert_silent() covers the negative case — for example that an ACL denies a topic or a message with retain=False is not replayed. It waits the full duration and fails as soon as something arrives.

Tests that read like specifications

With the fixtures in place, test bodies shrink to intent:

tests/test_telemetry.py

import json
import pytest


@pytest.mark.parametrize("qos", [0, 1, 2])
def test_roundtrip(mqtt_client, collector, topic, qos):
    pub = mqtt_client()
    pub.publish(f"{topic}/temp", "21.5", qos=qos).wait_for_publish(5)

    [msg] = collector.wait_for(1)
    assert msg.topic == f"{topic}/temp"
    assert msg.payload == b"21.5"


def test_ordering_within_one_topic(mqtt_client, collector, topic):
    pub = mqtt_client()
    for i in range(20):
        pub.publish(f"{topic}/seq", json.dumps({"i": i}), qos=1)

    received = [json.loads(m.payload)["i"] for m in collector.wait_for(20, timeout=10)]
    assert received == list(range(20))


def test_retained_status_reaches_late_subscriber(mqtt_client, subscriber, topic, retained):
    status = f"{topic}/status"
    retained(status)
    mqtt_client().publish(status, "online", qos=1, retain=True).wait_for_publish(5)

    late = subscriber(status)          # subscribes after the publish
    [msg] = late.wait_for(1)
    assert msg.payload == b"online" and msg.retain

Each test gets its own topic prefix such as testmqtt/pytest/3f9a1c2e/test_roundtrip_1_ (from test_roundtrip[1]), so parametrized variants never see each other. While debugging, subscribe to testmqtt/pytest/# in the online MQTT client to watch the suite's traffic live.

The retained test registers its topic before publishing. Even if an assertion fails, the retained teardown publishes an empty retained payload and the next run starts clean — see testing retained messages for why stale retained state is a common source of flaky results.

MQTT 5 options through the factory

Because extra keyword arguments go straight to connect(), MQTT 5 features need no new fixture:

tests/test_sessions.py

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties


def test_mqtt5_connect_with_session_expiry(mqtt_client):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 300
    client = mqtt_client(protocol=mqtt.MQTTv5, clean_start=False, properties=props)
    assert client.is_connected()

Use the same pattern for user properties, request-response or message expiry tests.

Running the suite

pytest.ini

[pytest]
testpaths = tests
markers =
    mqtt: tests that need a live MQTT broker

Run locally and in CI

pytest -q                                            # default broker
MQTT_HOST=localhost pytest -q                        # local Mosquitto
MQTT_PORT=8883 MQTT_TLS=1 MQTT_USER=$MQTT_USER MQTT_PASS=$MQTT_PASS pytest -q
pip install pytest-xdist && pytest -n 4              # parallel workers

Add pytest-timeout with a global timeout = 30 in pytest.ini as a last line of defense: a test stuck on a missing message fails instead of hanging the CI job. For a full pipeline, see MQTT testing in GitHub Actions.

Pitfalls

Assertions inside callbacks. An assert in on_message runs on paho's network thread; the failure is swallowed and the test passes. Collect in the callback, assert in the test.
Session-scoped clients. Faster, but a subscription made by one test keeps delivering into the next. If you share a client, unsubscribe and reset callbacks in a function-scoped wrapper.
Topic names with wildcards. Parametrize IDs can contain characters like + or #; the re.sub() in the topic fixture strips them.
Unit tests that need no broker. Logic that only formats payloads or picks topics belongs in fast unit tests with a fake client — see mock brokers for unit tests.

A dozen lines of fixtures pay for themselves quickly: tests stop sleeping, stop interfering with each other and clean up after themselves. Keep the conftest.py in your repository, select the broker with environment variables, and the same suite runs on a laptop, in CI and against staging without edits.

Frequently asked questions

Should MQTT fixtures be session-scoped or function-scoped?

Keep broker configuration session-scoped and clients function-scoped. A fresh client per test costs one CONNECT round trip but guarantees that no subscription, callback or queued message leaks from one test into the next. Only share a client across tests if connection time is a measured problem.

How do I wait for an MQTT message in pytest without sleep?

Use a collector that appends messages under a threading.Condition and exposes wait_for(count, timeout). The test blocks only until the message arrives and fails with a clear message on timeout, instead of sleeping a fixed time and hoping.

Can MQTT tests run in parallel with pytest-xdist?

Yes, if every test uses its own topic and client ID. Deriving the topic from a per-run ID plus the test node name, and generating random client IDs, keeps parallel workers from receiving each other's messages or kicking each other off the broker.

pytest Fixtures for MQTT: Clean, Fast, Isolated Tests