To unit-test code that talks MQTT, don't mock the broker — mock the boundary your code owns. Wrap the client behind a small interface, replace it with a mock or an in-memory fake in unit tests, and keep a real broker for a thinner layer of integration tests. Your unit tests then run in milliseconds, never flake on the network, and still prove that the right topic and payload go out and that incoming messages are handled correctly.
This guide shows the three test doubles that work in practice — a mocked client, a hand-written fake and an embedded in-memory broker — with Python (paho-mqtt 2.x) and JavaScript (MQTT.js 5) examples, and explains where each one stops being useful.
Pick the right test double
| Double | What it is | Good for | Blind spots |
|---|---|---|---|
| Mock / stub client | unittest.mock, jest.fn() in place of the library client | Asserting calls: topic, payload, QoS, retain flag | Wildcards, QoS, retained messages, reconnects — nothing protocol-level is real |
| Hand-written fake | A tiny in-memory pub/sub class with the same interface | Testing flows across several components, topic routing | Only as correct as your fake; no network, no session state |
| Embedded broker | aedes (Node.js) started inside the test process | Real MQTT packets without Docker, very fast | Not your production broker; ACLs, limits and clustering differ |
| Real broker | Mosquitto/EMQX in a container, or a hosted test instance | Integration and end-to-end tests | Slower, needs setup and cleanup |
The first two are unit tests. The last two are integration tests, even when they feel fast. Keeping that distinction clear stops you from writing "unit tests" that secretly depend on a running broker — and from mocking away the very behavior you meant to verify.
Design the seam first
Mocking is painless when your business code never calls paho.mqtt.client or mqtt.connect directly. Inject the client (or a thin wrapper) instead:
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)Nothing here opens a socket, so nothing here needs a broker to be tested.
Python: mocking paho-mqtt with unittest.mock
Use create_autospec so the mock has the real Client signature — a typo like client.publsh(...) then fails the test instead of passing silently.
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()Two techniques do most of the work: assert on publish calls for outgoing traffic, and call the callback yourself with a fake message for incoming traffic. A SimpleNamespace with topic, payload, qos and retain is enough, because your handler only reads those attributes. If your code calls client.connect() at import time, use unittest.mock.patch("paho.mqtt.client.Client") instead — but treat that as a hint to move the connection into an injectable factory.
JavaScript: mocking MQTT.js with Jest
An MQTT.js client is an EventEmitter, so the most faithful mock is a real EventEmitter with jest.fn() methods attached. Your code registers on('message') handlers as usual, and the test emits events.
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)
);
});The message event passes (topic, payload, packet) with payload as a Buffer — keep that shape in the mock, or your test will pass with strings while production receives Buffers. For more patterns (fake timers for reconnect logic, async helpers) see testing MQTT with Jest.
A hand-written in-memory fake
When several components talk to each other through topics, asserting individual publish calls gets brittle. A small fake that actually routes messages is often clearer. paho-mqtt ships topic_matches_sub, so wildcard matching is correct for free:
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))Keep the fake tiny and test it against a real broker once (the same scenario through both), otherwise you end up testing the fake. You can double-check wildcard expectations with the topic matcher.
Node.js: aedes as an in-memory broker
Aedes is a real MQTT broker you can start inside a Jest or Mocha process. It speaks the actual protocol over a local TCP port, so your MQTT.js code runs unchanged — no mocks at all:
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()]);
});Listening on port 0 gives each test file its own free port, so parallel Jest workers don't collide. Python has no equally common embedded broker; most teams use a Mosquitto container instead — see Testcontainers for MQTT and pytest fixtures.
Where unit tests end
Mocks are great at verifying your decisions. They cannot verify the broker's. Move a test to the integration layer when it depends on:
- QoS and sessions — redelivery after a disconnect, persistent sessions, duplicate handling.
- Retained messages and Last Will — they are broker features; a fake only reproduces your assumptions.
- Authentication and ACLs — a mock never returns
Not authorized; a real broker does, sometimes silently. - Payload limits, topic aliases, MQTT 5 properties — limits are set by the broker configuration.
- TLS, WebSockets, reconnect timing — network-level behavior needs a network.
A healthy pyramid has many fast mocked tests, a few dozen integration tests against a real broker (see MQTT integration testing), and a handful of end-to-end checks against the environment you ship on.
Checking the mock against reality
Before trusting a mocked test, confirm once that its assumptions match a real broker. Publish the same message from the online MQTT client to broker.connectmq.com and compare: the payload type, the retain flag on late subscribers, and the exact topic strings. If they differ from what your mock returns, fix the mock first.
Compare your mock with a real broker
Mock the client to test your code quickly, use a small fake or an embedded broker when components interact, and leave protocol guarantees to integration tests against a real broker. That split keeps the suite fast without letting the tests lie about what MQTT actually does.
Frequently asked questions
Should I mock the MQTT client or run a real broker in tests?
Both, at different layers. Mock the client to unit-test your own logic (which topic, which payload, how you react to a message) in milliseconds. Run a real broker in integration tests to verify protocol behavior such as QoS, retained messages, ACLs and reconnects.
Is aedes a mock or a real MQTT broker?
Aedes is a real MQTT broker written in Node.js. Embedded in a test process with in-memory persistence it behaves like a fast, disposable broker, so tests that use it are small integration tests rather than pure unit tests.
How do I simulate an incoming MQTT message in a unit test?
Call the message handler directly with a fake message object that has the fields your code reads (topic, payload, qos, retain). For paho-mqtt that means calling on_message(client, userdata, msg); for MQTT.js, emit a 'message' event on a mocked EventEmitter client.
