ops-warden/tests/test_case_1_3_integration.py

"""Integration: the real net-kingdom 1.3 candidate against this kernel and catalog.

Reads the actual 1.3 type, its forgejo example and the committed composition
artifact from net-kingdom, plus this branch's real routing catalog. Only two things
are simulated, both in memory: draft lanes flipped to active (promotion needs owner
confirmation) and a 1.3 PARAMETER_PINS entry (PIN-1 waits for the founder's
acknowledgment). Trust is fixture-only, and nothing is dispatched.
"""
from dataclasses import asdict
import hashlib
import json
import os
from pathlib import Path
from unittest import mock

import pytest
import yaml

from warden import case_inputs as inputs
from warden.case_receipt_state import Refused
from warden.case_run import Adapter, Descriptor, Executor, Journal, digest
from warden.case_type import check_follow_through
from warden.routing.catalog import load_catalog

NK = Path(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT",
                         Path(__file__).resolve().parents[2] / "net-kingdom"))
TYPE_DIR = NK / "security-case-type-revisions/1.3/oidc-browser-relying-party"
ARTIFACT = NK / "sso-mfa/k8s/keycape/descriptor-composition-1.3.candidate.json"
CATALOG = Path(__file__).resolve().parents[1] / "registry/routing/catalog.yaml"

pytestmark = pytest.mark.skipif(not (TYPE_DIR / "type.yaml").is_file() or not ARTIFACT.is_file(),
                                reason="net-kingdom 1.3 candidate not available")

DRAFT_LANES = {"keycape-attended-case-change", "platform-secret-delivery",
               "consumer-owner-deployment", "apps-consumer-handover"}


def encode(desc):  # net-kingdom compose-1.2-descriptors.py encode()
    return json.dumps(asdict(desc), sort_keys=True, separators=(",", ":")).encode()


def type_doc(status="acknowledged"):
    doc = yaml.safe_load((TYPE_DIR / "type.yaml").read_bytes())
    doc["spec"]["status"] = status
    return doc


def catalog_bytes(promote):
    raw = yaml.safe_load(CATALOG.read_bytes())
    if promote:
        for entry in raw["entries"]:
            if entry["id"] in DRAFT_LANES and entry.get("status") == "draft":
                entry["status"] = "active"
    return yaml.safe_dump(raw).encode()


def descriptors():
    art = json.loads(ARTIFACT.read_text())
    out = {}
    for row in art["steps"]:
        d = dict(row["descriptor"])
        for key in ("allowed_created", "revocation_scopes"):
            d[key] = tuple(d[key])
        out[row["step"]] = Descriptor(**d)
    return out


def pin_13():
    params = type_doc()["spec"]["parameters"]
    fp = hashlib.sha256(json.dumps(params, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    return mock.patch.dict(inputs.PARAMETER_PINS, {"1.3": fp})


def unused(*_):
    raise AssertionError("admission must not dispatch")


def admit(tmp_path, desc, *, promote=True, pinned=True, status="acknowledged"):
    case = yaml.safe_load((TYPE_DIR / "examples/forgejo.yaml").read_bytes())
    case["spec"].update(status="materialized", acceptance={
        "at": "2026-10-17", "record": "fixture-only acceptance, not an owner record",
        "by": [{"name": "fixture", "role": "owner"}, {"name": "fixture", "role": "founder"}]})
    adapter = Adapter(desc.adapter, desc.adapter_source, unused, unused)
    executor = Executor(Journal(tmp_path / desc.step), {adapter.reference: adapter},
                        lambda purpose, source, evidence: True,
                        browser_inputs=inputs.BrowserInputs(NK))
    with mock.patch.dict(os.environ), (pin_13() if pinned else mock.patch.dict({})):
        os.environ.pop("WARDEN_AGENT_ID", None)
        return executor.admit(yaml.safe_dump(case).encode(), yaml.safe_dump(type_doc(status)).encode(),
                              encode(desc), catalog_bytes(promote), principal="founder",
                              actor=desc.actor, lane=desc.lane,
                              acceptance_evidence="e-acceptance", materialization_evidence="e-mason",
                              descriptor_evidence="e-descriptor", session_evidence="e-session")


def test_artifact_descriptors_are_the_type_map():
    descs = descriptors()
    assert len(descs) == 7
    assert {s: digest(encode(d)) for s, d in descs.items()} == type_doc()["spec"]["execution_descriptors"]


def test_all_seven_admitted_once_promoted_and_pinned(tmp_path):
    for step, desc in descriptors().items():
        admitted, invocation, _ = admit(tmp_path, desc)
        assert admitted == desc
        assert (invocation.step, invocation.actor, invocation.target) == (step, "human", "forgejo")


def test_draft_lane_steps_refused_until_promoted(tmp_path):
    for step, desc in descriptors().items():
        if desc.lane in DRAFT_LANES:
            with pytest.raises(Refused, match="descriptor lane is not active in the catalog"):
                admit(tmp_path, desc, promote=False)
        else:
            admit(tmp_path, desc, promote=False)  # mint-secret's lane is already active


def test_unpinned_1_3_refused_by_browser_validation(tmp_path):
    with pytest.raises(Refused, match="failed exact-version validation"):
        admit(tmp_path, descriptors()["mint-secret"], pinned=False)


def test_proposed_type_refused(tmp_path):
    with pytest.raises(Refused, match="type is not acknowledged"):
        admit(tmp_path, descriptors()["mint-secret"], status="proposed")


def test_type_check_findings_against_the_real_catalog(tmp_path):
    def findings(promote):
        path = tmp_path / f"catalog-{promote}.yaml"
        path.write_bytes(catalog_bytes(promote))
        report = check_follow_through(type_doc("proposed"), load_catalog(path))
        return {(f.code, f.where) for f in report.findings}

    draft = findings(False)
    assert {code for code, _ in draft} == {"LANE-UNREVIEWED"}
    assert len(draft) == 6  # six steps use the four draft lanes; mint-secret's lane is active

    promoted = findings(True)
    # Only the two railiance-apps lanes remain: their session_lane awaits owner confirmation.
    assert {code for code, _ in promoted} == {"LANE-KIND"}
    assert len(promoted) == 2
    assert all("deliver-consumer-secret" in where or "hand-over" in where for _, where in promoted)