Test Which MQTT Features a Broker Supports (MQTT 5, QoS 2, Retain)

"""Report which MQTT features a broker supports. Requires paho-mqtt >= 2.0."""
import os, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USER, PASS = os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS")
BASE = f"testmqtt/probe/{uuid.uuid4().hex[:8]}"

DEFAULTS = {  # value that applies when the broker omits the property
    "MaximumQoS": 2, "RetainAvailable": 1, "WildcardSubscriptionAvailable": 1,
    "SharedSubscriptionAvailable": 1, "SubscriptionIdentifierAvailable": 1,
    "ReceiveMaximum": 65535, "MaximumPacketSize": "no limit",
    "TopicAliasMaximum": 0, "ServerKeepAlive": "client value",
}

def connect(protocol=mqtt.MQTTv5, timeout=10):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"probe-{uuid.uuid4().hex[:8]}", protocol=protocol)
    if USER:
        c.username_pw_set(USER, PASS)
    if PORT == 8883:
        c.tls_set()
    done, res = threading.Event(), {}
    def on_connect(client, userdata, flags, reason_code, properties):
        res.update(rc=reason_code, props=properties)
        done.set()
    c.on_connect = on_connect
    c.connect(HOST, PORT, keepalive=30)
    c.loop_start()
    if not done.wait(timeout):
        c.loop_stop()
        raise TimeoutError("no CONNACK")
    return c, res["rc"], res["props"]

def close(*clients):
    for c in clients:
        c.disconnect()
        c.loop_stop()

def subscribe(c, topic_filter, qos=2, timeout=5):
    done, res = threading.Event(), {}
    def on_subscribe(client, userdata, mid, reason_code_list, properties):
        res["rc"] = reason_code_list[0]
        done.set()
    c.on_subscribe = on_subscribe
    c.subscribe(topic_filter, qos=qos)
    return res["rc"] if done.wait(timeout) else "no SUBACK"

def show(label, value):
    print(f"  {label:34} {value}")

def main():
    print(f"Broker {HOST}:{PORT}  (test topics under {BASE})\n")

    print("Protocol versions")
    for name, proto in (("MQTT 3.1.1", mqtt.MQTTv311), ("MQTT 5", mqtt.MQTTv5)):
        try:
            c, rc, _ = connect(proto)
            show(name, "OK" if not rc.is_failure else f"refused: {rc}")
            close(c)
        except Exception as e:
            show(name, f"FAILED ({e})")

    c, rc, props = connect()
    if rc.is_failure:
        raise SystemExit(f"MQTT 5 connect refused: {rc}")

    print("\nCONNACK properties")
    for name, default in DEFAULTS.items():
        value = getattr(props, name, None)
        show(name, value if value is not None else f"{default} (default)")

    print("\nSubscriptions (granted QoS or reason code)")
    for q in (0, 1, 2):
        show(f"QoS {q} subscription", subscribe(c, f"{BASE}/qos/{q}", qos=q))
    show("Wildcard filter (+)", subscribe(c, f"{BASE}/+/wild", qos=1))
    show("Shared subscription ($share)", subscribe(c, f"$share/probe/{BASE}/shared", qos=1))

    print("\nActive tests")
    got = {}
    received = threading.Event()
    def on_message(client, userdata, msg):
        got[msg.topic] = msg
        received.set()
    c.on_message = on_message

    # End-to-end QoS 2: publish at 2 to a QoS 2 subscription, check delivered QoS
    received.clear()
    c.publish(f"{BASE}/qos/2", b"qos2-probe", qos=2)
    if received.wait(5):
        show("QoS 2 delivery", f"delivered at QoS {got[f'{BASE}/qos/2'].qos}")
    else:
        show("QoS 2 delivery", "NOT delivered")

    # Retained message: publish, then subscribe with a fresh client
    topic = f"{BASE}/retain"
    try:
        c.publish(topic, b"retained-probe", qos=1, retain=True).wait_for_publish(5)
        sub, _, _ = connect()
        hit = threading.Event()
        sub.on_message = lambda client, u, m: m.retain and hit.set()
        subscribe(sub, topic, qos=1)
        show("Retained message", "OK" if hit.wait(5) else "NOT delivered")
        close(sub)
        c.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)  # clear it
    except Exception as e:
        show("Retained message", f"FAILED ({e})")

    close(c)

if __name__ == "__main__":
    main()

pip install "paho-mqtt>=2.0"
MQTT_HOST=broker.connectmq.com MQTT_PORT=8883 \
MQTT_USER="$MQTT_USER" MQTT_PASS="$MQTT_PASS" python probe.py

import mqtt from 'mqtt';

const client = mqtt.connect('mqtts://broker.connectmq.com:8883', {
  protocolVersion: 5,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
});

client.on('connect', (connack) => {
  // Absent keys mean the spec default applies
  console.log(connack.properties ?? '(no properties: all defaults)');
  client.subscribe('$share/probe/testmqtt/probe/js', { qos: 2 }, (err, granted) => {
    console.log(err ?? granted); // granted[0].qos is the granted QoS or a reason code
    client.end();
  });
});

client.on('error', (err) => console.error('connect failed:', err.message));

python probe.py | sed -E 's#testmqtt/probe/[0-9a-f]+#testmqtt/probe/X#' > capabilities.txt
git diff --exit-code capabilities.txt