tegwick
eb10f54fd2
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 252470@bnt-lap001 Assistant-Session: ad628c8f-4be2-4979-94c9-76c7ed6f8241
159 lines
6.9 KiB
Python
159 lines
6.9 KiB
Python
import socket
|
|
|
|
import pytest
|
|
|
|
from mas_conformance import statute_document
|
|
from mas_core.parser import parse_address, parse_local_part
|
|
from mas_core.policy import resolve
|
|
from mas_core.statute import build_statute
|
|
|
|
EX = build_statute(statute_document("valid-example.json"))
|
|
GS = build_statute(statute_document("valid-great-service.json"))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def no_network(monkeypatch):
|
|
def deny(*a, **k):
|
|
raise AssertionError("policy attempted network I/O")
|
|
for name in ("socket", "create_connection", "getaddrinfo"):
|
|
monkeypatch.setattr(socket, name, deny)
|
|
|
|
|
|
def codes(sem):
|
|
return {f.code for f in sem.failures}
|
|
|
|
|
|
def test_no_statute_stays_opaque():
|
|
s = resolve(parse_address("bernd.worsch@example.org"))
|
|
assert s.kind == "opaque" and s.kind_source == "none" and not s.namespace_confirmed
|
|
assert any("punctuation" in u for u in s.unknowns)
|
|
|
|
|
|
def test_statute_registry_classifies():
|
|
assert resolve(parse_local_part("bernd.worsch"), EX).kind == "person"
|
|
assert resolve(parse_local_part("research-companion"), EX).kind == "actor"
|
|
assert resolve(parse_local_part("happy-helper.v2"), EX).kind == "opaque"
|
|
|
|
|
|
def test_registry_uses_base_target_not_faceted_address():
|
|
s = resolve(parse_local_part("support+urgent"), EX)
|
|
assert s.kind == "function" and s.registry_key == "support" and s.facet == "urgent"
|
|
|
|
|
|
def test_resource_with_statute():
|
|
s = resolve(parse_local_part("ticket=PR42"), EX)
|
|
assert (s.kind, s.resource_type, s.resource_id) == ("resource", "ticket", "PR42")
|
|
|
|
|
|
def test_resource_without_statute_is_unconfirmed():
|
|
s = resolve(parse_local_part("ticket=PR42"))
|
|
assert s.kind == "opaque" and s.ok and s.resource_type == "ticket"
|
|
assert any("no application semantics" in u for u in s.unknowns)
|
|
|
|
|
|
def test_unknown_and_disabled_resource_types():
|
|
assert codes(resolve(parse_local_part("order=1"), GS)) == {"UnknownResourceType"}
|
|
doc = statute_document("valid-great-service.json")
|
|
doc["resource_types"]["ticket"]["enabled"] = False
|
|
assert codes(resolve(parse_local_part("ticket=1"), build_statute(doc))) == {"ResourceTypeNotEnabled"}
|
|
|
|
|
|
def test_profile_not_enabled():
|
|
doc = statute_document("valid-great-service.json")
|
|
doc["profiles"]["p2"] = False
|
|
doc["interaction_modes"] = {}
|
|
s = resolve(parse_local_part("ticket!callback"), build_statute(doc))
|
|
assert codes(s) == {"ProfileNotEnabled"}
|
|
assert s.parsed.required_profile.value == "P2" # syntax still valid
|
|
|
|
|
|
def test_interaction_mode():
|
|
assert resolve(parse_local_part("ticket!callback"), GS).ok
|
|
assert codes(resolve(parse_local_part("ticket!chat"), GS)) == {"InteractionModeNotEnabled"}
|
|
assert any("unconfirmed" in u for u in resolve(parse_local_part("ticket!callback")).unknowns)
|
|
|
|
|
|
def test_equivalence_only_when_declared():
|
|
assert resolve(parse_local_part("Bernd.Worsch"), EX).kind == "opaque"
|
|
doc = statute_document("valid-example.json")
|
|
doc["equivalence"] = {"local_part_case": "ascii-insensitive", "dot_folding": True}
|
|
st = build_statute(doc)
|
|
s = resolve(parse_local_part("Bernd.Worsch"), st)
|
|
assert s.kind == "person" and s.equivalence_key == "berndworsch"
|
|
assert resolve(parse_local_part("ticket=GS.1"), st).equivalence_key == "ticket=gs.1"
|
|
|
|
|
|
def test_local_registry_fallback_and_precedence():
|
|
local = {"addresses": {"support": {"kind": "group"}, "ops": {"kind": "system"}},
|
|
"resource_types": {"order": {"enabled": True}}}
|
|
assert resolve(parse_local_part("support"), EX, local).kind == "function" # statute wins
|
|
s = resolve(parse_local_part("ops"), EX, local)
|
|
assert (s.kind, s.kind_source) == ("system", "local-registry")
|
|
assert resolve(parse_local_part("order=7"), None, local).kind == "resource"
|
|
|
|
|
|
def test_deterministic():
|
|
p = parse_local_part("ticket=PR42+comments")
|
|
assert resolve(p, EX) == resolve(p, EX)
|
|
|
|
|
|
def test_intake_for_is_reported():
|
|
# SCP-3: the statute declares which function is a type's intake
|
|
sem = resolve(parse_address("ticket!callback@great-service.example"), GS)
|
|
assert sem.kind == "function" and sem.intake_for == "ticket" and sem.interaction == "callback"
|
|
assert resolve(parse_address("support@great-service.example"), GS).intake_for is None
|
|
assert resolve(parse_address("ticket@great-service.example"), None).intake_for is None
|
|
|
|
|
|
# -- v0.2 stand-ins (spec v0.2 section 6.3) ----------------------------------------
|
|
|
|
def _v02_statute(resource=True, interaction=True, id_case="insensitive"):
|
|
doc = statute_document("valid-great-service.json")
|
|
doc["mas_version"] = "0.2"
|
|
doc["alternate_syntax"] = {"resource": resource, "interaction": interaction}
|
|
doc["equivalence"]["resource_id_case"] = id_case
|
|
return build_statute(doc, "great-service.example")
|
|
|
|
|
|
V2 = _v02_statute()
|
|
|
|
|
|
@pytest.mark.parametrize("raw, canonical, stand_ins", [
|
|
("ticket---callback", "ticket!callback", ("interaction",)),
|
|
("ticket--gs1---callback", "ticket=gs1!callback", ("interaction", "resource")),
|
|
("ticket--a---b", "ticket=a---b", ("resource",)), # b is not an enabled mode
|
|
("ticket--gs1+comments", "ticket=gs1+comments", ("resource",)),
|
|
("ticket--gs-85qj8wc09p+comments", "ticket=gs-85qj8wc09p+comments", ("resource",)),
|
|
("support---callback", "support!callback", ("interaction",)),
|
|
("invoice--x1", "invoice--x1", ()), # undeclared type
|
|
("ticket=gs1!callback", "ticket=gs1!callback", ()), # already canonical
|
|
])
|
|
def test_stand_in_table(raw, canonical, stand_ins):
|
|
sem = resolve(parse_address(f"{raw}@great-service.example"), V2)
|
|
assert sem.to_dict()["canonical_local_part"] == canonical and sem.stand_ins == stand_ins
|
|
|
|
|
|
def test_stand_in_resource_correlates_like_canonical():
|
|
alt = resolve(parse_address("ticket--GS-85QJ8WC09P+comments@great-service.example"), V2)
|
|
canon = resolve(parse_address("ticket=gs-85qj8wc09p@great-service.example"), V2)
|
|
assert alt.kind == "resource" and alt.resource_id == "GS-85QJ8WC09P" and alt.facet == "comments"
|
|
assert alt.equivalence_key == canon.equivalence_key == "ticket=gs-85qj8wc09p"
|
|
|
|
|
|
def test_stand_in_intake_reaches_registry():
|
|
sem = resolve(parse_address("ticket---callback@great-service.example"), V2)
|
|
assert sem.kind == "function" and sem.intake_for == "ticket" and sem.interaction == "callback"
|
|
|
|
|
|
def test_stand_ins_need_the_statute_switch():
|
|
off = _v02_statute(resource=False, interaction=False, id_case="exact")
|
|
for raw in ("ticket--gs1", "ticket---callback"):
|
|
sem = resolve(parse_address(f"{raw}@great-service.example"), off)
|
|
assert sem.stand_ins == () and sem.kind == "opaque"
|
|
assert resolve(parse_address("ticket--gs1@great-service.example"), GS).stand_ins == () # v0.1
|
|
assert resolve(parse_address("ticket--gs1@great-service.example"), None).stand_ins == ()
|
|
|
|
|
|
def test_resource_id_case_exact_keeps_case():
|
|
exact = _v02_statute(id_case="exact")
|
|
assert resolve(parse_address("ticket=GS-1@great-service.example"), exact).equivalence_key == "ticket=GS-1"
|