Spaces:
Paused
Paused
| """ | |
| E2E test klíčového požadavku v5: ZMĚNA NASTAVENÍ ZA BĚHU BEZ RESTARTU. | |
| Scénář: | |
| 1. Aplikace běží, engine načte model A (mock vLLM subprocess). | |
| 2. Passthrough /v1/chat/completions odpovídá přes model A. | |
| 3. POST /admin/settings změní model na B => engine se reloadne NA POZADÍ. | |
| 4. Aplikace (/health) zůstává dostupná PO CELOU DOBU reloadu. | |
| 5. Po reloadu odpovídá model B; aplikace se nikdy nerestartovala. | |
| """ | |
| import time | |
| from fastapi.testclient import TestClient | |
| def _wait_ready(engine, timeout=20): | |
| deadline = time.time() + timeout | |
| while time.time() < deadline: | |
| if engine.is_ready: | |
| return True | |
| if engine.status_dict()["state"] == "error": | |
| raise AssertionError(f"Engine error: {engine.status_dict()}") | |
| time.sleep(0.1) | |
| return False | |
| def test_hot_swap_without_app_restart(app_module, auth): | |
| c = TestClient(app_module.app) | |
| eng = app_module.ENGINE | |
| # 1) start s modelem A | |
| r = c.post("/admin/settings", json={"model": "org/model-A"}, headers=auth) | |
| assert r.json()["engine_reload_started"] is True | |
| assert _wait_ready(eng), eng.status_dict() | |
| assert eng.current_model == "org/model-A" | |
| # 2) inference přes model A (passthrough bez agentní smyčky) | |
| r = c.post("/v1/chat/completions", headers=auth, | |
| json={"model": "code-agent-llm", | |
| "messages": [{"role": "user", "content": "ping"}]}) | |
| assert r.status_code == 200 | |
| assert r.json()["choices"][0]["message"]["content"] == "mock:org/model-A:ping" | |
| # 3) hot-swap na model B | |
| r = c.post("/admin/settings", json={"model": "org/model-B"}, headers=auth) | |
| data = r.json() | |
| assert data["applied"] == {"model": "org/model-B"} | |
| assert data["engine_reload_started"] is True | |
| # 4) BĚHEM reloadu aplikace žije a hlásí stav | |
| saw_loading = False | |
| deadline = time.time() + 20 | |
| while time.time() < deadline: | |
| h = c.get("/health") | |
| assert h.status_code == 200 # aplikace se nikdy nerestartovala | |
| state = h.json()["engine"]["state"] | |
| if state == "starting": | |
| saw_loading = True | |
| if state == "ready" and eng.current_model == "org/model-B": | |
| break | |
| time.sleep(0.1) | |
| assert eng.is_ready, eng.status_dict() | |
| assert saw_loading, "očekával jsem přechodný stav 'starting' během reloadu" | |
| # 5) inference už jede přes model B | |
| r = c.post("/v1/chat/completions", headers=auth, | |
| json={"model": "code-agent-llm", | |
| "messages": [{"role": "user", "content": "pong"}]}) | |
| assert r.json()["choices"][0]["message"]["content"] == "mock:org/model-B:pong" | |
| # instant změna během provozu nezpůsobí reload | |
| rev = app_module.SETTINGS.revision | |
| r = c.post("/admin/settings", json={"max_steps": 33}, headers=auth) | |
| assert r.json()["engine_reload_started"] is False | |
| assert app_module.SETTINGS.revision == rev + 1 | |
| assert eng.current_model == "org/model-B" # engine nedotčen | |
| def test_engine_reload_endpoint(app_module, auth): | |
| c = TestClient(app_module.app) | |
| eng = app_module.ENGINE | |
| c.post("/admin/settings", json={"model": "org/model-C"}, headers=auth) | |
| assert _wait_ready(eng) | |
| r = c.post("/admin/engine/reload", headers=auth) | |
| assert r.json()["status"] == "reload_started" | |
| assert _wait_ready(eng) | |
| assert eng.current_model == "org/model-C" | |
| r = c.post("/admin/engine/stop", headers=auth) | |
| assert r.json()["status"] == "stopped" | |
| assert eng.status_dict()["state"] == "stopped" | |