"""Integration tests through the ASGI app with fake workers (no MLX required)."""

from __future__ import annotations

import asyncio
import json

import pytest

from .conftest import install_fake_adapters

pytestmark = pytest.mark.asyncio


# --------------------------------------------------------------------------- auth

async def test_auth_flow(client):
    r = await client.get("/api/auth/status")
    assert r.json()["needs_setup"] is False and r.json()["authenticated"] is False
    r = await client.get("/api/models")
    assert r.status_code == 401
    r = await client.post("/api/auth/login", json={"email": "admin@test.local", "password": "wrong"})
    assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_CREDENTIALS"
    r = await client.post("/api/auth/login", json={"email": "admin@test.local", "password": "correct-horse-battery"})
    assert r.status_code == 200
    r = await client.get("/api/models")
    assert r.status_code == 200
    # mutation without CSRF header is refused for sessions
    r = await client.post("/api/models/rescan")
    assert r.status_code == 403 and r.json()["error"]["code"] == "CSRF"


async def test_api_keys(admin):
    r = await admin.post("/api/keys", json={"name": "laptop", "scopes": ["inference"]})
    key = r.json()["key"]
    assert key.startswith("llm_live_")
    r = await admin.get("/api/keys")
    assert r.json()["keys"][0]["prefix"] == key[:16] and "key_hash" not in r.json()["keys"][0]
    # key works for inference endpoints, not for admin
    bare = admin
    bare.cookies.clear()
    r = await bare.get("/v1/models", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 200
    r = await bare.get("/api/models", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 403
    r = await bare.get("/v1/models", headers={"Authorization": "Bearer llm_live_bogus"})
    assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_API_KEY"
    r = await bare.get("/v1/models")
    assert r.status_code == 401


async def test_revoked_key(admin):
    r = await admin.post("/api/keys", json={"name": "k", "scopes": ["inference"]})
    key, kid = r.json()["key"], r.json()["record"]["id"]
    await admin.delete(f"/api/keys/{kid}")
    admin.cookies.clear()
    r = await admin.get("/v1/models", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 401


# --------------------------------------------------------------------------- registry

async def test_registry_scan(admin):
    r = await admin.get("/api/models")
    models = {m["id"]: m for m in r.json()["models"]}
    assert "qwen3-test-4bit" in models and "llama-test-4bit" in models and "gemma-test-q4-k-m" in models
    q = models["qwen3-test-4bit"]
    assert q["runtime"] == "mlx" and q["quantization"] == "4bit" and q["compatibility_status"] in ("compatible", "compatible_with_restrictions")
    assert q["thinking"] is True and q["kv_bytes_per_token"] == 2 * 4 * 2 * 64 * 2
    g = models["gemma-test-q4-k-m"]
    assert g["runtime"] == "llamacpp" and g["format"] == "gguf" and g["quantization"] == "Q4_K_M"
    huge = models["huge-test-4bit"]
    assert huge["compatibility_status"] in ("incompatible", "not_recommended") and huge["compatible"] is False


async def test_rescan_detects_missing(admin, root):
    import shutil
    shutil.rmtree(root / "models" / "mlx" / "llama")
    r = await admin.post("/api/models/rescan")
    assert r.status_code == 200
    r = await admin.get("/api/models?include_missing=true")
    m = next(x for x in r.json()["models"] if x["id"] == "llama-test-4bit")
    assert m["installed"] is False
    # files are never deleted by a scan
    assert (root / "models" / "mlx" / "qwen" / "Qwen3-Test-4bit" / "model.safetensors").exists()


async def test_model_detail_and_patch(admin):
    r = await admin.get("/api/models/qwen3-test-4bit")
    assert r.status_code == 200 and "memory_curve" in r.json()
    r = await admin.patch("/api/models/qwen3-test-4bit", json={"favorite": True, "tags": ["coding"]})
    assert r.json()["favorite"] is True and "coding" in r.json()["tags"]
    r = await admin.post("/api/models/qwen3-test-4bit/pin?pinned=true")
    assert r.json()["pinned"] is True
    r = await admin.get("/api/models/nope")
    assert r.status_code == 404 and r.json()["error"]["code"] == "MODEL_NOT_FOUND"


async def test_aliases(admin, api_key):
    r = await admin.put("/api/aliases", json={"alias": "fast", "model_id": "qwen3-test-4bit"})
    assert r.status_code == 200 and r.json()["aliases"]["fast"] == "qwen3-test-4bit"
    r = await admin.put("/api/aliases", json={"alias": "auto", "model_id": "qwen3-test-4bit"})
    assert r.status_code == 400
    admin.cookies.clear()
    r = await admin.get("/v1/models", headers={"Authorization": f"Bearer {api_key}"})
    ids = [m["id"] for m in r.json()["data"]]
    assert "fast" in ids and "qwen3-test-4bit" in ids


# --------------------------------------------------------------------------- inference (fake worker)

async def test_load_on_demand_chat_and_unload(admin, api_key, app):
    h = {"Authorization": f"Bearer {api_key}"}
    admin.cookies.clear()
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": [{"role": "user", "content": "hi"}]})
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["choices"][0]["message"]["content"] == "echo: hi" and d["usage"]["total_tokens"] == 10 and d["timings"]["generation_tps"] == 55.5
    assert app.state.manager.status_of("qwen3-test-4bit") == "ready"
    # streaming
    async with admin.stream("POST", "/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "stream": True,
                                                                             "messages": [{"role": "user", "content": "a b c"}]}) as s:
        assert s.status_code == 200
        lines = [l async for l in s.aiter_lines() if l.startswith("data: ")]
    assert lines[-1] == "data: [DONE]"
    chunks = [json.loads(l[6:]) for l in lines[:-1]]
    text = "".join((c["choices"][0]["delta"].get("content") or "") for c in chunks if c.get("choices"))
    assert text.strip() == "echo: a b c"
    assert chunks[-1]["usage"]["completion_tokens"] == 3
    # completions + embeddings
    r = await admin.post("/v1/completions", headers=h, json={"model": "qwen3-test-4bit", "prompt": "hello"})
    assert r.status_code == 200 and r.json()["choices"][0]["text"] == " world"
    r = await admin.post("/v1/embeddings", headers=h, json={"model": "qwen3-test-4bit", "input": ["a", "b"]})
    assert r.status_code == 200 and len(r.json()["data"]) == 2
    # request log recorded
    await asyncio.sleep(0.1)
    n = await app.state.db.scalar("SELECT COUNT(*) FROM inference_requests WHERE status=200")
    assert n >= 4
    # model switch: llama evicts qwen (max 1 model)
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "llama-test-4bit", "messages": [{"role": "user", "content": "yo"}]})
    assert r.status_code == 200
    assert app.state.manager.status_of("llama-test-4bit") == "ready"
    assert app.state.manager.status_of("qwen3-test-4bit") == "unloaded"
    # manual unload via admin
    await admin.post("/api/auth/login", json={"email": "admin@test.local", "password": "correct-horse-battery"})
    r = await admin.post("/api/models/llama-test-4bit/unload")
    assert r.json()["ok"] is True and app.state.manager.loaded == {}


async def test_model_not_found_and_validation(admin, api_key):
    h = {"Authorization": f"Bearer {api_key}"}
    admin.cookies.clear()
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "ghost", "messages": [{"role": "user", "content": "x"}]})
    assert r.status_code == 404 and r.json()["error"]["code"] == "MODEL_NOT_FOUND"
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": []})
    assert r.status_code == 400 and r.json()["error"]["type"] == "invalid_request_error"
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": [{"role": "user", "content": "x"}], "temperature": 9})
    assert r.status_code == 400 and r.json()["error"]["param"] == "temperature"
    r = await admin.post("/v1/chat/completions", headers=h, content=b"{not json")
    assert r.status_code == 400


async def test_memory_rejection(admin, api_key, app):
    h = {"Authorization": f"Bearer {api_key}"}
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "huge-test-4bit", "messages": [{"role": "user", "content": "x"}]})
    assert r.status_code in (422, 507)
    assert r.json()["error"]["code"] in ("MODEL_INCOMPATIBLE", "MODEL_TOO_LARGE")
    # lower the budget -> a normally fine model becomes too large for a manual load
    r = await admin.patch("/api/settings", json={"max_model_memory_gb": 1})
    assert r.status_code == 200
    r = await admin.post("/api/models/qwen3-test-4bit/load")
    assert r.status_code in (422, 507), r.text
    r = await admin.get("/api/models/qwen3-test-4bit")
    assert r.json()["compatible"] is False
    await admin.patch("/api/settings", json={"max_model_memory_gb": 45})


async def test_worker_crash_is_reported(admin, api_key, app):
    install_fake_adapters(app.state.manager, crash=True)
    h = {"Authorization": f"Bearer {api_key}"}
    r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": [{"role": "user", "content": "x"}]})
    # warm-up already generates -> crash during load -> clean structured error
    assert r.status_code in (502, 503), r.text
    assert r.json()["error"]["code"] in ("WORKER_CRASHED", "MODEL_LOAD_FAILED", "WORKER_UNREACHABLE")
    assert app.state.manager.loaded == {}
    install_fake_adapters(app.state.manager)


async def test_concurrent_requests_and_switch_lock(admin, api_key, app):
    h = {"Authorization": f"Bearer {api_key}"}
    admin.cookies.clear()

    async def ask(model, text):
        return await admin.post("/v1/chat/completions", headers=h, json={"model": model, "messages": [{"role": "user", "content": text}]})

    rs = await asyncio.gather(*[ask("qwen3-test-4bit", f"q{i}") for i in range(5)], ask("llama-test-4bit", "l1"), ask("qwen3-test-4bit", "q9"))
    assert all(r.status_code == 200 for r in rs), [r.text for r in rs if r.status_code != 200]
    # never more than one large model resident
    assert len(app.state.manager.loaded) <= 1
    assert app.state.manager.stats["loads"] >= 2


async def test_load_timeout(admin, app, monkeypatch):
    install_fake_adapters(app.state.manager, load_delay=60)
    monkeypatch.setattr(app.state.settings, "load_timeout_seconds", 2)
    r = await admin.post("/api/models/qwen3-test-4bit/load")
    assert r.status_code == 503 and r.json()["error"]["code"] == "MODEL_LOAD_TIMEOUT"
    assert app.state.manager.loaded == {}
    install_fake_adapters(app.state.manager)


async def test_delete_requires_confirmation(admin, root):
    r = await admin.request("DELETE", "/api/models/llama-test-4bit", json={"confirm": "nope"})
    assert r.status_code == 400 and r.json()["error"]["code"] == "CONFIRMATION_REQUIRED"
    assert (root / "models" / "mlx" / "llama" / "Llama-Test-4bit").exists()
    r = await admin.request("DELETE", "/api/models/llama-test-4bit", json={"confirm": "llama-test-4bit"})
    assert r.status_code == 200
    assert not (root / "models" / "mlx" / "llama" / "Llama-Test-4bit").exists()
    r = await admin.get("/api/models/llama-test-4bit")
    assert r.status_code == 404


async def test_system_endpoints(admin):
    for path in ("/api/system", "/api/system/memory", "/api/system/gpu", "/api/system/storage", "/api/system/processes", "/api/runtime/status", "/api/settings", "/api/system/metrics"):
        r = await admin.get(path)
        assert r.status_code == 200, path
    r = await admin.get("/health")
    d = r.json()
    assert d["status"] == "ok" and "chip" in d["hardware"] and "available_gb" in d["memory"]


async def test_settings_validation(admin):
    r = await admin.patch("/api/settings", json={"max_model_memory_gb": 999})
    assert r.status_code == 400
    r = await admin.patch("/api/settings", json={"log_prompts": True, "unknown_key": 1})
    assert r.json()["changed"] == {"log_prompts": True}


async def test_low_disk_warning_blocks_download(admin, app, monkeypatch):
    # Pretend the disk is almost full: inspect must report disk.ok False and start_download must refuse
    import shutil as _sh
    from llm_api import downloads as dl

    class FakeInfo:
        siblings = [type("S", (), {"rfilename": "model.safetensors", "size": 10 * 1024**3})(), type("S", (), {"rfilename": "config.json", "size": 100})()]
        tags = ["mlx", "base_model:quantized:Qwen/Qwen3-Test"]
        library_name = "mlx"
        config = {"model_type": "qwen3", "architectures": ["Qwen3ForCausalLM"], "num_hidden_layers": 4, "num_attention_heads": 4,
                  "num_key_value_heads": 2, "hidden_size": 256, "max_position_embeddings": 8192, "quantization": {"bits": 4}}
        pipeline_tag = "text-generation"
        downloads = 1000
        likes = 5
        last_modified = "2026-01-01"
        gated = False

    class FakeApi:
        def model_info(self, repo, files_metadata=True):
            return FakeInfo()

    monkeypatch.setattr(app.state.downloader, "_api", lambda: FakeApi())
    Usage = type("U", (), {})
    def fake_du(path):
        u = Usage(); u.total = 1000 * 1024**3; u.used = 950 * 1024**3; u.free = 50 * 1024**3
        return u
    monkeypatch.setattr(dl.shutil, "disk_usage", fake_du)
    r = await admin.post("/api/models/inspect", json={"repository": "mlx-community/Qwen3-Test-4bit"})
    assert r.status_code == 200 and r.json()["disk"]["ok"] is False
    r = await admin.post("/api/models/download", json={"repository": "mlx-community/Qwen3-Test-4bit"})
    assert r.status_code == 507 and r.json()["error"]["code"] == "INSUFFICIENT_DISK"
    r = await admin.post("/api/models/inspect", json={"repository": "../../etc/passwd"})
    assert r.status_code == 400


async def test_benchmark_job(admin, app):
    r = await admin.post("/api/models/qwen3-test-4bit/benchmark", json={"max_tokens": 8, "runs": 1, "long_prompt": False})
    job_id = r.json()["job"]["id"]
    for _ in range(100):
        await asyncio.sleep(0.2)
        j = app.state.jobs.get(job_id)
        if j and j.status in ("completed", "failed"):
            break
    assert j.status == "completed", j.error
    r = await admin.get("/api/models/qwen3-test-4bit/benchmarks")
    b = r.json()["benchmarks"][0]
    assert b["load_ms"] is not None and b["generation_tps"] is not None


async def test_restart_clears_stale_state(client, app, root):
    """A model marked loaded in a previous life must come back as unloaded (workers file cleanup)."""
    from llm_api.manager import ModelManager
    wf = root / "data" / "workers.json"
    wf.parent.mkdir(parents=True, exist_ok=True)
    wf.write_text(json.dumps({"qwen3-test-4bit": {"pid": 999999, "port": 18499, "runtime": "mlx"}}))
    m = ModelManager(app.state.settings, app.state.db, app.state.manager.registry)
    await m._recover_stale_workers()
    assert json.loads(wf.read_text()) == {}
    assert m.status_of("qwen3-test-4bit") == "unloaded"
    await m.client.aclose()