Skip to content

Testing Helpers

chumicro_mqtt.testing provides pre-baked broker-response byte sequences for unit tests. Tests typically drive MQTTClient against a chumicro_sockets.testing.FakeSocket — script the broker's responses with these helpers, let the client tick, then assert the wire-format on sock.sent.

The canned-bytes helpers stay in sync with the encoder/decoder, so a hand-rolled byte literal in a test doesn't drift if the wire format changes.

Usage

Enqueue each broker response just before the client action that expects it — the client drains the socket greedily, so a PUBACK sitting in the recv queue before its PUBLISH has been sent is treated as an unsolicited ack and faults the client.

from chumicro_mqtt import MQTTClient, ProtocolState
from chumicro_mqtt.testing import (
    canned_connack_bytes,
    canned_puback_bytes,
    canned_suback_bytes,
    canned_publish_bytes,
)
from chumicro_sockets.testing import FakeSocket

def test_connect_publish_subscribe_roundtrip():
    sock = FakeSocket()
    sock.enqueue_recv(canned_connack_bytes(return_code=0))

    client = MQTTClient(sock, client_id="test")
    client.connect()
    while client.state != ProtocolState.CONNECTED:
        client.handle(now_ms=0)

    sock.enqueue_recv(canned_suback_bytes(packet_id=1))
    client.subscribe("commands/+")
    for _ in range(3):
        client.handle(now_ms=0)

    acked = []
    sock.enqueue_recv(canned_puback_bytes(packet_id=2))
    client.publish(
        "sensors/temp", b"21.5", qos=1,
        on_publish=lambda topic, payload: acked.append(topic),
    )
    for _ in range(3):
        client.handle(now_ms=0)

    assert acked == ["sensors/temp"]
    # Inspect what the client wrote on the wire.  `sock.sent` is a
    # bytearray of everything send()'d in order.
    assert sock.sent[0] == 0x10  # CONNECT fixed header

Available helpers

Helper Returns
canned_connack_bytes(return_code=0, session_present=False) CONNACK packet — return_code=0 means accepted; 1-5 are rejection codes per MQTT 3.1.1 §3.2.2.3.
canned_puback_bytes(packet_id) PUBACK packet — drives QoS-1 publish acknowledgment.
canned_suback_bytes(packet_id, granted_qos=0) SUBACK packet — acknowledges a one-subscription SUBSCRIBE.
canned_unsuback_bytes(packet_id) UNSUBACK packet — acknowledges an UNSUBSCRIBE.
canned_pingresp_bytes() PINGRESP packet — keep-alive response.
canned_publish_bytes(topic, payload, qos=0, retain=False, packet_id=None) PUBLISH packet shaped like the broker would send — feed it to sock.enqueue_recv() to simulate an inbound message.

Simulating a broker rejection

def test_client_handles_connect_refusal():
    sock = FakeSocket()
    sock.enqueue_recv(canned_connack_bytes(return_code=4))   # bad credentials

    client = MQTTClient(sock, client_id="test")
    client.connect()
    client.handle(now_ms=0)

    assert client.state == ProtocolState.FAILED

Simulating an inbound message

def test_subscriber_receives_publish():
    sock = FakeSocket()
    sock.enqueue_recv(canned_connack_bytes(return_code=0))

    received = []
    client = MQTTClient(sock, client_id="test")
    client.on_message = lambda topic, payload: received.append((topic, payload))
    client.connect()
    while client.state != ProtocolState.CONNECTED:
        client.handle(now_ms=0)

    sock.enqueue_recv(canned_suback_bytes(packet_id=1))
    client.subscribe("sensors/+")
    for _ in range(3):
        client.handle(now_ms=0)

    # Broker now pushes a retained / fresh message on the subscription.
    sock.enqueue_recv(canned_publish_bytes("sensors/temp", b"22.1"))
    for _ in range(3):
        client.handle(now_ms=0)

    assert received == [("sensors/temp", b"22.1")]

Usage from other libraries

Libraries that depend on chumicro-mqtt can import the helpers directly in their own test suites:

from chumicro_mqtt.testing import canned_connack_bytes, canned_publish_bytes

Libraries that expose injectable services ship their own test fakes alongside the production code, so every consumer uses the same shared fake.

API Reference

chumicro_mqtt.testing

Pre-baked broker responses and client fixtures for unit tests.

canned_connack_bytes(*, return_code=0, session_present=False)

Return the four-byte CONNACK packet for return_code.

Parameters:

Name Type Description Default
return_code int

0 = accepted; 1-5 = rejection reasons (MQTT 3.1.1 §3.2.2.3).

 0
session_present bool

Bit 0 of the ack flags byte.

 False

canned_puback_bytes(packet_id)

Return the four-byte PUBACK packet for packet_id.

canned_suback_bytes(packet_id, granted_qos=0)

Return a one-subscription SUBACK with granted_qos.

canned_unsuback_bytes(packet_id)

Return the four-byte UNSUBACK packet for packet_id.

canned_pingresp_bytes()

Return the two-byte PINGRESP packet.

canned_publish_bytes(topic, payload, *, qos=0, retain=False, packet_id=None)

Return a PUBLISH packet shaped like the broker would send.

new_client(sock, ticks, **overrides)

Build an MQTTClient against sock and ticks with test defaults.

drive(client, ticks, count=1)

Call client.handle(ticks.ticks_ms()) count times in a row.

← Home

Source · \ PyPI · \ Bundle · \ Experimental Bundle