ops-mason/tests/test_case_plan.py

import shutil
from pathlib import Path
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"] == "oidc-browser-relying-party@1.0"
    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())
    front = yaml.safe_load(doc.split("---")[1])
    assert front["status"] == "reviewed"
    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)
    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