115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""Phase 7 component test: HTTP endpoints (/api/set-avatar, /api/idle-clip,
|
|
/api/set-video-mode, /api/reload-loras, WebSocket handshake video_mode msg).
|
|
|
|
Uses FastAPI's ``TestClient`` so we don't need a running uvicorn server.
|
|
Stubs the model manager to avoid loading Wan2.2 — we only care that the
|
|
HTTP surface is plumbed correctly.
|
|
|
|
Run:
|
|
docker compose exec voice-chat python -m tests.component.test_07_endpoints
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
|
|
from tests.component._common import get_logger
|
|
|
|
log = get_logger("test_07")
|
|
|
|
|
|
def _stub_video_engine():
|
|
class StubCfg:
|
|
mode = "reflective"
|
|
class StubEngine:
|
|
cfg = StubCfg()
|
|
avatar_path = None
|
|
def __init__(self): self.idle = b"FAKE_MP4"
|
|
def is_ready(self): return bool(self.avatar_path)
|
|
def get_idle_clip(self): return self.idle
|
|
def set_avatar(self, path): self.avatar_path = path
|
|
def load_loras(self, specs): self._last_loras = specs
|
|
return StubEngine()
|
|
|
|
|
|
def run():
|
|
from fastapi.testclient import TestClient
|
|
import server.main as main_mod
|
|
|
|
# Inject a stub engine so we never touch Wan2.2.
|
|
main_mod.model_mgr.video_engine = _stub_video_engine()
|
|
|
|
# Bypass the heavy lifespan (model loading) so TestClient starts fast.
|
|
main_mod.app.router.lifespan_context = None # type: ignore[attr-defined]
|
|
|
|
client = TestClient(main_mod.app)
|
|
|
|
# --- set-avatar ---
|
|
log.info("[case 1] POST /api/set-avatar")
|
|
fake_png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64 # minimal PNG header
|
|
resp = client.post(
|
|
"/api/set-avatar",
|
|
files={"image": ("avatar.png", io.BytesIO(fake_png), "image/png")},
|
|
)
|
|
assert resp.status_code == 200, f"got {resp.status_code}: {resp.text}"
|
|
data = resp.json()
|
|
assert data["status"] == "ok"
|
|
assert data["idle_clip_url"] == "/api/idle-clip"
|
|
log.info(" PASS: %s", data)
|
|
|
|
# --- idle-clip ---
|
|
log.info("[case 2] GET /api/idle-clip")
|
|
resp = client.get("/api/idle-clip")
|
|
assert resp.status_code == 200
|
|
assert resp.content == b"FAKE_MP4"
|
|
assert resp.headers["content-type"] == "video/mp4"
|
|
log.info(" PASS")
|
|
|
|
# --- set-video-mode ---
|
|
log.info("[case 3] POST /api/set-video-mode")
|
|
for mode in ("off", "library", "reflective"):
|
|
resp = client.post("/api/set-video-mode", data={"mode": mode})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["mode"] == mode
|
|
resp = client.post("/api/set-video-mode", data={"mode": "bogus"})
|
|
assert resp.status_code == 400
|
|
log.info(" PASS")
|
|
|
|
# --- reload-loras ---
|
|
log.info("[case 4] POST /api/reload-loras")
|
|
body = {
|
|
"loras": [
|
|
{"path": "/cache/loras/a.safetensors", "weight": 0.8,
|
|
"target": "both", "name": "test-a"},
|
|
{"path": "/cache/loras/b.safetensors", "weight": 0.4,
|
|
"target": "both"},
|
|
]
|
|
}
|
|
resp = client.post("/api/reload-loras", json=body)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["lora_count"] == 2
|
|
log.info(" PASS: %s", data)
|
|
|
|
# --- WebSocket video_mode handshake ---
|
|
log.info("[case 5] WebSocket /ws/chat → video_mode announcement")
|
|
with client.websocket_connect("/ws/chat") as websocket:
|
|
msgs = []
|
|
for _ in range(5):
|
|
try:
|
|
msg = websocket.receive_json()
|
|
msgs.append(msg)
|
|
if msg.get("type") == "video_mode":
|
|
break
|
|
except Exception:
|
|
break
|
|
assert any(m.get("type") == "video_mode" for m in msgs), msgs
|
|
log.info(" PASS")
|
|
|
|
log.info("ALL PASSED")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|