End-to-End Testing IoT Systems over MQTT — Device to UI

Unit tests prove that each service works. Customers care whether the reading a sensor sends shows up on their dashboard, and whether the button they press reaches the device. End-to-end testing for IoT systems over MQTT checks exactly that: a simulated device publishes through the real broker, and the test asserts on every hop, from the ingestion service and database to the API and the UI, then sends a command back and checks that the device received it.

This guide shows how to structure that test, with runnable pytest and Playwright code, and how to keep it fast and reliable enough to run on every merge.

The path an event travels

A typical MQTT-based IoT system has two directions, and an end-to-end suite should cover both:

Direction	Hops	What can break
Uplink (telemetry)	Device → broker → ingestion service → database → API → UI	Topic or payload changes, ACLs, parsing, time zones, caching
Downlink (commands)	UI → API → backend → broker → device → ack	Wrong topic, missing permissions, QoS, offline devices
Presence	Device disconnect → Last Will → backend → UI status	Keep-alive settings, LWT topic, stale state

Each hop already has its own tests (see MQTT integration testing for the broker boundary). The end-to-end suite is small, a handful of scenarios, but each one crosses every boundary at once.

Build a device simulator that behaves like the firmware

The simulator is the most important piece. It must use the same client ID scheme, topics, QoS, payload format and Last Will as the real firmware, otherwise you are testing a different system. A minimal Python version with paho-mqtt 2.x:

sim_device.py

import json, queue, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

class SimDevice:
    def __init__(self, device_id, host, user, password, prefix="testmqtt/e2e"):
        self.id = device_id
        self.base = f"{prefix}/{device_id}"
        self.commands = queue.Queue()
        self.subscribed = queue.Queue()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=device_id,
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(user, password)
        self.client.tls_set()
        self.client.will_set(f"{self.base}/status", "offline", qos=1, retain=True)
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, rc, p: self.subscribed.put(mid)
        self.client.on_message = self._on_message
        self.host = host
        self.crashed = False

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.publish(f"{self.base}/status", "online", qos=1, retain=True)
        client.subscribe(f"{self.base}/cmd", qos=1)

    def _on_message(self, client, userdata, msg):
        cmd = json.loads(msg.payload)
        self.commands.put(cmd)
        client.publish(f"{self.base}/cmd/ack", json.dumps({"id": cmd["id"], "ok": True}), qos=1)

    def start(self):
        self.client.connect(self.host, 8883, keepalive=30)
        self.client.loop_start()
        self.subscribed.get(timeout=10)        # SUBACK received: ready for commands

    def send_reading(self, temp_c):
        payload = {"schema": "telemetry/v1", "device": self.id, "temp_c": temp_c,
                   "ts": datetime.now(timezone.utc).isoformat()}
        self.client.publish(f"{self.base}/telemetry", json.dumps(payload), qos=1).wait_for_publish()

    def crash(self):
        """Drop the TCP connection without DISCONNECT, so the broker sends the Last Will."""
        self.crashed = True
        self.client.loop_stop()
        self.client.socket().close()

    def stop(self):
        if not self.crashed:
            # clear the retained status so the broker stays tidy
            self.client.publish(f"{self.base}/status", "", qos=1, retain=True).wait_for_publish()
            self.client.disconnect()
        self.client.loop_stop()

If you already have a fleet simulator, reuse it; see building an MQTT device simulator for scaling the same idea to many devices. For payload drift between firmware and simulator, a shared JSON Schema and contract tests keep both honest.

Wait on conditions, never on sleep

Asynchronous pipelines have variable latency. A fixed time.sleep(2) is either too short on a busy CI runner or wastes time everywhere else. Poll a condition until it holds or a deadline passes:

eventually.py

import time

def eventually(check, timeout=15.0, interval=0.25):
    """Call check() until it stops raising AssertionError, or re-raise after timeout."""
    deadline = time.monotonic() + timeout
    while True:
        try:
            return check()
        except AssertionError:
            if time.monotonic() > deadline:
                raise
            time.sleep(interval)

The uplink test: device to database to API

Each run gets a unique device ID, so parallel runs and leftovers from earlier runs never interfere:

test_e2e_uplink.py

import os, uuid
import httpx, psycopg, pytest
from eventually import eventually
from sim_device import SimDevice

HOST = "broker.connectmq.com"
API = os.environ.get("API_URL", "http://localhost:8000")
DSN = os.environ.get("DATABASE_URL", "postgresql://app:app@localhost:5432/app")

@pytest.fixture
def device():
    dev = SimDevice(f"e2e-{uuid.uuid4().hex[:8]}", HOST,
                    os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    dev.start()
    yield dev
    dev.stop()
    with psycopg.connect(DSN) as db:
        db.execute("DELETE FROM readings WHERE device_id = %s", (dev.id,))

def test_reading_reaches_database_and_api(device):
    device.send_reading(21.5)

    def stored():
        with psycopg.connect(DSN) as db:
            row = db.execute(
                "SELECT temp_c FROM readings WHERE device_id = %s ORDER BY ts DESC LIMIT 1",
                (device.id,),
            ).fetchone()
        assert row is not None, "reading not stored yet"
        assert row[0] == pytest.approx(21.5)
    eventually(stored)

    def served():
        r = httpx.get(f"{API}/devices/{device.id}/latest", timeout=5)
        assert r.status_code == 200
        assert r.json()["temp_c"] == pytest.approx(21.5)
    eventually(served)

The database check tells you ingestion works; the API check tells you that queries, serialization and caching work on top of it. When the test fails, the first failing assertion names the broken hop, which is the main value of asserting on every layer instead of only the UI. The MQTT to PostgreSQL guide shows a matching ingestion service.

The UI and downlink test with Playwright

The browser test checks what the user sees and drives the command path back to the device. With Playwright for Python:

test_e2e_ui.py

import os
from playwright.sync_api import Page, expect
from eventually import eventually

APP = os.environ.get("APP_URL", "http://localhost:3000")

def test_dashboard_shows_reading_and_sends_command(page: Page, device):
    device.send_reading(23.0)

    page.goto(f"{APP}/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)
    expect(page.get_by_test_id("temp-value")).to_have_text("23.0 °C", timeout=15_000)

    page.get_by_role("button", name="Reboot").click()

    cmd = device.commands.get(timeout=15)        # the simulator received it over MQTT
    assert cmd["action"] == "reboot"
    expect(page.get_by_test_id("last-command")).to_contain_text("acknowledged", timeout=15_000)

This reuses the device fixture (move it to conftest.py). Playwright's expect already retries until its timeout, so no extra polling is needed in the browser. Use data-testid attributes rather than CSS classes; they survive redesigns.

Presence: test the Last Will path

Offline detection is a feature users rely on and one that is rarely tested. Simulate a crash, not a clean disconnect, because a normal DISCONNECT suppresses the Last Will:

test_e2e_presence.py

from playwright.sync_api import Page, expect

def test_crashed_device_shows_offline(page: Page, device):
    page.goto(f"http://localhost:3000/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)

    device.crash()

    # a closed socket triggers the Will at once; a silently dead link takes up to 1.5 x keepalive
    expect(page.get_by_test_id("device-status")).to_have_text("offline", timeout=60_000)

A closed TCP socket triggers the Will immediately, but a device that loses power or coverage sends nothing, and the broker only notices after about 1.5 times the keep-alive interval, so keep the timeout above that. See testing Last Will for the timing rules. Watching testmqtt/e2e/# in the online MQTT client during a local run shows the status, telemetry, command and ack messages in order, which is the fastest way to debug a failing scenario.

Watch the end-to-end test traffic

Environment and CI

Run the backend, database and UI with Docker Compose and point them, and the simulator, at the same broker:

Run the suite locally

export MQTT_USER=username MQTT_PASS=password
docker compose up -d --wait          # api, ingest, postgres, web
pip install pytest pytest-playwright paho-mqtt httpx "psycopg[binary]"
playwright install chromium
pytest -q tests/e2e --maxfail=1

In CI, the same commands work in a job; see MQTT testing in GitHub Actions. Keep the broker configuration (TLS, ACLs, MQTT version) identical to production, because permission and TLS mistakes are exactly what end-to-end tests are supposed to catch.

Keeping the suite reliable

Unique IDs per run. Device ID, topic prefix and database rows all derive from one random suffix.
Wait for SUBACK. The simulator's start() blocks until the command subscription is confirmed; without it, the first command can arrive before anyone listens.
Clean up retained messages. Publishing an empty retained payload to the status topic in stop() keeps the broker tidy; see retained messages.
Few scenarios, full depth. Five end-to-end scenarios that cross every hop are worth more than fifty that duplicate unit tests.
Fail with the hop name. Write assertion messages like "reading not stored yet", so the report tells you where the chain broke.

An end-to-end suite for an MQTT system is a simulator, a polling helper and a few tests that follow one event from device to screen and back. It runs in minutes, catches the integration mistakes unit tests cannot see, and gives you a concrete answer to "does it work?" before every release.

Frequently asked questions

What is the difference between integration and end-to-end tests in IoT?

An integration test checks one boundary, for example that your ingestion service receives a message from the broker. An end-to-end test follows a single event through every component: device, broker, backend, database, API and UI, plus the command path back to the device.

Do I need real hardware for IoT end-to-end tests?

Not for most of them. A software device simulator that speaks the same MQTT topics and payloads as the firmware covers the cloud side reliably and runs in CI. Keep a small hardware-in-the-loop suite for radio, power and firmware-specific behaviour.

How do I keep MQTT end-to-end tests from being flaky?

Give every run a unique device ID and topic prefix, wait on conditions with a timeout instead of fixed sleeps, wait for SUBACK before publishing, and clean up the rows and retained messages the test created.

End-to-End Testing for IoT Systems over MQTT