MQTT Integration Testing: Test Your App Against a Real Broker

MQTT integration tests run your application against a real broker: your code connects, subscribes and publishes, and the test asserts on the messages that actually cross the wire. They catch what unit tests with mocks cannot — wrong topic filters, missing QoS acknowledgements, retained state, ACL denials and reconnect bugs.

This guide shows working patterns for pytest (paho-mqtt 2.x) and Jest (MQTT.js 5), and the five rules that keep these tests fast and free of flakiness.

Unit, integration or end-to-end?

Keep the mock-based tests for pure logic — mock brokers for unit tests covers that layer. Everything that depends on how a broker behaves (wildcards, retained messages, QoS 1 redelivery, session expiry) belongs in integration tests against a real one.

Choosing the test broker

You have three practical options:

• Local Mosquitto (installed or via Docker) — fast and private, but every developer and CI runner must start it.
• Ephemeral container per test run — fully isolated and disposable; see MQTT tests with Testcontainers.
• Hosted broker — no setup, real TLS and authentication, and the same endpoint from laptops, CI and staging devices.

Your test code should not care which one it talks to. Read the host, port and credentials from environment variables and default to something that works out of the box:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
# MQTT_PORT=8883 for TLS; set MQTT_USER / MQTT_PASS for a private broker

The five rules for stable MQTT tests

1. Unique topic prefix per test. Build every topic from testmqtt/it/<run-id>/<random-id>. Parallel test workers, CI jobs and developers then never see each other's messages.
2. Unique client ID per connection. Two connections with the same ID disconnect each other — a classic source of tests that "randomly" lose messages.
3. Wait for events, never sleep. Wait for CONNACK before subscribing, wait for SUBACK before publishing, and wait for a message with a timeout. A sleep(1) is either too long (slow suite) or too short (flaky suite).
4. Clean up what outlives the test. Retained messages and persistent sessions survive disconnects. Clear retained topics with an empty retained publish and use clean sessions unless you are testing sessions.
5. Assert on content, not on timing or global order. MQTT only guarantees order per topic and QoS from one publisher. Correlate messages with an ID in the payload.

pytest: a reusable broker client fixture

The fixture below gives each test a connected helper that waits for acknowledgements and collects messages in a thread-safe queue. Put it in conftest.py:

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

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN_ID = os.getenv("GITHUB_RUN_ID", uuid.uuid4().hex[:8])


class MqttProbe:
    def __init__(self):
        self.messages = queue.Queue()
        self.retained = set()
        self._connected = threading.Event()
        self._subacks = {}
        self._cond = threading.Condition()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"it-{uuid.uuid4().hex[:10]}")
        self.client.on_connect = lambda c, u, f, rc, p: self._connected.set()
        self.client.on_subscribe = self._on_subscribe
        self.client.on_message = lambda c, u, msg: self.messages.put(msg)
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        if PORT == 8883:
            self.client.tls_set()

    def _on_subscribe(self, client, userdata, mid, reason_codes, properties):
        with self._cond:
            self._subacks[mid] = reason_codes
            self._cond.notify_all()

    def connect(self, timeout=5):
        self.client.connect(HOST, PORT, keepalive=30)
        self.client.loop_start()
        assert self._connected.wait(timeout), "no CONNACK from broker"

    def subscribe(self, topic, qos=1, timeout=5):
        _, mid = self.client.subscribe(topic, qos)
        with self._cond:
            assert self._cond.wait_for(lambda: mid in self._subacks, timeout), f"no SUBACK for {topic}"
            code = self._subacks.pop(mid)[0]
        assert not code.is_failure, f"subscribe to {topic} refused: {code}"

    def publish(self, topic, payload, qos=1, retain=False, timeout=5):
        info = self.client.publish(topic, payload, qos=qos, retain=retain)
        info.wait_for_publish(timeout)
        assert info.is_published(), f"publish to {topic} not acknowledged"
        if retain:
            self.retained.add(topic)

    def expect(self, topic, timeout=5):
        deadline = time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                msg = self.messages.get(timeout=left)
            except queue.Empty:
                break
            if msg.topic == topic:
                return msg
        raise AssertionError(f"no message on {topic} within {timeout}s")

    def close(self):
        for topic in self.retained:  # clear retained state left by the test
            self.client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
        self.client.disconnect()
        self.client.loop_stop()


@pytest.fixture
def prefix():
    return f"testmqtt/it/{RUN_ID}/{uuid.uuid4().hex[:8]}"


@pytest.fixture
def mqtt_probe():
    probe = MqttProbe()
    probe.connect()
    yield probe
    probe.close()

A test for an application service — here a thermostat that listens for commands and reports its state — then reads like a specification:

import json
from conftest import HOST, PORT
from myapp.thermostat import ThermostatService


def test_setpoint_command_updates_state(mqtt_probe, prefix):
    service = ThermostatService(host=HOST, port=PORT, prefix=prefix)
    service.start()  # returns once the service has its own SUBACK
    try:
        mqtt_probe.subscribe(f"{prefix}/state")
        mqtt_probe.publish(f"{prefix}/cmd/setpoint", json.dumps({"value": 21.5}))

        msg = mqtt_probe.expect(f"{prefix}/state")
        assert json.loads(msg.payload)["setpoint"] == 21.5
    finally:
        service.stop()

The prefix is passed into the application, which is the single most useful design decision for testability: if your service builds topics from a configurable root, every test can run in its own namespace. Note that service.start() blocks until the service is subscribed — if your application does not expose such a readiness signal, add one (an event, a callback or a retained online status message). Otherwise the test's command can arrive before the service listens. pytest fixtures for MQTT shows more fixture variants, including session-scoped brokers.

Jest: promise helpers with MQTT.js 5

MQTT.js 5 ships connectAsync, subscribeAsync, publishAsync and endAsync, which make tests linear. The only helper you need is "wait for the next message on a topic":

const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

const URL = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const opts = () => ({
  clientId: `it-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  reconnectPeriod: 0,
});

function nextMessage(client, topic, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const onMessage = (t, payload, packet) => {
      if (t !== topic) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ payload: payload.toString(), retain: packet.retain });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on ${topic} within ${timeoutMs} ms`));
    }, timeoutMs);
    client.on('message', onMessage);
  });
}

let publisher;
let prefix;

beforeAll(async () => {
  publisher = await mqtt.connectAsync(URL, opts());
});
afterAll(() => publisher.endAsync());
beforeEach(() => {
  prefix = `testmqtt/it/${process.env.GITHUB_RUN_ID ?? 'local'}/${randomUUID().slice(0, 8)}`;
});

test('late subscribers receive the retained device config', async () => {
  const topic = `${prefix}/config`;
  await publisher.publishAsync(topic, JSON.stringify({ interval: 30 }), { qos: 1, retain: true });

  const device = await mqtt.connectAsync(URL, opts());
  try {
    const received = nextMessage(device, topic); // listen before subscribing
    await device.subscribeAsync(topic, { qos: 1 });
    const msg = await received;
    expect(msg.retain).toBe(true);
    expect(JSON.parse(msg.payload)).toEqual({ interval: 30 });
  } finally {
    await device.endAsync();
    await publisher.publishAsync(topic, '', { qos: 1, retain: true }); // clear retained
  }
}, 15000);

Two details prevent the most common Jest problems. The listener is registered before subscribeAsync, because the broker sends retained messages immediately after the SUBACK — sometimes before the awaited promise resolves. And every client is closed with endAsync(), so Jest does not hang with "Jest did not exit one second after the test run has completed". More Jest patterns are in testing MQTT with Jest.

Test the failure paths too

The happy path is the least interesting part of an MQTT client. Integration tests are where you can cheaply prove the behaviour that matters in the field:

• Malformed payloads. Publish invalid JSON to your command topic and assert that the service logs an error and keeps running, instead of crashing its network loop.
• Authorization. Connect with a user that is not allowed to publish to a topic and assert that your code reports the refusal. With MQTT 5 a QoS 1 publish returns a PUBACK reason code; with MQTT 3.1.1 the broker usually drops the message or closes the connection.
• Reconnects. Disconnect the application's client (or restart a local broker) and assert that it resubscribes and processes the next command. MQTT reconnect testing shows how to script this.

Each of these takes a few lines with the fixtures above and catches bugs that otherwise surface only in production logs.

Flaky test checklist

If a subscription is refused, the SUBACK reason code tells you why — 0x87 Not authorized usually means the test user's ACL does not cover the prefix. Decode any code with the MQTT reason code lookup.

Testing the tests

Before trusting a new suite, watch it from the outside. Subscribe to testmqtt/it/# in the online MQTT client and run the tests: you should see each test's prefix appear, its messages flow, and retained topics cleared (an empty retained message) at the end. Then run the suite twice in parallel — for example pytest -n 4 with pytest-xdist, or two CI jobs — and confirm nothing changes.

Watch integration test traffic

Integration tests against a real broker are the cheapest way to prove that your MQTT code behaves the way production will. Isolate every test by topic and client ID, wait for acknowledgements instead of sleeping, clean up retained state, and the suite will stay green for the right reasons. When you are ready to run it on every push, continue with testing MQTT in CI with GitHub Actions.

Frequently asked questions