tegwick
59815701c1
- mas_core: P2-maximal parser with required_profile, generator, statute schema + semantic lint + defaults-only inheritance, exact-host discovery with typed failures and same-origin redirects, pure policy resolver - mas_conformance: vendored spec fixtures (drift-tested) + property and statute cases - mas CLI: parse / explain / discover / lint-statute / test - mas-gateway: generic mail -> MAS event -> HTTP (MTA pipe + webhook) - examples: static great-service.example statute, reference ticket service - docs/experiment-review.md: first-pass findings; live MTA test still open Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 623586@bnt-lap001 Assistant-Session: 10b30928-5b3b-4c11-9934-5ecca07367d8
82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
import importlib.util
|
|
import json
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mas_cli.main import main
|
|
from mas_core.discovery import Response, discover, http_transport
|
|
from mas_core.statute import lint
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
STATIC = ROOT / "examples/statute/great-service.example"
|
|
STATUTE = STATIC / ".well-known/mas"
|
|
|
|
|
|
def run(capsys, *argv):
|
|
code = main(list(argv))
|
|
return code, capsys.readouterr().out
|
|
|
|
|
|
def test_parse_json(capsys):
|
|
code, out = run(capsys, "parse", "ticket!callback", "--json")
|
|
d = json.loads(out)
|
|
assert code == 0 and d["required_profile"] == "P2"
|
|
assert run(capsys, "parse", "ticket!callback", "--profile", "p1")[0] == 1
|
|
assert run(capsys, "parse", "ticket=PR42$1")[0] == 1
|
|
|
|
|
|
def test_explain_separates_layers_and_never_guesses(capsys, monkeypatch):
|
|
import mas_cli.main as cli
|
|
monkeypatch.setattr(cli, "discover", lambda host, **kw: discover(
|
|
host, transport=lambda u, t: Response(404, {}, b"")))
|
|
code, out = run(capsys, "explain", "bernd.worsch@example.org")
|
|
assert code == 0
|
|
for section in ("Syntax:", "Discovery:", "Semantics:", "Unknowns:"):
|
|
assert section in out
|
|
assert "no valid MAS statute found" in out and "NotFound" in out
|
|
assert "kind: opaque" in out and "person" not in out
|
|
|
|
|
|
def test_explain_with_statute_file(capsys):
|
|
code, out = run(capsys, "explain", "ticket=GS-41872+comments@great-service.example",
|
|
"--statute", str(STATUTE), "--json")
|
|
sem = json.loads(out)["semantics"]
|
|
assert sem["kind"] == "resource" and sem["resource_id"] == "GS-41872" and sem["facet"] == "comments"
|
|
|
|
|
|
def test_lint_exit_codes(capsys):
|
|
st = ROOT / "mas_conformance/statutes"
|
|
assert run(capsys, "lint-statute", str(STATUTE), "--host", "great-service.example")[0] == 0
|
|
code, out = run(capsys, "lint-statute", str(st / "semantic-p2-disabled-modes.json"))
|
|
assert code == 1 and "semantic error" in out
|
|
code, out = run(capsys, "lint-statute", str(st / "schema-version.json"))
|
|
assert code == 1 and "schema error" in out
|
|
|
|
|
|
def test_conformance_command(capsys):
|
|
code, out = run(capsys, "test")
|
|
assert code == 0 and "FAIL" not in out
|
|
|
|
|
|
def test_static_statute_lints_clean():
|
|
assert lint(json.loads(STATUTE.read_text()), "great-service.example") == []
|
|
|
|
|
|
def test_static_statute_served_and_discovered():
|
|
spec = importlib.util.spec_from_file_location("serve", ROOT / "examples/statute/serve.py")
|
|
serve = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(serve)
|
|
# Serve a copy whose namespace is the local test host.
|
|
srv = serve.make_server(str(STATIC), port=0)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
try:
|
|
port = srv.server_address[1]
|
|
r = discover("127.0.0.1", scheme="http", port=port)
|
|
assert not r.ok and r.kind.value == "NamespaceMismatch" # served for great-service.example
|
|
r = discover("great-service.example", scheme="http", port=port,
|
|
transport=lambda url, t: http_transport(url.replace("great-service.example", "127.0.0.1"), t))
|
|
assert r.ok and r.statute.interaction_mode_enabled("callback")
|
|
finally:
|
|
srv.shutdown()
|