lhl
80daadcd31
- 将 /api/sessions/{sid}/ws 端点移入 create_app（此前置于模块级导致整模块 import NameError，回归被验证拦截）
- register_loop + subscribe 调整至 accept 之前，缩小连接已开但未订阅期间的进度丢失窗口
- 新增 tests/test_verify_ws_real_flow.py：驱动真实 HTTP 聊天流程断言 WS 收到 agent 实际发射的 parse/impact 进度
- 同步 WebSocket 计划文档 Task 3 代码片段（标注端点必须位于 create_app 内）
- 全量 pytest 实测 583 passed / 99.03% 达标
103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
"""真实链路验证：确认 agent 实际发射的进度经 hub 送达 /ws。
|
||
|
||
不走手动 hub.emit，而是驱动真实 HTTP 流程（创建会话→上传真实样本→发送「生成概要设计书」），
|
||
断言 WS 收到 agent 在 _parse_and_confirm 中真正发射的进度事件。
|
||
"""
|
||
import io
|
||
import os
|
||
import threading
|
||
import time
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from genesis.server.app import create_app
|
||
from genesis.server.service import SessionStore
|
||
|
||
SAMPLE_DIR = Path(__file__).resolve().parent.parent / "sample"
|
||
|
||
|
||
def _zip_dir_bytes(src: Path) -> bytes:
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for root, _dirs, files in os.walk(src):
|
||
for f in files:
|
||
p = Path(root) / f
|
||
zf.write(p, p.relative_to(src))
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_ws_receives_agent_emitted_progress(tmp_path):
|
||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||
client = TestClient(create_app(
|
||
store=store, data_root=str(tmp_path / "data"), engine="fake"))
|
||
|
||
# 1) 创建会话（真实流程：sid 由服务端生成）
|
||
created = client.post("/api/sessions", json={"user_id": "default", "project": "测试项目"})
|
||
assert created.status_code == 200, created.text
|
||
sid = created.json()["session_id"]
|
||
|
||
# 2) 上传真实样本（端点 /api/sessions/{sid}/files，file_type 走表单字段）
|
||
samples = [
|
||
("requirements", "requirements.xlsx",
|
||
(SAMPLE_DIR / "requirements_newdev.xlsx").read_bytes(),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"),
|
||
("template", "template.docx",
|
||
(SAMPLE_DIR / "template_design_zh.docx").read_bytes(),
|
||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document"),
|
||
("existing_system", "existing_system.zip",
|
||
_zip_dir_bytes(SAMPLE_DIR / "existing-system"),
|
||
"application/zip"),
|
||
]
|
||
for ftype, fname, content, mime in samples:
|
||
r = client.post(
|
||
f"/api/sessions/{sid}/files",
|
||
data={"file_type": ftype},
|
||
files={"file": (fname, content, mime)},
|
||
)
|
||
assert r.status_code == 200, f"上传 {ftype} 失败: {r.text}"
|
||
|
||
# 3) 打开 WS 订阅该会话进度，再驱动聊天（POST 在子线程同步执行，期间 agent 会真实发射进度）
|
||
received: list = []
|
||
|
||
def trigger():
|
||
resp = client.post(
|
||
f"/api/chat/{sid}/messages",
|
||
json={"content": "请生成概要设计书"},
|
||
)
|
||
received.append(("post_status", resp.status_code))
|
||
|
||
def receiver(ws):
|
||
# 阻塞读取，直到连接关闭（连接关闭由主线程在 POST 完成后触发）
|
||
while True:
|
||
try:
|
||
received.append(ws.receive_json())
|
||
except Exception:
|
||
return
|
||
|
||
with client.websocket_connect(f"/api/sessions/{sid}/ws") as ws:
|
||
time.sleep(0.1) # 确保服务端已完成订阅（register_loop + subscribe 在 accept 之前）
|
||
t = threading.Thread(target=trigger)
|
||
t.start()
|
||
rth = threading.Thread(target=receiver, args=(ws,))
|
||
rth.start()
|
||
t.join(timeout=30) # 等待聊天请求（含解析/影响调查）完成
|
||
time.sleep(1.0) # 收尾缓冲：让服务端已发射的事件送达
|
||
try:
|
||
ws.close()
|
||
except Exception:
|
||
pass
|
||
rth.join(timeout=5)
|
||
|
||
post_status = None
|
||
for item in received:
|
||
if isinstance(item, tuple) and item[0] == "post_status":
|
||
post_status = item[1]
|
||
progress_events = [e for e in received if isinstance(e, dict) and e.get("type") == "progress"]
|
||
assert progress_events, (
|
||
f"WS 未收到任何 agent 实际发射的进度事件；post_status={post_status}; 收到={received}"
|
||
)
|
||
steps = [e.get("step") for e in progress_events]
|
||
assert "parse" in steps, f"至少应收到 parse 进度；实际 steps={steps}"
|