diff --git a/src/common/sync_manager.py b/src/common/sync_manager.py index d51bbb27..41cef988 100644 --- a/src/common/sync_manager.py +++ b/src/common/sync_manager.py @@ -101,6 +101,7 @@ class DisplaySyncManager: self._peer_chain: int = 0 self._last_heartbeat_time: float = 0.0 self._leader_width: int = 0 # set by display_controller after init + self._oversized_frame_warned: bool = False # Follower state self._follower_state = FollowerState.STANDALONE @@ -174,6 +175,10 @@ class DisplaySyncManager: continue except Exception as exc: self.logger.debug("Sync leader recv error: %s", exc) + # Brief backoff: a socket left in a bad state raises + # immediately, which would otherwise spin this thread at + # 100% CPU logging the same error. + time.sleep(0.1) def _handle_hello(self, msg: dict, sender_ip: str) -> None: hw = self._hw_config @@ -396,7 +401,7 @@ class DisplaySyncManager: data = header + arr.tobytes() if len(data) <= 65000: self._send_sock.sendto(data, (self._peer_ip, self.port)) - elif not getattr(self, '_oversized_frame_warned', False): + elif not self._oversized_frame_warned: self._oversized_frame_warned = True self.logger.warning( "Sync: frame too large for UDP (%d bytes, max 65000) — " @@ -451,41 +456,44 @@ class DisplaySyncManager: ) self.write_status_file() + def _handle_received_frame(self, img: Image.Image, sender_ip: str) -> None: + """Record a decoded leader frame and enter follower mode if needed.""" + with self._frame_lock: + self._latest_frame = img + self._last_leader_frame_time = time.time() + self._leader_ip = sender_ip + + if self._follower_state == FollowerState.STANDALONE: + self._follower_state = FollowerState.FOLLOWER + self.logger.info( + "Sync: leader active at %s — switching to follower mode", + sender_ip, + ) + self.write_status_file() + def _follower_recv_loop(self) -> None: while self._running: try: data, addr = self._recv_sock.recvfrom(65535) sender_ip = addr[0] - if data[:8] == _RAW_MAGIC or len(data) > 512: - # Frame data: prefer magic-tagged raw RGB; fall back to legacy PNG + if data[:8] == _RAW_MAGIC: + # Magic-tagged raw RGB frame — self-describing, no guessing. try: - if data[:8] == _RAW_MAGIC: - w, h = _RAW_HEADER.unpack(data[8:12]) - raw = data[12:] - img = Image.frombuffer( - "RGB", (w, h), raw, "raw", "RGB", 0, 1 - ) - else: - # Fallback: try legacy PNG - img = Image.open(io.BytesIO(data)) - img.load() - with self._frame_lock: - self._latest_frame = img - self._last_leader_frame_time = time.time() - self._leader_ip = sender_ip - - if self._follower_state == FollowerState.STANDALONE: - self._follower_state = FollowerState.FOLLOWER - self.logger.info( - "Sync: leader active at %s — switching to follower mode", - sender_ip, - ) - self.write_status_file() + w, h = _RAW_HEADER.unpack(data[8:12]) + raw = data[12:] + img = Image.frombuffer( + "RGB", (w, h), raw, "raw", "RGB", 0, 1 + ) + self._handle_received_frame(img, sender_ip) except Exception as exc: self.logger.debug("Sync: frame decode error: %s", exc) else: - # Control message + # No magic prefix: try control-message JSON, and treat a + # parse failure as a legacy (pre-magic) PNG frame. Both + # wire formats are self-describing, so no size heuristic + # is needed — a >512-byte control message used to be + # misrouted into image decode and silently dropped. try: msg = json.loads(data.decode("utf-8")) t = msg.get("t") @@ -518,12 +526,19 @@ class DisplaySyncManager: if self._on_new_cycle: self._on_new_cycle() except (json.JSONDecodeError, UnicodeDecodeError, KeyError): - pass + # Not a control message — try legacy PNG frame. + try: + img = Image.open(io.BytesIO(data)) + img.load() + self._handle_received_frame(img, sender_ip) + except Exception as exc: + self.logger.debug("Sync: frame decode error: %s", exc) except socket.timeout: continue except Exception as exc: self.logger.debug("Sync follower recv error: %s", exc) + time.sleep(0.1) def _follower_announce_loop(self) -> None: hw = self._hw_config diff --git a/test/test_sync_manager.py b/test/test_sync_manager.py new file mode 100644 index 00000000..7340c701 --- /dev/null +++ b/test/test_sync_manager.py @@ -0,0 +1,827 @@ +""" +Tests for src/common/sync_manager.py — the UDP leader/follower protocol +that synchronizes scrolling content across two LED matrix displays. + +This module had zero coverage: it only ever appeared in the suite as a +MagicMock() stand-in (test_vegas_continuous_refresh.py, +test_display_controller_vegas_tick.py), so none of its real framing, +handshake, or socket logic was exercised. + +Most tests build the manager via object.__new__() + manual attribute +assignment (the test_display_controller_vegas_tick.py bare-stub pattern) +so no real sockets open and no background threads start. Receive loops are +driven synchronously by once_then_stop(): the mocked socket call returns +one crafted packet, then flips _running False and raises socket.timeout, +so `while self._running:` exits after exactly one real iteration. + +Regression coverage for three fixed bugs: +- Both recv loops' generic `except Exception` retried with no delay, so a + socket stuck raising a non-timeout error spun the thread at 100% CPU. +- _follower_recv_loop dispatched on `data[:8] == _RAW_MAGIC or + len(data) > 512`, which routed any control message over 512 bytes into + the image decoder (dropping it) and any raw frame under 512 bytes into + the JSON parser. +- _oversized_frame_warned was read via getattr(self, ..., False) instead of + being initialized in __init__. +""" + +import io +import json +import socket +import threading +import time +from pathlib import Path +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest +from PIL import Image + +from src.common import sync_manager +from src.common.sync_manager import ( + DisplaySyncManager, + FollowerState, + LeaderState, + SyncRole, +) + + +@pytest.fixture(autouse=True) +def _isolated_status_file(tmp_path, monkeypatch): + # STATUS_FILE is a module-level fixed path under tempfile.gettempdir() — + # genuinely shared state between tests and even between processes. + monkeypatch.setattr( + sync_manager, "STATUS_FILE", str(tmp_path / "led_matrix_sync_status.json")) + + +def make_manager(role=SyncRole.STANDALONE, hw_config=None): + """Bare stub bypassing __init__'s socket/thread setup.""" + mgr = object.__new__(DisplaySyncManager) + mgr.role = role + mgr.logger = MagicMock() + mgr.port = sync_manager.SYNC_PORT + mgr._hw_config = hw_config or {"rows": 32, "cols": 64, "chain_length": 1} + + mgr._leader_state = LeaderState.NO_PEER + mgr._peer_ip = None + mgr._peer_compatible = False + mgr._peer_chain = 0 + mgr._last_heartbeat_time = 0.0 + mgr._leader_width = 0 + mgr._oversized_frame_warned = False + + mgr._follower_state = FollowerState.STANDALONE + mgr._latest_frame = None + mgr._latest_scroll_x = None + mgr._last_leader_frame_time = 0.0 + mgr._frame_lock = threading.Lock() + mgr._leader_ip = None + mgr._on_new_cycle = None + mgr._on_scroll_image = None + mgr._pending_scroll_image = None + mgr._scroll_image_lock = threading.Lock() + mgr._img_server_sock = None + + mgr._on_follower_connected = None + mgr._error_message = None + mgr._running = False + mgr._recv_sock = None + mgr._send_sock = None + return mgr + + +def once_then_stop(mgr, value): + """side_effect returning `value` once, then stopping the enclosing loop.""" + state = {"served": False} + + def _side_effect(*args, **kwargs): + if not state["served"]: + state["served"] = True + return value + mgr._running = False + raise socket.timeout() + + return _side_effect + + +def raise_n_then_stop(mgr, exc, count): + """side_effect raising `exc` `count` times, then stopping the loop.""" + state = {"n": 0} + + def _side_effect(*args, **kwargs): + state["n"] += 1 + if state["n"] <= count: + raise exc + mgr._running = False + raise socket.timeout() + + return _side_effect + + +def run_watchdog_once(monkeypatch, mgr, watchdog, now): + """Run exactly one watchdog iteration at a frozen wall-clock time.""" + monkeypatch.setattr(sync_manager.time, "time", lambda: now) + monkeypatch.setattr( + sync_manager.time, "sleep", lambda _: setattr(mgr, "_running", False)) + mgr._running = True + watchdog() + + +class FakeConn: + """Minimal TCP connection stand-in whose recv() drains a byte buffer.""" + + def __init__(self, payload: bytes): + self._buf = payload + self.closed = False + + def settimeout(self, _): + pass + + def recv(self, n): + chunk, self._buf = self._buf[:n], self._buf[n:] + return chunk + + def close(self): + self.closed = True + + +def png_bytes(size=(10, 10), color=(1, 2, 3)) -> bytes: + buf = io.BytesIO() + Image.new("RGB", size, color).save(buf, format="PNG") + return buf.getvalue() + + +def raw_frame_packet(width, height, color=(10, 20, 30)) -> bytes: + arr = np.asarray(Image.new("RGB", (width, height), color), dtype=np.uint8) + return _magic_header(width, height) + arr.tobytes() + + +def _magic_header(width, height) -> bytes: + return sync_manager._RAW_MAGIC + sync_manager._RAW_HEADER.pack(width, height) + + +def length_prefixed(payload: bytes) -> bytes: + return len(payload).to_bytes(4, "big") + payload + + +class TestRoleParsing: + def test_leader_role(self, monkeypatch): + monkeypatch.setattr(DisplaySyncManager, "_start_leader", lambda self: None) + assert DisplaySyncManager("leader", {}, {}, MagicMock()).role is SyncRole.LEADER + + def test_follower_role(self, monkeypatch): + monkeypatch.setattr(DisplaySyncManager, "_start_follower", lambda self: None) + assert DisplaySyncManager("follower", {}, {}, MagicMock()).role is SyncRole.FOLLOWER + + def test_standalone_starts_nothing(self): + mgr = DisplaySyncManager("standalone", {}, {}, MagicMock()) + assert mgr.role is SyncRole.STANDALONE + assert mgr._running is False + assert mgr._recv_sock is None + + def test_invalid_role_warns_and_falls_back(self): + logger = MagicMock() + assert DisplaySyncManager("bogus", {}, {}, logger).role is SyncRole.STANDALONE + assert logger.warning.called + + def test_role_matching_is_case_sensitive(self): + # Pinned: SyncRole's values are lowercase, so "LEADER" is not + # normalized — it is simply invalid and falls back to standalone. + logger = MagicMock() + assert DisplaySyncManager("LEADER", {}, {}, logger).role is SyncRole.STANDALONE + assert logger.warning.called + + def test_port_defaults_to_module_constant(self): + assert DisplaySyncManager("standalone", {}, {}, MagicMock()).port == sync_manager.SYNC_PORT + + def test_port_read_from_config(self): + assert DisplaySyncManager("standalone", {"port": 9999}, {}, MagicMock()).port == 9999 + + def test_oversized_frame_warned_initialized_in_init(self, monkeypatch): + # Regression: this attribute was only ever created on first use via + # getattr(self, '_oversized_frame_warned', False). + monkeypatch.setattr(DisplaySyncManager, "_start_leader", lambda self: None) + mgr = DisplaySyncManager("leader", {}, {}, MagicMock()) + assert mgr._oversized_frame_warned is False + + +class TestHandleHello: + def test_matching_panels_connect(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._send_sock = MagicMock() + mgr._handle_hello({"t": "hello", "rows": 32, "cols": 64, "chain": 3}, "10.0.0.5") + assert mgr._leader_state is LeaderState.CONNECTED + assert mgr._peer_ip == "10.0.0.5" + assert mgr._peer_compatible is True + assert mgr._peer_chain == 3 + assert mgr._error_message is None + + def test_ack_reports_compatibility(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._send_sock = MagicMock() + mgr._leader_width = 128 + mgr._handle_hello({"t": "hello", "rows": 32, "cols": 64, "chain": 1}, "10.0.0.5") + payload, dest = mgr._send_sock.sendto.call_args[0] + ack = json.loads(payload.decode("utf-8")) + assert ack["compatible"] is True + assert ack["leader_width"] == 128 + assert dest == ("10.0.0.5", mgr.port) + + def test_mismatched_panels_are_incompatible(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._send_sock = MagicMock() + mgr._handle_hello({"t": "hello", "rows": 16, "cols": 32, "chain": 1}, "10.0.0.5") + assert mgr._leader_state is LeaderState.INCOMPATIBLE + assert "Incompatible panels" in mgr._error_message + ack = json.loads(mgr._send_sock.sendto.call_args[0][0].decode("utf-8")) + assert ack["compatible"] is False + assert ack["error"] == mgr._error_message + + def test_chain_length_may_differ(self): + # Documented rule: rows/cols must match, chain_length need not. + mgr = make_manager(role=SyncRole.LEADER, hw_config={"rows": 32, "cols": 64, "chain_length": 1}) + mgr._send_sock = MagicMock() + mgr._handle_hello({"t": "hello", "rows": 32, "cols": 64, "chain": 4}, "10.0.0.5") + assert mgr._leader_state is LeaderState.CONNECTED + + def test_connect_callback_fires_only_on_first_transition(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._send_sock = MagicMock() + fired = threading.Event() + calls = [] + mgr._on_follower_connected = lambda: (calls.append(1), fired.set()) + + hello = {"t": "hello", "rows": 32, "cols": 64, "chain": 1} + mgr._handle_hello(hello, "10.0.0.5") + assert fired.wait(timeout=1) + assert len(calls) == 1 + + fired.clear() + mgr._handle_hello(hello, "10.0.0.5") # already CONNECTED + assert not fired.wait(timeout=0.2) + assert len(calls) == 1 + + def test_ack_send_failure_is_swallowed(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._send_sock = MagicMock() + mgr._send_sock.sendto.side_effect = OSError("network unreachable") + mgr._handle_hello({"t": "hello", "rows": 32, "cols": 64, "chain": 1}, "10.0.0.5") + assert mgr._leader_state is LeaderState.CONNECTED # state still updated + assert mgr.logger.debug.called + + +class TestWatchdogs: + def test_leader_drops_peer_after_heartbeat_timeout(self, monkeypatch): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + mgr._peer_compatible = True + mgr._last_heartbeat_time = 0.0 + run_watchdog_once(monkeypatch, mgr, mgr._leader_watchdog, + now=sync_manager.PEER_TIMEOUT + 1) + assert mgr._leader_state is LeaderState.NO_PEER + assert mgr._peer_ip is None + assert mgr._peer_compatible is False + + def test_leader_keeps_peer_within_timeout(self, monkeypatch): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + mgr._last_heartbeat_time = 100.0 + run_watchdog_once(monkeypatch, mgr, mgr._leader_watchdog, now=101.0) + assert mgr._leader_state is LeaderState.CONNECTED + assert mgr._peer_ip == "10.0.0.1" + + def test_leader_watchdog_ignores_disconnected_state(self, monkeypatch): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.INCOMPATIBLE + mgr._last_heartbeat_time = 0.0 + run_watchdog_once(monkeypatch, mgr, mgr._leader_watchdog, now=10_000) + assert mgr._leader_state is LeaderState.INCOMPATIBLE + + def test_follower_returns_to_standalone_after_frame_timeout(self, monkeypatch): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._follower_state = FollowerState.FOLLOWER + mgr._last_leader_frame_time = 0.0 + mgr._latest_frame = Image.new("RGB", (2, 2)) + run_watchdog_once(monkeypatch, mgr, mgr._follower_watchdog, + now=sync_manager.LEADER_TIMEOUT + 1) + assert mgr._follower_state is FollowerState.STANDALONE + assert mgr.get_latest_frame() is None + + def test_follower_keeps_frames_within_timeout(self, monkeypatch): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._follower_state = FollowerState.FOLLOWER + mgr._last_leader_frame_time = 100.0 + mgr._latest_frame = Image.new("RGB", (2, 2)) + run_watchdog_once(monkeypatch, mgr, mgr._follower_watchdog, now=101.0) + assert mgr._follower_state is FollowerState.FOLLOWER + assert mgr.get_latest_frame() is not None + + +class TestLeaderRecvLoop: + def _drive(self, mgr, payload, sender="10.0.0.8"): + mgr._recv_sock = MagicMock() + mgr._recv_sock.recvfrom.side_effect = once_then_stop(mgr, (payload, (sender, 1))) + mgr._running = True + mgr._leader_recv_loop() + + def test_hello_is_dispatched(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._send_sock = MagicMock() + self._drive(mgr, json.dumps( + {"t": "hello", "rows": 32, "cols": 64, "chain": 1}).encode()) + assert mgr._leader_state is LeaderState.CONNECTED + assert mgr._peer_ip == "10.0.0.8" + + def test_heartbeat_from_known_peer_refreshes_timer(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._peer_ip = "10.0.0.8" + with patch.object(sync_manager.time, "time", return_value=12345.0): + self._drive(mgr, json.dumps({"t": "hb"}).encode()) + assert mgr._last_heartbeat_time == 12345.0 + + def test_heartbeat_from_stranger_is_ignored(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._peer_ip = "10.0.0.8" + mgr._last_heartbeat_time = 5.0 + self._drive(mgr, json.dumps({"t": "hb"}).encode(), sender="10.0.0.99") + assert mgr._last_heartbeat_time == 5.0 + + def test_unknown_message_type_ignored(self): + mgr = make_manager(role=SyncRole.LEADER) + self._drive(mgr, json.dumps({"t": "who-knows"}).encode()) + assert mgr._leader_state is LeaderState.NO_PEER + + def test_malformed_json_is_swallowed(self): + mgr = make_manager(role=SyncRole.LEADER) + self._drive(mgr, b"{not json") + assert mgr._leader_state is LeaderState.NO_PEER + + def test_undecodable_bytes_are_swallowed(self): + mgr = make_manager(role=SyncRole.LEADER) + self._drive(mgr, b"\xff\xfe\x00bad") + assert mgr._leader_state is LeaderState.NO_PEER + + def test_backs_off_between_repeated_errors(self, monkeypatch): + # Regression: without a sleep this loop spun at 100% CPU whenever + # the socket raised a non-timeout error on every call. + mgr = make_manager(role=SyncRole.LEADER) + mgr._recv_sock = MagicMock() + mgr._recv_sock.recvfrom.side_effect = raise_n_then_stop(mgr, OSError("boom"), 3) + sleeps = MagicMock() + monkeypatch.setattr(sync_manager.time, "sleep", sleeps) + mgr._running = True + mgr._leader_recv_loop() + assert sleeps.call_count == 3 + sleeps.assert_called_with(0.1) + + +class TestFollowerRecvLoop: + def _drive(self, mgr, payload, sender="10.0.0.2"): + mgr._recv_sock = MagicMock() + mgr._recv_sock.recvfrom.side_effect = once_then_stop(mgr, (payload, (sender, 1))) + mgr._running = True + mgr._follower_recv_loop() + + def test_small_raw_frame_is_decoded(self): + # Regression: a raw frame under the old 512-byte threshold was sent + # to the JSON parser and dropped. + mgr = make_manager(role=SyncRole.FOLLOWER) + packet = raw_frame_packet(4, 3) + assert len(packet) <= 512 + self._drive(mgr, packet) + frame = mgr.get_latest_frame() + assert frame is not None and frame.size == (4, 3) + assert mgr._follower_state is FollowerState.FOLLOWER + + def test_large_raw_frame_is_decoded(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + packet = raw_frame_packet(64, 32) + assert len(packet) > 512 + self._drive(mgr, packet) + assert mgr.get_latest_frame().size == (64, 32) + + def test_large_control_message_is_not_routed_to_image_decode(self): + # Regression: the old `len(data) > 512` branch treated any large + # control message as frame data and silently discarded it. + mgr = make_manager(role=SyncRole.FOLLOWER) + long_error = "x" * 600 + payload = json.dumps( + {"t": "hello_ack", "compatible": False, "error": long_error}).encode() + assert len(payload) > 512 + self._drive(mgr, payload, sender="10.0.0.9") + assert mgr._leader_ip == "10.0.0.9" + assert mgr._peer_compatible is False + assert mgr._error_message == long_error + assert mgr.get_latest_frame() is None + assert mgr.logger.error.called + + def test_legacy_png_frame_without_magic_is_decoded(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + self._drive(mgr, png_bytes(size=(5, 5))) + frame = mgr.get_latest_frame() + assert frame is not None and frame.size == (5, 5) + assert mgr._follower_state is FollowerState.FOLLOWER + + def test_truncated_raw_frame_is_swallowed(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + self._drive(mgr, _magic_header(64, 32) + b"\x00" * 10) # far too short + assert mgr.get_latest_frame() is None + assert mgr.logger.debug.called + + def test_garbage_payload_is_swallowed(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + self._drive(mgr, b"neither json nor a png, just bytes 1234567890") + assert mgr.get_latest_frame() is None + + def test_hello_ack_updates_peer_state(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + self._drive(mgr, json.dumps( + {"t": "hello_ack", "compatible": True, "error": None}).encode(), + sender="10.0.0.6") + assert mgr._leader_ip == "10.0.0.6" + assert mgr._peer_compatible is True + assert mgr.logger.error.called is False + + def test_scroll_x_switches_to_follower_and_builds_cycle(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + calls = [] + mgr._on_new_cycle = lambda: calls.append(1) + self._drive(mgr, json.dumps({"t": "sx", "x": 12.34}).encode()) + assert mgr._follower_state is FollowerState.FOLLOWER + assert mgr.get_latest_scroll_x() == 12.34 + assert calls == [1] + + def test_scroll_x_while_already_following_does_not_rebuild(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._follower_state = FollowerState.FOLLOWER + calls = [] + mgr._on_new_cycle = lambda: calls.append(1) + self._drive(mgr, json.dumps({"t": "sx", "x": 5.0}).encode()) + assert mgr.get_latest_scroll_x() == 5.0 + assert calls == [] + + def test_new_cycle_message_triggers_callback(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._follower_state = FollowerState.FOLLOWER + calls = [] + mgr._on_new_cycle = lambda: calls.append(1) + self._drive(mgr, json.dumps({"t": "nc"}).encode()) + assert calls == [1] + + def test_scroll_x_missing_key_is_swallowed(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + self._drive(mgr, json.dumps({"t": "sx"}).encode()) # no "x" + assert mgr.get_latest_scroll_x() is None + + def test_backs_off_between_repeated_errors(self, monkeypatch): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._recv_sock = MagicMock() + mgr._recv_sock.recvfrom.side_effect = raise_n_then_stop(mgr, OSError("boom"), 3) + sleeps = MagicMock() + monkeypatch.setattr(sync_manager.time, "sleep", sleeps) + mgr._running = True + mgr._follower_recv_loop() + assert sleeps.call_count == 3 + sleeps.assert_called_with(0.1) + + +class TestSendFrame: + def _connected_leader(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + mgr._send_sock = MagicMock() + return mgr + + def test_frame_sent_with_magic_header(self): + mgr = self._connected_leader() + mgr.send_frame(Image.new("RGB", (8, 8))) + packet = mgr._send_sock.sendto.call_args[0][0] + assert packet[:8] == sync_manager._RAW_MAGIC + assert sync_manager._RAW_HEADER.unpack(packet[8:12]) == (8, 8) + + def test_oversized_frame_warns_once_and_is_dropped(self): + mgr = self._connected_leader() + big = Image.new("RGB", (300, 300)) # 270000 bytes > 65000 UDP cap + + mgr.send_frame(big) + assert mgr._oversized_frame_warned is True + assert mgr.logger.warning.call_count == 1 + assert not mgr._send_sock.sendto.called + + mgr.send_frame(big) + assert mgr.logger.warning.call_count == 1 # still warned only once + + def test_not_sent_when_no_peer(self): + mgr = self._connected_leader() + mgr._leader_state = LeaderState.NO_PEER + mgr.send_frame(Image.new("RGB", (8, 8))) + assert not mgr._send_sock.sendto.called + + def test_follower_never_sends(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._send_sock = MagicMock() + mgr.send_frame(Image.new("RGB", (8, 8))) + assert not mgr._send_sock.sendto.called + + def test_send_error_is_swallowed(self): + mgr = self._connected_leader() + mgr._send_sock.sendto.side_effect = OSError("no route") + mgr.send_frame(Image.new("RGB", (8, 8))) # must not raise + assert mgr.logger.debug.called + + +class TestSendControlMessages: + def _connected_leader(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + mgr._send_sock = MagicMock() + return mgr + + def test_send_scroll_x_rounds_to_two_places(self): + mgr = self._connected_leader() + mgr.send_scroll_x(3.14159) + msg = json.loads(mgr._send_sock.sendto.call_args[0][0].decode()) + assert msg == {"t": "sx", "x": 3.14} + + def test_send_new_cycle(self): + mgr = self._connected_leader() + mgr.send_new_cycle() + msg = json.loads(mgr._send_sock.sendto.call_args[0][0].decode()) + assert msg == {"t": "nc"} + + def test_control_messages_noop_when_disconnected(self): + mgr = self._connected_leader() + mgr._leader_state = LeaderState.NO_PEER + mgr.send_scroll_x(1.0) + mgr.send_new_cycle() + assert not mgr._send_sock.sendto.called + + def test_set_leader_width(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr.set_leader_width(256) + assert mgr._leader_width == 256 + + +class TestImageServerLoop: + def _drive(self, mgr, conn): + mgr._img_server_sock = MagicMock() + mgr._img_server_sock.accept.side_effect = once_then_stop( + mgr, (conn, ("10.0.0.1", 1))) + mgr._running = True + mgr._image_server_loop() + + def test_rejects_non_positive_length(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._on_scroll_image = MagicMock() + self._drive(mgr, FakeConn((0).to_bytes(4, "big"))) + assert mgr.logger.warning.called + mgr._on_scroll_image.assert_not_called() + + def test_rejects_oversized_length(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._on_scroll_image = MagicMock() + self._drive(mgr, FakeConn((11 * 1024 * 1024).to_bytes(4, "big"))) + assert mgr.logger.warning.called + mgr._on_scroll_image.assert_not_called() + + def test_rejects_oversized_dimensions(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._on_scroll_image = MagicMock() + self._drive(mgr, FakeConn(length_prefixed(png_bytes(size=(300, 300))))) + assert mgr.logger.warning.called + mgr._on_scroll_image.assert_not_called() + + def test_rejects_decompression_bomb(self, monkeypatch): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._on_scroll_image = MagicMock() + + class BombImage: + width = height = 10 + + def load(self): + raise Image.DecompressionBombError("too many pixels") + + monkeypatch.setattr(sync_manager.Image, "open", lambda *a, **kw: BombImage()) + self._drive(mgr, FakeConn(length_prefixed(png_bytes()))) + assert mgr.logger.warning.called + mgr._on_scroll_image.assert_not_called() + + def test_valid_image_invokes_callback(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + received = [] + mgr._on_scroll_image = received.append + self._drive(mgr, FakeConn(length_prefixed(png_bytes(size=(10, 10))))) + assert len(received) == 1 + assert received[0].size == (10, 10) + + def test_image_cached_when_callback_not_yet_registered(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._on_scroll_image = None + self._drive(mgr, FakeConn(length_prefixed(png_bytes(size=(6, 6))))) + assert mgr._pending_scroll_image is not None + assert mgr._pending_scroll_image.size == (6, 6) + + def test_short_header_is_skipped(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._on_scroll_image = MagicMock() + self._drive(mgr, FakeConn(b"\x00\x01")) # under the 4-byte prefix + mgr._on_scroll_image.assert_not_called() + + def test_connection_always_closed(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + conn = FakeConn(length_prefixed(png_bytes())) + self._drive(mgr, conn) + assert conn.closed is True + + +class TestScrollImageCallback: + def test_pending_image_delivered_on_late_registration(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + img = Image.new("RGB", (3, 3)) + mgr._pending_scroll_image = img + received = [] + mgr.set_on_scroll_image(received.append) + assert received == [img] + assert mgr._pending_scroll_image is None + + def test_no_pending_image_means_no_immediate_call(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + received = [] + mgr.set_on_scroll_image(received.append) + assert received == [] + + +class TestFollowerConnectedCallback: + def test_fires_immediately_when_already_connected(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + fired = threading.Event() + mgr.set_on_follower_connected(fired.set) + assert fired.wait(timeout=1) + + def test_does_not_fire_when_no_peer(self): + mgr = make_manager(role=SyncRole.LEADER) + fired = threading.Event() + mgr.set_on_follower_connected(fired.set) + assert not fired.wait(timeout=0.2) + + +class TestSendScrollImage: + def test_noop_when_not_connected(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.NO_PEER + with patch.object(sync_manager.socket, "socket") as sock: + mgr.send_scroll_image(Image.new("RGB", (4, 4))) + sock.assert_not_called() + + def test_noop_for_follower_role(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + with patch.object(sync_manager.socket, "socket") as sock: + mgr.send_scroll_image(Image.new("RGB", (4, 4))) + sock.assert_not_called() + + def test_sends_length_prefixed_png(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + fake_sock = MagicMock() + fake_sock.__enter__ = lambda s: s + fake_sock.__exit__ = lambda s, *a: False + with patch.object(sync_manager.socket, "socket", return_value=fake_sock): + mgr.send_scroll_image(Image.new("RGB", (4, 4))) + payload = fake_sock.sendall.call_args[0][0] + assert int.from_bytes(payload[:4], "big") == len(payload) - 4 + assert payload[4:8] == b"\x89PNG" + + def test_connection_error_is_swallowed(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + with patch.object(sync_manager.socket, "socket", side_effect=OSError("refused")): + mgr.send_scroll_image(Image.new("RGB", (4, 4))) # must not raise + assert mgr.logger.debug.called + + +class TestGetStatus: + def test_standalone_shape(self): + status = make_manager(role=SyncRole.STANDALONE).get_status() + assert status["role"] == "standalone" + assert status["state"] == "standalone" + assert status["local_rows"] == 32 and status["local_cols"] == 64 + + def test_leader_shape(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._leader_state = LeaderState.CONNECTED + mgr._peer_ip = "10.0.0.1" + mgr._peer_compatible = True + mgr._peer_chain = 2 + mgr._leader_width = 128 + status = mgr.get_status() + assert status["role"] == "leader" + assert status["state"] == "connected" + assert status["peer_ip"] == "10.0.0.1" + assert status["peer_chain"] == 2 + assert status["leader_width"] == 128 + + def test_follower_shape(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + mgr._follower_state = FollowerState.FOLLOWER + mgr._leader_ip = "10.0.0.2" + status = mgr.get_status() + assert status["role"] == "follower" + assert status["state"] == "follower" + assert status["leader_ip"] == "10.0.0.2" + assert "peer_chain" not in status + + def test_is_follower_active(self): + mgr = make_manager(role=SyncRole.FOLLOWER) + assert mgr.is_follower_active() is False + mgr._follower_state = FollowerState.FOLLOWER + assert mgr.is_follower_active() is True + + def test_leader_is_never_follower_active(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._follower_state = FollowerState.FOLLOWER + assert mgr.is_follower_active() is False + + +class TestWriteStatusFile: + def test_writes_status_and_cleans_up_temp(self): + mgr = make_manager(role=SyncRole.STANDALONE) + mgr.write_status_file() + data = json.loads(Path(sync_manager.STATUS_FILE).read_text()) + assert data["role"] == "standalone" + assert "ts" in data + assert not Path(sync_manager.STATUS_FILE + ".tmp").exists() + + def test_write_failure_is_swallowed(self, monkeypatch): + mgr = make_manager(role=SyncRole.STANDALONE) + monkeypatch.setattr("builtins.open", MagicMock(side_effect=OSError("disk full"))) + mgr.write_status_file() # must not raise + assert mgr.logger.debug.called + + +class TestStop: + def _stub_with_sockets(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._recv_sock = MagicMock() + mgr._send_sock = MagicMock() + mgr._img_server_sock = MagicMock() + return mgr + + def test_closes_every_socket(self): + mgr = self._stub_with_sockets() + mgr.stop() + assert mgr._running is False + mgr._recv_sock.close.assert_called_once() + mgr._send_sock.close.assert_called_once() + mgr._img_server_sock.close.assert_called_once() + + def test_is_idempotent(self): + mgr = self._stub_with_sockets() + mgr.stop() + mgr.stop() # must not raise + + def test_close_failure_is_swallowed(self): + mgr = make_manager(role=SyncRole.LEADER) + mgr._recv_sock = MagicMock() + mgr._recv_sock.close.side_effect = OSError("already closed") + mgr.stop() # must not raise + assert mgr.logger.debug.called + + def test_handles_unset_sockets(self): + make_manager(role=SyncRole.STANDALONE).stop() # all sockets None + + +class TestLoopbackHandshake: + def test_leader_and_follower_negotiate_over_real_sockets(self, monkeypatch): + # One end-to-end check that the wire format actually round-trips: + # every other test drives the loops with mocked sockets. + monkeypatch.setattr(sync_manager, "HELLO_INTERVAL", 0.02) + monkeypatch.setattr(sync_manager, "HEARTBEAT_INTERVAL", 0.02) + + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + probe.bind(("", 0)) + port = probe.getsockname()[1] + probe.close() + + hw = {"rows": 32, "cols": 64, "chain_length": 1} + leader = DisplaySyncManager("leader", {"port": port}, hw, MagicMock()) + follower = DisplaySyncManager("follower", {"port": port}, hw, MagicMock()) + try: + deadline = time.time() + 5.0 + while time.time() < deadline: + if (leader._leader_state is LeaderState.CONNECTED + and follower._peer_compatible): + break + time.sleep(0.02) + assert leader._leader_state is LeaderState.CONNECTED + assert follower._peer_compatible is True + assert follower._leader_ip is not None + finally: + leader.stop() + follower.stop()