LEDMatrix/test/test_display_dirty_tracking.py

"""Tests for update_display dirty tracking (src/display_manager.py).

Runs against RGBMatrixEmulator (EMULATOR=true), exercising the REAL
DisplayManager — not a mock — so the skip logic, its invalidation hooks,
and the kill switch are verified off-Pi.

The invariants:
- identical frames are pushed exactly once (SwapOnVSync not re-called)
- ANY pixel change pushes
- clear() and set_brightness() invalidate (the two paths that alter panel
  state outside the digest's view)
- the kill switch (display.dirty_tracking: false) restores always-push
"""

import os
import sys
import time

os.environ["EMULATOR"] = "true"

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))


@pytest.fixture(scope="module")
def dm(tmp_path_factory):
    """One real DisplayManager on the emulator (it's a process singleton)."""
    from src.display_manager import DisplayManager
    DisplayManager._instance = None
    manager = DisplayManager({
        "display": {
            "hardware": {"rows": 32, "cols": 64, "chain_length": 2,
                         "parallel": 1, "brightness": 90},
            "runtime": {"gpio_slowdown": 0},
        },
    }, suppress_test_pattern=True)
    # DisplayManager defaults _snapshot_path to the fixed /tmp/led_matrix_preview.png
    # that the web UI reads. That path is shared by every pytest process on the
    # machine, so two concurrent runs -- CI shards, a second worktree, a agent
    # running the suite alongside -- write over each other's snapshot and the
    # mtime assertions below stop meaning anything. Point it somewhere unique to
    # this session; the individual tests that care still override it further.
    manager._snapshot_path = str(
        tmp_path_factory.mktemp("dirty_tracking") / "led_matrix_preview.png")
    # _setup_matrix() swallows every construction failure and falls back to
    # matrix=None, so a broken environment reaches the tests as fifteen
    # identical "'NoneType' object has no attribute 'SwapOnVSync'" errors that
    # name neither this fixture nor the real cause. Fail here instead, once,
    # and say where to look.
    if manager.matrix is None:
        pytest.fail(
            "DisplayManager fell back to matrix=None: RGBMatrix construction "
            "raised (the 'Failed to initialize RGB Matrix' log line above "
            "carries the reason). Known causes: the emulator adapter losing a "
            "fixed TCP port to another process -- see pytest_configure in "
            "test/conftest.py, which pins the port-free 'raw' adapter -- or a "
            "patch('src.display_manager.RGBMatrix') leaked from an earlier "
            "test module.")
    yield manager
    DisplayManager._instance = None


class _SwapSpy:
    """Counts SwapOnVSync calls through the real matrix object."""

    def __init__(self, matrix):
        self.matrix = matrix
        self.count = 0
        self.last_frame_hold = None
        self.holds = []
        self._orig = matrix.SwapOnVSync

    def __enter__(self):
        def counting(canvas, *args):
            # *args carries framerate_fraction, which display_manager passes so
            # a frame can be held for several refreshes. Signature must match
            # the real binding or the spy hides a TypeError as a failed push.
            self.count += 1
            self.last_frame_hold = args[0] if args else 1
            self.holds.append(self.last_frame_hold)
            return self._orig(canvas, *args)
        self.matrix.SwapOnVSync = counting
        return self

    def __exit__(self, *exc):
        self.matrix.SwapOnVSync = self._orig


class TestDirtyTracking:
    def test_identical_frames_push_once(self, dm):
        dm.draw.rectangle([0, 0, 10, 10], fill=(255, 0, 0))
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()
            dm.update_display()
            dm.update_display()
        assert spy.count == 1

    def test_pixel_change_pushes(self, dm):
        dm.update_display()
        with _SwapSpy(dm.matrix) as spy:
            dm.draw.point((5, 5), fill=(0, 255, 0))
            dm.update_display()
            dm.update_display()  # unchanged again
        assert spy.count == 1

    def test_clear_invalidates(self, dm):
        dm.draw.rectangle([0, 0, 20, 20], fill=(0, 0, 255))
        dm.update_display()
        dm.clear()  # writes to the matrix directly; digest must reset
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()  # black frame after clear must still push
        assert spy.count == 1

    def test_brightness_change_forces_push(self, dm):
        dm.draw.rectangle([0, 0, 20, 20], fill=(200, 200, 200))
        dm.update_display()
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()          # identical -> skipped
            assert spy.count == 0
            dm.set_brightness(40)        # dim schedule scenario
            dm.update_display()          # same image, new brightness -> push
        assert spy.count == 1
        dm.set_brightness(90)

    def test_snapshot_still_written_on_skip(self, dm, tmp_path):
        """The web preview mirror must keep working through skipped panel
        pushes: _write_snapshot_if_due() still runs on the dirty-tracking
        skip path and applies its own write/touch policy rather than being
        bypassed entirely (see src/common/snapshot_policy.py — an unchanged
        frame is touched, not re-encoded, once TOUCH_INTERVAL elapses)."""
        dm._snapshot_path = str(tmp_path / "snap.png")
        dm._last_snapshot_ts = 0.0
        dm._last_snapshot_touch_ts = 0.0
        dm._last_snapshot_digest = None
        dm.draw.rectangle([0, 0, 30, 8], fill=(255, 255, 0))
        dm.update_display()   # push + snapshot write (first frame)
        assert os.path.exists(dm._snapshot_path)
        # Backdate the file so the "was it bumped?" check below cannot be
        # defeated by filesystem mtime granularity -- on Windows two writes in
        # the same tick get identical timestamps, which made this test fail
        # roughly two runs in three regardless of the code under test.
        os.utime(dm._snapshot_path, (time.time() - 60, time.time() - 60))
        first_mtime = os.path.getmtime(dm._snapshot_path)

        # Age the write/touch bookkeeping past TOUCH_INTERVAL so the next
        # identical frame is due for a touch, then push it again: dirty
        # tracking must skip the panel write, but the snapshot mirror must
        # still get its mtime bumped so the health check doesn't go stale.
        from src.common import snapshot_policy
        stale_ts = time.time() - snapshot_policy.TOUCH_INTERVAL - 1.0
        dm._last_snapshot_ts = stale_ts
        dm._last_snapshot_touch_ts = stale_ts
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()  # identical frame -> panel push skipped
        assert spy.count == 0
        assert os.path.getmtime(dm._snapshot_path) > first_mtime


class TestScrollLock:
    """Dirty tracking must not skip the panel push while a scroll is running.

    SwapOnVSync is what paces the render loop, so skipping it also skips the
    wait for the panel. A duplicate frame therefore returns early -- ~8ms
    instead of ~10ms on a 100Hz panel -- which advances the strip only 0.8px
    instead of 1.0px, which makes the NEXT frame more likely to be a duplicate
    too. That is self-sustaining: measured at ~20% duplicate frames mid-scroll
    on the odds ticker against essentially zero on a lighter plugin with
    identical scroll settings. Pushing an identical frame costs one canvas
    copy; falling out of vsync lock costs smooth motion.
    """

    def test_identical_frames_still_push_while_scrolling(self, dm):
        dm.draw.rectangle([0, 0, 12, 12], fill=(0, 0, 255))
        dm.update_display()
        dm.set_scrolling_state(True)
        try:
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
                dm.update_display()
                dm.update_display()
            assert spy.count == 3, "scrolling must stay locked to the panel"
        finally:
            dm.set_scrolling_state(False)

    def test_identical_frames_are_skipped_when_not_scrolling(self, dm):
        """The optimisation still applies to static content."""
        dm.set_scrolling_state(False)
        dm.draw.rectangle([0, 0, 14, 14], fill=(255, 0, 255))
        dm.update_display()
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()
            dm.update_display()
        assert spy.count == 0

    def test_stale_scrolling_state_stops_forcing_pushes(self, dm):
        """A plugin that stops scrolling without saying so must not pin the
        panel into always-push forever. is_currently_scrolling() expires on
        its own inactivity threshold, and the skip has to come back with it."""
        dm.draw.rectangle([0, 0, 16, 16], fill=(0, 255, 255))
        dm.update_display()
        dm.set_scrolling_state(True)
        try:
            # Backdate the activity marker past the inactivity threshold.
            dm._scrolling_state['last_scroll_activity'] = (
                time.time() - dm._scrolling_state['scroll_inactivity_threshold'] - 1.0)
            assert dm.is_currently_scrolling() is False
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
                dm.update_display()
            assert spy.count == 0
        finally:
            dm.set_scrolling_state(False)


class TestKillSwitch:
    def test_dirty_tracking_can_be_disabled(self, dm):
        dm._dirty_tracking_enabled = False
        try:
            dm.draw.rectangle([0, 0, 10, 10], fill=(1, 2, 3))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
                dm.update_display()
                dm.update_display()
            assert spy.count == 3  # always-push, exactly the old behavior
        finally:
            dm._dirty_tracking_enabled = True
            dm._last_pushed_digest = None

    def test_config_flag_wires_through(self):
        from src.display_manager import DisplayManager
        DisplayManager._instance = None
        try:
            manager = DisplayManager({
                "display": {
                    "hardware": {"rows": 32, "cols": 64, "chain_length": 1,
                                 "parallel": 1},
                    "runtime": {"gpio_slowdown": 0},
                    "dirty_tracking": False,
                },
            }, suppress_test_pattern=True)
            assert manager._dirty_tracking_enabled is False
        finally:
            DisplayManager._instance = None


if __name__ == "__main__":
    sys.exit(pytest.main([__file__, "-v"]))


class TestFrameHold:
    """Holding a frame for N refreshes is how a scroll runs slower than one
    pixel per refresh without fractional pixel positions."""

    def test_hold_reaches_swap_on_vsync(self, dm):
        dm.set_scrolling_state(True)
        dm.set_frame_hold(3)
        try:
            dm.draw.rectangle([0, 0, 9, 9], fill=(120, 0, 200))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
            assert spy.count == 1
            assert spy.last_frame_hold == 3
        finally:
            dm.set_scrolling_state(False)

    def test_default_is_every_refresh(self, dm):
        dm.set_scrolling_state(True)
        try:
            dm.draw.rectangle([0, 0, 11, 11], fill=(0, 120, 200))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
            assert spy.last_frame_hold == 1
        finally:
            dm.set_scrolling_state(False)

    def test_hold_resets_when_scrolling_stops(self, dm):
        """One plugin's pacing must not leak into whatever is on screen next."""
        dm.set_scrolling_state(True)
        dm.set_frame_hold(5)
        dm.set_scrolling_state(False)
        dm.set_scrolling_state(True)
        try:
            dm.draw.rectangle([0, 0, 13, 13], fill=(200, 120, 0))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
            assert spy.last_frame_hold == 1
        finally:
            dm.set_scrolling_state(False)

    @pytest.mark.parametrize("bad,expected", [(0, 1), (-4, 1), (None, 1), ("x", 1)])
    def test_unusable_holds_are_ignored_or_floored(self, dm, bad, expected):
        dm.set_frame_hold(bad)
        assert dm._frame_hold == expected


class TestFrameHoldLifetime:
    """The hold must last exactly as long as the scroll that asked for it.

    Plugins share one display manager. A hold applied at plugin construction is
    wiped the moment any *other* plugin finishes scrolling, so by the time the
    first plugin renders it is back to one pixel per refresh -- the speed reads
    correct in the log and is wrong on the panel.
    """

    def test_scrolling_state_carries_the_hold(self, dm):
        dm.set_scrolling_state(True, frame_hold=4)
        try:
            dm.draw.rectangle([0, 0, 7, 7], fill=(10, 200, 10))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
            assert spy.last_frame_hold == 4
        finally:
            dm.set_scrolling_state(False)

    def test_a_scroll_that_times_out_does_not_strand_a_hold(self, dm):
        """The hold must go when the state does, however the scroll ended.

        set_scrolling_state(False) is the polite exit. The other one is
        is_currently_scrolling() deciding, after scroll_inactivity_threshold
        of silence, that the scroll is over -- which is what happens when the
        rotation moves on mid-scroll or a plugin is torn down. That path used
        to clear the flag and keep the hold, so every later plugin, scrolling
        or static, was presented at refresh/N by whoever scrolled last.
        """
        dm.set_scrolling_state(True, frame_hold=5)
        # Age the scroll past the inactivity threshold rather than sleeping.
        dm._scrolling_state['last_scroll_activity'] -= (
            dm._scrolling_state['scroll_inactivity_threshold'] + 1.0)

        assert dm.is_currently_scrolling() is False
        dm.draw.rectangle([0, 0, 5, 5], fill=(10, 10, 200))
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()
        assert spy.last_frame_hold == 1, (
            "a timed-out scroll left its frame hold behind; the next plugin "
            "is being presented at a fraction of the refresh rate")

    def test_another_plugin_stopping_does_not_strand_a_hold(self, dm):
        dm.set_scrolling_state(True, frame_hold=3)
        dm.set_scrolling_state(False)          # some other plugin finishes
        dm.set_scrolling_state(True)           # a plugin that wants no hold
        try:
            dm.draw.rectangle([0, 0, 6, 6], fill=(200, 10, 10))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
            assert spy.last_frame_hold == 1
        finally:
            dm.set_scrolling_state(False)

    def test_default_keeps_previous_behaviour(self, dm):
        """Callers that never heard of frame holds get one frame per refresh."""
        dm.set_scrolling_state(True)
        try:
            assert dm._frame_hold == 1
        finally:
            dm.set_scrolling_state(False)


class TestSnapshotOffRenderThread:
    """Mid-scroll, the preview PNG is encoded off the render thread.

    At 512x64 the encode takes 12-14ms on a Pi 4 -- longer than a refresh --
    so doing it inline made the next swap miss its vsync five times a second
    whenever the web preview was open.
    """

    def _record_saves(self, dm, monkeypatch):
        import threading
        threads = []
        done = threading.Event()
        real = dm._save_snapshot

        def recording(image):
            threads.append(threading.current_thread().name)
            real(image)
            done.set()

        monkeypatch.setattr(dm, "_save_snapshot", recording)
        return threads, done

    def _due(self, dm, tmp_path, colour):
        dm._snapshot_path = str(tmp_path / "snap.png")
        dm._last_snapshot_ts = 0.0
        dm._last_snapshot_touch_ts = 0.0
        dm._last_snapshot_digest = None
        dm.draw.rectangle([0, 0, 10, 4], fill=colour)

    def test_scrolling_frames_are_encoded_on_the_writer_thread(
            self, dm, tmp_path, monkeypatch):
        import threading
        threads, done = self._record_saves(dm, monkeypatch)
        self._due(dm, tmp_path, (0, 255, 255))
        dm.set_scrolling_state(True)
        try:
            dm.update_display()
            assert done.wait(5), "the snapshot writer never wrote the frame"
        finally:
            dm.set_scrolling_state(False)
        assert threads == ["snapshot-writer"]
        assert threads[0] != threading.current_thread().name
        assert os.path.exists(dm._snapshot_path)

    def test_a_failed_background_write_is_retried_not_touched(
            self, dm, tmp_path, monkeypatch):
        # Queuing records the frame as written. If the writer then fails, an
        # unchanged frame must be written again, not mtime-touched: touching
        # would make a stale preview look healthy.
        import threading
        import time
        failed = threading.Event()

        def failing(image):
            failed.set()
            raise OSError("disk full")

        monkeypatch.setattr(dm, "_save_snapshot", failing)
        self._due(dm, tmp_path, (0, 255, 0))
        dm.set_scrolling_state(True)
        try:
            dm.update_display()
            assert failed.wait(5)
            deadline = time.time() + 5
            while dm._last_snapshot_digest is not None and time.time() < deadline:
                time.sleep(0.01)
        finally:
            dm.set_scrolling_state(False)
        assert dm._last_snapshot_digest is None

    def test_a_frame_not_yet_on_disk_is_written_not_touched(
            self, dm, tmp_path, monkeypatch):
        # The digest is recorded when a frame is queued. Until the writer has
        # saved it, an unchanged frame must not mtime-touch the older file on
        # disk into looking current.
        import zlib
        from src.common import snapshot_policy
        touched, saved = [], []
        self._due(dm, tmp_path, (9, 9, 9))
        digest = zlib.adler32(dm.image.tobytes())
        dm._last_snapshot_digest = digest        # queued earlier...
        dm._saved_snapshot_digest = 12345        # ...but an older frame is on disk
        monkeypatch.setattr(snapshot_policy, "decide",
                            lambda *a, **k: snapshot_policy.SnapshotAction.TOUCH)
        monkeypatch.setattr(os, "utime", lambda *a, **k: touched.append(a))
        monkeypatch.setattr(dm, "_save_snapshot", lambda image: saved.append(image))
        dm.set_scrolling_state(False)
        dm._write_snapshot_if_due(digest)
        assert touched == []
        assert len(saved) == 1
        assert dm._saved_snapshot_digest == digest

        # Once it is on disk, the same frame is only touched.
        dm._write_snapshot_if_due(digest)
        assert len(touched) == 1 and len(saved) == 1

    def test_a_static_frame_lands_after_a_queued_one_still_being_written(
            self, dm, tmp_path, monkeypatch):
        # The last frame of a scroll can still be encoding when the first
        # static frame is due; the older one must not land on top.
        import threading
        written, started, release = [], threading.Event(), threading.Event()
        real = dm._save_snapshot

        def slow_then_record(image):
            if threading.current_thread().name == "snapshot-writer":
                started.set()
                release.wait(5)
            written.append((threading.current_thread().name, image.getpixel((0, 0))))
            real(image)

        monkeypatch.setattr(dm, "_save_snapshot", slow_then_record)
        self._due(dm, tmp_path, (0, 0, 255))
        dm.set_scrolling_state(True)
        dm.update_display()                       # queued: the writer blocks mid-write
        assert started.wait(5)
        dm.set_scrolling_state(False)
        self._due(dm, tmp_path, (255, 0, 0))
        static = threading.Thread(target=dm.update_display)
        static.start()
        static.join(0.2)
        assert static.is_alive(), "the static save must wait for the write in flight"
        release.set()
        static.join(5)
        assert [colour for _, colour in written] == [(0, 0, 255), (255, 0, 0)]

    def test_cleanup_stops_the_writer(self, dm, tmp_path, monkeypatch):
        threads, done = self._record_saves(dm, monkeypatch)
        self._due(dm, tmp_path, (0, 255, 255))
        dm.set_scrolling_state(True)
        dm.update_display()
        assert done.wait(5)
        writer = dm._snapshot_thread
        dm.set_scrolling_state(False)
        dm._stop_snapshot_writer()
        writer.join(2)
        assert not writer.is_alive()

    def test_static_frames_are_still_written_inline(
            self, dm, tmp_path, monkeypatch):
        import threading
        threads, _ = self._record_saves(dm, monkeypatch)
        self._due(dm, tmp_path, (255, 0, 255))
        dm.set_scrolling_state(False)
        dm.update_display()
        assert threads == [threading.current_thread().name]


class TestLazyDigest:
    """Mid-scroll the frame is hashed only when the preview snapshot is due.

    The digest feeds two things: the dirty-tracking skip, which is never
    taken while scrolling, and the snapshot's changed-frame check, which acts
    at most once a second. Hashing every scrolled frame for them cost a
    tobytes() plus adler32 over the whole framebuffer on every frame (~0.3ms
    at 512x64 on a Pi 4). update_display() now leaves the digest out while
    scrolling, and _write_snapshot_if_due() hashes only when the policy could
    act on the answer.
    """

    @pytest.fixture
    def hashes(self, dm, tmp_path, monkeypatch):
        """Every adler32 display_manager takes, while no viewer is watching."""
        import types
        import zlib
        import src.display_manager as display_manager_module
        calls = []

        def counting(data, *args):
            calls.append(len(data))
            return zlib.adler32(data, *args)

        monkeypatch.setattr(display_manager_module, "zlib",
                            types.SimpleNamespace(adler32=counting))
        # The fixture's marker is the real /tmp one, which a preview open on
        # this machine would keep fresh. Point it somewhere nobody touches.
        monkeypatch.setattr(dm, "_viewer_marker_path", str(tmp_path / "no-viewer"))
        monkeypatch.setattr(dm, "_snapshot_path", str(tmp_path / "snap.png"))
        dm._viewer_check_ts = 0.0
        dm._viewer_fresh = False
        dm._viewer_was_fresh = False
        # TestSnapshotOffRenderThread's cleanup test stops this shared
        # manager's snapshot writer for good; queued writes here need one.
        with dm._snapshot_cond:
            dm._snapshot_stop = False
            dm._snapshot_pending = None
        yield calls
        dm.set_scrolling_state(False)
        dm._viewer_check_ts = 0.0
        dm._viewer_fresh = False
        dm._viewer_was_fresh = False

    def _scroll(self, dm, frames):
        """Present ``frames`` scrolled frames, each different from the last."""
        for shade in range(frames):
            dm.set_scrolling_state(True)       # as a scroller does every frame
            dm.draw.rectangle([0, 0, 10, 10], fill=(shade * 10 % 256, 40, 0))
            dm.update_display()

    def test_scrolled_frames_are_not_hashed_when_no_snapshot_is_due(
            self, dm, hashes):
        now = time.time()
        dm._last_snapshot_ts = now             # written a moment ago
        dm._last_snapshot_touch_ts = now
        with _SwapSpy(dm.matrix) as spy:
            self._scroll(dm, 20)
        assert spy.count == 20, "every scrolled frame must still be pushed"
        assert hashes == []

    def test_a_due_snapshot_hashes_the_frame_once(
            self, dm, hashes, monkeypatch):
        import threading
        wrote = threading.Event()
        monkeypatch.setattr(dm, "_save_snapshot", lambda image: wrote.set())
        dm._last_snapshot_ts = 0.0             # a write is due (no viewer:
        dm._last_snapshot_touch_ts = 0.0       # the idle keepalive)
        dm._last_snapshot_digest = None
        self._scroll(dm, 20)
        assert wrote.wait(5), "the due snapshot was never written"
        assert len(hashes) == 1

    def test_with_a_viewer_a_scroll_is_hashed_once_per_write(
            self, dm, hashes, tmp_path, monkeypatch):
        import threading
        from src.common import snapshot_policy
        marker = tmp_path / "viewer"
        marker.touch()
        monkeypatch.setattr(dm, "_viewer_marker_path", str(marker))
        writes = []
        wrote = threading.Event()

        def save(image):
            writes.append(image)
            wrote.set()

        monkeypatch.setattr(dm, "_save_snapshot", save)
        dm._last_snapshot_ts = time.time()     # the viewer opening resets it
        dm._last_snapshot_touch_ts = time.time()
        dm._last_snapshot_digest = None        # every frame here is new
        started = time.time()
        self._scroll(dm, 30)
        elapsed = time.time() - started
        assert wrote.wait(5)
        # One write as the viewer appears, then at most one per interval.
        assert 1 <= len(hashes) <= 1 + int(elapsed / snapshot_policy.VIEWER_INTERVAL)
        assert len(hashes) < 30

    def test_an_unchanged_scrolled_frame_is_not_re_encoded(
            self, dm, hashes, monkeypatch):
        # A write would be due for a changed frame, but this one is already
        # on disk: the lazy hash must re-ask the policy for an unchanged frame
        # (SKIP here) rather than act on the changed-frame answer (WRITE).
        import zlib
        saved, touched = [], []
        monkeypatch.setattr(dm, "_save_snapshot", lambda image: saved.append(image))
        monkeypatch.setattr(os, "utime", lambda *a, **k: touched.append(a))
        dm.set_scrolling_state(True)
        dm.draw.rectangle([0, 0, 10, 10], fill=(3, 3, 3))
        digest = zlib.adler32(dm.image.tobytes())
        dm._last_snapshot_digest = digest
        dm._saved_snapshot_digest = digest
        last_write = time.time() - 1000             # a changed frame would write
        dm._last_snapshot_ts = last_write
        dm._last_snapshot_touch_ts = time.time()    # no touch due
        dm.update_display()
        # A write, even one only queued, would have moved the bookkeeping.
        assert dm._last_snapshot_ts == last_write
        assert dm._snapshot_pending is None
        assert saved == [] and touched == []
        assert len(hashes) == 1

    def test_an_unchanged_scrolled_frame_is_touched_only_once_on_disk(
            self, dm, hashes, monkeypatch):
        # A touch vouches for the file on disk, so through the lazy hash too it
        # must only happen once the frame has actually been saved.
        import threading
        import zlib
        from src.common import snapshot_policy
        saved, touched = [], []
        wrote = threading.Event()

        def save(image):
            saved.append(image)
            wrote.set()

        monkeypatch.setattr(dm, "_save_snapshot", save)
        monkeypatch.setattr(os, "utime", lambda *a, **k: touched.append(a))
        dm.set_scrolling_state(True)
        dm.draw.rectangle([0, 0, 10, 10], fill=(5, 5, 5))
        digest = zlib.adler32(dm.image.tobytes())
        dm._last_snapshot_digest = digest      # queued earlier...
        dm._saved_snapshot_digest = 12345      # ...but an older frame is on disk
        # A touch is due; a write is not (no viewer: the idle interval).
        stale = time.time() - snapshot_policy.TOUCH_INTERVAL - 1.0
        dm._last_snapshot_ts = stale
        dm._last_snapshot_touch_ts = stale
        dm.update_display()
        assert wrote.wait(5), "a frame not yet on disk must be written"
        deadline = time.time() + 5
        while dm._saved_snapshot_digest != digest and time.time() < deadline:
            time.sleep(0.01)
        assert touched == [] and len(saved) == 1

        # Once it is on disk, the same frame is only touched.
        dm._last_snapshot_ts = stale
        dm._last_snapshot_touch_ts = stale
        dm.set_scrolling_state(True)
        dm.update_display()
        assert len(touched) == 1 and len(saved) == 1

    def test_the_first_static_frame_after_a_scroll_is_pushed_once(
            self, dm, hashes):
        # No digest is kept mid-scroll, so a static frame identical to the
        # scroll's last one is pushed once (one swap), and then dirty
        # tracking skips its repeats as before.
        dm.set_scrolling_state(True)
        dm.draw.rectangle([0, 0, 18, 18], fill=(90, 90, 0))
        dm.update_display()                    # the scroll's last frame
        dm.set_scrolling_state(False)
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()
            dm.update_display()
            dm.update_display()
        assert spy.count == 1

    def test_a_scrolled_push_forgets_the_last_static_frame(
            self, dm, hashes):
        # Scrolled frames carry no digest, so a scrolled push must clear the
        # last static one. Were it kept, a static screen drawing what it
        # showed before the scroll would be skipped as already on the panel,
        # and the scroll's last frame would stay up until the content changed.
        dm.set_scrolling_state(False)
        dm.draw.rectangle([0, 0, 18, 18], fill=(10, 120, 30))
        dm.update_display()                    # static frame A
        dm.set_scrolling_state(True)
        dm.draw.rectangle([0, 0, 18, 18], fill=(200, 0, 60))
        dm.update_display()                    # a scrolled frame B
        assert dm._last_pushed_digest is None
        dm.set_scrolling_state(False)
        dm.draw.rectangle([0, 0, 18, 18], fill=(10, 120, 30))
        with _SwapSpy(dm.matrix) as spy:
            dm.update_display()                # A again, over B
        assert spy.count == 1, "A must replace the scroll's last frame"


class TestOneScrollingAnswerPerFrame:
    """update_display() asks is_currently_scrolling() once per frame and uses
    that answer throughout. The call has side effects -- a scroll that has
    gone quiet past its threshold is expired there and its frame hold
    dropped -- so asking again partway through let one frame be presented at
    the scroll's hold and recorded as static."""

    def test_a_scroll_that_times_out_is_presented_without_its_hold(
            self, dm, monkeypatch):
        recorded = []
        monkeypatch.setattr(
            dm.frame_timing, "record",
            lambda blit, wait, hold, scrolling, at: recorded.append(
                (hold, scrolling)))
        dm.set_scrolling_state(True, frame_hold=3)
        # Quiet past the inactivity threshold, and nothing has asked since.
        dm._scrolling_state['last_scroll_activity'] -= (
            dm._scrolling_state['scroll_inactivity_threshold'] + 1.0)
        dm.draw.rectangle([0, 0, 8, 8], fill=(31, 62, 93))
        try:
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
        finally:
            dm.set_scrolling_state(False)
        assert spy.count == 1
        assert spy.last_frame_hold == 1
        assert recorded == [(1, False)]

    def test_scan_compensation_uses_the_frames_answer(
            self, dm, tmp_path, monkeypatch):
        from collections import deque
        monkeypatch.setattr(dm, "_scan_lag_bands", [(16, 32, 1)])
        monkeypatch.setattr(dm, "_scan_history", deque(maxlen=1))
        # No preview and no snapshot due, so only the frame itself asks.
        monkeypatch.setattr(dm, "_viewer_marker_path", str(tmp_path / "no-viewer"))
        dm._viewer_check_ts = 0.0
        dm._last_snapshot_ts = dm._last_snapshot_touch_ts = time.time()
        asked = []
        real = dm.is_currently_scrolling

        def counting():
            asked.append(1)
            return real()

        monkeypatch.setattr(dm, "is_currently_scrolling", counting)
        dm.set_scrolling_state(True)
        try:
            dm.draw.rectangle([0, 0, 8, 8], fill=(17, 34, 51))
            dm.update_display()
        finally:
            dm.set_scrolling_state(False)
            dm._viewer_check_ts = 0.0
        assert len(asked) == 1

    @staticmethod
    def _scan_banded(dm, tmp_path, monkeypatch):
        """Scan-order compensation on, no preview, no snapshot due, and every
        frame handed to the canvases kept (update_display alternates them)."""
        from collections import deque
        monkeypatch.setattr(dm, "_scan_lag_bands", [(16, 32, 1)])
        monkeypatch.setattr(dm, "_scan_history", deque(maxlen=1))
        monkeypatch.setattr(dm, "_viewer_marker_path", str(tmp_path / "no-viewer"))
        dm._viewer_check_ts = 0.0
        dm._last_snapshot_ts = dm._last_snapshot_touch_ts = time.time()
        shown = []
        for canvas in (dm.offscreen_canvas, dm.current_canvas):
            def capture(image, *args, _real=canvas.SetImage, **kwargs):
                shown.append(image.copy())
                return _real(image, *args, **kwargs)
            monkeypatch.setattr(canvas, "SetImage", capture)
        return shown

    def test_a_held_frame_is_split_on_the_frames_answer(
            self, dm, tmp_path, monkeypatch):
        # Scan-order compensation presents a held frame as two swaps, the
        # lagging half stepping one refresh after the rest. The split is
        # decided on the frame's one answer, exactly as when it asked itself.
        shown = self._scan_banded(dm, tmp_path, monkeypatch)
        asked, recorded = [], []
        real = dm.is_currently_scrolling

        def counting():
            asked.append(1)
            return real()

        monkeypatch.setattr(dm, "is_currently_scrolling", counting)
        monkeypatch.setattr(
            dm.frame_timing, "record",
            lambda blit, wait, hold, scrolling, at: recorded.append(
                (hold, scrolling)))
        dm.set_scrolling_state(True, frame_hold=2)
        try:
            # The fixture is shared: a slow blit left by an earlier test
            # would send this frame out whole and drop the history.
            dm._last_blit_seconds = 0.0
            dm.draw.rectangle([0, 0, dm.width - 1, dm.height - 1],
                              fill=(10, 0, 0))
            dm.update_display()                # the previous frame
            asked.clear()
            recorded.clear()
            before = len(shown)
            dm._last_blit_seconds = 0.0        # a blit that fits a refresh
            dm.set_scrolling_state(True, frame_hold=2)
            dm.draw.rectangle([0, 0, dm.width - 1, dm.height - 1],
                              fill=(20, 0, 0))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
        finally:
            dm.set_scrolling_state(False)
            dm._viewer_check_ts = 0.0
        assert len(asked) == 1
        assert spy.holds == [1, 1]
        first, second = shown[before:]
        assert first.getpixel((0, 0)) == (20, 0, 0)
        assert first.getpixel((0, 31)) == (10, 0, 0)    # lagging half: old
        assert second.getpixel((0, 31)) == (20, 0, 0)   # a refresh later
        assert recorded == [(2, True)]

    def test_a_held_scroll_that_times_out_goes_out_whole_at_hold_1(
            self, dm, tmp_path, monkeypatch):
        # The expiry frame with scan-order compensation on: one swap of the
        # frame as drawn, at hold 1, the history dropped -- and the pacing
        # gate, the swap and frame timing all see that one hold. (Asking
        # partway through instead had the swap use the dead scroll's hold
        # while the gate and frame timing were told 1.)
        shown = self._scan_banded(dm, tmp_path, monkeypatch)
        gate_holds, recorded = [], []

        class _Gate:
            def before_swap(self, hold):
                gate_holds.append(('before', hold))

            def after_swap(self, hold):
                gate_holds.append(('after', hold))

        monkeypatch.setattr(
            dm.frame_timing, "record",
            lambda blit, wait, hold, scrolling, at: recorded.append(
                (hold, scrolling)))
        dm.set_scrolling_state(True, frame_hold=3)
        try:
            dm.draw.rectangle([0, 0, dm.width - 1, dm.height - 1],
                              fill=(30, 0, 0))
            dm.update_display()                # the scroll's last frame
            recorded.clear()
            before = len(shown)
            # Quiet past the inactivity threshold, and nothing has asked since.
            dm._scrolling_state['last_scroll_activity'] -= (
                dm._scrolling_state['scroll_inactivity_threshold'] + 1.0)
            dm._last_blit_seconds = 0.0        # a split would fit
            monkeypatch.setattr(dm, "render_gate", _Gate())
            dm.draw.rectangle([0, 0, dm.width - 1, dm.height - 1],
                              fill=(40, 0, 0))
            with _SwapSpy(dm.matrix) as spy:
                dm.update_display()
        finally:
            dm.set_scrolling_state(False)
            dm._viewer_check_ts = 0.0
        assert spy.holds == [1]
        assert gate_holds == [('before', 1), ('after', 1)]
        assert recorded == [(1, False)]
        (frame,) = shown[before:]
        assert frame.getpixel((0, 31)) == (40, 0, 0)    # not composed
        assert len(dm._scan_history) == 0


class TestSnapshotEncoding:
    def test_the_snapshot_is_written_at_the_fastest_level_and_lossless(
            self, dm, tmp_path, monkeypatch):
        # The encode is paid on the render thread for a static screen, and
        # the file is read once; level 1 cuts the encode, and PNG is lossless
        # at any level, so the preview shows exactly the frame.
        from PIL import Image, ImageChops
        levels = []
        real_save = Image.Image.save

        def recording(image, fp, format=None, **params):
            levels.append(params.get("compress_level"))
            return real_save(image, fp, format, **params)

        monkeypatch.setattr(Image.Image, "save", recording)
        monkeypatch.setattr(dm, "_snapshot_path", str(tmp_path / "snap.png"))
        frame = Image.new("RGB", (dm.width, dm.height))
        frame.paste((200, 10, 90), (3, 4, 40, 20))
        frame.putpixel((0, 0), (1, 2, 3))
        dm._save_snapshot(frame)
        assert levels == [1]
        with Image.open(dm._snapshot_path) as saved:
            assert saved.format == "PNG"
            assert ImageChops.difference(
                saved.convert("RGB"), frame).getbbox() is None