25 lines
773 B
Python
25 lines
773 B
Python
import threading
|
|
from fastapi.testclient import TestClient
|
|
from genesis.server.app import create_app
|
|
from genesis.server.hub import hub
|
|
from genesis.server.store import SessionStore
|
|
|
|
|
|
def _client(tmp_path):
|
|
return TestClient(create_app(
|
|
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
|
data_root=str(tmp_path / "data"), engine="fake"))
|
|
|
|
|
|
def test_ws_streams_progress(tmp_path):
|
|
c = _client(tmp_path)
|
|
|
|
def trigger():
|
|
hub.emit("ws-s1", {"type": "progress", "step": "gen", "status": "ok", "detail": "生成中"})
|
|
|
|
with c.websocket_connect("/api/sessions/ws-s1/ws") as ws:
|
|
threading.Thread(target=trigger).start()
|
|
data = ws.receive_json()
|
|
assert data["type"] == "progress"
|
|
assert data["step"] == "gen"
|