sorrow404null
acbf73f2c3
Offscreen harnesses now resize+show before load, use per-instance webengine profile dirs and explicit exit codes; compaction and bash stream suites follow the renamed internals.
337 lines
14 KiB
Python
337 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""P1-01 渲染窗口 400 条消息规模诊断（offscreen；临时 DB + 临时配置；不启动真实 LLM）
|
||
|
||
运行: QT_QPA_PLATFORM=offscreen python tests/diag_render_scale.py [N]
|
||
N 默认 400（位置参数可覆盖）
|
||
|
||
验证项（manual 模式保证确定性；末尾追加 auto 模式抽查）：
|
||
1. 初始窗口：.message-wrapper == min(N, size) 且 ≤ size（默认 40）；
|
||
窗口 = 最新 size 条；load-older 可见、load-newer 隐藏；
|
||
2. 向上分页至头部：每页锚点误差 ≤ 2px（绝对顶部例外：scrollTop==0 且露出新页）；
|
||
全程窗口 ≤ size；到达头部后 hasMoreOlder=false；
|
||
全部 N 条消息可达（各步渲染 id 并集 == 链 id 集），无重复 id；
|
||
3. 自头部向下回翻 2 页：锚点稳定（≤2px）；
|
||
4. 规模报告：DOM 节点总数、页面高度、每页耗时采样（ms）；
|
||
5. auto 模式抽查：切 auto + 滚到顶部 → 自动补页发生。
|
||
|
||
调试铁律：本脚本自带总超时（QTimer 300s），bash 侧以 timeout=360 运行。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
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"
|
||
|
||
# 铁律：临时 DB + 临时配置（含 render_window_mode/size），先于任何 UI import
|
||
from tests._test_env import isolate, default_config # noqa: E402
|
||
|
||
N = int(sys.argv[1]) if len(sys.argv) > 1 else 400
|
||
_SIZE = 40
|
||
_PAGE = _SIZE // 2 # 页大小 = 半窗（与 main_window._rw_page_size 一致）
|
||
_cfg = dict(default_config())
|
||
_cfg["render_window_mode"] = "manual" # 确定性诊断；auto 行为末尾单独抽查
|
||
_cfg["render_window_size"] = _SIZE
|
||
_env = isolate("render_scale", config=_cfg)
|
||
|
||
import core.db_manager as _dbm # noqa: E402
|
||
from core.db_manager import DBManager # noqa: E402
|
||
|
||
# ---------- 造数：N 条链（含附件用户消息 / assistant 时间线 / 一个分支兄弟） ----------
|
||
db = DBManager()
|
||
sess = db.create_session(title=f"P1-01 scale {N}")
|
||
sid = sess["id"]
|
||
parent = None
|
||
chain_ids = []
|
||
for i in range(N):
|
||
role = "user" if i % 2 == 0 else "assistant"
|
||
content = f"消息 {i} —— " + "这是一段用于撑起 DOM 高度的填充段落。" * 6
|
||
atts = None
|
||
if role == "user" and i % 50 == 0:
|
||
atts = json.dumps({"user_text": content, "attachments": [
|
||
{"type": "pdf", "mode": "text", "name": f"spec_{i}.pdf",
|
||
"size_kb": 12, "pages": 3, "lines": 40, "content": "文本附件内容"}
|
||
]}, ensure_ascii=False)
|
||
timeline = None
|
||
if role == "assistant" and i % 75 == 3:
|
||
timeline = json.dumps([
|
||
{"t": "think", "text": f"思考片段 {i}"},
|
||
{"t": "text", "text": f"时间线正文 {i}"},
|
||
], ensure_ascii=False)
|
||
msg_id = f"scale-{i}"
|
||
db.add_message(session_id=sid, role=role, content=content, parent_id=parent,
|
||
msg_id=msg_id, attachment_metadata=atts, timeline=timeline)
|
||
parent = msg_id
|
||
chain_ids.append(msg_id)
|
||
# 分支兄弟：必须插在链中间（add_message 会把新消息设为叶子，
|
||
# 若放在主链之后会抢走叶子、截断可见链）
|
||
if i == 298 and N > 300:
|
||
db.add_message(session_id=sid, role="assistant", content="分支兄弟回复",
|
||
parent_id=parent, msg_id="scale-branch-sib")
|
||
db.mark_session_has_messages(sid)
|
||
|
||
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 = []
|
||
|
||
|
||
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
|
||
js = {"done": False}
|
||
|
||
|
||
def _run_js(code, timeout_s=15):
|
||
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"):
|
||
window.browser.execute_js_async(code, on_ret)
|
||
else:
|
||
window.browser.page().runJavaScript(code, on_ret)
|
||
t0 = time.time()
|
||
while not result["done"] and time.time() - t0 < timeout_s:
|
||
app.processEvents()
|
||
time.sleep(0.02)
|
||
assert result["done"], f"JS 执行超时: {code[:60]}"
|
||
return result["val"]
|
||
|
||
|
||
def wait_until(cond_js, timeout_s=30, desc=""):
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout_s:
|
||
if _run_js(cond_js, timeout_s=5):
|
||
return True
|
||
time.sleep(0.1)
|
||
raise AssertionError(f"等待超时: {desc}")
|
||
|
||
|
||
page_metrics = {"pages": [], "heights": [], "dom_nodes": []}
|
||
|
||
|
||
def try_load():
|
||
global window
|
||
window = MainWindow()
|
||
# offscreen 零视口（innerHeight=0）→ 给真实尺寸，滚动/锚点几何才有效
|
||
window.resize(1400, 950)
|
||
window.show()
|
||
for _ in range(10):
|
||
app.processEvents()
|
||
time.sleep(0.05)
|
||
# 等真实视口
|
||
wait_until("window.innerHeight > 0 && window.innerWidth > 0", 30, "视口尺寸")
|
||
# 等 JS 就绪 + 首轮（可能零视口）渲染完成
|
||
wait_until("window.jsReady === true", 30, "jsReady")
|
||
wait_until(f"document.querySelectorAll('#chat-container .message-wrapper').length >= {_SIZE}",
|
||
60, "首轮渲染")
|
||
# 真实视口下走完整 load_messages_to_web 路径重新窗口化加载
|
||
window.load_messages_to_web(window.current_session_id, show_loading=False)
|
||
wait_until(f"rwState !== null && rwState.generation >= 2 && "
|
||
f"document.querySelectorAll('#chat-container .message-wrapper').length >= {_SIZE}",
|
||
60, "视口重载")
|
||
wait_until("(window.__rwPageRendering === false && document.getElementById('load-older') !== null) ? 1 : 0",
|
||
10, "初始窗口收尾")
|
||
run_checks()
|
||
|
||
|
||
def run_checks():
|
||
sel = "#chat-container .message-wrapper"
|
||
|
||
# ---- 1) 初始窗口 ----
|
||
def initial():
|
||
cnt = _run_js(f"document.querySelectorAll('{sel}').length")
|
||
assert cnt == _SIZE, f"初始窗口 {cnt} != {_SIZE}"
|
||
st = _run_js("JSON.stringify({len: rwState.order.length, hidO: rwState.hiddenOlder, "
|
||
"hidN: rwState.hiddenNewer, older: rwState.hasMoreOlder, newer: rwState.hasMoreNewer, "
|
||
"mode: rwState.mode, size: rwState.size})")
|
||
assert st["mode"] == "manual" and st["size"] == _SIZE, st
|
||
assert st["len"] == _SIZE and st["hidO"] == N - _SIZE, st
|
||
assert st["newer"] is False and st["older"] is True, st
|
||
assert _run_js("document.getElementById('load-older').hidden") is False
|
||
assert _run_js("document.getElementById('load-newer').hidden") is True
|
||
# 窗口 = 最新 size 条
|
||
first = _run_js(f"document.querySelector('{sel}').id")
|
||
assert first == chain_ids[N - _SIZE], first
|
||
# 初始贴底
|
||
assert _run_js("isNearBottom()"), "初始窗口未对齐底部"
|
||
|
||
check(f"初始窗口 = 最新 {_SIZE} 条（链长 {N}）", initial)
|
||
|
||
# ---- 2a) 中部锚点保持（真实"窗口中部换页"路径，误差 ≤2px） ----
|
||
seen = set(chain_ids[-_SIZE:])
|
||
anchor_errors = []
|
||
|
||
def page_older_once():
|
||
"""执行一次向上换页（调用方保证视口位置）；断言窗口上限与锚点。"""
|
||
before = _run_js("JSON.stringify(rwCaptureAnchor() || {})")
|
||
before = json.loads(before) if isinstance(before, str) else (before or {})
|
||
scroll_before = _run_js("window.scrollY")
|
||
t0 = time.time()
|
||
_run_js("rwRequestPage('older')")
|
||
# 等待 pending 消费 + 批次渲染收尾
|
||
wait_until("rwState.pending === null && window.__rwPageRendering === false",
|
||
30, "换页完成")
|
||
dt = (time.time() - t0) * 1000
|
||
page_metrics["pages"].append(dt)
|
||
cnt = _run_js(f"document.querySelectorAll('{sel}').length")
|
||
assert cnt <= _SIZE, f"窗口超限 {cnt}"
|
||
if before.get("msgId"):
|
||
el = _run_js(f"document.getElementById({json.dumps(before['msgId'])}) ? "
|
||
f"document.getElementById({json.dumps(before['msgId'])}).getBoundingClientRect().top + window.scrollY : null")
|
||
assert el is not None, f"锚点消息被裁剪: {before['msgId']}"
|
||
if scroll_before <= 1:
|
||
# 绝对顶部：停在 0 露出新页（需求定义的例外：不做锚点断言）
|
||
assert _run_js("window.scrollY") <= 1, "绝对顶部未保持"
|
||
anchor_errors.append(0.0)
|
||
else:
|
||
expect = scroll_before + (el - before["docTop"])
|
||
actual = _run_js("window.scrollY")
|
||
err = abs(expect - actual)
|
||
anchor_errors.append(err)
|
||
assert err <= 2.0, f"锚点误差 {err}px (expect={expect} actual={actual})"
|
||
else:
|
||
anchor_errors.append(0.0)
|
||
ids_now = _run_js(f"Array.from(document.querySelectorAll('{sel}')).map(e => e.id)")
|
||
for i in ids_now:
|
||
seen.add(i)
|
||
page_metrics["dom_nodes"].append(_run_js("document.querySelectorAll('body *').length"))
|
||
page_metrics["heights"].append(_run_js("document.body.scrollHeight"))
|
||
|
||
def anchor_middle():
|
||
# 滚到窗口中部（35% 处）：既非绝对顶部也非贴底 → 走完整锚点数学
|
||
_run_js("window.scrollTo(0, Math.max(1, document.body.scrollHeight * 0.15)); 1")
|
||
time.sleep(0.2)
|
||
app.processEvents()
|
||
scroll_before = _run_js("window.scrollY")
|
||
assert scroll_before > 10, f"未到中部: {scroll_before}"
|
||
assert not _run_js("isNearBottom()"), "中部位置不应贴底（防振荡前置条件）"
|
||
page_older_once()
|
||
st = _run_js("JSON.stringify({hidO: rwState.hiddenOlder, hidN: rwState.hiddenNewer})")
|
||
assert st["hidN"] > 0, f"向上换页后应裁出较新隐藏区（无振荡）: {st}"
|
||
assert st["hidO"] == N - _SIZE - _PAGE, f"窗口应上移 _PAGE 条: {st}"
|
||
|
||
check("中部锚点保持（误差≤2px，无 newer 振荡）", anchor_middle)
|
||
|
||
# ---- 2b) 自顶部循环向上分页至头部（真实"load older"点击路径） ----
|
||
def page_up_to_head():
|
||
_run_js("window.scrollTo(0, 0); 1")
|
||
time.sleep(0.2)
|
||
app.processEvents()
|
||
steps = 0
|
||
while True:
|
||
if not _run_js("rwState.hasMoreOlder"):
|
||
break
|
||
page_older_once()
|
||
steps += 1
|
||
assert steps <= N // _SIZE + 10, "换页次数异常（疑似死循环）"
|
||
first = _run_js(f"document.querySelector('{sel}').id")
|
||
assert first == chain_ids[0], f"未到达头部: {first}"
|
||
|
||
check("自顶部向上分页至头部（绝对顶部例外）", page_up_to_head)
|
||
|
||
def head_state():
|
||
st = _run_js("JSON.stringify({len: rwState.order.length, hidO: rwState.hiddenOlder, "
|
||
"hidN: rwState.hiddenNewer, older: rwState.hasMoreOlder, newer: rwState.hasMoreNewer})")
|
||
assert st["hidO"] == 0 and st["older"] is False, st
|
||
assert st["newer"] is True and st["hidN"] == N - _SIZE, st
|
||
# 全链可达 + 无重复
|
||
all_ids = _run_js(f"Array.from(document.querySelectorAll('{sel}')).map(e => e.id)")
|
||
assert len(all_ids) == len(set(all_ids)), "DOM 出现重复消息 id"
|
||
assert len(seen) == N, f"可达 {len(seen)}/{N} 条"
|
||
|
||
check("头部状态 + 全链可达无重复", head_state)
|
||
|
||
# ---- 3) 自顶部向下回翻 2 页（窗口下移，视口留在顶部） ----
|
||
def page_down_two():
|
||
for k in range(2):
|
||
_run_js("window.scrollTo(0, 0); 1")
|
||
time.sleep(0.15)
|
||
app.processEvents()
|
||
_run_js("rwRequestPage('newer')")
|
||
wait_until("rwState.pending === null && window.__rwPageRendering === false",
|
||
30, f"向下换页{k}完成")
|
||
# 顶部回翻：锚点（最旧一条）随 trimHead 移除 → 视口留在顶部露出新页
|
||
assert _run_js("window.scrollY") <= 1, "顶部回翻应留在顶部"
|
||
first = _run_js(f"document.querySelector('{sel}').id")
|
||
assert first == chain_ids[_PAGE * (k + 1)], f"窗口未下移: {first}"
|
||
st = _run_js("rwState.hasMoreOlder")
|
||
assert st is True, "回翻后应仍有更旧消息"
|
||
|
||
check("自顶部向下回翻 2 页（窗口下移）", page_down_two)
|
||
|
||
# ---- 5) auto 模式抽查：顶部自动补页 ----
|
||
def auto_probe():
|
||
_run_js("window.scrollTo(0, 0); 1")
|
||
time.sleep(0.15)
|
||
app.processEvents()
|
||
_run_js("rwState.mode = 'auto'; 1")
|
||
hid_before = _run_js("rwState.hiddenOlder")
|
||
_run_js("rwAutoCheck(); 1") # 直接触发自动补页判定（事件接线另行覆盖）
|
||
# 给自动补页 12s（每页渲染约 0.5-2s）
|
||
t0 = time.time()
|
||
moved = False
|
||
while time.time() - t0 < 12:
|
||
app.processEvents()
|
||
time.sleep(0.2)
|
||
if _run_js("rwState.hiddenOlder") < hid_before:
|
||
moved = True
|
||
break
|
||
assert moved, "auto 模式顶部未自动补页"
|
||
|
||
check("auto 模式：顶部自动补页", auto_probe)
|
||
|
||
# ---- 4) 规模报告 ----
|
||
nodes = page_metrics["dom_nodes"]
|
||
heights = page_metrics["heights"]
|
||
pages_ms = page_metrics["pages"]
|
||
print("\n================ 规模报告 ================")
|
||
print(f"链长 N={N} | 窗口 size={_SIZE}")
|
||
print(f"换页数={len(pages_ms)} | 每页耗时 min/avg/max = "
|
||
f"{min(pages_ms):.0f}/{(sum(pages_ms) / len(pages_ms)):.0f}/{max(pages_ms):.0f} ms")
|
||
print(f"DOM 节点总数 min/max = {min(nodes)}/{max(nodes)}")
|
||
print(f"页面高度 min/max = {min(heights)}/{max(heights)} px")
|
||
if anchor_errors:
|
||
print(f"锚点误差 min/max = {min(anchor_errors):.2f}/{max(anchor_errors):.2f} px")
|
||
print("==========================================")
|
||
|
||
|
||
def finish(ok=None):
|
||
app.quit()
|
||
|
||
|
||
QTimer.singleShot(500, try_load)
|
||
QTimer.singleShot(300000, lambda: (print("[超时] 300s 总超时"), finish(False)))
|
||
app.exec()
|
||
try:
|
||
window.close()
|
||
except Exception:
|
||
pass
|
||
print(f"\n===== {'ALL PASS' if all(results) else 'HAS FAILURES'}: {sum(results)}/{len(results)} =====")
|
||
# QtWebEngine 退出段错误规避：汇总已打印，直接退出
|
||
os._exit(0 if all(results) else 1)
|