157 lines
7.0 KiB
Python
157 lines
7.0 KiB
Python
"""会话复制 · UI 入口端到端（offscreen）
|
||
链路：弹窗「📋 复制」按钮 → action_triggered("copy") → on_session_action →
|
||
DBManager.copy_session → 侧边栏重建 + 自动切到副本
|
||
|
||
运行: QT_QPA_PLATFORM=offscreen python tests/smoke_copy_session.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import shutil
|
||
|
||
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
|
||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||
|
||
# 铁律：测试不得污染真实 data/chat_history.db → 默认 DB 路径重定向到临时文件
|
||
import core.db_manager as _dbm # noqa: E402
|
||
_DB_TMP = os.path.join(tempfile.gettempdir(), f"haocode_test_copy_ui_{os.getpid()}.db")
|
||
if os.path.exists(_DB_TMP):
|
||
os.remove(_DB_TMP)
|
||
_dbm._DEFAULT_DB = _DB_TMP
|
||
|
||
_FS_TMP = tempfile.mkdtemp(prefix="hocode_copy_ui_files_") # 临时附件根
|
||
|
||
from PyQt6.QtWidgets import QApplication, QPushButton # noqa: E402
|
||
from PyQt6 import QtCore # noqa: E402
|
||
from ui.views.main_window import MainWindow, SessionContextPopup # noqa: E402
|
||
|
||
app = QApplication(sys.argv)
|
||
ok = True
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
global ok
|
||
print((" PASS " if cond else " FAIL ") + name + ("" if cond else f" {extra}"), flush=True)
|
||
if not cond:
|
||
ok = False
|
||
|
||
|
||
def chain_sig(sid):
|
||
return [(m["role"], m["content"]) for m in window.db.get_message_chain(sid)]
|
||
|
||
|
||
window = MainWindow()
|
||
window.db.files_root = _FS_TMP
|
||
|
||
|
||
# ======================================================================
|
||
# 1) 准备一个带内容的源会话（直接走 DB 层，不触发真实 API 请求）
|
||
# ======================================================================
|
||
window.on_new_chat_clicked()
|
||
src = window.current_session_id
|
||
for role, text in [("user", "帮我看看这个文件"), ("assistant", "好的，我先读一下")]:
|
||
window.db.add_message(src, role, text, window.db.get_session_leaf(src))
|
||
window.db.mark_session_has_messages(src)
|
||
window.rebuild_sidebar()
|
||
src_title = [s for s in window.db.get_all_sessions() if s["id"] == src][0]["title"]
|
||
check("源会话已就绪（2 条消息）", len(window.db.get_message_chain(src)) == 3,
|
||
str(len(window.db.get_message_chain(src))))
|
||
|
||
# 图片附件（验证 UI 链路里附件也被深拷贝）
|
||
os.makedirs(os.path.join(_FS_TMP, "data", "attachments"), exist_ok=True)
|
||
png = os.path.join(_FS_TMP, "data", "attachments", "ui_src.png")
|
||
with open(png, "wb") as f:
|
||
f.write(b"\x89PNG\r\n\x1a\nUI-IMAGE-BYTES")
|
||
meta = ('{"user_text":"看图","attachments":[{"type":"image","size_kb":0.1,'
|
||
'"local_path":"data/attachments/ui_src.png"}]}')
|
||
window.db.add_message(src, "user", "以及这张图", window.db.get_session_leaf(src),
|
||
attachment_metadata=meta)
|
||
|
||
# ======================================================================
|
||
# 2) 弹窗里存在「复制」按钮，且点击后发出 ("copy", sid)
|
||
# ======================================================================
|
||
popup = SessionContextPopup(src, False, window)
|
||
btns = popup.findChildren(QPushButton)
|
||
texts = [b.text() for b in btns]
|
||
check("弹窗含「📋 复制」按钮", any("复制" in t for t in texts), str(texts))
|
||
check("按钮顺序 编辑/星标/复制/删除",
|
||
len(texts) == 4 and "编辑" in texts[0] and "星标" in texts[1]
|
||
and "复制" in texts[2] and "删除" in texts[3], str(texts))
|
||
|
||
got = []
|
||
popup.action_triggered.connect(lambda a, s: got.append((a, s)))
|
||
copy_btn = [b for b in btns if "复制" in b.text()][0]
|
||
copy_btn.click()
|
||
check("点击复制按钮发出 copy 动作", got == [("copy", src)], str(got))
|
||
|
||
# ======================================================================
|
||
# 3) on_session_action("copy") → 真实克隆 + 侧边栏刷新 + 切到副本
|
||
# ======================================================================
|
||
n_before = len(window.db.get_all_sessions())
|
||
window.on_session_action("copy", src)
|
||
sessions = window.db.get_all_sessions()
|
||
check("会话数 +1", len(sessions) == n_before + 1, f"{n_before}→{len(sessions)}")
|
||
|
||
rows = [s for s in sessions if s["title"] == src_title + " (副本)"]
|
||
check("副本标题 = 源标题 + ' (副本)'", len(rows) == 1, str([s["title"] for s in sessions]))
|
||
copy_id = rows[0]["id"] if rows else None
|
||
|
||
if copy_id:
|
||
check("副本不是源（ID 不同）", copy_id != src)
|
||
check("消息链完全一致", chain_sig(src) == chain_sig(copy_id))
|
||
check("已自动切到副本", window.current_session_id == copy_id, str(window.current_session_id))
|
||
check("侧边栏含副本项",
|
||
any(window.history_list.item(i).data(QtCore.Qt.ItemDataRole.UserRole) == copy_id
|
||
for i in range(window.history_list.count())))
|
||
check("副本排在列表顶部（sort_order 最小）",
|
||
sessions[0]["id"] == copy_id, sessions[0]["title"])
|
||
check("副本不带星标", rows[0]["is_starred"] == 0)
|
||
check("副本 mode 跟随源（未锁定为 None）",
|
||
rows[0]["mode"] == window.db.get_session_mode(src))
|
||
|
||
# 附件深拷贝：新旧文件同时存在，且副本 metadata 指向新文件
|
||
import json as _json
|
||
c_rows = [m for m in window.db.get_message_chain(copy_id) if m.get("attachment_metadata")]
|
||
c_meta = _json.loads(c_rows[-1]["attachment_metadata"])
|
||
new_rel = c_meta["attachments"][0]["local_path"]
|
||
check("副本附件已改名", new_rel != "data/attachments/ui_src.png", new_rel)
|
||
check("新旧图片文件同时存在",
|
||
os.path.isfile(png) and os.path.isfile(os.path.join(_FS_TMP, new_rel)))
|
||
check("副本图片内容一致",
|
||
open(os.path.join(_FS_TMP, new_rel), "rb").read()
|
||
== open(png, "rb").read())
|
||
|
||
# 删副本后源完好（UI 链路下的删除隔离）
|
||
window.db.delete_session(copy_id)
|
||
check("删副本后源消息链不变", chain_sig(src) == [
|
||
("system", "你是一个优秀的助手！"), ("user", "帮我看看这个文件"),
|
||
("assistant", "好的，我先读一下"), ("user", "以及这张图")])
|
||
check("删副本后源图片仍在", os.path.isfile(png))
|
||
|
||
# ======================================================================
|
||
# 4) 生成中拒绝复制
|
||
# ======================================================================
|
||
n_before2 = len(window.db.get_all_sessions())
|
||
window._active_streams[src] = {"worker": None}
|
||
window.on_session_action("copy", src)
|
||
check("会话生成中 → 拒绝复制（无新会话）",
|
||
len(window.db.get_all_sessions()) == n_before2)
|
||
window._active_streams.pop(src, None)
|
||
|
||
# 5) 源不存在 → 静默失败，不崩
|
||
try:
|
||
window.on_session_action("copy", "sess_not_exist_zzz")
|
||
check("源不存在时不抛异常", True)
|
||
except Exception as e:
|
||
check("源不存在时不抛异常", False, str(e))
|
||
|
||
print("\n===== " + ("ALL PASS" if ok else "HAS FAILURES") + " =====", flush=True)
|
||
|
||
shutil.rmtree(_FS_TMP, ignore_errors=True)
|
||
if os.path.exists(_DB_TMP):
|
||
os.remove(_DB_TMP)
|
||
sys.exit(0 if ok else 1)
|