"""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()