# -*- coding: utf-8 -*- """时间线持久化冒烟测试（真实 DB + 真实 WebEngine 页面）： 1. 流式事件 → stream_state 时间线累积（思考/文本/工具 按序） 2. 入库（timeline 列）→ 切会话重载 → DOM 时间线还原（工具气泡不丢） 3. build_api_context 从时间线重建完整 API 链（assistant+tool_calls+tool） 4. 切回进行中的会话：restoreStreamingTimeline 续流 运行: QT_QPA_PLATFORM=offscreen python tests/smoke_persist.py """ import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") os.environ.setdefault("HAOCODE_RENDER", "software") os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = "--disable-gpu" # 绕过 AMD 核显 context lost from PyQt6.QtWidgets import QApplication # noqa: E402 from PyQt6.QtCore import QTimer # noqa: E402 from ui.views.main_window import MainWindow # noqa: E402 app = QApplication(sys.argv) results = [] window = None test = {"sid": None, "mid": None, "done": False} def check(name, ok, detail=""): print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" [{detail}]" if detail and not ok else "")) results.append(ok) def finish(): if test["done"]: return test["done"] = True # 清理测试会话 try: if test["sid"] and window: window.db.delete_session(test["sid"]) except Exception: pass print(f"\n===== {'ALL PASS' if all(results) else 'HAS FAILURES'}: " f"{sum(results)}/{len(results)} =====") app.quit() # JS: 校验重载后的时间线 DOM JS_VERIFY_RELOAD = r""" (function() { var out = {}; try { var mid = '__MID__'; var wrapper = document.getElementById(mid); if (!wrapper) return JSON.stringify({error: 'wrapper missing'}); var tl = wrapper.querySelector('.reply-content'); out.blocks = Array.prototype.map.call(tl.children, function(el) { return el.className.split(' ')[0]; }); var chip = wrapper.querySelector('.tool-chip'); var st = chip ? chip.querySelector('.tool-chip-status') : null; out.chipStatus = st ? st.textContent : null; out.chipOk = st ? st.classList.contains('ok') : false; out.chipCallId = chip ? chip.getAttribute('data-call-id') : null; var segs = wrapper.querySelectorAll('.md-segment'); out.lastSegText = segs.length ? segs[segs.length - 1].textContent : ''; out.thinkCount = wrapper.querySelectorAll('.think-block').length; out.thinkChevron = !!wrapper.querySelector('.think-block .chev'); } catch (e) { out.error = String(e); } return JSON.stringify(out); })() """ # JS: 续流测试 JS_RESUME_A = r""" (function() { try { var mid = '__MID__'; createMessage(mid, 'assistant', '', 'Resume'); restoreStreamingTimeline(mid, '__TLJSON__'); // 续流：接着最后一段文本写；思考新开一段 appendToken(mid, '（续流文本）'); appendReasoning(mid, '续思考内容'); return 'resumeA-done'; } catch (e) { return 'resumeA-err:' + String(e); } })() """ JS_RESUME_B = r""" (function() { try { var mid = '__MID__'; finishMessage(mid); var wrapper = document.getElementById(mid); var tl = wrapper.querySelector('.reply-content'); var out = {}; out.blocks = Array.prototype.map.call(tl.children, function(el) { return el.className.split(' ')[0]; }); var segs = wrapper.querySelectorAll('.md-segment'); out.lastSegText = segs.length ? segs[segs.length - 1].textContent : ''; out.thinkCount = wrapper.querySelectorAll('.think-block').length; out.chipCount = wrapper.querySelectorAll('.tool-chip').length; wrapper.parentNode.removeChild(wrapper); return JSON.stringify(out); } catch (e) { return 'resumeB-err:' + String(e); } })() """ def run_checks(): db = window.db # ---------- 建测试会话 + 模拟流式事件累积 ---------- sess = db.create_session("持久化测试会话") sid = sess["id"] test["sid"] = sid mid = "msg-persist-test" test["mid"] = mid user_row = db.add_message(sid, "user", "测试问题", None) user_id = user_row["id"] window.current_session_id = sid window.chat_bridge.create_message(mid, "assistant", "", "PersistTest") st = { "msg_id": mid, "content": "", "reasoning": "", "timeline": [], "tl_kind": None, "parent_id": user_id, "branch_info": None, "worker": None, "previous_leaf_id": user_id, } window._active_streams[sid] = st # 事件序列：思考 → 文本 → 工具 → 思考 → 文本 window.on_reasoning_received(sid, "先看一下") window.on_reasoning_received(sid, "目录结构。") window.on_chunk_received(sid, "我来执行") window._on_tool_started(sid, "call-p1", "bash", '{"command": "echo hi"}') window._on_tool_updated(sid, "call-p1", "hi\n") window._on_tool_finished(sid, "call-p1", "bash", True, "$ echo hi\nhi\n[exit 0]") window.on_reasoning_received(sid, "输出正常。") window.on_chunk_received(sid, "任务完成。") tl = st["timeline"] check("时间线累积: 5 段", len(tl) == 5, str(tl)) check("时间线类型序 think/text/tool/think/text", [e["t"] for e in tl] == ["think", "text", "tool", "think", "text"], str([e["t"] for e in tl])) tool_e = tl[2] if len(tl) > 2 else {} check("工具条目定格 ok+result", tool_e.get("ok") is True and "[exit 0]" in tool_e.get("result", ""), str(tool_e)) check("聚合 content/reasoning 正确", st["content"] == "我来执行任务完成。" and st["reasoning"] == "先看一下目录结构。输出正常。", f"{st['content']!r} / {st['reasoning']!r}") tl_json = json.dumps(tl, ensure_ascii=False) # ---------- 入库 ---------- row = db.add_message(sid, "assistant", st["content"], user_id, reasoning=st["reasoning"], msg_id=mid, timeline=tl_json) check("DB timeline 列写入", row.get("timeline") == tl_json) # ---------- 切走再切回（重载） ---------- del window._active_streams[sid] # 模拟流已结束（先于重载，避免 live-restore 重复渲染） window.load_messages_to_web(sid, show_loading=False) def verify_reload(res): data = json.loads(str(res)) if "error" in data: check("重载 DOM 校验", False, data["error"]) finish() return expected = ["think-block", "md-segment", "tool-chip", "think-block", "md-segment"] check("重载: 时间线块序还原", data["blocks"] == expected, str(data["blocks"])) check("重载: 工具气泡保留且成功", data["chipOk"] and data["chipStatus"] == "✓ 完成" and data["chipCallId"] == "call-p1", f"{data['chipStatus']} {data['chipCallId']}") check("重载: 思考块 x2 + 末段文本", data["thinkCount"] == 2 and "任务完成。" in data["lastSegText"], f"think={data['thinkCount']} seg={data['lastSegText']!r}") check("重载: SVG 箭头", data["thinkChevron"]) # ---------- build_api_context 重建 ---------- api = window.build_api_context(sid) roles = [m["role"] for m in api] check("API 链: user→assistant→tool→assistant", roles == ["user", "assistant", "tool", "assistant"], str(roles)) asst1 = api[1] if len(api) > 1 else {} check("API: 首条 assistant 带 tool_calls", asst1.get("content") == "我来执行" and len(asst1.get("tool_calls", [])) == 1 and asst1["tool_calls"][0]["function"]["name"] == "bash", str(asst1)[:200]) check("API: tool 消息 tool_call_id 对齐", len(api) > 2 and api[2].get("tool_call_id") == "call-p1" and "[exit 0]" in api[2].get("content", ""), str(api[2] if len(api) > 2 else {})[:200]) check("API: 末条 assistant = 最终回答", len(api) > 3 and api[3].get("content") == "任务完成。") check("API: 无 reasoning 字段", all("reasoning" not in m for m in api)) # ---------- 切回进行中会话: 续流 ---------- tljs = json.dumps(tl_json) js_a = (JS_RESUME_A .replace("'__MID__'", "'msg-resume-test'") .replace("restoreStreamingTimeline(mid, '__TLJSON__')", f"restoreStreamingTimeline(mid, {tljs})")) js_b = JS_RESUME_B.replace("'__MID__'", "'msg-resume-test'") def verify_resume_data(d2): if "error" in d2: check("续流 DOM 校验", False, d2["error"]) finish() return expected2 = ["think-block", "md-segment", "tool-chip", "think-block", "md-segment", "think-block"] check("续流: 块序正确", d2["blocks"] == expected2, str(d2["blocks"])) check("续流: 文本并入末段", "任务完成。（续流文本）" in d2["lastSegText"], d2["lastSegText"]) check("续流: 思考 x3 + 工具 x1", d2["thinkCount"] == 3 and d2["chipCount"] == 1, f"think={d2['thinkCount']} chip={d2['chipCount']}") finish() def run_b(): def got_b(v): vstr = "" if v is None else str(v) print(f" [resume B] {vstr[:120]}") if vstr.startswith("resumeB-err"): check("续流 finishMessage", False, vstr) finish() return try: d2 = json.loads(vstr) except Exception: check("续流 DOM 校验", False, repr(vstr)[:120]) finish() return verify_resume_data(d2) window.browser.page().runJavaScript(js_b, got_b) def run_a(): def got_a(v): vstr = "" if v is None else str(v) print(f" [resume A] {vstr[:120]}") if vstr.startswith("resumeA-err"): check("续流 restore", False, vstr) finish() return QTimer.singleShot(300, run_b) window.browser.page().runJavaScript(js_a, got_a) QTimer.singleShot(300, run_a) window.browser.page().runJavaScript( JS_VERIFY_RELOAD.replace("'__MID__'", f"'{mid}'"), verify_reload) def main(): global window try: window = MainWindow() except Exception: import traceback traceback.print_exc() app.quit() return QTimer.singleShot(6000, run_checks) QTimer.singleShot(45000, lambda: (check("超时", False, "45s 未完成"), finish()) if not test["done"] else None) app.exec() if __name__ == "__main__": main()