tmpl/multi-agent-mux
1
0 Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
325b056057fcdd717d2c81017fc5ef57183cb4d0
multi-agent-mux/tests/test_local_mqtt.py
T
GodopuandClaude Opus 5.5 b58edace11 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 <noreply@anthropic.com>
2026-10-01 01:04:19 +00:00

345 lines
14 KiB
Python
Raw Blame History

#!/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.
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)
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 _c0_conformance(broker):
root = "mam/t26-" + uuid.uuid4().hex[:8]
a, b = _Client("c0-pub"), _Client("c0-sub")
try:
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)"
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
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 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())
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) == ([], [])
Reference in new issue View Git Blame Copy Permalink