ops-warden/tests/test_case_sheet.py

"""Ordered sheet over admitted steps. Fixture adapters only; nothing native."""

from dataclasses import asdict, replace
import json

import pytest
import yaml

from warden.case_receipt_state import Outcome, Refused
from warden.case_run import Adapter, Descriptor, Executor, Journal, OwnerResult, digest
from warden.case_sheet import rollback_sheet, run_sheet
from tests.test_case_run import lane_entry


SOURCE = "sha256:" + "1" * 64


def _result(inv, outcome, created=()):
    return OwnerResult(
        outcome,
        "owner-receipt",
        created,
        True,
        session_revocation_evidence="session-revoked",
        grant_revocation_evidence="grant-revoked",
        attempt_id=inv.attempt_id,
        operation=inv.operation,
    )


def _spec(step, mode, outcome=Outcome.SUCCESS, created=True, lane=None):
    return {
        "step": step,
        "mode": mode,
        "outcome": outcome,
        "created": created,
        "lane": lane,
    }


def admit_sheet(tmp_path, specs, *, case_id="case-sheet", executor=None):
    """Admit fixture steps on one executor. No step is dispatched."""
    modes = {
        "warden": (
            "issuer-lane",
            True,
            False,
            "mint",
            ("parent-session", "broker-grant"),
            None,
        ),
        "attended": ("login-lane", False, True, "change", ("parent-session",), "login"),
        "owner": ("issuer-lane", False, False, "change", ("parent-session",), None),
    }
    descriptors = {}
    follow = []
    entries = {}
    adapters = {}
    calls = []
    undos = []
    for spec in specs:
        lane, executes, attended, effect, scopes, kind = modes[spec["mode"]]
        lane = spec["lane"] or lane
        if spec["mode"] == "attended" and spec["lane"]:
            kind = "login"
        object_id = "obj-" + spec["step"]
        desc = Descriptor(
            "warden.case.execution.v1",
            spec["step"],
            "owner",
            "owner/" + spec["step"] + "@v1",
            SOURCE,
            "human",
            lane,
            "client_id",
            effect,
            (object_id,),
            False,
            scopes,
        )
        encoded = json.dumps(asdict(desc)).encode()
        descriptors[spec["step"]] = (desc, encoded)
        follow.append(
            {
                "id": spec["step"],
                "owner": "owner",
                "lane": lane,
                "warden_executes": executes,
                "attended": attended,
            }
        )
        entry = lane_entry(lane)
        if kind:
            entry["lane"] = kind
        entries[lane] = entry

        def invoke(inv, spec=spec, object_id=object_id):
            calls.append(spec["step"])
            created = (
                (object_id,)
                if spec["created"] and spec["outcome"] is Outcome.SUCCESS
                else ()
            )
            if spec["outcome"] == "raise":
                raise RuntimeError("synthetic lost reply")
            return _result(inv, spec["outcome"], created)

        def undo(inv, objects, spec=spec):
            undos.append(spec["step"])
            if spec.get("undo") == "raise":
                raise RuntimeError("synthetic lost undo")
            return _result(inv, Outcome.ROLLED_BACK, ())

        adapters[desc.adapter] = Adapter(
            desc.adapter, SOURCE, invoke, lambda *_: True, undo
        )

    typ = yaml.safe_dump(
        {
            "spec": {
                "id": "test",
                "version": "1",
                "status": "acknowledged",
                "parameters": [
                    {
                        "name": "client_id",
                        "sensitivity": "operational",
                        "case_settable": True,
                    }
                ],
                "execution_descriptors": {
                    step: digest(encoded) for step, (_, encoded) in descriptors.items()
                },
                "follow_through": follow,
            }
        }
    ).encode()
    case = yaml.safe_dump(
        {
            "spec": {
                "id": case_id,
                "type": "test@1",
                "status": "materialized",
                "acceptance": "acceptance",
                "parameters": {"client_id": "fixture-client"},
            }
        }
    ).encode()
    catalog = yaml.safe_dump({"entries": list(entries.values())}).encode()
    if executor is None:
        executor = Executor(Journal(tmp_path / "private"), adapters, lambda *_: True)
    else:
        executor.adapters.update(adapters)
    admitted = []
    for spec in specs:
        desc, encoded = descriptors[spec["step"]]
        admitted.append(
            executor.admit(
                case,
                typ,
                encoded,
                catalog,
                principal="operator",
                actor="human",
                lane=desc.lane,
                acceptance_evidence="acceptance",
                materialization_evidence="materialization",
                descriptor_evidence="descriptor",
                session_evidence="session",
            )
        )
    return executor, tuple(admitted), calls, undos


def _session(events):
    def session(lane, dispatch):
        events.append(("session", lane))
        dispatch()
        events.append(("session-end", lane))

    return session


def test_stops_before_later_steps_when_a_step_does_not_close(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, calls, _ = admit_sheet(
        tmp_path,
        [
            _spec("mint-secret", "warden", Outcome.FAILED_BEFORE, created=False),
            _spec("bind-env", "warden"),
        ],
    )
    seen = []
    receipt = run_sheet(
        executor,
        steps,
        verifications=(("consumer-enforces-groups", lambda: seen.append(1)),),
    )
    assert calls == ["mint-secret"]
    assert seen == []
    assert receipt.closed is False
    assert receipt.steps == (("mint-secret", "failed-before-change"),)
    assert receipt.verifications == ()


def test_attended_steps_on_one_lane_share_one_session(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, calls, _ = admit_sheet(
        tmp_path,
        [
            _spec("ensure-groups", "attended"),
            _spec("bind-env", "attended"),
            _spec("mint-secret", "warden"),
        ],
    )
    events = []

    def session(lane, dispatch):
        events.append(("session", lane, tuple(calls)))
        dispatch()
        events.append(("session-end", lane, tuple(calls)))

    receipt = run_sheet(executor, steps, session=session)
    assert events == [
        ("session", "login-lane", ()),
        ("session-end", "login-lane", ("ensure-groups", "bind-env")),
    ]
    assert calls == ["ensure-groups", "bind-env", "mint-secret"]
    assert receipt.closed is True
    assert receipt.verifications == ()


def test_different_lanes_and_a_break_are_separate_sessions(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, calls, _ = admit_sheet(
        tmp_path,
        [
            _spec("ensure-groups", "attended"),
            _spec("mint-secret", "warden"),
            _spec("register-client", "attended", lane="other-login"),
        ],
    )
    events = []
    run_sheet(executor, steps, session=_session(events))
    assert [lane for kind, lane in events if kind == "session"] == [
        "login-lane",
        "other-login",
    ]
    assert calls == ["ensure-groups", "mint-secret", "register-client"]


def test_attended_dispatch_outside_the_session_is_refused(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, calls, _ = admit_sheet(
        tmp_path, [_spec("ensure-groups", "attended")]
    )
    with pytest.raises(Refused, match="contained session"):
        run_sheet(executor, steps)
    assert calls == []

    def silent(lane, dispatch):
        return None

    with pytest.raises(Refused, match="left the contained session"):
        run_sheet(executor, steps, session=silent)
    assert calls == []


def test_verification_runs_only_after_every_step_closes(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, _, _ = admit_sheet(
        tmp_path, [_spec("mint-secret", "warden"), _spec("bind-env", "owner")]
    )
    checks = []

    def check(outcome):
        def run():
            checks.append(outcome)
            return outcome

        return run

    receipt = run_sheet(
        executor,
        steps,
        verifications=(
            ("custody-allowed", check(Outcome.SUCCESS)),
            ("consumer-enforces-groups", check(Outcome.FAILED_BEFORE)),
            ("not-reached", check(Outcome.SUCCESS)),
        ),
    )
    assert [item.value for item in checks] == [
        "verified-success",
        "failed-before-change",
    ]
    assert receipt.closed is False
    assert receipt.verifications[-1][0] == "consumer-enforces-groups"


def test_unknown_verification_result_is_uncertain_and_not_stored_as_text(
    tmp_path, monkeypatch
):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, _, _ = admit_sheet(tmp_path, [_spec("mint-secret", "warden")])

    def leak():
        raise RuntimeError("synthetic-secret-value")

    receipt = run_sheet(executor, steps, verifications=(("readback", leak),))
    assert receipt.closed is False
    assert receipt.verifications == (("readback", "uncertain"),)
    assert "synthetic-secret-value" not in repr(receipt)

    def wrong():
        return "verified-success"

    again, again_steps, _, _ = admit_sheet(
        tmp_path / "again", [_spec("mint-secret", "warden")]
    )
    receipt = run_sheet(again, again_steps, verifications=(("readback", wrong),))
    assert receipt.verifications == (("readback", "uncertain"),)


def test_refuses_unadmitted_duplicate_and_foreign_pins(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, calls, _ = admit_sheet(tmp_path, [_spec("mint-secret", "warden")])
    moved = (steps[0][0], replace(steps[0][1], target="other-target"), steps[0][2])
    with pytest.raises(Refused, match="not admitted"):
        run_sheet(executor, (moved,))
    with pytest.raises(Refused, match="duplicate"):
        run_sheet(executor, (steps[0], steps[0]))
    _, others, _, _ = admit_sheet(
        tmp_path,
        [_spec("bind-env", "owner")],
        case_id="other-case",
        executor=executor,
    )
    with pytest.raises(Refused, match="do not share"):
        run_sheet(executor, (steps[0], others[0]))
    assert calls == []


def test_rollback_skips_no_change_and_stops_on_uncertain_undo(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, _, undos = admit_sheet(
        tmp_path,
        [
            _spec("mint-secret", "warden"),
            _spec("bind-env", "owner", Outcome.NO_CHANGE, created=False),
            dict(_spec("register-client", "owner"), undo="raise"),
        ],
    )
    assert run_sheet(executor, steps).closed is True
    done = rollback_sheet(executor, steps)
    assert undos == ["register-client"]
    assert done[-1].last.outcome is Outcome.ROLLBACK_UNCERTAIN
    assert "mint-secret" not in undos


def test_uncertain_step_blocks_sheet_rollback(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, _, undos = admit_sheet(
        tmp_path,
        [
            _spec("mint-secret", "warden"),
            _spec("bind-env", "owner", outcome="raise", created=False),
        ],
    )
    receipt = run_sheet(executor, steps)
    assert receipt.steps[-1][1] == "uncertain"
    with pytest.raises(Refused, match="owner reconciliation"):
        rollback_sheet(executor, steps)
    assert undos == []


def test_attended_rollback_stays_inside_the_session(tmp_path, monkeypatch):
    monkeypatch.delenv("WARDEN_AGENT_ID", raising=False)
    executor, steps, _, undos = admit_sheet(
        tmp_path,
        [_spec("ensure-groups", "attended"), _spec("bind-env", "attended")],
    )
    run_sheet(executor, steps, session=_session([]))
    with pytest.raises(Refused, match="contained session"):
        rollback_sheet(executor, steps)
    assert undos == []
    events = []
    done = rollback_sheet(executor, steps, session=_session(events))
    assert [lane for kind, lane in events if kind == "session"] == ["login-lane"]
    assert undos == ["bind-env", "ensure-groups"]
    assert all(state.last.outcome is Outcome.ROLLED_BACK for state in done)