import json
import os
import subprocess
import sys
import tempfile
import zipfile
import pytest
from tests.conftest import FAKE_SECRET, run_job_inline


async def test_cross_project_reuse_without_rediscovery(app_client, pool, mock_api):
    """Project A registers; fresh project B (own token, no chat) finds the AKP, downloads the pinned client and executes it."""
    c = app_client; h = c.auth_headers
    await c.post("/api/v1/projects", json={"slug": "proj-a", "name": "A", "policy_mode": "development"}, headers=h)
    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"})
    await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True})
    await c.post("/api/v1/projects", json={"slug": "proj-b", "name": "B", "policy_mode": "development"}, headers=h)
    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']}"}
    n_apis = (await c.get("/api/v1/overview", headers=h)).json()["counts"]["apis"]
    hits = (await c.get("/api/v1/search", params={"q": "mock items"}, headers=hb)).json()["results"]
    assert any(x["type"] == "api" for x in hits)
    pkg = (await c.get("/api/v1/akps/mock.mock-data-api", headers=hb)).json(); assert pkg["lifecycle"] == "tested"
    src = (await c.get(f"/api/v1/akps/{pkg['id']}/files/client.py", headers=hb)).text
    assert f"AKP_VERSION = {pkg['version']}" in src
    # execute the pinned client's operation through the gateway as project B
    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)
    assert r.status_code == 200 and r.json()["state"] == "succeeded"
    assert (await c.get("/api/v1/overview", headers=h)).json()["counts"]["apis"] == n_apis   # nothing rediscovered
    # the generated client itself runs (direct transport) against the mock API with no AKOS involvement
    with tempfile.TemporaryDirectory() as d:
        z = (await c.get(f"/api/v1/akps/{pkg['id']}/bundle.zip", headers=hb)).content
        zipfile.ZipFile(__import__("io").BytesIO(z)).extractall(d)
        root = os.path.join(d, os.listdir(d)[0])
        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)
        assert out.returncode == 0 and out.stdout.startswith("200 1.15"), out.stdout + out.stderr
        assert out.stdout.strip().endswith("[]")
        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)
        assert t.returncode == 0, t.stdout[-1500:]


async def test_credentials_scoped_rotation_revocation_no_leak(app_client, pool, mock_api, caplog):
    c = app_client; h = c.auth_headers
    await c.post("/api/v1/projects", json={"slug": "cred-ok", "name": "ok", "policy_mode": "discovery"}, headers=h)
    await c.post("/api/v1/projects", json={"slug": "cred-no", "name": "no", "policy_mode": "discovery"}, headers=h)
    r = await c.post("/api/v1/credentials", json={"name": "mock-bearer", "scheme": "bearer", "value": FAKE_SECRET, "allowed_projects": ["cred-ok"], "environment": "development"}, headers=h)
    assert r.status_code == 200, r.text; cred = r.json(); assert cred["ref"].startswith("vault://") and FAKE_SECRET not in r.text
    url = mock_api["base"] + "/secure"
    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"] == 200
    ok_exec = r.json()
    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 refuses
    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 project
    r = await c.post(f"/api/v1/credentials/{cred['id']}/rotate", json={"value": FAKE_SECRET + "-v2"}, headers=h); assert r.json()["version"] == 2
    r = await c.post("/api/v1/execute", json={"project_id": "cred-ok", "url": url, "credential_ref_id": cred["id"], "snapshot": True}, headers=h)
    body = (await c.get(f"/api/v1/executions/{r.json()['id']}/body", headers=h)).json(); assert body["version"] == "v2" and r.json()["credential_version"] == 2
    r = await c.post(f"/api/v1/credentials/{cred['id']}/revoke", headers=h); assert r.json()["revoked_at"]
    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.text
    aud = (await c.get(f"/api/v1/credentials/{cred['id']}/audit", headers=h)).json()["events"]
    assert {e["action"] for e in aud} >= {"credential.create", "credential.use", "credential.rotate", "credential.revoke"}
    # --- leakage checks: DB, API responses, logs, export, generated code
    async with pool.acquire() as conn:
        for t in ("executions", "audit_events", "credential_refs", "policy_decisions", "verification_evidence", "jobs", "egress_log"):
            n = await conn.fetchval(f"SELECT count(*) FROM {t} WHERE {t}::text LIKE '%' || $1 || '%'", FAKE_SECRET)
            assert n == 0, f"secret leaked into table {t}"
    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"):
        assert FAKE_SECRET not in (await c.get(path, headers=h)).text, path
    assert FAKE_SECRET not in caplog.text
    from akos.settings import settings
    assert not subprocess.run(["grep", "-rIl", FAKE_SECRET, str(settings.root / "akos"), str(settings.root / "docs")], capture_output=True, text=True).stdout


async def test_dataset_export_replays_and_gates(app_client, pool, mock_api):
    c = app_client; h = c.auth_headers
    await c.post("/api/v1/projects", json={"slug": "rep", "name": "rep", "policy_mode": "development"}, headers=h)
    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"})
    await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True})
    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()
    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()
    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)
    assert r.status_code == 200, r.text; ds = r.json(); assert ds["row_count"] == 3 and ds["verification"] == "verified", ds["assessment"]
    q = (await c.get(f"/api/v1/datasets/{ds['id']}/quality", headers=h)).json(); assert q["row_count"] == 3 and q["duplicate_identifier_rows"] == 0
    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"}
    out = await run_job_inline(pool, "export", {"dataset_id": ds["id"], "verified": True, "created_by": "t"})
    assert out["verification_label"] == "verified"
    z = (await c.get(f"/api/v1/artifacts/{out['artifact_id']}", headers=h)).content
    with tempfile.TemporaryDirectory() as d:
        zipfile.ZipFile(__import__("io").BytesIO(z)).extractall(d)
        root = os.path.join(d, os.listdir(d)[0])
        m = json.load(open(os.path.join(root, "manifest.json")))
        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)
        for f in ("acquire.py", "transform.py", "requirements.lock", "quality-report.json", "lineage.json", "METHODS.md", "README.md", "tests/test_replay.py"):
            assert os.path.exists(os.path.join(root, f)), f
        rr = subprocess.run([sys.executable, "-I", "transform.py", "--offline"], capture_output=True, text=True, cwd=root, timeout=60)
        assert rr.returncode == 0 and "REPRODUCED" in rr.stdout, rr.stdout + rr.stderr
    # gate: dataset built from a provisional (no-AKP) execution cannot be exported as verified
    await c.patch("/api/v1/projects/rep", json={"policy_mode": "discovery"}, headers=h)
    r = await c.post("/api/v1/execute", json={"project_id": "rep", "url": mock_api["base"] + "/items", "snapshot": True}, headers=h); prov = r.json()
    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()
    assert ds2["verification"] == "unverified"
    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"])
    out3 = await run_job_inline(pool, "export", {"dataset_id": ds2["id"], "verified": False}); assert out3["verification_label"] == "unverified"