coulomb/ops-mason
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Enforce readiness tiers and reconcile blocked workplans

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e75a-fc5c-7913-9dba-9846210c766d
This commit is contained in:
tegwick 2026-09-28 11:40:57 +02:00
parent 370e1f84c7
commit 36445ae679
14 changed files with 652 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

166
tests/test_kubernetes_plane.py
View file

@ -1,10 +1,13 @@
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,
@ -16,6 +19,9 @@ from ops_mason.kubernetes_plane import (
)
READINESS = "bound_rapps:\n - rapp_id: rapp-test\n readiness_state: verified\n"
REVISION = "a" * 40
ROOT = Path(__file__).resolve().parents[1]
@ -72,6 +78,11 @@ def _fixture(tmp_path: Path, *, approved: bool = True, kind: str = "Namespace")
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",
@ -148,6 +159,14 @@ class FakeCluster:
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"]:
@ -176,7 +195,8 @@ class FakeCluster:
),
)
if command[:4] == ["kubectl", "-n", "whitehat", "get"]:
return CommandResult(0, json.dumps({"items": []}))
assert command[-2:] == ["-o", "name"]
return CommandResult(0, "")
raise AssertionError(f"unexpected command: {command}")
@ -264,3 +284,147 @@ def test_rollback_is_generated_but_never_executed(tmp_path: Path) -> None:
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"]