Testland
Browse all skills & agents

mqtt-tests

Test MQTT v5.0 with Mosquitto broker in CI + paho-mqtt clients - QoS 0 / 1 / 2 delivery semantics, retained messages, Last Will and Testament (LWT), shared subscriptions ($share/group/topic), $SYS topic introspection. Critical for IoT, embedded, and M2M systems where wire-level guarantees matter. Use when a product speaks MQTT on the wire and QoS 1 / 2 redelivery, retained-message state, or LWT behavior needs a broker-backed test - including smoke-testing a new broker auth / ACL / persistence config.

Install with skills.sh (any agent)

npx skills add testland/qa --skill mqtt-tests
View source

mqtt-tests

This skill covers the v5.0 surfaces tests must exercise: QoS 0 / 1 / 2 delivery semantics, retained messages, Last Will and Testament (LWT), shared subscriptions, and $SYS topic introspection - all per the MQTT v5.0 spec (opens in new window).

When to use

  • IoT / sensor / M2M product where MQTT is the wire protocol.
  • Pre-deploy gate: QoS 1 + 2 redelivery semantics correct, retained-message + LWT setup right.
  • Smoke test new broker config (auth, ACL, persistence).

How to use

  1. Stand up an eclipse-mosquitto:2 broker as a CI service on port 1883 with a persistence-enabled config (Step 1).
  2. Wire a paho-mqtt v5 client with callback_api_version=VERSION2 (Step 2).
  3. Exercise the QoS matrix - assert QoS 1 redelivers buffered messages to a reconnecting clean_start=False session (Step 3).
  4. Verify a retained message reaches a late subscriber, then clear it with an empty retained payload (Step 4).
  5. Cover the advanced broker behaviors - LWT on abnormal disconnect, shared-subscription round-robin, and $SYS diagnostics (references/lwt-shared-subs-sys.md).
  6. Use a unique client_id per test and clean up retained state between runs (Anti-patterns).

Step 1 - Run Mosquitto broker in CI

# GitHub Actions service
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - 1883:1883
    volumes:
      - ./tests/mosquitto.conf:/mosquitto/config/mosquitto.conf

tests/mosquitto.conf:

listener 1883
allow_anonymous true
persistence true
persistence_location /mosquitto/data/
log_dest stdout

Step 2 - paho-mqtt client setup (Python)

pip install paho-mqtt
import paho.mqtt.client as mqtt

def test_connect_v5():
    client = mqtt.Client(
        client_id="test-1",
        protocol=mqtt.MQTTv5,
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    )
    client.connect("localhost", 1883)
    client.loop_start()
    # ... exercise ...
    client.disconnect()
    client.loop_stop()

Step 3 - QoS matrix tests

Per the MQTT v5.0 spec (opens in new window):

QoSGuaranteeUse
0At most once (best effort, may be lost)High-frequency sensor where loss is acceptable
1At least once (may duplicate)Most application messaging
2Exactly once (slowest, full PUBREC/PUBREL/PUBCOMP handshake)Billing, payments, anything dedup-required
def test_qos1_redelivers_after_disconnect():
    received = []
    sub = mqtt.Client(client_id="sub", protocol=mqtt.MQTTv5,
                       clean_start=False, callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    sub.on_message = lambda c, u, msg: received.append(msg.payload)
    sub.connect("localhost", 1883)
    sub.subscribe("sensors/temp", qos=1)
    sub.loop_start()
    time.sleep(0.5)

    # Disconnect subscriber, publish messages, reconnect
    sub.disconnect()
    sub.loop_stop()

    pub = mqtt.Client(client_id="pub", protocol=mqtt.MQTTv5,
                       callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    pub.connect("localhost", 1883)
    pub.publish("sensors/temp", "22.5", qos=1).wait_for_publish()
    pub.disconnect()

    # Reconnect; broker delivers buffered QoS 1 messages
    sub.reconnect()
    sub.loop_start()
    time.sleep(2.0)
    sub.disconnect()
    sub.loop_stop()

    assert b"22.5" in received

clean_start=False is required for the broker to retain offline session state; without it, QoS 1 messages are lost.

Step 4 - Retained message test

Per the MQTT v5.0 spec (opens in new window), "servers store and distribute the most recent message on a topic to new subscribers, enabling state sharing without republishing."

def test_retained_message_delivered_to_late_subscriber():
    # Publisher sets retain=True
    pub = mqtt.Client(client_id="pub", protocol=mqtt.MQTTv5,
                       callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    pub.connect("localhost", 1883)
    pub.publish("device/status", "online", qos=1, retain=True).wait_for_publish()
    pub.disconnect()

    # New subscriber connects 5s later - still receives retained msg
    time.sleep(5)
    received = []
    sub = mqtt.Client(client_id="sub-late", protocol=mqtt.MQTTv5,
                       callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    sub.on_message = lambda c, u, msg: received.append(msg.payload)
    sub.connect("localhost", 1883)
    sub.subscribe("device/status", qos=1)
    sub.loop_start()
    time.sleep(1)
    sub.disconnect()
    sub.loop_stop()

    assert received == [b"online"]

To clear: publish empty payload with retain=True.

LWT, shared subscriptions, and $SYS introspection

See references/lwt-shared-subs-sys.md for Last Will and Testament on abnormal disconnect, shared-subscription round-robin ($share/<group>/<topic>), and $SYS broker-diagnostics tests.

Worked example

A sensor gateway publishes temperature to sensors/temp at QoS 1. QA needs to confirm a subscriber that drops offline still receives messages published during the outage.

  1. Connect a subscriber with clean_start=False and subscribe to sensors/temp at QoS 1 (Step 3).
  2. Disconnect the subscriber to simulate an outage.
  3. From a separate publisher, publish("sensors/temp", "22.5", qos=1).wait_for_publish().
  4. Reconnect the subscriber with reconnect() and pump the loop for ~2s.
  5. Assert b"22.5" in received - the broker delivered the buffered QoS 1 message on reconnect.

Result: offline-session persistence is verified end to end. Had the test used clean_start=True, the broker would have discarded the session and the message would have been lost - the exact regression this test guards.

Anti-patterns

Anti-patternWhy it failsFix
Test only QoS 0 ("works on my machine")QoS 1/2 redelivery bugs shipStep 3 covers matrix
Use clean_start=True then expect persistenceBroker discards session; QoS 1 buffer lostclean_start=False (Step 3)
Skip LWT testStale "online" status persists when clients crashLWT reference
Hardcode same client_id across testsBroker disconnects existing on connect; flakeUnique client_id per test
Forget retained-message cleanupSubsequent test runs see stale statePublish empty retained payload between tests

Limitations

  • MQTT v3.1.1 vs v5.0 differs on Will Properties + reason codes; pin client + broker version per test.
  • Mosquitto auth + ACL are not tested here; enable in a separate test suite if security requirements demand.
  • Some advanced v5 features (request/response, topic alias) require newer paho-mqtt; verify minimum version per the paho-mqtt reference (opens in new window).

References

  • MQTT v5.0 spec (opens in new window) - QoS, retained, LWT, shared subscriptions, $SYS
  • websocket-tests - alternative for browser-side bidirectional
  • webhook-replay-tests - HTTP alternative for at-least-once delivery

MQTT LWT, shared subscriptions, and $SYS introspection

View source (opens in new window)

MQTT LWT, shared subscriptions, and $SYS introspection

Advanced broker behaviors beyond core QoS and retained-message delivery: Last Will and Testament, shared subscriptions, and $SYS diagnostics.

Last Will and Testament (LWT)

Per the MQTT v5.0 spec (opens in new window), "When clients disconnect abnormally, servers automatically publish predetermined messages to notify other clients of unavailability."

def test_lwt_published_on_abnormal_disconnect():
    # Subscriber listens for status updates
    received = []
    monitor = mqtt.Client(client_id="monitor", protocol=mqtt.MQTTv5,
                           callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    monitor.on_message = lambda c, u, msg: received.append(msg.payload)
    monitor.connect("localhost", 1883)
    monitor.subscribe("device/+/status", qos=1)
    monitor.loop_start()

    # Device sets LWT then crashes
    device = mqtt.Client(client_id="device-1", protocol=mqtt.MQTTv5,
                          callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    device.will_set("device/1/status", "offline", qos=1, retain=False)
    device.connect("localhost", 1883)
    device.publish("device/1/status", "online", qos=1, retain=True).wait_for_publish()
    # Simulate crash (no clean disconnect)
    device._sock.close()

    time.sleep(3)  # broker keepalive timeout
    monitor.disconnect()
    monitor.loop_stop()

    assert b"offline" in received

Shared subscriptions ($share/...)

Per the MQTT v5.0 spec (opens in new window), shared subscriptions distribute messages among group members rather than broadcasting:

$share/<groupname>/<topic-filter>
def test_shared_subscription_round_robin():
    received_a = []
    received_b = []
    sub_a = mqtt.Client(client_id="sub-a", protocol=mqtt.MQTTv5,
                         callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    sub_b = mqtt.Client(client_id="sub-b", protocol=mqtt.MQTTv5,
                         callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    sub_a.on_message = lambda c, u, msg: received_a.append(msg.payload)
    sub_b.on_message = lambda c, u, msg: received_b.append(msg.payload)

    sub_a.connect("localhost", 1883); sub_a.subscribe("$share/workers/jobs", qos=1); sub_a.loop_start()
    sub_b.connect("localhost", 1883); sub_b.subscribe("$share/workers/jobs", qos=1); sub_b.loop_start()
    time.sleep(0.5)

    pub = mqtt.Client(client_id="pub", protocol=mqtt.MQTTv5,
                       callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    pub.connect("localhost", 1883)
    for i in range(10):
        pub.publish("jobs", f"job-{i}", qos=1).wait_for_publish()
    pub.disconnect()
    time.sleep(1)

    sub_a.disconnect(); sub_a.loop_stop()
    sub_b.disconnect(); sub_b.loop_stop()

    # Each got some, neither got all
    assert 0 < len(received_a) < 10
    assert 0 < len(received_b) < 10
    assert len(received_a) + len(received_b) == 10

$SYS topic introspection

Per the MQTT v5.0 spec (opens in new window), $SYS/... reserved topics provide broker diagnostics. Useful for monitoring tests:

def test_broker_reports_connected_clients():
    received = []
    monitor = mqtt.Client(client_id="monitor", protocol=mqtt.MQTTv5,
                           callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    monitor.on_message = lambda c, u, msg: received.append((msg.topic, msg.payload))
    monitor.connect("localhost", 1883)
    monitor.subscribe("$SYS/broker/clients/connected", qos=0)
    monitor.loop_start()
    time.sleep(15)  # $SYS update interval default = 10s
    monitor.disconnect()
    monitor.loop_stop()

    assert any(b for _, b in received if int(b) >= 1)

$SYS/... topic set varies per broker - Mosquitto + EMQX + HiveMQ each publish slightly different metrics.

Related skills

grpc-streaming-tests

Test gRPC streaming RPCs - Server-streaming (server returns sequence), Client-streaming (client sends sequence), Bidirectional (both sides stream independently). Cover deadline + cancellation + flow control + status codes (CANCELLED, DEADLINE_EXCEEDED) + metadata. Use ghz for load, grpcurl for ad-hoc, language-native test stubs for unit/integration. Use when a service exposes server-, client-, or bidirectional-streaming RPCs and deadline, cancellation, or partial-stream status-code behavior is unverified.

server-sent-events-tests

Test Server-Sent Events (SSE) flows, one-way server-to-client push only (not bidirectional, use websocket-tests for client-to-server messaging): `EventSource` API on the browser side (`onmessage`, `onerror`, `readyState` 0/1/2), event stream format (`data:`, `event:`, `id:`, `retry:`), `Last-Event-ID` reconnect-with-replay header, content-type `text/event-stream`, and HTTP/1.1 connection-pool limits. Use Playwright for browser-side, raw HTTP client for server-side stream tests. Use when a feature pushes updates over `text/event-stream` and the reconnect interval, `Last-Event-ID` replay, or per-origin connection ceiling has no coverage.

sse-load-tests

Load-tests SSE endpoints at scale with k6 - measures concurrent-stream capacity, connection churn, and server memory pressure. Covers the HTTP/1.1 6-connection-per-origin browser ceiling vs HTTP/2 multiplexing, a custom k6 SSE client built on ReadableStream, and threshold gates for TTFB and data throughput. Use when validating whether a server can sustain N concurrent EventSource connections without connection starvation or memory growth.

stomp-amqp-tests

Tests STOMP over WebSocket (Spring, ActiveMQ, RabbitMQ Web STOMP) and AMQP 0-9-1 (RabbitMQ Java client) - frame connect/subscribe/send/ack sequences, ack modes (auto/client/client-individual), exchange and queue declarations, binding routing, Testcontainers RabbitMQ broker, and delivery assertion. Use when validating enterprise Spring or RabbitMQ messaging stacks before deploy.

webhook-replay-tests

Tests inbound webhook receivers for replay-attack resistance: capture incoming webhook payloads + headers, replay against the receiver under test, validate the Standard Webhooks signature scheme (svix-id + svix-timestamp + svix-signature, HMAC-SHA256 over `{id}.{timestamp}.{payload}`), svix-id idempotency dedup, and 5-minute timestamp-window enforcement by signing fixtures at runtime. Does NOT cover outbound delivery, retry-on-5xx, or failure-event exhaustion - those belong to an outbound webhook delivery harness. Use when testing the receiving side of a webhook integration.

websocket-tests

Test WebSocket protocol behavior - opening handshake (HTTP Upgrade with Sec-WebSocket-Key + Sec-WebSocket-Version: 13), control frames (ping 0x9 / pong 0xA / close 0x8), close-frame status codes (1000 normal, 1001 going-away, 1006 abnormal, 1011 server error), subprotocol negotiation, backpressure, and reconnect with jitter. Works with ws (Node), websockets (Python), or Playwright frame inspection per language. Use when a feature holds a long-lived WebSocket open and reconnect, close-code, or backpressure behavior is unverified.