From 6cb92e379884ef7aed2b000d62932f2fdcf9b43e Mon Sep 17 00:00:00 2001
From: lhl <liuhaili1203@126.com>
Date: Sat, 29 Aug 2026 12:15:02 +0800
Subject: [PATCH] =?UTF-8?q?test(chat):=20WebSocket=20=E8=BF=9B=E5=BA=A6?=
 =?UTF-8?q?=E6=B5=81=E7=AB=AF=E5=88=B0=E7=AB=AF=E5=86=92=E7=83=9F=20+=20de?=
 =?UTF-8?q?sign.md=20=E8=AE=B0=E5=BD=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/design.md             |   4 ++
 tests/test_progress_e2e.py | 100 +++++++++++++++++++++++++++++++++++++
 2 files changed, 104 insertions(+)
 create mode 100644 tests/test_progress_e2e.py

diff --git a/docs/design.md b/docs/design.md
index b32e33e..e32e6d0 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -1917,3 +1917,7 @@ Document（注入后 Word 文档）
 - `ImpactAgent._cross_ref_design_docs`：以既有系统解析出的标识符（类/方法/模块名，小写键）为锚，在 `design_docs` 的 `markdown_content` 中做**大小写不敏感**子串检索；命中则记录原始大小写 token 与前后文片段。**无 LLM 参与**，纯字符串匹配。
 - 序列化：`impact_report_to_dict` 输出包含 `design_references`；影响调查书下载 JSON 同步包含。
 - 说明：设计文档作为 Type A 辅助证据，**不进入写入规则**，不引入额外 LLM 调用，保持影响调查零幻觉目标。
+
+### 12.8 WebSocket 实时进度流（2026-08-29，分支 feat/websocket-progress）
+
+新增 `ProgressHub` 进程内发布/订阅单例 + `/api/sessions/{sid}/ws` 端点 + `chat_ws.js` 前端实时渲染；进度/错误事件实时推送，既有 `role='progress'/'error'` 持久化兜底保留（重载仍可见）。**单进程假设**：hub 为进程内单例，多 worker 部署下跨进程不互通（后续可迭代 Redis 总线）。
diff --git a/tests/test_progress_e2e.py b/tests/test_progress_e2e.py
new file mode 100644
index 0000000..068e301
--- /dev/null
+++ b/tests/test_progress_e2e.py
@@ -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