scripts/test_api.py

"""Offline tests for the FastAPI backend (`api.py`).

Exercises every endpoint EXCEPT a live triage call: the billed `/api/triage` path is verified by
injecting a synthetic `GatedDecision` into the in-memory cache, so this runs with no API key and no
spend. The override-governance rule (a contradiction needs a written reason) is checked here because
it is enforced server-side.

    python -m scripts.test_api
"""
from __future__ import annotations

import os
import sys
import tempfile

# Point the audit log at a throwaway file BEFORE importing anything that reads CONFIG.
_TMP = tempfile.mkdtemp()
os.environ["AUDIT_LOG_PATH"] = os.path.join(_TMP, "audit.jsonl")
# Force the JSONL backend even if a DATABASE_URL is set in .env — these are offline tests and must
# not touch a real Postgres. (load_dotenv() won't override an env var that's already present.)
os.environ["DATABASE_URL"] = ""

from fastapi.testclient import TestClient  # noqa: E402

import api  # noqa: E402
from src.schemas import GatedDecision, RuleViolation, TriageDecision  # noqa: E402

client = TestClient(api.app)
results: list[tuple[str, str, str]] = []
PASS, FAIL = "PASS", "FAIL"


def check(name, fn):
    try:
        fn()
        results.append((PASS, name, ""))
    except Exception as e:
        results.append((FAIL, name, f"{type(e).__name__}: {e}"))


def t_stats():
    r = client.get("/api/stats")
    assert r.status_code == 200, r.status_code
    body = r.json()
    assert "policy_rules" in body and "precedent_cases" in body
    assert body["provider"] in body["providers"]
    assert isinstance(body["public_demo"], bool)  # drives the UI's provider-toggle visibility


def t_list_campaigns():
    r = client.get("/api/campaigns")
    assert r.status_code == 200
    body = r.json()
    assert len(body) == 18, f"expected 18 campaigns, got {len(body)}"
    assert all("id" in c and "decided" in c for c in body)


def t_get_campaign_and_404():
    r = client.get("/api/campaigns/camp-001")
    assert r.status_code == 200 and r.json()["story"], "camp-001 should load with a story"
    assert client.get("/api/campaigns/camp-999").status_code == 404


def t_policy_sections():
    from src.policy import valid_rule_ids

    r = client.get("/api/policy")
    assert r.status_code == 200
    sections = r.json()
    prefixes = [s["prefix"] for s in sections]
    assert prefixes == ["ELIG", "PROH", "COMP", "CONT", "DEC"], prefixes
    assert all(s["name"] and s["description"] for s in sections), "each section needs a name + blurb"
    n_rules = sum(len(s["rules"]) for s in sections)
    assert n_rules == len(valid_rule_ids()), f"policy drawer should expose every rule, got {n_rules}"


def t_decision_requires_triage_first():
    # No cached triage for this id => 409.
    r = client.post("/api/decisions", json={"campaign_id": "camp-018", "human_decision": "APPROVE", "moderator": "Mod"})
    assert r.status_code == 409, f"expected 409 without triage, got {r.status_code}"


def t_decision_requires_moderator():
    _seed_cache("camp-005")
    # A decision with no moderator name is rejected server-side (a human must own it).
    r = client.post("/api/decisions", json={"campaign_id": "camp-005", "human_decision": "REJECT"})
    assert r.status_code == 400, f"missing moderator should 400, got {r.status_code}"


def _seed_cache(cid="camp-005"):
    td = TriageDecision(
        recommendation="REJECT", confidence="high",
        rule_violations=[RuleViolation(rule_id="PROH-3", severity="hard", evidence="12% guaranteed return")],
        risk_signals=[], rationale="riba investment", questions_for_submitter=[], manipulation_detected=False,
    )
    api._triage_cache[cid] = {"provider": "anthropic", "gated": GatedDecision(decision=td, llm_recommendation="REJECT", overrides=[])}


def t_triage_returns_enriched_rule_text():
    _seed_cache("camp-005")
    # Cached + same provider + not force => returns cache without a live call; rule text enriched.
    r = client.post("/api/triage", json={"campaign_id": "camp-005", "provider": "anthropic"})
    assert r.status_code == 200
    rv = r.json()["decision"]["rule_violations"][0]
    assert rv["rule_id"] == "PROH-3"
    assert rv["rule_text"] and "interest" in rv["rule_text"].lower(), "PROH-3 rule text should be enriched in"


def t_public_demo_caps_spend():
    # With PUBLIC_DEMO on, the billed endpoint must neutralize the two spend faucets: force-rerun
    # (cache bypass) and a non-Anthropic provider. A force+ollama request should therefore be served
    # from the seeded cache — a live call would 502 here (no key), so a 200 proves both were ignored.
    _seed_cache("camp-005")  # provider "anthropic"
    object.__setattr__(api.CONFIG, "public_demo", True)
    try:
        r = client.post("/api/triage", json={"campaign_id": "camp-005", "provider": "ollama", "force": True})
        assert r.status_code == 200, f"lockdown should serve cache, got {r.status_code}: {r.text}"
        body = r.json()
        assert body["provider"] == "anthropic", "lockdown forces the Anthropic provider"
        assert body["decision"]["rule_violations"][0]["rule_id"] == "PROH-3", "served the cached decision"
    finally:
        object.__setattr__(api.CONFIG, "public_demo", False)


def t_decision_agree_and_override_governance():
    _seed_cache("camp-005")
    # Agreeing with the AI (REJECT) needs no reason, but still needs a moderator.
    r = client.post("/api/decisions", json={"campaign_id": "camp-005", "human_decision": "REJECT", "moderator": "Mod"})
    assert r.status_code == 200 and r.json()["is_override"] is False
    assert r.json()["moderator"] == "Mod", "the moderator should be recorded"

    # Overriding (APPROVE) WITHOUT a reason is rejected server-side.
    r = client.post("/api/decisions", json={"campaign_id": "camp-005", "human_decision": "APPROVE", "moderator": "Mod"})
    assert r.status_code == 400, f"unreasoned override should 400, got {r.status_code}"

    # Overriding WITH a reason is accepted and flagged.
    r = client.post("/api/decisions", json={"campaign_id": "camp-005", "human_decision": "APPROVE", "moderator": "Mod", "reason": "manual check cleared it"})
    assert r.status_code == 200 and r.json()["is_override"] is True and r.json()["reason"]


def t_decisions_log_roundtrip():
    r = client.get("/api/decisions")
    assert r.status_code == 200
    assert any(rec["campaign_id"] == "camp-005" for rec in r.json()), "logged decisions should load back"


def main() -> int:
    for name, fn in [
        ("GET /api/stats", t_stats),
        ("GET /api/campaigns (18)", t_list_campaigns),
        ("GET /api/campaigns/{id} + 404", t_get_campaign_and_404),
        ("GET /api/policy (5 sections, all rules)", t_policy_sections),
        ("POST /api/decisions 409 before triage", t_decision_requires_triage_first),
        ("POST /api/decisions requires moderator", t_decision_requires_moderator),
        ("POST /api/triage enriches rule text", t_triage_returns_enriched_rule_text),
        ("POST /api/triage PUBLIC_DEMO caps spend", t_public_demo_caps_spend),
        ("POST /api/decisions override governance", t_decision_agree_and_override_governance),
        ("GET /api/decisions round-trip", t_decisions_log_roundtrip),
    ]:
        check(name, fn)

    print("\n" + "=" * 60)
    for status, name, msg in results:
        print(f"  {status}  {name}" + (f"  — {msg}" if msg else ""))
    n_fail = sum(1 for s, _, _ in results if s == FAIL)
    print("=" * 60)
    print(f"  {len(results) - n_fail} passed, {n_fail} failed")
    return 1 if n_fail else 0


if __name__ == "__main__":
    sys.exit(main())