Hardware-in-the-Loop Testing with MQTT

Simulators and mock brokers catch logic bugs, but they never reproduce a brown-out during a TLS handshake or a Wi-Fi stack that gives up after the third reconnect. Hardware-in-the-loop (HIL) testing with MQTT closes that gap: real boards running real firmware connect to a test broker, and a script on a PC drives them, listens to every message and decides pass or fail.

This guide shows how to build a small HIL rig, automate it with pytest and paho-mqtt 2.x, inject failures such as power loss, and turn the results into reports your CI can show.

What an MQTT HIL rig looks like

A rig does not need to be expensive. A shelf with a few boards, a USB hub and a small Linux machine is enough to start.

The runner is both the conductor and the observer: it switches the device on, subscribes to everything the device publishes, sends commands like a backend would, and checks timing and content.

Choose the broker for the rig

The broker decides how realistic your rig is. A local broker on the runner removes the internet from the equation, which is good for testing protocol logic deterministically. A remote broker with your production settings also tests DNS, TLS certificates, authentication and the real network path, which is where field failures usually come from. Many teams run both: a local broker for fast firmware feedback and a remote one for nightly runs.

Whatever you pick, connect to it once by hand with the online MQTT client and subscribe to the device's topics before automating anything. If you cannot see the device's messages manually, no test script will either.

Isolate every run

A rig is shared over time: yesterday's run left retained messages, a colleague's bench publishes to similar topics, and two CI jobs may run at once. Give every bench and every run its own topic root, for example testmqtt/hil/<bench>/<run-id>/..., and make the firmware's topic prefix configurable (over the serial console, a provisioning file or a build flag) instead of hard-coding it. The full pattern is described in naming test topics so tests never collide.

If the prefix cannot change at runtime, use at least one fixed prefix per bench and clear retained messages under it before every test.

The harness: an MQTT recorder

The heart of the rig is a small class that subscribes to the device's prefix, keeps a transcript of every message and lets tests wait for a specific message with a timeout. Waiting on events instead of sleep() is the single biggest fix for 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()

Power and serial control

Every test should start from a cold boot, so the rig needs a way to cut power. For USB-powered boards on a hub that supports per-port power switching, uhubctl is enough; for mains-powered devices, a relay board or a switchable PDU does the same job. Serial logs are captured in a background thread so a failing test can attach exactly what the firmware printed.

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):
    # Replace with your relay or PDU call. Example: port 2 of 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()

The firmware in this example publishes online to <prefix>/status after connecting, sets a Last Will of offline on the same topic, listens on <prefix>/cmd/# and answers on <prefix>/ack.

Write the tests

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 gives up after 1.5 x 30 s
    rig.mqtt.expect("status", timeout=60, match=lambda p: p == "offline")

The last test is something no simulator really proves: an abrupt power cut, a broker that only notices through keep-alive expiry, and a Last Will that arrives at the backend. See testing Last Will for the details of when the broker sends it.

Inject failures

The value of HIL grows with every failure mode you can trigger on demand:

• Power loss mid-publish — cut power right after a command and check that the device recovers and does not report a stale state.
• Broker restart — restart the local broker and verify the device reconnects with backoff, resubscribes and resumes publishing. The scenarios are covered in MQTT reconnect testing.
• Network loss without power loss — switch off the Wi-Fi access point or the switch port. The device keeps running, its TCP connection silently dies, and you find out whether your keep-alive and reconnect logic actually work.
• Bad credentials or a revoked ACL — rotate the device's password on the test broker and confirm the firmware logs the CONNACK reason instead of looping silently. The reason code lookup helps decode what the device reports.

Reporting

Run the suite with pytest --junitxml=report.xml so any CI system can display pass/fail per test. On failure, attach the serial log and the MQTT transcript from the recorder: together they tell you whether the device never sent the message, sent it to the wrong topic, or the broker refused it. A self-hosted CI runner connected to the rig lets every firmware merge trigger a hardware run — the workflow side is the same as in MQTT testing in GitHub Actions.

A good HIL report contains:

1. Firmware version and git commit of the device under test.
2. Bench name, broker endpoint and run ID (the topic prefix).
3. Per-test result and duration, plus the measured timings.
4. Serial log and MQTT transcript for every failed test.

Keep the rig honest

• Re-flash the device at the start of each run so the test always runs against the build you think it does.
• Clear retained messages under the bench prefix before the run, or a stale online can make a broken boot look fine.
• Keep one known-good firmware image; if it starts failing, the rig is broken, not the code.
• Quarantine flaky tests instead of retrying them blindly, and fix the cause — usually a missing wait or a shared topic.

Watch your rig's traffic in the browser

Hardware-in-the-loop testing turns "it worked on my desk" into a repeatable check that runs on every build. Start with one board, one power switch and three tests — boot, command round trip and power loss — and grow the failure catalogue as the field teaches you new ones.

Frequently asked questions