986 lines
42 KiB
Python
986 lines
42 KiB
Python
# -*- 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'<div class="listener"[^>]*>', html)
|
|
self.assertIsNotNone(listener, "侧栏监听开关整块不见了")
|
|
markup = listener.group(0)
|
|
self.assertIn('data-action="toggle-listen"', markup)
|
|
self.assertIn('id="listener"', markup)
|
|
self.assertIn('role="button"', markup)
|
|
# 看得出来能点：手型光标
|
|
self.assertIn(".listener{", html)
|
|
self.assertIn("cursor:pointer", html.split(".listener{", 1)[1][:400])
|
|
|
|
def test_the_console_script_handles_that_action(self):
|
|
script = Path(CONSOLE_HTML).with_name("console-app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn('a === "toggle-listen"', script)
|
|
self.assertIn('callBridge("toggleListen")', script)
|
|
# 提示语要跟着运行状态走，别让人点之前猜
|
|
self.assertIn('listener.title = STATE.listening', script)
|
|
# div 不会因为按 Enter 就变成 click，键盘路径要自己补
|
|
self.assertIn('closest("#listener")', script)
|
|
|
|
def test_no_layout_scrolls_anywhere(self):
|
|
"""整套界面不许出现滚动条——内容多了就往下铺，由整体缩放来兜。
|
|
|
|
原来的做法是"舞台写死 1630×965，装不下的部分在面板内部滚"。代价是每张
|
|
卡片多一行就冒出一根内层滚动条，或者干脆被裁掉：「人机共存」加了草稿超时
|
|
直发的两行之后，那张卡就是这么坏的。现在高度一律是下限，内容自己往下铺，
|
|
缩放按真实内容高度算，所以任何一页都不需要滚动条。
|
|
"""
|
|
html = Path(CONSOLE_HTML).read_text(encoding="utf-8")
|
|
for banned in ("overflow:auto", "overflow-y:auto", "overflow-x:auto",
|
|
"overflow:scroll", "overflow-y:scroll"):
|
|
self.assertNotIn(banned, html, f"还有 {banned}，说明某处仍会出现滚动条")
|
|
|
|
def test_the_stage_and_pages_grow_with_their_content(self):
|
|
"""舞台和单页都必须是"下限高度"，不能写死——写死就必然要么滚要么裁。"""
|
|
html = Path(CONSOLE_HTML).read_text(encoding="utf-8")
|
|
self.assertIn(".screen{min-height:876px;overflow:visible", html)
|
|
self.assertIn(".main{min-width:0;padding-top:24px;min-height:900px", html)
|
|
# 画布左上角起、等比缩放；宽高由 fit() 按窗口给定
|
|
self.assertIn("#stage{position:absolute;left:0;top:0;width:1630px;min-height:965px;height:auto", html)
|
|
self.assertIn("transform-origin:top left", html)
|
|
|
|
def test_the_scale_never_depends_on_which_page_is_open(self):
|
|
"""缩放只能由窗口尺寸决定，绝不能看当前页内容有多高。
|
|
|
|
踩过一次：按"这一页内容多高"算缩放之后，切一次页面字号和卡片就变一次
|
|
大小，同一个软件看着像两套界面。所以 fit() 里不许出现任何量内容高度的
|
|
动作——一旦量了，缩放就会随页面漂移。
|
|
"""
|
|
script = Path(CONSOLE_HTML).with_name("console-app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
body = script[script.index("fit = function()"):]
|
|
body = body[:body.index("function connectQt(")]
|
|
self.assertIn("innerHeight / DESIGN_H", body, "缩放不是按窗口高度定的")
|
|
self.assertNotIn("scrollHeight", body,
|
|
"fit() 又去量内容高度了，缩放会随页面变化")
|
|
self.assertNotIn("getBoundingClientRect", body,
|
|
"fit() 又去量元素尺寸了，缩放会随页面变化")
|
|
|
|
def test_every_save_button_gives_the_user_feedback(self):
|
|
"""点了保存必须看得见反馈。
|
|
|
|
现场：AI 设置页点「保存并发布」，界面一动不动——存成功没提示、存失败更
|
|
没提示，只能等下一次同步才发现改动根本没生效。人自然以为按钮坏了，
|
|
或者反复点。三个保存按钮都必须走带反馈的调用。
|
|
"""
|
|
script = Path(CONSOLE_HTML).with_name("console-app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
for action in ("save-ai", "save-automation", "save-system"):
|
|
index = script.index(f'"{action}":')
|
|
body = script[index:index + 260]
|
|
self.assertIn(
|
|
"callBridgeWithFeedback", body,
|
|
f"{action} 还是调完就完，用户看不到任何回执",
|
|
)
|
|
|
|
def test_the_feedback_helper_covers_pending_result_and_failure(self):
|
|
"""反馈要三段齐全：立刻响应、真实结果、失败也要说清。"""
|
|
script = Path(CONSOLE_HTML).with_name("console-app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
helper = script[script.index("function callBridgeWithFeedback"):]
|
|
helper = helper[:2400] # 这个函数体不到 2400 字符
|
|
self.assertIn("is-busy", helper, "按钮点下去没有处理中的样子")
|
|
self.assertIn("button.disabled = true", helper, "没有防连点")
|
|
self.assertIn("options.pending", helper, "点击瞬间没有提示")
|
|
self.assertIn("data.ok", helper, "没有读后台回传的真实结果")
|
|
self.assertIn("界面没有连上后台", helper, "后台不在时没有说清")
|
|
# 兜底不许假装成功
|
|
self.assertIn("没收到后台确认", helper, "拿不到确认时不能说得像成功了")
|
|
|
|
def test_the_save_slots_report_their_result_back(self):
|
|
"""Qt 侧的保存槽必须把结果回传，否则网页那边永远只能瞎猜。"""
|
|
source = Path(__file__).with_name("wechat_gui_qt.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
for slot in ("saveAi", "saveAutomationSettings", "setSystemField"):
|
|
index = source.index(f"def {slot}(")
|
|
head = source[max(0, index - 120):index]
|
|
self.assertIn(
|
|
"result=str", head,
|
|
f"{slot} 没有返回值，控制台拿不到成功/失败",
|
|
)
|
|
|
|
def test_the_side_rail_height_does_not_follow_page_content(self):
|
|
"""侧栏在每一页必须一样高。
|
|
|
|
之前工作区是被内容撑高的，侧栏跟着拉伸——七个页面的内容高矮不同，侧栏
|
|
就长短不一，切一次页面它就变一次长度。改成：舞台纵向弹性、工作区吃满
|
|
除标题栏外的全部高度、侧栏铺满工作区，于是侧栏高度只由画布决定，和当前
|
|
是哪一页无关。
|
|
"""
|
|
html = Path(CONSOLE_HTML).read_text(encoding="utf-8")
|
|
self.assertIn("display:flex;flex-direction:column", html,
|
|
"舞台不是纵向弹性，工作区没法吃满高度")
|
|
self.assertIn(".workspace{flex:1 1 auto", html,
|
|
"工作区没有吃满剩余高度，侧栏会被内容带着变长")
|
|
self.assertIn(".rail{height:100%", html,
|
|
"侧栏没有铺满工作区，高度会跟着内容走")
|
|
|
|
def test_the_canvas_fills_the_window_in_both_directions(self):
|
|
"""画布要铺满窗口：高度按设计画布等比缩放，宽度靠加宽画布本身补齐。
|
|
|
|
宽度不补的话，页面一长高就只能整体缩小，右边空出一大块——那正是
|
|
"不适配屏幕"的直接原因。
|
|
"""
|
|
script = Path(CONSOLE_HTML).with_name("console-app.js").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("Math.max(DESIGN_W, innerWidth / scale)", script,
|
|
"画布宽度没有撑到铺满窗口")
|
|
self.assertIn("Math.max(DESIGN_H, innerHeight / scale)", script,
|
|
"画布高度没有撑到铺满窗口")
|
|
# 缩放必须同时受宽高约束：只看高度的话，窗口一窄内容就横着被切掉
|
|
self.assertIn("Math.min(innerWidth / DESIGN_W, innerHeight / DESIGN_H)", script,
|
|
"缩放没有同时受宽高两个方向约束")
|
|
|
|
|
|
|
|
def test_page_grids_use_proportional_columns(self):
|
|
"""各栏必须按比例分配宽度。写死像素的话，舞台变宽它们也不会跟着变宽，
|
|
右边照样留白——那正是"不适配屏幕"的直接原因。"""
|
|
import re as _re
|
|
|
|
html = Path(CONSOLE_HTML).read_text(encoding="utf-8")
|
|
for grid in (".auto-grid", ".queue-view-grid", ".archive-grid", ".ai-top",
|
|
".automation-top", ".automation-mid", ".logs-grid", ".log-mid"):
|
|
rule = _re.search(_re.escape(grid) + r"\{[^}]*\}", html)
|
|
self.assertIsNotNone(rule, f"{grid} 规则不见了")
|
|
cols = _re.search(r"grid-template-columns:([^;]*);", rule.group(0))
|
|
self.assertIsNotNone(cols, f"{grid} 没有列定义")
|
|
self.assertNotRegex(
|
|
cols.group(1), r"\d+px",
|
|
f"{grid} 还在用写死像素的列宽，舞台变宽时右边会留白",
|
|
)
|
|
|
|
def test_no_layout_container_keeps_a_hard_coded_height(self):
|
|
"""去掉滚动条之后，"装不下"不再表现为滚动条，而是内容画到别的面板身上。
|
|
|
|
这次真的发生了：会话归档和 AI 设置两页的面板互相压在一起。原因是只把
|
|
一部分容器改成了下限高度，剩下那些还写死着高度——`overflow:visible` 时
|
|
它们既不滚也不裁，直接往下面那块面板上画。所以布局容器必须一个不落。
|
|
"""
|
|
import re as _re
|
|
|
|
html = Path(CONSOLE_HTML).read_text(encoding="utf-8")
|
|
for grid in (".auto-grid", ".queue-view-grid", ".archive-grid", ".ai-top",
|
|
".ai-bottom", ".automation-top", ".automation-mid", ".log-mid",
|
|
".system-mid"):
|
|
rule = _re.search(_re.escape(grid) + r"\{[^}]*\}", html)
|
|
self.assertIsNotNone(rule, f"{grid} 规则不见了")
|
|
self.assertNotRegex(
|
|
rule.group(0), r"(?<!min-)height:\d+px",
|
|
f"{grid} 还写死着高度，内容一多就会画到别的面板上",
|
|
)
|
|
|
|
|
|
|
|
|
|
def test_editable_fields_look_editable(self):
|
|
"""能改的地方要看得出来能改。
|
|
|
|
自动化设置里那几个数字原来是 border:0;background:transparent;outline:none
|
|
——跟旁边的粗体蓝字长得一模一样。没人知道它能改，点进去也没有任何反馈。
|
|
"""
|
|
html = Path(CONSOLE_HTML).read_text(encoding="utf-8")
|
|
block = html.split("可输入的地方要看得出来能输入", 1)
|
|
self.assertEqual(len(block), 2, "内联数字框的可编辑外观样式不见了")
|
|
style = block[1]
|
|
for prop in ("border:1px solid", "cursor:text", "border-radius:10px"):
|
|
self.assertIn(prop, style, f"数字框缺少 {prop}")
|
|
# 聚焦必须有反馈，而且权重要压得过那几条三层类名的静止规则
|
|
self.assertIn(".automation-top .flow-pill.live-flow-pill .num:focus", style)
|
|
self.assertIn(".search:focus-within", style)
|
|
# 裸 select 要跟整体设计一致，只读的要看得出来不能改
|
|
self.assertIn(".system-line select{", style)
|
|
self.assertIn(".textarea[readonly]", style)
|
|
|
|
def test_number_fields_tell_you_the_allowed_range(self):
|
|
script = Path(CONSOLE_HTML).with_name("console-app.js").read_text(encoding="utf-8")
|
|
self.assertIn("annotateEditableFields", script)
|
|
self.assertIn('input[type="number"]', script)
|
|
self.assertIn("可修改：", script)
|
|
# 必须挂在渲染之后，否则换页就没了
|
|
pattern = "syncOrb();" + chr(10) + " annotateEditableFields();"
|
|
self.assertIn(pattern, script, "注解没有挂在每次渲染之后")
|
|
|
|
def test_toggle_listen_starts_and_stops_the_host(self):
|
|
host = QMainWindow()
|
|
host._running = False
|
|
host.start_monitoring = mock.Mock()
|
|
host.stop_monitoring = mock.Mock()
|
|
host._push_console_state = mock.Mock()
|
|
bridge = ConsoleBridge(host)
|
|
|
|
bridge.toggleListen()
|
|
host.start_monitoring.assert_called_once()
|
|
host.stop_monitoring.assert_not_called()
|
|
|
|
host._running = True
|
|
bridge.toggleListen()
|
|
host.stop_monitoring.assert_called_once()
|
|
self.assertEqual(host.start_monitoring.call_count, 1)
|
|
bridge.deleteLater()
|
|
host.deleteLater()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|