# -*- coding: utf-8 -*- """通用运行设置与连续消息合并窗口测试。""" import json import queue import tempfile from pathlib import Path from unittest import TestCase, main, mock import wechat_gui from gui_runtime import LogQueue, delete_pending_reply_file from wechat_bot import WeChatBot class RuntimeSettingsTest(TestCase): def test_legacy_settings_without_batch_window_keep_twenty_second_default(self): payload = { "auto_reply_text": "在的", "poll_interval": 2.0, "mouse_idle_enabled": True, "mouse_idle_seconds": 5.0, } with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "app_settings.json" path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") app = wechat_gui.App.__new__(wechat_gui.App) with mock.patch.object(wechat_gui, "APP_SETTINGS_FILE", str(path)): settings = app._load_runtime_settings() self.assertEqual(settings["message_batch_window_seconds"], 20.0) def test_invalid_saved_batch_window_falls_back_to_twenty_seconds(self): self.assertEqual( wechat_gui.normalize_message_batch_window_seconds(True), 20.0, ) payload = { **wechat_gui.App._runtime_defaults(), "message_batch_window_seconds": 999, } with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "app_settings.json" path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") app = wechat_gui.App.__new__(wechat_gui.App) with mock.patch.object(wechat_gui, "APP_SETTINGS_FILE", str(path)): settings = app._load_runtime_settings() self.assertEqual(settings["message_batch_window_seconds"], 20.0) def test_runtime_thread_syncs_updated_batch_window_before_next_poll(self): observed = [] fake_bot = mock.Mock() fake_bot.connect.return_value = True fake_bot._window_ready = True fake_bot.hwnd = 100 fake_bot.L = fake_bot.T = 0 fake_bot.R = 1600 fake_bot.B = 900 fake_bot.input_x = 1200 fake_bot.input_y = 780 fake_bot.security_verification_required = False fake_bot.reply_count = 0 fake_bot.false_pos_rows = set() thread = wechat_gui.BotThread( queue.Queue(), "在的", 0.001, message_batch_window_seconds=7, ) def poll_once(): observed.append(fake_bot.message_batch_window_seconds) if len(observed) == 1: thread.set_message_batch_window_seconds(9) else: thread.stop_event.set() fake_bot._poll_once.side_effect = poll_once with mock.patch("wechat_bot.WeChatBot", return_value=fake_bot): thread.run() self.assertEqual(observed, [7.0, 9]) def test_classic_settings_save_persists_and_updates_live_thread(self): app = wechat_gui.App.__new__(wechat_gui.App) app._runtime_save_job = None app._last_runtime_settings = {} app._reply_var = mock.Mock() app._reply_var.get.return_value = "在的" app._poll_var = mock.Mock() app._poll_var.get.return_value = "2" app._idle_seconds_var = mock.Mock() app._idle_seconds_var.get.return_value = "5" app._batch_window_var = mock.Mock() app._batch_window_var.get.return_value = "12.5" app._mouse_idle_var = mock.Mock() app._mouse_idle_var.get.return_value = True app._thread = mock.Mock() app._thread.is_alive.return_value = True with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "app_settings.json" with mock.patch.object(wechat_gui, "APP_SETTINGS_FILE", str(path)): self.assertTrue(app._save_runtime_settings(silent=True)) saved = json.loads(path.read_text(encoding="utf-8")) self.assertEqual(saved["message_batch_window_seconds"], 12.5) app._thread.set_message_batch_window_seconds.assert_called_once_with(12.5) def test_bot_uses_instance_batch_window_when_call_has_no_override(self): bot = WeChatBot.__new__(WeChatBot) bot.message_batch_window_seconds = 7 bot._active_session_fp = b"target" bot._active_identity_signature = b"identity" bot._chat_identity_signature = mock.Mock(return_value=b"identity") bot._chat_surface_signature = mock.Mock(return_value=b"surface") bot._raw_selected_session_fingerprint = mock.Mock(return_value=b"target") bot._selected_session_fingerprint = mock.Mock(return_value=b"target") bot._stop_check = mock.Mock() bot._stop_check.wait.return_value = True with mock.patch("builtins.print") as output: self.assertFalse(bot._wait_for_message_batch(b"target")) rendered = "\n".join( " ".join(str(part) for part in call.args) for call in output.call_args_list ) self.assertIn("7 秒", rendered) def test_classic_capsule_renders_each_progress_message_as_a_numbered_step(self): app = wechat_gui.App.__new__(wechat_gui.App) app._running = True app._capsule_progress_text = "" app._capsule_progress_step = 0 app._capsule_progress = mock.Mock() app._update_capsule_progress("消息读取 6/8 · 复制聊天第 1/1 屏") app._capsule_progress.configure.assert_called_once_with( text="步骤 001 · 消息读取 6/8 · 复制聊天第 1/1 屏" ) self.assertEqual(app._capsule_progress_step, 1) self.assertEqual(app._capsule_progress_text, "消息读取 6/8 · 复制聊天第 1/1 屏") def test_classic_queue_consumes_the_runtime_progress_channel(self): app = wechat_gui.App.__new__(wechat_gui.App) app._queue = queue.Queue() app._queue.put(("progress", "视觉识别 3/4 · 等待多模态模型分析")) app._sync_console_dpi = mock.Mock() app._update_capsule_progress = mock.Mock() app._start_time = None app._running = True app.after = mock.Mock() app._process_queue() app._update_capsule_progress.assert_called_once_with( "视觉识别 3/4 · 等待多模态模型分析" ) def test_atomic_operation_progress_includes_phase_step_detail_and_customer(self): bot = WeChatBot.__new__(WeChatBot) fp = b"customer" bot._pending_reply_sessions = { fp.hex(): {"display_name": "高瑞@微信"} } bot._progress_text = "" bot.progress_cb = mock.Mock() bot.report_operation("发送", 9, 12, "按 Ctrl+V 粘贴回复", fp) bot.progress_cb.assert_called_once_with( "发送 9/12 · 按 Ctrl+V 粘贴回复 · 高瑞@微信" ) def test_tagged_low_level_logs_are_mirrored_to_capsule_progress(self): target = queue.Queue() log = LogQueue.__new__(LogQueue) log.target_queue = target log._handle = None log.write(" [未读扫描] 正在从列表顶部逐页查找未置顶未读…") self.assertEqual(target.get_nowait()[0], "log") self.assertEqual( target.get_nowait(), ("progress", "未读扫描 · 正在从列表顶部逐页查找未置顶未读…"), ) def test_customer_chat_body_is_not_mirrored_into_floating_capsule(self): self.assertEqual( LogQueue.progress_from_log("[AI] 本次提取的新内容:\n客户隐私正文"), "", ) def test_message_copy_retries_when_wecom_writes_clipboard_late(self): bot = WeChatBot.__new__(WeChatBot) bot.report_operation = mock.Mock() sentinel = WeChatBot._CLIP_SENTINEL clipboard_reads = [sentinel] + [sentinel] * 6 + ["客户刚发的消息"] with mock.patch("wechat_bot.pyperclip.copy"), \ mock.patch("wechat_bot.pyperclip.paste", side_effect=clipboard_reads), \ mock.patch("wechat_bot.pyautogui.hotkey") as hotkey, \ mock.patch("wechat_bot.time.sleep"): copied = bot._copy_selection() self.assertEqual(copied, "客户刚发的消息") self.assertEqual(hotkey.call_count, 2) def test_manual_queue_delete_removes_only_selected_full_key_atomically(self): first = "a" * 80 second = "b" * 80 payload = { "pending": { first: {"display_name": "甲", "created_at": 1}, second: {"display_name": "乙", "created_at": 2}, }, "version": 1, } with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "pending_replies.json" path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") with mock.patch("queue_log.QueueLog"): result = delete_pending_reply_file([first], str(path)) saved = json.loads(path.read_text(encoding="utf-8")) self.assertEqual(result["deleted"], [first]) self.assertNotIn(first, saved["pending"]) self.assertIn(second, saved["pending"]) self.assertEqual(saved["version"], 1) if __name__ == "__main__": main()