141 lines
6.7 KiB
Python
141 lines
6.7 KiB
Python
"""GUI regressions for the WebEngine model path, with synthetic local data only."""
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import time
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
from PySide6.QtWidgets import QApplication
|
|
import wechat_gui_qt as gui
|
|
from review_state import pending_attention_counts
|
|
|
|
|
|
class ConsoleHost:
|
|
_console_people = gui.MainWindow._console_people
|
|
_console_task = gui.MainWindow._console_task
|
|
_console_select_task = gui.MainWindow._console_select_task
|
|
_console_stage_durations = gui.MainWindow._console_stage_durations
|
|
_console_queue_wait_label = staticmethod(gui.MainWindow._console_queue_wait_label)
|
|
_console_seconds_label = staticmethod(gui.MainWindow._console_seconds_label)
|
|
_console_clock_label = staticmethod(gui.MainWindow._console_clock_label)
|
|
|
|
def __init__(self, page):
|
|
self.queue_page = page
|
|
self.runtime_settings = {"mask_sensitive": False}
|
|
self._console_selected_key = ""
|
|
|
|
|
|
class DesktopModelTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def setUp(self):
|
|
self.temp = self.enterContext(tempfile.TemporaryDirectory())
|
|
self.root = Path(self.temp)
|
|
self.path = self.root / "pending.json"
|
|
self.events = []
|
|
self.enterContext(patch.object(gui, "SCRIPT_DIR", self.root))
|
|
self.enterContext(patch.object(gui, "queue_path_for_transport", return_value=self.path))
|
|
log = SimpleNamespace(recent=lambda *args: copy.deepcopy(self.events))
|
|
self.enterContext(patch("queue_log.QueueLog", return_value=log))
|
|
self.write({})
|
|
self.page = gui.QueuePage(build_task_cards=False)
|
|
self.host = ConsoleHost(self.page)
|
|
self.addCleanup(self.page.deleteLater)
|
|
|
|
def write(self, rows):
|
|
self.path.write_text(json.dumps({"pending": rows}), encoding="utf-8")
|
|
|
|
@staticmethod
|
|
def row(index, **updates):
|
|
return dict(display_name=f"模拟客户{index}", created_at=1700000000+index,
|
|
stage="queued", confirmed_unread=True, **updates)
|
|
|
|
def test_large_queue_without_hidden_cards_preserves_updates_and_selection(self):
|
|
rows = {f"M:{n}": self.row(n) for n in range(300)}
|
|
self.write(rows)
|
|
with patch.object(self.page, "_add_task_list_item", side_effect=AssertionError("hidden cards rebuilt")):
|
|
self.page.refresh_data()
|
|
self.assertEqual(len(self.host._console_people()), 300)
|
|
self.assertEqual(self.page.task_list.count(), 0)
|
|
original = self.page.queue_table.item(299, 1)
|
|
rows["M:299"].update(stage="awaiting_review", staged_user_text="新来的消息",
|
|
staged_reply_text="已生成的草稿", awaiting_review=True)
|
|
self.write(rows)
|
|
self.page.refresh_data()
|
|
self.assertIs(self.page.queue_table.item(299, 1), original)
|
|
self.assertTrue(self.host._console_select_task("M:299"))
|
|
task = self.host._console_task(self.host._console_people())
|
|
self.assertEqual(task["key"], "M:299")
|
|
self.assertEqual(task["messages"][0]["text"], "新来的消息")
|
|
self.assertEqual(task["answer"], "已生成的草稿")
|
|
self.assertTrue(task["canApprove"])
|
|
|
|
def test_filters_and_review_counters_without_list_widgets(self):
|
|
rows = {"review": self.row(1), "human": self.row(2), "normal": self.row(3)}
|
|
rows["review"].update(stage="awaiting_review", staged_reply_text="草稿")
|
|
rows["human"].update(stage="manual_takeover")
|
|
self.write(rows)
|
|
self.page.refresh_data()
|
|
for label, expected in [("待审核", ["review"]), ("需人工", ["human"]), ("等待中", ["normal"])]:
|
|
self.page._set_task_filter(label)
|
|
self.assertEqual([row["key"] for row in self.host._console_people()], expected)
|
|
self.assertEqual(pending_attention_counts(self.page.pending_states),
|
|
{"review": 1, "human": 1, "total": 2})
|
|
self.page._set_task_filter("全部")
|
|
self.assertEqual(len(self.host._console_people()), 3)
|
|
|
|
def test_completion_empty_queue_and_new_arrival_have_no_stale_rows(self):
|
|
for rows in ({"old": self.row(0)}, {}, {"new": self.row(1)}):
|
|
self.write(rows)
|
|
self.page.refresh_data()
|
|
self.assertEqual([row["key"] for row in self.host._console_people()], list(rows))
|
|
self.assertFalse(self.host._console_select_task("old"))
|
|
self.assertTrue(self.host._console_select_task("new"))
|
|
|
|
def test_history_refresh_reuses_unchanged_rows_and_recounts_at_midnight(self):
|
|
self.events = [{"ts": time.time(), "name": "模拟客户", "event": "已发送", "detail": "测试"}]
|
|
self.page._refresh_history()
|
|
original = self.page.history_table.item(0, 0)
|
|
self.page._refresh_history()
|
|
self.assertIs(self.page.history_table.item(0, 0), original)
|
|
self.assertEqual(self.page.sent_metric.value.text(), "1")
|
|
localtime, tomorrow = time.localtime, time.time() + 86400
|
|
with patch.object(gui.time, "localtime", side_effect=lambda seconds=None: localtime(tomorrow if seconds is None else seconds)):
|
|
self.page._refresh_history()
|
|
self.assertIsNot(self.page.history_table.item(0, 0), original)
|
|
self.assertEqual(self.page.sent_metric.value.text(), "0")
|
|
self.events.clear()
|
|
self.page._refresh_history()
|
|
self.assertEqual(self.page.sent_metric.value.text(), "0")
|
|
|
|
def test_log_bursts_keep_history_counters_search_and_export(self):
|
|
page = gui.LogPage(render_event_cards=False)
|
|
self.addCleanup(page.deleteLater)
|
|
with patch.object(page, "_assign_event_row", side_effect=AssertionError("hidden log card repaint")):
|
|
for n in range(200):
|
|
page.append(f"模拟事件 {n}", "notify")
|
|
page.append("模型超时", "warn")
|
|
page.append("发送失败", "err")
|
|
self.assertEqual(len(page._event_history), 80)
|
|
self.assertEqual(page._timeout_count, 1)
|
|
self.assertEqual(page._send_fail_count, 1)
|
|
self.assertEqual(page._warning_count, 2)
|
|
self.assertIn("模拟事件 0", page._visible_log_text())
|
|
self.assertIn("模拟事件 199", page._visible_log_text())
|
|
page.event_search.setText("模型超时")
|
|
host = SimpleNamespace(log_page=page, runtime_settings={})
|
|
result = gui.MainWindow._console_logs(host)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]["detail"], "模型超时")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|