298 lines
11 KiB
Python
298 lines
11 KiB
Python
# -*- 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()
|