Testing MQTT Clients: Unit, Integration and End-to-End Tests

Testing MQTT clients is awkward because the interesting behaviour happens across a network, inside callbacks, on someone else's thread. This guide shows a layered approach that works: fast unit tests with a mocked client, integration tests against a throwaway broker container in CI, and a short end-to-end smoke test against a real cloud broker before release.

Why MQTT Client Code Is Hard to Test

MQTT code mixes three concerns that fail in different ways:

1. Business logic: parsing payloads, deciding what to publish, mapping topics to devices.
2. Protocol behaviour: QoS, retained messages, reconnects, resubscribing, last will.
3. Environment: TLS certificates, credentials, ports, firewalls, broker quotas.

Most flaky MQTT test suites try to cover all three in one test that connects to a broker and sleeps. Splitting them lets each layer use the cheapest tool that can catch its bugs.

The examples use Python with paho-mqtt 2.x and pytest, but the structure carries over to any language.

Step 1: Design Code That Can Be Tested

The single most useful change is to keep the MQTT client at the edge. Put decisions in plain functions or classes that take a publish function as a dependency, and let a thin adapter wire them to paho.

thermostat.py:

import json
from dataclasses import dataclass
from typing import Callable

Publish = Callable[[str, str, int, bool], None]


@dataclass
class Thermostat:
    prefix: str
    publish: Publish
    setpoint: float = 21.0
    hysteresis: float = 0.5
    heating: bool = False

    def on_temperature(self, topic: str, payload: bytes) -> None:
        try:
            value = float(json.loads(payload)["value"])
        except (ValueError, KeyError, TypeError, json.JSONDecodeError):
            return  # ignore malformed messages

        room = topic.split("/")[-2]
        if not self.heating and value < self.setpoint - self.hysteresis:
            self.heating = True
        elif self.heating and value > self.setpoint + self.hysteresis:
            self.heating = False
        else:
            return

        self.publish(f"{self.prefix}/{room}/heater/set",
                     "ON" if self.heating else "OFF", 1, True)

adapter.py:

import threading

import paho.mqtt.client as mqtt
from thermostat import Thermostat


def build_client(host, port, client_id, prefix, username=None, password=None, tls=False):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if username:
        client.username_pw_set(username, password)
    if tls:
        client.tls_set()

    thermostat = Thermostat(
        prefix=prefix,
        publish=lambda t, p, q, r: client.publish(t, p, qos=q, retain=r),
    )

    def on_connect(client, userdata, flags, reason_code, properties):
        if not reason_code.is_failure:
            client.subscribe(f"{prefix}/+/temperature", qos=1)

    ready = threading.Event()

    def on_subscribe(client, userdata, mid, reason_codes, properties):
        ready.set()

    def on_message(client, userdata, msg):
        thermostat.on_temperature(msg.topic, msg.payload)

    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    return client, thermostat, ready

Subscribing inside on_connect is deliberate: it restores subscriptions after every reconnect, and the returned ready event lets tests wait until the subscription is active instead of guessing with sleep(). That is exactly the sort of wiring the integration tests will verify.

Step 2: Unit Tests With a Mocked Client

Because Thermostat takes publish as a parameter, a Mock replaces the network entirely.

test_thermostat.py:

import json
from unittest.mock import Mock

from thermostat import Thermostat

PREFIX = "home/test"


def make():
    pub = Mock()
    return Thermostat(prefix=PREFIX, publish=pub), pub


def msg(value):
    return json.dumps({"value": value}).encode()


def test_turns_heater_on_below_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    pub.assert_called_once_with(f"{PREFIX}/kitchen/heater/set", "ON", 1, True)


def test_no_publish_inside_hysteresis_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.2))
    pub.assert_not_called()


def test_turns_heater_off_above_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.6))
    assert pub.call_args.args[1] == "OFF"


def test_ignores_garbage():
    t, pub = make()
    for bad in [b"", b"not json", b'{"temp": 3}', b'{"value": "hot"}']:
        t.on_temperature(f"{PREFIX}/kitchen/temperature", bad)
    pub.assert_not_called()

These run in milliseconds and cover the logic that most often has bugs. Resist the urge to mock paho itself in detail (asserting that client.subscribe was called with exact arguments): such tests mirror the implementation, break on refactors and still do not prove the broker accepts what you send. That is the integration layer's job.

Step 3: Integration Tests Against a Broker Container

A real broker in a container gives honest protocol behaviour with no shared state and no internet dependency. The official Eclipse Mosquitto image includes a config file that allows anonymous connections, which is ideal for tests:

docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf

Now write tests that exercise the real adapter. A helper collects messages on a second client and waits on a condition instead of sleeping for a fixed time.

test_integration.py:

import os
import queue
import uuid

import paho.mqtt.client as mqtt
import pytest

from adapter import build_client

HOST = os.getenv("MQTT_TEST_HOST", "localhost")
PORT = int(os.getenv("MQTT_TEST_PORT", "1883"))


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


@pytest.fixture
def probe(prefix):
    received = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: received.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/#", qos=1)
    c.loop_start()
    yield c, received
    c.loop_stop()
    c.disconnect()


def wait_for(q, predicate, timeout=5.0):
    while True:
        topic, payload, retain = q.get(timeout=timeout)  # raises queue.Empty on timeout
        if predicate(topic, payload):
            return topic, payload, retain


def test_temperature_drives_heater(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/office/temperature", '{"value": 18.5}', qos=1)
        topic, payload, _ = wait_for(received, lambda t, p: t.endswith("/heater/set"))
        assert topic == f"{prefix}/office/heater/set"
        assert payload == "ON"
    finally:
        client.loop_stop()
        client.disconnect()


def test_heater_command_is_retained(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/hall/temperature", '{"value": 15}', qos=1)
        wait_for(received, lambda t, p: t.endswith("/heater/set"))
    finally:
        client.loop_stop()
        client.disconnect()

    # A late subscriber must get the retained command immediately
    late = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"late-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: late.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/hall/heater/set", qos=1)
    c.loop_start()
    try:
        _, payload, retain = late.get(timeout=5)
        assert payload == "ON" and retain is True
    finally:
        c.loop_stop()
        c.disconnect()

Three habits keep integration tests stable:

• Unique topic prefix per test. Retained messages from one test cannot leak into the next.
• Wait on conditions, not sleep(). A queue with a timeout fails fast and never waits longer than necessary.
• Always stop loops and disconnect in finally. Leaked network threads make later tests hang.

To test reconnect logic, stop and start the container mid-test (docker restart test-broker) and assert that the client resubscribes and receives a message published after the restart.

Running the Broker in CI

GitHub Actions can start the container as a step before the tests:

name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Start MQTT broker
        run: |
          docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
            mosquitto -c /mosquitto-no-auth.conf
      - run: pip install "paho-mqtt>=2,<3" pytest
      - run: pytest -q

Unit tests and integration tests can share one pytest run; mark the slow ones with @pytest.mark.integration if you want to skip them locally.

Step 4: End-to-End Smoke Tests Against a Real Broker

A container cannot tell you that your TLS settings, CA bundle, firewall rules or production credentials work. A short smoke test against the real endpoint can. For open-source projects or quick checks, the public MQTT test broker at broker.connectmq.com is enough; it accepts anonymous clients over TLS on port 8883. Keep the payloads synthetic, because the public area is readable by everyone.

smoke_test.py:

import os
import sys
import threading
import uuid

import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
USER = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"smoke-{uuid.uuid4().hex[:8]}"
TOPIC = f"{os.getenv('MQTT_PREFIX', 'connectmq/demo')}/smoke-{uuid.uuid4().hex[:8]}/ping"

got = threading.Event()
nonce = uuid.uuid4().hex
errors = []


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        errors.append(f"connect failed: {reason_code}")
        got.set()  # wake the main thread; callbacks cannot exit the process
        return
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    client.publish(TOPIC, nonce, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == nonce:
        got.set()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USER:
    client.username_pw_set(USER, PASSWORD)
client.tls_set()
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 8883, keepalive=30)
client.loop_start()
ok = got.wait(timeout=15) and not errors
for e in errors:
    print(e)
client.loop_stop()
client.disconnect()
print("smoke test passed" if ok else "smoke test FAILED: no echo within 15 s")
sys.exit(0 if ok else 1)

For production, point the same script at your private broker by setting MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID and MQTT_PREFIX from CI secrets. On a fully managed MQTT service like ConnectMQ, client IDs are bound to issued credentials, so create a dedicated credential set for CI rather than borrowing a device's identity; otherwise the smoke test will disconnect that device.

Run smoke tests before releases and on a schedule (for example hourly), not on every commit. Quotas apply on shared brokers: the public broker allows 10 messages per second per client and 10 subscriptions per client, and the free ConnectMQ Serverless plan caps concurrent connections at 25.

Common Anti-Patterns

A few habits make MQTT test suites slow and flaky. Watch for them in code review:

• Fixed sleeps. time.sleep(2) after publishing is either too long, wasting minutes across a suite, or too short on a busy CI runner. Wait on a queue or event with a timeout instead.
• Shared topics between tests. Retained messages and lingering subscriptions from one test leak into the next. Generate a unique prefix per test.
• Swallowed callback errors. Exceptions inside MQTT callbacks run on the network thread, so they do not fail the test directly. Capture them into a list and assert it is empty during teardown.
• Device credentials in CI. Borrowing a real device's identity for tests disconnects that device. Create separate credentials for automation.
• Testing the library instead of your code. You do not need to prove that paho delivers QoS 1 messages; you need to prove that your code subscribes to the right topics and reacts correctly.
• One giant end-to-end test. When it fails, nobody knows which layer broke. Keep end-to-end tests few and narrow, and push detail down into unit tests.

Testing Checklist

1. Business logic lives outside MQTT callbacks and is unit tested with a mock publish.
2. Malformed, empty and oversized payloads are tested and ignored safely.
3. Integration tests use a fresh broker container and a unique topic prefix per test.
4. Reconnect and resubscribe behaviour is tested by restarting the broker.
5. Retained messages and last will behaviour are asserted with a late subscriber.
6. A TLS smoke test with real credentials runs before every release.
7. CI credentials are separate from device credentials.

For language-specific details, see the PHP MQTT client tutorial, the Paho MQTT C client tutorial and the broader MQTT client libraries guide.

Get Started with ConnectMQ

Give your test pipeline a real broker to talk to. Create a free Serverless broker at app.connectmq.com, no credit card required, add a dedicated CI credential set, and run the smoke test against it on every release. The portal's live topic explorer and in-browser client make it easy to see exactly what your tests published. ConnectMQ runs the broker, you keep the tests.

Frequently Asked Questions

Should I mock the MQTT library in unit tests?

Mock at your own boundary, such as a publish function you inject, rather than mocking the library's internals. Verify real protocol behaviour with an integration test against a broker container.

Which broker should I use for MQTT integration tests in CI?

A throwaway Eclipse Mosquitto container is the simplest choice: it starts in about a second, needs no credentials with /mosquitto-no-auth.conf, and leaves no state behind between runs.

Can I use broker.connectmq.com in automated tests?

Yes for occasional smoke tests with synthetic data and unique topics, within its limits of 10 messages per second and 10 subscriptions per client. For heavy test suites, use a local container or your own private broker.

How do I test MQTT reconnect logic?

Restart the broker container during the test and assert that the client reconnects, resubscribes and receives a message published after the restart.