MQTT-Integrationstests: Deine App gegen einen echten Broker testen

MQTT-Integrationstests lassen deine Anwendung gegen einen echten Broker laufen: Dein Code verbindet sich, abonniert und publisht, und der Test prüft die Nachrichten, die tatsächlich über die Leitung gehen. Sie finden, was Unit-Tests mit Mocks nicht finden können – falsche Topic-Filter, fehlende QoS-Bestätigungen, Retained-Zustand, ACL-Ablehnungen und Reconnect-Bugs.

Diese Anleitung zeigt funktionierende Muster für pytest (paho-mqtt 2.x) und Jest (MQTT.js 5) sowie die fünf Regeln, mit denen diese Tests schnell und frei von Flakiness bleiben.

Unit, Integration oder End-to-End?

Behalte die Mock-basierten Tests für reine Logik – Mock-Broker für Unit-Tests deckt diese Ebene ab. Alles, was davon abhängt, wie sich ein Broker verhält (Wildcards, Retained Messages, QoS-1-Neuzustellung, Session Expiry), gehört in Integrationstests gegen einen echten Broker.

Den Test-Broker wählen

Du hast drei praktikable Optionen:

• Lokaler Mosquitto (installiert oder per Docker) – schnell und privat, aber jeder Entwickler und jeder CI-Runner muss ihn starten.
• Kurzlebiger Container pro Testlauf – vollständig isoliert und wegwerfbar; siehe MQTT-Tests mit Testcontainers.
• Gehosteter Broker – kein Setup, echtes TLS und echte Authentifizierung, und derselbe Endpunkt von Laptops, CI und Staging-Geräten aus.

Deinem Testcode sollte egal sein, mit welchem er spricht. Lies Host, Port und Zugangsdaten aus Umgebungsvariablen und setze einen Standard, der sofort funktioniert:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
# MQTT_PORT=8883 for TLS; set MQTT_USER / MQTT_PASS for a private broker

Die fünf Regeln für stabile MQTT-Tests

1. Eindeutiges Topic-Präfix pro Test. Baue jedes Topic aus testmqtt/it/<run-id>/<random-id>. Parallele Test-Worker, CI-Jobs und Entwickler sehen dann nie die Nachrichten der anderen.
2. Eindeutige Client-ID pro Verbindung. Zwei Verbindungen mit derselben ID werfen sich gegenseitig raus – eine klassische Quelle für Tests, die „zufällig" Nachrichten verlieren.
3. Auf Events warten, nie schlafen. Warte auf CONNACK, bevor du abonnierst, auf SUBACK, bevor du publisht, und auf eine Nachricht mit Timeout. Ein sleep(1) ist entweder zu lang (langsame Suite) oder zu kurz (flaky Suite).
4. Aufräumen, was den Test überlebt. Retained Messages und persistente Sessions überleben Verbindungsabbrüche. Lösche Retained-Topics mit einem leeren Retained-Publish und nutze Clean Sessions, außer du testest Sessions.
5. Inhalte prüfen, nicht Timing oder globale Reihenfolge. MQTT garantiert die Reihenfolge nur pro Topic und QoS von einem Publisher. Korreliere Nachrichten über eine ID in der Payload.

pytest: eine wiederverwendbare Client-Fixture

Die folgende Fixture gibt jedem Test einen verbundenen Helper, der auf Bestätigungen wartet und Nachrichten in einer thread-sicheren Queue sammelt. Leg sie in conftest.py ab:

import os, queue, threading, time, uuid
import pytest
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN_ID = os.getenv("GITHUB_RUN_ID", uuid.uuid4().hex[:8])


class MqttProbe:
    def __init__(self):
        self.messages = queue.Queue()
        self.retained = set()
        self._connected = threading.Event()
        self._subacks = {}
        self._cond = threading.Condition()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"it-{uuid.uuid4().hex[:10]}")
        self.client.on_connect = lambda c, u, f, rc, p: self._connected.set()
        self.client.on_subscribe = self._on_subscribe
        self.client.on_message = lambda c, u, msg: self.messages.put(msg)
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        if PORT == 8883:
            self.client.tls_set()

    def _on_subscribe(self, client, userdata, mid, reason_codes, properties):
        with self._cond:
            self._subacks[mid] = reason_codes
            self._cond.notify_all()

    def connect(self, timeout=5):
        self.client.connect(HOST, PORT, keepalive=30)
        self.client.loop_start()
        assert self._connected.wait(timeout), "no CONNACK from broker"

    def subscribe(self, topic, qos=1, timeout=5):
        _, mid = self.client.subscribe(topic, qos)
        with self._cond:
            assert self._cond.wait_for(lambda: mid in self._subacks, timeout), f"no SUBACK for {topic}"
            code = self._subacks.pop(mid)[0]
        assert not code.is_failure, f"subscribe to {topic} refused: {code}"

    def publish(self, topic, payload, qos=1, retain=False, timeout=5):
        info = self.client.publish(topic, payload, qos=qos, retain=retain)
        info.wait_for_publish(timeout)
        assert info.is_published(), f"publish to {topic} not acknowledged"
        if retain:
            self.retained.add(topic)

    def expect(self, topic, timeout=5):
        deadline = time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                msg = self.messages.get(timeout=left)
            except queue.Empty:
                break
            if msg.topic == topic:
                return msg
        raise AssertionError(f"no message on {topic} within {timeout}s")

    def close(self):
        for topic in self.retained:  # clear retained state left by the test
            self.client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
        self.client.disconnect()
        self.client.loop_stop()


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


@pytest.fixture
def mqtt_probe():
    probe = MqttProbe()
    probe.connect()
    yield probe
    probe.close()

Ein Test für einen Anwendungsdienst – hier ein Thermostat, der auf Befehle hört und seinen Zustand meldet – liest sich dann wie eine Spezifikation:

import json
from conftest import HOST, PORT
from myapp.thermostat import ThermostatService


def test_setpoint_command_updates_state(mqtt_probe, prefix):
    service = ThermostatService(host=HOST, port=PORT, prefix=prefix)
    service.start()  # returns once the service has its own SUBACK
    try:
        mqtt_probe.subscribe(f"{prefix}/state")
        mqtt_probe.publish(f"{prefix}/cmd/setpoint", json.dumps({"value": 21.5}))

        msg = mqtt_probe.expect(f"{prefix}/state")
        assert json.loads(msg.payload)["setpoint"] == 21.5
    finally:
        service.stop()

Das prefix wird an die Anwendung übergeben, und das ist die nützlichste Designentscheidung für Testbarkeit: Wenn dein Dienst seine Topics aus einer konfigurierbaren Wurzel aufbaut, kann jeder Test in seinem eigenen Namensraum laufen. Beachte, dass service.start() blockiert, bis der Dienst abonniert hat – falls deine Anwendung kein solches Bereitschaftssignal anbietet, füge eines hinzu (ein Event, einen Callback oder eine Retained-Statusnachricht online). Sonst kann der Befehl des Tests ankommen, bevor der Dienst zuhört. pytest-Fixtures für MQTT zeigt weitere Fixture-Varianten, einschließlich Brokern mit Session-Scope.

Jest: Promise-Helper mit MQTT.js 5

MQTT.js 5 bringt connectAsync, subscribeAsync, publishAsync und endAsync mit, wodurch Tests linear werden. Der einzige Helper, den du brauchst, ist „auf die nächste Nachricht auf einem Topic warten":

const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

const URL = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const opts = () => ({
  clientId: `it-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  reconnectPeriod: 0,
});

function nextMessage(client, topic, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const onMessage = (t, payload, packet) => {
      if (t !== topic) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ payload: payload.toString(), retain: packet.retain });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on ${topic} within ${timeoutMs} ms`));
    }, timeoutMs);
    client.on('message', onMessage);
  });
}

let publisher;
let prefix;

beforeAll(async () => {
  publisher = await mqtt.connectAsync(URL, opts());
});
afterAll(() => publisher.endAsync());
beforeEach(() => {
  prefix = `testmqtt/it/${process.env.GITHUB_RUN_ID ?? 'local'}/${randomUUID().slice(0, 8)}`;
});

test('late subscribers receive the retained device config', async () => {
  const topic = `${prefix}/config`;
  await publisher.publishAsync(topic, JSON.stringify({ interval: 30 }), { qos: 1, retain: true });

  const device = await mqtt.connectAsync(URL, opts());
  try {
    const received = nextMessage(device, topic); // listen before subscribing
    await device.subscribeAsync(topic, { qos: 1 });
    const msg = await received;
    expect(msg.retain).toBe(true);
    expect(JSON.parse(msg.payload)).toEqual({ interval: 30 });
  } finally {
    await device.endAsync();
    await publisher.publishAsync(topic, '', { qos: 1, retain: true }); // clear retained
  }
}, 15000);

Zwei Details verhindern die häufigsten Jest-Probleme. Der Listener wird vor subscribeAsync registriert, weil der Broker Retained Messages direkt nach dem SUBACK schickt – manchmal bevor das erwartete Promise aufgelöst ist. Und jeder Client wird mit endAsync() geschlossen, damit Jest nicht mit „Jest did not exit one second after the test run has completed" hängen bleibt. Weitere Jest-Muster findest du in MQTT mit Jest testen.

Teste auch die Fehlerpfade

Der Happy Path ist der uninteressanteste Teil eines MQTT-Clients. In Integrationstests kannst du günstig das Verhalten beweisen, auf das es im Feld ankommt:

• Fehlerhafte Payloads. Publishe ungültiges JSON auf dein Befehls-Topic und prüfe, dass der Dienst einen Fehler loggt und weiterläuft, statt seine Netzwerkschleife abstürzen zu lassen.
• Autorisierung. Verbinde dich mit einem User, der auf ein Topic nicht publishen darf, und prüfe, dass dein Code die Ablehnung meldet. Mit MQTT 5 liefert ein QoS-1-Publish einen Reason Code im PUBACK; mit MQTT 3.1.1 verwirft der Broker die Nachricht meist oder schließt die Verbindung.
• Reconnects. Trenne den Client der Anwendung (oder starte einen lokalen Broker neu) und prüfe, dass er neu abonniert und den nächsten Befehl verarbeitet. MQTT-Reconnect-Tests zeigt, wie du das skriptest.

Jeder dieser Fälle braucht mit den Fixtures oben nur ein paar Zeilen und findet Bugs, die sonst erst in Produktionslogs auftauchen.

Checkliste gegen Flaky Tests

Wird eine Subscription abgelehnt, sagt dir der Reason Code im SUBACK, warum – 0x87 Not authorized bedeutet meist, dass die ACL des Test-Users das Präfix nicht abdeckt. Jeden Code entschlüsselst du mit der MQTT-Reason-Code-Suche.

Die Tests testen

Bevor du einer neuen Suite vertraust, beobachte sie von außen. Abonniere testmqtt/it/# im Online-MQTT-Client und starte die Tests: Du solltest sehen, wie das Präfix jedes Tests auftaucht, seine Nachrichten fließen und am Ende die Retained-Topics gelöscht werden (eine leere Retained Message). Lass die Suite dann zweimal parallel laufen – zum Beispiel pytest -n 4 mit pytest-xdist oder zwei CI-Jobs – und bestätige, dass sich nichts ändert.

Traffic der Integrationstests ansehen

Integrationstests gegen einen echten Broker sind der günstigste Weg zu beweisen, dass sich dein MQTT-Code so verhält wie in Produktion. Isoliere jeden Test über Topic und Client-ID, warte auf Bestätigungen statt zu schlafen, räume Retained-Zustand auf, und die Suite bleibt aus den richtigen Gründen grün. Wenn du sie bei jedem Push laufen lassen willst, geht es weiter mit MQTT in CI mit GitHub Actions testen.

Häufig gestellte Fragen