"""Tests for the routing pointer catalog and `warden route` CLI. No test here requires a live subsystem — routing is a read-only pointer layer. """ import json import re from pathlib import Path import pytest import yaml from typer.testing import CliRunner from warden.cli import app from warden.routing import CatalogError, load_catalog from warden.routing.catalog import find_catalog_path runner = CliRunner() def _repo_catalog() -> Path: return find_catalog_path() def _write_catalog(tmp_path: Path, entries: list[dict]) -> Path: path = tmp_path / "catalog.yaml" path.write_text(yaml.dump({"version": 1, "entries": entries})) return path SSH_ENTRY = { "id": "ssh-cert-host-access", "title": "SSH cert", "need_keywords": ["ssh", "cert", "sign"], "owner_repo": "ops-warden", "subsystem": "ops-warden", "warden_executes": True, "wiki_ref": "wiki/AccessRouting.md#issue-vs-route", "canon_ref": "net-kingdom/docs/x.md", "reviewed": "2026-06-18", "status": "active", "cert_command": "warden sign --pubkey ", "steps": ["confirm inventory", "sign"], } ROUTED_ENTRY = { "id": "openbao-api-key", "title": "API key", "need_keywords": ["api", "key", "openbao"], "owner_repo": "railiance-platform", "subsystem": "OpenBao", "warden_executes": False, "wiki_ref": "wiki/CredentialRouting.md#routing-table", "canon_ref": "net-kingdom/docs/x.md", "reviewed": "2026-06-18", "status": "active", } # --------------------------------------------------------------------------- # Catalog load + validation # --------------------------------------------------------------------------- def test_real_catalog_loads(): catalog = load_catalog(_repo_catalog()) assert len(catalog.entries) >= 6 ssh = catalog.get("ssh-cert-host-access") assert ssh is not None and ssh.warden_executes is True assert ssh.cert_command and "warden sign" in ssh.cert_command def test_real_catalog_has_one_executed_lane(): catalog = load_catalog(_repo_catalog()) executed = [e for e in catalog.entries if e.warden_executes] assert [e.id for e in executed] == ["ssh-cert-host-access"] def test_no_double_source_rule_rejects_routed_steps(tmp_path): bad = dict(ROUTED_ENTRY) bad["steps"] = ["do a thing on OpenBao"] # non-SSH entry must not carry steps path = _write_catalog(tmp_path, [SSH_ENTRY, bad]) with pytest.raises(CatalogError, match="no-double-source"): load_catalog(path) def test_routed_cert_command_rejected(tmp_path): bad = dict(ROUTED_ENTRY) bad["cert_command"] = "warden secret get" path = _write_catalog(tmp_path, [bad]) with pytest.raises(CatalogError, match="cert_command"): load_catalog(path) def test_duplicate_id_rejected(tmp_path): path = _write_catalog(tmp_path, [ROUTED_ENTRY, dict(ROUTED_ENTRY)]) with pytest.raises(CatalogError, match="duplicate"): load_catalog(path) def test_missing_field_rejected(tmp_path): bad = {k: v for k, v in ROUTED_ENTRY.items() if k != "owner_repo"} path = _write_catalog(tmp_path, [bad]) with pytest.raises(CatalogError, match="owner_repo"): load_catalog(path) def test_missing_catalog_file(): with pytest.raises(CatalogError): load_catalog(Path("/nonexistent/catalog.yaml")) # --------------------------------------------------------------------------- # find ranking # --------------------------------------------------------------------------- def test_find_active_excludes_draft(): catalog = load_catalog(_repo_catalog()) ids = [e.id for e in catalog.find("issue core api key")] assert "issue-core-ingestion-api-key" not in ids def test_find_all_includes_draft(): catalog = load_catalog(_repo_catalog()) ids = [e.id for e in catalog.find("issue core api key", include_draft=True)] assert "issue-core-ingestion-api-key" in ids def test_find_ssh_tunnel_top_match(): catalog = load_catalog(_repo_catalog()) matches = catalog.find("ssh tunnel") assert matches and matches[0].id == "ops-bridge-tunnel" # --------------------------------------------------------------------------- # CLI (uses the repo catalog via env override) # --------------------------------------------------------------------------- @pytest.fixture def repo_catalog_env(monkeypatch): monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog())) def test_cli_list_active_only(repo_catalog_env): result = runner.invoke(app, ["route", "list", "--json"]) assert result.exit_code == 0 ids = [e["id"] for e in json.loads(result.stdout)] assert "issue-core-ingestion-api-key" not in ids def test_cli_list_all_includes_draft(repo_catalog_env): result = runner.invoke(app, ["route", "list", "--all", "--json"]) ids = [e["id"] for e in json.loads(result.stdout)] assert "issue-core-ingestion-api-key" in ids def test_cli_show_ssh_json_includes_cert_pattern(repo_catalog_env): result = runner.invoke(app, ["route", "show", "ssh-cert-host-access", "--json"]) assert result.exit_code == 0 data = json.loads(result.stdout) assert data["warden_executes"] is True assert "warden sign" in data["cert_command"] assert data["steps"] def test_cli_show_routed_has_next_action_not_steps(repo_catalog_env): result = runner.invoke(app, ["route", "show", "openbao-api-key", "--json"]) data = json.loads(result.stdout) assert data["warden_executes"] is False assert "steps" not in data assert "next_action" in data def test_cli_show_unknown_exits_one(repo_catalog_env): result = runner.invoke(app, ["route", "show", "does-not-exist"]) assert result.exit_code == 1 def test_cli_find_json(repo_catalog_env): result = runner.invoke(app, ["route", "find", "ssh tunnel", "--json"]) assert result.exit_code == 0 ids = [e["id"] for e in json.loads(result.stdout)] assert "ops-bridge-tunnel" in ids # --------------------------------------------------------------------------- # T5 drift guard — every wiki_ref anchor resolves, every entry has a reviewed date # --------------------------------------------------------------------------- def _github_slug(heading: str) -> str: """Approximate GitHub's heading-anchor slug algorithm.""" text = heading.strip().lower() text = re.sub(r"[^\w\s-]", "", text) # drop punctuation (em-dash, parens, etc.) text = text.replace(" ", "-") return text def _heading_anchors(md_path: Path) -> set[str]: anchors: set[str] = set() for line in md_path.read_text().splitlines(): m = re.match(r"^#{1,6}\s+(.*)$", line) if m: anchors.add(_github_slug(m.group(1))) return anchors def test_every_wiki_ref_anchor_resolves(): catalog = load_catalog(_repo_catalog()) repo_root = _repo_catalog().parents[2] # registry/routing/catalog.yaml -> repo root failures = [] for entry in catalog.entries: rel, _, anchor = entry.wiki_ref.partition("#") md_path = repo_root / rel if not md_path.exists(): failures.append(f"{entry.id}: wiki file missing: {rel}") continue if anchor and anchor not in _heading_anchors(md_path): failures.append(f"{entry.id}: anchor #{anchor} not found in {rel}") assert not failures, "\n".join(failures) def test_every_entry_has_reviewed_date(): catalog = load_catalog(_repo_catalog()) for entry in catalog.entries: assert re.match(r"^\d{4}-\d{2}-\d{2}$", entry.reviewed), ( f"{entry.id}: reviewed must be YYYY-MM-DD, got {entry.reviewed!r}" )