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-testsmqtt-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
How to use
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.conftests/mosquitto.conf:
listener 1883
allow_anonymous true
persistence true
persistence_location /mosquitto/data/
log_dest stdoutStep 2 - paho-mqtt client setup (Python)
pip install paho-mqttimport 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):
| QoS | Guarantee | Use |
|---|---|---|
| 0 | At most once (best effort, may be lost) | High-frequency sensor where loss is acceptable |
| 1 | At least once (may duplicate) | Most application messaging |
| 2 | Exactly 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 receivedclean_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.
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-pattern | Why it fails | Fix |
|---|---|---|
| Test only QoS 0 ("works on my machine") | QoS 1/2 redelivery bugs ship | Step 3 covers matrix |
Use clean_start=True then expect persistence | Broker discards session; QoS 1 buffer lost | clean_start=False (Step 3) |
| Skip LWT test | Stale "online" status persists when clients crash | LWT reference |
Hardcode same client_id across tests | Broker disconnects existing on connect; flake | Unique client_id per test |
| Forget retained-message cleanup | Subsequent test runs see stale state | Publish empty retained payload between tests |
Limitations
References
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.