tegwick
ecb1ba47f6
conformance/FLOWS.md defines setup, steps, captures and expectations, all observable at the contract boundary. flows.json carries setups (peer ids), fixed test keys, statute, scenario and explicit assign steps; scenarios.json holds the callback scenarios (the stub loads them). CONTRACT.md defines the case.assign response. Runner reads only the bundle and rejects undefined keys; new flow s-006 (held over cap). 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
51 lines
2.2 KiB
Python
51 lines
2.2 KiB
Python
"""Mode contract bundle (MAS-WP-0010 T02/T03/T07): vendored copies, fixtures, flows,
|
|
and the gateway event schema checked against real gateway output."""
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mas_core.discovery import Response, discover
|
|
from mas_gateway.core import Resolver, build_event, message_meta_from_rfc822
|
|
from mas_modes import event_errors
|
|
from mas_modes import _load
|
|
from mas_modes.runner import EXPECT, STEPS, run_mode_corpus, undefined_keys
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BUNDLE = ROOT / "mas-modes-v0.1-spec"
|
|
VENDORED = ROOT / "mas_modes"
|
|
|
|
|
|
@pytest.mark.parametrize("rel", sorted(str(p.relative_to(BUNDLE)) for p in BUNDLE.glob("*/*.*")))
|
|
def test_vendored_copy_matches_bundle(rel):
|
|
assert (BUNDLE / rel).read_bytes() == (VENDORED / rel).read_bytes()
|
|
|
|
|
|
def test_flows_are_self_describing():
|
|
doc = _load("conformance/flows.json")
|
|
scenarios = _load("conformance/scenarios.json")["scenarios"]
|
|
for flow in doc["flows"]:
|
|
assert undefined_keys(flow) == [], flow["id"]
|
|
assert flow["setup"] in doc["setups"] and flow["scenario"] in scenarios, flow["id"]
|
|
text = (BUNDLE / "conformance/FLOWS.md").read_text()
|
|
for key in STEPS | EXPECT:
|
|
assert f"`{key}`" in text or f'"{key}"' in text, f"{key} is not defined in FLOWS.md"
|
|
|
|
|
|
def test_mode_corpus():
|
|
rep = run_mode_corpus()
|
|
assert rep.ok, rep.failures
|
|
assert rep.passed >= 170
|
|
|
|
|
|
@pytest.mark.parametrize("recipient", [
|
|
"ticket!callback@great-service.example", "ticket=GS-1+comments@great-service.example",
|
|
"bernd.worsch@great-service.example", "ticket=PR42$4@great-service.example",
|
|
"x@unknown.invalid"])
|
|
def test_gateway_events_match_schema(recipient):
|
|
raw = (b"Authentication-Results: great-service.example; dmarc=pass header.from=example.net\r\n"
|
|
b"From: c@example.net\r\nTo: x@great-service.example\r\nSubject: s\r\n\r\nbody\r\n")
|
|
offline = lambda host, **kw: discover(host, transport=lambda u, t: Response(404, {}, b"")) # noqa: E731
|
|
resolver = Resolver.with_statute_files(
|
|
[ROOT / "examples/statute/great-service.example/.well-known/mas"], discover_fn=offline)
|
|
event = build_event(recipient, message_meta_from_rfc822(raw, "great-service.example"), resolver)
|
|
assert event_errors(event) == [], event
|