ops-mason/tests/test_case_build.py

import json
import shutil
from pathlib import Path

import pytest
import yaml

from ops_mason import case_build as cb
from ops_mason import case_plan as cp

NK = cp.DEFAULT_NK_ROOT
TYPE = NK / "security-case-types" / "oidc-browser-relying-party"
CASE = NK / "security-cases" / "forgejo-netkingdom-oidc.yaml"
pytestmark = pytest.mark.skipif(not CASE.is_file(), reason="net-kingdom checkout not present")


@pytest.fixture
def types_root(tmp_path):
    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, **spec):
    case = yaml.safe_load(CASE.read_text())
    case["spec"].update(spec)
    p = tmp_path / "case.yaml"
    p.write_text(yaml.safe_dump(case, sort_keys=False))
    return p


def _plan(tmp_path, case, types_root, source="live"):
    doc, _, _ = cp.plan_from_case(case, types_root=types_root, structure=cp.Structure(source=source))
    p = tmp_path / "plan.md"
    p.write_text(doc)
    return p


def test_builds_bundle_naming_case_type_and_acceptance(tmp_path, types_root):
    case = _case(tmp_path)
    out, log = tmp_path / "bundle", tmp_path / "audit.jsonl"
    r = cb.build_from_case(case, _plan(tmp_path, case, types_root), out, types_root=types_root, audit_log=log)
    assert r["case"] == "forgejo-netkingdom-oidc" and r["type"] == "oidc-browser-relying-party@1.0"
    ccrs = {Path(p).name: yaml.safe_load(Path(p).read_text()) for p in r["ccrs"]}
    eso = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-ESO.yaml"]
    assert eso["status"] == "proposed" and eso["openbao"]["auth"]["role"] == "external-secrets-keycape-forgejo"
    assert eso["openbao"]["auth"]["bound_claims_confirmed"] is False
    consumer = ccrs["CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml"]
    assert consumer["status"] == "in_flight" and consumer["in_flight"]["missing_fields"] == ["openbao.auth"]
    for doc in ccrs.values():
        assert doc["security_case"]["acceptance_record"] and "case acceptance replaces" in doc["requester"]["reason"]
    hcl = (out / "openbao/policies/workload-kv-read-keycape-forgejo-eso.hcl").read_text()
    assert 'path "platform/data/workloads/keycape/forgejo"' in hcl and '["read"]' in hcl and "forgejo-netkingdom-oidc" in hcl
    # token_self_lifecycle: true needs the self stanzas, or the platform applier refuses the file.
    assert 'path "auth/token/lookup-self"' in hcl and 'path "auth/token/revoke-self"' in hcl
    stanzas = [l for l in hcl.splitlines() if l.startswith("path ")]
    assert len(stanzas) == 3
    rollback = yaml.safe_load((out / "rollback.yaml").read_text())
    assert rollback["removes_exactly"][-1] == {"kind": "kv_path", "name": "platform/workloads/keycape/forgejo"}
    assert rollback["removes_exactly"][0]["kind"] == "catalog_entry"
    manifest = json.loads((out / "manifest.json").read_text())
    assert set(manifest["files"]) >= {"rollback.yaml", "catalog-proposal.yaml"}
    record = json.loads(log.read_text().splitlines()[-1])
    assert record["objects"]["event"] == "case-build-prepared" and record["approved_by"].startswith("case acceptance")


def test_refuses_case_without_acceptance(tmp_path, types_root):
    case = _case(tmp_path, status="proposed")
    case.write_text(case.read_text())  # acceptance block remains but status is proposed
    with pytest.raises(cb.CaseBuildRefused):
        cb.build_from_case(case, _plan(tmp_path, case, types_root), tmp_path / "b", types_root=types_root,
                           audit_log=tmp_path / "a.jsonl")


def test_refuses_plan_that_is_not_reviewed(tmp_path, types_root):
    case = _case(tmp_path)
    with pytest.raises(cb.CaseBuildRefused, match="not 'reviewed'"):
        cb.build_from_case(case, _plan(tmp_path, case, types_root, source="none"), tmp_path / "b",
                           types_root=types_root, audit_log=tmp_path / "a.jsonl")
    assert not (tmp_path / "b").exists()


def test_refuses_when_case_changed_since_review(tmp_path, types_root):
    case = _case(tmp_path)
    plan = _plan(tmp_path, case, types_root)
    data = yaml.safe_load(case.read_text())
    data["spec"]["parameters"]["display_name"] = "Forgejo (renamed)"
    case.write_text(yaml.safe_dump(data, sort_keys=False))
    with pytest.raises(cb.CaseBuildRefused, match="digest"):
        cb.build_from_case(case, plan, tmp_path / "b", types_root=types_root, audit_log=tmp_path / "a.jsonl")


def test_refuses_to_overwrite_an_existing_bundle(tmp_path, types_root):
    case = _case(tmp_path)
    plan = _plan(tmp_path, case, types_root)
    (tmp_path / "b").mkdir()
    with pytest.raises(FileExistsError):
        cb.build_from_case(case, plan, tmp_path / "b", types_root=types_root, audit_log=tmp_path / "a.jsonl")


def test_platform_validator_accepts_the_drafts(tmp_path, types_root):
    if not (cb.RAILIANCE_PLATFORM / "scripts" / "credential-change.py").is_file():
        pytest.skip("railiance-platform checkout not present")
    case = _case(tmp_path)
    r = cb.build_from_case(case, _plan(tmp_path, case, types_root), tmp_path / "b", types_root=types_root,
                           audit_log=tmp_path / "a.jsonl")
    assert cb.validate_with_platform([Path(p) for p in r["ccrs"]]) == []