MQTT-Clients testen: Unit-, Integrations- und E2E-Tests

MQTT-Clients zu testen ist umständlich, weil das interessante Verhalten über ein Netzwerk hinweg passiert, in Callbacks, auf dem Thread eines anderen. Dieser Leitfaden zeigt einen geschichteten Ansatz, der funktioniert: schnelle Unit-Tests mit einem gemockten Client, Integrationstests gegen einen Wegwerf-Broker-Container in der CI und ein kurzer End-to-End-Smoke-Test gegen einen echten Cloud-Broker vor dem Release.

Warum MQTT-Client-Code schwer zu testen ist

MQTT-Code vermischt drei Belange, die auf unterschiedliche Weise scheitern:

Geschäftslogik: Payloads parsen, entscheiden, was publiziert wird, Topics auf Geräte abbilden.
Protokollverhalten: QoS, Retained Messages, Reconnects, erneutes Abonnieren, Last Will.
Umgebung: TLS-Zertifikate, Zugangsdaten, Ports, Firewalls, Broker-Kontingente.

Die meisten wackeligen MQTT-Testsuites versuchen, alle drei in einem einzigen Test abzudecken, der sich mit einem Broker verbindet und schläft. Trennt man sie, kann jede Ebene das günstigste Werkzeug nutzen, das ihre Bugs findet.

Ebene	Was sie findet	Broker nötig	Geschwindigkeit	Läuft
Unit	Parsing, Topic-Zuordnung, Entscheidungslogik	Nein (gemockt)	Millisekunden	Bei jedem Speichern
Integration	Verdrahtung von Subscribe/Publish, QoS, Retained, Reconnect	Lokaler Container	Sekunden	Bei jedem Pull Request
End-to-End-Smoke	TLS, Zugangsdaten, Ports, echtes Netzwerk	Echter Cloud-Broker	Sekunden	Vor dem Release, nach Zeitplan

Die Beispiele verwenden Python mit paho-mqtt 2.x und pytest, doch die Struktur lässt sich auf jede Sprache übertragen.

Schritt 1: Testbaren Code entwerfen

Die nützlichste Einzeländerung ist, den MQTT-Client an den Rand zu verlagern. Legen Sie Entscheidungen in einfache Funktionen oder Klassen, die eine Publish-Funktion als Abhängigkeit erhalten, und lassen Sie einen dünnen Adapter sie mit paho verbinden.

thermostat.py:

import json
from dataclasses import dataclass
from typing import Callable

Publish = Callable[[str, str, int, bool], None]


@dataclass
class Thermostat:
    prefix: str
    publish: Publish
    setpoint: float = 21.0
    hysteresis: float = 0.5
    heating: bool = False

    def on_temperature(self, topic: str, payload: bytes) -> None:
        try:
            value = float(json.loads(payload)["value"])
        except (ValueError, KeyError, TypeError, json.JSONDecodeError):
            return  # fehlerhafte Nachrichten ignorieren

        room = topic.split("/")[-2]
        if not self.heating and value < self.setpoint - self.hysteresis:
            self.heating = True
        elif self.heating and value > self.setpoint + self.hysteresis:
            self.heating = False
        else:
            return

        self.publish(f"{self.prefix}/{room}/heater/set",
                     "ON" if self.heating else "OFF", 1, True)

adapter.py:

import threading

import paho.mqtt.client as mqtt
from thermostat import Thermostat


def build_client(host, port, client_id, prefix, username=None, password=None, tls=False):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if username:
        client.username_pw_set(username, password)
    if tls:
        client.tls_set()

    thermostat = Thermostat(
        prefix=prefix,
        publish=lambda t, p, q, r: client.publish(t, p, qos=q, retain=r),
    )

    def on_connect(client, userdata, flags, reason_code, properties):
        if not reason_code.is_failure:
            client.subscribe(f"{prefix}/+/temperature", qos=1)

    ready = threading.Event()

    def on_subscribe(client, userdata, mid, reason_codes, properties):
        ready.set()

    def on_message(client, userdata, msg):
        thermostat.on_temperature(msg.topic, msg.payload)

    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    return client, thermostat, ready

Dasselbe Muster trägt auch in der Industrie, etwa wenn ein Edge-Dienst SPS-Rohwerte skaliert und in die ISA-95-basierte Topic-Struktur eines Unified Namespace (UNS) wie acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur übersetzt: Die Übersetzungslogik lässt sich ohne Broker testen, nur der Adapter braucht einen.

Das Abonnieren in on_connect ist Absicht: Es stellt Abonnements nach jedem Reconnect wieder her, und das zurückgegebene Event ready erlaubt Tests, zu warten, bis das Abonnement aktiv ist, statt mit sleep() zu raten. Genau diese Art von Verdrahtung prüfen die Integrationstests.

Schritt 2: Unit-Tests mit einem gemockten Client

Da Thermostat den Parameter publish entgegennimmt, ersetzt ein Mock das Netzwerk vollständig.

test_thermostat.py:

import json
from unittest.mock import Mock

from thermostat import Thermostat

PREFIX = "home/test"


def make():
    pub = Mock()
    return Thermostat(prefix=PREFIX, publish=pub), pub


def msg(value):
    return json.dumps({"value": value}).encode()


def test_turns_heater_on_below_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    pub.assert_called_once_with(f"{PREFIX}/kitchen/heater/set", "ON", 1, True)


def test_no_publish_inside_hysteresis_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.2))
    pub.assert_not_called()


def test_turns_heater_off_above_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.6))
    assert pub.call_args.args[1] == "OFF"


def test_ignores_garbage():
    t, pub = make()
    for bad in [b"", b"not json", b'{"temp": 3}', b'{"value": "hot"}']:
        t.on_temperature(f"{PREFIX}/kitchen/temperature", bad)
    pub.assert_not_called()

Diese Tests laufen in Millisekunden und decken die Logik ab, die am häufigsten Bugs enthält. Widerstehen Sie der Versuchung, paho selbst detailliert zu mocken (etwa zu prüfen, dass client.subscribe mit exakten Argumenten aufgerufen wurde): Solche Tests spiegeln nur die Implementierung, brechen bei Refactorings und beweisen trotzdem nicht, dass der Broker akzeptiert, was Sie senden. Das ist Aufgabe der Integrationsebene.

Schritt 3: Integrationstests gegen einen Broker-Container

Ein echter Broker in einem Container liefert ehrliches Protokollverhalten, ohne geteilten Zustand und ohne Abhängigkeit vom Internet. Das offizielle Eclipse-Mosquitto-Image enthält eine Konfigurationsdatei, die anonyme Verbindungen erlaubt, was sich ideal für Tests eignet:

docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf

Schreiben Sie nun Tests, die den echten Adapter ausführen. Eine Hilfsfunktion sammelt Nachrichten auf einem zweiten Client und wartet auf eine Bedingung, statt eine feste Zeit zu schlafen.

test_integration.py:

import os
import queue
import uuid

import paho.mqtt.client as mqtt
import pytest

from adapter import build_client

HOST = os.getenv("MQTT_TEST_HOST", "localhost")
PORT = int(os.getenv("MQTT_TEST_PORT", "1883"))


@pytest.fixture
def prefix():
    return f"test/{uuid.uuid4().hex[:8]}"


@pytest.fixture
def probe(prefix):
    received = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: received.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/#", qos=1)
    c.loop_start()
    yield c, received
    c.loop_stop()
    c.disconnect()


def wait_for(q, predicate, timeout=5.0):
    while True:
        topic, payload, retain = q.get(timeout=timeout)  # löst bei Timeout queue.Empty aus
        if predicate(topic, payload):
            return topic, payload, retain


def test_temperature_drives_heater(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/office/temperature", '{"value": 18.5}', qos=1)
        topic, payload, _ = wait_for(received, lambda t, p: t.endswith("/heater/set"))
        assert topic == f"{prefix}/office/heater/set"
        assert payload == "ON"
    finally:
        client.loop_stop()
        client.disconnect()


def test_heater_command_is_retained(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/hall/temperature", '{"value": 15}', qos=1)
        wait_for(received, lambda t, p: t.endswith("/heater/set"))
    finally:
        client.loop_stop()
        client.disconnect()

    # Ein später Subscriber muss den retained Befehl sofort erhalten
    late = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"late-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: late.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/hall/heater/set", qos=1)
    c.loop_start()
    try:
        _, payload, retain = late.get(timeout=5)
        assert payload == "ON" and retain is True
    finally:
        c.loop_stop()
        c.disconnect()

Drei Gewohnheiten halten Integrationstests stabil:

Eindeutiges Topic-Präfix pro Test. Retained Messages eines Tests können nicht in den nächsten durchsickern.
Auf Bedingungen warten, nicht auf sleep(). Eine Queue mit Timeout scheitert schnell und wartet nie länger als nötig.
Schleifen immer in finally stoppen und trennen. Verwaiste Netzwerk-Threads lassen spätere Tests hängen.

Um die Reconnect-Logik zu testen, stoppen und starten Sie den Container mitten im Test (docker restart test-broker) und prüfen, dass der Client neu abonniert und eine nach dem Neustart publizierte Nachricht empfängt.

Den Broker in der CI betreiben

GitHub Actions kann den Container als Schritt vor den Tests starten:

name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Start MQTT broker
        run: |
          docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
            mosquitto -c /mosquitto-no-auth.conf
      - run: pip install "paho-mqtt>=2,<3" pytest
      - run: pytest -q

Unit- und Integrationstests können sich einen pytest-Lauf teilen; markieren Sie die langsamen mit @pytest.mark.integration, wenn Sie sie lokal überspringen möchten.

Schritt 4: End-to-End-Smoke-Tests gegen einen echten Broker

Ein Container kann Ihnen nicht sagen, ob Ihre TLS-Einstellungen, Ihr CA-Bundle, Ihre Firewall-Regeln oder Ihre Produktionszugangsdaten funktionieren. Ein kurzer Smoke-Test gegen den echten Endpunkt kann das. Für Open-Source-Projekte oder schnelle Prüfungen reicht der Öffentliche MQTT-Broker unter broker.connectmq.com; er akzeptiert anonyme Clients über TLS auf Port 8883. Verwenden Sie nur synthetische Payloads, denn der öffentliche MQTT-Broker ist für alle lesbar und nur für Tests und Prototypen gedacht.

smoke_test.py:

import os
import sys
import threading
import uuid

import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
USER = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"smoke-{uuid.uuid4().hex[:8]}"
TOPIC = f"{os.getenv('MQTT_PREFIX', 'connectmq/demo')}/smoke-{uuid.uuid4().hex[:8]}/ping"

got = threading.Event()
nonce = uuid.uuid4().hex
errors = []


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        errors.append(f"connect failed: {reason_code}")
        got.set()  # Haupt-Thread wecken; Callbacks können den Prozess nicht beenden
        return
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    client.publish(TOPIC, nonce, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == nonce:
        got.set()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USER:
    client.username_pw_set(USER, PASSWORD)
client.tls_set()
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 8883, keepalive=30)
client.loop_start()
ok = got.wait(timeout=15) and not errors
for e in errors:
    print(e)
client.loop_stop()
client.disconnect()
print("smoke test passed" if ok else "smoke test FAILED: no echo within 15 s")
sys.exit(0 if ok else 1)

Für die Produktion richten Sie dasselbe Skript auf Ihren privaten Broker aus, indem Sie MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID und MQTT_PREFIX aus CI-Secrets setzen. Bei einem verwalteten MQTT-Broker wie ConnectMQ sind Client IDs an die ausgegebenen Zugangsdaten gebunden. Legen Sie daher einen eigenen Satz Zugangsdaten für die CI an, statt die Identität eines Geräts auszuleihen; sonst trennt der Smoke-Test dieses Gerät.

Führen Sie Smoke-Tests vor Releases und nach Zeitplan aus (zum Beispiel stündlich), nicht bei jedem Commit. Auf geteilten Brokern gelten Kontingente: Der öffentliche MQTT-Broker erlaubt 10 Nachrichten pro Sekunde und 10 Abonnements pro Client, und der kostenlose ConnectMQ Serverless-Plan begrenzt gleichzeitige Verbindungen auf 25.

Häufige Anti-Patterns

Einige Gewohnheiten machen MQTT-Testsuites langsam und wackelig. Achten Sie im Code-Review darauf:

Feste Sleeps. Ein time.sleep(2) nach dem Publizieren ist entweder zu lang und verschwendet über eine ganze Suite Minuten, oder zu kurz auf einem ausgelasteten CI-Runner. Warten Sie stattdessen mit Timeout auf eine Queue oder ein Event.
Geteilte Topics zwischen Tests. Retained Messages und verbliebene Abonnements eines Tests sickern in den nächsten durch. Erzeugen Sie pro Test ein eindeutiges Präfix.
Verschluckte Callback-Fehler. Exceptions in MQTT-Callbacks laufen auf dem Netzwerk-Thread und lassen den Test daher nicht direkt fehlschlagen. Sammeln Sie sie in einer Liste und prüfen Sie beim Teardown, dass sie leer ist.
Gerätezugangsdaten in der CI. Wer für Tests die Identität eines echten Geräts ausleiht, trennt dieses Gerät. Legen Sie eigene Zugangsdaten für die Automatisierung an.
Die Bibliothek testen statt Ihres Codes. Sie müssen nicht beweisen, dass paho QoS-1-Nachrichten zustellt; Sie müssen beweisen, dass Ihr Code die richtigen Topics abonniert und korrekt reagiert.
Ein riesiger End-to-End-Test. Schlägt er fehl, weiß niemand, welche Ebene kaputt ist. Halten Sie End-to-End-Tests wenige und schmal, und verlagern Sie Details in Unit-Tests.

Checkliste zum Testen

Die Geschäftslogik liegt außerhalb der MQTT-Callbacks und wird mit einem gemockten publish per Unit-Test geprüft.
Fehlerhafte, leere und übergroße Payloads werden getestet und sicher ignoriert.
Integrationstests nutzen einen frischen Broker-Container und ein eindeutiges Topic-Präfix pro Test.
Reconnect und erneutes Abonnieren werden durch einen Neustart des Brokers getestet.
Retained Messages und das Verhalten des Last Will werden mit einem späten Subscriber geprüft.
Vor jedem Release läuft ein TLS-Smoke-Test mit echten Zugangsdaten.
CI-Zugangsdaten sind von Gerätezugangsdaten getrennt.

Sprachspezifische Details finden Sie im PHP MQTT Client Tutorial, im Paho MQTT C Client Tutorial und im umfassenderen Leitfaden zu MQTT-Client-Bibliotheken.

Starten Sie mit ConnectMQ

Geben Sie Ihrer Test-Pipeline einen echten Broker als Gegenüber. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless Broker, ganz ohne Kreditkarte, legen Sie einen eigenen Satz CI-Zugangsdaten an und führen Sie den Smoke-Test bei jedem Release dagegen aus. Der Live-Topic-Explorer und der Browser-Client des Portals zeigen Ihnen genau, was Ihre Tests publiziert haben. ConnectMQ betreibt den Broker, Sie behalten die Tests.

Häufig gestellte Fragen

Sollte ich die MQTT-Bibliothek in Unit-Tests mocken?

Mocken Sie an Ihrer eigenen Grenze, etwa an einer injizierten Funktion publish, statt die Interna der Bibliothek zu mocken. Prüfen Sie echtes Protokollverhalten mit einem Integrationstest gegen einen Broker-Container.

Welchen Broker sollte ich für MQTT-Integrationstests in der CI verwenden?

Ein Wegwerf-Container mit Eclipse Mosquitto ist die einfachste Wahl: Er startet in etwa einer Sekunde, braucht mit /mosquitto-no-auth.conf keine Zugangsdaten und hinterlässt zwischen den Läufen keinen Zustand.

Kann ich broker.connectmq.com in automatisierten Tests verwenden?

Ja, für gelegentliche Smoke-Tests mit synthetischen Daten und eindeutigen Topics, innerhalb seiner Limits von 10 Nachrichten pro Sekunde und 10 Abonnements pro Client. Für umfangreiche Testsuites nutzen Sie einen lokalen Container oder Ihren eigenen privaten Broker.

Wie teste ich die Reconnect-Logik von MQTT?

Starten Sie den Broker-Container während des Tests neu und prüfen Sie, dass der Client sich neu verbindet, neu abonniert und eine nach dem Neustart publizierte Nachricht empfängt.

MQTT-Clients testen: Unit-, Integrations- und End-to-End-Tests