import ast import importlib.util import os import subprocess import sys import tempfile import unittest from datetime import timedelta from pathlib import Path from unittest.mock import patch from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter from user_engine.adapters.local import LocalAuthorizationCheckPort from user_engine.domain import AuthorizationEffect, utc_now from user_engine.layer_yaml import load_mapping, load_mapping_text from user_engine.pep_stance import ( ALLOW_BINDING, ALLOW_LIFETIME, STANCE, UNREACHABLE_STANCE, VERDICT_CACHING, ) ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts" / "check_layer_conformance.py" LAYER = ROOT / "layer.yaml" STANCE_FILE = ROOT / "pep-stance.yaml" RUNTIME = ROOT / "src" / "user_engine" / "runtime.py" class LayerDeclarationTests(unittest.TestCase): def test_declares_engine_pip_in_own_voice(self): data = load_mapping(LAYER) self.assertEqual(data["repository"], "user-engine") self.assertEqual(_fold(data["layer"]), "engine") self.assertEqual(data["role"], "pip") self.assertEqual(data["tooling_contacts"], []) self.assertEqual(data["pep_stance"], "pep-stance.yaml") self.assertTrue(data["pep_shape"]) self.assertEqual(data["declared_by"], "INTENT.md") def test_intent_frontmatter_matches_declaration(self): text = (ROOT / "INTENT.md").read_text() front = text.split("---", 2)[1] intent = load_mapping_text(front) decl = load_mapping(LAYER) # Assert the fold, not equality: neither form is re-spelled, and a # real layer divergence still fails (GH-DEC-2026-017 §2). self.assertEqual(_fold(intent["layer"]), _fold(decl["layer"])) self.assertEqual(_fold(intent["role"]), _fold(decl["role"])) def test_sidecar_is_marked_derived_from_intent(self): decl = load_mapping(LAYER) self.assertIs(decl["derived"], True) self.assertEqual(decl["derived_from"], "INTENT.md") def test_declaration_carries_no_standard_version(self): decl = load_mapping(LAYER) front = load_mapping_text((ROOT / "INTENT.md").read_text().split("---", 2)[1]) self.assertNotIn("standard_version", decl) self.assertNotIn("standard_version", front) def test_intent_standard_path_is_unversioned(self): # GH-DEC-2026-020 §1: a version inside the standard: path is a # standard version under A12 r2; companion_version is one too (§2). front = load_mapping_text((ROOT / "INTENT.md").read_text().split("---", 2)[1]) decl = load_mapping(LAYER) self.assertEqual( front["standard"], "net-kingdom/canon/standards/security-layer-model" ) for mapping in (front, decl): self.assertNotIn("companion_version", mapping) module = _checker() self.assertEqual(module.find_versions(front, "INTENT.md"), []) self.assertEqual(module.find_versions(decl, "layer.yaml"), []) def test_version_scan_leaves_schema_version_and_stance_alone(self): module = _checker() self.assertEqual(module.find_versions({"schema_version": "0.1"}, "x"), []) self.assertNotIn("pep-stance", module.SCOPE) # pep-stance.yaml keeps its version fields (GH-DEC-2026-020 §3). self.assertTrue(module.find_versions(load_mapping(STANCE_FILE), "stance")) def test_every_run_prints_version_and_scope(self): module = _checker() for argv in ([], ["--report"]): result = subprocess.run( [sys.executable, str(SCRIPT), *argv], capture_output=True, text=True, cwd=ROOT, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn(module.VALIDATED_AGAINST, result.stdout) self.assertIn(module.SCOPE, result.stdout) def test_vocabulary_is_four_tokens_compared_after_fold(self): module = _checker() self.assertEqual( module.LAYER_VOCABULARY, {"taxonomy", "tooling", "engine", "staff"} ) for token in ("Taxonomy", "TOOLING", "engine", "Staff"): self.assertIn(module.fold(token), module.LAYER_VOCABULARY) self.assertNotIn(module.fold("surface"), module.LAYER_VOCABULARY) def test_checker_rejects_standard_version_and_divergence(self): module = _checker() good = load_mapping(LAYER) cases = { "standard_version": dict(good, standard_version="0.7"), "companion_version": dict(good, companion_version="0.2"), "versioned standard path": dict( good, standard="net-kingdom/canon/standards/security-layer-model_v0.7.md", ), "versioned companion path": dict( good, companion="net-kingdom/SECURITY-COMPANION.md v0.2" ), "nested version key": dict( good, catalog_entry=dict(good["catalog_entry"], standard_version="0.8") ), "not derived": dict(good, derived=False), "divergent layer": dict(good, layer="staff"), } for label, data in cases.items(): with self.subTest(label), patch.object( module, "load_mapping", return_value=data ), patch("sys.stderr"): with self.assertRaises(SystemExit) as caught: module.load_declaration() self.assertEqual(caught.exception.code, 2) with patch.object( module, "load_mapping", return_value=dict(good, layer="ENGINE") ): self.assertEqual(module.load_declaration()["layer"], "ENGINE") def test_own_store_is_declared(self): data = load_mapping(LAYER) self.assertTrue(data["own_store"]) self.assertEqual(data["own_store"][0]["import_root"], "psycopg") def test_checker_passes_on_the_real_tree(self): result = subprocess.run( [sys.executable, str(SCRIPT)], capture_output=True, text=True, cwd=ROOT, ) self.assertEqual(result.returncode, 0, result.stderr) def test_checker_catches_an_undeclared_tooling_client(self): spec = importlib.util.spec_from_file_location( "check_layer_conformance", SCRIPT ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) with tempfile.TemporaryDirectory() as directory: fake_src = Path(directory) / "src" / "user_engine" fake_src.mkdir(parents=True) (fake_src / "secrets.py").write_text( "import hvac\n\n\ndef leak():\n hvac.Client().read('secret')\n" ) original_src = module.SRC try: module.SRC = fake_src hits = module.scan() finally: module.SRC = original_src self.assertTrue(hits, "a direct OpenBao client was not detected") self.assertEqual(hits[0][1], "hvac") class PepStanceTests(unittest.TestCase): def test_published_map_equals_shipped_behaviour(self): published = load_mapping(STANCE_FILE) self.assertEqual(published["stance"], STANCE) self.assertEqual(published["allow_binding"], ALLOW_BINDING) self.assertEqual( timedelta(seconds=published["allow_lifetime_seconds"]), ALLOW_LIFETIME ) self.assertEqual(published["verdict_caching"], VERDICT_CACHING) self.assertTrue( all(value == UNREACHABLE_STANCE for value in published["stance"].values()) ) def test_stance_is_total_over_the_zone_model(self): required = { "z0-experimental", "z1-operational", "z2-protected", "z2-continuity", "z3-critical", "unknown", "not-applicable", } self.assertEqual(required, set(STANCE)) def test_flex_auth_unavailable_applies_fail_closed_without_decision_id(self): from urllib.error import URLError adapter = FlexAuthHTTPAdapter(base_url="http://flex-auth") with patch( "user_engine.adapters.flex_auth.urlopen", side_effect=URLError("down") ): decision = adapter.check(_request()) self.assertEqual(decision.effect, AuthorizationEffect.DENY) self.assertIsNone(decision.decision_id) self.assertEqual(decision.stance_applied, UNREACHABLE_STANCE) self.assertEqual(decision.reason, "authorization service unavailable") def test_flex_auth_allow_carries_published_lifetime(self): import io import json class _Response(io.BytesIO): def __enter__(self): return self def __exit__(self, *_args): self.close() body = _Response(json.dumps({"id": "decision:1", "effect": "allow"}).encode()) adapter = FlexAuthHTTPAdapter(base_url="http://flex-auth") with patch("user_engine.adapters.flex_auth.urlopen", return_value=body): decision = adapter.check(_request()) self.assertEqual(decision.effect, AuthorizationEffect.ALLOW) self.assertEqual(decision.decision_id, "decision:1") self.assertEqual(decision.binding, ALLOW_BINDING) self.assertEqual(decision.lifetime, ALLOW_LIFETIME) self.assertIsNotNone(decision.issued_at) self.assertFalse(decision.expired(now=utc_now())) def test_production_runtime_does_not_import_local_authorization(self): tree = ast.parse(RUNTIME.read_text()) imported = set() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): imported.update(alias.name for alias in node.names) elif isinstance(node, ast.Import): imported.update(alias.name for alias in node.names) self.assertIn("FlexAuthHTTPAdapter", imported) self.assertNotIn("LocalAuthorizationCheckPort", imported) self.assertNotIn("StaticAuthorizationCheckPort", imported) def test_local_authorization_port_cannot_construct_with_production_token(self): with patch.dict( os.environ, {"USER_ENGINE_FLEX_AUTH_TOKEN_FILE": "/var/run/token"} ): with self.assertRaisesRegex(RuntimeError, "cannot be constructed"): LocalAuthorizationCheckPort() def _fold(value): return "".join(chr(ord(c) + 32) if "A" <= c <= "Z" else c for c in str(value)) def _checker(): spec = importlib.util.spec_from_file_location("check_layer_conformance", SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _request(): from user_engine.domain import Actor, AuthorizationRequest, PrincipalType actor = Actor( issuer="https://issuer", subject="subject-1", tenant="tenant-a", principal_type=PrincipalType.HUMAN, audience=("user-engine",), ) return AuthorizationRequest( actor=actor, resource_type="user-engine:user", resource_id="user-1", action="user.update", tenant="tenant-a", correlation_id="corr-1", ) if __name__ == "__main__": unittest.main()