From 2826e709ae5aea7badb1f8ffa26299c97a0c0df6 Mon Sep 17 00:00:00 2001 From: Godopu Date: Wed, 30 Sep 2026 15:54:58 +0000 Subject: [PATCH] test(issue-26): no test depends on a third-party MQTT broker - tests/_local_mqtt.py: in-process MQTT 3.1.1 broker on 127.0.0.1 (what the repo uses: protocol level 4, clean session, QoS 0/1, retained). - conftest: one session broker named by MQTT_BROKER/MQTT_PORT/MQTT_TLS; other redirecting/auth MQTT_* dropped and restored afterwards. - Harnesses of the 22 broker-dependent tests (#18 acceptance1/2/3/5, #6 t1_1/2/7/8/14/15/16/20, #19 MQTT loop x2, #33 b3/r3/b7, #30 s9/s10/r2) use it; the "broker unreachable" skips are gone (a local broker that does not deliver is a failure). acceptance1/5 and #19 also assert on the subscriber, runner and audit logs the real producers write. - tests/test_local_mqtt.py: conformance with the real clients and the real publish_event -> job_subscriber path (C0), an AST guard against public broker literals and MQTT_* deleted without a replacement (C2), and the session's broker config (C3). Co-Authored-By: Claude Opus 5.5 --- tests/_local_mqtt.py | 272 ++++++++++++++++ tests/conftest.py | 32 ++ tests/test_issue18_liveness_contract.py | 62 ++-- ...est_issue19_subscriber_daemon_lifecycle.py | 9 +- tests/test_issue30_stall_watch.py | 13 +- tests/test_issue33_budget.py | 10 +- tests/test_issue6_fail_closed.py | 3 + tests/test_local_mqtt.py | 304 ++++++++++++++++++ 8 files changed, 656 insertions(+), 49 deletions(-) create mode 100644 tests/_local_mqtt.py create mode 100644 tests/test_local_mqtt.py diff --git a/tests/_local_mqtt.py b/tests/_local_mqtt.py new file mode 100644 index 0000000..e61583d --- /dev/null +++ b/tests/_local_mqtt.py @@ -0,0 +1,272 @@ +"""Issue #26 - a test-only, in-process MQTT 3.1.1 broker. + +The suite used to round-trip through the public default broker +(mqtt_common.BrokerConfig.host). conftest starts one of these per session on +127.0.0.1 and points MQTT_BROKER/MQTT_PORT/MQTT_TLS at it, so every real +producer (publish_event, job_subscriber, the wrapper, the remote runner, the +monitor daemon) talks to a broker the test process owns. + +Scope is what the repo uses: paho-mqtt 2.x clients with its defaults +(protocol level 4 = 3.1.1, clean session, no will), QoS 0/1, retained +messages. No TLS, no auth checks, no persistent sessions, no QoS 2, no +redelivery of unacknowledged QoS 1 messages. The name does not start with +`test_`, so pytest never collects it. +""" +from __future__ import annotations + +import collections +import socket +import struct +import threading + +CONNECT, CONNACK, PUBLISH, PUBACK = 1, 2, 3, 4 +SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK = 8, 9, 10, 11 +PINGREQ, PINGRESP, DISCONNECT = 12, 13, 14 +NAMES = {CONNECT: "CONNECT", PUBLISH: "PUBLISH", PUBACK: "PUBACK", SUBSCRIBE: "SUBSCRIBE", + UNSUBSCRIBE: "UNSUBSCRIBE", PINGREQ: "PINGREQ", DISCONNECT: "DISCONNECT"} + +_CURRENT = None + + +def topic_matches(filt: str, topic: str) -> bool: + f, t = filt.split("/"), topic.split("/") + if topic.startswith("$") and f[0] in ("+", "#"): + return False + for i, part in enumerate(f): + if part == "#": + return True + if i >= len(t) or (part != "+" and part != t[i]): + return False + return len(f) == len(t) + + +def _varint(n: int) -> bytes: + out = bytearray() + while True: + n, digit = divmod(n, 128) + out.append(digit | (0x80 if n else 0)) + if not n: + return bytes(out) + + +def _packet(first: int, body: bytes) -> bytes: + return bytes([first]) + _varint(len(body)) + body + + +def _str(data: bytes, i: int): + (n,) = struct.unpack_from("!H", data, i) + return data[i + 2:i + 2 + n], i + 2 + n + + +class _Conn: + def __init__(self, sock): + self.sock = sock + self.lock = threading.Lock() + self.subs = {} # filter -> granted qos + self.next_pid = 0 + self.client_id = None + + def send(self, data: bytes) -> None: + with self.lock: + try: + self.sock.sendall(data) + except OSError: + pass + + def recv_exact(self, n: int) -> bytes: + buf = b"" + while len(buf) < n: + chunk = self.sock.recv(n - len(buf)) + if not chunk: + raise ConnectionError("closed") + buf += chunk + return buf + + +class LocalBroker: + def __init__(self): + self._lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._lsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._lsock.bind(("127.0.0.1", 0)) + self._lsock.listen(128) + self.host, self.port = self._lsock.getsockname() + self._lock = threading.Lock() + self._conns = set() + self.retained = {} # topic -> (payload, qos) + self.connections = 0 + self.protocol_levels = [] # one entry per CONNECT + self.client_ids = [] + self.packets = collections.Counter() # inbound packets by name + self._stopped = False + + # -- lifecycle ----------------------------------------------------------- + def start(self) -> "LocalBroker": + threading.Thread(target=self._accept_loop, name="local-mqtt-accept", daemon=True).start() + return self + + def stop(self) -> None: + self._stopped = True + try: + self._lsock.close() + except OSError: + pass + with self._lock: + conns = list(self._conns) + for c in conns: + try: + c.sock.shutdown(socket.SHUT_RDWR) + c.sock.close() + except OSError: + pass + + def env(self) -> dict: + return {"MQTT_BROKER": self.host, "MQTT_PORT": str(self.port), "MQTT_TLS": "0"} + + def stats(self) -> dict: + with self._lock: + return {"connections": self.connections, "protocol_levels": list(self.protocol_levels), + "packets": dict(self.packets), "open": len(self._conns)} + + # -- network ------------------------------------------------------------- + def _accept_loop(self) -> None: + while not self._stopped: + try: + sock, _ = self._lsock.accept() + except OSError: + return + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + threading.Thread(target=self._serve, args=(_Conn(sock),), name="local-mqtt-conn", daemon=True).start() + + def _serve(self, conn: _Conn) -> None: + with self._lock: + self._conns.add(conn) + try: + while True: + first = conn.recv_exact(1)[0] + length, mult = 0, 1 + while True: + b = conn.recv_exact(1)[0] + length += (b & 0x7F) * mult + mult *= 128 + if not b & 0x80: + break + body = conn.recv_exact(length) if length else b"" + ptype = first >> 4 + with self._lock: + self.packets[NAMES.get(ptype, str(ptype))] += 1 + if conn.client_id is None and ptype != CONNECT: + return + if not self._handle(conn, ptype, first & 0x0F, body): + return + except (ConnectionError, OSError, struct.error, IndexError): + return + finally: + with self._lock: + self._conns.discard(conn) + try: + conn.sock.close() + except OSError: + pass + + def _handle(self, conn: _Conn, ptype: int, flags: int, body: bytes) -> bool: + if ptype == CONNECT: + _name, i = _str(body, 0) + level, cflags = body[i], body[i + 1] + client_id, i = _str(body, i + 4) + with self._lock: + self.connections += 1 + self.protocol_levels.append(level) + self.client_ids.append(client_id.decode("utf-8", "replace")) + if level not in (3, 4): # 3.1 / 3.1.1 only (no MQTT v5 in the repo) + conn.send(_packet(CONNACK << 4, b"\x00\x01")) + return False + conn.client_id = client_id.decode("utf-8", "replace") + conn.send(_packet(CONNACK << 4, b"\x00\x00")) + return True + if ptype == PUBLISH: + qos, retain = (flags >> 1) & 3, bool(flags & 1) + topic, i = _str(body, 0) + if qos == 2: # the repo publishes QoS 0/1 only + return False + if qos == 1: + pid = body[i:i + 2] + i += 2 + conn.send(_packet(PUBACK << 4, pid)) + self._route(topic.decode("utf-8"), body[i:], qos, retain) + return True + if ptype == SUBSCRIBE: + pid, i, granted, new = body[:2], 2, bytearray(), [] + while i < len(body): + filt, i = _str(body, i) + q = min(body[i], 1) + i += 1 + conn.subs[filt.decode("utf-8")] = q + granted.append(q) + new.append((filt.decode("utf-8"), q)) + conn.send(_packet(SUBACK << 4, pid + bytes(granted))) + with self._lock: + retained = list(self.retained.items()) + for filt, q in new: + for topic, (payload, rq) in retained: + if topic_matches(filt, topic): + self._deliver(conn, topic, payload, min(rq, q), True) + return True + if ptype == UNSUBSCRIBE: + pid, i = body[:2], 2 + while i < len(body): + filt, i = _str(body, i) + conn.subs.pop(filt.decode("utf-8"), None) + conn.send(_packet(UNSUBACK << 4, pid)) + return True + if ptype == PINGREQ: + conn.send(_packet(PINGRESP << 4, b"")) + return True + if ptype == PUBACK: # a subscriber acknowledging our QoS 1 delivery + return True + return False # DISCONNECT, or anything unsupported + + def _route(self, topic: str, payload: bytes, qos: int, retain: bool) -> None: + with self._lock: + if retain: + if payload: + self.retained[topic] = (payload, qos) + else: + self.retained.pop(topic, None) + conns = list(self._conns) + for c in conns: + granted = [q for f, q in list(c.subs.items()) if topic_matches(f, topic)] + if granted: + self._deliver(c, topic, payload, min(qos, max(granted)), False) + + def _deliver(self, conn: _Conn, topic: str, payload: bytes, qos: int, retain: bool) -> None: + t = topic.encode("utf-8") + body = struct.pack("!H", len(t)) + t + if qos: + with conn.lock: + conn.next_pid = conn.next_pid % 65535 + 1 + pid = conn.next_pid + body += struct.pack("!H", pid) + conn.send(_packet((PUBLISH << 4) | (qos << 1) | int(retain), body + payload)) + + +def start_broker() -> LocalBroker: + global _CURRENT + _CURRENT = LocalBroker().start() + return _CURRENT + + +def stop_broker() -> None: + global _CURRENT + if _CURRENT is not None: + _CURRENT.stop() + _CURRENT = None + + +def current_broker() -> LocalBroker: + assert _CURRENT is not None, "the session's local MQTT broker is not running (conftest)" + return _CURRENT + + +def local_broker_env() -> dict: + """MQTT_BROKER / MQTT_PORT / MQTT_TLS for this session's local broker.""" + return current_broker().env() diff --git a/tests/conftest.py b/tests/conftest.py index fe597ee..ab7560b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1550,3 +1550,35 @@ def delegation_tree_provenance_guard(tmp_path_factory): "(these would shadow the real tree for any later first-time import):\n " + "\n ".join(polluted) ) + + +# ============================================================================== +# Issue #26: no test reaches a third-party MQTT broker. One in-process broker +# (tests/_local_mqtt.py) per session on 127.0.0.1; MQTT_BROKER/MQTT_PORT/ +# MQTT_TLS name it, and every other MQTT_* that would redirect or authenticate +# a client is dropped. Subprocesses inherit it through os.environ; a harness +# that builds its own env takes _local_mqtt.local_broker_env(). Tests that +# deliberately name an unreachable broker (127.0.0.1:9, :65534) still win. +# ============================================================================== +import _local_mqtt # noqa: E402 + +_MQTT_DROPPED = ("MQTT_HOST", "MQTT_USERNAME", "MQTT_PASSWORD", "MQTT_CA_CERTS", "MQTT_CERTFILE", "MQTT_KEYFILE") + + +@pytest.fixture(scope="session", autouse=True) +def local_mqtt_broker(): + broker = _local_mqtt.start_broker() + keys = tuple(broker.env()) + _MQTT_DROPPED + previous = {k: os.environ.get(k) for k in keys} + for k in _MQTT_DROPPED: + os.environ.pop(k, None) + os.environ.update(broker.env()) + try: + yield broker + finally: + for k, v in previous.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + _local_mqtt.stop_broker() diff --git a/tests/test_issue18_liveness_contract.py b/tests/test_issue18_liveness_contract.py index 8bd4b02..461352c 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -18,11 +18,12 @@ Contract (registry fields, wrapper options): its own registry and broker credentials runs it, and the orchestrator's subscriber mirrors its events into the orchestrator's registry. -Tests marked "broker" talk to the configured broker (the public default in the -sandbox); each stays under a minute. +Tests marked "broker" talk to the session's in-process broker (tests/_local_mqtt.py, +Issue #26); each stays under a minute. """ from __future__ import annotations +import hashlib import json import os import re @@ -36,6 +37,8 @@ from pathlib import Path import pytest +from _local_mqtt import local_broker_env + REPO_ROOT = Path(__file__).resolve().parent.parent SKILLS = REPO_ROOT / ".agents" / "skills" LIB_SH = SKILLS / "lib.sh" @@ -59,6 +62,7 @@ def _env(cwd: Path, extra=None) -> dict: env["PYTHONPATH"] = str(SKILLS) for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"): env.pop(k, None) + env.update(local_broker_env()) # Issue #26: the session's in-process broker if extra: env.update({k: str(v) for k, v in extra.items()}) return env @@ -1091,6 +1095,12 @@ def test_acceptance1_headless_agent_survives_on_heartbeats_alone(tmp_path, mock_ assert _prompt_calls(Path(mock_herdr)) == 1 beats = [e for e in _events(reg, jid) if e["event"] == "progress"] assert len(beats) >= 5, len(beats) + # Issue #26: the wrapper's own subscriber saw them on the wire (its console + # record, written by job_subscriber, not by the test) + sub = (reg / f"{jid}.subscriber.out").read_text() + assert f"SUBSCRIBED {_job(reg, jid)['topic_prefix']}/events" in sub, sub + assert len(re.findall(rf"job={jid} seq=\d+ progress\s+heartbeat$", sub, re.M)) >= 5, sub + assert re.search(rf"job={jid} seq=\d+ completed\s+PASS$", sub, re.M), sub brief = (reg / jid / "brief.md").read_text() assert "heartbeat every 2 s" in brief and "after 2 missed heartbeats" in brief and "within 40 s" in brief @@ -1161,31 +1171,6 @@ def test_acceptance3_lost_heartbeats_close_the_job_and_a_late_completed_is_refus # 6. remote transport (broker): no shared disk, no pane, only the wire # --------------------------------------------------------------------------- -def _skip_if_broker_did_not_deliver(reg: Path, runner_log: Path) -> None: - """acceptance5 runs over a public broker. The remote runner already covers - the idle window with `started --heartbeat` (remote.py handle_dispatch). The - observed failures were of one kind: the runner published started and - completed successfully, the orchestrator had subscribed to the right topic, - and it received nothing at all for the job - the broker did not deliver. - Only that proven non-delivery skips; a drop (HMAC, identity), a wrong - topic, or a runner that failed to publish still fails.""" - log = runner_log.read_text(errors="replace") - m = re.search(r"published started seq=1 job=(\w+)", log) - if not m: - return - jid = m.group(1) - rec = reg / f"{jid}.json" - if not rec.exists(): - return - topic = json.loads(rec.read_text()).get("topic_prefix") or "" - out = reg / f"{jid}.subscriber.out" - sub = out.read_text(errors="replace") if out.exists() else "" - if (topic and f"SUBSCRIBED {topic}/events" in sub - and f"job={jid}" not in sub and f"drop event for job {jid}" not in sub): - pytest.skip(f"public broker did not deliver the runner's events to the orchestrator " - f"(runner published started for {jid}; subscriber received nothing for it)") - - def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path): orch = tmp_path / "orchestrator" remote = tmp_path / "remote-host" @@ -1210,8 +1195,7 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path deadline = time.time() + 45 while time.time() < deadline and "SUBSCRIBED" not in runner_log.read_text(errors="replace"): time.sleep(0.2) - if "SUBSCRIBED" not in runner_log.read_text(errors="replace"): - pytest.skip("broker unreachable: " + runner_log.read_text(errors="replace")[-300:]) + assert "SUBSCRIBED" in runner_log.read_text(errors="replace"), runner_log.read_text(errors="replace") reg = orch / "jobs" proc, env = _submit(orch, reg, f"remote:{session}", "remote", @@ -1219,11 +1203,6 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path env_extra={"MAM_DISPATCH_SECRET": secret}, timeout_s=120) stdout, stderr = proc.communicate(timeout=180) jid = _wait_job_id(reg) - if proc.returncode != 0: - if ("no CONNACK from broker" in stderr - and "published started" not in runner_log.read_text(errors="replace")): - pytest.skip("broker unreachable when publishing the dispatch (no CONNACK); the runner never saw the job") - _skip_if_broker_did_not_deliver(reg, runner_log) assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}" # delivery is confirmed by the protocol: the subscriber's attributed # audit record of the remote `started` (review R1: the registry status @@ -1246,6 +1225,21 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path # the agent command's last stdout line became the completed detail assert "remote work done for " + jid in stdout, stdout assert runner.returncode == 0 + # Issue #26: both ends' own logs. The runner (and the publish_event it + # runs) wrote runner.log; the orchestrator's subscriber wrote the audit + # record of what it received, attributed to this dispatch. + rlog = runner_log.read_text(errors="replace") + assert f"DISPATCH job={jid} " in rlog, rlog + assert f"published started seq=1 job={jid} " in rlog, rlog + assert re.search(rf"published completed seq=\d+ job={jid} ", rlog), rlog + assert f"DONE job={jid} rc=0" in rlog, rlog + audit = [json.loads(l) for l in (orch / "logs" / jid / "events.ndjson").read_text().splitlines() if l.strip()] + fp = hashlib.sha256(_job(reg, jid)["dispatch_token"].encode("utf-8")).hexdigest() + for ev in ("started", "completed"): + got = [a for a in audit if a.get("event") == "received" and a.get("source_event") == ev] + assert got and got[0]["dispatch_verified"] is True and got[0]["dispatch_fingerprint"] == fp, (ev, audit) + sub = (reg / f"{jid}.subscriber.out").read_text() + assert f"SUBSCRIBED {_job(reg, jid)['topic_prefix']}/events" in sub, sub assert "살아있는 에이전트 세션" not in stdout, "no pane injection path for a remote session" finally: if runner.poll() is None: diff --git a/tests/test_issue19_subscriber_daemon_lifecycle.py b/tests/test_issue19_subscriber_daemon_lifecycle.py index 7b37e0e..11bb16f 100644 --- a/tests/test_issue19_subscriber_daemon_lifecycle.py +++ b/tests/test_issue19_subscriber_daemon_lifecycle.py @@ -28,6 +28,8 @@ from pathlib import Path import pytest +from _local_mqtt import local_broker_env + FIXTURES = Path(__file__).parent / "fixtures" / "issue_5" / "phase1" DELEGATION_MODES = ("0", "1") @@ -415,11 +417,12 @@ def test_sub_stop_trap_exits_143_before_the_first_background_child(): # --------------------------------------------------------------------------- def _start_connected(mam_sandbox, log): - proc = _start_subscriber(mam_sandbox, log, {}) - if not _wait_for(log, r"MQTT Monitor: subscribed to ", 45): + # Issue #26: the session's in-process broker is always up - not reaching it fails. + proc = _start_subscriber(mam_sandbox, log, local_broker_env()) + if not _wait_for(log, r"MQTT Monitor: subscribed to mam/[0-9a-f]{12}/jobs/\+/events", 45): proc.kill() proc.wait(timeout=10) - pytest.skip("broker unreachable from this host: " + log.read_text()[-300:]) + pytest.fail("the monitor daemon never subscribed on the local broker: " + log.read_text()[-300:]) return proc diff --git a/tests/test_issue30_stall_watch.py b/tests/test_issue30_stall_watch.py index 1be7138..27790be 100644 --- a/tests/test_issue30_stall_watch.py +++ b/tests/test_issue30_stall_watch.py @@ -488,12 +488,7 @@ def test_scoped_env_drops_the_pane_inherited_workspace_for_name_lookups_only(): import test_issue18_liveness_contract as t18 # noqa: E402 -LIVE_BROKER = ("MQTT_BROKER", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD") - - -def _live_broker() -> dict: - """This environment's broker (the #18 harness defaults to a public one).""" - return {k: os.environ[k] for k in LIVE_BROKER if k in os.environ} +from _local_mqtt import local_broker_env # noqa: E402 (Issue #26: the session's in-process broker) def _watchers(jid: str) -> list: @@ -507,7 +502,7 @@ def test_s9_the_wrapper_runs_the_watcher_for_the_jobs_life(tmp_path, mock_herdr, reg.mkdir() t18._headless_agent(Path(mock_herdr), "headless-codex") proc, env = t18._submit(tmp_path, reg, "herdr:headless-codex", "codex", ["--ack-timeout", "40"], - env_extra={"MAM_STALL_WATCH": switch, **_live_broker()}, timeout_s=60) + env_extra={"MAM_STALL_WATCH": switch, **local_broker_env()}, timeout_s=60) jid = t18._wait_job_id(reg) try: time.sleep(1) @@ -533,7 +528,7 @@ def test_s10_remote_and_human_jobs_are_not_watched(tmp_path, agent, session, why reg = tmp_path / "jobs" reg.mkdir() proc, env = t18._submit(tmp_path, reg, session, agent, ["--ack-timeout", "40"], timeout_s=60, - env_extra={"MAM_DISPATCH_SECRET": "s10-secret", **_live_broker()}) + env_extra={"MAM_DISPATCH_SECRET": "s10-secret", **local_broker_env()}) jid = t18._wait_job_id(reg) try: time.sleep(1) @@ -620,7 +615,7 @@ def test_r2_the_loop_relays_a_blocked_notice_while_the_job_is_open(mam_sandbox, "SKS_IDLE_TIMEOUT_MS": "500", "SKS_EMPTY_GIVEUP": "1", "MAM_ACK_TIMEOUT_S": "40", "MAM_JOBS_DIR": str(reg), "PYTHONPATH": str(SKILLS), "PYTHONUNBUFFERED": "1"}) - env.update(_live_broker()) + env.update(local_broker_env()) script = '''from lib_py.loop.runner import LoopRunner r = LoopRunner(["--creator", "headless-codex", "--task", "review check", "--job-timeout", "60"]) r.parse_args() diff --git a/tests/test_issue33_budget.py b/tests/test_issue33_budget.py index 269c002..ce86226 100644 --- a/tests/test_issue33_budget.py +++ b/tests/test_issue33_budget.py @@ -25,6 +25,8 @@ from pathlib import Path import pytest +from _local_mqtt import local_broker_env + REPO_ROOT = Path(__file__).resolve().parent.parent SKILLS = REPO_ROOT / ".agents" / "skills" WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" @@ -48,10 +50,12 @@ def _env(home: Path, extra=None, broker: bool = False) -> dict: for k in ("MAM_JOBS_DIR", "MAM_REAL_ROOT", "MAM_JOB_ID", "MAM_JOB_TIMEOUT_S", "MAM_CHILD_WAIT_S", "MAM_WAIT_GRACE_S"): env.pop(k, None) - if not broker: + for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"): + env.pop(k, None) + if broker: + env.update(local_broker_env()) # Issue #26: the session's in-process broker + else: env["MAM_ENV_FILE"] = str(home / ".mam.env.absent") - for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"): - env.pop(k, None) env.update(UNREACHABLE) if extra: env.update({k: str(v) for k, v in extra.items()}) diff --git a/tests/test_issue6_fail_closed.py b/tests/test_issue6_fail_closed.py index 3d42f1e..308b51d 100644 --- a/tests/test_issue6_fail_closed.py +++ b/tests/test_issue6_fail_closed.py @@ -16,6 +16,8 @@ from unittest.mock import MagicMock import pytest +from _local_mqtt import local_broker_env + REPO_ROOT = Path(__file__).resolve().parent.parent SKILLS = REPO_ROOT / ".agents" / "skills" WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" @@ -25,6 +27,7 @@ PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_even def _run_wrapper(args, env=None, cwd=None, timeout=60): run_env = dict(os.environ) + run_env.update(local_broker_env()) # Issue #26: the session's in-process broker run_env["DELEGATE_JOB_PYTHON"] = sys.executable run_env["AGENT_PYTHON_BIN"] = sys.executable if cwd: diff --git a/tests/test_local_mqtt.py b/tests/test_local_mqtt.py new file mode 100644 index 0000000..4252687 --- /dev/null +++ b/tests/test_local_mqtt.py @@ -0,0 +1,304 @@ +#!/usr/bin/env python3 +"""Issue #26 - the suite's in-process MQTT broker (tests/_local_mqtt.py). + +C0 conformance with the clients the repo really runs (mqtt_common.make_client, + publish_event.py, job_subscriber.py); +C2 a static guard that keeps tests off third-party brokers; +C3 the session fixture points the product's broker config at the local one. +""" +from __future__ import annotations + +import ast +import json +import os +import queue +import subprocess +import sys +import time +import uuid +from pathlib import Path + +import pytest + +import _local_mqtt +from _local_mqtt import local_broker_env + +REPO_ROOT = Path(__file__).resolve().parent.parent +TESTS = Path(__file__).resolve().parent +SCRIPTS = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts" +SKILLS = REPO_ROOT / ".agents" / "skills" +if str(SKILLS) not in sys.path: + sys.path.insert(0, str(SKILLS)) + +from lib_py.delegation import mqtt_common # noqa: E402 + + +# --------------------------------------------------------------------------- +# C3 - the session fixture +# --------------------------------------------------------------------------- + +def test_c3_the_product_config_resolves_to_the_local_broker(): + cfg = mqtt_common.broker_config_from_env() + broker = _local_mqtt.current_broker() + assert (cfg.host, cfg.port, cfg.tls) == ("127.0.0.1", broker.port, False) + assert cfg.username is None and cfg.password is None + assert "MQTT_HOST" not in os.environ + assert local_broker_env() == {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": str(broker.port), "MQTT_TLS": "0"} + + +# --------------------------------------------------------------------------- +# C0 - conformance with the repo's real clients +# --------------------------------------------------------------------------- + +class _Client: + """A real mqtt_common.make_client client, connected and looping.""" + + def __init__(self, role): + self.msgs = queue.Queue() + self.subacks = queue.Queue() + self.c = mqtt_common.make_client(role, mqtt_common.broker_config_from_env()) + 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]) + cfg = mqtt_common.broker_config_from_env() + self.c.connect(cfg.host, cfg.port, cfg.keepalive) + self.c.loop_start() + + def sub(self, filt, qos=1): + self.c.subscribe(filt, qos=qos) + return self.subacks.get(timeout=5) + + def get(self, timeout=5): + return self.msgs.get(timeout=timeout) + + def none(self, timeout=1.0): + try: + self.msgs.get(timeout=timeout) + except queue.Empty: + return True + return False + + def close(self): + self.c.loop_stop() + 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"] + 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" + 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) + 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 + a.c.publish(f"{root}/j2/state", b"retained", qos=1, retain=True).wait_for_publish(5) + c = _Client("c0-late") + try: + assert c.sub(f"{root}/+/state") == [1] + assert c.get() == (f"{root}/j2/state", b"retained", 1, True) + finally: + c.close() + a.c.publish(f"{root}/j2/state", b"", qos=1, retain=True).wait_for_publish(5) + d = _Client("c0-later") + try: + d.sub(f"{root}/+/state") + assert d.none(), "an empty retained payload deletes the retained message" + finally: + d.close() + assert f"{root}/j2/state" not in broker.retained + finally: + a.close() + b.close() + + +def _env(tmp_path: Path) -> dict: + env = dict(os.environ) + env.update(local_broker_env()) + env["PYTHONPATH"] = str(SKILLS) + env["DELEGATE_JOB_LOGS_DIR"] = str(tmp_path / "logs") + env["MAM_ENV_FILE"] = str(tmp_path / ".mam.env.absent") + return env + + +@pytest.mark.parametrize("terminal,rc", [("completed", 0), ("error", 1)]) +def test_c0_publish_event_reaches_job_subscriber_over_the_local_broker(tmp_path, terminal, rc): + """The real producer -> the real consumer; the subscriber's exit code and + console record are the evidence. `error` is the failing twin (RC_ERROR).""" + env = _env(tmp_path) + reg = tmp_path / "jobs" + jid = subprocess.run([sys.executable, str(SCRIPTS / "registry.py"), "--registry-dir", str(reg), "register", + "--prompt", "c0"], capture_output=True, text=True, env=env, cwd=str(tmp_path), + check=True).stdout.strip() + out = tmp_path / "sub.out" + with open(out, "wb") as fh: + sub = subprocess.Popen([sys.executable, str(SCRIPTS / "job_subscriber.py"), "--registry-dir", str(reg), + "--job", jid, "--timeout", "40", "--idle-timeout", "30"], + env=env, cwd=str(tmp_path), stdout=fh, stderr=subprocess.STDOUT) + try: + deadline = time.time() + 20 + while time.time() < deadline and "SUBSCRIBED" not in out.read_text(errors="replace"): + time.sleep(0.1) + record = json.loads((reg / f"{jid}.json").read_text()) + assert f"SUBSCRIBED {record['topic_prefix']}/events" in out.read_text(), out.read_text() + for ev in ("started", terminal): + res = subprocess.run([sys.executable, str(SCRIPTS / "publish_event.py"), "--registry-dir", str(reg), + "--job", jid, "--event", ev, "--detail", f"c0 {ev}", + "--expect-dispatch-token", record["dispatch_token"]], + capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=60) + assert res.returncode == 0, res.stderr + assert sub.wait(timeout=30) == rc, out.read_text() + finally: + if sub.poll() is None: + sub.kill() + sub.wait(timeout=10) + text = out.read_text() + assert f"job={jid} seq=1 started" in text, text + assert f"job={jid} seq=2 {terminal}" in text and f"c0 {terminal}" in text, text + + +# --------------------------------------------------------------------------- +# C2 - static guard over tests/*.py +# --------------------------------------------------------------------------- + +# Tests that pin the product's public default broker by name. None today; +# add one here (with its reason) rather than weakening the scan. +PUBLIC_BROKER_ALLOWLIST: dict = {} +PUBLIC_BROKER = "broker." + "hivemq.com" +REDIRECT_KEYS = {"MQTT_BROKER", "MQTT_HOST"} + + +def _const(node): + return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else None + + +def _dict_has_key(node) -> bool: + return isinstance(node, ast.Dict) and any(_const(k) in REDIRECT_KEYS for k in node.keys if k is not None) + + +def _module_env_dicts(tree) -> set: + """Module-level names bound to a dict literal that names a broker (UNREACHABLE).""" + names = set() + for node in tree.body: + if isinstance(node, ast.Assign) and _dict_has_key(node.value): + names.update(t.id for t in node.targets if isinstance(t, ast.Name)) + return names + + +def _names_broker(node, env_dicts) -> bool: + if _dict_has_key(node): + return True + if isinstance(node, ast.Name) and node.id in env_dicts: + return True + if isinstance(node, ast.Call): + fn = node.func.id if isinstance(node.func, ast.Name) else getattr(node.func, "attr", None) + if fn == "local_broker_env": + return True + if fn == "dict" and any(k.arg in REDIRECT_KEYS for k in node.keywords): + return True + return False + + +def _deletes_and_resets(func, env_dicts): + loop_keys = {} # loop variable -> iterates over a literal that names a redirect key + for node in ast.walk(func): + if isinstance(node, ast.For) and isinstance(node.target, ast.Name) and isinstance(node.iter, (ast.Tuple, ast.List)): + if any(_const(e) in REDIRECT_KEYS for e in node.iter.elts): + loop_keys[node.target.id] = True + deletes = resets = False + for node in ast.walk(func): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + attr, args = node.func.attr, node.args + first = args[0] if args else None + names_key = _const(first) in REDIRECT_KEYS or (isinstance(first, ast.Name) and first.id in loop_keys) + if attr in ("pop", "delenv") and names_key: + deletes = True + if attr == "setenv" and _const(first) in REDIRECT_KEYS: + resets = True + if attr == "update" and (any(_names_broker(a, env_dicts) for a in args) + or any(k.arg in REDIRECT_KEYS for k in node.keywords)): + resets = True + elif isinstance(node, ast.Delete): + if any(isinstance(t, ast.Subscript) and _const(t.slice) in REDIRECT_KEYS for t in node.targets): + deletes = True + elif isinstance(node, (ast.Assign, ast.AugAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + if any(isinstance(t, ast.Subscript) and _const(t.slice) in REDIRECT_KEYS for t in targets): + resets = True + elif isinstance(node, ast.Dict) and any(k is None and _names_broker(v, env_dicts) + for k, v in zip(node.keys, node.values)): + resets = True # {**os.environ, **UNREACHABLE} + return deletes, resets + + +def _scan(path: Path): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + public, stripping = [], [] + for node in ast.walk(tree): + if _const(node) is not None and PUBLIC_BROKER in node.value: + public.append(node.lineno) + env_dicts = _module_env_dicts(tree) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + deletes, resets = _deletes_and_resets(node, env_dicts) + if deletes and not resets: + stripping.append(f"{node.name}:{node.lineno}") + return public, stripping + + +def test_c2_no_test_names_the_public_broker_or_strips_the_local_one(): + """(a) the public default broker's name appears only in allowlisted + tests; (b) no helper deletes MQTT_BROKER/MQTT_HOST without naming a + broker again in the same function - without MQTT_BROKER/MQTT_HOST the + product falls back to the public default.""" + public, stripping = {}, {} + for path in sorted(TESTS.glob("*.py")): + p, s = _scan(path) + if p and path.name not in PUBLIC_BROKER_ALLOWLIST: + public[path.name] = p + if s: + stripping[path.name] = s + assert not public, f"tests naming the public broker: {public}" + assert not stripping, f"helpers that drop the broker and fall back to the public default: {stripping}" + + +def test_c2_the_guard_catches_what_it_is_for(tmp_path): + """The failing twins: each forbidden shape is reported, and the allowed + shapes (pop then UNREACHABLE / local_broker_env / setenv) are not.""" + bad = tmp_path / "bad.py" + bad.write_text( + "import os\n" + f"URL = '{PUBLIC_BROKER}'\n" + "def strip_loop(env):\n" + " for k in ('MQTT_BROKER', 'MQTT_PORT'):\n" + " env.pop(k, None)\n" + " return env\n" + "def strip_direct(mp):\n" + " mp.delenv('MQTT_HOST', raising=False)\n" + "def strip_del(env):\n" + " del env['MQTT_BROKER']\n") + good = tmp_path / "good.py" + good.write_text( + "UNREACHABLE = {'MQTT_BROKER': '127.0.0.1', 'MQTT_PORT': '9'}\n" + "def a(env):\n" + " for k in ('MQTT_BROKER', 'MQTT_HOST'):\n" + " env.pop(k, None)\n" + " env.update(UNREACHABLE)\n" + "def b(env):\n" + " env.pop('MQTT_BROKER', None)\n" + " env.update(local_broker_env())\n" + "def c(mp):\n" + " mp.delenv('MQTT_HOST')\n" + " mp.setenv('MQTT_BROKER', 'x')\n") + public, stripping = _scan(bad) + assert public == [2] + assert stripping == ["strip_loop:3", "strip_direct:7", "strip_del:9"] + assert _scan(good) == ([], [])