1import json2import os3import subprocess4import sys5import tempfile6import zipfile7import pytest8from tests.conftest import FAKE_SECRET, run_job_inline91011async def test_cross_project_reuse_without_rediscovery(app_client, pool, mock_api):12 """Project A registers; fresh project B (own token, no chat) finds the AKP, downloads the pinned client and executes it."""13 c = app_client; h = c.auth_headers14 await c.post("/api/v1/projects", json={"slug": "proj-a", "name": "A", "policy_mode": "development"}, headers=h)15 res = await run_job_inline(pool, "discover", {"project_id": "proj-a", "openapi_url": mock_api["base"] + "/openapi.json", "name": "Mock Data API", "provider_slug": "mock", "license": "CC0-1.0", "created_by": "a@test"})16 await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True})17 await c.post("/api/v1/projects", json={"slug": "proj-b", "name": "B", "policy_mode": "development"}, headers=h)18 rb = await c.post("/api/v1/auth/tokens", json={"name": "b", "scopes": ["read", "execute"], "project_id": "proj-b"}); hb = {"Authorization": f"Bearer {rb.json()['token']}"}19 n_apis = (await c.get("/api/v1/overview", headers=h)).json()["counts"]["apis"]20 hits = (await c.get("/api/v1/search", params={"q": "mock items"}, headers=hb)).json()["results"]21 assert any(x["type"] == "api" for x in hits)22 pkg = (await c.get("/api/v1/akps/mock.mock-data-api", headers=hb)).json(); assert pkg["lifecycle"] == "tested"23 src = (await c.get(f"/api/v1/akps/{pkg['id']}/files/client.py", headers=hb)).text24 assert f"AKP_VERSION = {pkg['version']}" in src25 # execute the pinned client's operation through the gateway as project B26 r = await c.post("/api/v1/execute", json={"project_id": "proj-b", "package_id": pkg["id"], "operation_id": "ratesByYear", "path_params": {"year": 2019}, "caller": "generated-client"}, headers=hb)27 assert r.status_code == 200 and r.json()["state"] == "succeeded"28 assert (await c.get("/api/v1/overview", headers=h)).json()["counts"]["apis"] == n_apis # nothing rediscovered29 # the generated client itself runs (direct transport) against the mock API with no AKOS involvement30 with tempfile.TemporaryDirectory() as d:31 z = (await c.get(f"/api/v1/akps/{pkg['id']}/bundle.zip", headers=hb)).content32 zipfile.ZipFile(__import__("io").BytesIO(z)).extractall(d)33 root = os.path.join(d, os.listdir(d)[0])34 out = subprocess.run([sys.executable, "-c", f"import sys; sys.path.insert(0, {root!r}); import client as C; r = C.Client(transport='direct').ratesbyyear(2018); print(r.status, r.data['rate'], r.response_hash[:8]); print(C.Client(transport='direct').validate(r))"], capture_output=True, text=True, timeout=60)35 assert out.returncode == 0 and out.stdout.startswith("200 1.15"), out.stdout + out.stderr36 assert out.stdout.strip().endswith("[]")37 t = subprocess.run([sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", os.path.join(root, "tests")], capture_output=True, text=True, timeout=120, cwd=root)38 assert t.returncode == 0, t.stdout[-1500:]394041async def test_credentials_scoped_rotation_revocation_no_leak(app_client, pool, mock_api, caplog):42 c = app_client; h = c.auth_headers43 await c.post("/api/v1/projects", json={"slug": "cred-ok", "name": "ok", "policy_mode": "discovery"}, headers=h)44 await c.post("/api/v1/projects", json={"slug": "cred-no", "name": "no", "policy_mode": "discovery"}, headers=h)45 r = await c.post("/api/v1/credentials", json={"name": "mock-bearer", "scheme": "bearer", "value": FAKE_SECRET, "allowed_projects": ["cred-ok"], "environment": "development"}, headers=h)46 assert r.status_code == 200, r.text; cred = r.json(); assert cred["ref"].startswith("vault://") and FAKE_SECRET not in r.text47 url = mock_api["base"] + "/secure"48 r = await c.post("/api/v1/execute", json={"project_id": "cred-ok", "url": url, "credential_ref_id": cred["id"]}, headers=h); assert r.status_code == 200 and r.json()["status_code"] == 20049 ok_exec = r.json()50 r = await c.post("/api/v1/execute", json={"project_id": "cred-ok", "url": url}, headers=h); assert r.json()["status_code"] == 401 # no credential → provider refuses51 r = await c.post("/api/v1/execute", json={"project_id": "cred-no", "url": url, "credential_ref_id": cred["id"]}, headers=h); assert r.status_code == 403 # unauthorized project52 r = await c.post(f"/api/v1/credentials/{cred['id']}/rotate", json={"value": FAKE_SECRET + "-v2"}, headers=h); assert r.json()["version"] == 253 r = await c.post("/api/v1/execute", json={"project_id": "cred-ok", "url": url, "credential_ref_id": cred["id"], "snapshot": True}, headers=h)54 body = (await c.get(f"/api/v1/executions/{r.json()['id']}/body", headers=h)).json(); assert body["version"] == "v2" and r.json()["credential_version"] == 255 r = await c.post(f"/api/v1/credentials/{cred['id']}/revoke", headers=h); assert r.json()["revoked_at"]56 r = await c.post("/api/v1/execute", json={"project_id": "cred-ok", "url": url, "credential_ref_id": cred["id"]}, headers=h); assert r.status_code == 403 and "revoked" in r.text57 aud = (await c.get(f"/api/v1/credentials/{cred['id']}/audit", headers=h)).json()["events"]58 assert {e["action"] for e in aud} >= {"credential.create", "credential.use", "credential.rotate", "credential.revoke"}59 # --- leakage checks: DB, API responses, logs, export, generated code60 async with pool.acquire() as conn:61 for t in ("executions", "audit_events", "credential_refs", "policy_decisions", "verification_evidence", "jobs", "egress_log"):62 n = await conn.fetchval(f"SELECT count(*) FROM {t} WHERE {t}::text LIKE '%' || $1 || '%'", FAKE_SECRET)63 assert n == 0, f"secret leaked into table {t}"64 for path in (f"/api/v1/executions/{ok_exec['id']}", "/api/v1/credentials", "/api/v1/audit?limit=500", "/api/v1/executions?limit=500", "/api/v1/policy/decisions"):65 assert FAKE_SECRET not in (await c.get(path, headers=h)).text, path66 assert FAKE_SECRET not in caplog.text67 from akos.settings import settings68 assert not subprocess.run(["grep", "-rIl", FAKE_SECRET, str(settings.root / "akos"), str(settings.root / "docs")], capture_output=True, text=True).stdout697071async def test_dataset_export_replays_and_gates(app_client, pool, mock_api):72 c = app_client; h = c.auth_headers73 await c.post("/api/v1/projects", json={"slug": "rep", "name": "rep", "policy_mode": "development"}, headers=h)74 res = await run_job_inline(pool, "discover", {"project_id": "rep", "openapi_url": mock_api["base"] + "/openapi.json", "name": "Mock Data API", "provider_slug": "mock", "license": "CC0-1.0", "created_by": "t"})75 await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True})76 r = await c.post("/api/v1/execute", json={"project_id": "rep", "package_id": res["package_id"], "operation_id": "listItems", "query": {"page": 1}}, headers=h); e1 = r.json()77 r = await c.post("/api/v1/execute", json={"project_id": "rep", "package_id": res["package_id"], "operation_id": "listItems", "query": {"page": 2}}, headers=h); e2 = r.json()78 r = await c.post("/api/v1/datasets", json={"project_id": "rep", "name": "items", "execution_ids": [e1["id"], e2["id"]], "records_path": "$.items", "identifiers": ["id"], "units": {"value": "USD"}, "granularity": "item"}, headers=h)79 assert r.status_code == 200, r.text; ds = r.json(); assert ds["row_count"] == 3 and ds["verification"] == "verified", ds["assessment"]80 q = (await c.get(f"/api/v1/datasets/{ds['id']}/quality", headers=h)).json(); assert q["row_count"] == 3 and q["duplicate_identifier_rows"] == 081 lin = (await c.get(f"/api/v1/datasets/{ds['id']}/lineage", headers=h)).json(); assert {n["type"] for n in lin["nodes"]} >= {"dataset", "execution", "raw", "akp", "api", "provider"}82 out = await run_job_inline(pool, "export", {"dataset_id": ds["id"], "verified": True, "created_by": "t"})83 assert out["verification_label"] == "verified"84 z = (await c.get(f"/api/v1/artifacts/{out['artifact_id']}", headers=h)).content85 with tempfile.TemporaryDirectory() as d:86 zipfile.ZipFile(__import__("io").BytesIO(z)).extractall(d)87 root = os.path.join(d, os.listdir(d)[0])88 m = json.load(open(os.path.join(root, "manifest.json")))89 assert m["expected_output_hash"] == ds["hash"] and all(e["snapshot_included"] for e in m["executions"]) and FAKE_SECRET not in json.dumps(m)90 for f in ("acquire.py", "transform.py", "requirements.lock", "quality-report.json", "lineage.json", "METHODS.md", "README.md", "tests/test_replay.py"):91 assert os.path.exists(os.path.join(root, f)), f92 rr = subprocess.run([sys.executable, "-I", "transform.py", "--offline"], capture_output=True, text=True, cwd=root, timeout=60)93 assert rr.returncode == 0 and "REPRODUCED" in rr.stdout, rr.stdout + rr.stderr94 # gate: dataset built from a provisional (no-AKP) execution cannot be exported as verified95 await c.patch("/api/v1/projects/rep", json={"policy_mode": "discovery"}, headers=h)96 r = await c.post("/api/v1/execute", json={"project_id": "rep", "url": mock_api["base"] + "/items", "snapshot": True}, headers=h); prov = r.json()97 r = await c.post("/api/v1/datasets", json={"project_id": "rep", "name": "provisional", "execution_ids": [prov["id"]], "records_path": "$.items"}, headers=h); ds2 = r.json()98 assert ds2["verification"] == "unverified"99 out2 = await run_job_inline(pool, "export", {"dataset_id": ds2["id"], "verified": True}); assert out2["blocked"] and any("akp_tested" in x for x in out2["reasons"])100 out3 = await run_job_inline(pool, "export", {"dataset_id": ds2["id"], "verified": False}); assert out3["verification_label"] == "unverified"101