590 lines
22 KiB
Python
590 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""QWebChannel control-console bridge regression tests."""
|
|
|
|
import json
|
|
import os
|
|
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_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()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|