mail-address-semantics/tests/test_policy.py

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)