107 lines
5.0 KiB
Python
107 lines
5.0 KiB
Python
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.human_gate_store import HumanGateStore
|
|
|
|
|
|
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("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_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("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"]
|