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())