import asyncio
import os
import subprocess
import pytest
from tests.conftest import run_job_inline


async def test_opaque_mcp_labelled_and_blocked(app_client, pool):
    c = app_client; h = c.auth_headers
    await c.post("/api/v1/projects", json={"slug": "mcp", "name": "mcp", "policy_mode": "discovery"}, headers=h)
    mc = {"server": "mock-mcp", "tools": [{"name": "lookup", "description": "hidden implementation", "inputSchema": {"type": "object", "properties": {"q": {"type": "string"}}, "required": ["q"]}}],
          "invocations": [{"tool": "lookup", "input": {"q": "a"}, "output": {"q": "a", "answer": 42}}]}
    res = await run_job_inline(pool, "discover", {"project_id": "mcp", "kind": "mcp", "name": "Mock MCP", "mcp_contract": mc, "created_by": "t"})
    api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json()
    assert api["kind"] == "mcp" and api["opaque_internal_implementation"] is True
    assert any(d["kind"] == "opaque_mcp" and d["severity"] == "critical" for d in api["debt"])
    pkg = (await c.get(f"/api/v1/akps/{res['package_id']}", headers=h)).json(); assert pkg["manifest"]["opaque_internal_implementation"] is True
    ex = [e for e in (await c.get("/api/v1/executions", params={"project_id": "mcp"}, headers=h)).json()["executions"]]
    assert ex and all(e["observed"] is False for e in ex)
    r = await c.post("/api/v1/datasets", json={"project_id": "mcp", "name": "mcp-out", "execution_ids": [ex[0]["id"]]}, headers=h); ds = r.json()
    assert ds["verification"] == "unverified" and any(ch["check"].endswith("not_opaque") and not ch["ok"] for ch in ds["assessment"]["checks"])
    out = await run_job_inline(pool, "export", {"dataset_id": ds["id"], "verified": True}); assert out["blocked"]
    # a research-mode project refuses to execute an opaque package? allowed but flagged; verified export stays blocked
    assert (await run_job_inline(pool, "export", {"dataset_id": ds["id"], "verified": False}))["verification_label"] == "unverified"


async def test_drift_detection_versioning_revalidation(app_client, pool, mock_api):
    c = app_client; h = c.auth_headers
    await c.post("/api/v1/projects", json={"slug": "drift", "name": "drift", "policy_mode": "development"}, headers=h)
    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"})
    await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True})
    await c.post("/api/v1/execute", json={"project_id": "drift", "package_id": res["package_id"], "operation_id": "listItems", "query": {"page": 1}}, headers=h)  # dependent project
    ok = await run_job_inline(pool, "drift_check", {"api_version_id": res["version_id"]}); assert not ok["drifted"]
    mock_api["state"]["drift"] = True
    try:
        bad = await run_job_inline(pool, "drift_check", {"api_version_id": res["version_id"]})
        assert bad["drifted"] and bad["alert_id"]
        proj = (await c.get("/api/v1/projects/drift", headers=h)).json()["id"]
        assert proj in bad["affected_projects"]
        api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json(); assert api["lifecycle"] == "stale"
        assert any(d["kind"] == "stale_contract" and d["status"] == "open" for d in api["debt"])
        pkg = (await c.get(f"/api/v1/akps/{res['package_id']}", headers=h)).json(); assert pkg["lifecycle"] == "stale"
        alerts = (await c.get("/api/v1/drift", headers=h)).json()["alerts"]; assert alerts[0]["status"] == "open"
    finally:
        mock_api["state"]["drift"] = False
    # fix: re-verify (upstream restored) → version revalidated, alert resolved, debt resolved, lifecycle back to tested
    v2 = await run_job_inline(pool, "verify", {"package_id": res["package_id"], "live": True}); assert v2["promoted_to"] == "tested"
    api = (await c.get(f"/api/v1/apis/{res['api_id']}", headers=h)).json(); assert api["lifecycle"] == "tested"
    alerts = (await c.get("/api/v1/drift", headers=h)).json()["alerts"]; assert alerts[0]["status"] == "resolved"
    assert not any(d["kind"] == "stale_contract" and d["status"] == "open" for d in api["debt"])


async def test_worker_crash_recovery_and_audit_intact(pool):
    """A job whose worker died (stale heartbeat) is requeued and completed by another worker; audit chain stays valid."""
    from akos.jobs import queue
    from akos import audit
    async with pool.acquire() as conn:
        job = await queue.enqueue(conn, "maintenance", {}, idempotency_key="recovery-test")
        claimed = await queue.claim(conn, "worker-A", ["maintenance"]); assert claimed and claimed["id"] == job["id"]
        # worker-A crashes: no completion, heartbeat goes stale
        await conn.execute("UPDATE jobs SET heartbeat_at = now() - interval '10 minutes' WHERE id = $1", job["id"])
        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 "")
        await queue.complete(conn, job["id"], {"ok": True})
        row = await conn.fetchrow("SELECT state, attempts FROM jobs WHERE id = $1", job["id"]); assert row["state"] == "done" and row["attempts"] == 2
        # max attempts → dead letter, retry endpoint semantics
        j2 = await queue.enqueue(conn, "maintenance", {}, idempotency_key="dead-test", max_attempts=1)
        cl = await queue.claim(conn, "w", ["maintenance"]); assert (await queue.fail(conn, cl, "boom")) == "dead"
        assert (await queue.retry(conn, j2["id"]))["state"] == "queued"
        assert (await audit.verify_chain(conn))["ok"]


@pytest.mark.node
def test_backup_and_restore_scripts():
    """pg_dump/pg_restore round trip into akos_restore_test (needs Postgres tools on the host)."""
    from akos.settings import settings
    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"}
    r = subprocess.run(["/bin/bash", str(settings.root / "scripts" / "backup.sh")], capture_output=True, text=True, env=env, timeout=300)
    assert r.returncode == 0, r.stderr
    dumps = sorted(settings.backup_dir.glob("akos-*.dump")); assert dumps
    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)
    assert r2.returncode == 0 and "restored" in r2.stdout, r2.stderr
    assert "audit events" in r2.stdout