ops-warden/tests/test_plan.py
tegwick 0fae0904ce fix: contain attended OpenBao login output
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0290b-3241-74c3-b868-6049545af836
2026-08-23 01:31:05 +02:00

121 lines
4.7 KiB
Python

"""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 -- <reviewed-command>"
)
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 -- <reviewed-command>"
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"