From b58edace115205500c1197d9ce59cb5c6d724893 Mon Sep 17 00:00:00 2001 From: Godopu Date: Thu, 1 Oct 2026 01:04:19 +0000 Subject: [PATCH] test(issue-26): C0 reads broker state only after the matching ack; the local broker acks a PUBLISH after routing it Review (fa45b0e6): paho's connect() does not wait for the CONNACK, so C0 could read protocol_levels before the broker had handled the second CONNECT, and the `before` slice could take in a previous test's connection. - _Client waits for CONNACK (and UNSUBACK, instead of sleeping 0.3 s); C0 checks the levels of its own two client ids (stats() levels_by_client). - The local broker now routes and stores a PUBLISH before its PUBACK, so a publisher holding the ack may rely on the retained state it asked for (late subscriber replay, retained delete). - test_c0_holds_when_the_broker_is_slow runs C0 with CONNECT/UNSUBSCRIBE delayed and PUBLISH routed late; the old C0 fails it ([] == [4, 4]), as do an ack-before-route broker and a fixed-sleep unsubscribe. Co-Authored-By: Claude Opus 5.5 --- tests/_local_mqtt.py | 7 +++++- tests/test_local_mqtt.py | 54 ++++++++++++++++++++++++++++++++++------ 2 files changed, 53 insertions(+), 8 deletions(-) diff --git a/tests/_local_mqtt.py b/tests/_local_mqtt.py index e61583d..b6dbcd6 100644 --- a/tests/_local_mqtt.py +++ b/tests/_local_mqtt.py @@ -125,6 +125,7 @@ class LocalBroker: def stats(self) -> dict: with self._lock: return {"connections": self.connections, "protocol_levels": list(self.protocol_levels), + "levels_by_client": dict(zip(self.client_ids, self.protocol_levels)), "packets": dict(self.packets), "open": len(self._conns)} # -- network ------------------------------------------------------------- @@ -188,11 +189,15 @@ class LocalBroker: topic, i = _str(body, 0) if qos == 2: # the repo publishes QoS 0/1 only return False + pid = b"" if qos == 1: pid = body[i:i + 2] i += 2 - conn.send(_packet(PUBACK << 4, pid)) + # Route (and store a retained message) BEFORE the PUBACK: a publisher + # that has its ack may rely on the broker state it asked for. self._route(topic.decode("utf-8"), body[i:], qos, retain) + if qos == 1: + conn.send(_packet(PUBACK << 4, pid)) return True if ptype == SUBSCRIBE: pid, i, granted, new = body[:2], 2, bytearray(), [] diff --git a/tests/test_local_mqtt.py b/tests/test_local_mqtt.py index 4252687..3693cc7 100644 --- a/tests/test_local_mqtt.py +++ b/tests/test_local_mqtt.py @@ -51,22 +51,37 @@ def test_c3_the_product_config_resolves_to_the_local_broker(): # --------------------------------------------------------------------------- class _Client: - """A real mqtt_common.make_client client, connected and looping.""" + """A real mqtt_common.make_client client, connected and looping. + + Every step that changes broker state waits for the broker's ack (CONNACK, + SUBACK, UNSUBACK, PUBACK), and the broker records/applies the change + before it acks - so a test may read broker state right after a step. + paho's connect() only sends CONNECT; it does not wait for the CONNACK.""" def __init__(self, role): self.msgs = queue.Queue() self.subacks = queue.Queue() + self.unsubacks = queue.Queue() + self.connacks = queue.Queue() self.c = mqtt_common.make_client(role, mqtt_common.broker_config_from_env()) + self.client_id = self.c._client_id.decode() + self.c.on_connect = lambda _c, _u, _f, rc, _p: self.connacks.put(mqtt_common.reason_code_value(rc)) self.c.on_message = lambda _c, _u, m: self.msgs.put((m.topic, m.payload, m.qos, m.retain)) self.c.on_subscribe = lambda _c, _u, mid, rcs, _p: self.subacks.put([mqtt_common.reason_code_value(r) for r in rcs]) + self.c.on_unsubscribe = lambda *_a: self.unsubacks.put(True) cfg = mqtt_common.broker_config_from_env() self.c.connect(cfg.host, cfg.port, cfg.keepalive) self.c.loop_start() + assert self.connacks.get(timeout=5) == 0, "CONNACK refused" def sub(self, filt, qos=1): self.c.subscribe(filt, qos=qos) return self.subacks.get(timeout=5) + def unsub(self, filt): + self.c.unsubscribe(filt) + return self.unsubacks.get(timeout=5) + def get(self, timeout=5): return self.msgs.get(timeout=timeout) @@ -82,21 +97,19 @@ class _Client: self.c.disconnect() -def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained(): - broker = _local_mqtt.current_broker() - before = broker.stats()["connections"] +def _c0_conformance(broker): root = "mam/t26-" + uuid.uuid4().hex[:8] a, b = _Client("c0-pub"), _Client("c0-sub") try: - assert broker.stats()["protocol_levels"][before:] == [4, 4], "paho's default MQTT 3.1.1" + levels = broker.stats()["levels_by_client"] + assert (levels.get(a.client_id), levels.get(b.client_id)) == (4, 4), "paho's default MQTT 3.1.1" assert b.sub(f"{root}/+/events", qos=1) == [1] a.c.publish(f"{root}/j1/events", b"one", qos=1).wait_for_publish(5) assert b.get() == (f"{root}/j1/events", b"one", 1, False) assert b.sub(f"{root}/#", qos=0) == [0] a.c.publish(f"{root}/j1/deep/x", b"two", qos=1).wait_for_publish(5) assert b.get() == (f"{root}/j1/deep/x", b"two", 0, False), "min(pub qos, sub qos)" - b.c.unsubscribe(f"{root}/#") - time.sleep(0.3) + assert b.unsub(f"{root}/#") a.c.publish(f"{root}/j1/deep/y", b"three", qos=1).wait_for_publish(5) assert b.none(), "unsubscribed" # retained: replayed to a late subscriber, deleted by an empty payload @@ -120,6 +133,33 @@ def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained(): b.close() +def test_c0_make_client_speaks_3_1_1_wildcards_qos1_and_retained(): + _c0_conformance(_local_mqtt.current_broker()) + + +def test_c0_holds_when_the_broker_is_slow(monkeypatch): + """The C0 races, made deterministic. The broker takes its time on CONNECT + and UNSUBSCRIBE, and routes each PUBLISH later still (longer than a new + client takes to connect and subscribe). A test that reads broker state + before the matching ack, or sleeps a fixed time, fails here; so does a + broker that acks a PUBLISH before routing it.""" + broker = _local_mqtt.current_broker() + handle, route = broker._handle, broker._route + delay = {_local_mqtt.CONNECT: 0.3, _local_mqtt.UNSUBSCRIBE: 1.5} + + def slow_handle(conn, ptype, flags, body): + time.sleep(delay.get(ptype, 0)) + return handle(conn, ptype, flags, body) + + def slow_route(*args): + time.sleep(0.8) + return route(*args) + + monkeypatch.setattr(broker, "_handle", slow_handle) + monkeypatch.setattr(broker, "_route", slow_route) + _c0_conformance(broker) + + def _env(tmp_path: Path) -> dict: env = dict(os.environ) env.update(local_broker_env())