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"]