MQTT in Unit-Tests mocken: Fakes, Stubs, In-Memory-Broker

Wenn du Code, der MQTT spricht, mit Unit-Tests abdecken willst, gilt: Mocke nicht den Broker – mocke die Grenze, die dein Code besitzt. Kapsle den Client hinter einem kleinen Interface, ersetze ihn in Unit-Tests durch einen Mock oder ein In-Memory-Fake und behalte einen echten Broker für eine dünnere Schicht von Integrationstests. Dann laufen deine Unit-Tests in Millisekunden, scheitern nie am Netzwerk und beweisen trotzdem, dass das richtige Topic und der richtige Payload rausgehen und eingehende Nachrichten korrekt verarbeitet werden.

Dieser Guide zeigt die drei Test-Doubles, die sich in der Praxis bewähren – einen gemockten Client, ein handgeschriebenes Fake und einen eingebetteten In-Memory-Broker – mit Beispielen in Python (paho-mqtt 2.x) und JavaScript (MQTT.js 5), und erklärt, wo jedes davon an seine Grenzen kommt.

Das passende Test-Double wählen

Double	Was es ist	Gut für	Blinde Flecken
Mock- / Stub-Client	`unittest.mock`, `jest.fn()` anstelle des Library-Clients	Aufrufe prüfen: Topic, Payload, QoS, Retain-Flag	Wildcards, QoS, Retained Messages, Reconnects – nichts auf Protokollebene ist echt
Handgeschriebenes Fake	Eine winzige In-Memory-Pub/Sub-Klasse mit demselben Interface	Abläufe über mehrere Komponenten, Topic-Routing	Nur so korrekt wie dein Fake; kein Netzwerk, kein Session-State
Eingebetteter Broker	aedes (Node.js), im Testprozess gestartet	Echte MQTT-Pakete ohne Docker, sehr schnell	Nicht dein Produktions-Broker; ACLs, Limits und Clustering unterscheiden sich
Echter Broker	Mosquitto/EMQX im Container oder eine gehostete Testinstanz	Integrations- und End-to-End-Tests	Langsamer, braucht Setup und Aufräumen

Die ersten beiden sind Unit-Tests. Die letzten beiden sind Integrationstests, auch wenn sie sich schnell anfühlen. Wer diese Unterscheidung klar hält, schreibt keine „Unit-Tests“, die heimlich von einem laufenden Broker abhängen – und mockt nicht genau das Verhalten weg, das er eigentlich prüfen wollte.

Zuerst die Nahtstelle entwerfen

Mocken ist schmerzfrei, wenn dein Business-Code nie direkt paho.mqtt.client oder mqtt.connect aufruft. Injiziere stattdessen den Client (oder einen dünnen Wrapper):

telemetry.py

import json

class TelemetryPublisher:
    def __init__(self, client, device_id: str):
        self.client = client
        self.topic = f"testmqtt/devices/{device_id}/telemetry"

    def report(self, temperature: float) -> None:
        if not -40 <= temperature <= 125:
            raise ValueError("sensor out of range")
        payload = json.dumps({"t": round(temperature, 2)})
        self.client.publish(self.topic, payload, qos=1, retain=False)

def handle_command(client, userdata, msg):
    """paho-mqtt on_message callback: toggle a relay on 'on'/'off' commands."""
    command = msg.payload.decode().strip().lower()
    if command not in ("on", "off"):
        return
    userdata["relay"].set(command == "on")
    client.publish(msg.topic.replace("/cmd", "/state"), command, qos=1, retain=True)

Hier öffnet nichts einen Socket, also braucht hier auch nichts einen Broker, um getestet zu werden.

Python: paho-mqtt mit unittest.mock mocken

Verwende create_autospec, damit der Mock die echte Signatur von Client hat – ein Tippfehler wie client.publsh(...) lässt den Test dann fehlschlagen, statt still durchzulaufen.

test_telemetry.py

import json
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, create_autospec

import paho.mqtt.client as mqtt
from telemetry import TelemetryPublisher, handle_command

class TelemetryTests(unittest.TestCase):
    def setUp(self):
        self.client = create_autospec(mqtt.Client, instance=True)

    def test_report_publishes_json_at_qos1(self):
        TelemetryPublisher(self.client, "dev-42").report(21.456)
        self.client.publish.assert_called_once_with(
            "testmqtt/devices/dev-42/telemetry", json.dumps({"t": 21.46}), qos=1, retain=False
        )

    def test_out_of_range_is_not_published(self):
        with self.assertRaises(ValueError):
            TelemetryPublisher(self.client, "dev-42").report(500)
        self.client.publish.assert_not_called()

    def test_command_sets_relay_and_reports_state(self):
        relay = MagicMock()
        msg = SimpleNamespace(topic="testmqtt/devices/dev-42/cmd", payload=b" ON ", qos=1, retain=False)
        handle_command(self.client, {"relay": relay}, msg)
        relay.set.assert_called_once_with(True)
        self.client.publish.assert_called_once_with(
            "testmqtt/devices/dev-42/state", "on", qos=1, retain=True
        )

if __name__ == "__main__":
    unittest.main()

Zwei Techniken erledigen fast alles: Prüfe die publish-Aufrufe für ausgehenden Traffic und rufe den Callback selbst auf mit einer Fake-Nachricht für eingehenden Traffic. Ein SimpleNamespace mit topic, payload, qos und retain genügt, weil dein Handler nur diese Attribute liest. Wenn dein Code client.connect() schon beim Import aufruft, nutze unittest.mock.patch("paho.mqtt.client.Client") – betrachte das aber als Hinweis, den Verbindungsaufbau in eine injizierbare Factory zu verschieben.

JavaScript: MQTT.js mit Jest mocken

Ein MQTT.js-Client ist ein EventEmitter. Der realistischste Mock ist deshalb ein echter EventEmitter mit angehängten jest.fn()-Methoden. Dein Code registriert wie gewohnt on('message')-Handler, und der Test löst die Events aus.

alarm.test.js

import { EventEmitter } from 'node:events';
import { connect } from 'mqtt';
import { startAlarm } from './alarm.js'; // subscribes to sensors, publishes alarms

jest.mock('mqtt', () => ({ connect: jest.fn() }));

function fakeClient() {
  const client = new EventEmitter();
  client.subscribe = jest.fn((topic, opts, cb) => cb?.(null, [{ topic, qos: opts?.qos ?? 0 }]));
  client.publish = jest.fn((topic, payload, opts, cb) => cb?.());
  client.end = jest.fn();
  return client;
}

test('publishes an alarm when temperature exceeds the threshold', () => {
  const client = fakeClient();
  connect.mockReturnValue(client);

  startAlarm({ url: 'mqtts://broker.connectmq.com:8883', threshold: 30 });
  client.emit('connect');
  expect(client.subscribe).toHaveBeenCalledWith('testmqtt/sensors/+/temp', { qos: 1 }, expect.any(Function));

  client.emit('message', 'testmqtt/sensors/kitchen/temp', Buffer.from('31.5'), { retain: false });
  expect(client.publish).toHaveBeenCalledWith(
    'testmqtt/alarms/kitchen', expect.stringContaining('"temp":31.5'), { qos: 1 }, expect.any(Function)
  );
});

Das message-Event liefert (topic, payload, packet) mit payload als Buffer – behalte diese Form im Mock bei, sonst besteht dein Test mit Strings, während die Produktion Buffers erhält. Weitere Muster (Fake-Timer für Reconnect-Logik, Async-Helper) findest du unter MQTT mit Jest testen.

Ein handgeschriebenes In-Memory-Fake

Wenn mehrere Komponenten über Topics miteinander sprechen, werden Assertions auf einzelne publish-Aufrufe schnell brüchig. Ein kleines Fake, das Nachrichten tatsächlich routet, ist oft verständlicher. paho-mqtt bringt topic_matches_sub mit, sodass das Wildcard-Matching ohne Aufwand korrekt ist:

fake_mqtt.py

from types import SimpleNamespace
from paho.mqtt.client import topic_matches_sub

class FakeMqtt:
    """In-memory pub/sub with the subset of the paho API our code uses."""
    def __init__(self):
        self.subscriptions, self.retained, self.published = [], {}, []
        self.on_message = None

    def subscribe(self, topic, qos=0):
        self.subscriptions.append(topic)
        for t, payload in self.retained.items():
            if topic_matches_sub(topic, t):
                self._deliver(t, payload, retain=True)

    def publish(self, topic, payload, qos=0, retain=False):
        data = payload.encode() if isinstance(payload, str) else payload
        self.published.append((topic, data, qos, retain))
        if retain and data:
            self.retained[topic] = data
        elif retain:
            self.retained.pop(topic, None)   # empty retained payload clears
        if any(topic_matches_sub(s, topic) for s in self.subscriptions):
            self._deliver(topic, data, retain=False)

    def _deliver(self, topic, payload, retain):
        if self.on_message:
            self.on_message(self, None, SimpleNamespace(topic=topic, payload=payload, qos=0, retain=retain))

Halte das Fake winzig und teste es einmal gegen einen echten Broker (dasselbe Szenario über beide Wege), sonst testest du am Ende nur das Fake. Deine Wildcard-Erwartungen kannst du mit dem Topic Matcher gegenprüfen.

Node.js: aedes als In-Memory-Broker

Aedes ist ein echter MQTT-Broker, den du in einem Jest- oder Mocha-Prozess starten kannst. Er spricht das echte Protokoll über einen lokalen TCP-Port, dein MQTT.js-Code läuft also unverändert – ganz ohne Mocks:

broker.test.js

import { createServer } from 'node:net';
import { Aedes } from 'aedes';
import mqtt from 'mqtt';

let broker, server, url;

beforeAll(async () => {
  broker = await Aedes.createBroker();      // aedes 0.x: require('aedes')()
  server = createServer(broker.handle);
  await new Promise((resolve) => server.listen(0, resolve)); // random free port
  url = `mqtt://127.0.0.1:${server.address().port}`;
});

afterAll(async () => {
  await new Promise((resolve) => server.close(resolve));
  await new Promise((resolve) => broker.close(resolve));
});

test('retained state is delivered to late subscribers', async () => {
  const pub = await mqtt.connectAsync(url);
  await pub.publishAsync('testmqtt/unit/state', 'ready', { qos: 1, retain: true });

  const sub = await mqtt.connectAsync(url);
  const received = new Promise((resolve) => sub.on('message', (t, p, pkt) => resolve([t, p.toString(), pkt.retain])));
  await sub.subscribeAsync('testmqtt/unit/#', { qos: 1 });

  expect(await received).toEqual(['testmqtt/unit/state', 'ready', true]);
  await Promise.all([pub.endAsync(), sub.endAsync()]);
});

Mit Port 0 bekommt jede Testdatei ihren eigenen freien Port, sodass parallele Jest-Worker nicht kollidieren. Für Python gibt es keinen ähnlich verbreiteten eingebetteten Broker; die meisten Teams nehmen stattdessen einen Mosquitto-Container – siehe Testcontainers für MQTT und pytest-Fixtures.

Wo Unit-Tests enden

Mocks sind hervorragend darin, deine Entscheidungen zu prüfen. Die des Brokers können sie nicht prüfen. Verschiebe einen Test in die Integrationsschicht, wenn er abhängt von:

QoS und Sessions – erneute Zustellung nach einem Disconnect, Persistent Sessions, Umgang mit Duplikaten.
Retained Messages und Last Will – das sind Broker-Features; ein Fake bildet nur deine Annahmen nach.
Authentifizierung und ACLs – ein Mock liefert nie Not authorized; ein echter Broker schon, manchmal stillschweigend.
Payload-Limits, Topic Aliases, MQTT-5-Properties – Limits legt die Broker-Konfiguration fest.
TLS, WebSockets, Reconnect-Timing – Netzwerkverhalten braucht ein Netzwerk.

Eine gesunde Testpyramide hat viele schnelle Tests mit Mocks, einige Dutzend Integrationstests gegen einen echten Broker (siehe MQTT-Integrationstests) und eine Handvoll End-to-End-Checks gegen die Umgebung, in die du auslieferst.

Den Mock mit der Realität abgleichen

Bevor du einem Test mit Mocks vertraust, prüfe einmal, ob seine Annahmen zu einem echten Broker passen. Veröffentliche dieselbe Nachricht mit dem Online-MQTT-Client auf broker.connectmq.com und vergleiche: den Typ des Payloads, das Retain-Flag bei späten Subscribern und die exakten Topic-Strings. Weichen sie von dem ab, was dein Mock liefert, korrigiere zuerst den Mock.

Vergleiche deinen Mock mit einem echten Broker

Mocke den Client, um deinen Code schnell zu testen, nimm ein kleines Fake oder einen eingebetteten Broker, wenn Komponenten zusammenspielen, und überlasse Protokollgarantien den Integrationstests gegen einen echten Broker. Diese Aufteilung hält die Suite schnell, ohne dass die Tests darüber lügen, was MQTT tatsächlich tut.

Häufig gestellte Fragen

Soll ich den MQTT-Client mocken oder in Tests einen echten Broker verwenden?

Beides, auf unterschiedlichen Ebenen. Mocke den Client, um deine eigene Logik (welches Topic, welcher Payload, wie du auf eine Nachricht reagierst) in Millisekunden zu testen. Nutze einen echten Broker in Integrationstests, um Protokollverhalten wie QoS, Retained Messages, ACLs und Reconnects zu prüfen.

Ist aedes ein Mock oder ein echter MQTT-Broker?

Aedes ist ein echter MQTT-Broker in Node.js. Eingebettet in den Testprozess mit In-Memory-Persistenz verhält er sich wie ein schneller Wegwerf-Broker. Tests damit sind daher kleine Integrationstests und keine reinen Unit-Tests.

Wie simuliere ich eine eingehende MQTT-Nachricht in einem Unit-Test?

Rufe den Message-Handler direkt mit einem Fake-Nachrichtenobjekt auf, das die Felder hat, die dein Code liest (topic, payload, qos, retain). Bei paho-mqtt heißt das on_message(client, userdata, msg) aufrufen, bei MQTT.js ein 'message'-Event auf einem gemockten EventEmitter-Client auslösen.

MQTT in Unit-Tests mocken: Fakes, Stubs und In-Memory-Broker