End-to-End-Tests für IoT über MQTT: vom Gerät bis zur UI

Unit-Tests beweisen, dass jeder Dienst funktioniert. Kunden interessiert, ob der Messwert eines Sensors in ihrem Dashboard ankommt und ob der Button, den sie drücken, das Gerät erreicht. End-to-End-Tests für IoT-Systeme über MQTT prüfen genau das: Ein simuliertes Gerät publisht über den echten Broker, und der Test prüft jeden Hop, vom Ingestion-Service und der Datenbank bis zur API und UI. Danach schickt er einen Befehl zurück und prüft, ob das Gerät ihn erhalten hat.

Dieser Guide zeigt, wie du einen solchen Test aufbaust, mit lauffähigem pytest- und Playwright-Code, und wie er schnell und zuverlässig genug bleibt, um bei jedem Merge zu laufen.

Der Weg eines Ereignisses

Ein typisches MQTT-basiertes IoT-System hat zwei Richtungen, und eine End-to-End-Suite sollte beide abdecken:

Richtung	Hops	Was brechen kann
Uplink (Telemetrie)	Gerät → Broker → Ingestion-Service → Datenbank → API → UI	Topic- oder Payload-Änderungen, ACLs, Parsing, Zeitzonen, Caching
Downlink (Befehle)	UI → API → Backend → Broker → Gerät → Ack	Falscher Topic, fehlende Berechtigungen, QoS, Offline-Geräte
Präsenz	Geräte-Disconnect → Last Will → Backend → Status in der UI	Keep-Alive-Einstellungen, LWT-Topic, veralteter Zustand

Jeder Hop hat bereits eigene Tests (siehe MQTT-Integrationstests für die Grenze zum Broker). Die End-to-End-Suite ist klein, eine Handvoll Szenarien, aber jedes davon überquert alle Grenzen auf einmal.

Einen Simulator bauen, der sich wie die Firmware verhält

Der Simulator ist das wichtigste Teil. Er muss dasselbe Client-ID-Schema, dieselben Topics, QoS, dasselbe Payload-Format und denselben Last Will wie die echte Firmware verwenden, sonst testest du ein anderes System. Eine minimale Python-Version mit paho-mqtt 2.x:

sim_device.py

import json, queue, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

class SimDevice:
    def __init__(self, device_id, host, user, password, prefix="testmqtt/e2e"):
        self.id = device_id
        self.base = f"{prefix}/{device_id}"
        self.commands = queue.Queue()
        self.subscribed = queue.Queue()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=device_id,
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(user, password)
        self.client.tls_set()
        self.client.will_set(f"{self.base}/status", "offline", qos=1, retain=True)
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, rc, p: self.subscribed.put(mid)
        self.client.on_message = self._on_message
        self.host = host
        self.crashed = False

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.publish(f"{self.base}/status", "online", qos=1, retain=True)
        client.subscribe(f"{self.base}/cmd", qos=1)

    def _on_message(self, client, userdata, msg):
        cmd = json.loads(msg.payload)
        self.commands.put(cmd)
        client.publish(f"{self.base}/cmd/ack", json.dumps({"id": cmd["id"], "ok": True}), qos=1)

    def start(self):
        self.client.connect(self.host, 8883, keepalive=30)
        self.client.loop_start()
        self.subscribed.get(timeout=10)        # SUBACK erhalten: bereit für Befehle

    def send_reading(self, temp_c):
        payload = {"schema": "telemetry/v1", "device": self.id, "temp_c": temp_c,
                   "ts": datetime.now(timezone.utc).isoformat()}
        self.client.publish(f"{self.base}/telemetry", json.dumps(payload), qos=1).wait_for_publish()

    def crash(self):
        """TCP-Verbindung ohne DISCONNECT trennen, damit der Broker den Last Will sendet."""
        self.crashed = True
        self.client.loop_stop()
        self.client.socket().close()

    def stop(self):
        if not self.crashed:
            # Retained-Status löschen, damit der Broker sauber bleibt
            self.client.publish(f"{self.base}/status", "", qos=1, retain=True).wait_for_publish()
            self.client.disconnect()
        self.client.loop_stop()

Wenn du schon einen Flottensimulator hast, verwende ihn weiter; unter einen MQTT-Gerätesimulator bauen erfährst du, wie du dieselbe Idee auf viele Geräte skalierst. Gegen auseinanderlaufende Payloads zwischen Firmware und Simulator helfen ein gemeinsames JSON Schema und Contract-Tests.

Auf Bedingungen warten, nie auf Sleep

Asynchrone Pipelines haben schwankende Latenz. Ein festes time.sleep(2) ist auf einem ausgelasteten CI-Runner zu kurz oder verschwendet überall sonst Zeit. Frage eine Bedingung ab, bis sie erfüllt ist oder eine Frist abläuft:

eventually.py

import time

def eventually(check, timeout=15.0, interval=0.25):
    """check() aufrufen, bis kein AssertionError mehr kommt, sonst nach Timeout erneut werfen."""
    deadline = time.monotonic() + timeout
    while True:
        try:
            return check()
        except AssertionError:
            if time.monotonic() > deadline:
                raise
            time.sleep(interval)

Der Uplink-Test: vom Gerät in die Datenbank und die API

Jeder Lauf bekommt eine eindeutige Geräte-ID, damit parallele Läufe und Reste früherer Läufe sich nie in die Quere kommen:

test_e2e_uplink.py

import os, uuid
import httpx, psycopg, pytest
from eventually import eventually
from sim_device import SimDevice

HOST = "broker.connectmq.com"
API = os.environ.get("API_URL", "http://localhost:8000")
DSN = os.environ.get("DATABASE_URL", "postgresql://app:app@localhost:5432/app")

@pytest.fixture
def device():
    dev = SimDevice(f"e2e-{uuid.uuid4().hex[:8]}", HOST,
                    os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    dev.start()
    yield dev
    dev.stop()
    with psycopg.connect(DSN) as db:
        db.execute("DELETE FROM readings WHERE device_id = %s", (dev.id,))

def test_reading_reaches_database_and_api(device):
    device.send_reading(21.5)

    def stored():
        with psycopg.connect(DSN) as db:
            row = db.execute(
                "SELECT temp_c FROM readings WHERE device_id = %s ORDER BY ts DESC LIMIT 1",
                (device.id,),
            ).fetchone()
        assert row is not None, "reading not stored yet"
        assert row[0] == pytest.approx(21.5)
    eventually(stored)

    def served():
        r = httpx.get(f"{API}/devices/{device.id}/latest", timeout=5)
        assert r.status_code == 200
        assert r.json()["temp_c"] == pytest.approx(21.5)
    eventually(served)

Die Datenbankprüfung zeigt dir, dass die Ingestion funktioniert; die API-Prüfung, dass Abfragen, Serialisierung und Caching darüber funktionieren. Schlägt der Test fehl, benennt die erste fehlgeschlagene Assertion den kaputten Hop. Genau das ist der Hauptnutzen davon, jede Schicht zu prüfen statt nur die UI. Der Guide MQTT nach PostgreSQL zeigt einen passenden Ingestion-Service.

Der UI- und Downlink-Test mit Playwright

Der Browsertest prüft, was der Nutzer sieht, und treibt den Befehlsweg zurück zum Gerät. Mit Playwright für Python:

test_e2e_ui.py

import os
from playwright.sync_api import Page, expect
from eventually import eventually

APP = os.environ.get("APP_URL", "http://localhost:3000")

def test_dashboard_shows_reading_and_sends_command(page: Page, device):
    device.send_reading(23.0)

    page.goto(f"{APP}/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)
    expect(page.get_by_test_id("temp-value")).to_have_text("23.0 °C", timeout=15_000)

    page.get_by_role("button", name="Reboot").click()

    cmd = device.commands.get(timeout=15)        # der Simulator hat ihn über MQTT empfangen
    assert cmd["action"] == "reboot"
    expect(page.get_by_test_id("last-command")).to_contain_text("acknowledged", timeout=15_000)

Hier wird die Fixture device wiederverwendet (verschiebe sie nach conftest.py). Playwrights expect wiederholt bereits bis zum Timeout, im Browser ist also kein zusätzliches Polling nötig. Verwende data-testid-Attribute statt CSS-Klassen; sie überleben Redesigns.

Präsenz: den Last-Will-Pfad testen

Offline-Erkennung ist ein Feature, auf das sich Nutzer verlassen und das selten getestet wird. Simuliere einen Absturz, keinen sauberen Disconnect, denn ein normales DISCONNECT unterdrückt den Last Will:

test_e2e_presence.py

from playwright.sync_api import Page, expect

def test_crashed_device_shows_offline(page: Page, device):
    page.goto(f"http://localhost:3000/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)

    device.crash()

    # ein geschlossener Socket löst den Will sofort aus; eine still tote Verbindung bis zu 1,5 x Keepalive
    expect(page.get_by_test_id("device-status")).to_have_text("offline", timeout=60_000)

Ein geschlossener TCP-Socket löst den Will sofort aus. Ein Gerät, das Strom oder Empfang verliert, sendet dagegen nichts, und der Broker merkt es erst nach etwa dem 1,5-Fachen des Keep-Alive-Intervalls. Halte den Timeout also darüber. Die Timing-Regeln findest du unter Last Will testen. Wenn du bei einem lokalen Lauf testmqtt/e2e/# im Online-MQTT-Client beobachtest, siehst du Status-, Telemetrie-, Befehls- und Ack-Nachrichten in der richtigen Reihenfolge. Das ist der schnellste Weg, ein fehlschlagendes Szenario zu debuggen.

Traffic des End-to-End-Tests ansehen

Umgebung und CI

Starte Backend, Datenbank und UI mit Docker Compose und richte sie, genau wie den Simulator, auf denselben Broker aus:

Suite lokal ausführen

export MQTT_USER=username MQTT_PASS=password
docker compose up -d --wait          # api, ingest, postgres, web
pip install pytest pytest-playwright paho-mqtt httpx "psycopg[binary]"
playwright install chromium
pytest -q tests/e2e --maxfail=1

In der CI funktionieren dieselben Befehle in einem Job; siehe MQTT-Tests in GitHub Actions. Halte die Broker-Konfiguration (TLS, ACLs, MQTT-Version) identisch zur Produktion, denn Berechtigungs- und TLS-Fehler sind genau das, was End-to-End-Tests aufdecken sollen.

Die Suite zuverlässig halten

Eindeutige IDs pro Lauf. Geräte-ID, Topic-Präfix und Datenbankzeilen leiten sich alle von einem zufälligen Suffix ab.
Auf das SUBACK warten. start() des Simulators blockiert, bis das Befehls-Abo bestätigt ist; ohne das kann der erste Befehl ankommen, bevor jemand zuhört.
Retained Messages aufräumen. Eine leere Retained Payload auf dem Status-Topic in stop() hält den Broker sauber; siehe Retained Messages.
Wenige Szenarien, volle Tiefe. Fünf End-to-End-Szenarien, die jeden Hop überqueren, sind mehr wert als fünfzig, die Unit-Tests duplizieren.
Mit dem Namen des Hops scheitern. Schreibe Assertion-Meldungen wie "reading not stored yet", damit der Report zeigt, wo die Kette gerissen ist.

Eine End-to-End-Suite für ein MQTT-System besteht aus einem Simulator, einem Polling-Helper und einigen Tests, die ein Ereignis vom Gerät bis auf den Bildschirm und zurück verfolgen. Sie läuft in Minuten, findet die Integrationsfehler, die Unit-Tests nicht sehen können, und liefert vor jedem Release eine konkrete Antwort auf die Frage „Funktioniert es?“.

Häufig gestellte Fragen

Was ist der Unterschied zwischen Integrations- und End-to-End-Tests im IoT?

Ein Integrationstest prüft eine Grenze, zum Beispiel ob dein Ingestion-Service eine Nachricht vom Broker erhält. Ein End-to-End-Test verfolgt ein einzelnes Ereignis durch alle Komponenten: Gerät, Broker, Backend, Datenbank, API und UI, dazu den Befehlsweg zurück zum Gerät.

Brauche ich echte Hardware für IoT-End-to-End-Tests?

Für die meisten nicht. Ein Software-Gerätesimulator, der dieselben MQTT-Topics und Payloads wie die Firmware verwendet, deckt die Cloud-Seite zuverlässig ab und läuft in der CI. Behalte eine kleine Hardware-in-the-Loop-Suite für Funk, Energie und firmwarespezifisches Verhalten.

Wie verhindere ich, dass MQTT-End-to-End-Tests flaky werden?

Gib jedem Lauf eine eindeutige Geräte-ID und ein eindeutiges Topic-Präfix, warte mit Timeout auf Bedingungen statt mit festen Sleeps, warte vor dem Publish auf das SUBACK und räume die Zeilen und Retained Messages auf, die der Test angelegt hat.

End-to-End-Tests für IoT-Systeme über MQTT