# -*- coding: utf-8 -*- """QWebChannel control-console bridge regression tests.""" import json import os import re from pathlib import Path from types import SimpleNamespace from unittest import TestCase, main, mock, skipUnless os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") try: from PySide6.QtCore import QEventLoop, QTimer from PySide6.QtWidgets import QApplication, QCheckBox, QMainWindow except ImportError: QApplication = None else: from wechat_gui_qt import ( CONSOLE_HTML, CONSOLE_VIEWS, ConsoleBridge, ConsoleShell, MainWindow, ) class _FakeTimer: def __init__(self, active=False, remaining=-1): self.active = active self.remaining = remaining self.started = [] self.stop_count = 0 def isActive(self): return self.active def remainingTime(self): return self.remaining def start(self, delay): self.active = True self.remaining = delay self.started.append(delay) def stop(self): self.active = False self.remaining = -1 self.stop_count += 1 @skipUnless(QApplication is not None, "当前测试环境未安装 PySide6") class ConsoleBridgeTest(TestCase): @classmethod def setUpClass(cls): cls.app = QApplication.instance() or QApplication([]) def test_request_state_forces_webchannel_handshake_payload(self): host = QMainWindow() host._push_console_state = mock.Mock() host.console_shell = SimpleNamespace(channel_ready=False) bridge = ConsoleBridge(host) bridge.requestState() self.assertTrue(host.console_shell.channel_ready) host._push_console_state.assert_called_once_with(force=True) bridge.deleteLater() host.deleteLater() def test_console_push_timer_coalesces_against_earliest_deadline(self): timer = _FakeTimer() host = SimpleNamespace( _console_push_timer=timer, _console_push_urgent=False, ) MainWindow._schedule_console_push(host) self.assertEqual(timer.started, [1000]) timer.remaining = 700 MainWindow._schedule_console_push(host) self.assertEqual(timer.started, [1000], "普通更新不应不断延后计时器") MainWindow._schedule_console_push(host, urgent=True) self.assertEqual(timer.started, [1000, 80]) self.assertTrue(host._console_push_urgent) timer.remaining = 45 MainWindow._schedule_console_push(host, urgent=True) self.assertEqual(timer.started, [1000, 80], "紧急更新已有更早截止时间时不重启") def test_state_is_not_cached_before_webchannel_handshake(self): timer = _FakeTimer(active=True, remaining=40) shell = SimpleNamespace( page_ready=True, channel_ready=False, set_state=mock.Mock(return_value=True), ) host = SimpleNamespace( _console_push_timer=timer, _console_push_urgent=True, _console_state_pending=False, _console_last_payload="previous", _console_snapshot=mock.Mock(return_value={"view": "system"}), console_shell=shell, ) MainWindow._push_console_state(host) self.assertEqual(timer.stop_count, 1) self.assertTrue(host._console_state_pending) self.assertEqual(host._console_last_payload, "previous") host._console_snapshot.assert_not_called() shell.set_state.assert_not_called() def test_ready_state_delivery_updates_cache_only_after_emit(self): timer = _FakeTimer() shell = SimpleNamespace( page_ready=True, channel_ready=True, set_state=mock.Mock(return_value=True), ) host = SimpleNamespace( _console_push_timer=timer, _console_push_urgent=False, _console_state_pending=True, _console_last_payload="", _console_snapshot=mock.Mock(return_value={"view": "system"}), console_shell=shell, ) MainWindow._push_console_state(host, force=True) shell.set_state.assert_called_once_with('{"view": "system"}') self.assertEqual(host._console_last_payload, '{"view": "system"}') self.assertFalse(host._console_state_pending) def test_set_view_waits_for_loaded_document_and_replays_once(self): page = SimpleNamespace(runJavaScript=mock.Mock()) shell = SimpleNamespace( page_ready=False, _desired_view="auto", _applied_view=None, _view_command_serial=0, page=lambda: page, ) ConsoleShell.set_view(shell, "system") self.assertEqual(shell._desired_view, "system") page.runJavaScript.assert_not_called() shell.page_ready = True ConsoleShell.set_view(shell, "system") page.runJavaScript.assert_called_once() callback = page.runJavaScript.call_args.args[1] callback(None) self.assertEqual(shell._applied_view, "system") ConsoleShell.set_view(shell, "system") page.runJavaScript.assert_called_once() ConsoleShell.remember_view(shell, "logs") self.assertEqual(shell._desired_view, "logs") self.assertEqual(shell._applied_view, "logs") ConsoleShell.set_view(shell, "logs") page.runJavaScript.assert_called_once() def test_every_html_navigation_schedules_authoritative_state(self): shell = SimpleNamespace(remember_view=mock.Mock()) host = SimpleNamespace( _console_view="auto", console_shell=shell, _schedule_console_push=mock.Mock(), ) for view in CONSOLE_VIEWS: with self.subTest(view=view): shell.remember_view.reset_mock() host._schedule_console_push.reset_mock() MainWindow._console_navigate(host, view) self.assertEqual(host._console_view, view) shell.remember_view.assert_called_once_with(view) host._schedule_console_push.assert_called_once_with(urgent=True) def test_console_html_and_javascript_assets_are_packaged_together(self): html_path = Path(CONSOLE_HTML) script_path = html_path.with_name("console-app.js") self.assertTrue(html_path.is_file(), html_path) self.assertTrue(script_path.is_file(), script_path) html = html_path.read_text(encoding="utf-8") self.assertIn("console-app.js", html) self.assertIn( "html.embedded .rail{-webkit-backdrop-filter:none;backdrop-filter:none}", html, ) self.assertIn( "html.embedded .panel{-webkit-backdrop-filter:none;backdrop-filter:none}", html, ) self.assertIn( "html.embedded .nav-btn,html.embedded .nav-icon{transition:none}", html, ) self.assertIn( "html.embedded .nav-btn:hover .nav-icon{transform:none}", html, ) def test_same_view_state_push_preserves_sidebar_and_main_dom_identity(self): host = QMainWindow() host._push_console_state = mock.Mock() shell = ConsoleShell(host) shell.resize(1630, 920) shell.show() if not shell.page_ready: loaded = [] loop = QEventLoop() def finished(ok): loaded.append(bool(ok)) loop.quit() shell.loadFinished.connect(finished) QTimer.singleShot(10000, loop.quit) loop.exec() shell.loadFinished.disconnect(finished) self.assertEqual(loaded, [True]) def run_javascript(source): values = [] loop = QEventLoop() def completed(value): values.append(value) loop.quit() shell.page().runJavaScript(source, completed) QTimer.singleShot(5000, loop.quit) loop.exec() self.assertEqual(len(values), 1) return values[0] payload = { "view": "auto", "demo": False, "listening": True, "statusLabel": "监听中", "wecomConnected": True, "aiOk": True, "people": [], "task": None, "archives": [], "chat": [], "memory": {}, "settle": [], "journey": [], "chain": [], "activities": [], "logs": [], "exceptions": {"sendFail": 0, "timeout": 0, "windowLost": 0}, "ai": {}, "automation": {}, "system": { "storage": "1 MB", "archiveSize": "1 KB", "logSize": "1 KB", "cacheSize": "1 KB", "noticeTime": "16:33", }, "metrics": { "processing": "1", "waiting": "3", "retry": "1", "avgWait": "12s", "failed": "0", "warnings": "0", }, } for view_name in CONSOLE_VIEWS: with self.subTest(view=view_name): payload["view"] = view_name encoded = json.dumps(payload, ensure_ascii=False) unchanged = run_javascript( f""" window.applyState({encoded}); window.__domIdentity = {{ rail: document.querySelector('.rail'), screen: document.querySelector('#main > .screen'), nav: document.querySelector('[data-view="{view_name}"]') }}; window.applyState({encoded}); Boolean(window.__domIdentity.rail && window.__domIdentity.screen && window.__domIdentity.nav && window.__domIdentity.rail === document.querySelector('.rail') && window.__domIdentity.screen === document.querySelector('#main > .screen') && window.__domIdentity.nav === document.querySelector('[data-view="{view_name}"]') && window.__domIdentity.nav.classList.contains('active')); """ ) self.assertTrue(unchanged) payload["view"] = "queue" encoded = json.dumps(payload, ensure_ascii=False) self.assertTrue(run_javascript( f""" window.applyState({encoded}); window.__domIdentity = {{ rail: document.querySelector('.rail'), screen: document.querySelector('#main > .screen'), nav: document.querySelector('[data-view="queue"]') }}; document.querySelector('[data-live-metric="avgWait"]').textContent === '12s'; """ )) payload["metrics"]["avgWait"] = "13s" payload["system"]["storage"] = "2 MB" payload["archives"] = [{"key": "background-change", "name": "后台归档变化"}] payload["exceptions"]["timeout"] = 1 volatile = json.dumps(payload, ensure_ascii=False) patched = run_javascript( f""" window.applyState({volatile}); window.__domIdentity.rail === document.querySelector('.rail') && window.__domIdentity.screen === document.querySelector('#main > .screen') && window.__domIdentity.nav === document.querySelector('[data-view="queue"]') && document.querySelector('[data-live-metric="avgWait"]').textContent === '13s'; """ ) self.assertTrue(patched) payload["view"] = "system" payload["system"]["storage"] = "3 MB" payload["system"]["logSize"] = "2 MB" system_initial = json.dumps(payload, ensure_ascii=False) self.assertTrue(run_javascript( f""" window.applyState({system_initial}); window.__systemScreen = document.querySelector('#main > .screen'); document.querySelector('[data-live-system-storage]').textContent === '3'; """ )) payload["system"]["storage"] = "4 MB" payload["system"]["logSize"] = "3 MB" payload["system"]["noticeTime"] = "16:34" system_volatile = json.dumps(payload, ensure_ascii=False) self.assertTrue(run_javascript( f""" window.applyState({system_volatile}); window.__systemScreen === document.querySelector('#main > .screen') && document.querySelector('[data-live-system-storage]').textContent === '4' && document.querySelector('[data-live-system-notice]').textContent === '16:34' && [...document.querySelectorAll('.system-top > section:nth-child(2) .stack .row > span:last-child')] .some(node => node.textContent === '3 MB'); """ )) log_payload = json.loads(json.dumps(payload, ensure_ascii=False)) log_payload.update( { "view": "logs", "runtime": "00:00:12", "chain": [ {"icon": "activity", "title": "扫描未读", "time": "0.3s"}, {"icon": "message", "title": "读取消息", "time": "1.2s"}, {"icon": "brain", "title": "请求 AI", "time": "2.4s"}, {"icon": "send", "title": "回填发送", "time": "0.6s"}, ], "exceptions": {"sendFail": 0, "timeout": 0, "windowLost": 0}, "logDetail": {"title": "未读扫描", "text": "检查完成", "tags": ["成功"]}, "logs": [ { "time": "16:33:27", "title": "未读扫描", "detail": "检查完成", "state": "成功", "tone": "green", } ], } ) log_payload["metrics"].update({"avgResp": "2.4s", "warnings": "0"}) log_initial = json.dumps(log_payload, ensure_ascii=False) self.assertTrue(run_javascript( f""" window.applyState({log_initial}); window.__logScreen = document.querySelector('#main > .screen'); Boolean(window.__logScreen && document.querySelector('[data-live-runtime]')); """ )) shell._last_state = log_payload shell.channel_ready = True live_payload = json.loads(json.dumps(log_payload, ensure_ascii=False)) live_payload["runtime"] = "00:00:13" live_payload["metrics"].update({"avgWait": "14s", "avgResp": "3.1s", "warnings": "1"}) live_payload["chain"][2]["time"] = "3.1s" live_payload["exceptions"]["sendFail"] = 2 live_payload["logDetail"] = { "title": "发送失败", "text": "已加入重试队列", "tags": ["警告", "发送"], } live_payload["logs"].append( { "time": "16:33:28", "title": "发送失败", "detail": "已加入重试队列", "state": "警告", "tone": "orange", } ) # Queue/task churn is present in every backend snapshot but is not # structural content of the currently visible logs page. live_payload["task"] = { "key": "task-1", "action": "AI 生成回复中", "stepIndex": 3, "answer": "", "messageCount": "已读取 1 条消息", "confidence": "生成中", "latency": "3.1s", } self.assertTrue(shell.set_state(json.dumps(live_payload, ensure_ascii=False))) loop = QEventLoop() QTimer.singleShot(100, loop.quit) loop.exec() self.assertTrue(run_javascript( """ window.__logScreen === document.querySelector('#main > .screen') && document.querySelector('[data-live-runtime]').textContent === '00:00:13' && document.querySelector('[data-live-metric="avgResp"]').textContent === '3.1s' && document.querySelector('[data-live-metric="warnings"]').textContent === '1' && document.querySelector('.exception-grid .exception b').textContent === '2' && document.querySelector('.log-mid > section:first-child .row b').textContent === '发送失败' && document.querySelector('#log-list').textContent.includes('发送失败') && window.__coreFp === coreFingerprint(window.STATE); """ )) shell.close() shell.deleteLater() host.deleteLater() self.app.processEvents() def test_auto_send_header_selects_auto_mode_before_starting(self): host = QMainWindow() host.runtime_settings = {"send_mode": "review"} host.settings_page = mock.Mock() host.settings_page.auto_send_mode = QCheckBox(host) host.settings_page.review_send_mode = QCheckBox(host) host.settings_page.review_send_mode.setChecked(True) host._running = False host.start_monitoring = mock.Mock() host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) bridge.startListen() self.assertTrue(host.settings_page.auto_send_mode.isChecked()) self.assertFalse(host.settings_page.review_send_mode.isChecked()) host.settings_page._emit_save.assert_called_once_with() host.start_monitoring.assert_called_once_with() bridge.deleteLater() host.deleteLater() def test_handoff_only_reports_tasks_the_backend_accepted(self): host = SimpleNamespace( _delete_queue_tasks=mock.Mock( return_value={ "deleted": ["done"], "scheduled": [], "protected": ["protected"], "error": "", } ), append_log=mock.Mock(), ) MainWindow._handoff_queue_tasks(host, ["done", "protected"]) host.append_log.assert_called_once_with( "已将 1 条会话转交人工,自动回复不会继续发送。", "notify", ) def test_queue_actions_select_exact_key_before_forwarding(self): host = QMainWindow() host.queue_page = mock.Mock() host._console_select_task = mock.Mock(return_value=True) host._schedule_console_push = mock.Mock() host._push_console_state = mock.Mock() host.append_log = mock.Mock() bridge = ConsoleBridge(host) actions = ( (bridge.cancelTask, host.queue_page._request_delete_selected), (bridge.retryTask, host.queue_page._request_retry_selected), (bridge.handoffTask, host.queue_page._request_handoff_selected), ) for invoke, forwarded in actions: with self.subTest(action=invoke.__name__): host.queue_page.reset_mock() host._console_select_task.reset_mock() host._console_select_task.return_value = True host._schedule_console_push.reset_mock() invoke("task-key") host._console_select_task.assert_called_once_with("task-key") forwarded.assert_called_once_with() host._schedule_console_push.assert_called_once_with(urgent=True) bridge.selectTask("selected-key") host._console_select_task.assert_called_with("selected-key") host._push_console_state.assert_called_once_with() bridge.deleteLater() host.deleteLater() def test_approve_task_reaches_the_host_with_the_exact_key(self): """「通过并发送」必须真的连到后端。 补这条是因为这个按钮以前根本不存在:审核模式下草稿贴进输入框、任务标成 "等待人工审核",界面上却只有"终止本任务"和"转人工"——没有任何放行入口。 看到的现象就是"AI 回复生成了、聊天框里也有字,但就是不发送"。 """ host = QMainWindow() host.queue_page = mock.Mock() host._console_select_task = mock.Mock(return_value=True) host._approve_queue_tasks = mock.Mock() host._schedule_console_push = mock.Mock() host._push_console_state = mock.Mock() host.append_log = mock.Mock() bridge = ConsoleBridge(host) bridge.approveTask("review-key") host._console_select_task.assert_called_once_with("review-key") host._approve_queue_tasks.assert_called_once_with(["review-key"]) bridge.deleteLater() host.deleteLater() def test_approving_a_task_that_left_the_queue_sends_nothing(self): """任务已经不在队列里就别再放行——否则可能把草稿发到别的会话去。""" host = QMainWindow() host.queue_page = mock.Mock() host._console_select_task = mock.Mock(return_value=False) host._approve_queue_tasks = mock.Mock() host._push_console_state = mock.Mock() host._schedule_console_push = mock.Mock() host.append_log = mock.Mock() bridge = ConsoleBridge(host) bridge.approveTask("stale-key") host._approve_queue_tasks.assert_not_called() host.append_log.assert_called_once_with( "所选任务已离开队列,未执行放行。", "warn", ) host._push_console_state.assert_called_once_with(force=True) bridge.deleteLater() host.deleteLater() def test_connect_backend_forwards_the_url_to_the_host(self) -> None: """控制台上填的后台地址要真的到得了后端。 这是桌面端唯一需要人填的东西。补这条测试是因为这个入口以前根本不存在: `BackendLoginDialog` 里有个「后台地址」输入框,但**没有任何地方打开过它** ——界面上改不了,只能去手工编辑 backend_connection.json。 """ host = QMainWindow() host.system_page = mock.Mock() host.system_page._connect_backend = mock.Mock( return_value={ "ok": True, "server_url": "https://x.example.com", "version": 12, "gateway_url": "https://x.example.com/gateway/v1/answer", "message": "ok", } ) host.append_log = mock.Mock() host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) bridge.connectBackend("https://x.example.com") host.system_page._connect_backend.assert_called_once_with("https://x.example.com") host._push_console_state.assert_called_once_with(force=True) logged = " ".join(str(call) for call in host.append_log.call_args_list) self.assertIn("https://x.example.com", logged) self.assertIn("v12", logged) bridge.deleteLater() host.deleteLater() def test_a_failed_connection_says_why_in_the_log(self) -> None: """失败必须留下原因。只说"连接失败"的话,人不知道是地址错了还是后台没起。""" host = QMainWindow() host.system_page = mock.Mock() host.system_page._connect_backend = mock.Mock( return_value={"ok": False, "message": "无法连接后台:timed out"} ) host.append_log = mock.Mock() host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) bridge.connectBackend("http://127.0.0.1:9") logged = " ".join(str(call) for call in host.append_log.call_args_list) self.assertIn("timed out", logged) self.assertIn("err", logged) bridge.deleteLater() host.deleteLater() def test_save_ai_only_applies_the_agent_name(self) -> None: """AI 设置页现在只有客服昵称是真正的本机设置。 温度、最大 tokens、当前模型搬去了后台的「模型清单 / 角色编排」;上下文轮数 和 MCP 随后台配置下发。真按客户端提交的值写下去,界面会提示"已保存并生效", 而下一次同步(每 5 分钟一次)就把它们盖回去——改了、提示成功了、什么都没 发生。这条测试挡的就是旧版客户端和手工调接口把它们塞回来。 """ host = QMainWindow() host.persona_page = mock.Mock() host.persona_page.save_config = mock.Mock(return_value=True) host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) bridge.saveAi(json.dumps({ "name": "贴心管家", "temperature": 1.9, "maxTokens": 9999, "context": 42, "mcpRounds": 19, "replyLength": 2, })) host.persona_page.agent_name.setText.assert_called_once_with("贴心管家") host.persona_page.temperature.setValue.assert_not_called() host.persona_page.max_tokens.setValue.assert_not_called() host.persona_page.rounds.setValue.assert_not_called() host.persona_page.mcp_rounds.setValue.assert_not_called() host.persona_page._reply_length_changed.assert_not_called() host.persona_page.save_config.assert_called_once_with() bridge.deleteLater() host.deleteLater() def test_an_over_long_agent_name_is_truncated(self) -> None: """昵称会拼进系统提示词,不能让它无限长。""" host = QMainWindow() host.persona_page = mock.Mock() host.persona_page.save_config = mock.Mock(return_value=True) host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) bridge.saveAi(json.dumps({"name": "长" * 200})) applied = host.persona_page.agent_name.setText.call_args.args[0] self.assertEqual(len(applied), 40) bridge.deleteLater() host.deleteLater() def test_a_blank_name_leaves_the_current_one_alone(self) -> None: """空名字是"没填",不是"改成空"——改成空会让提示词里出现一个没有名字的客服。""" host = QMainWindow() host.persona_page = mock.Mock() host.persona_page.save_config = mock.Mock(return_value=True) host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) bridge.saveAi(json.dumps({"name": " "})) host.persona_page.agent_name.setText.assert_not_called() bridge.deleteLater() host.deleteLater() def test_stale_queue_key_never_mutates_previous_selection(self): host = QMainWindow() host.queue_page = mock.Mock() host._console_select_task = mock.Mock(return_value=False) host._push_console_state = mock.Mock() host._schedule_console_push = mock.Mock() host.append_log = mock.Mock() bridge = ConsoleBridge(host) bridge.cancelTask("stale-key") host.queue_page._request_delete_selected.assert_not_called() host.append_log.assert_called_once_with( "所选任务已离开队列,未执行取消。", "warn", ) host._push_console_state.assert_called_once_with(force=True) bridge.deleteLater() host.deleteLater() def test_archive_selection_only_commits_an_existing_visible_key(self): host = QMainWindow() host.business_page = mock.Mock() host._console_selected_archive = "previous-key" host._select_archive_session = mock.Mock(return_value=False) host._push_console_state = mock.Mock() host.append_log = mock.Mock() bridge = ConsoleBridge(host) bridge.selectArchive("stale-key") self.assertEqual(host._console_selected_archive, "") host.business_page.refresh_data.assert_called_once_with() host.append_log.assert_called_once_with( "所选会话已不在归档中,已刷新会话列表。", "warn", ) host._push_console_state.assert_called_once_with(force=True) host.business_page.reset_mock() host._push_console_state.reset_mock() host._select_archive_session.return_value = True bridge.selectArchive("live-key") self.assertEqual(host._console_selected_archive, "live-key") host.business_page.refresh_data.assert_not_called() host._push_console_state.assert_called_once_with(force=True) bridge.deleteLater() host.deleteLater() def test_archive_date_button_uses_real_business_filter(self): host = QMainWindow() host.business_page = mock.Mock() host.business_page.archive_date = "2026-08-17" host._push_console_state = mock.Mock() bridge = ConsoleBridge(host) with mock.patch( "wechat_gui_qt.QInputDialog.getText", return_value=("2026-08-18", True), ): bridge.chooseArchiveDate() host.business_page._set_archive_date.assert_called_once_with("2026-08-18") host._push_console_state.assert_called_once_with(force=True) bridge.deleteLater() host.deleteLater() def test_sidebar_orb_is_wired_to_the_listen_toggle(self): """侧栏那颗球是用户最常点的开关,HTML 上必须真的绑着动作。 它曾经只是一个纯装饰的 div:没有 data-action,前端的点击派发只认 `[data-view]` 和 `[data-action]`,所以点它、点"已停止"三个字都毫无反应, 而界面上又没有任何提示——看上去就是"这个按钮坏了"。 """ html = Path(CONSOLE_HTML).read_text(encoding="utf-8") listener = re.search(r'