fix(testing): clear get_cached_data_with_strategy_calls in MockCacheManager.reset()

reset() cleared get_calls/set_calls/delete_calls but not the newer
get_cached_data_with_strategy_calls tracker, so a reused mock (e.g. across
test cases sharing a fixture) retained stale strategy-call records after
reset().

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ

src/cache/disk_cache.py

@@ -10,7 +10,6 @@ import time
 import tempfile
 import logging
 import threading
-import zlib
 from typing import Dict, Any, Optional, Protocol
 from datetime import datetime
 
@@ -54,11 +53,6 @@ class DiskCache:
         self.cache_dir = cache_dir
         self.logger = logger or logging.getLogger(__name__)
         self._lock = threading.Lock()
-        # key -> adler32 of the last payload successfully written to the
-        # primary cache path; lets set() skip rewriting identical data
-        # (per-process only — worst case another process rewrites, never
-        # a missed write). Guarded by _lock.
-        self._write_digests: Dict[str, int] = {}
     
     def get_cache_path(self, key: str) -> Optional[str]:
         """
@@ -161,35 +155,10 @@ class DiskCache:
         cache_path = self.get_cache_path(key)
         if not cache_path:
             return
-
+        
-        # Serialize once, compact (no indent): the payload is reused by every
-        # write path below, and cache files are machine-read only — indenting
-        # them just multiplied the bytes written to the SD card.
-        try:
-            payload = json.dumps(data, cls=DateTimeEncoder)
-        except (TypeError, ValueError) as e:
-            self.logger.warning("Cache data for key '%s' not serializable: %s", key, e)
-            return
-
-        digest = zlib.adler32(payload.encode('utf-8'))
-
         try:
             # Atomic write to avoid partial/corrupt files
             with self._lock:
-                # Skip the disk entirely when this exact payload was already
-                # written for this key (plugins re-save unchanged API data
-                # every update cycle — each write is real SD-card wear).
-                # Refresh the file mtime so records that rely on it for TTL
-                # (no embedded 'timestamp') don't expire early; a metadata
-                # touch is journal-cheap compared to rewriting the data.
-                if self._write_digests.get(key) == digest:
-                    try:
-                        os.utime(cache_path, None)
-                        return
-                    except OSError:
-                        # File vanished or perms changed — fall through and write
-                        self._write_digests.pop(key, None)
-
                 tmp_dir = os.path.dirname(cache_path)
                 # Try to create temp file in cache directory first
                 # If that fails due to permissions, fall back to direct write
@@ -212,17 +181,13 @@ class DiskCache:
                         fd = None
                     
                     if tmp_path and fd is not None:
-                        # Atomic write with temp file. No fsync: os.replace
+                        # Use atomic write with temp file
-                        # already guarantees readers never see a torn file,
-                        # and cache data is re-fetchable — forcing a disk
-                        # flush per write was the single biggest SD-card
-                        # wear source (dozens of fsyncs/min on API-heavy
-                        # installs) for data that can be re-downloaded.
                         try:
                             with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file:
-                                tmp_file.write(payload)
+                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
+                                tmp_file.flush()
+                                os.fsync(tmp_file.fileno())
                             os.replace(tmp_path, cache_path)
-                            self._write_digests[key] = digest
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -238,8 +203,9 @@ class DiskCache:
                         # Fallback: direct write (not atomic, but better than failing)
                         try:
                             with open(cache_path, 'w', encoding='utf-8') as cache_file:
-                                cache_file.write(payload)
+                                json.dump(data, cache_file, indent=4, cls=DateTimeEncoder)
-                            self._write_digests[key] = digest
+                                cache_file.flush()
+                                os.fsync(cache_file.fileno())
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -263,12 +229,9 @@ class DiskCache:
                             pass
                         
                         if os.path.isdir(fallback_dir) and os.access(fallback_dir, os.W_OK):
-                            # NOTE: no digest record here — the fallback file
-                            # is a different path, so future sets must keep
-                            # retrying the primary location.
                             fallback_path = os.path.join(fallback_dir, os.path.basename(cache_path))
                             with open(fallback_path, 'w', encoding='utf-8') as tmp_file:
-                                tmp_file.write(payload)
+                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(fallback_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -309,7 +272,6 @@ class DiskCache:
         
         with self._lock:
             if key:
-                self._write_digests.pop(key, None)
                 cache_path = self.get_cache_path(key)
                 if cache_path and os.path.exists(cache_path):
                     try:
@@ -318,7 +280,6 @@ class DiskCache:
                         self.logger.warning("Could not remove cache file %s: %s", cache_path, e)
             else:
                 # Clear all cache files
-                self._write_digests.clear()
                 if os.path.exists(self.cache_dir):
                     for filename in os.listdir(self.cache_dir):
                         if filename.endswith('.json'):

src/common/snapshot_policy.py

@@ -1,68 +0,0 @@
-"""Snapshot write policy for the display preview mirror.
-
-The display service mirrors frames to /tmp/led_matrix_preview.png, which
-serves two consumers with different needs:
-
-- The web UI's live preview (SSE reader in web_interface/app.py) wants
-  fresh frames — but only while a browser is actually watching.
-- The health check (web_interface/blueprints/api_v3.py, hardware status)
-  uses the file's AGE as a liveness proxy: age >= 60s reads as degraded.
-
-PNG-encoding every frame at 5 fps forever — identical frames, no viewers —
-was one of the biggest fixed CPU costs on the Pi. This module is the pure
-decision logic (extracted so it's unit-testable off-Pi; display_manager
-imports rgbmatrix unconditionally and can't be):
-
-    WRITE  — encode + atomically replace the snapshot file
-    TOUCH  — os.utime only: keeps the health-check mtime fresh and lets
-             the SSE reader (mtime-gated) resend at a low rate, without
-             paying for a PNG encode of an unchanged frame
-    SKIP   — do nothing
-
-Policy:
-- With a fresh viewer marker: changed frames write at up to 1/VIEWER_INTERVAL.
-- Without viewers: changed frames still write at 1/IDLE_INTERVAL so the
-  preview page shows something recent on open.
-- Unchanged frames are never re-encoded; the mtime is touched every
-  TOUCH_INTERVAL so the health check (60s threshold) never degrades.
-
-If any constant here changes, re-check the health threshold in
-api_v3.py (get_hardware_status) — TOUCH_INTERVAL must stay well under it.
-"""
-
-from enum import Enum
-
-# Snapshot cadence with a browser preview open (seconds).
-VIEWER_INTERVAL = 0.2
-# Snapshot cadence with no viewers — cheap freshness for page-open (seconds).
-IDLE_INTERVAL = 30.0
-# Max age of the last write/touch before bumping mtime for the health
-# check. MUST stay well under api_v3's 60s degraded threshold.
-TOUCH_INTERVAL = 20.0
-# A viewer marker older than this no longer counts as a live viewer.
-VIEWER_MARKER_FRESH_SEC = 5.0
-
-
-class SnapshotAction(Enum):
-    WRITE = "write"
-    TOUCH = "touch"
-    SKIP = "skip"
-
-
-def decide(now: float, last_write_ts: float, last_touch_ts: float,
-           viewer_fresh: bool, frame_changed: bool) -> SnapshotAction:
-    """Decide what to do with the current frame.
-
-    Args:
-        now: current monotonic-ish timestamp (same clock as the ts args)
-        last_write_ts: when a frame was last actually encoded+written
-        last_touch_ts: when the file mtime was last bumped (write or touch)
-        viewer_fresh: a browser preview is currently watching
-        frame_changed: the frame differs from the last WRITTEN frame
-    """
-    interval = VIEWER_INTERVAL if viewer_fresh else IDLE_INTERVAL
-    if frame_changed and (now - last_write_ts) >= interval:
-        return SnapshotAction.WRITE
-    if (now - max(last_write_ts, last_touch_ts)) >= TOUCH_INTERVAL:
-        return SnapshotAction.TOUCH
-    return SnapshotAction.SKIP

src/display_controller.py

@@ -199,10 +199,6 @@ class DisplayController:
         self.wifi_status_file = WIFI_STATUS_FILE
         self.wifi_status_active = False
         self.wifi_status_expires_at: Optional[float] = None
-        # _check_wifi_status_message throttle state (checked at frame rate,
-        # stat'd at most once per second)
-        self._wifi_status_check_ts = 0.0
-        self._wifi_status_last_result: Optional[Dict[str, Any]] = None
 
         # Plugin display() signature cache — must be initialised before the plugin
         # loading loop below so the .pop() invalidation at load time is always safe.
@@ -1688,7 +1684,7 @@ class DisplayController:
                     self._sleep_with_plugin_updates(60)
                     continue
                 
-                logger.debug("Display active, processing mode: %s", self.current_display_mode)
+                logger.info(f"Display active, processing mode: {self.current_display_mode}")
                 
                 # Plugins update on their own schedules - no forced sync updates needed
                 # Each plugin has its own update_interval and background services
@@ -1856,7 +1852,7 @@ class DisplayController:
                         if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
                             should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
                             if should_skip:
-                                logger.info("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode)
+                                logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
                                 display_result = False
                                 # Skip to next mode - let existing logic handle it
                                 manager_to_display = None
@@ -1914,7 +1910,7 @@ class DisplayController:
                             if isinstance(result, bool):
                                 display_result = result
                                 if not display_result:
-                                    logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
+                                    logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
                         
                         # Record success if display completed without exception
                         if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
@@ -2407,16 +2403,6 @@ class DisplayController:
             Returns None on any error or if message is expired/invalid.
         """
         try:
-            # Throttle the existence stat to ~1 Hz: this runs on every render
-            # iteration (60+ fps), and the file usually doesn't exist — the
-            # status message's lifetime is measured in seconds anyway.
-            # Both attributes are initialised in __init__.
-            now = time.time()
-            if (now - self._wifi_status_check_ts) < 1.0:
-                return self._wifi_status_last_result
-            self._wifi_status_check_ts = now
-            self._wifi_status_last_result = None
-
             # Check if file exists
             if not self.wifi_status_file or not self.wifi_status_file.exists():
                 return None
@@ -2467,14 +2453,13 @@ class DisplayController:
                     pass
                 return None
             
-            # Message is valid and not expired — cache for the throttle window
+            # Message is valid and not expired
-            self._wifi_status_last_result = {
+            return {
                 'message': message,
                 'timestamp': timestamp,
                 'duration': duration,
                 'expires_at': expires_at
             }
-            return self._wifi_status_last_result
             
         except Exception as e:
             # Catch-all for any unexpected errors - log but don't break the display

src/display_manager.py

@@ -31,24 +31,13 @@ if os.getenv("EMULATOR", "false") == "true":
 else:
     from rgbmatrix import RGBMatrix, RGBMatrixOptions
 from contextlib import contextmanager
-from pathlib import Path
 from PIL import Image, ImageDraw, ImageFont
 import time
-from collections import OrderedDict
+from typing import Dict, Any, List, Optional
-from typing import Dict, Any, List, Optional, Tuple
 import logging
 import math
-import zlib
 import freetype
 
-from src.common import snapshot_policy
-from src.common.permission_utils import (
-    ensure_directory_permissions,
-    ensure_file_permissions,
-    get_assets_dir_mode,
-    get_assets_file_mode,
-)
-
 # Get logger without configuring
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)  # Set to INFO level
@@ -191,34 +180,14 @@ class DisplayManager:
         # the logical image is blitted to the matrix unchanged.
         self._double_sided = None  # dict {copies, axis, logical_width, logical_height} or None
         self._physical_image = None  # full-chain buffer reused each frame when tiling
-        # Text-width measurement cache: (text, id(font)) -> (width, font_ref)
+        # Text-width measurement cache: (text, id(font)) -> pixel_width
         # Avoids re-measuring the same string+font on every display() call.
-        # LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
-        # live score) would otherwise grow it forever on a 24/7 service.
-        # Entries hold a strong reference to the font so its id() can't be
-        # recycled by a different font object — an id-keyed cache without
-        # the reference can return the WRONG width after garbage collection.
         # Cleared on _load_fonts() so stale entries don't survive a font reload.
-        self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
+        self._text_width_cache: Dict[tuple, int] = {}
-        self._TEXT_WIDTH_CACHE_MAX = 1024
+        # Snapshot settings for web preview integration (service writes, web reads)
-        # Snapshot mirror for web preview + health check (service writes, web
-        # reads). Cadence/skip decisions live in src/common/snapshot_policy.py:
-        # full rate only while the web SSE broadcaster keeps the viewer marker
-        # fresh; unchanged frames are never re-encoded, only mtime-touched.
         self._snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path intentional; web UI reads same path
-        self._viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - touched by web SSE broadcaster
+        self._snapshot_min_interval_sec = 0.2  # max ~5 fps
         self._last_snapshot_ts = 0.0
-        self._last_snapshot_touch_ts = 0.0
-        self._last_snapshot_digest: Optional[int] = None
-        self._snapshot_dir_prepared = False
-        self._viewer_check_ts = 0.0
-        self._viewer_fresh = False
-        self._viewer_was_fresh = False
-        # Snapshot failures are logged as warnings, rate-limited so a
-        # persistent failure (e.g. an unwritable file) can't spam the log —
-        # but is never silent: the snapshot's mtime doubles as the web UI's
-        # hardware-liveness signal, so a quiet failure makes health checks lie.
-        self._snapshot_fail_log_ts = 0.0
         
         # Scrolling state tracking for graceful updates
         self._scrolling_state = {
@@ -730,15 +699,12 @@ class DisplayManager:
 
         Results are cached by (text, font identity) so plugins that measure
         the same string every frame (e.g. to centre a score) pay only one
-        measurement per unique (text, font) pair. The entry keeps the font
+        measurement per unique (text, font) pair.
-        alive so its id() can't be recycled, and the cache is LRU-bounded so
-        ever-changing text (clocks, tickers) can't grow it without limit.
         """
         cache_key = (text, id(font))
         cached = self._text_width_cache.get(cache_key)
         if cached is not None:
-            self._text_width_cache.move_to_end(cache_key)
+            return cached
-            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -753,9 +719,7 @@ class DisplayManager:
             logger.error("Error getting text width: %s", e)
             return 0
 
-        self._text_width_cache[cache_key] = (width, font)
+        self._text_width_cache[cache_key] = width
-        while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
-            self._text_width_cache.popitem(last=False)
         return width
 
     def get_font_height(self, font):
@@ -1164,56 +1128,27 @@ class DisplayManager:
             'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
         }
 
-    def _viewer_is_fresh(self, now: float) -> bool:
-        """True when a browser preview is watching (marker file touched by
-        the web SSE broadcaster). The marker is stat'd at most once per
-        second — at 125 fps loops a per-call stat would be pure overhead."""
-        if (now - self._viewer_check_ts) >= 1.0:
-            self._viewer_check_ts = now
-            try:
-                marker_age = now - os.stat(self._viewer_marker_path).st_mtime
-                self._viewer_fresh = marker_age < snapshot_policy.VIEWER_MARKER_FRESH_SEC
-            except OSError:
-                self._viewer_fresh = False
-        return self._viewer_fresh
-
     def _write_snapshot_if_due(self) -> None:
-        """Mirror the current frame to the preview snapshot when the policy
+        """Write the current image to a PNG snapshot file at a limited frequency."""
-        says it's worth it — see src/common/snapshot_policy.py. Unchanged
-        frames are never re-encoded; without viewers the cadence drops to
-        the idle keepalive."""
         try:
             now = time.time()
-            viewer_fresh = self._viewer_is_fresh(now)
+            if (now - self._last_snapshot_ts) < self._snapshot_min_interval_sec:
-            if viewer_fresh and not self._viewer_was_fresh:
-                # A preview just opened: let the next changed frame through
-                # immediately instead of waiting out the idle interval.
-                self._last_snapshot_ts = 0.0
-            self._viewer_was_fresh = viewer_fresh
-
-            digest = zlib.adler32(self.image.tobytes())
-            action = snapshot_policy.decide(
-                now, self._last_snapshot_ts, self._last_snapshot_touch_ts,
-                viewer_fresh, digest != self._last_snapshot_digest)
-            if action is snapshot_policy.SnapshotAction.SKIP:
                 return
-            if action is snapshot_policy.SnapshotAction.TOUCH:
+            # Ensure directory exists with proper permissions
-                # mtime bump only: keeps the health check (snapshot age)
+            from pathlib import Path
-                # green without paying for a PNG encode of an unchanged frame
+            from src.common.permission_utils import (
-                os.utime(self._snapshot_path, None)
+                ensure_directory_permissions,
-                self._last_snapshot_touch_ts = now
+                ensure_file_permissions,
-                return
+                get_assets_dir_mode,
-
+                get_assets_file_mode
-            # WRITE: ensure directory permissions once, not per frame
+            )
             snapshot_path_obj = Path(self._snapshot_path)
-            if not self._snapshot_dir_prepared:
+            # Only ensure permissions on non-system directories
-                # Never modify /tmp permissions - it has special system
+            # Never modify /tmp permissions - it has special system permissions (1777)
-                # permissions (1777) that must not be changed or it breaks
+            # that must not be changed or it breaks apt and other system tools
-                # apt and other system tools
+            parent_dir = snapshot_path_obj.parent
-                parent_dir = snapshot_path_obj.parent
+            if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
-                if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
+                ensure_directory_permissions(parent_dir, get_assets_dir_mode())
-                    ensure_directory_permissions(parent_dir, get_assets_dir_mode())
-                self._snapshot_dir_prepared = True
             # Write atomically: temp then replace
             tmp_path = f"{self._snapshot_path}.tmp"
             self.image.save(tmp_path, format='PNG')
@@ -1228,18 +1163,6 @@ class DisplayManager:
             except Exception:
                 pass
             self._last_snapshot_ts = now
-            self._last_snapshot_touch_ts = now
-            self._last_snapshot_digest = digest
         except Exception as e:
-            # Snapshot failures must never break display — but they must not
+            # Snapshot failures should never break display; log at debug to avoid noise
-            # be silent either: the snapshot's mtime is the web UI's display
+            logger.debug(f"Snapshot write skipped: {e}")
-            # mirror AND its hardware-liveness proxy, so a quietly failing
-            # write freezes the mirror and makes health checks lie (seen in
-            # the field: a stale root-owned /tmp file froze it for a day).
-            # Warn at most once per 5 minutes to avoid log spam.
-            if (now - self._snapshot_fail_log_ts) > 300:
-                self._snapshot_fail_log_ts = now
-                logger.warning("Snapshot write failing (web preview/health "
-                               "mirror is stale): %s", e)
-            else:
-                logger.debug(f"Snapshot write skipped: {e}")

src/font_manager.py

@@ -35,7 +35,6 @@ import urllib.request
 import zipfile
 import tempfile
 import time
-from collections import OrderedDict
 from pathlib import Path
 from PIL import ImageFont
 from typing import Dict, Tuple, Optional, Union, Any, List
@@ -59,13 +58,7 @@ class FontManager:
         # Font discovery and catalog
         self.font_catalog: Dict[str, str] = {}  # family_name -> file_path
         self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {}  # (family, size) -> font
-        # (text, id(font)) -> ((width, height, baseline), font_ref).
+        self.metrics_cache: Dict[str, Tuple[int, int, int]] = {}  # (text, font_id) -> (width, height, baseline)
-        # LRU-bounded — keys embed the measured TEXT, so changing strings
-        # (clocks, live scores) would otherwise grow it forever. Entries
-        # keep the font alive so its id() can't be recycled by a different
-        # font object (which would silently return wrong metrics).
-        self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
-        self._METRICS_CACHE_MAX = 1024
 
         # Plugin font management
         self.plugin_fonts: Dict[str, Dict[str, Any]] = {}  # plugin_id -> font_manifest
@@ -562,14 +555,10 @@ class FontManager:
         Returns:
             Tuple of (width, height, baseline_offset)
         """
-        # Key on the text itself (hash(text) could collide) + font identity;
+        cache_key = f"{hash(text)}_{id(font)}"
-        # the entry below keeps the font referenced so the id stays valid.
-        cache_key = (text, id(font))
 
-        cached = self.metrics_cache.get(cache_key)
+        if cache_key in self.metrics_cache:
-        if cached is not None:
+            return self.metrics_cache[cache_key]
-            self.metrics_cache.move_to_end(cache_key)
-            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -606,9 +595,7 @@ class FontManager:
             baseline = 10
 
         result = (width, height, baseline)
-        self.metrics_cache[cache_key] = (result, font)
+        self.metrics_cache[cache_key] = result
-        while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
-            self.metrics_cache.popitem(last=False)
         return result
 
     def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:

src/plugin_system/testing/mocks.py

@@ -71,6 +71,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
         # Real temp dir for plugins that write/read files under cache_dir.
         # Registered for cleanup so each mock instance doesn't leak a tmp dir.
         self.cache_dir = tempfile.mkdtemp(prefix="ledmatrix-mock-cache-")
@@ -108,6 +109,24 @@ class MockCacheManager:
         self.delete_calls.append(key)
         if key in self._cache:
             del self._cache[key]
+
+    def get_cached_data_with_strategy(self, key: str, data_type: str = 'default') -> Optional[Any]:
+        """Mock of CacheManager.get_cached_data_with_strategy (src/cache_manager.py).
+
+        The real method picks a max_age/memory_ttl strategy per data_type
+        (and extends it during market-closed hours for market data) before
+        delegating to get_cached_data(). None of that timing nuance matters
+        for a mock -- plugins under test just need the method to exist and
+        return whatever was cached, so this delegates straight to get().
+        """
+        self.get_cached_data_with_strategy_calls.append({'key': key, 'data_type': data_type})
+        return self.get(key)
+
+    def save_cache(self, key: str, data: Any) -> None:
+        """Mock of CacheManager.save_cache (src/cache_manager.py) -- the
+        write-side counterpart to get_cached_data_with_strategy, used by the
+        same real-CacheManager-oriented plugins. Delegates to set()."""
+        self.set(key, data)
         if key in self._cache_timestamps:
             del self._cache_timestamps[key]
     
@@ -118,6 +137,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
 
 
 class MockConfigManager:

src/vegas_mode/render_pipeline.py

@@ -66,10 +66,6 @@ class RenderPipeline:
             else display_manager.height
         )
 
-        # Reusable blank frame for cycle-end pushes (allocated lazily,
-        # re-blacked before each reuse)
-        self._blank_frame = None
-
         # ScrollHelper for optimized scrolling
         self.scroll_helper = ScrollHelper(
             self.display_width,
@@ -238,19 +234,11 @@ class RenderPipeline:
                     )
                     # Push blank immediately so the hardware never shows any
                     # post-wrap content while the coordinator recomposes the
-                    # next cycle (~100 ms). The blank is allocated once and
+                    # next cycle (~100 ms).
-                    # reused across cycle wraps (fresh paste each time in case
-                    # a consumer drew on the previous one).
                     try:
-                        if self._blank_frame is None or self._blank_frame.size != (
+                        from PIL import Image as _Image
-                                self.display_width, self.display_height):
+                        blank = _Image.new('RGB', (self.display_width, self.display_height))
-                            self._blank_frame = Image.new(
+                        self.display_manager.image = blank
-                                'RGB', (self.display_width, self.display_height))
-                        else:
-                            self._blank_frame.paste(
-                                (0, 0, 0),
-                                (0, 0, self.display_width, self.display_height))
-                        self.display_manager.image = self._blank_frame
                         self.display_manager.update_display()
                     except Exception:
                         logger.exception("Failed to write blank frame to display at cycle end")

test/conftest.py

@@ -38,11 +38,7 @@ def mock_cache_manager():
     mock._memory_cache_timestamps = {}
     mock.cache_dir = "/tmp/test_cache"
     
-    def mock_get(key: str, max_age: Optional[int] = 300,
+    def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
-                 memory_ttl: Optional[int] = None) -> Optional[Dict]:
-        # Signature mirrors CacheManager.get — keep in sync or callers
-        # passing keyword args (health tracker, resource monitor) break
-        # only in tests, hiding real-API compatibility.
         return mock._memory_cache.get(key)
     
     def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:

test/test_cache_manager.py

@@ -400,61 +400,3 @@ class TestDiskCache:
         assert stats['fetch_count'] == 3
         assert stats['total_fetch_time'] == 1.8
         assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
-
-
-class TestDiskCacheWriteEconomy:
-    """SD-card wear guards: identical payloads skip the disk, files are
-    compact, and TTL semantics survive the skip (see PR: fix/diskcache-sd-wear)."""
-
-    def test_identical_set_skips_rewrite(self, tmp_path):
-        import os
-        cache = DiskCache(cache_dir=str(tmp_path))
-        cache.set("k", {"data": "v"})
-        path = cache.get_cache_path("k")
-        first = os.stat(path)
-        os.utime(path, (first.st_atime - 100, first.st_mtime - 100))  # age it
-        aged_mtime = os.stat(path).st_mtime
-        ino_before = os.stat(path).st_ino
-        cache.set("k", {"data": "v"})  # identical payload
-        after = os.stat(path)
-        # mtime refreshed (TTL for mtime-based records preserved)...
-        assert after.st_mtime > aged_mtime
-        # ...but the file was NOT rewritten (same inode: no replace happened)
-        assert after.st_ino == ino_before
-
-    def test_changed_data_rewrites(self, tmp_path):
-        import os
-        cache = DiskCache(cache_dir=str(tmp_path))
-        cache.set("k", {"data": "v1"})
-        cache.set("k", {"data": "v2"})
-        assert cache.get("k") == {"data": "v2"}
-
-    def test_clear_resets_digest(self, tmp_path):
-        import os
-        cache = DiskCache(cache_dir=str(tmp_path))
-        cache.set("k", {"data": "v"})
-        cache.clear("k")
-        assert cache.get("k") is None
-        cache.set("k", {"data": "v"})  # same payload after clear must WRITE
-        assert cache.get("k") == {"data": "v"}
-
-    def test_skip_self_heals_when_file_deleted_externally(self, tmp_path):
-        import os
-        cache = DiskCache(cache_dir=str(tmp_path))
-        cache.set("k", {"data": "v"})
-        os.remove(cache.get_cache_path("k"))  # e.g. expiry cleanup
-        cache.set("k", {"data": "v"})  # digest matches but file is gone
-        assert cache.get("k") == {"data": "v"}
-
-    def test_files_are_compact_json(self, tmp_path):
-        cache = DiskCache(cache_dir=str(tmp_path))
-        cache.set("k", {"a": 1, "b": [1, 2, 3]})
-        raw = open(cache.get_cache_path("k")).read()
-        assert "\n" not in raw.strip()  # no indent
-        assert cache.get("k") == {"a": 1, "b": [1, 2, 3]}
-
-    def test_datetime_round_trip_still_works(self, tmp_path):
-        from datetime import datetime
-        cache = DiskCache(cache_dir=str(tmp_path))
-        cache.set("k", {"when": datetime(2026, 7, 12, 10, 30)})
-        assert cache.get("k") == {"when": "2026-07-12T10:30:00"}

test/test_snapshot_policy.py

@@ -1,93 +0,0 @@
-"""Tests for the snapshot write policy (src/common/snapshot_policy.py).
-
-The invariants that matter:
-- unchanged frames are NEVER re-encoded (the old code PNG-encoded identical
-  frames at 5 fps, 24/7)
-- the file mtime never goes stale enough to trip the health check's 60s
-  degraded threshold (api_v3 get_hardware_status)
-- a viewer gets full cadence; no viewer drops to the idle keepalive
-"""
-
-import os
-import sys
-
-import pytest
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
-from src.common.snapshot_policy import (  # noqa: E402
-    IDLE_INTERVAL,
-    TOUCH_INTERVAL,
-    VIEWER_INTERVAL,
-    SnapshotAction,
-    decide,
-)
-
-
-class TestViewerCadence:
-    def test_changed_frame_with_viewer_writes_at_full_rate(self):
-        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL,
-                      last_touch_ts=0, viewer_fresh=True,
-                      frame_changed=True) is SnapshotAction.WRITE
-
-    def test_changed_frame_with_viewer_respects_min_interval(self):
-        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL / 2,
-                      last_touch_ts=100.0, viewer_fresh=True,
-                      frame_changed=True) is SnapshotAction.SKIP
-
-    def test_unchanged_frame_with_viewer_never_writes(self):
-        """A static screen with a viewer must not burn PNG encodes."""
-        assert decide(now=100.0, last_write_ts=90.0, last_touch_ts=90.0,
-                      viewer_fresh=True,
-                      frame_changed=False) is SnapshotAction.SKIP
-
-
-class TestIdleCadence:
-    def test_changed_frame_without_viewer_waits_for_idle_interval(self):
-        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL / 2,
-                      last_touch_ts=100.0, viewer_fresh=False,
-                      frame_changed=True) is SnapshotAction.SKIP
-
-    def test_changed_frame_without_viewer_writes_at_idle_rate(self):
-        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL,
-                      last_touch_ts=0, viewer_fresh=False,
-                      frame_changed=True) is SnapshotAction.WRITE
-
-
-class TestHealthKeepalive:
-    def test_stale_mtime_gets_touched(self):
-        """Whatever else happens, mtime must be bumped within TOUCH_INTERVAL
-        so the health check (60s threshold) never reads the display as dead."""
-        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL,
-                      last_touch_ts=100.0 - TOUCH_INTERVAL, viewer_fresh=False,
-                      frame_changed=False) is SnapshotAction.TOUCH
-
-    def test_touch_applies_with_viewer_too(self):
-        """Viewer watching a static screen: no writes, but health stays green."""
-        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL - 1,
-                      last_touch_ts=100.0 - TOUCH_INTERVAL - 1, viewer_fresh=True,
-                      frame_changed=False) is SnapshotAction.TOUCH
-
-    def test_recent_touch_suppresses_another(self):
-        assert decide(now=100.0, last_write_ts=0.0,
-                      last_touch_ts=100.0 - TOUCH_INTERVAL / 2, viewer_fresh=False,
-                      frame_changed=False) is SnapshotAction.SKIP
-
-    def test_touch_interval_stays_under_health_threshold(self):
-        """api_v3's hardware status treats snapshot age >= 60s as degraded.
-        Keep a 2x margin so scheduling jitter can't trip it."""
-        assert TOUCH_INTERVAL <= 30
-
-    def test_worst_case_mtime_age_is_bounded(self):
-        """Simulate any interleaving: from any state, within one policy call
-        after TOUCH_INTERVAL elapses, mtime gets refreshed (WRITE or TOUCH)."""
-        for viewer in (True, False):
-            for changed in (True, False):
-                action = decide(now=1000.0, last_write_ts=900.0,
-                                last_touch_ts=900.0, viewer_fresh=viewer,
-                                frame_changed=changed)
-                assert action in (SnapshotAction.WRITE, SnapshotAction.TOUCH)
-
-
-if __name__ == "__main__":
-    sys.exit(pytest.main([__file__, "-v"]))

test/test_testing_mocks.py

@@ -0,0 +1,45 @@
+"""
+Unit tests for src/plugin_system/testing/mocks.py.
+
+MockCacheManager/MockPluginManager stand in for the real production
+managers under the plugin safety harness -- a missing method here isn't a
+harness bug in the abstract, it's a plugin silently failing to render
+under test (confirmed on ledmatrix-leaderboard, which calls
+get_cached_data_with_strategy() and previously hit an AttributeError that
+its own broad except swallowed, producing an empty-but-green render).
+"""
+
+from src.plugin_system.testing.mocks import MockCacheManager
+
+
+class TestMockCacheManagerStrategyMethod:
+    def test_get_cached_data_with_strategy_returns_cached_value(self):
+        cm = MockCacheManager()
+        cm.set("standings_nfl", {"teams": ["KC", "BUF"]})
+        result = cm.get_cached_data_with_strategy("standings_nfl", "sports_live")
+        assert result == {"teams": ["KC", "BUF"]}
+
+    def test_get_cached_data_with_strategy_returns_none_when_missing(self):
+        cm = MockCacheManager()
+        assert cm.get_cached_data_with_strategy("missing_key") is None
+
+    def test_get_cached_data_with_strategy_defaults_data_type(self):
+        cm = MockCacheManager()
+        cm.set("k", "v")
+        assert cm.get_cached_data_with_strategy("k") == "v"
+
+    def test_calls_are_tracked(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        assert cm.get_cached_data_with_strategy_calls == [{"key": "k", "data_type": "sports_live"}]
+
+    def test_save_cache_is_readable_via_strategy_lookup(self):
+        cm = MockCacheManager()
+        cm.save_cache("standings_nfl", {"teams": ["KC", "BUF"]})
+        assert cm.get_cached_data_with_strategy("standings_nfl") == {"teams": ["KC", "BUF"]}
+
+    def test_reset_clears_strategy_call_tracking(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        cm.reset()
+        assert cm.get_cached_data_with_strategy_calls == []

web_interface/app.py

@@ -608,23 +608,9 @@ def display_preview_generator():
     import base64
     from PIL import Image
     import io
-
+    
     snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path matches display_manager; only read here
-    # Viewer marker: this generator only runs while the broadcaster has
-    # subscribers (it exits with no clients), so touching the marker each
-    # loop tells the DISPLAY service a browser is actually watching — it
-    # only pays for full-rate PNG snapshot encodes while this stays fresh
-    # (see src/common/snapshot_policy.py).
-    viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - fixed path matches display_manager
     last_modified = None
-
-    def _touch_viewer_marker():
-        try:
-            with open(viewer_marker_path, 'a'):
-                pass
-            os.utime(viewer_marker_path, None)
-        except OSError:
-            pass  # display side treats a missing marker as "no viewer"
     
     # Get display dimensions from config
     try:
@@ -641,7 +627,6 @@ def display_preview_generator():
     
     while True:
         try:
-            _touch_viewer_marker()
             # Check if snapshot file exists and has been modified
             if os.path.exists(snapshot_path):
                 current_modified = os.path.getmtime(snapshot_path)