user-engine/tests/test_layer_conformance.py

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(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)
        self.assertEqual(intent["layer"].lower(), decl["layer"])
        self.assertEqual(intent["role"].lower(), decl["role"])

    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 _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()