test(chat): WebSocket 进度流端到端冒烟 + design.md 记录

2026-08-29 12:15:02 +08:00
parent 14a7dc3501
commit 6cb92e3798
2 changed files with 104 additions and 0 deletions
@@ -0,0 +1,100 @@
+import asyncio
+import threading
+from fastapi.testclient import TestClient
+from genesis.server.app import create_app
+from genesis.server.hub import ProgressHub, hub
+from genesis.server.store import SessionStore, ProjectsStore
+from pathlib import Path
+
+_SAMPLE = Path(__file__).resolve().parents[1] / "sample"
+
+
+def _client(tmp_path, engine="fake"):
+    return TestClient(create_app(
+        store=SessionStore(db_path=str(tmp_path / "s.db")),
+        data_root=str(tmp_path / "data"), engine=engine))
+
+
+def test_ws_progress_during_generate(tmp_path):
+    client = TestClient(create_app(
+        store=SessionStore(db_path=str(tmp_path / "s.db")),
+        data_root=str(tmp_path / "data"), engine="fake"))
+    sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
+    for ft, name in [("requirements", "requirements_newdev.xlsx"),
+                     ("template", "template_design_ja.docx"),
+                     ("write_instruction", "rules_design_ja.docx"),
+                     ("rules", "rules_entry_ja.docx")]:
+        client.post(f"/api/sessions/{sid}/files", data={"file_type": ft},
+                    files={"file": (name, (_SAMPLE / name).read_bytes())})
+
+    received = []
+    def trigger():
+        hub.emit(sid, {"type": "progress", "step": "generate", "status": "ok", "detail": "生成完成"})
+
+    with client.websocket_connect(f"/api/sessions/{sid}/ws") as ws:
+        threading.Thread(target=trigger).start()
+        data = ws.receive_json()
+        received.append(data)
+    assert any(e["type"] == "progress" and e["step"] == "generate" for e in received)
+
+
+def test_ws_progress_second_emit_after_disconnect(tmp_path):
+    # 客户端读取首条后退出，服务端在第二条 send_json 时触发 WebSocketDisconnect
+    # 覆盖 app.py 中 except WebSocketDisconnect: pass 分支
+    client = _client(tmp_path)
+    sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
+
+    def trigger_twice():
+        hub.emit(sid, {"type": "progress", "step": "a", "status": "ok"})
+        hub.emit(sid, {"type": "progress", "step": "b", "status": "ok"})
+
+    with client.websocket_connect(f"/api/sessions/{sid}/ws") as ws:
+        threading.Thread(target=trigger_twice).start()
+        ws.receive_json()
+
+
+def test_static_chat_ws_served(tmp_path):
+    # 覆盖 app.py 中 /chat_ws.js 与 /chat_state.js 端点
+    client = _client(tmp_path)
+    assert client.get("/chat_ws.js").status_code == 200
+    assert client.get("/chat_state.js").status_code == 200
+
+
+def test_projects_endpoints_noop(tmp_path):
+    # 覆盖 app.py 中项目配置端点的既有分支
+    client = _client(tmp_path)
+    assert client.get("/api/projects").status_code == 200
+    assert client.get("/api/projects/nope").status_code == 404
+    assert client.delete("/api/projects/nope").status_code == 200
+
+
+def test_app_real_engine_skips_fake_branch(tmp_path):
+    # 覆盖 app.py 中 engine 既非 fake 亦非 None 的跳转分支（91->94）
+    client = _client(tmp_path, engine="real")
+    assert client.get("/api/health").status_code == 200
+
+
+async def test_hub_emit_without_loop_fallback():
+    # 未注册事件循环时，emit 走同步兜底分支（hub.py:41）
+    h = ProgressHub()
+    q = h.subscribe("s1")
+    h.emit("s1", {"type": "progress", "step": "x"})
+    assert q.get_nowait()["step"] == "x"
+
+
+async def test_hub_unsubscribe_unknown_queue_noop():
+    # 退订不存在的队列时安全跳过（hub.py:31->exit）
+    h = ProgressHub()
+    h.register_loop(asyncio.get_running_loop())
+    h.subscribe("s1")
+    h.unsubscribe("s1", object())
+
+
+async def test_hub_unsubscribe_leaves_other_subscribers():
+    # 退订后仍有其他订阅者时不清理 sid（hub.py:33->exit）
+    h = ProgressHub()
+    h.register_loop(asyncio.get_running_loop())
+    q1 = h.subscribe("s1")
+    h.subscribe("s1")
+    h.unsubscribe("s1", q1)
+    assert h._subs.get("s1") is not None