ops-warden/tests/test_layer_conformance.py

"""Layer-model conformance (security-layer-model_v0.8 §5, §11).

Two things are checked here. §11 makes one of them mechanical: every direct
Tooling client maps to a declared shape. §5.2 asks for the other: the conduit's
supplied-authority property covered by a test.

Deliberately absent: any assertion on a §5.3 review date. A date-triggered
failure breaks the build on a calendar day with no code change, punishing
whoever commits next rather than whoever owns the gap — the same reasoning
recorded in WARDEN-WP-0033-T05 for blocker staleness.
"""
from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]


def _decl() -> dict:
    return yaml.safe_load((ROOT / "layer.yaml").read_text())


def _intent_frontmatter() -> dict:
    lines = (ROOT / "INTENT.md").read_text().splitlines()
    assert lines[0].strip() == "---", "INTENT.md must carry frontmatter — it is the declaration"
    end = next(i for i, ln in enumerate(lines[1:], 1) if ln.strip() == "---")
    return yaml.safe_load("\n".join(lines[1:end]))


def _checker():
    import importlib.util

    spec = importlib.util.spec_from_file_location(
        "check_layer_conformance", ROOT / "scripts" / "check_layer_conformance.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _fold(value: str) -> str:
    return str(value).strip().encode("ascii", "ignore").decode().lower()


class TestDeclaration:
    def test_declares_staff_layer_in_its_own_voice(self):
        d = _decl()
        assert d["repository"] == "ops-warden"
        assert _fold(d["layer"]) == "staff"
        # §11: "only the repository's own file, in its own voice, conforms."
        assert d["declared_by"] == "docs/adr/ADR-0010"

    def test_intent_md_carries_the_governing_declaration(self):
        """GH-DEC-2026-017 §1 / A11: INTENT.md's frontmatter is the declaration."""
        assert _fold(_intent_frontmatter()["layer"]) == "staff"

    def test_sidecar_is_marked_derived_and_names_its_source(self):
        """§11's derived-artifact rule, applied to the layer sidecar."""
        d = _decl()
        assert d["derived"] is True
        assert d["derived_from"] == "INTENT.md"

    def test_the_two_forms_agree_once_case_is_folded(self):
        """A11: the derived form must agree; A9: comparison folds case.

        `Staff` in INTENT.md and `staff` in layer.yaml are the SAME value. This
        test is deliberately a fold rather than an equality: the ruling asked
        nobody to re-spell anything, and an equality assertion here would be this
        repository quietly doing the re-spelling the ruling declined to order.
        The next *real* divergence — a different layer — still fails.
        """
        assert _fold(_decl()["layer"]) == _fold(_intent_frontmatter()["layer"])

    def test_layer_is_in_section_3_closed_vocabulary(self):
        """A9: {Taxonomy, Tooling, Engine, Staff}, closed, case-insensitive."""
        vocabulary = {"taxonomy", "tooling", "engine", "staff"}
        assert _fold(_intent_frontmatter()["layer"]) in vocabulary
        assert _fold(_decl()["layer"]) in vocabulary

    def test_no_declaration_carries_a_standard_version(self):
        """GH-DEC-2026-017 §5 / A12 — and the regression guard on its return.

        The field was removed from the estate's reference form, not just from
        this file. A field that is present will be branched on, so absence is
        asserted rather than trusted.
        """
        assert "standard_version" not in _decl()
        assert "standard_version" not in _intent_frontmatter()

    def test_no_version_anywhere_in_either_declaration(self):
        """A12 r2 / GH-DEC-2026-020 §1-§2: content, not a key name.

        A versioned `standard:` path or a `companion_version` is the same pin as
        `standard_version`, so the guard walks every key and value of both forms.
        """
        checker = _checker()
        assert checker.find_version_pins(_intent_frontmatter()) == []
        assert checker.find_version_pins(_decl()) == []
        assert not str(_intent_frontmatter()["standard"]).endswith(".md")

    def test_checker_catches_a_versioned_standard_path(self):
        checker = _checker()
        pins = checker.find_version_pins(
            {"layer": "Staff", "standard": "net-kingdom/canon/standards/security-layer-model_v0.7.md"}
        )
        assert pins and pins[0].startswith("standard")

    def test_checker_catches_a_companion_version(self):
        checker = _checker()
        assert checker.find_version_pins({"layer": "Staff", "companion_version": "0.2"})
        assert checker.find_version_pins({"nested": {"standard_version": "0.7"}})

    def test_version_token_in_identity_value_is_a_pin(self):
        """GH-DEC-2026-021 §3: any `v?N.N` in a standard:/companion: value is a pin."""
        checker = _checker()
        pins = checker.find_version_pins({"standard": "security-layer-model v0.7"})
        assert pins and pins[0].startswith("standard")
        assert checker.find_version_pins({"companion": "SECURITY-COMPANION 0.2"})
        assert checker.find_version_pins({"nested": {"standard": ["security-layer-model v0.8"]}})

    def test_prose_citation_and_intent_version_are_not_reached(self):
        """GH-DEC-2026-021 §1 (A12 r3): prose provenance and `intent_version` pass."""
        checker = _checker()
        assert checker.find_version_pins(
            {"layer": "Staff", "note": "Outside §5 by the v0.5 scope rule", "intent_version": "0.1.0"}
        ) == []

    def test_schema_version_is_not_reached(self):
        assert _checker().find_version_pins({"schema_version": "0.2", "layer": "Staff"}) == []

    def test_stance_map_is_outside_the_run(self):
        """GH-DEC-2026-020 §3: a stance map keeps its version; the run must not read it."""
        stance = yaml.safe_load((ROOT / "pep-stance.yaml").read_text())
        assert "standard_version" in stance, "pep-stance.yaml keeps its clause-scoped version"
        assert "pep-stance" not in _checker().SCOPE

    def test_every_run_states_version_and_scope(self):
        """GH-DEC-2026-020 §4: the version belongs to the run, printed every time."""
        checker = _checker()
        out = subprocess.run(
            [sys.executable, str(ROOT / "scripts" / "check_layer_conformance.py")],
            capture_output=True,
            text=True,
        ).stdout
        assert f"validated against: {checker.VALIDATED_AGAINST}" in out
        assert f"scope: {checker.SCOPE}" in out
        pass_line = next(ln for ln in out.splitlines() if ln.startswith("PASS"))
        assert checker.VALIDATED_AGAINST in pass_line

    def test_every_tooling_contact_maps_to_a_declared_shape(self):
        """§11 mechanical check — the guard against a new undeclared client."""
        result = subprocess.run(
            [sys.executable, str(ROOT / "scripts" / "check_layer_conformance.py")],
            capture_output=True,
            text=True,
        )
        assert result.returncode == 0, (
            f"undeclared Tooling contact — a finding under §11, not a tracked gap:\n"
            f"{result.stdout}{result.stderr}"
        )

    def test_declared_gaps_carry_all_four_fields(self):
        """§5.3 is machine-readable or it is prose wearing a schema."""
        for c in _decl()["tooling_contacts"]:
            if c["shape"] == "5.3":
                for field in ("capability", "intended_owner", "blocked_on", "review"):
                    assert c.get(field), f"{c['id']} missing {field}"

    def test_gaps_are_not_counted_as_conformance(self):
        """§11: a declared gap is tracked non-conformance. Keep that visible."""
        text = (ROOT / "layer.yaml").read_text()
        assert "TRACKED NON-CONFORMANCE" in text.upper()


class TestConduitSuppliesNoAuthority:
    """§5.2: 'MUST NOT present its own credential, MUST NOT widen what the
    caller could already do.' The standard says this SHOULD be covered by a
    test; this is that test."""

    def test_conduit_supplies_no_authority_of_its_own(self, monkeypatch):
        from warden import proxy

        monkeypatch.setenv("VAULT_TOKEN", "caller-own-token")
        monkeypatch.setenv("HOME", "/home/nobody")
        before = dict(os.environ)

        env = proxy._caller_env()

        # The child environment IS the caller's environment — nothing added,
        # nothing removed, no ops-warden credential injected.
        assert env == before, (
            "conduit altered the caller's environment; §5.2 requires it to "
            "supply no authority of its own"
        )
        assert env["VAULT_TOKEN"] == "caller-own-token"

    def test_conduit_declares_supplied_authority_none(self):
        conduits = [c for c in _decl()["tooling_contacts"] if c["shape"] == "5.2"]
        assert conduits, "no §5.2 conduit declared — proxy.py is one"
        for c in conduits:
            assert c["supplied_authority"] == "none"

    def test_proxy_holds_no_credential_constant(self):
        """A conduit that presents its own token is not a conduit (§5.2)."""
        src = (ROOT / "src" / "warden" / "proxy.py").read_text()
        # It may name token ENV VARS to detect caller auth; it must not carry a
        # token value or mint one.
        for forbidden in ("X-Vault-Token", "auth/approle/login", "token create"):
            assert forbidden not in src, (
                f"proxy.py references {forbidden!r} — that is presenting or "
                f"minting authority, not conducting the caller's"
            )


class TestPepStanceMap:
    """§6.4: every PEP-shaped consumer MUST publish its unreachable-engine
    stance map, total and per zone, 'published rather than held in code'.
    ADR-0009 is named as the reference shape, so it should actually hold."""

    def _stance(self) -> dict:
        return yaml.safe_load((ROOT / "pep-stance.yaml").read_text())

    def test_published_map_equals_shipped_behaviour(self):
        """The whole point. A published map that may drift from the code is
        worse than none, because it invites reliance it cannot support."""
        from warden.config import PolicyConfig

        assert self._stance()["stance"] == PolicyConfig().failure_modes

    def test_stance_is_total_over_the_zone_model(self):
        """§6.4 obligation 3: total, no implicit default."""
        stance = self._stance()["stance"]
        required = {
            "z0-experimental", "z1-operational", "z2-protected",
            "z2-continuity", "z3-critical", "unknown", "not-applicable",
        }
        assert required <= set(stance), f"stance not total; missing {required - set(stance)}"
        assert set(stance.values()) <= {"fail_open", "fail_closed"}

    def test_critical_zone_fails_closed(self):
        """ADR-0009's one non-negotiable row."""
        assert self._stance()["stance"]["z3-critical"] == "fail_closed"

    def test_verdict_is_never_cached(self):
        """§6.4 obligation 2: caching an input claim is permitted; caching the
        answer is a second decision point deciding early (§6.1)."""
        assert self._stance()["verdict_caching"] == "none"

    def test_revocation_visibility_deadline_equals_enforced_ttl_policy(self):
        """§9.7.2: a published replay window must not drift from issuance."""
        from warden.models import ActorType, MAX_TTL_HOURS

        published = self._stance()["revocation_visibility"]
        expected = {actor.value: MAX_TTL_HOURS[actor] for actor in ActorType}
        assert published["deadline_hours"] == expected
        assert published["mechanism"] == "ttl_expiry"
        assert published["revocation_channel"] == "none"

    def test_attributive_emission_cadence_deferral_carries_measurement(self):
        cadence = self._stance()["emission_cadence"]
        assert cadence["classification"] == "attributive"
        assert cadence["status"] == "deferred"
        assert cadence["observed_window"]["signature_records"] == 3
        assert cadence["observed_window"]["active_signature_days"] == 2
        assert cadence["reason"]


# --- classification coverage (v0.8 §6.4 obligation 3) -------------------------

def test_published_coverage_equals_measured_coverage():
    """The published figure must equal what the repo actually measures.

    ops-warden asked gate-house for §13.1's Coverage column and its figures are
    that column's first entries, so their accuracy is ours to hold. The register
    explicitly does not compute anyone's coverage, and a stale number beside a
    marked cell is worse than a blank -- a blank at least reads as "not reported".

    This is the same property that makes the stance map worth publishing (the map
    equals PolicyConfig.failure_modes by test), applied one level up.
    """
    import importlib.util

    repo = Path(__file__).resolve().parents[1]
    spec = importlib.util.spec_from_file_location(
        "report_coverage", repo / "scripts" / "report_coverage.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    published = yaml.safe_load((repo / "pep-stance.yaml").read_text())[
        "classification_coverage"
    ]
    measured = module.measure()

    for population in ("signing_targets", "routing_lanes"):
        assert published[population] == measured[population], population


def test_the_unknown_cell_is_marked_as_a_declared_gap():
    """A non-conformant cell must say so where it is declared, not only in a review.

    §11's marking obligation, which ops-warden argued for in the v0.6 round and
    then acquired a marked cell under. If the cell is ever flipped to fail_closed
    this test fails, which is the correct time to remove the marking.
    """
    repo = Path(__file__).resolve().parents[1]
    text = (repo / "pep-stance.yaml").read_text()
    stance = yaml.safe_load(text)["stance"]

    if stance["unknown"] == "fail_open":
        assert "DECLARED GAP" in text
        assert "WARDEN-WP-0040" in text
    else:
        assert stance["unknown"] == "fail_closed"


def test_accepted_v08_run_provenance_and_catalog_scope():
    checker = _checker()
    assert "security-layer-model_v0.8.md" in checker.VALIDATED_AGAINST
    assert "sha256:3d8298560069ef471fba28d2b1245fb2309328f231dad1a667cdf0ce5a5a7c8b" in checker.VALIDATED_AGAINST
    assert "single repository ops-warden (catalogued in §4)" in checker.SCOPE


def test_reference_semantics_propagate_identity_through_nested_sequences():
    checker = _checker()
    for data in (
        {"nested": {"standard": ["Security Layer Model v0.8"]}},
        {"companion": [["Security Companion 0.3"]]},
        {"standard": {"alternate": ["Security Layer Model v0.7"]}},
    ):
        # A dictionary begins a fresh identity key; only sequence descendants
        # inherit the immediate standard:/companion: identity in the reference.
        if isinstance(data.get("standard"), dict):
            assert checker.find_version_pins(data) == []
        else:
            assert checker.find_version_pins(data), data
    assert checker.find_version_pins({"rationale": "Assented at v0.7; migrated at v0.8."}) == []
    assert checker.find_version_pins({"intent_version": "0.2.0", "schema_version": "0.1"}) == []


def test_ascii_folding_never_discards_non_ascii_letters():
    checker = _checker()
    assert checker._fold("Staff") == "staff"
    assert checker._fold("Engine") == "engine"
    # Deleting non-ASCII bytes would turn this fifth token into Staff.
    assert checker._fold("Stαaff") not in checker.LAYER_VOCABULARY