ops-mason/tests/test_kubernetes_plane.py

import hashlib
import json
from pathlib import Path
from datetime import date

import pytest
import yaml

from ops_mason.readiness import inspect_readiness, resolve_tier

from ops_mason.kubernetes_plane import (
    CommandResult,
    PlaneBundle,
    PlaneRefused,
    _json_path,
    apply,
    preflight,
    rollback_plan,
)


READINESS = "bound_rapps:\n  - rapp_id: rapp-test\n    readiness_state: verified\n"
REVISION = "a" * 40

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


def test_whitehat_bundle_is_exactly_four_allowlisted_objects() -> None:
    bundle = PlaneBundle.load(ROOT / "bundles/whitehat-foundational-plane.yaml")
    assert bundle.digest
    assert [ref.display for ref in bundle.allowed_objects] == [
        "Namespace/whitehat",
        "NetworkPolicy/whitehat/default-deny",
        "NetworkPolicy/whitehat/allow-audit-core-e2",
        "ServiceAccount/whitehat/whitehat-runner",
    ]
    plan = bundle.plan()
    assert plan.status == "built"
    assert plan.approved_by == "Bernd Worsch"
    assert plan.approved_at == "2026-08-22"
    assert {doc["kind"] for doc in bundle.documents} == {
        "Namespace",
        "NetworkPolicy",
        "ServiceAccount",
    }


def _fixture(tmp_path: Path, *, approved: bool = True, kind: str = "Namespace") -> Path:
    (tmp_path / "bundles").mkdir()
    (tmp_path / "manifests").mkdir()
    (tmp_path / "plans").mkdir()
    manifest = tmp_path / "manifests/plane.yaml"
    if kind == "Namespace":
        document = {
            "apiVersion": "v1",
            "kind": "Namespace",
            "metadata": {
                "name": "whitehat",
                "labels": {"pod-security.kubernetes.io/enforce": "restricted"},
            },
        }
        resource = "namespaces"
        namespace = None
    else:
        document = {
            "apiVersion": "v1",
            "kind": kind,
            "metadata": {"name": "forbidden", "namespace": "whitehat"},
        }
        resource = kind.lower() + "s"
        namespace = "whitehat"
    manifest.write_text(yaml.safe_dump(document, sort_keys=False))
    digest = hashlib.sha256(manifest.read_bytes()).hexdigest()
    plan = tmp_path / "plans/plane.md"
    status = "approved" if approved else "reviewed"
    approval = 'approved_by: "Bernd"\napproved_at: "2026-08-22"\n' if approved else ""
    plan.write_text(f"---\nid: plane\nstatus: {status}\n{approval}---\n# Plan\n")
    descriptor = {
        "schema_version": "ops-mason.kubernetes-plane/v1",
        "id": "plane",
        "readiness": {
            "target": {"kind": "rapp", "rapp_id": "rapp-test", "namespace": "whitehat"},
            "source": {"repo": "reef-railiance", "path": "bindings/rapps.yaml",
                       "revision": REVISION, "sha256": hashlib.sha256(READINESS.encode()).hexdigest()},
        },
        "plan": "../plans/plane.md",
        "expected_context": "default",
        "expected_namespace": "whitehat",
        "source_repo": "whitehat-security",
        "source_revision": "abc",
        "implementation_revision": "def",
        "evidence_path": "../evidence/plane.json",
        "forbidden_kinds": ["Pod", "Secret"],
        "manifests": [
            {
                "path": "../manifests/plane.yaml",
                "source_path": "plane.yaml",
                "sha256": digest,
            }
        ],
        "allowed_objects": [
            {
                "api_version": "v1",
                "kind": kind,
                "resource": resource,
                "namespace": namespace,
                "name": document["metadata"]["name"],
            }
        ],
    }
    bundle_path = tmp_path / "bundles/plane.yaml"
    bundle_path.write_text(yaml.safe_dump(descriptor, sort_keys=False))
    return bundle_path


def test_bundle_refuses_forbidden_pod_even_when_allowlisted(tmp_path: Path) -> None:
    with pytest.raises(PlaneRefused, match="forbidden Kubernetes kind"):
        PlaneBundle.load(_fixture(tmp_path, kind="Pod"))


def test_bundle_refuses_manifest_hash_drift(tmp_path: Path) -> None:
    bundle_path = _fixture(tmp_path)
    manifest = tmp_path / "manifests/plane.yaml"
    manifest.write_text(manifest.read_text() + "# drift\n")
    with pytest.raises(PlaneRefused, match="digest mismatch"):
        PlaneBundle.load(bundle_path)


def test_bundle_refuses_secret_bearing_key_even_on_other_kind(tmp_path: Path) -> None:
    bundle_path = _fixture(tmp_path)
    manifest = tmp_path / "manifests/plane.yaml"
    document = yaml.safe_load(manifest.read_text())
    document["stringData"] = {"token": "must-never-enter-mason"}
    manifest.write_text(yaml.safe_dump(document, sort_keys=False))
    descriptor = yaml.safe_load(bundle_path.read_text())
    descriptor["manifests"][0]["sha256"] = hashlib.sha256(manifest.read_bytes()).hexdigest()
    bundle_path.write_text(yaml.safe_dump(descriptor, sort_keys=False))
    with pytest.raises(PlaneRefused, match="secret-bearing"):
        PlaneBundle.load(bundle_path)


def test_json_pointer_supports_label_keys_with_slashes_and_dots() -> None:
    value = {"labels": {"kubernetes.io/metadata.name": "whitehat"}}
    assert _json_path(value, "/labels/kubernetes.io~1metadata.name") == "whitehat"


class FakeCluster:
    def __init__(
        self, *, context: str = "default", drift: bool = False, dirty: bool = False
    ) -> None:
        self.context = context
        self.drift = drift
        self.dirty = dirty
        self.applied = False
        self.calls: list[list[str]] = []

    def __call__(self, args) -> CommandResult:
        command = list(args)
        self.calls.append(command)
        if command[:4] == ["git", "-C", command[2], "status"]:
            return CommandResult(0, " M src/ops_mason/kubernetes_plane.py\n" if self.dirty else "")
        if command[0] == "git":
            if command[3] == "rev-parse":
                return CommandResult(0, "false\n")
            if command[3] == "show":
                return CommandResult(0, READINESS)
            if command[3] == "log":
                return CommandResult(0, REVISION + "\n")
            return CommandResult(0)
        if command == ["kubectl", "config", "current-context"]:
            return CommandResult(0, self.context + "\n")
        if command[:4] == ["kubectl", "auth", "can-i", "create"]:
            return CommandResult(0, "yes\n")
        if "apply" in command:
            if "--dry-run=client" not in command and "--dry-run=server" not in command:
                self.applied = True
            return CommandResult(0, "configured\n")
        if command[:3] == ["kubectl", "get", "namespaces"]:
            if not self.applied:
                return CommandResult(1, "", 'Error from server (NotFound): namespaces "whitehat" not found')
            labels = {"pod-security.kubernetes.io/enforce": "baseline" if self.drift else "restricted"}
            return CommandResult(
                0,
                json.dumps(
                    {
                        "apiVersion": "v1",
                        "kind": "Namespace",
                        "metadata": {
                            "name": "whitehat",
                            "labels": labels,
                            "uid": "uid-1",
                            "resourceVersion": "10",
                        },
                    }
                ),
            )
        if command[:4] == ["kubectl", "-n", "whitehat", "get"]:
            assert command[-2:] == ["-o", "name"]
            return CommandResult(0, "")
        raise AssertionError(f"unexpected command: {command}")


def test_preflight_refuses_wrong_context_before_kubectl_apply(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path))
    cluster = FakeCluster(context="wrong")
    with pytest.raises(PlaneRefused, match="context mismatch"):
        preflight(bundle, cluster)
    assert not any("apply" in call for call in cluster.calls)


def test_apply_refuses_unapproved_plan_without_calling_runner(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path, approved=False))
    calls = []

    def runner(args):
        calls.append(args)
        raise AssertionError("runner must not be called")

    with pytest.raises(PlaneRefused, match="not approved"):
        apply(bundle, confirm_plan_id="plane", expected_digest=bundle.digest, runner=runner)
    assert calls == []


def test_apply_refuses_digest_mismatch_without_calling_runner(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path))
    calls = []

    def runner(args):
        calls.append(args)
        raise AssertionError("runner must not be called")

    with pytest.raises(PlaneRefused, match="digest confirmation mismatch"):
        apply(bundle, confirm_plan_id="plane", expected_digest="wrong", runner=runner)
    assert calls == []


def test_apply_refuses_dirty_repository_before_kubectl(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path))
    cluster = FakeCluster(dirty=True)
    with pytest.raises(PlaneRefused, match="committed and clean"):
        apply(
            bundle,
            confirm_plan_id="plane",
            expected_digest=bundle.digest,
            runner=cluster,
        )
    assert not any(call and call[0] == "kubectl" for call in cluster.calls)


def test_apply_runs_guarded_path_and_writes_metadata_only_evidence(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path))
    cluster = FakeCluster()
    evidence = apply(
        bundle,
        confirm_plan_id="plane",
        expected_digest=bundle.digest,
        runner=cluster,
    )
    assert evidence["verification"]["negative_scope"] == {"pods": 0, "secrets": 0}
    assert evidence["apply"] == {
        "field_manager": "ops-mason",
        "server_validated_manifests": ["manifests/plane.yaml"],
        "persisted_manifests": ["manifests/plane.yaml"],
    }
    assert bundle.evidence_path.exists()
    text = bundle.evidence_path.read_text()
    assert "uid-1" in text
    assert "data" not in evidence["verification"]
    mutating = [call for call in cluster.calls if "apply" in call and "--dry-run=server" not in call and "--dry-run=client" not in call]
    assert len(mutating) == 1
    assert "--field-manager=ops-mason" in mutating[0]


def test_preflight_refuses_unmanaged_live_drift(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path))
    cluster = FakeCluster(drift=True)
    cluster.applied = True
    with pytest.raises(PlaneRefused, match="unmanaged live drift"):
        preflight(bundle, cluster)


def test_rollback_is_generated_but_never_executed(tmp_path: Path) -> None:
    bundle = PlaneBundle.load(_fixture(tmp_path))
    result = rollback_plan(bundle)
    assert result["object_scoped_commands"] == []
    assert result["conditional_namespace_commands"] == ["kubectl delete namespaces whitehat"]



class ReadinessCluster(FakeCluster):
    def __init__(self, content=READINESS, *, old=None, changed=False, shallow=False):
        super().__init__()
        self.content, self.old, self.changed, self.shallow = content, old, changed, shallow

    def __call__(self, args):
        if args[0] == "git" and args[3] != "status":
            self.calls.append(list(args))
            if args[3] == "show":
                return CommandResult(0, self.old if args[4].startswith("b" * 40) else self.content)
            if args[3] == "log":
                return CommandResult(0, REVISION + "\n" + ("b" * 40 + "\n" if self.old else ""))
            if args[3] == "diff":
                return CommandResult(1 if self.changed else 0)
            if args[3] == "rev-parse":
                return CommandResult(0, "true" if self.shallow else "false")
            return CommandResult(0)
        return super().__call__(args)


def readiness_bundle(tmp_path, content=READINESS):
    bundle = PlaneBundle.load(_fixture(tmp_path))
    bundle.readiness["source"]["sha256"] = hashlib.sha256(content.encode()).hexdigest()
    return bundle


@pytest.mark.parametrize("case", ["missing", "platform", "unlisted", "digest", "changed", "unknown", "shallow", "namespace"])
def test_unverifiable_readiness_refuses_before_kubectl(tmp_path, case):
    content = READINESS.replace("verified", "mystery") if case == "unknown" else READINESS
    bundle = readiness_bundle(tmp_path, content)
    cluster = ReadinessCluster(content, changed=case == "changed", shallow=case == "shallow")
    if case == "missing":
        bundle.readiness = None
    elif case == "platform":
        bundle.readiness["target"]["kind"] = "platform"
    elif case == "unlisted":
        bundle.readiness["target"]["rapp_id"] = "absent"
    elif case == "digest":
        bundle.readiness["source"]["sha256"] = "0" * 64
    elif case == "namespace":
        bundle.readiness["target"]["namespace"] = "other"
    with pytest.raises(PlaneRefused, match="production tier"):
        apply(bundle, confirm_plan_id="plane", expected_digest=bundle.digest, runner=cluster)
    assert not any(c[0] == "kubectl" for c in cluster.calls)


def test_production_preflight_reports_but_apply_refuses(tmp_path):
    content = READINESS.replace("verified", "production-approved")
    bundle = readiness_bundle(tmp_path, content)
    cluster = ReadinessCluster(content)
    assert preflight(bundle, cluster)["readiness"]["tier"] == "production"
    with pytest.raises(PlaneRefused, match="production tier"):
        apply(bundle, confirm_plan_id="plane", expected_digest=bundle.digest, runner=cluster)
    assert not cluster.applied


def test_break_glass_requires_reason_and_records_reconciliation(tmp_path):
    content = READINESS.replace("verified", "production-approved")
    bundle = readiness_bundle(tmp_path, content)
    cluster = ReadinessCluster(content)
    with pytest.raises(PlaneRefused, match="non-empty reason"):
        apply(bundle, confirm_plan_id="plane", expected_digest=bundle.digest,
              runner=cluster, activation="BREAK_GLASS", break_glass_reason=" ")
    assert not cluster.calls
    result = apply(bundle, confirm_plan_id="plane", expected_digest=bundle.digest,
                   runner=cluster, activation="BREAK_GLASS", break_glass_reason="Incident test")
    assert result["readiness"]["tier"] == "production"
    assert result["break_glass"]["reason"] == "Incident test"
    assert result["break_glass"]["actor"] and result["break_glass"]["recorded_at"]
    assert "ArgoCD" in result["break_glass"]["follow_up"]


@pytest.mark.parametrize("day,allowed", [(20, True), (21, False), (22, False)])
def test_policy_nexus_transition_expires_at_review_date(tmp_path, day, allowed):
    content = READINESS.replace("rapp-test", "rapp-policy-nexus").replace("verified", "production-approved")
    bundle = readiness_bundle(tmp_path, content)
    bundle.readiness["target"]["rapp_id"] = "rapp-policy-nexus"
    cluster = ReadinessCluster(content)
    kwargs = dict(confirm_plan_id="plane", expected_digest=bundle.digest, runner=cluster, today=date(2026, 12, day))
    if allowed:
        result = apply(bundle, **kwargs)
        assert result["readiness"]["transition"] and result["readiness"]["tier"] == "production"
    else:
        with pytest.raises(PlaneRefused, match="production tier"):
            apply(bundle, **kwargs)
        assert not cluster.applied


@pytest.mark.parametrize("state,old,tier", [("verified", "production-approved", "production"),
    ("deprecated", "production-approved", "production"), ("deprecated", "verified", "non-production")])
def test_lapse_and_deprecation_retain_previous_tier(tmp_path, state, old, tier):
    content = READINESS.replace("verified", state)
    bundle = readiness_bundle(tmp_path, content)
    cluster = ReadinessCluster(content, old=READINESS.replace("verified", old))
    assert inspect_readiness(bundle, cluster, None)["tier"] == tier


def test_whitehat_explicit_placement_and_binding_supersession(tmp_path):
    bundle = readiness_bundle(tmp_path)
    bundle.id = "whitehat-foundational-plane"
    bundle.readiness["target"] = {"kind": "namespace", "namespace": "whitehat"}
    assert inspect_readiness(bundle, ReadinessCluster(), None)["tier"] == "non-production"
    content = READINESS.replace("rapp-test", "rapp-whitehat")
    bundle.readiness["source"]["sha256"] = hashlib.sha256(content.encode()).hexdigest()
    assert inspect_readiness(bundle, ReadinessCluster(content), None)["tier"] == "production"


def test_unknown_activation_never_allows_apply():
    assert not resolve_tier("verified", {}, "anything", date.today())["direct_apply_allowed"]


def test_real_git_history_retains_promotion_even_after_file_reverted(tmp_path):
    """A clean file equal to its pin is not proof it was never production."""
    import subprocess
    from ops_mason.kubernetes_plane import subprocess_runner

    bundle = readiness_bundle(tmp_path)
    repo = tmp_path / "reef"
    repo.mkdir()
    def git(*args):
        return subprocess.run(["git", "-C", str(repo), *args], check=True,
                              capture_output=True, text=True).stdout.strip()
    git("init")
    git("config", "user.name", "Test")
    git("config", "user.email", "test@example.invalid")
    (repo / "bindings").mkdir()
    source = repo / "bindings/rapps.yaml"
    def commit(content, message):
        source.write_text(content)
        git("add", "bindings/rapps.yaml")
        git("commit", "-m", message)
    commit(READINESS, "verified")
    bundle.readiness["source"]["revision"] = git("rev-parse", "HEAD")
    assert inspect_readiness(bundle, subprocess_runner, repo)["tier"] == "non-production"
    commit(READINESS.replace("verified", "production-approved"), "promote")
    commit(READINESS, "evidence lapse")
    result = inspect_readiness(bundle, subprocess_runner, repo)
    assert result["tier"] == "production"
    assert result["reason"] == "production tier retained from binding history"
    source.write_text(READINESS + "# uncommitted\n")
    assert "uncommitted" in inspect_readiness(bundle, subprocess_runner, repo)["reason"]