pytest-Fixtures für MQTT: isolierte Broker-Tests

Die meisten MQTT-Testsuiten fangen gleich an: ein Test, der sich verbindet, zwei Sekunden schläft, publisht, wieder schläft und eine globale Liste prüft. Am ersten Tag funktioniert das, am zehnten ist es flaky. pytest-Fixtures lösen das Problem, indem sie Verbindungsaufbau, Topic-Benennung, Warten und Aufräumen in ein paar wiederverwendbare Bausteine verlagern – jeder Testkörper besteht dann aus drei Zeilen Absicht.

Diese Anleitung baut eine conftest.py für paho-mqtt 2.x mit per Umgebungsvariablen gewähltem Broker, einer Factory für verbundene Clients mit automatischem Teardown, einem eindeutigen Topic pro Test, einem Nachrichten-Collector mit Timeouts und einem Aufräumer für Retained Messages.

Was ein gutes MQTT-Fixture-Set leistet

Fixture	Scope	Aufgabe
`broker`	session	Host, Port, Zugangsdaten und TLS aus Umgebungsvariablen; Skip, wenn nicht erreichbar
`run_id`	session	Zufällige ID, die diesen Lauf von allen anderen auf demselben Broker trennt
`topic`	function	Ein eindeutiges Topic-Präfix pro Test, abgeleitet vom Testnamen
`mqtt_client`	function	Factory, die verbundene Clients liefert und danach trennt
`subscriber`	function	Factory: beliebigen Filter abonnieren, Collector zurückbekommen
`collector`	function	Ein Subscriber auf dem Topic des Tests, mit `wait_for()` und Timeout
`retained`	function	Merkt sich Retained-Topics und leert sie im Teardown

Drei Eigenschaften zählen: Isolation (kein Test sieht Nachrichten eines anderen), keine festen Sleeps (auf Ereignisse warten, mit Deadline) und garantiertes Aufräumen (Teardown läuft auch, wenn der Test fehlschlägt).

Die conftest.py

Installiere die Abhängigkeiten: pip install pytest paho-mqtt. Dann kommt Folgendes 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()

Ein paar bewusste Entscheidungen:

Der Broker kommt aus der Umgebung. Lokal zeigt MQTT_HOST=localhost auf einen Mosquitto-Container; in CI laufen dieselben Tests gegen einen Testcontainers-Broker oder eine Staging-Instanz. Standard ist der geteilte Endpoint broker.connectmq.com, der für schnelle Läufe reicht.
Broker nicht erreichbar → Skip, kein Fehler. Wer Docker gerade nicht laufen hat, bekommt einen klaren Skip-Grund statt dreißig Verbindungsfehlern. Soll CI hart fehlschlagen, setze ein Flag, das den Skip in pytest.fail umwandelt.
Die Factory wartet auf CONNACK. connect() sendet nur das CONNECT-Paket; das Event wartet auf die Antwort des Brokers, und die Assertion zeigt den Reason Code, falls der Broker ablehnt.
Der Teardown trennt die Verbindung, bevor er den Loop stoppt, damit das DISCONNECT-Paket wirklich gesendet wird und der Broker kein Last Will für einen Test-Client veröffentlicht.
Zufällige Client-IDs. Zwei Tests mit derselben ID würden sich gegenseitig vom Broker werfen – siehe Best Practices für Client-IDs, oder erzeuge lesbare IDs mit dem Client-ID-Generator.

Auf Nachrichten warten ohne sleep

Der Collector ist das Teil, das time.sleep() aus deinen Tests entfernt. Er sammelt eingehende Nachrichten unter einer Condition Variable und lässt den Test warten, bis eine bestimmte Anzahl angekommen ist:

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)

Auf das SUBACK zu warten ist genauso wichtig wie auf die Nachrichten: Publisht ein Test, bevor die Subscription aktiv ist, verwirft der Broker die Nachricht, und der Test schlägt zufällig fehl. Die Fixtures subscriber und collector kehren erst zurück, wenn der Broker die Subscription bestätigt hat.

assert_silent() deckt den negativen Fall ab – etwa dass eine ACL ein Topic verweigert oder eine Nachricht mit retain=False nicht erneut ausgeliefert wird. Sie wartet die volle Dauer und schlägt fehl, sobald etwas ankommt.

Tests, die sich wie Spezifikationen lesen

Mit den Fixtures schrumpfen Testkörper auf die Absicht:

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

Jeder Test bekommt sein eigenes Topic-Präfix wie testmqtt/pytest/3f9a1c2e/test_roundtrip_1_ (aus test_roundtrip[1]), parametrisierte Varianten sehen sich also nie gegenseitig. Beim Debuggen abonnierst du testmqtt/pytest/# im Online-MQTT-Client und siehst den Verkehr der Suite live.

Der Retained-Test registriert sein Topic vor dem Publish. Selbst wenn eine Assertion fehlschlägt, publisht der retained-Teardown einen leeren Retained-Payload, und der nächste Lauf startet sauber – warum veralteter Retained-Zustand eine häufige Quelle für flaky Ergebnisse ist, erklärt Retained Messages testen.

MQTT-5-Optionen über die Factory

Da zusätzliche Keyword-Argumente direkt an connect() gehen, brauchen MQTT-5-Features keine neue 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()

Dasselbe Muster funktioniert für Tests von User Properties, Request-Response oder Message Expiry.

Die Suite ausführen

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

Ergänze pytest-timeout mit einem globalen timeout = 30 in der pytest.ini als letzte Verteidigungslinie: Ein Test, der auf eine nie eintreffende Nachricht wartet, schlägt fehl, statt den CI-Job hängen zu lassen. Eine komplette Pipeline zeigt MQTT-Tests in GitHub Actions.

Stolperfallen

Assertions in Callbacks. Ein assert in on_message läuft im Netzwerk-Thread von paho; der Fehler wird verschluckt, und der Test besteht. Im Callback sammeln, im Test prüfen.
Clients mit Session-Scope. Schneller, aber eine Subscription aus einem Test liefert weiter Nachrichten in den nächsten. Wenn du einen Client teilst, deabonniere und setze Callbacks in einem Wrapper mit Function-Scope zurück.
Topic-Namen mit Wildcards. Parametrize-IDs können Zeichen wie + oder # enthalten; das re.sub() in der topic-Fixture entfernt sie.
Unit-Tests ohne Broker. Logik, die nur Payloads formatiert oder Topics wählt, gehört in schnelle Unit-Tests mit einem Fake-Client – siehe Mock-Broker für Unit-Tests.

Ein Dutzend Zeilen Fixtures zahlt sich schnell aus: Tests schlafen nicht mehr, stören sich nicht mehr gegenseitig und räumen hinter sich auf. Leg die conftest.py ins Repository, wähle den Broker per Umgebungsvariablen, und dieselbe Suite läuft ohne Änderungen auf dem Laptop, in CI und gegen Staging.

Häufig gestellte Fragen

Sollten MQTT-Fixtures Session- oder Function-Scope haben?

Broker-Konfiguration mit Session-Scope, Clients mit Function-Scope. Ein frischer Client pro Test kostet einen CONNECT-Roundtrip, garantiert aber, dass keine Subscription, kein Callback und keine wartende Nachricht von einem Test in den nächsten durchsickert. Teile einen Client nur, wenn die Verbindungszeit nachweislich ein Problem ist.

Wie warte ich in pytest ohne sleep auf eine MQTT-Nachricht?

Mit einem Collector, der Nachrichten unter einer threading.Condition sammelt und wait_for(count, timeout) anbietet. Der Test blockiert nur, bis die Nachricht da ist, und schlägt bei Timeout mit einer klaren Meldung fehl, statt eine feste Zeit zu schlafen und zu hoffen.

Können MQTT-Tests mit pytest-xdist parallel laufen?

Ja, wenn jeder Test ein eigenes Topic und eine eigene Client-ID verwendet. Leitest du das Topic aus einer ID pro Lauf plus dem Namen des Testknotens ab und erzeugst zufällige Client-IDs, empfangen parallele Worker weder fremde Nachrichten noch werfen sie sich gegenseitig vom Broker.

pytest-Fixtures für MQTT: saubere, schnelle, isolierte Tests