"""Tests for warden plan (WARDEN-WP-0029 T01).""" from __future__ import annotations import json from pathlib import Path import pytest from typer.testing import CliRunner from warden.cli import app from warden.plan import build_plan from warden.posture import load_posture from warden.routing.catalog import load_catalog runner = CliRunner() REPO = Path(__file__).resolve().parents[1] @pytest.fixture(autouse=True) def _catalog_env(monkeypatch): monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(REPO / "registry/routing/catalog.yaml")) monkeypatch.setenv("WARDEN_POSTURE_CATALOG", str(REPO / "registry/policy/security-posture.yaml")) def test_plan_forgejo_deploy_key_autonomous(): plan = build_plan("forgejo deploy key for binky-control") assert plan.verdict == "autonomous" assert plan.organization_posture == "build" assert plan.lane_id == "agent-harness-forgejo-deploy" assert plan.commands assert plan.founder_act is None assert plan.catalog.get("content_hash") def test_plan_forgejo_admin_autonomous(): plan = build_plan("forgejo admin api token") assert plan.verdict == "autonomous" assert plan.lane_id == "forgejo-admin-api-token" assert any("warden access forgejo-admin-api-token" in c for c in plan.commands) def test_plan_new_secret_founder_required(): plan = build_plan("provision a new secret token for a tenant workload") assert plan.verdict == "founder_required" assert plan.founder_act is not None assert plan.founder_act.kind in ("paste_once_provision", "approve", "oidc_login") def test_plan_login_founder_required(): plan = build_plan("oidc login mfa key-cape") assert plan.verdict == "founder_required" assert plan.founder_act is not None assert plan.founder_act.kind == "oidc_login" def test_plan_first_time_openbao_database_admin_uses_platform_admin_login(): plan = build_plan( "first-time OpenBao database engine administration for " "database/config/platform-pg-2 dynamic roles policies and token roles; " "requires attended platform-admin handoff" ) assert plan.verdict == "founder_required" assert plan.lane_id == "openbao-platform-admin-login" assert plan.founder_act is not None assert plan.founder_act.kind == "oidc_login" command = plan.founder_act.details["fetch_command"] assert command == ( "warden access openbao-platform-admin-login --exec -- " ) assert "financials" not in command assert "paste_once" not in plan.founder_act.details["desk_hint"] assert any( item == "warden access openbao-platform-admin-login --exec -- " for item in plan.commands ) assert not any("--fetch" in item for item in plan.commands) assert not any("--out" in item or "--wrap" in item for item in plan.commands) def test_plan_openbao_shamir_recovery_uses_approval_ceremony_not_secret_provision(): plan = build_plan( "coordinate one attended production OpenBao emergency seal/unseal drill " "with a fresh encrypted off-host Raft snapshot receipt, verified " "provider-console access, two-of-three Shamir custodian quorum, named " "driver and abort operator, without exposing credential values" ) assert plan.verdict == "founder_required" assert plan.lane_id == "openbao-shamir-recovery-ceremony" assert plan.founder_act is not None assert plan.founder_act.kind == "approve" assert "openbao-shamir-recovery-ceremony" in plan.founder_act.details["desk_hint"] assert "paste_once" not in plan.founder_act.details["desk_hint"] assert all("warden access" not in item for item in plan.commands) assert any("openbao-shamir-recovery-ceremony" in item for item in plan.commands) def test_plan_unroutable(): # Zero keyword overlap with catalog (avoid tokens like secret/key/token) plan = build_plan("xyzzy-plugh-fnord-qqq-zzzz") assert plan.verdict == "unroutable" assert plan.ccr_stub is not None assert plan.lane_id is None def test_plan_composes_catalog_find(): """Plan must use Catalog.find — exact id match wins.""" cat = load_catalog() plan = build_plan("ssh-cert-host-access", catalog=cat, posture=load_posture()) assert plan.verdict == "autonomous" assert plan.lane_id == "ssh-cert-host-access" assert any("warden sign" in c for c in plan.commands) def test_cli_plan_json(): r = runner.invoke(app, ["plan", "forgejo deploy key for binky-control", "--json"]) assert r.exit_code == 0, r.stdout + r.stderr payload = json.loads(r.stdout) assert payload["verdict"] == "autonomous" assert payload["organization_posture"] == "build" assert payload["lane_id"] == "agent-harness-forgejo-deploy" # --- mutate intent (WARDEN-WP-0038) ------------------------------------------- # # `warden plan` scored a need by keyword overlap with no notion of what the # caller wanted to DO, so "generate a successor secret and CAS-write it to two # custodians" matched the lane that READS that path and inherited its # `autonomous` verdict, answered with --out/--exec/--wrap. Reported independently # by key-cape (2026-09-08) and railiance-platform (2026-09-09), who added: "this # is the same verdict your warden plan should have returned; until that is fixed, # do not let a plan result stand in for this answer." QONTO_ROTATION_NEED = ( "generate a successor client secret for rapp-qonto keycape client and " "CAS-write it to OpenBao platform/workloads/rapp-qonto/keycape-client " "and sso/keycape-rapp-qonto-client" ) def test_reported_qonto_write_no_longer_returns_autonomous(): """The exact need from the report. Regression, in the manner of WP-0033-T06.""" plan = build_plan(QONTO_ROTATION_NEED) assert plan.verdict == "founder_required" assert plan.lane_id == "rapp-qonto-keycape-client" assert plan.founder_act is not None assert plan.founder_act.kind == "approve" assert plan.founder_act.details["write_owner"] == "key-cape" def test_a_write_need_is_never_answered_with_a_read_transport(): """The half that made the wrong verdict actionable rather than merely wrong.""" plan = build_plan(QONTO_ROTATION_NEED) joined = " ".join(plan.commands) for read_transport in ("--out", "--wrap", "--fetch", "bao kv get", "--exec"): assert read_transport not in joined, read_transport def test_mutating_need_names_why_it_escalated(): """WP-0029's property: a verdict carries the reasons that produced it.""" plan = build_plan(QONTO_ROTATION_NEED) assert any("change a credential" in r for r in plan.reasons) assert any("write authority" in r for r in plan.reasons) # The lane records automatable: false; a driver must be told so. assert any("not automatable" in r for r in plan.reasons) def test_ops_warden_still_proceeds_autonomously_on_the_lane_it_owns(): """Signing IS a mutating act. The test is ownership, not the absence of a verb. `delegation.mode: permanent` is the whole distinction — if this ever fails, the guard has started refusing ops-warden's own front door. """ for need in ( "sign an ssh certificate for agt-state-hub-bridge", "issue a short-lived ssh cert for adm", ): plan = build_plan(need) assert plan.verdict == "autonomous", need assert plan.lane_id == "ssh-cert-host-access" def test_reading_the_same_lane_is_unaffected(): plan = build_plan("I need the npm token to publish whynot-design") assert plan.verdict == "autonomous" assert plan.lane_id == "whynot-design-npm-publish" @pytest.mark.parametrize( "verb", ["rotate", "revoke", "regenerate", "reset", "replace"] ) def test_mutating_verbs_escalate_on_a_lane_ops_warden_does_not_own(verb): plan = build_plan(f"{verb} the forgejo admin api token") assert plan.verdict == "founder_required" assert plan.founder_act.details["write_owner"] == "railiance-platform" def test_intent_classifier_matches_tokens_not_substrings(): from warden.plan import _need_intent assert _need_intent("rotate the forgejo token") == "mutate" assert _need_intent("CAS-write to two custodians") == "mutate" # "created" and "updated" are not the verbs; a need describing state is a read. assert _need_intent("read the token created for whynot-design") == "read" assert _need_intent("which subsystem owns the npm token") == "read"