import sqlite3 import httpx import pytest from src import main from src.human_gate_store import HumanGateStore from src.security_event_store import SecurityEventStore, SecurityEventStoreError CANDIDATE = { "source": "release-bot", "project": "stackchain/dashboard", "candidate_hash": "abc123", "title": "Release candidate", "priority": 7, "artifacts": [{"name": "manifest", "url": "https://forge.example/manifest"}], "links": [{"label": "pull", "url": "https://forge.example/pulls/1"}], "checks": [{"name": "unit", "state": "success", "required": True}], "score": {"value": 98, "provenance": "eval/v1"}, "provenance": {"run": "9"}, } CHECKLIST = {"exact_hash": True, "artifacts_reviewed": True, "provenance_reviewed": True} @pytest.fixture def gate_api(monkeypatch, tmp_path): ticks = iter(range(100, 120)) store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: next(ticks)) monkeypatch.setattr(main, "_human_gate_store", lambda: store, raising=False) async def identity(): return {"id": 1, "login": "timmy"} monkeypatch.setattr(main, "current_user", identity) return store @pytest.mark.anyio async def test_intake_list_and_detail_are_account_bound_and_no_store(gate_api): transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: created = await client.post("/api/v1/human-gates/intake", json=CANDIDATE, headers={"Idempotency-Key": "run-9"}) repeated = await client.post("/api/v1/human-gates/intake", json=CANDIDATE, headers={"Idempotency-Key": "run-9"}) listing = await client.get("/api/v1/human-gates") detail = await client.get(f"/api/v1/human-gates/{created.json()['id']}") assert created.status_code == 201 assert repeated.status_code == 200 assert repeated.json()["id"] == created.json()["id"] assert listing.json()["pending_count"] == 1 assert detail.json()["candidate_hash"] == "abc123" assert detail.json()["history"][0]["action"] == "intake" assert all(response.headers["cache-control"] == "no-store" for response in (created, repeated, listing, detail)) @pytest.mark.anyio async def test_decision_requires_revision_and_returns_durable_receipt(gate_api): gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-9") transport = httpx.ASGITransport(app=main.app) payload = {"expected_revision": gate["revision"], "decision": "release", "reason": "", "override_reason": "", "checklist": CHECKLIST} async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: decided = await client.post(f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-9"}) repeated = await client.post(f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-9"}) receipt = await client.get(f"/api/v1/human-gate-receipts/{decided.json()['receipt_id']}") stale = await client.post(f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-10"}) assert decided.status_code == 201 assert repeated.status_code == 200 assert receipt.json() == decided.json() assert stale.status_code == 409 assert all(response.headers["cache-control"] == "no-store" for response in (decided, repeated, receipt, stale)) @pytest.mark.anyio async def test_decision_requires_fresh_authorization_bound_to_the_exact_gate(monkeypatch, gate_api): gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-authorized") authorization_calls = [] async def require_step_up(request, grant, *, action, target): authorization_calls.append({"grant": grant, "action": action, "target": target}) if grant != "one-time-grant": raise main.HTTPException( status_code=428, detail={ "detail": "Fresh authorization required", "code": "step_up_required", "action": action, "target": target, }, ) monkeypatch.setattr(main, "_require_step_up", require_step_up) payload = { "expected_revision": gate["revision"], "decision": "hold", "reason": "Needs another browser run", "override_reason": "", "checklist": CHECKLIST, } transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: rejected = await client.post( f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-authorized"}, ) pending = await client.get(f"/api/v1/human-gates/{gate['id']}") accepted = await client.post( f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={ "Idempotency-Key": "decision-authorized", "X-Step-Up-Grant": "one-time-grant", }, ) assert rejected.status_code == 428 assert pending.json()["state"] == "pending" assert accepted.status_code == 201 assert authorization_calls == [ {"grant": None, "action": "decide_human_gate", "target": gate["id"]}, {"grant": "one-time-grant", "action": "decide_human_gate", "target": gate["id"]}, ] @pytest.mark.anyio async def test_successful_decision_records_one_completed_privacy_safe_security_event( monkeypatch, gate_api, tmp_path ): gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-audited") journal_path = tmp_path / "security.sqlite3" journal = SecurityEventStore(journal_path, clock=lambda: 200) monkeypatch.setattr(main, "_security_event_store", lambda: journal) private_reason = "Private launch context must not enter the security journal" payload = { "expected_revision": gate["revision"], "decision": "hold", "reason": private_reason, "override_reason": "", "checklist": CHECKLIST, } transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: decided = await client.post( f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-audited"}, ) events = journal.list().events assert decided.status_code == 201 assert [ {"kind": event.kind, "method": event.method, "target": event.target, "status": event.status} for event in events ] == [ { "kind": "human_gate_decision", "method": "hold", "target": gate["id"], "status": "completed", } ] assert private_reason.encode() not in journal_path.read_bytes() @pytest.mark.anyio async def test_decision_fails_closed_when_security_event_cannot_be_reserved( monkeypatch, gate_api ): gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-no-journal") class UnavailableJournal: def reserve(self, *args, **kwargs): raise SecurityEventStoreError("private journal path") monkeypatch.setattr(main, "_security_event_store", UnavailableJournal) payload = { "expected_revision": gate["revision"], "decision": "release", "reason": "", "override_reason": "", "checklist": CHECKLIST, } transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: rejected = await client.post( f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-no-journal"}, ) pending = await client.get(f"/api/v1/human-gates/{gate['id']}") assert rejected.status_code == 503 assert rejected.json() == {"detail": "Security activity is temporarily unavailable"} assert pending.json()["state"] == "pending" @pytest.mark.anyio async def test_rejected_decision_discards_its_pending_security_event( monkeypatch, gate_api, tmp_path ): gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-conflict") journal = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 200) monkeypatch.setattr(main, "_security_event_store", lambda: journal) payload = { "expected_revision": gate["revision"] + 1, "decision": "hold", "reason": "Outdated review", "override_reason": "", "checklist": CHECKLIST, } transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: rejected = await client.post( f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-conflict"}, ) assert rejected.status_code == 409 assert journal.list().events == [] @pytest.mark.anyio async def test_recycled_login_cannot_read_another_principal_gates(monkeypatch, tmp_path): store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: 100) monkeypatch.setattr(main, "_human_gate_store", lambda: store, raising=False) principal = {"id": 1, "login": "timmy"} async def identity(): return principal.copy() monkeypatch.setattr(main, "current_user", identity) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: created = await client.post( "/api/v1/human-gates/intake", json=CANDIDATE, headers={"Idempotency-Key": "principal-1"}, ) principal["id"] = 2 listing = await client.get("/api/v1/human-gates") assert created.status_code == 201 assert listing.json()["pending_count"] == 0 @pytest.mark.anyio async def test_gate_store_failure_is_sanitized_no_store(monkeypatch): def unavailable(): raise sqlite3.OperationalError("sensitive database path") monkeypatch.setattr(main, "_human_gate_store", unavailable) async def identity(): return {"id": 1, "login": "timmy"} monkeypatch.setattr(main, "current_user", identity) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/api/v1/human-gates") assert response.status_code == 503 assert response.headers["cache-control"] == "no-store" assert response.json() == {"detail": "Human Gates are temporarily unavailable"} @pytest.mark.anyio async def test_gate_mutations_require_idempotency_key_and_validate_override(gate_api): failing = {**CANDIDATE, "candidate_hash": "fail123", "checks": [{"name": "browser", "state": "failure", "required": True}]} gate = gate_api.intake("1:timmy", failing, idempotency_key="run-fail") transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: missing_key = await client.post("/api/v1/human-gates/intake", json=CANDIDATE) no_override = await client.post( f"/api/v1/human-gates/{gate['id']}/decision", json={"expected_revision": 1, "decision": "release", "reason": "", "override_reason": "", "checklist": CHECKLIST}, headers={"Idempotency-Key": "decision-fail"}, ) assert missing_key.status_code == 422 assert no_override.status_code == 422 assert "override reason" in no_override.json()["detail"]