tegwick
f52c94cd4e
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
154 lines
7.7 KiB
Python
154 lines
7.7 KiB
Python
import shutil
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from ops_mason import case_plan as cp
|
|
|
|
NK = cp.DEFAULT_NK_ROOT
|
|
TYPE = NK / "security-case-types" / "oidc-browser-relying-party"
|
|
CASE = TYPE / "examples" / "forgejo.yaml"
|
|
pytestmark = pytest.mark.skipif(not CASE.is_file(), reason="net-kingdom checkout not present")
|
|
|
|
|
|
@pytest.fixture
|
|
def types_root(tmp_path):
|
|
"""The reference type as it will be once implemented; the live one is only acknowledged."""
|
|
root = tmp_path / "types"
|
|
shutil.copytree(TYPE, root / TYPE.name)
|
|
f = root / TYPE.name / "type.yaml"
|
|
data = yaml.safe_load(f.read_text())
|
|
data["spec"]["status"] = "implemented"
|
|
f.write_text(yaml.safe_dump(data, sort_keys=False))
|
|
return root
|
|
|
|
|
|
def _case(tmp_path, **changes):
|
|
case = yaml.safe_load(CASE.read_text())
|
|
for key, value in changes.items():
|
|
if key == "acceptance":
|
|
case["spec"]["status"] = "accepted"
|
|
case["spec"]["acceptance"] = value
|
|
else:
|
|
case["spec"]["parameters"][key] = value
|
|
path = tmp_path / "case.yaml"
|
|
path.write_text(yaml.safe_dump(case, sort_keys=False))
|
|
return path
|
|
|
|
|
|
def test_renders_reference_case_as_derived_plan(tmp_path, types_root):
|
|
doc, collisions, rendered = cp.plan_from_case(_case(tmp_path), types_root=types_root, structure=cp.Structure())
|
|
assert collisions == []
|
|
front = yaml.safe_load(doc.split("---")[1])
|
|
assert front["derived_from_type"] == yaml.safe_load(CASE.read_text())["spec"]["type"]
|
|
assert front["render_digest"] == rendered["digest"] and front["status"] == "draft"
|
|
names = {i["template"] for i in rendered["materialization"]}
|
|
assert {"platform/workloads/keycape/forgejo", "workload-kv-read-keycape-forgejo-eso",
|
|
"workload-kv-read-keycape-forgejo-forgejo", "external-secrets-keycape-forgejo",
|
|
"forgejo-keycape-client"} <= names
|
|
assert "Not accepted yet" in doc and "{client_id}" not in doc
|
|
|
|
|
|
def test_acceptance_replaces_executive_summary(tmp_path, types_root):
|
|
acc = {"at": "2026-10-08", "record": "informed-decision:forgejo-oidc", "by": [{"role": "owner"}, {"role": "founder"}]}
|
|
doc, _, _ = cp.plan_from_case(_case(tmp_path, acceptance=acc), types_root=types_root, structure=cp.Structure(source="live"))
|
|
front = yaml.safe_load(doc.split("---")[1])
|
|
assert front["status"] == "reviewed"
|
|
offline, _, _ = cp.plan_from_case(_case(tmp_path, acceptance=acc), types_root=types_root, structure=cp.Structure())
|
|
assert yaml.safe_load(offline.split("---")[1])["status"] == "draft"
|
|
assert front["case_acceptance"]["record"] == "informed-decision:forgejo-oidc"
|
|
|
|
|
|
def test_refuses_case_whose_type_is_not_implemented(tmp_path):
|
|
with pytest.raises(cp.CasePlanError) as e:
|
|
cp.plan_from_case(_case(tmp_path), structure=cp.Structure())
|
|
assert any("only implemented types" in i for i in e.value.issues)
|
|
|
|
|
|
def test_refuses_parameter_outside_type_validation(tmp_path, types_root):
|
|
with pytest.raises(cp.CasePlanError) as e:
|
|
cp.plan_from_case(_case(tmp_path, redirect_uris="http://forgejo.example/cb"), types_root=types_root,
|
|
structure=cp.Structure())
|
|
assert any("redirect_uris" in i for i in e.value.issues)
|
|
|
|
|
|
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,
|
|
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
|
|
|
|
|
|
def test_live_survey_failure_is_not_an_empty_structure(tmp_path, types_root):
|
|
_, _, rendered = cp.render(_case(tmp_path), types_root=types_root)
|
|
with patch.object(cp.subprocess, "run", return_value=MagicMock(returncode=2, stdout="", stderr="denied")):
|
|
with pytest.raises(cp.CasePlanError):
|
|
cp.live_structure(rendered, catalog=tmp_path / "none.yaml")
|
|
|
|
|
|
def test_live_survey_reads_only_lists_and_metadata(tmp_path, types_root):
|
|
_, _, rendered = cp.render(_case(tmp_path), types_root=types_root)
|
|
calls = []
|
|
|
|
def fake(args, **kw):
|
|
calls.append(args)
|
|
return MagicMock(returncode=0 if args[1] in ("policy", "list") else 2, stdout='["default"]', stderr="")
|
|
with patch.object(cp.subprocess, "run", side_effect=fake):
|
|
s = cp.live_structure(rendered, catalog=tmp_path / "none.yaml")
|
|
assert s.policies == {"default"} and s.source == "live"
|
|
assert all(a[1] in ("policy", "list", "kv") for a in calls)
|
|
assert not any(a[1] == "kv" and a[2] != "metadata" for a in calls)
|
|
|
|
|
|
def test_cli_plan_refuses_unimplemented_type_with_exit_2(tmp_path, capsys):
|
|
from ops_mason.cli import main
|
|
assert main(["plan", "--case", str(_case(tmp_path)), "--no-live"]) == 2
|
|
assert "only implemented types" in capsys.readouterr().err
|
|
|
|
|
|
def test_reference_case_renders_against_acknowledged_type_only_when_asked(tmp_path):
|
|
doc, _, _ = cp.plan_from_case(_case(tmp_path), structure=cp.Structure(), allow_unimplemented=True)
|
|
front = yaml.safe_load(doc.split("---")[1])
|
|
assert front["reference_case"] is True and "**Reference case:**" in doc
|
|
|
|
|
|
def test_allow_unimplemented_refused_for_non_acknowledged_type(tmp_path, types_root):
|
|
f = types_root / TYPE.name / "type.yaml"
|
|
data = yaml.safe_load(f.read_text())
|
|
data["spec"]["status"] = "proposed"
|
|
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"]
|