From ab17d859cb6fe0b61080f99f8829237762e1ea8e Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 10 Oct 2026 13:17:22 +0200 Subject: [PATCH] plan --case: a case's own built objects are not collisions on re-render Objects listed in the case's committed build bundle (same case id) are shown as already present from this case; others still collide. 2 tests; suite 177. Forgejo plan marked stale after net-kingdom's type change (new render digest). Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 327203@bnt-lap001 Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0 --- src/ops_mason/case_plan.py | 54 +++++++++++++++---- tests/test_case_plan.py | 31 ++++++++++- ...N-WP-0010-security-case-materialization.md | 7 +++ 3 files changed, 81 insertions(+), 11 deletions(-) diff --git a/src/ops_mason/case_plan.py b/src/ops_mason/case_plan.py index 9d77f88..a77a95d 100644 --- a/src/ops_mason/case_plan.py +++ b/src/ops_mason/case_plan.py @@ -130,22 +130,49 @@ def _role_mount(item: dict) -> str: return str(body.get("auth") or "kubernetes") -def review(rendered: dict, structure: Structure) -> list[str]: - """Collisions with existing structure. A collision is a finding for a human, not an auto-merge.""" +DEFAULT_BUILDS = Path("builds") + + +def own_objects(case_id: str, builds_root: Path = DEFAULT_BUILDS) -> set[tuple[str, str]]: + """Objects this case's own committed build bundle created (kind, name). + + A live re-render of a partly built case finds them in OpenBao; they are this + case's work, not someone else's structure, so they are not collisions. + """ + rollback = Path(builds_root) / f"case-{case_id}" / "rollback.yaml" + if not rollback.is_file(): + return set() + data = yaml.safe_load(rollback.read_text()) or {} + if (data.get("security_case") or {}).get("case") != case_id: + return set() + return {(o["kind"], o["name"]) for o in data.get("removes_exactly", [])} + + +def _existing(rendered: dict, structure: Structure) -> list[tuple[dict, str]]: found = [] for item in rendered["materialization"]: name, kind = item["template"], item["kind"] if kind == "openbao_policy" and name in structure.policies: - found.append(f"policy {name} already exists") + found.append((item, f"policy {name} already exists")) elif kind == "openbao_role" and f"{_role_mount(item)}/{name}" in structure.roles: - found.append(f"role auth/{_role_mount(item)}/role/{name} already exists") + found.append((item, f"role auth/{_role_mount(item)}/role/{name} already exists")) elif kind == "kv_path" and name in structure.kv_paths: - found.append(f"KV path {name} already holds an entry") + found.append((item, f"KV path {name} already holds an entry")) elif kind == "catalog_entry" and name in structure.catalog_ids: - found.append(f"catalog entry {name} already exists") + found.append((item, f"catalog entry {name} already exists")) return found +def review(rendered: dict, structure: Structure, own: set[tuple[str, str]] = frozenset()) -> list[str]: + """Collisions with existing structure. A collision is a finding for a human, not an auto-merge.""" + return [msg for item, msg in _existing(rendered, structure) if (item["kind"], item["template"]) not in own] + + +def already_built(rendered: dict, structure: Structure, own: set[tuple[str, str]]) -> list[str]: + return [f"{item['kind']} {item['template']}" for item, _ in _existing(rendered, structure) + if (item["kind"], item["template"]) in own] + + def acceptance_ref(case: dict) -> dict | None: acc = (case.get("spec") or {}).get("acceptance") or {} if not acc.get("record"): @@ -153,7 +180,8 @@ def acceptance_ref(case: dict) -> dict | None: return {"record": acc["record"], "at": acc.get("at"), "by": [a.get("role") for a in acc.get("by", [])]} -def plan_document(case: dict, type_data: dict, rendered: dict, structure: Structure, collisions: list[str]) -> str: +def plan_document(case: dict, type_data: dict, rendered: dict, structure: Structure, collisions: list[str], + built: list[str] = ()) -> str: cspec, tspec = case["spec"], type_data["spec"] acc = acceptance_ref(case) front = { @@ -181,6 +209,9 @@ def plan_document(case: dict, type_data: dict, rendered: dict, structure: Struct "## 2. Existing-structure survey", "", f"Survey source: {structure.source} (metadata only).", ""] lines += [f"- **Collision:** {c}" for c in collisions] or ["No collision with existing policies, roles, KV paths or catalog entries."] + if built: + lines += ["", "Already present from this case's own build bundle (not collisions):", ""] + lines += [f"- {b}" for b in built] lines += ["", "## 3. Objects (from the type's materialization)", "", "| Item | Kind | Name |", "|---|---|---|"] lines += [f"| {i['id']} | {i['kind']} | `{i['template']}` |" for i in rendered["materialization"]] @@ -205,9 +236,12 @@ def plan_document(case: dict, type_data: dict, rendered: dict, structure: Struct def plan_from_case(case_path: Path, *, nk_root: Path = DEFAULT_NK_ROOT, types_root: Path | None = None, structure: Structure | None = None, catalog: Path = DEFAULT_CATALOG, - allow_unimplemented: bool = False) -> tuple[str, list[str], dict]: + allow_unimplemented: bool = False, + builds_root: Path = DEFAULT_BUILDS) -> tuple[str, list[str], dict]: case, type_data, rendered = render(case_path, nk_root, types_root, allow_unimplemented) if structure is None: structure = live_structure(rendered, catalog) - collisions = review(rendered, structure) - return plan_document(case, type_data, rendered, structure, collisions), collisions, rendered + own = own_objects(case["spec"]["id"], builds_root) + collisions = review(rendered, structure, own) + built = already_built(rendered, structure, own) + return plan_document(case, type_data, rendered, structure, collisions, built), collisions, rendered diff --git a/tests/test_case_plan.py b/tests/test_case_plan.py index 253306c..6a5a0f7 100644 --- a/tests/test_case_plan.py +++ b/tests/test_case_plan.py @@ -78,7 +78,8 @@ def test_flags_collision_with_existing_structure(tmp_path, types_root): existing = cp.Structure(policies={"workload-kv-read-keycape-forgejo-eso"}, roles={"kubernetes/external-secrets-keycape-forgejo"}, catalog_ids={"forgejo-keycape-client"}, source="test") - doc, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing) + doc, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing, + builds_root=tmp_path / "no-builds") assert len(collisions) == 3 assert yaml.safe_load(doc.split("---")[1])["status"] == "draft" assert "**Collision:** policy workload-kv-read-keycape-forgejo-eso already exists" in doc @@ -124,3 +125,31 @@ def test_allow_unimplemented_refused_for_non_acknowledged_type(tmp_path, types_r f.write_text(yaml.safe_dump(data, sort_keys=False)) with pytest.raises(cp.CasePlanError): cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=cp.Structure(), allow_unimplemented=True) + + +def test_objects_from_the_cases_own_bundle_are_not_collisions(tmp_path, types_root): + builds = tmp_path / "builds" / "case-forgejo-netkingdom-oidc" + builds.mkdir(parents=True) + (builds / "rollback.yaml").write_text(yaml.safe_dump({ + "security_case": {"case": "forgejo-netkingdom-oidc"}, + "removes_exactly": [{"kind": "openbao_policy", "name": "workload-kv-read-keycape-forgejo-eso"}, + {"kind": "openbao_role", "name": "external-secrets-keycape-forgejo"}]})) + existing = cp.Structure(policies={"workload-kv-read-keycape-forgejo-eso", "workload-kv-read-keycape-forgejo-forgejo"}, + roles={"kubernetes/external-secrets-keycape-forgejo"}, source="live") + doc, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing, + builds_root=tmp_path / "builds") + assert collisions == ["policy workload-kv-read-keycape-forgejo-forgejo already exists"] + assert "Already present from this case's own build bundle" in doc + assert "- openbao_role external-secrets-keycape-forgejo" in doc + + +def test_another_cases_bundle_does_not_excuse_a_collision(tmp_path, types_root): + builds = tmp_path / "builds" / "case-forgejo-netkingdom-oidc" + builds.mkdir(parents=True) + (builds / "rollback.yaml").write_text(yaml.safe_dump({ + "security_case": {"case": "someone-else"}, + "removes_exactly": [{"kind": "openbao_policy", "name": "workload-kv-read-keycape-forgejo-eso"}]})) + existing = cp.Structure(policies={"workload-kv-read-keycape-forgejo-eso"}, source="live") + _, collisions, _ = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=existing, + builds_root=tmp_path / "builds") + assert collisions == ["policy workload-kv-read-keycape-forgejo-eso already exists"] diff --git a/workplans/MASON-WP-0010-security-case-materialization.md b/workplans/MASON-WP-0010-security-case-materialization.md index 32577f7..2eb9a2a 100644 --- a/workplans/MASON-WP-0010-security-case-materialization.md +++ b/workplans/MASON-WP-0010-security-case-materialization.md @@ -202,3 +202,10 @@ net-kingdom changed the type to `KEYCAPE_{client_id:env}_CLIENT_SECRET` (ops-warden `23ba584`), so the render digest will change; and a live re-render of a partly applied case will report its own objects as collisions. The KV value waits on mint-secret (RPF-WP-0054). + +2026-10-10 (T03 follow-up): `plan --case` now recognises objects listed in the +case's own committed bundle (`builds/case-/rollback.yaml`, same case id) as +"already present from this case's own build", not collisions; anything else +existing still collides (2 tests, suite 177 passed). The committed Forgejo plan +is stale: net-kingdom's `{client_id:env}` change moves the render digest from +`fb8b114d…` to `5e97931e…`. Re-render it live under the next survey grant.