# -*- coding: utf-8 -*- """视觉监听、稳定确认与通话保护的定向回归。""" import threading from unittest import TestCase, mock import numpy as np from wechat_bot import WeChatBot class SystemEntryFilterTest(TestCase): def test_known_system_entry_is_removed_before_fingerprint_and_click(self): bot = WeChatBot.__new__(WeChatBot) bot.identity_by_name = True bot.detect_badge_rows = mock.Mock(return_value=[18]) bot._row_display_name = mock.Mock(return_value="行业资讯") bot._log_system_entry_skip = mock.Mock() bot._fp_from_name = mock.Mock(return_value=b"s" * 40) bot._session_fingerprint = mock.Mock() non_conversations = set() found = bot._target_from_session_image( np.zeros((100, 200, 4), dtype=np.uint8), set(), non_conversations, ) self.assertIsNone(found) bot._session_fingerprint.assert_not_called() bot._log_system_entry_skip.assert_called_once_with("行业资讯") self.assertIn(b"s" * 40, non_conversations) def _ambiguous_bot(self, decision): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot.identity_by_name = True bot._flat_visual_proof_fps = set() bot.detect_badge_rows = mock.Mock(return_value=[24]) bot._row_display_name = mock.Mock(return_value="新应用提醒") bot._session_fingerprint = mock.Mock(return_value=b"a" * 40) bot._flat_session_rejected = mock.Mock(return_value=False) bot._is_real_conversation = mock.Mock(return_value=False) bot._flat_row_requires_visual_proof = mock.Mock(return_value=True) bot._classify_ambiguous_session_row = mock.Mock(return_value=decision) bot._queue_log_instance = mock.Mock() return bot def test_multimodal_system_row_is_never_clicked_to_probe_it(self): bot = self._ambiguous_bot( {"kind": "system_entry", "reply_capable": False, "confidence": 0.97} ) excluded = set() found = bot._target_from_session_image( np.zeros((100, 200, 4), dtype=np.uint8), set(), excluded ) self.assertIsNone(found) self.assertIn(b"a" * 40, excluded) self.assertNotIn((b"a" * 40).hex(), bot._flat_visual_proof_fps) def test_unknown_row_fails_closed_and_keeps_its_unread_for_next_round(self): bot = self._ambiguous_bot( {"kind": "unknown", "reply_capable": False, "confidence": 0.31} ) excluded = set() found = bot._target_from_session_image( np.zeros((100, 200, 4), dtype=np.uint8), set(), excluded ) self.assertIsNone(found) self.assertIn(b"a" * 40, excluded) def test_high_confidence_customer_row_enters_post_click_proof_flow(self): bot = self._ambiguous_bot( {"kind": "customer_chat", "reply_capable": True, "confidence": 0.96} ) found = bot._target_from_session_image( np.zeros((100, 200, 4), dtype=np.uint8), set(), set() ) self.assertEqual(found, (24, b"a" * 40)) self.assertIn((b"a" * 40).hex(), bot._flat_visual_proof_fps) class StableUnreadTest(TestCase): def test_strict_mode_uses_the_second_frame_coordinates(self): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot.identity_by_name = False first = np.zeros((120, 220, 4), dtype=np.uint8) second = np.ones((120, 220, 4), dtype=np.uint8) fp = b"u" * 40 bot.capture_session_list = mock.Mock(return_value=second) bot.detect_badge_rows = mock.Mock(return_value=[76]) bot._session_fingerprint = mock.Mock(return_value=fp) bot._session_fp_matches = mock.Mock(return_value=True) with mock.patch("wechat_bot.time.sleep"): result = bot._stabilize_unread_target(first, (18, fp)) self.assertIs(result[0], second) self.assertEqual(result[1:], (76, fp)) class ModelWaitObserverTest(TestCase): def test_new_unread_is_queued_only_after_two_matching_frames(self): bot = WeChatBot.__new__(WeChatBot) bot.identity_by_name = True bot._pending_reply_sessions = {} bot._observer_candidate_counts = {} bot.detect_badge_rows = mock.Mock(return_value=[24]) bot._row_display_name = mock.Mock(return_value="客户甲") bot._session_fingerprint = mock.Mock(return_value=b"c" * 40) bot._is_real_conversation = mock.Mock(return_value=True) bot._flat_row_requires_visual_proof = mock.Mock(return_value=False) bot._mark_reply_pending = mock.Mock(return_value=True) bot._save_task_visual_evidence = mock.Mock(return_value="") frame = np.zeros((100, 200, 4), dtype=np.uint8) bot._observe_unreads_during_model_wait(frame) bot._mark_reply_pending.assert_not_called() bot._observe_unreads_during_model_wait(frame) bot._mark_reply_pending.assert_called_once_with( b"c" * 40, confirmed_unread=True, requires_visual_proof=False, bind_identity=False, display_name="客户甲", stage="queued", ) def test_ambiguous_app_row_is_not_queued_while_reply_model_is_busy(self): bot = WeChatBot.__new__(WeChatBot) bot.identity_by_name = True bot._pending_reply_sessions = {} bot._observer_candidate_counts = {} bot.detect_badge_rows = mock.Mock(return_value=[24]) bot._row_display_name = mock.Mock(return_value="新应用提醒") bot._session_fingerprint = mock.Mock(return_value=b"x" * 40) bot._is_real_conversation = mock.Mock(return_value=False) bot._flat_row_requires_visual_proof = mock.Mock(return_value=True) bot._mark_reply_pending = mock.Mock(return_value=True) frame = np.zeros((100, 200, 4), dtype=np.uint8) bot._observe_unreads_during_model_wait(frame) bot._observe_unreads_during_model_wait(frame) bot._mark_reply_pending.assert_not_called() def test_model_wait_flag_is_set_only_during_the_remote_call(self): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot._model_waiting = threading.Event() bot._observer_stop = threading.Event() bot._ensure_visual_observer = mock.Mock() def remote_call(): self.assertTrue(bot._model_waiting.is_set()) return "回复" self.assertEqual(bot._call_model_with_observer(remote_call), "回复") self.assertFalse(bot._model_waiting.is_set()) class StrictModalTest(TestCase): def test_centered_business_popup_is_detected_without_dimmed_scrim(self): frame = np.zeros((900, 1400, 4), dtype=np.uint8) with ( mock.patch("wechat_bot.find_blocking_modal_close", return_value=(760, 210)), mock.patch("wechat_bot.looks_like_modal_scrim", return_value=False), ): self.assertEqual( WeChatBot._workspace_modal_candidate(frame), (760, 210), ) def test_window_titlebar_close_is_not_mistaken_for_business_popup(self): frame = np.zeros((900, 1400, 4), dtype=np.uint8) with mock.patch( "wechat_bot.find_blocking_modal_close", return_value=(1380, 12) ): self.assertIsNone(WeChatBot._workspace_modal_candidate(frame)) def test_non_message_popup_gets_one_escape_after_two_stable_frames(self): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot._workspace_modal_action_key = "" bot._workspace_modal_action_ts = 0.0 bot.hwnd = 123 first = np.zeros((500, 900, 4), dtype=np.uint8) second = np.zeros_like(first) after = np.ones_like(first) bot._capture_full_window = mock.Mock(side_effect=[second, after]) bot.wait_for_mouse_idle = mock.Mock(return_value=True) bot._security_gate_visible = mock.Mock(return_value=False) bot._begin_bot_mouse = mock.Mock() bot._end_bot_mouse = mock.Mock() with ( mock.patch("wechat_bot.looks_like_security_verification", return_value=False), mock.patch("wechat_bot.find_blocking_modal_close", return_value=(620, 180)), mock.patch("wechat_bot.looks_like_modal_scrim", return_value=False), mock.patch("wechat_bot.safe_set_foreground", return_value=True), mock.patch("wechat_bot.win32gui.GetForegroundWindow", return_value=123), mock.patch("wechat_bot.pyautogui.press") as press, mock.patch("wechat_bot.pyautogui.click") as click, mock.patch("wechat_bot.time.sleep"), ): self.assertTrue(bot._dismiss_workspace_modal("测试", full=first)) press.assert_called_once_with("esc") click.assert_not_called() def test_escape_resistant_popup_uses_only_the_three_frame_close_button(self): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot._workspace_modal_action_key = "" bot._workspace_modal_action_ts = 0.0 bot.hwnd = 123 first = np.zeros((500, 900, 4), dtype=np.uint8) unchanged = np.zeros_like(first) closed = np.ones_like(first) bot._capture_full_window = mock.Mock( side_effect=[unchanged.copy(), unchanged.copy(), closed] ) bot.wait_for_mouse_idle = mock.Mock(return_value=True) bot._security_gate_visible = mock.Mock(return_value=False) bot._begin_bot_mouse = mock.Mock() bot._end_bot_mouse = mock.Mock() with ( mock.patch("wechat_bot.looks_like_security_verification", return_value=False), mock.patch("wechat_bot.find_blocking_modal_close", return_value=(620, 180)), mock.patch("wechat_bot.looks_like_modal_scrim", return_value=False), mock.patch("wechat_bot.safe_set_foreground", return_value=True), mock.patch("wechat_bot.win32gui.GetForegroundWindow", return_value=123), mock.patch( "wechat_bot.win32gui.GetWindowRect", return_value=(100, 200, 1000, 700), ), mock.patch("wechat_bot.pyautogui.press") as press, mock.patch("wechat_bot.pyautogui.click") as click, mock.patch("wechat_bot.time.sleep"), ): self.assertTrue(bot._dismiss_workspace_modal("测试", full=first)) press.assert_called_once_with("esc") click.assert_called_once_with(720, 380) def test_workspace_recovery_closes_popup_before_opening_messages(self): bot = WeChatBot.__new__(WeChatBot) frame = np.zeros((500, 900, 4), dtype=np.uint8) bot._message_workspace_ready = mock.Mock(side_effect=[False, True]) bot._security_gate_visible = mock.Mock(return_value=False) bot._visible_call_window_title = mock.Mock(return_value="") bot._ui_guard_surface_signature = mock.Mock(return_value="business-popup") bot._workspace_modal_candidate = mock.Mock(return_value=(620, 180)) bot._dismiss_workspace_modal = mock.Mock(return_value=True) bot._capture_full_window = mock.Mock(return_value=frame) bot._open_messages_page = mock.Mock(return_value=True) bot._run_ai_page_guard = mock.Mock(return_value=False) with mock.patch("wechat_bot.looks_like_security_verification", return_value=False): self.assertTrue(bot._recover_message_workspace("测试", full=frame)) bot._dismiss_workspace_modal.assert_called_once() bot._open_messages_page.assert_not_called() def test_stable_modal_gets_one_escape_and_never_a_guessed_click(self): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot._internal_blocker_stable_key = "" bot._internal_blocker_stable_count = 0 bot._internal_blocker_action_key = "" bot._internal_blocker_action_ts = 0.0 bot._internal_blocker_false_positives = {} bot._last_internal_blocker_signature = "" bot._last_internal_blocker_ts = 0.0 bot._active_session_fp = None bot.hwnd = 123 bot._capture_full_window = mock.Mock(return_value=np.zeros((120, 180, 4), np.uint8)) bot._message_nav_selected = mock.Mock(return_value=True) bot._ui_guard_surface_signature = mock.Mock(return_value="same-modal") bot.wait_for_mouse_idle = mock.Mock(return_value=True) bot._begin_bot_mouse = mock.Mock() bot._end_bot_mouse = mock.Mock() bot._blocker_area_changed = mock.Mock(return_value=True) candidate = (90, 30) with ( mock.patch("wechat_bot.looks_like_security_verification", return_value=False), mock.patch("wechat_bot.looks_like_modal_scrim", return_value=True), mock.patch( "wechat_bot.find_blocking_modal_close", side_effect=[candidate, candidate, None, candidate], ), mock.patch("wechat_bot.safe_set_foreground", return_value=True), mock.patch("wechat_bot.win32gui.GetForegroundWindow", return_value=123), mock.patch("wechat_bot.pyautogui.press") as press, mock.patch("wechat_bot.pyautogui.click") as click, mock.patch("wechat_bot.time.sleep"), ): self.assertTrue(bot._dismiss_internal_blocker("测试")) press.assert_not_called() self.assertTrue(bot._dismiss_internal_blocker("测试")) press.assert_called_once_with("esc") self.assertTrue(bot._dismiss_internal_blocker("测试")) press.assert_called_once_with("esc") click.assert_not_called() class CallWindowProtectionTest(TestCase): def test_send_is_paused_while_a_video_window_is_visible(self): bot = WeChatBot.__new__(WeChatBot) bot._strict_visual_actions = True bot._set_task_stage = mock.Mock() bot._visible_call_window_title = mock.Mock(return_value="企业微信视频通话") fp = b"v" * 40 self.assertFalse(bot.send_reply("稍后回复", expected_fp=fp)) self.assertIn("通话", bot.last_send_failure_reason()) bot._set_task_stage.assert_any_call( fp, "call_paused", detail="检测到通话窗口：企业微信视频通话", ) class ConstrainedAIProtocolTest(TestCase): def test_compound_non_message_modal_can_only_close_the_modal(self): from ai_chat import _parse_ui_guard_decision decision = _parse_ui_guard_decision( '{"state":"non_message_modal","action":"close_modal",' '"confidence":0.96,"reason":"离职继承页有弹窗"}' ) self.assertEqual(decision["state"], "non_message_modal") self.assertEqual(decision["action"], "close_modal") def test_ai_cannot_smuggle_a_click_or_send_action_into_page_recovery(self): from ai_chat import _parse_ui_guard_decision decision = _parse_ui_guard_decision( '{"state":"non_message_page","action":"click_customer",' '"confidence":1.0,"reason":"随便点"}' ) self.assertEqual(decision["action"], "none") def test_low_confidence_row_vision_is_rechecked_by_text_model(self): from ai_chat import classify_wecom_session_row with ( mock.patch("ai_chat._provider_type", return_value="openai"), mock.patch( "ai_chat._call_vision_classifier", return_value=( '{"kind":"system_entry","reply_capable":false,' '"confidence":0.55,"reason":"画面模糊"}' ), ), mock.patch( "ai_chat._chat_completion", return_value={ "content": ( '{"kind":"system_entry","reply_capable":false,' '"confidence":0.94,"reason":"OCR为审批入口"}' ) }, ), ): decision = classify_wecom_session_row( b"row-image", recognized_name="审批", preview_text="你有1条审批待处理", ) self.assertEqual(decision["kind"], "system_entry") self.assertEqual(decision["confidence"], 0.94) if __name__ == "__main__": import unittest unittest.main()