94 lines
3.8 KiB
Python
94 lines
3.8 KiB
Python
"""Lost final-draw events must not strand an isolated incoming message."""
|
|
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from test_wechat_local_ocr import MODULE
|
|
|
|
|
|
class ReconcileTests(unittest.TestCase):
|
|
def test_final_drawing_without_damage_is_captured_and_duplicates_are_suppressed(self):
|
|
class Monitor:
|
|
window_id = 42
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
pass
|
|
|
|
def wait_for_message_change(self):
|
|
return next(self.changes)
|
|
|
|
changes = iter((True, False, False))
|
|
|
|
buffer = MODULE.CaptureFrameBuffer(5)
|
|
outcomes = []
|
|
offer = buffer.offer
|
|
|
|
def observe(frame):
|
|
outcome = offer(frame)
|
|
outcomes.append(outcome)
|
|
if len(outcomes) == 3:
|
|
buffer.close()
|
|
return outcome
|
|
|
|
images = [MODULE.Image.new("RGB", (2, 2), color) for color in ("white", "white", "black")]
|
|
recorder = MODULE.Recorder.__new__(MODULE.Recorder)
|
|
with mock.patch.object(MODULE, "XDamageMonitor", return_value=Monitor()), \
|
|
mock.patch.object(MODULE, "window_id", return_value="42"), \
|
|
mock.patch.object(MODULE, "capture_image", side_effect=images) as captures, \
|
|
mock.patch.object(MODULE, "message_view_at_loaded_bottom", return_value=True), \
|
|
mock.patch.object(MODULE, "message_image", side_effect=lambda image: image), \
|
|
mock.patch.object(MODULE, "incoming_frame_hashes", side_effect=[
|
|
("old-layout", "old-visual"), ("old-layout", "old-visual"), ("new-layout", "new-visual")]), \
|
|
mock.patch.object(buffer, "offer", side_effect=observe), \
|
|
mock.patch.object(MODULE, "telemetry_event") as telemetry:
|
|
recorder.capture_damage_frames(buffer)
|
|
self.assertEqual(captures.call_count, 3)
|
|
self.assertEqual(outcomes, ["enqueued", "duplicate", "enqueued"])
|
|
self.assertEqual(buffer.get().visual_hash, "old-visual")
|
|
self.assertEqual(buffer.get().visual_hash, "new-visual")
|
|
self.assertEqual([call.kwargs["source"] for call in telemetry.call_args_list
|
|
if call.args == ("ocr.capture",)], ["damage", "reconcile", "reconcile"])
|
|
self.assertLessEqual(MODULE.DAMAGE_HEARTBEAT_SECONDS, 2)
|
|
|
|
def test_reconciliation_does_not_bypass_window_identity_check(self):
|
|
class Monitor:
|
|
window_id = 42
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
pass
|
|
|
|
def wait_for_message_change(self):
|
|
return False
|
|
|
|
buffer = MODULE.CaptureFrameBuffer(2)
|
|
recorder = MODULE.Recorder.__new__(MODULE.Recorder)
|
|
with mock.patch.object(MODULE, "XDamageMonitor", return_value=Monitor()), \
|
|
mock.patch.object(MODULE, "window_id", return_value="43"), \
|
|
mock.patch.object(MODULE, "capture_image") as capture:
|
|
recorder.capture_damage_frames(buffer)
|
|
capture.assert_not_called()
|
|
with self.assertRaises(MODULE.CaptureProducerError):
|
|
buffer.get()
|
|
|
|
def test_xlib_buffered_event_is_drained_before_waiting_for_socket(self):
|
|
monitor = MODULE.XDamageMonitor.__new__(MODULE.XDamageMonitor)
|
|
monitor._display = object()
|
|
monitor._descriptor = 42
|
|
monitor._x11 = mock.Mock()
|
|
monitor._x11.XPending.side_effect = [1, 0]
|
|
with mock.patch.object(monitor, "_drain_events", return_value=True) as drain, \
|
|
mock.patch.object(MODULE.select, "select", return_value=([], [], [])) as select:
|
|
self.assertTrue(monitor.wait_for_message_change())
|
|
drain.assert_called_once()
|
|
select.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|