File size: 1,778 Bytes
829be0a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | import asyncio
import httpx
import pytest
from server.main import build_app
pytestmark = pytest.mark.asyncio
async def test_activate_then_status_loaded(monkeypatch, fake_classes):
monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
monkeypatch.setattr("server.main.select_device", lambda: "cpu")
app = build_app()
from tests.conftest import lifespan_ctx
transport = httpx.ASGITransport(app=app)
async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
r = await c.post("/api/models/fake/activate")
assert r.status_code in (200, 202)
for _ in range(20):
s = (await c.get("/api/models/active")).json()
if s["status"] == "loaded":
break
await asyncio.sleep(0.05)
assert s["id"] == "fake"
assert s["status"] == "loaded"
async def test_activate_unknown_returns_404(monkeypatch, fake_classes):
monkeypatch.setattr("server.main._discover_adapter_classes", lambda: fake_classes)
monkeypatch.setattr("server.main.select_device", lambda: "cpu")
app = build_app()
from tests.conftest import lifespan_ctx
transport = httpx.ASGITransport(app=app)
async with lifespan_ctx(app), httpx.AsyncClient(transport=transport, base_url="http://t") as c:
r = await c.post("/api/models/nope/activate")
assert r.status_code == 404
assert r.json()["error"]["code"] == "model_not_found"
# Note: integration test for /api/models/active/events SSE stream is omitted.
# Registry event emission is unit-tested in tests/test_registry.py
# (test_emits_loading_then_loaded_events). The /api/models/active/events
# endpoint is a thin sse-starlette wrapper around that generator.
|