ops-warden/tests/test_flex_auth_registry.py

"""Tests for scripts/build_flex_auth_registry.py."""
import json
import subprocess
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "build_flex_auth_registry.py"
INVENTORY = ROOT / "examples" / "inventory.seed.yaml"


def test_build_registry_from_inventory_seed(tmp_path):
    out = tmp_path / "registry.json"
    subprocess.run(
        [sys.executable, str(SCRIPT), str(INVENTORY), "-o", str(out)],
        check=True,
        cwd=ROOT,
    )
    registry = json.loads(out.read_text())
    actors = yaml.safe_load(INVENTORY.read_text())["actors"]

    assert len(registry["subjects"]) == len(actors)
    assert len(registry["resource_manifests"][0]["resources"]) == len(actors)

    bridge = next(
        r
        for r in registry["resource_manifests"][0]["resources"]
        if r["id"] == "ssh-cert:actor/agt-state-hub-bridge"
    )
    assert bridge["attributes"]["actor_type"] == "agt"
    assert bridge["attributes"]["max_ttl_hours"] == 24
    assert "agt-task-bridge" in bridge["attributes"]["allowed_principals"]
    assert "trust_zone" not in bridge
    assert bridge["attributes"]["security_zone"] == "unknown"
    assert bridge["attributes"]["security_zone_admission"] == "unknown"
    assert bridge["attributes"]["workload_id"] == "ops-bridge-tunnel"

    human = next(
        r
        for r in registry["resource_manifests"][0]["resources"]
        if r["id"] == "ssh-cert:actor/adm-example"
    )
    assert human["attributes"]["security_zone_admission"] == "not-applicable"


def test_compiler_joins_explicit_workload_reference_to_resolved_zone(tmp_path):
    zones = tmp_path / "zones.json"
    zones.write_text(json.dumps({
        "records": [{
            "workload_id": "ops-bridge-tunnel",
            "declared_zone": "z2-continuity",
            "admission": "satisfied",
            "admission_reason": "admission_floor_met",
            "effective_zone": "z2-continuity",
            "membership_revision": "sha256:zone-revision",
        }]
    }))
    out = tmp_path / "registry.json"
    subprocess.run(
        [
            sys.executable,
            str(SCRIPT),
            str(INVENTORY),
            "--zone-resolutions",
            str(zones),
            "-o",
            str(out),
        ],
        check=True,
        cwd=ROOT,
    )
    registry = json.loads(out.read_text())
    bridge = next(
        r
        for r in registry["resource_manifests"][0]["resources"]
        if r["id"] == "ssh-cert:actor/agt-state-hub-bridge"
    )
    attrs = bridge["attributes"]
    assert attrs["security_zone"] == "z2-continuity"
    assert attrs["security_zone_admission"] == "satisfied"
    assert attrs["security_zone_revision"] == "sha256:zone-revision"


# --- manifest ceiling coverage (FLEX-DEC-2026-012) ----------------------------
#
# flex-auth fixed enrichment so registry facts beat caller-supplied ones, then
# asked every consumer to confirm the ceiling keys are actually declared: the
# fix wins only where the registry HAS a value. These assert the two ways a
# ceiling gets handed back to the caller.

import importlib.util  # noqa: E402

import pytest  # noqa: E402

_REPO = ROOT
_spec = importlib.util.spec_from_file_location(
    "check_flex_auth_manifest_coverage",
    _REPO / "scripts" / "check_flex_auth_manifest_coverage.py",
)
_coverage = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_coverage)
REQUIRED_CEILING_KEYS = _coverage.REQUIRED_CEILING_KEYS
audit = _coverage.audit

_build_spec = importlib.util.spec_from_file_location("build_flex_auth_registry", SCRIPT)
_builder = importlib.util.module_from_spec(_build_spec)
_build_spec.loader.exec_module(_builder)
build_registry = _builder.build_registry


def _repo_inventory_and_registry():
    inventory = yaml.safe_load((_REPO / "examples" / "inventory.seed.yaml").read_text())
    registry = json.loads(
        (_REPO / "registry" / "flex-auth" / "production_registry_snapshot.json").read_text()
    )
    return inventory, registry


def test_shipped_snapshot_declares_every_ceiling_key():
    """The confirmation flex-auth asked for, as a test rather than a reading."""
    report = audit(*_repo_inventory_and_registry())
    assert report["unregistered_actors"] == []
    assert report["resources_missing_keys"] == []
    assert report["covered"] is True


def test_actor_with_no_manifest_resource_is_a_coverage_gap():
    """`warden sign` names ssh-cert:actor/<name> whether or not the snapshot was rebuilt."""
    inventory, registry = _repo_inventory_and_registry()
    inventory["actors"]["adm-added-after-the-last-build"] = {
        "type": "adm", "principals": ["adm-full"], "ttl_hours": 48,
    }
    report = audit(inventory, registry)
    assert report["unregistered_actors"] == ["adm-added-after-the-last-build"]
    assert report["covered"] is False


@pytest.mark.parametrize("key", REQUIRED_CEILING_KEYS)
def test_dropping_any_single_ceiling_key_is_a_coverage_gap(key):
    inventory, registry = _repo_inventory_and_registry()
    resource = registry["resource_manifests"][0]["resources"][0]
    resource["attributes"].pop(key)
    report = audit(inventory, registry)
    assert report["resources_missing_keys"] == [
        {"resource": resource["id"], "missing": [key]}
    ]
    assert report["covered"] is False


@pytest.mark.parametrize("key", REQUIRED_CEILING_KEYS)
def test_a_null_ceiling_is_not_a_declaration(key):
    """flex-auth overlays a registry value only where one exists."""
    inventory, registry = _repo_inventory_and_registry()
    resource = registry["resource_manifests"][0]["resources"][0]
    resource["attributes"][key] = None
    report = audit(inventory, registry)
    assert report["covered"] is False


def test_builder_output_is_covered_by_construction(tmp_path):
    """Every resource the builder emits carries the full ceiling set."""
    inventory = yaml.safe_load((_REPO / "examples" / "inventory.seed.yaml").read_text())
    report = audit(inventory, build_registry(inventory))
    assert report["covered"] is True