1import pytest2from tests.conftest import run_job_inline345async def _set_mode(c, slug, mode):6 r = await c.patch(f"/api/v1/projects/{slug}", json={"policy_mode": mode}, headers=c.auth_headers); assert r.status_code == 200, r.text789async def test_policy_modes_and_provenance(app_client, pool, mock_api):10 c = app_client; h = c.auth_headers11 await c.post("/api/v1/projects", json={"slug": "pol", "name": "pol", "policy_mode": "discovery"}, headers=h)12 # discovery: unregistered URL allowed, recorded provisional + debt, intent/outcome audit, hash + snapshot13 r = await c.post("/api/v1/execute", json={"project_id": "pol", "url": mock_api["base"] + "/items?page=1", "purpose": "explore"}, headers=h); assert r.status_code == 200, r.text14 ex = r.json(); assert ex["state"] == "succeeded" and ex["provisional"] and ex["response_hash"] and ex["policy_decision"]["decision"] == "allow"15 d = (await c.get(f"/api/v1/executions/{ex['id']}", headers=h)).json()16 assert [a["action"] for a in d["audit"]] == ["execution.intent", "execution.outcome"]17 debt = (await c.get("/api/v1/debt", headers=h)).json()["debt"]18 assert any(x["kind"] == "undocumented_endpoint" and x["subject_id"] == ex["id"] for x in debt)19 # development: unregistered denied (blocked execution persisted with decision)20 await _set_mode(c, "pol", "development")21 r = await c.post("/api/v1/execute", json={"project_id": "pol", "url": mock_api["base"] + "/items"}, headers=h); assert r.status_code == 40322 blocked = r.json()["error"]["execution"]; assert blocked["state"] == "blocked" and blocked["policy_decision"]["decision"] == "deny"23 # register the mock API via discovery → development allowed through endpoint24 res = await run_job_inline(pool, "discover", {"project_id": "pol", "openapi_url": mock_api["base"] + "/openapi.json", "name": "Mock Data API", "provider_slug": "mock", "license": "CC0-1.0", "created_by": "test"})25 api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json(); ep = next(e for e in api["endpoints"] if e["operation_id"] == "listItems")26 r = await c.post("/api/v1/execute", json={"project_id": "pol", "api_id": res["api_id"], "endpoint_id": ep["id"], "query": {"page": 2}, "idempotency_key": "idem-1"}, headers=h); assert r.status_code == 200, r.text27 first = r.json(); assert first["state"] == "succeeded" and not first["provisional"]28 r2 = await c.post("/api/v1/execute", json={"project_id": "pol", "api_id": res["api_id"], "endpoint_id": ep["id"], "query": {"page": 2}, "idempotency_key": "idem-1"}, headers=h)29 assert r2.json()["id"] == first["id"] # idempotent replay, no second call30 # research: requires tested AKP → deny before verification31 await _set_mode(c, "pol", "research")32 r = await c.post("/api/v1/execute", json={"project_id": "pol", "package_id": res["package_id"], "operation_id": "listItems"}, headers=h); assert r.status_code == 403 and "needs tested" in r.json()["error"]["message"]33 vres = await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True, "created_by": "test"})34 assert vres["promoted_to"] == "tested", vres35 r = await c.post("/api/v1/execute", json={"project_id": "pol", "package_id": res["package_id"], "operation_id": "listItems", "query": {"page": 1}}, headers=h); assert r.status_code == 200, r.text36 # production: needs approved akp pinned on the project37 await _set_mode(c, "pol", "production")38 r = await c.post("/api/v1/execute", json={"project_id": "pol", "package_id": res["package_id"], "operation_id": "listItems"}, headers=h); assert r.status_code == 40339 r = await c.patch("/api/v1/projects/pol", json={"approved_akps": [res["package_id"]]}, headers=h); assert r.status_code == 20040 r = await c.post("/api/v1/execute", json={"project_id": "pol", "package_id": res["package_id"], "operation_id": "listItems"}, headers=h); assert r.status_code == 403 # lifecycle tested, needs registered41 r = await c.post(f"/api/v1/akps/{res['package_id']}/lifecycle", json={"to": "registered"}, headers=h); assert r.status_code == 200, r.text42 r = await c.post("/api/v1/execute", json={"project_id": "pol", "package_id": res["package_id"], "operation_id": "listItems", "query": {"page": 1}}, headers=h); assert r.status_code == 200, r.text43 assert r.json()["policy_decision"]["reasons"] == ["production: ok"]44 # response size limit + egress grant cleanup45 await _set_mode(c, "pol", "discovery")46 from akos.settings import settings47 big = await c.post("/api/v1/execute", json={"project_id": "pol", "url": mock_api["base"] + "/big"}, headers=h)48 assert big.status_code == 200 and big.json()["state"] == "succeeded"49 async with pool.acquire() as conn:50 assert await conn.fetchval("SELECT count(*) FROM egress_grants WHERE expires_at > now() AND execution_id IS NOT NULL") == 051 assert (await conn.fetchval("SELECT count(*) FROM executions WHERE state='intent'")) == 0525354async def test_lifecycle_transition_refused_without_evidence(app_client, pool, mock_api):55 c = app_client; h = c.auth_headers56 r = await c.post("/api/v1/providers", json={"name": "Manual Prov"}, headers=h); pid = r.json()["id"]57 r = await c.post("/api/v1/apis", json={"provider_id": pid, "name": "Manual API", "base_url": mock_api["base"]}, headers=h); aid = r.json()["id"]58 r = await c.post(f"/api/v1/apis/{aid}/lifecycle", json={"to": "tested"}, headers=h); assert r.status_code == 40959 r = await c.post(f"/api/v1/apis/{aid}/lifecycle", json={"to": "inspected"}, headers=h); assert r.status_code == 20060