464 lines
19 KiB
Python
464 lines
19 KiB
Python
# -*- 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 BotThread, LogQueue, delete_pending_reply_file
|
||
from wechat_bot import WeChatBot
|
||
from test_support import queue_log_redirect
|
||
|
||
|
||
|
||
def setUpModule():
|
||
# 别把测试写进开发机真实的 queue_events.json——现场排查时，那些假会话
|
||
# （b"customer"、0x76 重复）混在真实流水里，看着就像机器人在反复重发。
|
||
queue_log_redirect.start()
|
||
|
||
|
||
def tearDownModule():
|
||
queue_log_redirect.stop()
|
||
|
||
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)
|
||
self.assertEqual(settings["send_delay_seconds"], 1.0)
|
||
self.assertEqual(settings["send_mode"], "auto")
|
||
self.assertEqual(settings["mouse_idle_seconds"], 5.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_invalid_saved_send_delay_falls_back_to_one_second(self):
|
||
self.assertEqual(wechat_gui.normalize_send_delay_seconds(True), 1.0)
|
||
payload = {
|
||
**wechat_gui.App._runtime_defaults(),
|
||
"send_delay_seconds": 31,
|
||
}
|
||
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["send_delay_seconds"], 1.0)
|
||
self.assertEqual(settings["mouse_idle_seconds"], 5.0)
|
||
|
||
def test_invalid_send_mode_falls_back_to_automatic(self):
|
||
self.assertEqual(wechat_gui.normalize_send_mode("unknown"), "auto")
|
||
self.assertEqual(wechat_gui.normalize_send_mode("review"), "review")
|
||
|
||
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,
|
||
send_delay_seconds=1.5,
|
||
send_mode="review",
|
||
)
|
||
|
||
def poll_once(scan_unread=True): # noqa: ARG001 # 适配 gui_runtime 新调用契约
|
||
observed.append((
|
||
fake_bot.message_batch_window_seconds,
|
||
fake_bot.send_delay_seconds,
|
||
fake_bot.send_mode,
|
||
))
|
||
if len(observed) == 1:
|
||
thread.set_message_batch_window_seconds(9)
|
||
thread.set_send_delay_seconds(2.5)
|
||
thread.set_send_mode("auto")
|
||
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, 1.5, "review"), (9, 2.5, "auto")],
|
||
)
|
||
|
||
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._runtime_settings = {
|
||
"send_delay_seconds": 3.5,
|
||
"send_mode": "review",
|
||
}
|
||
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)
|
||
self.assertEqual(saved["send_delay_seconds"], 3.5)
|
||
self.assertEqual(saved["send_mode"], "review")
|
||
app._thread.set_message_batch_window_seconds.assert_called_once_with(12.5)
|
||
app._thread.set_send_delay_seconds.assert_called_once_with(3.5)
|
||
app._thread.set_send_mode.assert_called_once_with("review")
|
||
|
||
def test_send_delay_wait_is_independent_and_interruptible(self):
|
||
bot = WeChatBot.__new__(WeChatBot)
|
||
bot.send_delay_seconds = 1.25
|
||
bot._stop_check = mock.Mock()
|
||
bot._stop_check.wait.side_effect = [False, True]
|
||
|
||
with mock.patch("builtins.print"):
|
||
self.assertFalse(bot.wait_before_send())
|
||
|
||
self.assertAlmostEqual(bot._stop_check.wait.call_args_list[0].args[0], 0.2)
|
||
|
||
def test_review_pending_state_survives_restart(self):
|
||
fp = b"r" * 40
|
||
with tempfile.TemporaryDirectory() as folder:
|
||
path = Path(folder) / "pending_replies.json"
|
||
bot = WeChatBot.__new__(WeChatBot)
|
||
bot._pending_reply_path = str(path)
|
||
bot._pending_reply_sessions = {
|
||
fp.hex(): {
|
||
"batch_ready": True,
|
||
"confirmed_unread": True,
|
||
"created_at": 1.0,
|
||
"updated_at": 2.0,
|
||
"awaiting_review": True,
|
||
"review_requested_at": 2.0,
|
||
"reply_text": "请人工确认",
|
||
"stage": "manual_review",
|
||
}
|
||
}
|
||
self.assertTrue(bot._persist_pending_replies())
|
||
|
||
restored_bot = WeChatBot.__new__(WeChatBot)
|
||
restored_bot._pending_reply_path = str(path)
|
||
restored = restored_bot._load_pending_replies()
|
||
|
||
self.assertTrue(restored[fp.hex()]["awaiting_review"])
|
||
self.assertEqual(restored[fp.hex()]["reply_text"], "请人工确认")
|
||
self.assertEqual(restored[fp.hex()]["stage"], "manual_review")
|
||
|
||
def test_manual_retry_only_requeues_retry_wait_tasks(self):
|
||
bot = WeChatBot.__new__(WeChatBot)
|
||
retry_fp = b"r" * 40
|
||
normal_fp = b"n" * 40
|
||
review_fp = b"v" * 40
|
||
uncertain_fp = b"u" * 40
|
||
bot._pending_reply_path = ""
|
||
bot._pending_reply_sessions = {
|
||
retry_fp.hex(): {
|
||
"stage": "retry_wait",
|
||
"stage_started_at": 1.0,
|
||
"updated_at": 1.0,
|
||
"last_error": "temporary",
|
||
"resume_failures": 3,
|
||
"last_resume_attempt": 999.0,
|
||
"display_name": "待重试客户",
|
||
},
|
||
normal_fp.hex(): {"stage": "generating", "updated_at": 1.0},
|
||
review_fp.hex(): {
|
||
"stage": "manual_review",
|
||
"awaiting_review": True,
|
||
"updated_at": 1.0,
|
||
},
|
||
uncertain_fp.hex(): {
|
||
"stage": "retry_wait",
|
||
"send_state": "uncertain",
|
||
"updated_at": 1.0,
|
||
},
|
||
}
|
||
# 对着真正的 `queue_log` 访问器断言，而不是私有的 `_queue_log_instance`：
|
||
# 模块级的队列日志重定向覆盖的就是这个属性，盯着私有字段会漏掉真实调用。
|
||
recorder = mock.Mock()
|
||
with mock.patch.object(
|
||
type(bot), "queue_log", property(lambda _self, _log=recorder: _log)
|
||
):
|
||
result = bot.retry_pending_replies(
|
||
[retry_fp.hex(), normal_fp.hex(), review_fp.hex(), uncertain_fp.hex()]
|
||
)
|
||
|
||
self.assertEqual(result["retried"], [retry_fp.hex()])
|
||
self.assertEqual(result["not_retryable"], [normal_fp.hex()])
|
||
self.assertEqual(
|
||
set(result["protected"]), {review_fp.hex(), uncertain_fp.hex()}
|
||
)
|
||
retried = bot._pending_reply_sessions[retry_fp.hex()]
|
||
self.assertEqual(retried["stage"], "queued")
|
||
self.assertEqual(retried["last_resume_attempt"], 0.0)
|
||
self.assertNotIn("last_error", retried)
|
||
self.assertEqual(retried["resume_failures"], 3)
|
||
recorder.append.assert_called_once()
|
||
|
||
def test_retry_request_queues_until_bot_startup_finishes(self):
|
||
key = (b"q" * 40).hex()
|
||
thread = BotThread(queue.Queue(), "在的", 1.0)
|
||
|
||
result = thread.retry_pending_tasks([key])
|
||
|
||
self.assertEqual(result["scheduled"], [key])
|
||
fake_bot = mock.Mock()
|
||
thread._drain_queued_retries(fake_bot)
|
||
fake_bot.retry_pending_replies.assert_called_once_with({key})
|
||
|
||
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)
|
||
|
||
|
||
class LogRetentionTest(TestCase):
|
||
"""设置页的「数据保留 N 天」必须真的删东西，不能只是个摆设。"""
|
||
|
||
def _log_dir(self):
|
||
import os
|
||
|
||
root = Path(tempfile.mkdtemp())
|
||
patcher = mock.patch.object(LogQueue, "LOG_DIR", str(root))
|
||
patcher.start()
|
||
self.addCleanup(patcher.stop)
|
||
os.makedirs(root, exist_ok=True)
|
||
return root
|
||
|
||
def _write(self, root, name, age_days):
|
||
import os
|
||
import time
|
||
|
||
path = root / name
|
||
path.write_text("旧日志", encoding="utf-8")
|
||
stamp = time.time() - age_days * 86400.0
|
||
os.utime(path, (stamp, stamp))
|
||
return path
|
||
|
||
def test_out_of_range_values_fall_back_to_the_documented_defaults(self):
|
||
self.assertEqual(LogQueue.normalize_retention_days(30), 30)
|
||
self.assertEqual(LogQueue.normalize_retention_days(0), LogQueue.MIN_RETENTION_DAYS)
|
||
self.assertEqual(LogQueue.normalize_retention_days(9999), LogQueue.MAX_RETENTION_DAYS)
|
||
for junk in (None, "", "abc", True):
|
||
self.assertEqual(
|
||
LogQueue.normalize_retention_days(junk),
|
||
LogQueue.DEFAULT_RETENTION_DAYS,
|
||
junk,
|
||
)
|
||
|
||
def test_logs_older_than_the_retention_window_are_deleted(self):
|
||
root = self._log_dir()
|
||
stale = self._write(root, "gui_20250101_010101.log", age_days=120)
|
||
fresh = self._write(root, "gui_20260820_010101.log", age_days=3)
|
||
self.addCleanup(LogQueue(queue.Queue(), retention_days=90).close)
|
||
self.assertFalse(stale.exists())
|
||
self.assertTrue(fresh.exists())
|
||
|
||
def test_a_shorter_window_deletes_more(self):
|
||
root = self._log_dir()
|
||
older = self._write(root, "gui_a.log", age_days=20)
|
||
newer = self._write(root, "gui_b.log", age_days=3)
|
||
self.addCleanup(LogQueue(queue.Queue(), retention_days=7).close)
|
||
self.assertFalse(older.exists())
|
||
self.assertTrue(newer.exists())
|
||
|
||
def test_the_file_count_cap_still_applies(self):
|
||
root = self._log_dir()
|
||
for index in range(LogQueue.KEEP_FILES + 5):
|
||
self._write(root, f"gui_{index:03d}.log", age_days=1)
|
||
self.addCleanup(LogQueue(queue.Queue(), retention_days=365).close)
|
||
remaining = sorted(root.glob("gui_*.log"))
|
||
self.assertLessEqual(len(remaining), LogQueue.KEEP_FILES)
|
||
|
||
def test_a_clock_rollback_never_deletes_a_fresh_log(self):
|
||
# mtime 落在未来（时钟被回拨）时 age 为负，绝不能当成"很旧"删掉
|
||
root = self._log_dir()
|
||
future = self._write(root, "gui_future.log", age_days=-30)
|
||
self.addCleanup(LogQueue(queue.Queue(), retention_days=7).close)
|
||
self.assertTrue(future.exists())
|
||
|
||
def test_unreadable_directory_never_breaks_startup(self):
|
||
root = self._log_dir()
|
||
self._write(root, "gui_x.log", age_days=1)
|
||
with mock.patch("gui_runtime.glob.glob", side_effect=OSError("盘掉了")):
|
||
self.addCleanup(LogQueue(queue.Queue(), retention_days=90).close) # 不抛异常即可
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|