1"""Integration tests through the ASGI app with fake workers (no MLX required)."""23from __future__ import annotations45import asyncio6import json78import pytest910from .conftest import install_fake_adapters1112pytestmark = pytest.mark.asyncio131415# --------------------------------------------------------------------------- auth1617async def test_auth_flow(client):18 r = await client.get("/api/auth/status")19 assert r.json()["needs_setup"] is False and r.json()["authenticated"] is False20 r = await client.get("/api/models")21 assert r.status_code == 40122 r = await client.post("/api/auth/login", json={"email": "admin@test.local", "password": "wrong"})23 assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_CREDENTIALS"24 r = await client.post("/api/auth/login", json={"email": "admin@test.local", "password": "correct-horse-battery"})25 assert r.status_code == 20026 r = await client.get("/api/models")27 assert r.status_code == 20028 # mutation without CSRF header is refused for sessions29 r = await client.post("/api/models/rescan")30 assert r.status_code == 403 and r.json()["error"]["code"] == "CSRF"313233async def test_api_keys(admin):34 r = await admin.post("/api/keys", json={"name": "laptop", "scopes": ["inference"]})35 key = r.json()["key"]36 assert key.startswith("llm_live_")37 r = await admin.get("/api/keys")38 assert r.json()["keys"][0]["prefix"] == key[:16] and "key_hash" not in r.json()["keys"][0]39 # key works for inference endpoints, not for admin40 bare = admin41 bare.cookies.clear()42 r = await bare.get("/v1/models", headers={"Authorization": f"Bearer {key}"})43 assert r.status_code == 20044 r = await bare.get("/api/models", headers={"Authorization": f"Bearer {key}"})45 assert r.status_code == 40346 r = await bare.get("/v1/models", headers={"Authorization": "Bearer llm_live_bogus"})47 assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_API_KEY"48 r = await bare.get("/v1/models")49 assert r.status_code == 401505152async def test_revoked_key(admin):53 r = await admin.post("/api/keys", json={"name": "k", "scopes": ["inference"]})54 key, kid = r.json()["key"], r.json()["record"]["id"]55 await admin.delete(f"/api/keys/{kid}")56 admin.cookies.clear()57 r = await admin.get("/v1/models", headers={"Authorization": f"Bearer {key}"})58 assert r.status_code == 401596061# --------------------------------------------------------------------------- registry6263async def test_registry_scan(admin):64 r = await admin.get("/api/models")65 models = {m["id"]: m for m in r.json()["models"]}66 assert "qwen3-test-4bit" in models and "llama-test-4bit" in models and "gemma-test-q4-k-m" in models67 q = models["qwen3-test-4bit"]68 assert q["runtime"] == "mlx" and q["quantization"] == "4bit" and q["compatibility_status"] in ("compatible", "compatible_with_restrictions")69 assert q["thinking"] is True and q["kv_bytes_per_token"] == 2 * 4 * 2 * 64 * 270 g = models["gemma-test-q4-k-m"]71 assert g["runtime"] == "llamacpp" and g["format"] == "gguf" and g["quantization"] == "Q4_K_M"72 huge = models["huge-test-4bit"]73 assert huge["compatibility_status"] in ("incompatible", "not_recommended") and huge["compatible"] is False747576async def test_rescan_detects_missing(admin, root):77 import shutil78 shutil.rmtree(root / "models" / "mlx" / "llama")79 r = await admin.post("/api/models/rescan")80 assert r.status_code == 20081 r = await admin.get("/api/models?include_missing=true")82 m = next(x for x in r.json()["models"] if x["id"] == "llama-test-4bit")83 assert m["installed"] is False84 # files are never deleted by a scan85 assert (root / "models" / "mlx" / "qwen" / "Qwen3-Test-4bit" / "model.safetensors").exists()868788async def test_model_detail_and_patch(admin):89 r = await admin.get("/api/models/qwen3-test-4bit")90 assert r.status_code == 200 and "memory_curve" in r.json()91 r = await admin.patch("/api/models/qwen3-test-4bit", json={"favorite": True, "tags": ["coding"]})92 assert r.json()["favorite"] is True and "coding" in r.json()["tags"]93 r = await admin.post("/api/models/qwen3-test-4bit/pin?pinned=true")94 assert r.json()["pinned"] is True95 r = await admin.get("/api/models/nope")96 assert r.status_code == 404 and r.json()["error"]["code"] == "MODEL_NOT_FOUND"979899async def test_aliases(admin, api_key):100 r = await admin.put("/api/aliases", json={"alias": "fast", "model_id": "qwen3-test-4bit"})101 assert r.status_code == 200 and r.json()["aliases"]["fast"] == "qwen3-test-4bit"102 r = await admin.put("/api/aliases", json={"alias": "auto", "model_id": "qwen3-test-4bit"})103 assert r.status_code == 400104 admin.cookies.clear()105 r = await admin.get("/v1/models", headers={"Authorization": f"Bearer {api_key}"})106 ids = [m["id"] for m in r.json()["data"]]107 assert "fast" in ids and "qwen3-test-4bit" in ids108109110# --------------------------------------------------------------------------- inference (fake worker)111112async def test_load_on_demand_chat_and_unload(admin, api_key, app):113 h = {"Authorization": f"Bearer {api_key}"}114 admin.cookies.clear()115 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": [{"role": "user", "content": "hi"}]})116 assert r.status_code == 200, r.text117 d = r.json()118 assert d["choices"][0]["message"]["content"] == "echo: hi" and d["usage"]["total_tokens"] == 10 and d["timings"]["generation_tps"] == 55.5119 assert app.state.manager.status_of("qwen3-test-4bit") == "ready"120 # streaming121 async with admin.stream("POST", "/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "stream": True,122 "messages": [{"role": "user", "content": "a b c"}]}) as s:123 assert s.status_code == 200124 lines = [l async for l in s.aiter_lines() if l.startswith("data: ")]125 assert lines[-1] == "data: [DONE]"126 chunks = [json.loads(l[6:]) for l in lines[:-1]]127 text = "".join((c["choices"][0]["delta"].get("content") or "") for c in chunks if c.get("choices"))128 assert text.strip() == "echo: a b c"129 assert chunks[-1]["usage"]["completion_tokens"] == 3130 # completions + embeddings131 r = await admin.post("/v1/completions", headers=h, json={"model": "qwen3-test-4bit", "prompt": "hello"})132 assert r.status_code == 200 and r.json()["choices"][0]["text"] == " world"133 r = await admin.post("/v1/embeddings", headers=h, json={"model": "qwen3-test-4bit", "input": ["a", "b"]})134 assert r.status_code == 200 and len(r.json()["data"]) == 2135 # request log recorded136 await asyncio.sleep(0.1)137 n = await app.state.db.scalar("SELECT COUNT(*) FROM inference_requests WHERE status=200")138 assert n >= 4139 # model switch: llama evicts qwen (max 1 model)140 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "llama-test-4bit", "messages": [{"role": "user", "content": "yo"}]})141 assert r.status_code == 200142 assert app.state.manager.status_of("llama-test-4bit") == "ready"143 assert app.state.manager.status_of("qwen3-test-4bit") == "unloaded"144 # manual unload via admin145 await admin.post("/api/auth/login", json={"email": "admin@test.local", "password": "correct-horse-battery"})146 r = await admin.post("/api/models/llama-test-4bit/unload")147 assert r.json()["ok"] is True and app.state.manager.loaded == {}148149150async def test_model_not_found_and_validation(admin, api_key):151 h = {"Authorization": f"Bearer {api_key}"}152 admin.cookies.clear()153 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "ghost", "messages": [{"role": "user", "content": "x"}]})154 assert r.status_code == 404 and r.json()["error"]["code"] == "MODEL_NOT_FOUND"155 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": []})156 assert r.status_code == 400 and r.json()["error"]["type"] == "invalid_request_error"157 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": [{"role": "user", "content": "x"}], "temperature": 9})158 assert r.status_code == 400 and r.json()["error"]["param"] == "temperature"159 r = await admin.post("/v1/chat/completions", headers=h, content=b"{not json")160 assert r.status_code == 400161162163async def test_memory_rejection(admin, api_key, app):164 h = {"Authorization": f"Bearer {api_key}"}165 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "huge-test-4bit", "messages": [{"role": "user", "content": "x"}]})166 assert r.status_code in (422, 507)167 assert r.json()["error"]["code"] in ("MODEL_INCOMPATIBLE", "MODEL_TOO_LARGE")168 # lower the budget -> a normally fine model becomes too large for a manual load169 r = await admin.patch("/api/settings", json={"max_model_memory_gb": 1})170 assert r.status_code == 200171 r = await admin.post("/api/models/qwen3-test-4bit/load")172 assert r.status_code in (422, 507), r.text173 r = await admin.get("/api/models/qwen3-test-4bit")174 assert r.json()["compatible"] is False175 await admin.patch("/api/settings", json={"max_model_memory_gb": 45})176177178async def test_worker_crash_is_reported(admin, api_key, app):179 install_fake_adapters(app.state.manager, crash=True)180 h = {"Authorization": f"Bearer {api_key}"}181 r = await admin.post("/v1/chat/completions", headers=h, json={"model": "qwen3-test-4bit", "messages": [{"role": "user", "content": "x"}]})182 # warm-up already generates -> crash during load -> clean structured error183 assert r.status_code in (502, 503), r.text184 assert r.json()["error"]["code"] in ("WORKER_CRASHED", "MODEL_LOAD_FAILED", "WORKER_UNREACHABLE")185 assert app.state.manager.loaded == {}186 install_fake_adapters(app.state.manager)187188189async def test_concurrent_requests_and_switch_lock(admin, api_key, app):190 h = {"Authorization": f"Bearer {api_key}"}191 admin.cookies.clear()192193 async def ask(model, text):194 return await admin.post("/v1/chat/completions", headers=h, json={"model": model, "messages": [{"role": "user", "content": text}]})195196 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"))197 assert all(r.status_code == 200 for r in rs), [r.text for r in rs if r.status_code != 200]198 # never more than one large model resident199 assert len(app.state.manager.loaded) <= 1200 assert app.state.manager.stats["loads"] >= 2201202203async def test_load_timeout(admin, app, monkeypatch):204 install_fake_adapters(app.state.manager, load_delay=60)205 monkeypatch.setattr(app.state.settings, "load_timeout_seconds", 2)206 r = await admin.post("/api/models/qwen3-test-4bit/load")207 assert r.status_code == 503 and r.json()["error"]["code"] == "MODEL_LOAD_TIMEOUT"208 assert app.state.manager.loaded == {}209 install_fake_adapters(app.state.manager)210211212async def test_delete_requires_confirmation(admin, root):213 r = await admin.request("DELETE", "/api/models/llama-test-4bit", json={"confirm": "nope"})214 assert r.status_code == 400 and r.json()["error"]["code"] == "CONFIRMATION_REQUIRED"215 assert (root / "models" / "mlx" / "llama" / "Llama-Test-4bit").exists()216 r = await admin.request("DELETE", "/api/models/llama-test-4bit", json={"confirm": "llama-test-4bit"})217 assert r.status_code == 200218 assert not (root / "models" / "mlx" / "llama" / "Llama-Test-4bit").exists()219 r = await admin.get("/api/models/llama-test-4bit")220 assert r.status_code == 404221222223async def test_system_endpoints(admin):224 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"):225 r = await admin.get(path)226 assert r.status_code == 200, path227 r = await admin.get("/health")228 d = r.json()229 assert d["status"] == "ok" and "chip" in d["hardware"] and "available_gb" in d["memory"]230231232async def test_settings_validation(admin):233 r = await admin.patch("/api/settings", json={"max_model_memory_gb": 999})234 assert r.status_code == 400235 r = await admin.patch("/api/settings", json={"log_prompts": True, "unknown_key": 1})236 assert r.json()["changed"] == {"log_prompts": True}237238239async def test_low_disk_warning_blocks_download(admin, app, monkeypatch):240 # Pretend the disk is almost full: inspect must report disk.ok False and start_download must refuse241 import shutil as _sh242 from llm_api import downloads as dl243244 class FakeInfo:245 siblings = [type("S", (), {"rfilename": "model.safetensors", "size": 10 * 1024**3})(), type("S", (), {"rfilename": "config.json", "size": 100})()]246 tags = ["mlx", "base_model:quantized:Qwen/Qwen3-Test"]247 library_name = "mlx"248 config = {"model_type": "qwen3", "architectures": ["Qwen3ForCausalLM"], "num_hidden_layers": 4, "num_attention_heads": 4,249 "num_key_value_heads": 2, "hidden_size": 256, "max_position_embeddings": 8192, "quantization": {"bits": 4}}250 pipeline_tag = "text-generation"251 downloads = 1000252 likes = 5253 last_modified = "2026-01-01"254 gated = False255256 class FakeApi:257 def model_info(self, repo, files_metadata=True):258 return FakeInfo()259260 monkeypatch.setattr(app.state.downloader, "_api", lambda: FakeApi())261 Usage = type("U", (), {})262 def fake_du(path):263 u = Usage(); u.total = 1000 * 1024**3; u.used = 950 * 1024**3; u.free = 50 * 1024**3264 return u265 monkeypatch.setattr(dl.shutil, "disk_usage", fake_du)266 r = await admin.post("/api/models/inspect", json={"repository": "mlx-community/Qwen3-Test-4bit"})267 assert r.status_code == 200 and r.json()["disk"]["ok"] is False268 r = await admin.post("/api/models/download", json={"repository": "mlx-community/Qwen3-Test-4bit"})269 assert r.status_code == 507 and r.json()["error"]["code"] == "INSUFFICIENT_DISK"270 r = await admin.post("/api/models/inspect", json={"repository": "../../etc/passwd"})271 assert r.status_code == 400272273274async def test_benchmark_job(admin, app):275 r = await admin.post("/api/models/qwen3-test-4bit/benchmark", json={"max_tokens": 8, "runs": 1, "long_prompt": False})276 job_id = r.json()["job"]["id"]277 for _ in range(100):278 await asyncio.sleep(0.2)279 j = app.state.jobs.get(job_id)280 if j and j.status in ("completed", "failed"):281 break282 assert j.status == "completed", j.error283 r = await admin.get("/api/models/qwen3-test-4bit/benchmarks")284 b = r.json()["benchmarks"][0]285 assert b["load_ms"] is not None and b["generation_tps"] is not None286287288async def test_restart_clears_stale_state(client, app, root):289 """A model marked loaded in a previous life must come back as unloaded (workers file cleanup)."""290 from llm_api.manager import ModelManager291 wf = root / "data" / "workers.json"292 wf.parent.mkdir(parents=True, exist_ok=True)293 wf.write_text(json.dumps({"qwen3-test-4bit": {"pid": 999999, "port": 18499, "runtime": "mlx"}}))294 m = ModelManager(app.state.settings, app.state.db, app.state.manager.registry)295 await m._recover_stale_workers()296 assert json.loads(wf.read_text()) == {}297 assert m.status_of("qwen3-test-4bit") == "unloaded"298 await m.client.aclose()299