207 lines
7.5 KiB
Python
207 lines
7.5 KiB
Python
"""offscreen 冒烟测试：主窗口实例化 + 核心链路（不启动真实 LLM）
|
||
运行: QT_QPA_PLATFORM=offscreen python tests/smoke_offscreen.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
|
||
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 # 软渲染，离屏最稳
|
||
|
||
# 铁律：测试不得污染真实 data/chat_history.db → DBManager 默认路径重定向到临时文件
|
||
import tempfile as _tf # noqa: E402
|
||
import core.db_manager as _dbm # noqa: E402
|
||
_dbm._DEFAULT_DB = os.path.join(_tf.gettempdir(), f"haocode_test_smoke_offscreen_{os.getpid()}.db")
|
||
|
||
# P1-03：QtWebEngine 独立 profile 重定向到临时目录（不占项目 data/webengine/）
|
||
os.environ.setdefault(
|
||
"HAOCODE_WEBENGINE_PROFILE_DIR",
|
||
os.path.join(_tf.gettempdir(), f"haocode_test_smoke_offscreen_{os.getpid()}_profile"))
|
||
|
||
import ctypes # noqa: E402
|
||
|
||
from PyQt6.QtWidgets import QApplication # noqa: E402
|
||
from PyQt6.QtCore import QTimer # noqa: E402
|
||
|
||
# QtWebEngine 必须在 QApplication 创建前 import
|
||
from ui.views.main_window import MainWindow # noqa: E402
|
||
|
||
app = QApplication(sys.argv)
|
||
|
||
results = []
|
||
|
||
|
||
def check(name, fn):
|
||
try:
|
||
fn()
|
||
print(f" PASS {name}")
|
||
results.append(True)
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
print(f" FAIL {name}: {e}")
|
||
results.append(False)
|
||
|
||
|
||
window = None
|
||
page_ready = {"ok": False}
|
||
|
||
|
||
def on_page_load_progress(v):
|
||
pass
|
||
|
||
|
||
def on_js_console(level, msg, line, src):
|
||
pass
|
||
|
||
|
||
def try_load():
|
||
global window
|
||
try:
|
||
window = MainWindow()
|
||
page_ready["ok"] = True
|
||
except Exception:
|
||
import traceback
|
||
traceback.print_exc()
|
||
finish(False)
|
||
return
|
||
# MainWindow 构造完成时 HTML/JS 已就绪（loadFinished 可能早于连接）
|
||
QTimer.singleShot(6000, run_checks) # 等 JS 引擎 + 历史渲染完成
|
||
|
||
|
||
def run_checks():
|
||
# 1) 窗口已创建
|
||
check("MainWindow 实例化", lambda: (_ for _ in ()).throw(AssertionError("no window")) if window is None else None)
|
||
|
||
# 2) DB 链路
|
||
def db_chain():
|
||
assert window.db is not None
|
||
sessions = window.db.get_all_sessions()
|
||
assert isinstance(sessions, list)
|
||
|
||
check("DB 会话列表", db_chain)
|
||
|
||
# 3) build_api_context（P0 修复验证：不再有 reasoning 字段）
|
||
def ctx_build():
|
||
sid = window.current_session_id
|
||
if not sid:
|
||
return
|
||
msgs = window.build_api_context(sid)
|
||
assert isinstance(msgs, list)
|
||
for m in msgs:
|
||
assert "reasoning" not in m, f"reasoning 字段仍在 API payload 里: {m.get('role')}"
|
||
|
||
check("build_api_context 无 reasoning 字段", ctx_build)
|
||
|
||
# 4) CJK token 估算
|
||
def est():
|
||
n = window._estimate_token_count([
|
||
{"role": "user", "content": "你好，世界！这是一段中文测试。"},
|
||
{"role": "assistant", "content": "hello world " * 20},
|
||
])
|
||
assert isinstance(n, int) and n > 0
|
||
|
||
check("CJK token 估算", est)
|
||
|
||
# 5) AgentWorker 可构造 + 信号齐全
|
||
def worker():
|
||
from core.llm_engine import AgentWorker, TitleWorker
|
||
w = AgentWorker(window.current_provider, window.current_model,
|
||
[{"role": "user", "content": "x"}])
|
||
for sig in ("chunk_received", "reasoning_received", "error_occurred",
|
||
"tool_execution_started", "tool_execution_updated",
|
||
"tool_execution_finished", "context_compacted"):
|
||
assert hasattr(w, sig), sig
|
||
t = TitleWorker(window.current_provider, window.current_model,
|
||
[{"role": "user", "content": "x"}])
|
||
assert hasattr(t, "chunk_received")
|
||
|
||
check("AgentWorker/TitleWorker 构造", worker)
|
||
|
||
# 6) bridge 工具方法
|
||
def bridge():
|
||
b = window.chat_bridge
|
||
for m in ("tool_execution_started", "tool_execution_updated",
|
||
"tool_execution_finished", "show_note"):
|
||
assert hasattr(b, m), m
|
||
|
||
check("ChatBridge 工具事件方法", bridge)
|
||
|
||
# 7) 离屏渲染 JS 就绪
|
||
check("Web 页面加载完成", lambda: (_ for _ in ()).throw(AssertionError("page not ready")) if not page_ready["ok"] else None)
|
||
|
||
# 8) 🌟 KaTeX 公式渲染（真实页面上下文：资源加载 + [...] 供应商格式 + 行内 $）
|
||
def _run_js(js, timeout_s=10):
|
||
result = {"val": None, "done": False}
|
||
|
||
def on_ret(val):
|
||
if isinstance(val, str):
|
||
try:
|
||
val = json.loads(val)
|
||
except Exception:
|
||
pass
|
||
result["val"] = val
|
||
result["done"] = True
|
||
|
||
if hasattr(window.browser, "execute_js_async"): # WebView2 路径（cb 收 JSON 字符串）
|
||
window.browser.execute_js_async(js, on_ret)
|
||
else: # QtWebEngine 路径（cb 收 Python 对象）
|
||
window.browser.page().runJavaScript(js, on_ret)
|
||
t0 = time.time()
|
||
while not result["done"] and time.time() - t0 < timeout_s:
|
||
app.processEvents()
|
||
time.sleep(0.05)
|
||
assert result["done"], "JS 执行超时"
|
||
return result["val"]
|
||
|
||
def katex_render():
|
||
# 等页面 JS 就绪（WV2 冷启动可能慢；app.js 就绪时置 window.jsReady=true）
|
||
for _ in range(30):
|
||
if _run_js("window.jsReady === true ? 1 : 0", timeout_s=3) == 1:
|
||
break
|
||
time.sleep(0.5)
|
||
else:
|
||
raise AssertionError("页面 JS 未就绪（jsReady）")
|
||
ver = _run_js("typeof katex !== 'undefined' ? katex.version : null")
|
||
assert ver, "katex 全局缺失（资源加载失败？）"
|
||
js = ("(function(){var s = '结果：" + chr(92) + "n[" + chr(92) + "nP_4=" + chr(92)*2 + "operatorname{BRF}(M_4,M_5)" + chr(92) + "n]" + chr(92) + "n" + chr(92) + "n行内 $E=mc^2$ 结束。';"
|
||
"var html = safeHtml(marked.parse(s));return {"
|
||
"has: html.indexOf('katex') !== -1,"
|
||
"disp: (html.match(/katex-display/g) || []).length,"
|
||
'inline: (html.match(/class=\"katex\"/g) || []).length,'
|
||
"leaked: html.indexOf('@@K') !== -1,"
|
||
"vis: html.indexOf('katex-html') !== -1,"
|
||
"tex: html.indexOf('\\operatorname{BRF}') !== -1};})()")
|
||
r = _run_js(js)
|
||
assert isinstance(r, dict), f"意外返回: {r!r}"
|
||
assert r.get("has"), "未产生 KaTeX HTML"
|
||
assert r.get("disp", 0) >= 1, "块公式未渲染为 katex-display"
|
||
assert r.get("inline", 0) >= 1, "行内公式未渲染"
|
||
assert not r.get("leaked"), "占位符泄漏"
|
||
assert r.get("vis"), "缺少 katex-html 可视层（KaTeX 未真正渲染）"
|
||
assert r.get("tex"), "tex 未正确传入（转义错误）"
|
||
print(f" [info] KaTeX {ver} | display={r.get('disp')} inline={r.get('inline')}")
|
||
|
||
check("KaTeX 公式渲染（[...] 供应商格式 + 行内 $）", katex_render)
|
||
|
||
print(f"\n===== {'ALL PASS' if all(results) else 'HAS FAILURES'}: {sum(results)}/{len(results)} =====")
|
||
try:
|
||
window.close()
|
||
except Exception:
|
||
pass
|
||
app.quit()
|
||
|
||
|
||
def finish(ok=None):
|
||
app.quit()
|
||
|
||
|
||
QTimer.singleShot(500, try_load)
|
||
QTimer.singleShot(90000, finish) # 总超时
|
||
app.exec()
|
||
sys.exit(0 if all(results) else 1)
|