coulomb/mail-address-semantics
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

MAS-WP-0009 T01: independent JS implementation (impl/js), differential test, SCP-9

Browse source 
impl/js (Node, zero deps, written from the specs): parser, v0.2
statute-aware rules, signing, outside master. Passes addresses.json,
statute-aware.json and auth.json unchanged; its master passes the 11
slave-setup mode flows against the Python email side (ExternalMaster).
Differential testing vs mas_core found two unstated stand-in rules, now
SCP-9 in the v0.2 draft with fixtures si-011..013, dt-109; ABNF comments
no longer say v0.1. Runner: external masters, step errors fail the flow.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5787@bnt-lap001
Assistant-Session: 5db26b82-138b-40cc-9856-4d5eb0f752fd
This commit is contained in:
tegwick 2026-10-19 13:22:22 +02:00
parent ecb1ba47f6
commit a841ceec73
23 changed files with 938 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

81
mas_modes/runner.py
View file

@ -206,8 +206,9 @@ def _master(h, cmd, ctx):
raise KeyError(f"unknown master command {c!r}")
def run_flow(flow: dict, doc: Optional[dict] = None) -> list[str]:
"""Run one flow from the bundle (FLOWS.md). Returns the problems found."""
def run_flow(flow: dict, doc: Optional[dict] = None, master_factory=None) -> list[str]:
"""Run one flow from the bundle (FLOWS.md). Returns the problems found.
master_factory swaps in another outside master (e.g. the JS one)."""
doc = doc or _load("conformance/flows.json")
bad = undefined_keys(flow)
if bad:
@ -215,41 +216,51 @@ def run_flow(flow: dict, doc: Optional[dict] = None) -> list[str]:
keys = {p: bytes.fromhex(k) for p, k in doc["peers"].items()}
with resources.as_file(resources.files(__package__).joinpath("conformance", doc["statute"])) as st:
h = Harness(st, _roles(doc["setups"][flow["setup"]]), scenario=flow["scenario"],
auto_assign=False, max_sent=flow.get("max_sent"), keys=keys)
auto_assign=False, max_sent=flow.get("max_sent"), keys=keys,
master_factory=master_factory)
ctx: dict = {}
problems = []
try:
for n, step in enumerate(flow["steps"], 1):
seen = (len(h.callback.received), len(h.master.received))
event, response = None, None
if "mail" in step:
m = _fill(step["mail"], ctx)
event = h.mail(m["to"], m.get("text", ""), subject=m.get("subject", "Help needed"),
dmarc=m.get("dmarc", "pass"))
elif "mode" in step:
h.callback.run()
elif "advance" in step:
h.advance(step["advance"])
elif "master" in step:
response = _master(h, _fill(step["master"], ctx), ctx)
elif "inject" in step:
response = _inject(h, step["inject"], ctx)
_capture(ctx, h.callback.received[seen[0]:] + h.master.received[seen[1]:])
if response is not None:
_capture(ctx, [response[1]])
for key, want in step.get("expect", {}).items():
got = _observe(h, key, event, response)
if key == "master_received":
got = {t: got.get(t, 0) for t in want}
elif key == "last_mail_contains":
got, want = (want in (got or "")), True
if got != want:
problems.append(f"step {n} {key}: got {got!r}, expected {want!r}")
try:
_step(h, n, step, ctx, problems)
except Exception as e: # a broken peer must fail the flow, not the run
problems.append(f"step {n} raised {type(e).__name__}: {e}")
break
finally:
h.close()
return problems
def _step(h, n, step, ctx, problems):
"""One flow step (FLOWS.md), then its expectations."""
seen = (len(h.callback.received), len(h.master.received))
event, response = None, None
if "mail" in step:
m = _fill(step["mail"], ctx)
event = h.mail(m["to"], m.get("text", ""), subject=m.get("subject", "Help needed"),
dmarc=m.get("dmarc", "pass"))
elif "mode" in step:
h.callback.run()
elif "advance" in step:
h.advance(step["advance"])
elif "master" in step:
response = _master(h, _fill(step["master"], ctx), ctx)
elif "inject" in step:
response = _inject(h, step["inject"], ctx)
_capture(ctx, h.callback.received[seen[0]:] + h.master.received[seen[1]:])
if response is not None:
_capture(ctx, [response[1]])
for key, want in step.get("expect", {}).items():
got = _observe(h, key, event, response)
if key == "master_received":
got = {t: got.get(t, 0) for t in want}
elif key == "last_mail_contains":
got, want = (want in (got or "")), True
if got != want:
problems.append(f"step {n} {key}: got {got!r}, expected {want!r}")
def _observe(h, key, event, response):
body = (response[1] if response else None) or {}
if key == "event_accepted":
@ -284,13 +295,19 @@ def _observe(h, key, event, response):
raise KeyError(f"unknown expectation {key!r}")
def run_mode_corpus() -> Report:
def run_mode_corpus(master_factory=None, roles=("master", "slave")) -> Report:
"""All fixtures and flows. With master_factory, only the flows of the given
setup roles run, against that outside master."""
report = Report()
check_messages(report)
check_auth(report)
if master_factory is None:
check_messages(report)
check_auth(report)
doc = _load("conformance/flows.json")
for flow in doc["flows"]:
problems = run_flow(flow, doc)
setup_roles = {c["role"] for c in doc["setups"][flow["setup"]]["roles"].values()}
if not setup_roles & set(roles):
continue
problems = run_flow(flow, doc, master_factory)
if problems:
report.failures.append((flow["id"], "; ".join(problems)))
else: