Hardware-in-the-Loop-Tests mit MQTT

Simulatoren und Mock-Broker finden Logikfehler, aber sie reproduzieren nie einen Brown-out mitten im TLS-Handshake oder einen Wi-Fi-Stack, der nach dem dritten Reconnect aufgibt. Hardware-in-the-Loop-Tests (HIL) mit MQTT schließen diese Lücke: Echte Boards mit echter Firmware verbinden sich mit einem Test-Broker, und ein Skript auf einem PC steuert sie, hört jede Nachricht mit und entscheidet über Erfolg oder Fehlschlag.

Dieser Guide zeigt, wie du einen kleinen HIL-Prüfstand aufbaust, ihn mit pytest und paho-mqtt 2.x automatisierst, Fehler wie Stromausfälle injizierst und die Ergebnisse in Reports verwandelst, die deine CI anzeigen kann.

So sieht ein MQTT-HIL-Prüfstand aus

Ein Prüfstand muss nicht teuer sein. Ein Regal mit ein paar Boards, ein USB-Hub und ein kleiner Linux-Rechner reichen für den Anfang.

Der Runner ist Dirigent und Beobachter zugleich: Er schaltet das Gerät ein, abonniert alles, was es publiziert, sendet Befehle wie ein Backend und prüft Timing und Inhalt.

Den Broker für den Prüfstand wählen

Der Broker entscheidet, wie realistisch dein Prüfstand ist. Ein lokaler Broker auf dem Runner nimmt das Internet aus der Gleichung – gut, um die Protokolllogik deterministisch zu testen. Ein entfernter Broker mit deinen Produktionseinstellungen testet zusätzlich DNS, TLS-Zertifikate, Authentifizierung und den echten Netzwerkpfad, und genau dort entstehen die meisten Fehler im Feld. Viele Teams nutzen beides: einen lokalen Broker für schnelles Firmware-Feedback und einen entfernten für die nächtlichen Läufe.

Egal, wofür du dich entscheidest: Verbinde dich einmal von Hand mit dem Online-MQTT-Client und abonniere die Topics des Geräts, bevor du irgendetwas automatisierst. Wenn du die Nachrichten des Geräts manuell nicht siehst, sieht sie auch kein Testskript.

Jeden Lauf isolieren

Ein Prüfstand wird über die Zeit geteilt: Der Lauf von gestern hat Retained Messages hinterlassen, der Prüfstand eines Kollegen publiziert auf ähnliche Topics, und zwei CI-Jobs laufen vielleicht gleichzeitig. Gib jedem Prüfstand und jedem Lauf eine eigene Topic-Wurzel, zum Beispiel testmqtt/hil/<bench>/<run-id>/..., und mache das Topic-Präfix der Firmware konfigurierbar (über die serielle Konsole, eine Provisioning-Datei oder ein Build-Flag), statt es fest einzucodieren. Das vollständige Muster beschreibt Test-Topics so benennen, dass Tests nie kollidieren.

Wenn sich das Präfix zur Laufzeit nicht ändern lässt, nutze mindestens ein festes Präfix pro Prüfstand und lösche vor jedem Test die Retained Messages darunter.

Das Harness: ein MQTT-Recorder

Das Herz des Prüfstands ist eine kleine Klasse, die das Präfix des Geräts abonniert, ein Protokoll jeder Nachricht führt und Tests mit Timeout auf eine bestimmte Nachricht warten lässt. Auf Events zu warten statt sleep() aufzurufen, ist mit Abstand die wirksamste Maßnahme gegen flaky Hardware-Tests.

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

class Recorder:
    def __init__(self, host, port, prefix):
        self.prefix = prefix
        self.inbox = queue.Queue()
        self.transcript = []
        self.ready = threading.Event()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"hil-runner-{uuid.uuid4().hex[:8]}")
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, codes, props: self.ready.set()
        self.client.on_message = self._on_message
        self.client.connect(host, port, keepalive=30)
        self.client.loop_start()
        if not self.ready.wait(10):
            raise RuntimeError("runner could not subscribe to the test broker")

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.subscribe(f"{self.prefix}/#", qos=1)

    def _on_message(self, client, userdata, msg):
        entry = (time.time(), msg.topic, msg.payload.decode(errors="replace"), msg.retain)
        self.transcript.append(entry)
        self.inbox.put(entry)

    def expect(self, suffix, timeout=10, match=lambda payload: True):
        want, deadline = f"{self.prefix}/{suffix}", time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                _, topic, payload, _ = self.inbox.get(timeout=left)
            except queue.Empty:
                break
            if topic == want and match(payload):
                return payload
        raise AssertionError(f"no matching message on {want} within {timeout}s")

    def publish(self, suffix, payload, qos=1):
        self.client.publish(f"{self.prefix}/{suffix}", payload, qos=qos).wait_for_publish(5)

    def drain(self):
        while not self.inbox.empty():
            self.inbox.get_nowait()

    def close(self):
        self.client.loop_stop()
        self.client.disconnect()

Strom und serielle Schnittstelle steuern

Jeder Test sollte mit einem Kaltstart beginnen, also braucht der Prüfstand eine Möglichkeit, den Strom abzuschalten. Für USB-versorgte Boards an einem Hub mit Port-Power-Switching genügt uhubctl; für netzbetriebene Geräte erledigt eine Relaisplatine oder eine schaltbare PDU denselben Job. Serielle Logs werden in einem Hintergrund-Thread aufgezeichnet, damit ein fehlgeschlagener Test genau das anhängen kann, was die Firmware ausgegeben hat.

import os, subprocess, threading, time
import pytest, serial
from hil.recorder import Recorder

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
BENCH = os.getenv("HIL_BENCH", "bench1")
RUN_ID = os.getenv("GITHUB_RUN_ID", time.strftime("%Y%m%dT%H%M%S"))

def set_power(on: bool):
    # Durch den Aufruf deines Relais oder deiner PDU ersetzen. Beispiel: Port 2 von Hub 1-1.
    subprocess.run(["uhubctl", "-l", "1-1", "-p", "2", "-a", "on" if on else "off"], check=True)

class Rig:
    def __init__(self, mqtt):
        self.mqtt = mqtt
    def power_off(self):
        set_power(False)
    def power_cycle(self, off_seconds=2):
        set_power(False); time.sleep(off_seconds)
        self.mqtt.drain(); set_power(True)

def capture(port, log, stop):
    while not stop.is_set():
        log.write(port.read(512))

@pytest.fixture
def rig(tmp_path):
    prefix = f"testmqtt/hil/{BENCH}/{RUN_ID}/dut"
    rec = Recorder(HOST, 1883, prefix)
    log = open(tmp_path / "serial.log", "wb")
    port = serial.Serial(os.getenv("HIL_SERIAL", "/dev/ttyUSB0"), 115200, timeout=0.2)
    stop = threading.Event()
    threading.Thread(target=capture, args=(port, log, stop), daemon=True).start()
    yield Rig(rec)
    stop.set(); time.sleep(0.3); port.close(); log.close()
    rec.close()

Die Firmware in diesem Beispiel publiziert nach dem Verbinden online auf <prefix>/status, setzt einen Last Will offline auf demselben Topic, lauscht auf <prefix>/cmd/# und antwortet auf <prefix>/ack.

Die Tests schreiben

import json

def test_boot_reports_online(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")

def test_command_round_trip(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")
    rig.mqtt.publish("cmd/led", json.dumps({"id": "t-1", "state": "on"}))
    ack = json.loads(rig.mqtt.expect("ack", timeout=5))
    assert ack["id"] == "t-1" and ack["ok"] is True

def test_power_loss_publishes_last_will(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")
    rig.power_off()
    # Keep-Alive 30 s -> Broker gibt nach 1,5 x 30 s auf
    rig.mqtt.expect("status", timeout=60, match=lambda p: p == "offline")

Der letzte Test beweist etwas, das kein Simulator wirklich zeigt: einen abrupten Stromausfall, einen Broker, der es erst über den Ablauf des Keep-Alive bemerkt, und einen Last Will, der beim Backend ankommt. Wann genau der Broker ihn sendet, steht in Last Will testen.

Fehler injizieren

Der Wert von HIL wächst mit jedem Fehlerfall, den du auf Knopfdruck auslösen kannst:

• Stromausfall mitten im Publish – schalte den Strom direkt nach einem Befehl ab und prüfe, ob sich das Gerät erholt und keinen veralteten Zustand meldet.
• Broker-Neustart – starte den lokalen Broker neu und prüfe, ob das Gerät sich mit Backoff neu verbindet, erneut abonniert und weiter publiziert. Die Szenarien behandelt MQTT-Reconnect-Tests.
• Netzausfall ohne Stromausfall – schalte den Wi-Fi-Access-Point oder den Switch-Port ab. Das Gerät läuft weiter, seine TCP-Verbindung stirbt still, und du erfährst, ob deine Keep-Alive- und Reconnect-Logik wirklich funktioniert.
• Falsche Zugangsdaten oder entzogene ACL – ändere das Passwort des Geräts auf dem Test-Broker und prüfe, ob die Firmware den CONNACK-Grund loggt, statt still in einer Schleife zu hängen. Der Reason-Code-Lookup hilft beim Entschlüsseln dessen, was das Gerät meldet.

Reporting

Starte die Suite mit pytest --junitxml=report.xml, damit jedes CI-System das Ergebnis pro Test anzeigen kann. Hänge bei einem Fehlschlag das serielle Log und das MQTT-Protokoll des Recorders an: Zusammen zeigen sie, ob das Gerät die Nachricht nie gesendet hat, sie auf das falsche Topic geschickt hat oder der Broker sie abgelehnt hat. Ein Self-hosted-CI-Runner am Prüfstand lässt jeden Firmware-Merge einen Hardware-Lauf auslösen; die Workflow-Seite ist dieselbe wie bei MQTT-Tests in GitHub Actions.

Ein guter HIL-Report enthält:

1. Firmware-Version und Git-Commit des Device under Test.
2. Name des Prüfstands, Broker-Endpoint und Run-ID (das Topic-Präfix).
3. Ergebnis und Dauer jedes Tests sowie die gemessenen Zeiten.
4. Serielles Log und MQTT-Protokoll für jeden fehlgeschlagenen Test.

Den Prüfstand ehrlich halten

• Flashe das Gerät zu Beginn jedes Laufs neu, damit der Test immer gegen den Build läuft, den du erwartest.
• Lösche vor dem Lauf die Retained Messages unter dem Präfix des Prüfstands, sonst lässt ein altes online einen kaputten Boot gut aussehen.
• Behalte ein bekannt gutes Firmware-Image; wenn es plötzlich fehlschlägt, ist der Prüfstand kaputt, nicht der Code.
• Stelle flaky Tests unter Quarantäne, statt sie blind zu wiederholen, und behebe die Ursache – meist ein fehlendes Warten oder ein geteiltes Topic.

Traffic deines Prüfstands im Browser ansehen

Hardware-in-the-Loop-Tests machen aus „auf meinem Schreibtisch lief es“ eine wiederholbare Prüfung, die bei jedem Build läuft. Fang mit einem Board, einem Stromschalter und drei Tests an – Boot, Befehls-Roundtrip und Stromausfall – und erweitere den Fehlerkatalog, wenn dir das Feld neue Fälle beibringt.

Häufig gestellte Fragen