#!/usr/bin/env python3 import importlib.util import subprocess import sys import tempfile import unittest from contextlib import nullcontext from datetime import datetime, timezone from pathlib import Path from unittest import mock PROJECT_ROOT = Path(__file__).resolve().parents[1] MODULE_PATH = PROJECT_ROOT / "wechat_local_ocr.py" SPEC = importlib.util.spec_from_file_location("wechat_local_ocr", MODULE_PATH) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = MODULE SPEC.loader.exec_module(MODULE) class TelemetryTests(unittest.TestCase): def test_telemetry_hashes_identifiers_and_excludes_unknown_fields(self): with self.assertLogs(level="INFO") as captured: MODULE.telemetry_event( "ocr.scan", event_id="private-event", group_id="private-group", elapsed_ms=12.5, count=2, message_body="must-not-appear", ) output = "\n".join(captured.output) self.assertIn('"name":"ocr.scan"', output) self.assertNotIn("private-event", output) self.assertNotIn("private-group", output) self.assertNotIn("message_body", output) self.assertNotIn("must-not-appear", output) class LayoutTests(unittest.TestCase): def test_sender_and_body_are_paired(self): lines = [ MODULE.OcrLine("Alice", 0.99, 62, 150, 105, 173), MODULE.OcrLine("@JWeBot test", 0.98, 77, 179, 164, 201), MODULE.OcrLine("03:19", 0.99, 399, 229, 441, 246), MODULE.OcrLine("self reply", 0.99, 609, 275, 751, 296), ] bubbles = [MODULE.Bubble(64, 172, 107, 34, 3461)] messages = MODULE.messages_from_layout(lines, bubbles) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].sender, "Alice") self.assertEqual(messages[0].text, "@JWeBot test") self.assertEqual(MODULE.trigger_question(messages[0].text), "test") def test_trigger_allows_ocr_to_remove_mention_space(self): self.assertEqual( MODULE.trigger_question("@JWeBot现在几点了"), "现在几点了", ) def test_unknown_sender_is_preserved_but_low_confidence(self): lines = [MODULE.OcrLine("hello", 0.99, 77, 179, 164, 201)] bubbles = [MODULE.Bubble(64, 172, 107, 34, 3461)] messages = MODULE.messages_from_layout(lines, bubbles) self.assertEqual(messages[0].sender, "unknown") self.assertEqual(messages[0].confidence, 0.0) def test_single_line_wide_incoming_bubble_is_eligible(self): bubble = MODULE.Bubble(66, 242, 573, 37, 20939) self.assertTrue(MODULE.is_incoming_bubble(bubble)) def test_oversized_or_right_aligned_bubbles_remain_ineligible(self): self.assertFalse( MODULE.is_incoming_bubble( MODULE.Bubble(66, 242, MODULE.INCOMING_MAX_WIDTH + 1, 37, 30000) ) ) self.assertFalse( MODULE.is_incoming_bubble( MODULE.Bubble(500, 242, 573, 37, 20939) ) ) class MonitorTests(unittest.TestCase): def test_message_damage_intersection(self): self.assertTrue( MODULE.damage_intersects_message_box(371, 549, 96, 96) ) def test_input_cursor_damage_is_ignored(self): self.assertFalse( MODULE.damage_intersects_message_box(321, 742, 8, 22) ) def test_target_chat_unread_damage_is_monitored(self): self.assertTrue( MODULE.damage_intersects_message_box(95, 88, 24, 24) ) def test_zero_area_damage_is_ignored(self): self.assertFalse( MODULE.damage_intersects_message_box(400, 200, 0, 10) ) class CaptureBufferTests(unittest.TestCase): def frame(self, sequence, visual_hash=None): return MODULE.CapturedFrame( image=MODULE.Image.new("RGB", (2, 2), "white"), layout_hash=f"layout-{sequence}", visual_hash=visual_hash or f"visual-{sequence}", observed_at=f"2026-10-10T00:00:{sequence:02d}+08:00", captured_monotonic=float(sequence), sequence=sequence, latest_synced=True, ) def test_frames_remain_fifo_when_capacity_is_available(self): buffer = MODULE.CaptureFrameBuffer(3) for sequence in (1, 2, 3): self.assertEqual(buffer.offer(self.frame(sequence)), "enqueued") self.assertEqual([buffer.get().sequence for _ in range(3)], [1, 2, 3]) def test_overflow_discards_backlog_and_keeps_latest_frame(self): buffer = MODULE.CaptureFrameBuffer(3) for sequence in (1, 2, 3): buffer.offer(self.frame(sequence)) self.assertEqual(buffer.offer(self.frame(4)), "reset_to_latest") self.assertEqual(buffer.get().sequence, 4) self.assertEqual(buffer.stats()["coalesced"], 1) self.assertEqual(buffer.stats()["stale_discarded"], 3) self.assertEqual(buffer.stats()["max_depth"], 3) def test_get_latest_discards_queued_stale_frames(self): buffer = MODULE.CaptureFrameBuffer(3) for sequence in (1, 2, 3): buffer.offer(self.frame(sequence)) self.assertEqual(buffer.get_latest().sequence, 3) self.assertEqual(buffer.stats()["stale_discarded"], 2) self.assertEqual(buffer.stats()["depth"], 0) def test_adjacent_duplicate_visual_frame_is_not_queued(self): buffer = MODULE.CaptureFrameBuffer(2) self.assertEqual(buffer.offer(self.frame(1, "same")), "enqueued") self.assertEqual(buffer.offer(self.frame(2, "same")), "duplicate") self.assertEqual(buffer.get().sequence, 1) self.assertEqual(buffer.stats()["duplicates"], 1) def test_graceful_close_drains_frames_then_stops(self): buffer = MODULE.CaptureFrameBuffer(2) buffer.offer(self.frame(1)) buffer.close() self.assertEqual(buffer.get().sequence, 1) with self.assertRaises(MODULE.CaptureBufferClosed): buffer.get() def test_producer_error_is_propagated_without_draining_stale_frames(self): buffer = MODULE.CaptureFrameBuffer(2) buffer.offer(self.frame(1)) cause = RuntimeError("capture failed") buffer.close(cause) with self.assertRaises(MODULE.CaptureProducerError) as raised: buffer.get() self.assertIs(raised.exception.__cause__, cause) def test_capture_producer_reports_monitor_failure(self): class BrokenMonitor: window_id = 1 def __enter__(self): return self def __exit__(self, *_): return None def wait_for_message_change(self): raise RuntimeError("damage stream failed") recorder = MODULE.Recorder.__new__(MODULE.Recorder) buffer = MODULE.CaptureFrameBuffer(2) with mock.patch.object(MODULE, "XDamageMonitor", return_value=BrokenMonitor()): recorder.capture_damage_frames(buffer) with self.assertRaises(MODULE.CaptureProducerError) as raised: buffer.get() self.assertIsInstance(raised.exception.__cause__, RuntimeError) def test_capture_producer_reports_follow_latest_steps(self): class OneFrameMonitor: window_id = 42 def __enter__(self): return self def __exit__(self, *_): return None def wait_for_message_change(self): return True class OneFrameBuffer: capacity = 1 closed = False def offer(self, _frame): self.closed = True return "closed" def stats(self): return { "depth": 0, "coalesced": 0, "stale_discarded": 0, "duplicates": 0, } def close(self, _error=None): self.closed = True recorder = MODULE.Recorder.__new__(MODULE.Recorder) first = MODULE.Image.new( "RGB", (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), "white", ) second = first.copy() with mock.patch.object( MODULE, "XDamageMonitor", return_value=OneFrameMonitor(), ), mock.patch.object( MODULE, "window_id", return_value="42", ), mock.patch.object( MODULE, "capture_image", side_effect=[first, second], ), mock.patch.object( MODULE, "message_view_at_loaded_bottom", side_effect=[False, True], ), mock.patch.object( MODULE, "follow_latest_messages", return_value=True, ), mock.patch.object( MODULE, "incoming_frame_hashes", return_value=("layout", "visual"), ), mock.patch.object( MODULE, "telemetry_event", ) as telemetry: recorder.capture_damage_frames(OneFrameBuffer()) capture_steps = [ call for call in telemetry.call_args_list if call.args == ("ocr.capture.step",) ] self.assertEqual( [call.kwargs["phase"] for call in capture_steps], ["initial_capture", "follow_latest", "recapture"], ) self.assertEqual(capture_steps[1].kwargs["outcome"], "followed") def test_visual_hash_changes_when_text_pixels_change(self): bubble = MODULE.Bubble(50, 50, 80, 30, 2000) mask = MODULE.Image.new("L", (160, 120), 0) first = MODULE.Image.new("RGB", (160, 120), "white") second = first.copy() second.putpixel((70, 65), (0, 0, 0)) with mock.patch.object( MODULE, "incoming_mask_and_bubbles", return_value=(mask, [bubble]), ): first_layout, first_visual = MODULE.incoming_frame_hashes(first) second_layout, second_visual = MODULE.incoming_frame_hashes(second) self.assertEqual(first_layout, second_layout) self.assertNotEqual(first_visual, second_visual) class GuiCoordinationTests(unittest.TestCase): def scrollbar_image(self, start, end): image = MODULE.Image.new( "RGB", (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), "white", ) for y in range(start, end + 1): image.putpixel( (MODULE.FOLLOW_LATEST_MOUSE_X, y), MODULE.SCROLLBAR_THUMB_COLOR, ) return image def test_main_window_is_selected_by_exact_title(self): search = subprocess.CompletedProcess([], 0, "11\n22\n", "") child = subprocess.CompletedProcess([], 0, "Search chat history\n", "") main = subprocess.CompletedProcess([], 0, "Weixin\n", "") with mock.patch.object( MODULE, "as_bridge_user", side_effect=[search, child, main], ): self.assertEqual(MODULE.window_id(), "22") def test_geometry_drift_is_restored_by_maximizing(self): completed = subprocess.CompletedProcess([], 0, "", "") with mock.patch.object( MODULE, "window_geometry", side_effect=[ (900, 700), (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), ], ), mock.patch.object( MODULE, "as_bridge_user", return_value=completed, ) as bridge, mock.patch.object( MODULE.time, "sleep", ): geometry = MODULE.ensure_maximized_window("42") self.assertEqual( geometry, (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), ) command = bridge.call_args.args[0] self.assertIn("wmctrl", command[0]) self.assertIn("add,maximized_vert,maximized_horz", command) def test_expected_geometry_does_not_touch_window_state(self): with mock.patch.object( MODULE, "window_geometry", return_value=(MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), ), mock.patch.object( MODULE, "as_bridge_user", ) as bridge: geometry = MODULE.ensure_maximized_window("42") self.assertEqual( geometry, (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), ) bridge.assert_not_called() def test_capture_reuses_a_verified_window_identifier(self): payload = MODULE.io.BytesIO() MODULE.Image.new( "RGB", (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), "white", ).save(payload, format="PNG") completed = subprocess.CompletedProcess( [], 0, payload.getvalue(), b"", ) with mock.patch.object( MODULE, "ensure_directories", ), mock.patch.object( MODULE, "window_id", ) as find_window, mock.patch.object( MODULE, "ensure_maximized_window", return_value=(MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), ) as maximize, mock.patch.object( MODULE, "as_bridge_user_bytes", return_value=completed, ) as bridge: image = MODULE.capture_image("42") self.assertEqual( image.size, (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), ) find_window.assert_not_called() maximize.assert_called_once_with("42") command = bridge.call_args.args[0] self.assertEqual(command[command.index("--window") + 1], "42") def test_window_bounds_preserve_negative_window_position(self): completed = subprocess.CompletedProcess( [], 0, "WINDOW=42\nX=-12\nY=27\nWIDTH=1440\nHEIGHT=873\nSCREEN=0\n", "", ) with mock.patch.object( MODULE, "as_bridge_user", return_value=completed, ): self.assertEqual( MODULE.window_bounds("42"), (-12, 27, 1440, 873), ) def test_title_fingerprint_uses_direct_region_capture(self): payload = MODULE.io.BytesIO() MODULE.Image.new("RGB", (600, 47), (20, 30, 40)).save( payload, format="PNG", ) completed = subprocess.CompletedProcess( [], 0, payload.getvalue(), b"", ) with mock.patch.object( MODULE, "window_id", return_value="42", ), mock.patch.object( MODULE, "window_bounds", return_value=(0, 27, 1440, 873), ), mock.patch.object( MODULE, "as_bridge_user_bytes", return_value=completed, ) as bridge: signature = MODULE.capture_title_fingerprint() self.assertEqual(len(signature), 64) command = bridge.call_args.args[0] self.assertIn("--autoselect", command) self.assertIn("300,60,600,47", command) self.assertNotIn("--window", command) def test_bottom_scrollbar_thumb_is_detected(self): image = self.scrollbar_image( MODULE.MESSAGE_BOX[3] - 40, MODULE.MESSAGE_BOX[3] - 4, ) self.assertTrue(MODULE.message_view_at_loaded_bottom(image)) def test_historical_scrollbar_thumb_is_not_at_bottom(self): image = self.scrollbar_image(170, 317) self.assertFalse(MODULE.message_view_at_loaded_bottom(image)) def test_hidden_scrollbar_is_not_assumed_to_be_at_bottom(self): image = MODULE.Image.new( "RGB", (MODULE.EXPECTED_WIDTH, MODULE.EXPECTED_HEIGHT), "white", ) self.assertFalse(MODULE.message_view_at_loaded_bottom(image)) def test_follow_latest_scrolls_message_pane_when_lock_is_available(self): completed = subprocess.CompletedProcess([], 0, "", "") with mock.patch.object( MODULE, "gui_interaction_lock", return_value=nullcontext(True), ), mock.patch.object( MODULE, "as_bridge_user", return_value=completed, ) as bridge, mock.patch.object( MODULE.time, "sleep", ): self.assertTrue(MODULE.follow_latest_messages("42")) command = bridge.call_args.args[0] self.assertIn("windowactivate", command) self.assertIn("mousemove", command) self.assertIn(str(MODULE.FOLLOW_LATEST_MOUSE_X), command) self.assertIn(str(MODULE.FOLLOW_LATEST_MOUSE_Y), command) self.assertIn("click", command) self.assertIn(str(MODULE.FOLLOW_LATEST_SCROLL_STEPS), command) delay_index = command.index("--delay") self.assertEqual( command[delay_index + 1], str(MODULE.FOLLOW_LATEST_SCROLL_DELAY_MS), ) self.assertEqual(MODULE.FOLLOW_LATEST_SCROLL_DELAY_MS, 1) self.assertEqual(command[-1], "5") self.assertNotIn("ctrl+End", command) def test_follow_latest_skips_when_responder_holds_lock(self): with mock.patch.object( MODULE, "gui_interaction_lock", return_value=nullcontext(False), ), mock.patch.object( MODULE, "as_bridge_user", ) as bridge: self.assertFalse(MODULE.follow_latest_messages("42")) bridge.assert_not_called() class OcrFallbackTests(unittest.TestCase): def test_low_confidence_line_uses_detection_fallback(self): engine = MODULE.OcrEngine.__new__(MODULE.OcrEngine) bubble = MODULE.Bubble(50, 50, 132, 68, 5000) image = MODULE.Image.new("RGB", (300, 180), "white") with mock.patch.object( engine, "recognize_line", side_effect=[("Alice", 0.95), ("broken", 0.36)], ), mock.patch.object( engine, "recognize_block", return_value=[("@JWeBot recovered", 0.98)], ) as fallback, mock.patch.object( MODULE, "text_line_regions", return_value=[(0, 20)], ), mock.patch.object( MODULE, "run", return_value=subprocess.CompletedProcess([], 0, "", ""), ): messages, _ = engine.recognize_messages(image, [bubble]) fallback.assert_called_once() self.assertEqual(len(messages), 1) self.assertEqual(messages[0].text, "@JWeBot recovered") self.assertEqual(messages[0].body_confidence, 0.98) self.assertEqual(len(messages[0].visual_signature), 64) class BubbleSequenceSignatureTests(unittest.TestCase): def test_signature_ignores_small_color_and_antialias_drift(self): first_sender = MODULE.Image.new("RGB", (12, 6), (238, 238, 240)) first_body = MODULE.Image.new("RGB", (20, 10), (238, 238, 240)) second_sender = MODULE.Image.new("RGB", (12, 6), (242, 242, 244)) second_body = MODULE.Image.new("RGB", (20, 10), (242, 242, 244)) first_sender.putpixel((2, 2), (20, 20, 20)) first_body.putpixel((4, 4), (30, 30, 30)) second_sender.putpixel((2, 2), (35, 35, 35)) second_body.putpixel((4, 4), (45, 45, 45)) first = MODULE.bubble_sequence_signature(first_sender, first_body) second = MODULE.bubble_sequence_signature(second_sender, second_body) self.assertEqual(first, second) self.assertNotEqual( MODULE.OcrEngine._cache_key(first_sender, first_body), MODULE.OcrEngine._cache_key(second_sender, second_body), ) def test_signature_changes_when_text_structure_changes(self): sender = MODULE.Image.new("RGB", (12, 6), "white") first_body = MODULE.Image.new("RGB", (20, 10), "white") second_body = first_body.copy() first_body.putpixel((4, 4), (0, 0, 0)) second_body.putpixel((5, 4), (0, 0, 0)) self.assertNotEqual( MODULE.bubble_sequence_signature(sender, first_body), MODULE.bubble_sequence_signature(sender, second_body), ) class OcrCacheTests(unittest.TestCase): @staticmethod def message(y): return MODULE.IncomingMessage( "Alice", "@JWeBot cached", 0.98, 0.99, 0.98, y, ) def test_unchanged_bubble_reuses_ocr_without_dropping_occurrences(self): engine = MODULE.OcrEngine.__new__(MODULE.OcrEngine) image = MODULE.Image.new("RGB", (300, 240), "white") bubbles = [ MODULE.Bubble(50, 60, 132, 40, 5000), MODULE.Bubble(50, 120, 132, 40, 5000), ] with mock.patch.object( engine, "_recognize_bubble", return_value=self.message(60), ) as recognize: messages, _ = engine.recognize_messages(image, bubbles) self.assertEqual(recognize.call_count, 1) self.assertEqual([message.y for message in messages], [60, 120]) self.assertEqual(engine.last_cache_hits, 1) self.assertEqual(engine.last_cache_misses, 1) def test_changed_bubble_pixels_force_fresh_ocr(self): engine = MODULE.OcrEngine.__new__(MODULE.OcrEngine) bubble = MODULE.Bubble(50, 60, 132, 40, 5000) first = MODULE.Image.new("RGB", (300, 180), "white") second = first.copy() second.putpixel((70, 75), (0, 0, 0)) with mock.patch.object( engine, "_recognize_bubble", return_value=self.message(60), ) as recognize: engine.recognize_messages(first, [bubble]) engine.recognize_messages(second, [bubble]) self.assertEqual(recognize.call_count, 2) self.assertEqual(engine.last_cache_hits, 0) self.assertEqual(engine.last_cache_misses, 1) def test_periodic_refresh_revalidates_cached_bubbles(self): engine = MODULE.OcrEngine.__new__(MODULE.OcrEngine) image = MODULE.Image.new("RGB", (300, 180), "white") bubble = MODULE.Bubble(50, 60, 132, 40, 5000) with mock.patch.object( engine, "_recognize_bubble", return_value=self.message(60), ) as recognize: engine.recognize_messages(image, [bubble]) engine._scan_count = MODULE.OCR_CACHE_REFRESH_SCANS - 1 engine.recognize_messages(image, [bubble]) self.assertEqual(recognize.call_count, 2) self.assertTrue(engine.last_cache_refresh) def test_warmup_initializes_recognition_and_detection_paths(self): engine = MODULE.OcrEngine.__new__(MODULE.OcrEngine) with mock.patch.object( engine, "recognize_line", return_value=("", 0.0), ) as line, mock.patch.object( engine, "recognize_block", return_value=[], ) as block: engine.warmup() line.assert_called_once() block.assert_called_once() class SequenceTests(unittest.TestCase): def message(self, sender, text, y, visual_signature=""): return MODULE.IncomingMessage( sender, text, 0.99, 0.99, 0.99, y, visual_signature, ) def test_sliding_overlap_returns_only_new_suffix(self): first = self.message("A", "one", 10) second = self.message("B", "two", 20) third = self.message("A", "three", 30) new, current, overlap, resynced = MODULE.new_sequence_suffix( [first.signature, second.signature], [second, third], ) self.assertEqual(new, [third]) self.assertEqual(current, [second.signature, third.signature]) self.assertEqual(overlap, 1) self.assertFalse(resynced) def test_identical_repeated_message_is_a_new_occurrence(self): repeated = self.message("A", "same", 10) new, _, overlap, resynced = MODULE.new_sequence_suffix( [repeated.signature], [repeated, repeated], ) self.assertEqual(new, [repeated]) self.assertEqual(overlap, 1) self.assertFalse(resynced) def test_signature_ignores_ocr_case_drift(self): upper = self.message("USERA", "Hello JWEBOT", 10) mixed = self.message("uSeRa", "hello jwebot", 10) self.assertEqual(upper.signature, mixed.signature) def test_visual_signature_survives_ocr_text_drift(self): first = self.message("Alice", "broken OCR", 10, "visual-a") corrected = self.message( "Alice", "@JWeBot /steam", 10, "visual-a", ) self.assertEqual(first.signature, corrected.signature) def test_identical_text_with_distinct_visual_occurrences_stays_distinct(self): first = self.message("Alice", "@JWeBot /steam", 10, "visual-a") second = self.message("Alice", "@JWeBot /steam", 20, "visual-b") self.assertNotEqual(first.signature, second.signature) def test_ordered_fallback_keeps_new_suffix_after_ocr_omission(self): first = self.message("A", "one", 10) omitted = self.message("B", "two", 20) third = self.message("C", "three", 30) extra_top = self.message("X", "older", 5) latest = self.message("D", "@JWeBot latest", 40) new, _, overlap, resynced = MODULE.new_sequence_suffix( [first.signature, omitted.signature, third.signature], [extra_top, first, third, latest], ) self.assertEqual(new, [latest]) self.assertEqual(overlap, 2) self.assertTrue(resynced) def test_no_overlap_fails_closed(self): old = self.message("A", "old", 10) new_message = self.message("B", "new", 20) new, _, overlap, resynced = MODULE.new_sequence_suffix( [old.signature], [new_message], ) self.assertEqual(new, []) self.assertEqual(overlap, 0) self.assertTrue(resynced) def test_no_overlap_recovers_one_unseen_trigger_at_the_bottom(self): old = self.message("A", "old", 10, "visual-old") trigger = self.message( "B", "@JWeBot test", 20, "visual-new-trigger", ) new, _, overlap, resynced = MODULE.new_sequence_suffix( [old.signature], [trigger], ) self.assertEqual(new, [trigger]) self.assertEqual(overlap, 0) self.assertTrue(resynced) def test_no_overlap_does_not_replay_a_recent_trigger(self): old = self.message("A", "old", 10, "visual-old") trigger = self.message( "B", "@JWeBot test", 20, "visual-seen-trigger", ) new, _, overlap, resynced = MODULE.new_sequence_suffix( [old.signature], [trigger], recent_trigger_signatures=[trigger.signature], ) self.assertEqual(new, []) self.assertEqual(overlap, 0) self.assertTrue(resynced) def test_no_overlap_recovery_is_disabled_for_startup_baseline(self): old = self.message("A", "old", 10, "visual-old") trigger = self.message( "B", "@JWeBot test", 20, "visual-new-trigger", ) new, _, overlap, resynced = MODULE.new_sequence_suffix( [old.signature], [trigger], recover_unseen_trigger=False, ) self.assertEqual(new, []) self.assertEqual(overlap, 0) self.assertTrue(resynced) class RecorderSignatureMigrationTests(unittest.TestCase): def test_legacy_text_signatures_reset_to_a_safe_visual_baseline(self): engine = mock.Mock() with mock.patch.object( MODULE, "ensure_directories", ), mock.patch.object( MODULE, "OcrEngine", return_value=engine, ), mock.patch.object( MODULE, "read_json", return_value={ "initialized": True, "last_visual_hash": "old-frame", "visible_signatures": ["old-text-signature"], "recent_trigger_signatures": ["old-trigger-signature"], }, ): recorder = MODULE.Recorder(queue_enabled=True) self.assertFalse(recorder.state["initialized"]) self.assertEqual(recorder.state["last_visual_hash"], "") self.assertEqual(recorder.state["visible_signatures"], []) self.assertEqual(recorder.state["recent_trigger_signatures"], []) self.assertFalse(recorder.no_overlap_recovery_ready) engine.warmup.assert_called_once() class StorageTests(unittest.TestCase): def test_elapsed_since_iso_is_timezone_aware_and_bounded(self): now = datetime(2026, 10, 14, 12, 0, 1, tzinfo=timezone.utc) self.assertEqual( MODULE.elapsed_since_iso_ms( "2026-10-14T12:00:00+00:00", now=now, ), 1000.0, ) self.assertEqual( MODULE.elapsed_since_iso_ms( "2026-10-14T12:00:02+00:00", now=now, ), 0.0, ) self.assertIsNone(MODULE.elapsed_since_iso_ms("not-a-timestamp")) def test_jsonl_contains_text_only_event_fields(self): message = MODULE.IncomingMessage( "Alice", "@JWeBot hello", 0.98, 0.99, 0.98, 10, ) event = MODULE.history_event( message, observed_at="2026-10-10T03:23:38+08:00", group_fingerprint="group", ordinal=0, ) self.assertEqual(event["sender"], "Alice") self.assertEqual(event["text"], "@JWeBot hello") self.assertTrue(event["mention_bot"]) self.assertNotIn("image", event) self.assertNotIn("screenshot", event) def test_atomic_json_roundtrip(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "state.json" MODULE.write_json(path, {"ok": True}) self.assertEqual(MODULE.read_json(path), {"ok": True}) def test_queued_event_records_its_queue_entry_time(self): with tempfile.TemporaryDirectory() as directory: queue = Path(directory) incoming = { "event_id": "event-1", "observed_at": "2026-10-14T11:59:59+00:00", } with mock.patch.object( MODULE, "QUEUE_DIR", queue, ), mock.patch.object( MODULE, "now_iso", return_value="2026-10-14T12:00:00+00:00", ): MODULE.QUEUE_WAKE_EVENT.clear() MODULE.queue_event(incoming) stored = MODULE.read_json(queue / "event-1.json") self.assertEqual( stored["queued_at"], "2026-10-14T12:00:00+00:00", ) self.assertTrue(MODULE.QUEUE_WAKE_EVENT.is_set()) MODULE.QUEUE_WAKE_EVENT.clear() class RoutingTests(unittest.TestCase): def event(self): return { "event_id": "event-route", "sender": "Alice", "text": "@JWeBot test", "group_fingerprint": "fixed-group", } def test_router_decision_accepts_direct_reply(self): decision = MODULE.router_decision( '{"route":"reply","text":"收到，测试正常。"}' ) self.assertEqual( decision, {"route": "reply", "text": "收到，测试正常喵"}, ) def test_router_decision_removes_english_full_stop(self): decision = MODULE.router_decision( '{"route":"reply","text":"你急了嘎嘎嘎."}' ) self.assertEqual( decision, {"route": "reply", "text": "你急了嘎嘎嘎喵"}, ) def test_router_decision_does_not_duplicate_meow_suffix(self): decision = MODULE.router_decision( '{"route":"reply","text":"收到喵～"}' ) self.assertEqual( decision, {"route": "reply", "text": "收到喵"}, ) def test_router_decision_rejects_main_route(self): with self.assertRaises(ValueError): MODULE.router_decision( '```json\n{"route":"main","task":"use browser"}\n```' ) def test_direct_reply_does_not_start_main_agent(self): with mock.patch.object( MODULE, "invoke_openclaw_agent", return_value='{"route":"reply","text":"fast reply"}', ) as invoke: reply = MODULE.ask_openclaw(self.event()) self.assertEqual(reply, "fast reply喵") self.assertEqual(invoke.call_count, 1) self.assertEqual(invoke.call_args.kwargs["agent"], MODULE.ROUTER_AGENT) def test_tool_escalation_fails_closed_without_main_agent(self): with mock.patch.object( MODULE, "invoke_openclaw_agent", return_value='{"route":"main","task":"use Computer Use"}', ) as invoke: reply = MODULE.ask_openclaw(self.event()) self.assertEqual(reply, "JWeBot Error: Invalid router reply") self.assertEqual(invoke.call_count, 1) self.assertEqual(invoke.call_args.kwargs["agent"], MODULE.ROUTER_AGENT) def test_router_invocation_failure_returns_visible_error(self): with mock.patch.object( MODULE, "invoke_openclaw_agent", side_effect=RuntimeError("router unavailable"), ) as invoke: reply = MODULE.ask_openclaw(self.event()) self.assertEqual(reply, "JWeBot Error: Router unavailable") self.assertEqual(invoke.call_count, 1) self.assertEqual(invoke.call_args.kwargs["agent"], MODULE.ROUTER_AGENT) def test_router_error_message_classifies_gateway_failure(self): result = subprocess.CompletedProcess( [], 255, "", "ssh: connect to host: Host is down", ) self.assertEqual( MODULE.router_error_message(result=result), "JWeBot Error: Gateway unreachable", ) def test_router_error_message_classifies_timeout(self): self.assertEqual( MODULE.router_error_message( exc=RuntimeError("operation timed out") ), "JWeBot Error: Router timeout", ) def test_router_can_be_disabled_without_fallback(self): with mock.patch.dict( MODULE.os.environ, {"WECHAT_ROUTER_ENABLED": "0"}, ), mock.patch.object( MODULE, "invoke_openclaw_agent", ) as invoke: reply = MODULE.ask_openclaw(self.event()) self.assertEqual(reply, "NO_REPLY") invoke.assert_not_called() class ResponderTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() root = Path(self.temporary.name) self.paths = { "STATE_ROOT": root, "HISTORY_DIR": root / "history", "QUEUE_DIR": root / "queue", "FAILED_DIR": root / "failed", "RECORDER_STATE": root / "recorder-state.json", "RESPONDER_STATE": root / "responder-state.json", "GUI_LOCK_FILE": root / "gui-interaction.lock", } self.original_paths = { name: getattr(MODULE, name) for name in self.paths } for name, value in self.paths.items(): setattr(MODULE, name, value) MODULE.ensure_directories() def tearDown(self): for name, value in self.original_paths.items(): setattr(MODULE, name, value) self.temporary.cleanup() def event(self): return { "event_id": "event-1", "side": "incoming", "mention_bot": True, "sender": "Alice", "sender_confidence": 0.99, "text": "@JWeBot hello", "group_fingerprint": "fixed-group", "attempts": 0, "next_attempt_at": 0, } def queue(self, event=None): path = self.paths["QUEUE_DIR"] / "event-1.json" MODULE.write_json(path, event or self.event()) return path def test_successful_delivery_is_completed_once(self): path = self.queue() send_calls = [] def fake_send(reply, expected_group, *, before_send, after_send, event_id): self.assertEqual(reply, "reply") self.assertEqual(expected_group, "fixed-group") self.assertEqual(event_id, "event-1") before_send() send_calls.append(reply) after_send() with mock.patch.object(MODULE, "route_event", return_value="reply"), mock.patch.object( MODULE, "send_reply", side_effect=fake_send ): responder = MODULE.Responder() responder.process(path) self.assertEqual(send_calls, ["reply"]) self.assertEqual(list(self.paths["QUEUE_DIR"].iterdir()), []) self.assertEqual(responder.state["sent_total"], 1) self.assertEqual(responder.state["processed_total"], 1) def test_uncertain_delivery_is_not_retried(self): path = self.queue() def uncertain_send(reply, expected_group, *, before_send, after_send, event_id): self.assertEqual(event_id, "event-1") before_send() raise RuntimeError("lost send acknowledgement") with mock.patch.object(MODULE, "route_event", return_value="reply"), mock.patch.object( MODULE, "send_reply", side_effect=uncertain_send ): responder = MODULE.Responder() responder.process(path) self.assertEqual(list(self.paths["QUEUE_DIR"].iterdir()), []) failed = MODULE.read_json(self.paths["FAILED_DIR"] / "event-1.json") self.assertEqual(failed["delivery_outcome"], "uncertain") self.assertEqual(responder.state["uncertain_total"], 1) def test_restart_quarantines_send_started_event(self): event = self.event() event["delivery_state"] = "send_started" processing = self.paths["QUEUE_DIR"] / "event-1.processing" MODULE.write_json(processing, event) responder = MODULE.Responder() self.assertFalse(processing.exists()) self.assertEqual(list(self.paths["QUEUE_DIR"].iterdir()), []) self.assertTrue((self.paths["FAILED_DIR"] / "event-1.json").exists()) self.assertEqual(responder.state["recovered_uncertain_total"], 1) def test_group_is_rechecked_after_drafting(self): draft = subprocess.CompletedProcess( [], 0, ( '{"confirmation_token":"token","timings":[' '{"phase":"draft.total","elapsed_ms":1.2}]}' "\n" ), "", ) cleared = subprocess.CompletedProcess([], 0, '{"state":"cleared"}\n', "") calls = [] def fake_run(argv, **kwargs): calls.append(argv) return draft if argv[1] == "draft" else cleared with mock.patch.object(MODULE, "capture_image", side_effect=[object(), object()]), mock.patch.object( MODULE, "title_fingerprint", side_effect=["fixed-group", "other-group"] ), mock.patch.object(MODULE, "run", side_effect=fake_run): with self.assertRaisesRegex(RuntimeError, "changed after draft"): MODULE.send_reply("reply", "fixed-group") self.assertEqual([call[1] for call in calls], ["draft", "clear"]) def test_send_reply_emits_bridge_step_timings(self): draft = subprocess.CompletedProcess( [], 0, ( '{"confirmation_token":"token","timings":[' '{"phase":"draft.window","elapsed_ms":2.0,"width":1440,' '"height":873},' '{"phase":"draft.confirm_poll","elapsed_ms":55.0,' '"polls":2,"pixels":120,"threshold":80,' '"timeout_ms":800,"outcome":"confirmed"}]}' "\n" ), "", ) sent = subprocess.CompletedProcess( [], 0, ( '{"state":"sent","timings":[' '{"phase":"send.return_key","elapsed_ms":52.0,' '"method":"vnc-enter","x":441,"y":786},' '{"phase":"send.return_confirm_poll","elapsed_ms":50.0,' '"polls":1,"pixels":140,"threshold":80,' '"timeout_ms":900,"outcome":"confirmed",' '"roi":"input_text","flag":true,' '"flag_wait_ms":50.0,' '"confirmed_by":"input_roi_cleared"}]}' "\n" ), "", ) def fake_run(argv, **kwargs): return draft if argv[1] == "draft" else sent with mock.patch.object( MODULE, "capture_image", side_effect=[object(), object()], ), mock.patch.object( MODULE, "title_fingerprint", side_effect=["fixed-group", "fixed-group"], ), mock.patch.object( MODULE, "run", side_effect=fake_run, ), mock.patch.object( MODULE, "telemetry_event", ) as telemetry: MODULE.send_reply("reply", "fixed-group", event_id="event-1") names = [call.args[0] for call in telemetry.call_args_list] self.assertIn("gui.send.step", names) self.assertIn("gui.bridge.step", names) self.assertIn("gui.send.total", names) bridge_phases = [ call.kwargs["phase"] for call in telemetry.call_args_list if call.args[0] == "gui.bridge.step" ] self.assertIn("draft.confirm_poll", bridge_phases) self.assertIn("send.return_confirm_poll", bridge_phases) class CaptureProcessTests(unittest.TestCase): def test_root_capture_uses_bridge_user_owned_inner_pipe(self): completed = subprocess.CompletedProcess([], 0, b"png", b"") with mock.patch.object(MODULE.os, "geteuid", return_value=0), mock.patch.object( MODULE.subprocess, "run", return_value=completed ) as run_mock: result = MODULE.as_bridge_user_bytes(["/usr/bin/scrot", "/dev/stdout"]) command = run_mock.call_args.args[0] self.assertEqual(result.stdout, b"png") self.assertEqual( command[:4], ["/usr/sbin/runuser", "-u", MODULE.BRIDGE_USER, "--"], ) self.assertIn("/usr/bin/python3", command) self.assertEqual(command[-2:], ["/usr/bin/scrot", "/dev/stdout"]) if __name__ == "__main__": unittest.main()