tegwick
a2465643ba
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
105 lines
4.2 KiB
Python
105 lines
4.2 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
|