test(node): record what the session does with every message it can be sent

A characterisation test over every MNP type, peer state and message shape,
and every signed operation's answer: the reply, the audit, the handler
started. It pins dispatch order and the admin table so they can be moved.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 221 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_dispatch_golden.py b/packages/meshbay-node/tests/test_dispatch_golden.py
new file mode 100644
index 0000000..b31b63b
--- /dev/null
+++ b/packages/meshbay-node/tests/test_dispatch_golden.py
@@ -0,0 +1,221 @@
+"""
+What the session does with each message it can be sent, recorded once and held.
+
+A characterisation test: it states no rule of its own, it pins the behaviour
+the node has, so that code can be moved and restructured underneath it —
+`_dispatch_message` and `_do_admin_response` especially, whose order of checks
+is itself the rule (a bundle fetch is served before the proof, a join request
+with a nonce, everything else only after it).
+
+Every message type the protocol names, plus two that it does not, is sent to a
+fresh session in each state a peer can be in, in four shapes: bare, and with every
+field any handler reads set to a string, to a number and to a list. What is recorded
+is the reply the peer would receive (with its `req_id`), what was audited, which
+coroutine was started in the background — started, not run: what those do is
+the business of the tests written for them — and the lines logged, as
+templates, since their arguments carry object addresses.
+
+Then every signed operation's answer is sent against a challenge that is
+pending for it, which is the table `_do_admin_response` dispatches on.
+
+The recording is `golden/dispatch.json`. When a change to this behaviour is
+intended, regenerate it with MESHBAY_GOLDEN_WRITE=1 and read the diff: that
+diff is the behaviour change, stated message by message.
+"""
+
+import base64
+import json
+import logging
+import os
+import struct
+import time
+from pathlib import Path
+
+import msgpack
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common import adminop
+from meshbay_common.protocol import MNP
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+GOLDEN = Path(__file__).parent / "golden" / "dispatch.json"
+GROUP = "g" * 32
+REQ_ID = 4242
+# Values that differ between runs and mean nothing for the comparison.
+VOLATILE = {"nonce", "op_id", "ts"}
+
+
+class _Channel:
+    readyState = "open"
+
+    def __init__(self):
+        self.sent = []
+
+    def send(self, data: bytes) -> None:
+        (n,) = struct.unpack(">I", data[:4])
+        self.sent.append(msgpack.unpackb(data[4:4 + n], raw=False))
+
+
+class _PC:
+    connectionState = "connected"
+    iceConnectionState = "connected"
+    remoteDescription = None
+    localDescription = None
+    sctp = None
+
+
+class _Log(logging.Handler):
+    def __init__(self):
+        super().__init__(level=logging.DEBUG)
+        self.lines = []
+
+    def emit(self, record):
+        if record.levelno >= logging.INFO:
+            self.lines.append(f"{record.levelname} {record.msg}")
+
+
+def _clean(value):
+    if isinstance(value, dict):
+        return {k: ("<volatile>" if k in VOLATILE else _clean(v))
+                for k, v in sorted(value.items())}
+    if isinstance(value, (list, tuple)):
+        return [_clean(v) for v in value]
+    if isinstance(value, bytes):
+        return f"<{len(value)} bytes>"
+    return value
+
+
+def _ctx():
+    return {
+        "sk_node": Ed25519PrivateKey.from_private_bytes(b"\x01" * 32),
+        "groups": {GROUP: {}},
+    }
+
+
+def _session(state: str):
+    s = WebRTCPeerSession(_PC(), _ctx(), peer_id="peer")
+    s._channel = _Channel()
+    s.spawned, s.audited = [], []
+
+    def spawn(coro):
+        # The function's name, not its qualified name: which class it sits in is
+        # exactly what may change underneath this recording.
+        s.spawned.append(coro.cr_code.co_name)
+        coro.close()
+
+    s._spawn = spawn
+    s._audit = lambda event, detail="": s.audited.append([event, detail])
+    if state in ("challenged", "member", "operator"):
+        s._gek_challenge = b"\x02" * 32
+        s._nonce_node = b"\x02" * 32
+        s._nonce_client = b"\x03" * 32
+    if state in ("member", "operator"):
+        s._gek_challenge = None
+        s._user_id = "alice"
+        s._username = "Alice"
+        s._group_id = GROUP
+        s._device_confirmed = True
+        s._pinned_pk = base64.b64encode(b"\x04" * 32).decode()
+    if state == "operator":
+        s._ctx["has_admin_authority"] = True
+    return s
+
+
+def _run(s, msg: dict) -> dict:
+    handler = _Log()
+    logger = logging.getLogger("meshbay_node.transport.webrtc_server")
+    # Pinned rather than assumed: an earlier test may have raised the level or
+    # disabled the logger, and a recording that depends on that is not one.
+    saved = logger.level, logger.disabled, logger.propagate
+    logger.setLevel(logging.DEBUG)
+    logger.disabled = False
+    logger.propagate = False
+    logger.addHandler(handler)
+    try:
+        s._handle_message(msg)
+    finally:
+        logger.removeHandler(handler)
+        logger.level, logger.disabled, logger.propagate = saved
+    return _clean({"sent": s._channel.sent, "audit": s.audited,
+                   "spawned": s.spawned, "log": handler.lines})
+
+
+def _message_types() -> list[str]:
+    names = {v for k, v in vars(MNP).items() if k.isupper() and isinstance(v, str)}
+    return sorted(names | {"client_diag", "no_such_type"})
+
+
+def _shapes(keys: list[str]) -> dict[str, dict]:
+    return {
+        "bare": {},
+        "strings": {k: "x" for k in keys},
+        "numbers": {k: 7 for k in keys},
+        "lists": {k: ["x"] for k in keys},
+    }
+
+
+def _admin_ops() -> list[str]:
+    return sorted(v for k, v in vars(adminop).items()
+                  if k.startswith("OP_") and isinstance(v, str))
+
+
+def _record(keys: list[str]) -> dict:
+    out = {"keys": keys, "dispatch": {}, "admin": {}}
+    for mtype in _message_types():
+        for state in ("fresh", "challenged", "member", "operator"):
+            for shape, fields in _shapes(keys).items():
+                msg = {**fields, "type": mtype, "req_id": REQ_ID}
+                out["dispatch"][f"{mtype} | {state} | {shape}"] = _run(_session(state), msg)
+    for op in _admin_ops() + ["no_such_op"]:
+        s = _session("operator")
+        s._admin_ops["the-op"] = {"op": op, "subject": "subject", "nonce": b"\x05" * 32,
+                                  "ts": int(time.time()), "payload": {}, "group_id": GROUP}
+        out["admin"][op] = _run(s, {"type": MNP.ADMIN_RESPONSE, "op_id": "the-op",
+                                    "signature": base64.b64encode(b"\x06" * 64).decode(),
+                                    "req_id": REQ_ID})
+    s = _session("operator")
+    s._admin_ops["old"] = {"op": adminop.OP_FILE_DELETE, "subject": "s", "nonce": b"",
+                           "ts": 0, "payload": {}, "group_id": GROUP}
+    out["admin"]["<expired>"] = _run(s, {"type": MNP.ADMIN_RESPONSE, "op_id": "old",
+                                         "signature": "AAAA", "req_id": REQ_ID})
+    s = _session("operator")
+    s._admin_ops["bad"] = {"op": adminop.OP_FILE_DELETE, "subject": "s", "nonce": b"",
+                           "ts": int(time.time()), "payload": {}, "group_id": GROUP}
+    out["admin"]["<bad signature encoding>"] = _run(
+        s, {"type": MNP.ADMIN_RESPONSE, "op_id": "bad", "signature": "%%%", "req_id": REQ_ID})
+    return out
+
+
+def test_every_message_is_handled_as_recorded():
+    golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
+    if os.environ.get("MESHBAY_GOLDEN_WRITE") == "1":
+        # The field list is kept from the recording, not re-derived from the
+        # handlers: it has to be the same input before and after a change.
+        GOLDEN.write_text(json.dumps(_record(golden["keys"]), indent=1, sort_keys=True)
+                          + "\n", encoding="utf-8")
+        golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
+    now = _record(golden["keys"])
+
+    assert sorted(now["dispatch"]) == sorted(golden["dispatch"]), (
+        "the set of message types changed; regenerate on purpose if it should")
+    differ = [k for k in golden["dispatch"] if now["dispatch"][k] != golden["dispatch"][k]]
+    differ += [f"admin {k}" for k in sorted(set(golden["admin"]) | set(now["admin"]))
+               if now["admin"].get(k) != golden["admin"].get(k)]
+    detail = "\n".join(
+        f"  {k}\n    was: {golden['dispatch'].get(k) or golden['admin'].get(k[6:])}\n"
+        f"    now: {now['dispatch'].get(k) or now['admin'].get(k[6:])}"
+        for k in differ[:10])
+    assert not differ, f"{len(differ)} case(s) behave differently:\n{detail}"
+
+
+def test_the_recording_exercises_the_session():
+    """A recording of nothing would hold nothing."""
+    golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
+    cases = golden["dispatch"].values()
+    assert len(golden["keys"]) > 50
+    assert sum(1 for c in cases if c["spawned"]) > 100, "no handler was ever started"
+    assert sum(1 for c in cases if c["sent"]) > 300, "nothing was ever answered"
+    assert all(any(m.get("req_id") == REQ_ID for m in c["sent"])
+               for c in cases if c["sent"]), "a reply went out without its req_id"
+    spawned = {c["spawned"][0] for c in golden["admin"].values() if c["spawned"]}
+    assert len(spawned) == len(_admin_ops()), (
+        "some signed operation reaches no executor, or two reach the same one")