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. The descriptor
map and metadata come from those committed files. All authority is fixture-only:
we change the proposed type to acknowledged, change the example case to
materialized and replace its acceptance record, promote draft lanes in memory,
and temporarily install the 1.3 parameter pin. Stub adapters replace native
adapters; an unconditional evidence verifier simulates every acceptance and caller
session. WARDEN_AGENT_ID is removed to simulate a human runtime. No adapter is
dispatched. This proves conditional source admission, not authenticated owner
acceptance, native execution or readiness; unresolved Apps sessions stay refused.
"""
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_conditional_admission_keeps_unresolved_apps_sessions_refused(tmp_path):
    for step, desc in descriptors().items():
        if step in ("deliver-consumer-secret", "hand-over"):
            with pytest.raises(Refused, match="does not resolve to an active, reviewed login lane"):
                admit(tmp_path, desc)
        else:
            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)