1import asyncio2import os3import subprocess4import pytest5from tests.conftest import run_job_inline678async def test_opaque_mcp_labelled_and_blocked(app_client, pool):9 c = app_client; h = c.auth_headers10 await c.post("/api/v1/projects", json={"slug": "mcp", "name": "mcp", "policy_mode": "discovery"}, headers=h)11 mc = {"server": "mock-mcp", "tools": [{"name": "lookup", "description": "hidden implementation", "inputSchema": {"type": "object", "properties": {"q": {"type": "string"}}, "required": ["q"]}}],12 "invocations": [{"tool": "lookup", "input": {"q": "a"}, "output": {"q": "a", "answer": 42}}]}13 res = await run_job_inline(pool, "discover", {"project_id": "mcp", "kind": "mcp", "name": "Mock MCP", "mcp_contract": mc, "created_by": "t"})14 api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json()15 assert api["kind"] == "mcp" and api["opaque_internal_implementation"] is True16 assert any(d["kind"] == "opaque_mcp" and d["severity"] == "critical" for d in api["debt"])17 pkg = (await c.get(f"/api/v1/akps/{res['package_id']}", headers=h)).json(); assert pkg["manifest"]["opaque_internal_implementation"] is True18 ex = [e for e in (await c.get("/api/v1/executions", params={"project_id": "mcp"}, headers=h)).json()["executions"]]19 assert ex and all(e["observed"] is False for e in ex)20 r = await c.post("/api/v1/datasets", json={"project_id": "mcp", "name": "mcp-out", "execution_ids": [ex[0]["id"]]}, headers=h); ds = r.json()21 assert ds["verification"] == "unverified" and any(ch["check"].endswith("not_opaque") and not ch["ok"] for ch in ds["assessment"]["checks"])22 out = await run_job_inline(pool, "export", {"dataset_id": ds["id"], "verified": True}); assert out["blocked"]23 # a research-mode project refuses to execute an opaque package? allowed but flagged; verified export stays blocked24 assert (await run_job_inline(pool, "export", {"dataset_id": ds["id"], "verified": False}))["verification_label"] == "unverified"252627async def test_drift_detection_versioning_revalidation(app_client, pool, mock_api):28 c = app_client; h = c.auth_headers29 await c.post("/api/v1/projects", json={"slug": "drift", "name": "drift", "policy_mode": "development"}, headers=h)30 res = await run_job_inline(pool, "discover", {"project_id": "drift", "openapi_url": mock_api["base"] + "/openapi.json", "name": "Mock Data API", "provider_slug": "mock", "license": "CC0-1.0", "created_by": "t"})31 await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True})32 await c.post("/api/v1/execute", json={"project_id": "drift", "package_id": res["package_id"], "operation_id": "listItems", "query": {"page": 1}}, headers=h) # dependent project33 ok = await run_job_inline(pool, "drift_check", {"api_version_id": res["version_id"]}); assert not ok["drifted"]34 mock_api["state"]["drift"] = True35 try:36 bad = await run_job_inline(pool, "drift_check", {"api_version_id": res["version_id"]})37 assert bad["drifted"] and bad["alert_id"]38 proj = (await c.get("/api/v1/projects/drift", headers=h)).json()["id"]39 assert proj in bad["affected_projects"]40 api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json(); assert api["lifecycle"] == "stale"41 assert any(d["kind"] == "stale_contract" and d["status"] == "open" for d in api["debt"])42 pkg = (await c.get(f"/api/v1/akps/{res['package_id']}", headers=h)).json(); assert pkg["lifecycle"] == "stale"43 alerts = (await c.get("/api/v1/drift", headers=h)).json()["alerts"]; assert alerts[0]["status"] == "open"44 finally:45 mock_api["state"]["drift"] = False46 # fix: re-verify (upstream restored) → version revalidated, alert resolved, debt resolved, lifecycle back to tested47 v2 = await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True}); assert v2["promoted_to"] == "tested"48 api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json(); assert api["lifecycle"] == "tested"49 alerts = (await c.get("/api/v1/drift", headers=h)).json()["alerts"]; assert alerts[0]["status"] == "resolved"50 assert not any(d["kind"] == "stale_contract" and d["status"] == "open" for d in api["debt"])515253async def test_worker_crash_recovery_and_audit_intact(pool):54 """A job whose worker died (stale heartbeat) is requeued and completed by another worker; audit chain stays valid."""55 from akos.jobs import queue56 from akos import audit57 async with pool.acquire() as conn:58 job = await queue.enqueue(conn, "maintenance", {}, idempotency_key="recovery-test")59 claimed = await queue.claim(conn, "worker-A", ["maintenance"]); assert claimed and claimed["id"] == job["id"]60 # worker-A crashes: no completion, heartbeat goes stale61 await conn.execute("UPDATE jobs SET heartbeat_at = now() - interval '10 minutes' WHERE id = $1", job["id"])62 again = await queue.claim(conn, "worker-B", ["maintenance"]); assert again and again["id"] == job["id"] and again["attempts"] == 2 and "requeued" in (again["error"] or "")63 await queue.complete(conn, job["id"], {"ok": True})64 row = await conn.fetchrow("SELECT state, attempts FROM jobs WHERE id = $1", job["id"]); assert row["state"] == "done" and row["attempts"] == 265 # max attempts → dead letter, retry endpoint semantics66 j2 = await queue.enqueue(conn, "maintenance", {}, idempotency_key="dead-test", max_attempts=1)67 cl = await queue.claim(conn, "w", ["maintenance"]); assert (await queue.fail(conn, cl, "boom")) == "dead"68 assert (await queue.retry(conn, j2["id"]))["state"] == "queued"69 assert (await audit.verify_chain(conn))["ok"]707172@pytest.mark.node73def test_backup_and_restore_scripts():74 """pg_dump/pg_restore round trip into akos_restore_test (needs Postgres tools on the host)."""75 from akos.settings import settings76 env = {**os.environ, "DATABASE_URL": settings.database_url, "AKOS_BACKUP_DIR": str(settings.backup_dir), "AKOS_DATA_DIR": str(settings.data_dir), "AKOS_BACKUP_REMOTE": "", "PATH": "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin"}77 r = subprocess.run(["/bin/bash", str(settings.root / "scripts" / "backup.sh")], capture_output=True, text=True, env=env, timeout=300)78 assert r.returncode == 0, r.stderr79 dumps = sorted(settings.backup_dir.glob("akos-*.dump")); assert dumps80 r2 = subprocess.run(["/bin/bash", str(settings.root / "scripts" / "restore.sh"), str(dumps[-1]), "akos_restore_test"], capture_output=True, text=True, env=env, timeout=300)81 assert r2.returncode == 0 and "restored" in r2.stdout, r2.stderr82 assert "audit events" in r2.stdout83