1232 lines
43 KiB
Python
Executable File
1232 lines
43 KiB
Python
Executable File
#!/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()
|