From 3f8edf511378e6a02477a14e75cd5ba47c5a05c2 Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Wed, 23 Sep 2026 12:58:11 -0400 Subject: [PATCH 1/2] fix(logos): one hardened logo download path; shared, real HTTP headers (#612) * fix(logos): harden the plugin logo download and share core HTTP headers download_missing_logo / LogoDownloader.download_logo, the path the scoreboard plugins use, read response.content with no size cap and wrote straight to the final path, so a failed or corrupt download could be left in place and cached as the logo. It now goes through fetch_logo: streamed with a 10 MB cap, image/* only, decoded by Pillow, converted to RGBA once, and moved into place atomically. A failure leaves no partial or temp file and keeps any logo already on disk. LogoHelper._download_logo delegates to the same code. Public signatures and return values are unchanged; saved files are pixel-identical to before (RGBA, palette+tRNS, L+tRNS, LA, JPEG). download_missing_logo reuses one downloader per thread instead of a new Session per logo. Per thread rather than behind a lock: Session is not documented thread-safe, and a lock would serialise every plugin's downloads behind the slowest one. Placeholders are written atomically, without the test_write.tmp probe. The logo downloader and background data service now send the real ChuckBuilds User-Agent from src.common.api_helper (USER_AGENT, DEFAULT_HTTP_HEADERS) instead of a yourusername/contact@example.com placeholder, and no longer hand-set Accept-Encoding: br (brotli is not installed). Co-Authored-By: Claude Opus 5.5 * fix(http): drop APIHelper's hand-set brotli encoding; LogoHelper sends the real UA APIHelper advertised `br` though brotli isn't installed, so a server that honoured it would send a body requests can't decode. LogoHelper sent a bare `LEDMatrix-Common/1.0`, the kind of User-Agent ESPN has been rejecting. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- CHANGELOG.md | 29 ++++ src/background_data_service.py | 12 +- src/common/api_helper.py | 25 ++- src/common/logo_helper.py | 73 ++------- src/logo_downloader.py | 270 ++++++++++++++++++++++++--------- test/test_http_headers.py | 80 ++++++++++ test/test_logo_downloader.py | 267 ++++++++++++++++++++++++++++++++ test/test_logo_helper.py | 19 ++- 8 files changed, 622 insertions(+), 153 deletions(-) create mode 100644 test/test_http_headers.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 36c80b86..8b2d3927 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,35 @@ core, the monorepo or the registry's third-party plugins calls them: instead of handing `config.json` to root, and an install path with "secrets" in a directory name no longer makes `config.json` mode 0640. +New names in existing modules (no new modules; a plugin importing these must +floor on the release that ships them): + +- `src.common.api_helper`: `USER_AGENT`, `DEFAULT_HTTP_HEADERS` (read-only). +- `src.logo_downloader`: `fetch_logo`, `save_png_atomically`, + `shared_downloader`. + +### Logo downloads + +- `download_missing_logo` / `LogoDownloader.download_logo` (the path the + scoreboard plugins use) now stream the logo with a 10 MB cap, accept only an + `image/*` response that Pillow can decode, and move the finished RGBA PNG + into place atomically. A failed, oversized or non-image download no longer + leaves a partial file behind, and no longer replaces a logo already on disk. + `LogoHelper._download_logo` goes through the same code. Signatures and return + values are unchanged; saved files are pixel-identical to before. +- `download_missing_logo` reuses one downloader (one `requests.Session`) per + thread instead of building a new one for every logo. +- Placeholder logos are written atomically, without the `test_write.tmp` + probe file. + +### HTTP headers + +- The logo downloader and the background data service send the real + `LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)` User-Agent + instead of a `yourusername` / `contact@example.com` placeholder, and no + longer set `Accept-Encoding: ... br` by hand (brotli is not installed, so a + `br` response could not be decoded); requests picks the encodings. + ## 3.5.0 New modules a plugin may import via `src.*` (floor on 3.5.0): diff --git a/src/background_data_service.py b/src/background_data_service.py index f12b3750..22989a56 100644 --- a/src/background_data_service.py +++ b/src/background_data_service.py @@ -172,14 +172,10 @@ class BackgroundDataService: self.session.mount('http://', requests.adapters.HTTPAdapter(max_retries=3)) self.session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3)) - # Default headers - self.default_headers = { - 'User-Agent': 'LEDMatrix/1.0 (https://github.com/yourusername/LEDMatrix)', - 'Accept': 'application/json', - 'Accept-Language': 'en-US,en;q=0.9', - 'Accept-Encoding': 'gzip, deflate, br', - 'Connection': 'keep-alive' - } + # Default headers: core's shared set (real User-Agent, no hand-set + # Accept-Encoding) -- see src/common/api_helper.py. + from src.common.api_helper import DEFAULT_HTTP_HEADERS + self.default_headers = dict(DEFAULT_HTTP_HEADERS) logger.info(f"BackgroundDataService initialized with {max_workers} workers") diff --git a/src/common/api_helper.py b/src/common/api_helper.py index 88bdbe11..d025b378 100644 --- a/src/common/api_helper.py +++ b/src/common/api_helper.py @@ -8,14 +8,32 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins. import logging import time from datetime import datetime +from types import MappingProxyType from src.common.espn_dates import ESPN_MAX_LIMIT -from typing import Any, Dict, Optional +from typing import Any, Dict, Mapping, Optional import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry +#: The User-Agent core sends to ESPN and other data APIs. It names the client +#: and links to it: around 2026-08-04 ESPN began 403ing bare custom tokens +#: (and browser strings), and this form is what it accepts. +USER_AGENT = 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)' + +#: Base headers for core's JSON API requests. Read-only; pass +#: ``{**DEFAULT_HTTP_HEADERS, ...}`` to add to it. There is deliberately no +#: Accept-Encoding: requests advertises only what urllib3 can decode here +#: (``br`` needs the optional brotli package, which is not a requirement), so a +#: hand-set ``br`` invites a body the client cannot read. +DEFAULT_HTTP_HEADERS: Mapping[str, str] = MappingProxyType({ + 'User-Agent': USER_AGENT, + 'Accept': 'application/json', + 'Accept-Language': 'en-US,en;q=0.9', +}) + + class APIHelper: """ Helper class for HTTP requests, caching, and ESPN API integration. @@ -57,12 +75,9 @@ class APIHelper: # Default headers self.session.headers.update({ - # Identifies the client and links to it: ESPN began 403ing bare - # custom tokens (and browser strings) around 2026-08-04. - 'User-Agent': 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)', + 'User-Agent': USER_AGENT, 'Accept': 'application/json', 'Accept-Language': 'en-US,en;q=0.9', - 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive' }) diff --git a/src/common/logo_helper.py b/src/common/logo_helper.py index c8a5c221..a574b89c 100644 --- a/src/common/logo_helper.py +++ b/src/common/logo_helper.py @@ -6,19 +6,16 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins. """ import logging -import os -import tempfile import time from pathlib import Path from typing import Dict, List, Optional, Union import requests from PIL import Image +from src.common.api_helper import USER_AGENT from src.common.permission_utils import ( ensure_directory_permissions, - ensure_file_permissions, get_assets_dir_mode, - get_assets_file_mode ) # How long a missing logo stays remembered as missing. @@ -61,6 +58,7 @@ def _usable_scale(scale) -> float: # Well above any real team logo; bounds what a remote URL can write to disk. +# The cap for every logo download: src.logo_downloader.fetch_logo uses it too. MAX_LOGO_BYTES = 10 * 1024 * 1024 @@ -107,7 +105,7 @@ class LogoHelper: # Session for HTTP requests self.session = requests.Session() self.session.headers.update({ - 'User-Agent': 'LEDMatrix-Common/1.0', + 'User-Agent': USER_AGENT, 'Accept': 'image/*', }) @@ -408,64 +406,23 @@ class LogoHelper: self._cache_order.append(cache_key) def _download_logo(self, url: str, file_path: Path) -> None: - """Download logo from URL. + """Download a logo from ``url`` to ``file_path``; raises on failure. - The response size is capped and the saved file is verified as a - decodable image before it is left on disk: a logo URL is remote - input, and without this an oversized or malformed response would - be cached for every later load_logo() call to trip over. + Delegates to ``src.logo_downloader.fetch_logo``, the same hardened + download the scoreboard plugins use: streamed and capped at + ``MAX_LOGO_BYTES``, ``image/*`` only, decoded by Pillow, stored as an + RGBA PNG, and moved into place atomically -- a failure leaves neither + a partial file nor a temp file. Uses this helper's own session. - The body is streamed and counted as it arrives rather than read - through response.content, which buffers the whole thing first — - a server that omits Content-Length and never stops sending would - exhaust memory before any size check could run. Nothing lands at - file_path until the download completes and decodes, so a failed - download cannot leave a truncated logo behind either. + Imported lazily: src.logo_downloader imports src.common, so a + module-level import here would be circular. """ + from src.logo_downloader import fetch_logo + # Ensure directory exists with proper permissions ensure_directory_permissions(file_path.parent, get_assets_dir_mode()) - - # A unique temp name, not a fixed ".part": two plugins can - # ask for the same logo at once, and a shared name would let them - # interleave writes into one file, publish the mixture, or delete - # each other's partial. Same directory, so os.replace stays atomic. - fd, tmp_name = tempfile.mkstemp( - dir=str(file_path.parent), prefix=file_path.name + '.', suffix='.part') - tmp_path = Path(tmp_name) - try: - # fdopen outermost so the descriptor mkstemp handed back is - # always adopted and closed, including when the request itself - # raises — load_logo_with_download swallows that, so a leak - # here would accumulate quietly on a URL that keeps failing. - with os.fdopen(fd, 'wb') as f: - with self.session.get(url, timeout=30, stream=True) as response: - response.raise_for_status() - downloaded = 0 - for chunk in response.iter_content(chunk_size=64 * 1024): - if not chunk: - continue - downloaded += len(chunk) - if downloaded > MAX_LOGO_BYTES: - raise ValueError( - f"Logo at {url} exceeds the " - f"{MAX_LOGO_BYTES}-byte limit; not saved") - f.write(chunk) - - # Verify it decodes before it becomes the cached logo. PIL - # raises DecompressionBombError past its own pixel limit; a - # partial or non-image response raises UnidentifiedImageError - # (an OSError subclass). - with Image.open(tmp_path) as probe: - probe.load() - - os.replace(tmp_path, file_path) - except BaseException: - tmp_path.unlink(missing_ok=True) - raise - - # Set proper file permissions after saving - ensure_file_permissions(file_path, get_assets_file_mode()) - + fetch_logo(self.session, url, file_path, timeout=30, + max_bytes=MAX_LOGO_BYTES) self.logger.debug(f"Downloaded logo to {file_path}") def _create_placeholder_logo(self, team_abbr: str, diff --git a/src/logo_downloader.py b/src/logo_downloader.py index 43b8ff1a..c74096df 100644 --- a/src/logo_downloader.py +++ b/src/logo_downloader.py @@ -7,17 +7,21 @@ with special support for FCS teams and other NCAA divisions. import os import re +import tempfile +import threading import time import logging import requests import json from typing import Dict, List, Optional, Tuple from pathlib import Path -from PIL import Image, ImageDraw, ImageFont +from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError from src.common.font_layout import load_truetype from PIL.PngImagePlugin import PngInfo from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry +from src.common.api_helper import DEFAULT_HTTP_HEADERS +from src.common.logo_helper import MAX_LOGO_BYTES from src.common.permission_utils import ( ensure_directory_permissions, ensure_file_permissions, @@ -27,6 +31,142 @@ from src.common.permission_utils import ( logger = logging.getLogger(__name__) +#: Accept header for logo image requests (the JSON default is for the API). +LOGO_ACCEPT = 'image/png,image/*;q=0.8' + + +def _is_image_content_type(content_type: str) -> bool: + """True for an ``image/*`` media type, ignoring parameters and case. + + Only a cheap gate before the body is read; Pillow decoding the bytes is + what actually decides whether they are an image. + """ + return content_type.split(';', 1)[0].strip().lower().startswith('image/') + + +def _to_rgba(img: Image.Image) -> Image.Image: + """``img`` as RGBA, keeping its transparency. + + One conversion covers every mode: Pillow folds a palette's or a + greyscale/RGB image's ``transparency`` entry into the alpha channel when + converting to RGBA, and an image without one gets an opaque alpha. Plugins + paste logos with the image as its own mask, so the alpha is the part that + has to survive. + """ + return img.copy() if img.mode == 'RGBA' else img.convert('RGBA') + + +def _publish(tmp_path: Path, filepath: Path) -> None: + """Give a finished temp file the asset mode, then move it into place. + + ``mkstemp`` creates files 0600, so the mode is set before the rename: the + logo must never be visible under its real name unreadable to the web + service's user. ``os.replace`` within one directory is atomic, so a reader + sees the old file or the new one, never a half-written one. + """ + ensure_file_permissions(tmp_path, get_assets_file_mode()) + os.replace(tmp_path, filepath) + + +def _temp_beside(filepath: Path) -> Tuple[int, Path]: + """A unique temp file in ``filepath``'s directory. + + Unique rather than a fixed ``.part`` because two plugins can ask for + the same logo at once; a shared name would let them interleave writes into + one file, or delete each other's partial. The same directory keeps + ``os.replace`` atomic. + """ + fd, tmp_name = tempfile.mkstemp( + dir=str(filepath.parent), prefix=filepath.name + '.', suffix='.part') + return fd, Path(tmp_name) + + +def save_png_atomically(image: Image.Image, filepath: Path, + pnginfo: Optional[PngInfo] = None) -> None: + """Save ``image`` as a PNG at ``filepath`` without ever exposing a partial file. + + Raises on failure (including PermissionError for an unwritable + directory), leaving any previous file at ``filepath`` untouched and no + temp file behind. + """ + filepath = Path(filepath) + fd, tmp_path = _temp_beside(filepath) + try: + with os.fdopen(fd, 'wb') as f: + if pnginfo is not None: + image.save(f, 'PNG', pnginfo=pnginfo) + else: + image.save(f, 'PNG') + _publish(tmp_path, filepath) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise + + +def fetch_logo(session: requests.Session, url: str, filepath: Path, *, + headers: Optional[Dict[str, str]] = None, timeout: float = 30, + max_bytes: Optional[int] = None) -> None: + """Download the image at ``url`` and save it at ``filepath`` as an RGBA PNG. + + The one hardened logo download; ``LogoDownloader.download_logo`` and + ``LogoHelper._download_logo`` both go through it. A logo URL is remote + input, and whatever lands at ``filepath`` is cached and loaded on every + later frame, so nothing is written there until the bytes have been + size-checked and decoded: + + - the response must be ``image/*`` and is streamed, counted as it + arrives, and abandoned past ``max_bytes`` (``MAX_LOGO_BYTES`` by + default) -- ``response.content`` would buffer a body that never ends + before any check could run; + - the bytes go to a unique temp file beside ``filepath`` and must decode + with Pillow (which also enforces its decompression-bomb limit); + - the image is converted to RGBA once, rewritten as PNG, and moved into + place atomically. + + Raises on any failure, with no temp file left and any previous file at + ``filepath`` intact. The caller creates the directory. + """ + if max_bytes is None: + max_bytes = MAX_LOGO_BYTES + filepath = Path(filepath) + fd, tmp_path = _temp_beside(filepath) + try: + # fdopen outermost so the descriptor is adopted and closed even when + # the request itself raises. + with os.fdopen(fd, 'wb') as f: + with session.get(url, headers=headers, timeout=timeout, + stream=True) as response: + response.raise_for_status() + content_type = response.headers.get('content-type') or '' + if not _is_image_content_type(content_type): + raise ValueError( + f"Logo at {url} is not an image " + f"(content-type {content_type!r}); not saved") + received = 0 + for chunk in response.iter_content(chunk_size=64 * 1024): + if not chunk: + continue + received += len(chunk) + if received > max_bytes: + raise ValueError( + f"Logo at {url} exceeds the " + f"{max_bytes}-byte limit; not saved") + f.write(chunk) + + # UnidentifiedImageError (an OSError) for bytes that are not an + # image, DecompressionBombError past Pillow's pixel limit, OSError for + # a truncated one. + with Image.open(tmp_path) as img: + img.load() + rgba = _to_rgba(img) + with open(tmp_path, 'wb') as f: + rgba.save(f, 'PNG') + _publish(tmp_path, filepath) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise + + #: PNG text key stamped into a generated placeholder so a later run can tell it #: apart from a real logo that happens to be small. PLACEHOLDER_MARKER = "ledmatrix_placeholder" @@ -199,14 +339,8 @@ class LogoDownloader: self.session.mount("https://", adapter) self.session.mount("http://", adapter) - # Set up headers - self.headers = { - 'User-Agent': 'LEDMatrix/1.0 (https://github.com/yourusername/LEDMatrix; contact@example.com)', - 'Accept': 'application/json', - 'Accept-Language': 'en-US,en;q=0.9', - 'Accept-Encoding': 'gzip, deflate, br', - 'Connection': 'keep-alive' - } + # Core's shared API headers; a plain dict so callers may adjust theirs. + self.headers = dict(DEFAULT_HTTP_HEADERS) @staticmethod def normalize_abbreviation(abbreviation: str) -> str: @@ -301,50 +435,19 @@ class LogoDownloader: return False def download_logo(self, logo_url: str, filepath: Path, team_abbreviation: str) -> bool: - """Download a single logo from URL and save to filepath.""" + """Download a single logo from URL and save it to filepath as an RGBA PNG. + + Returns False (and logs why) on any failure; see ``fetch_logo`` for + the guarantees -- in particular a failure never leaves a partial file, + and never replaces a logo already at ``filepath``. + """ + filepath = Path(filepath) try: - response = self.session.get(logo_url, headers=self.headers, timeout=self.request_timeout) - response.raise_for_status() - - # Verify it's actually an image - content_type = response.headers.get('content-type', '').lower() - if not any(img_type in content_type for img_type in ['image/png', 'image/jpeg', 'image/jpg', 'image/gif']): - logger.warning(f"Downloaded content for {team_abbreviation} is not an image: {content_type}") - return False - - with open(filepath, 'wb') as f: - f.write(response.content) - - # Verify and convert the downloaded image to RGBA format - try: - with Image.open(filepath) as img: - # Convert to RGBA to avoid PIL warnings about palette images with transparency - if img.mode in ('P', 'LA', 'L'): - # Convert palette or grayscale images to RGBA - img = img.convert('RGBA') - elif img.mode == 'RGB': - # Convert RGB to RGBA (add alpha channel) - img = img.convert('RGBA') - elif img.mode != 'RGBA': - # For any other mode, convert to RGBA - img = img.convert('RGBA') - - # Save the converted image - img.save(filepath, 'PNG') - - # Set proper file permissions after saving - ensure_file_permissions(filepath, get_assets_file_mode()) - - logger.info(f"Successfully downloaded and converted logo for {team_abbreviation} -> {filepath.name}") - return True - except Exception as e: - logger.error(f"Downloaded file for {team_abbreviation} is not a valid image or conversion failed: {e}") - try: - os.remove(filepath) # Remove invalid file - except OSError: - pass - return False - + fetch_logo(self.session, logo_url, filepath, + headers={**self.headers, 'Accept': LOGO_ACCEPT}, + timeout=self.request_timeout) + logger.info(f"Successfully downloaded and converted logo for {team_abbreviation} -> {filepath.name}") + return True except PermissionError as e: logger.error(f"Permission denied downloading logo for {team_abbreviation}: {e}") logger.error("Please run: sudo ./scripts/fix_perms/fix_assets_permissions.sh") @@ -352,10 +455,13 @@ class LogoDownloader: except requests.exceptions.RequestException as e: logger.error(f"Failed to download logo for {team_abbreviation}: {e}") return False + except (ValueError, UnidentifiedImageError, Image.DecompressionBombError) as e: + logger.error(f"Rejected downloaded logo for {team_abbreviation}: {e}") + return False except Exception as e: logger.error(f"Unexpected error downloading logo for {team_abbreviation}: {e}") return False - + # Allowlist for the league_code segment interpolated into ESPN API URLs _SAFE_LEAGUE_CODE_RE = re.compile(r'^[a-z0-9_-]+$') @@ -728,20 +834,7 @@ class LogoDownloader: filename = f"{self.normalize_abbreviation(team_abbreviation)}.png" filepath = Path(logo_dir) / filename - - # Check if we can write to the directory - try: - # Test write permissions by creating a temporary file - test_file = filepath.parent / "test_write.tmp" - test_file.touch() - test_file.unlink() # Remove the test file - except PermissionError: - logger.error(f"Permission denied: Cannot write to directory {logo_dir}") - return False - except Exception as e: - logger.error(f"Directory access error for {logo_dir}: {e}") - return False - + # Create a simple placeholder logo logo = Image.new('RGBA', (64, 64), (100, 100, 100, 255)) # Gray background draw = ImageDraw.Draw(logo) @@ -774,14 +867,16 @@ class LogoDownloader: # proof the logo was fetched. metadata = PngInfo() metadata.add_text(PLACEHOLDER_MARKER, str(time.time())) - logo.save(filepath, "PNG", pnginfo=metadata) - - # Set proper file permissions after saving - ensure_file_permissions(filepath, get_assets_file_mode()) + # Atomic, and it sets the asset mode; an unwritable directory + # surfaces here as PermissionError. + save_png_atomically(logo, filepath, pnginfo=metadata) logger.info(f"Created placeholder logo for {team_abbreviation} at {filepath}") return True - + + except PermissionError as e: + logger.error(f"Permission denied: Cannot write placeholder logo to {logo_dir}: {e}") + return False except Exception as e: logger.error(f"Failed to create placeholder logo for {team_abbreviation}: {e}") return False @@ -837,6 +932,31 @@ def get_soccer_league_key(league_code: str) -> str: return f"soccer_{league_code}" +_thread_state = threading.local() + + +def shared_downloader() -> LogoDownloader: + """The calling thread's reusable LogoDownloader. + + ``download_missing_logo`` used to build a new downloader -- a new + ``requests.Session``, retry adapter and connection pool -- for every logo. + Reusing one keeps connections to ESPN's CDN alive between logos. + + One per thread rather than one behind a lock: ``requests.Session`` is not + documented as safe for concurrent use, and plugins download on worker + threads (football-scoreboard runs a small pool precisely so downloads + overlap). A lock would serialise every plugin's downloads behind the + slowest one -- up to 30s per attempt, with retries. Pool threads persist, + so each still reuses its own session; a thread's downloader goes with the + thread. + """ + downloader = getattr(_thread_state, 'downloader', None) + if downloader is None: + downloader = LogoDownloader() + _thread_state.downloader = downloader + return downloader + + # Convenience function for easy integration def download_missing_logo(league: str, team_id: str, team_abbreviation: str, logo_path: Path, logo_url: str | None = None, create_placeholder: bool = True) -> bool: """ @@ -852,8 +972,8 @@ def download_missing_logo(league: str, team_id: str, team_abbreviation: str, log Returns: True if logo exists or was successfully downloaded, False otherwise """ - downloader = LogoDownloader() - + downloader = shared_downloader() + # Use the directory from the logo_path parameter (respects config settings) logo_path = Path(logo_path) if not logo_path.is_absolute(): @@ -920,5 +1040,5 @@ def download_all_logos_for_league(league: str, force_download: bool = False) -> Returns: Tuple of (downloaded_count, failed_count) """ - downloader = LogoDownloader() + downloader = shared_downloader() return downloader.download_missing_logos_for_league(league, force_download) diff --git a/test/test_http_headers.py b/test/test_http_headers.py new file mode 100644 index 00000000..bc5836a1 --- /dev/null +++ b/test/test_http_headers.py @@ -0,0 +1,80 @@ +"""One set of HTTP headers for core's ESPN/data requests. + +The logo downloader and the background data service each carried their own +header dict with a placeholder User-Agent (``yourusername/LEDMatrix; +contact@example.com``) -- the kind of nonconforming token ESPN began 403ing +around 2026-08-04 -- and a hand-set ``Accept-Encoding: ... br`` although brotli +is not installed, so a ``br`` body could not have been decoded. +""" + +from unittest.mock import MagicMock + +import pytest + +from src.common.api_helper import DEFAULT_HTTP_HEADERS, USER_AGENT, APIHelper + + +def _lower_keys(headers): + return {k.lower(): v for k, v in headers.items()} + + +class TestSharedHeaders: + def test_user_agent_names_the_project(self): + assert USER_AGENT == 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)' + assert DEFAULT_HTTP_HEADERS['User-Agent'] == USER_AGENT + + def test_no_hand_set_accept_encoding(self): + assert 'accept-encoding' not in _lower_keys(DEFAULT_HTTP_HEADERS) + + def test_is_read_only(self): + with pytest.raises(TypeError): + DEFAULT_HTTP_HEADERS['User-Agent'] = 'x' # type: ignore[index] + + def test_api_helper_sends_the_same_user_agent(self): + assert APIHelper().session.headers['User-Agent'] == USER_AGENT + + def test_api_helper_does_not_hand_set_brotli(self): + assert 'br' not in APIHelper().session.headers.get('Accept-Encoding', '') + + def test_logo_helper_sends_the_same_user_agent(self): + from src.common.logo_helper import LogoHelper + assert LogoHelper(display_width=64, display_height=32).session.headers['User-Agent'] == USER_AGENT + + +class TestLogoDownloaderHeaders: + def test_uses_the_shared_headers(self): + from src.logo_downloader import LogoDownloader + headers = LogoDownloader().headers + assert headers['User-Agent'] == USER_AGENT + assert 'accept-encoding' not in _lower_keys(headers) + assert 'yourusername' not in str(headers) + + def test_instance_headers_are_a_private_copy(self): + from src.logo_downloader import LogoDownloader + downloader = LogoDownloader() + downloader.headers['X-Test'] = '1' + assert 'X-Test' not in DEFAULT_HTTP_HEADERS + assert 'X-Test' not in LogoDownloader().headers + + def test_logo_request_sends_the_user_agent_and_asks_for_an_image(self, tmp_path): + from src.logo_downloader import LogoDownloader + downloader = LogoDownloader() + downloader.session.get = MagicMock(side_effect=RuntimeError("stop")) + downloader.download_logo("http://x/a.png", tmp_path / "A.png", "A") + sent = _lower_keys(downloader.session.get.call_args.kwargs['headers']) + assert sent['user-agent'] == USER_AGENT + assert sent['accept'].startswith('image/') + assert 'accept-encoding' not in sent + + +class TestBackgroundDataServiceHeaders: + def test_uses_the_shared_headers(self): + from src.background_data_service import BackgroundDataService + service = BackgroundDataService(MagicMock(), max_workers=1) + try: + headers = service.default_headers + assert headers['User-Agent'] == USER_AGENT + assert 'accept-encoding' not in _lower_keys(headers) + assert 'yourusername' not in str(headers) + finally: + service.shutdown(wait=False) diff --git a/test/test_logo_downloader.py b/test/test_logo_downloader.py index c14b3391..a303b087 100644 --- a/test/test_logo_downloader.py +++ b/test/test_logo_downloader.py @@ -7,16 +7,20 @@ ensure_logo_directory, and the download_missing_logo function path (with HTTP mocked). """ +import io import os +import threading import time import pytest +import requests from pathlib import Path from unittest.mock import patch, Mock, MagicMock from PIL import Image from PIL.PngImagePlugin import PngInfo +import src.logo_downloader as logo_downloader_module from src.logo_downloader import ( PLACEHOLDER_BG, PLACEHOLDER_MARKER, @@ -351,3 +355,266 @@ class TestRefreshPlaceholderTimestamp: def test_missing_file_is_not_an_error(self, tmp_path): assert refresh_placeholder_timestamp(tmp_path / "nope.png") is False + + +# --------------------------------------------------------------------------- +# download_logo: the download the scoreboard plugins actually use +# +# It used to read response.content with no size cap, write straight to the +# final path (so a failed or corrupt download could be left there and then be +# cached as the logo), and build a fresh Session for every logo. It now goes +# through fetch_logo, the hardened download LogoHelper also uses. +# --------------------------------------------------------------------------- + +def _png(img: Image.Image) -> bytes: + buf = io.BytesIO() + img.save(buf, "PNG") + return buf.getvalue() + + +def _stream(body: bytes, content_type: str = "image/png", chunk: int = 1024, + die_after: int | None = None): + """A streamed requests.Response stand-in. + + ``die_after`` makes the transfer fail after that many bytes, the way a + reset connection does mid-body. + """ + response = MagicMock() + response.__enter__.return_value = response + response.__exit__.return_value = False + response.raise_for_status = MagicMock() + response.headers = {"content-type": content_type} + + def _iter_content(*_args, **_kwargs): + sent = 0 + for i in range(0, len(body), chunk): + if die_after is not None and sent >= die_after: + raise requests.exceptions.ChunkedEncodingError("connection reset") + piece = body[i:i + chunk] + sent += len(piece) + yield piece + + response.iter_content = _iter_content + return response + + +@pytest.fixture +def fresh_thread_state(monkeypatch): + """Each test gets its own per-thread downloader cache.""" + monkeypatch.setattr(logo_downloader_module, "_thread_state", threading.local()) + + +@pytest.fixture +def downloader(): + return LogoDownloader() + + +@pytest.fixture +def old_logo(tmp_path): + """A logo already on disk that a failed re-download must not damage.""" + path = tmp_path / "PHI.png" + Image.new("RGBA", (30, 30), (1, 2, 3, 255)).save(path) + return path, path.read_bytes() + + +def _leftovers(directory: Path, keep: str | None = None): + return sorted(p.name for p in directory.iterdir() if p.name != keep) + + +class TestDownloadLogoHardening: + def test_valid_logo_is_saved_as_rgba_png(self, downloader, tmp_path): + target = tmp_path / "PHI.png" + downloader.session.get = MagicMock( + return_value=_stream(_png(Image.new("RGB", (20, 10), (9, 8, 7))))) + assert downloader.download_logo("http://x/phi.png", target, "PHI") is True + with Image.open(target) as img: + assert img.format == "PNG" and img.mode == "RGBA" + assert img.getpixel((0, 0)) == (9, 8, 7, 255) + assert _leftovers(tmp_path, keep="PHI.png") == [] + # Streamed, so the size cap applies before the body is buffered. + assert downloader.session.get.call_args.kwargs["stream"] is True + + def test_oversized_response_is_rejected_and_leaves_no_file( + self, downloader, tmp_path, monkeypatch): + monkeypatch.setattr(logo_downloader_module, "MAX_LOGO_BYTES", 4096) + target = tmp_path / "BIG.png" + body = _png(Image.new("RGB", (8, 8))) + b"\x00" * 8192 + downloader.session.get = MagicMock(return_value=_stream(body)) + assert downloader.download_logo("http://x/big.png", target, "BIG") is False + assert list(tmp_path.iterdir()) == [] + + def test_oversized_response_does_not_replace_an_existing_logo( + self, downloader, old_logo, monkeypatch): + path, before = old_logo + monkeypatch.setattr(logo_downloader_module, "MAX_LOGO_BYTES", 4096) + downloader.session.get = MagicMock( + return_value=_stream(b"\x89PNG" + b"\x00" * 8192)) + assert downloader.download_logo("http://x/big.png", path, "PHI") is False + assert path.read_bytes() == before + assert _leftovers(path.parent, keep=path.name) == [] + + def test_mid_download_failure_leaves_no_partial_file(self, downloader, tmp_path): + target = tmp_path / "CUT.png" + body = _png(Image.new("RGB", (200, 200), (5, 5, 5))) + b"\x00" * 4096 + downloader.session.get = MagicMock( + return_value=_stream(body, chunk=256, die_after=512)) + assert downloader.download_logo("http://x/cut.png", target, "CUT") is False + assert list(tmp_path.iterdir()) == [] + + def test_mid_download_failure_keeps_the_previous_logo(self, downloader, old_logo): + # The old code opened the final path for writing before the body + # arrived, so a dropped connection truncated the logo it was replacing. + path, before = old_logo + body = _png(Image.new("RGB", (200, 200), (5, 5, 5))) + b"\x00" * 4096 + downloader.session.get = MagicMock( + return_value=_stream(body, chunk=256, die_after=512)) + assert downloader.download_logo("http://x/cut.png", path, "PHI") is False + assert path.read_bytes() == before + assert _leftovers(path.parent, keep=path.name) == [] + + def test_non_image_content_type_is_rejected(self, downloader, old_logo): + # Rejected on the label alone, before the body is trusted: these bytes + # would decode, so only the content-type check stops them. + path, before = old_logo + body = _png(Image.new("RGB", (8, 8), (250, 0, 0))) + downloader.session.get = MagicMock( + return_value=_stream(body, content_type="text/html")) + assert downloader.download_logo("http://x/404", path, "PHI") is False + assert path.read_bytes() == before + assert _leftovers(path.parent, keep=path.name) == [] + + def test_bytes_that_do_not_decode_are_rejected(self, downloader, old_logo): + # A server that labels an error page image/png must not get it cached. + path, before = old_logo + downloader.session.get = MagicMock( + return_value=_stream(b"oops", content_type="image/png")) + assert downloader.download_logo("http://x/lie.png", path, "PHI") is False + assert path.read_bytes() == before + assert _leftovers(path.parent, keep=path.name) == [] + + def test_http_error_keeps_the_previous_logo(self, downloader, old_logo): + path, before = old_logo + response = _stream(b"") + response.raise_for_status.side_effect = requests.exceptions.HTTPError("503") + downloader.session.get = MagicMock(return_value=response) + assert downloader.download_logo("http://x/phi.png", path, "PHI") is False + assert path.read_bytes() == before + assert _leftovers(path.parent, keep=path.name) == [] + + def test_unwritable_directory_returns_false(self, downloader, tmp_path): + downloader.session.get = MagicMock() + with patch("src.logo_downloader.tempfile.mkstemp", + side_effect=PermissionError("read-only")): + assert downloader.download_logo( + "http://x/phi.png", tmp_path / "PHI.png", "PHI") is False + downloader.session.get.assert_not_called() + + +class TestDownloadLogoTransparency: + """Plugins paste logos with the image as its own mask; alpha must survive.""" + + def _download(self, downloader, tmp_path, body, content_type="image/png"): + target = tmp_path / "LOGO.png" + downloader.session.get = MagicMock(return_value=_stream(body, content_type)) + assert downloader.download_logo("http://x/logo", target, "LOGO") is True + with Image.open(target) as img: + img.load() + return img.copy(), img.format + + def test_rgba_alpha_is_kept_exactly(self, downloader, tmp_path): + src = Image.new("RGBA", (16, 16), (0, 0, 0, 0)) + for x in range(16): + src.putpixel((x, 3), (200, 100, 50, x * 16)) + out, _ = self._download(downloader, tmp_path, _png(src)) + assert out.mode == "RGBA" + assert list(out.getdata()) == list(src.getdata()) + + def test_palette_transparency_becomes_alpha(self, downloader, tmp_path): + src = Image.new("P", (8, 8), 0) + src.putpalette([0, 0, 0, 255, 0, 0] + [0] * (254 * 3)) + src.putpixel((4, 4), 1) + buf = io.BytesIO() + src.save(buf, "PNG", transparency=0) + out, _ = self._download(downloader, tmp_path, buf.getvalue()) + assert out.mode == "RGBA" + assert out.getpixel((0, 0))[3] == 0 + assert out.getpixel((4, 4)) == (255, 0, 0, 255) + + def test_greyscale_transparency_becomes_alpha(self, downloader, tmp_path): + src = Image.new("L", (8, 8), 0) + src.putpixel((2, 2), 255) + buf = io.BytesIO() + src.save(buf, "PNG", transparency=0) + out, _ = self._download(downloader, tmp_path, buf.getvalue()) + assert out.getpixel((0, 0))[3] == 0 + assert out.getpixel((2, 2)) == (255, 255, 255, 255) + + def test_jpeg_is_stored_as_opaque_rgba_png(self, downloader, tmp_path): + buf = io.BytesIO() + Image.new("RGB", (8, 8), (10, 200, 30)).save(buf, "JPEG", quality=95) + out, fmt = self._download(downloader, tmp_path, buf.getvalue(), "image/jpeg") + assert fmt == "PNG" and out.mode == "RGBA" + assert out.getchannel("A").getextrema() == (255, 255) + + +class TestSharedDownloader: + def test_download_missing_logo_reuses_one_session(self, tmp_path, fresh_thread_state): + sessions = [] + real_session = requests.Session + + def counting_session(*args, **kwargs): + s = real_session(*args, **kwargs) + sessions.append(s) + return s + + with patch("src.logo_downloader.requests.Session", side_effect=counting_session): + with patch.object(LogoDownloader, "download_logo", return_value=True) as dl: + for abbr in ("AAA", "BBB", "CCC"): + assert download_missing_logo( + "nfl", "1", abbr, tmp_path / f"{abbr}.png", + logo_url=f"http://x/{abbr}.png", + create_placeholder=False) is True + assert dl.call_count == 3 + assert len(sessions) == 1 + + def test_each_thread_gets_its_own_downloader(self, fresh_thread_state): + here = logo_downloader_module.shared_downloader() + assert logo_downloader_module.shared_downloader() is here + seen = [] + t = threading.Thread(target=lambda: seen.append( + logo_downloader_module.shared_downloader())) + t.start() + t.join() + assert seen and seen[0] is not here + assert seen[0].session is not here.session + + +class TestPlaceholderWrite: + def test_no_write_probe_file_is_created(self, tmp_path): + created = [] + real_touch = Path.touch + + def spy_touch(self, *args, **kwargs): + created.append(self.name) + return real_touch(self, *args, **kwargs) + + with patch.object(Path, "touch", spy_touch): + assert LogoDownloader().create_placeholder_logo("COLL", str(tmp_path)) is True + assert "test_write.tmp" not in created + assert sorted(p.name for p in tmp_path.iterdir()) == ["COLL.png"] + + def test_unwritable_directory_returns_false(self, tmp_path): + with patch.object(LogoDownloader, "ensure_logo_directory", return_value=True), \ + patch("src.logo_downloader.tempfile.mkstemp", + side_effect=PermissionError("read-only")): + assert LogoDownloader().create_placeholder_logo("COLL", str(tmp_path)) is False + assert list(tmp_path.iterdir()) == [] + + def test_failed_save_keeps_the_previous_file(self, tmp_path): + path = tmp_path / "COLL.png" + Image.new("RGBA", (30, 30), (1, 2, 3, 255)).save(path) + before = path.read_bytes() + with patch("src.logo_downloader.os.replace", side_effect=OSError("disk full")): + assert LogoDownloader().create_placeholder_logo("COLL", str(tmp_path)) is False + assert path.read_bytes() == before + assert sorted(p.name for p in tmp_path.iterdir()) == ["COLL.png"] diff --git a/test/test_logo_helper.py b/test/test_logo_helper.py index 0ae7ec44..9ce4eeaf 100644 --- a/test/test_logo_helper.py +++ b/test/test_logo_helper.py @@ -33,7 +33,7 @@ def _no_real_chmod(monkeypatch): # Keep the permission helpers out of the way: their own env detection # is not what these tests are about. monkeypatch.setattr("src.common.logo_helper.ensure_directory_permissions", MagicMock()) - monkeypatch.setattr("src.common.logo_helper.ensure_file_permissions", MagicMock()) + monkeypatch.setattr("src.logo_downloader.ensure_file_permissions", MagicMock()) @pytest.fixture @@ -48,7 +48,8 @@ def write_logo(path: Path, size=(20, 20), color=(255, 0, 0), fmt="PNG") -> Path: return path -def fake_response(content: bytes, chunk_size: int = 64 * 1024): +def fake_response(content: bytes, chunk_size: int = 64 * 1024, + content_type: str = "image/png"): """Stand-in for a streamed requests.Response. _download_logo opens `with session.get(..., stream=True)` and reads @@ -61,6 +62,7 @@ def fake_response(content: bytes, chunk_size: int = 64 * 1024): response.__enter__.return_value = response response.__exit__.return_value = False response.raise_for_status = MagicMock() + response.headers = {"content-type": content_type} def _iter_content(*_args, **_kwargs): for i in range(0, len(content), chunk_size): @@ -80,6 +82,7 @@ def endless_response(chunk: bytes = b"\x00" * 65536): response.__enter__.return_value = response response.__exit__.return_value = False response.raise_for_status = MagicMock() + response.headers = {"content-type": "image/png"} def _iter_content(*_args, **_kwargs): while True: @@ -208,7 +211,7 @@ class TestLoadLogoWithDownload: # stream=True is load-bearing: it is what lets the size cap apply # before the body is buffered. helper.session.get.assert_called_once_with( - "http://x/logo.png", timeout=30, stream=True) + "http://x/logo.png", headers=None, timeout=30, stream=True) def test_download_failure_falls_back_to_placeholder(self, helper, tmp_path): helper.session.get = MagicMock( @@ -241,7 +244,7 @@ class TestDownloadLogo: path.parent.mkdir() helper.session.get = MagicMock(return_value=fake_response(png_bytes())) with patch("src.common.logo_helper.ensure_directory_permissions") as dirs, \ - patch("src.common.logo_helper.ensure_file_permissions") as files: + patch("src.logo_downloader.ensure_file_permissions") as files: helper._download_logo("http://x/logo.png", path) assert path.exists() dirs.assert_called_once() @@ -284,6 +287,7 @@ class TestDownloadLogo: response.__enter__.return_value = response response.__exit__.return_value = False response.raise_for_status = MagicMock() + response.headers = {"content-type": "image/png"} response.iter_content = _dies_midway helper.session.get = MagicMock(return_value=response) @@ -305,7 +309,7 @@ class TestDownloadLogo: seen.append(name) return fd, name - with patch("src.common.logo_helper.tempfile.mkstemp", side_effect=record): + with patch("src.logo_downloader.tempfile.mkstemp", side_effect=record): helper.session.get = MagicMock(return_value=fake_response(png_bytes())) helper._download_logo("http://x/logo.png", path) helper.session.get = MagicMock(return_value=fake_response(png_bytes())) @@ -351,7 +355,7 @@ class TestDownloadLogo: def load(self): raise Image.DecompressionBombError("too many pixels") - monkeypatch.setattr("src.common.logo_helper.Image.open", lambda *a, **kw: Bomb()) + monkeypatch.setattr("src.logo_downloader.Image.open", lambda *a, **kw: Bomb()) with pytest.raises(Image.DecompressionBombError): helper._download_logo("http://x/bomb.png", path) assert not path.exists() @@ -420,7 +424,8 @@ class TestPlaceholderLogo: class TestSessionConfiguration: def test_user_agent_and_accept_headers(self, helper): - assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0" + from src.common.api_helper import USER_AGENT + assert helper.session.headers["User-Agent"] == USER_AGENT assert helper.session.headers["Accept"] == "image/*" From cd5a4e2251b89c5b6f0aa4bc811bd82d0efd9e7f Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Wed, 23 Sep 2026 12:58:39 -0400 Subject: [PATCH 2/2] fix(display): apply on-demand, brightness and schedule changes mid-screen (#618) * fix(display): apply on-demand, brightness and schedule changes mid-screen The main loop read the on-demand mailbox, the on/off schedule and the brightness target once per pass -- once per screen. A dwell can be a minute and a Vegas iteration runs for max_cycle_duration (240s), so on a Pi an on-demand request posted at 10:54:27 was activated at 10:57:24, and two brightness saves 12s apart inside one 30s screen never reached the panel. During Vegas nothing read the mailbox at all: _check_vegas_interrupt only checked on_demand_active, which only the main-loop read sets. _service_pending_changes does the main loop's on-demand poll, expiry, schedule and brightness steps, throttled to PENDING_CHANGES_INTERVAL (the existing 0.25s mailbox floor), on the display thread. It runs from the Vegas interrupt checker, the high-FPS and once-a-second render loops (replacing their direct on-demand poll) and _sleep_with_plugin_updates; between passes it costs one monotonic compare. A brightness change re-pushes the current frame, since the panel only shows it from the next push. Callers act on what it leaves behind: Vegas yields on an on-demand start or the display being scheduled off (and the main loop then blanks instead of rendering a screen), the render loops break on a schedule-off as they already did on a mode change, and the dwell sleep returns early on an on-demand start/stop or a schedule flip -- so the 60s scheduled-off sleep now wakes for an on-demand request. The main loop no longer rotates after a dwell that ended that way, which advanced a new on-demand session past the mode that was asked for. A brightness set_brightness() refuses is not retried until the target changes, so the 4Hz pass doesn't log the same failure (fallback mode) four times a second. Co-Authored-By: Claude Opus 5.5 * test(display): a screen scheduled off midway stops rendering Covers the schedule-off break added to the high-FPS and once-a-second render loops: with the display scheduled off halfway through a 120s screen, neither loop renders for more than one redraw plus one service interval past the boundary. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- src/display_controller.py | 164 ++++++-- test/test_display_pending_changes.py | 592 +++++++++++++++++++++++++++ 2 files changed, 732 insertions(+), 24 deletions(-) create mode 100644 test/test_display_pending_changes.py diff --git a/src/display_controller.py b/src/display_controller.py index dea182fa..1124f01b 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -207,6 +207,12 @@ class DisplayController: # _poll_on_demand_requests. None means "never polled", so the first # call always goes through. self._last_on_demand_poll: Optional[float] = None + # Monotonic stamp of the last _service_pending_changes pass; same + # "None means never" convention as _last_on_demand_poll. + self._last_pending_service: Optional[float] = None + # A brightness set_brightness() refused, so the periodic service pass + # doesn't retry (and log) the same failure several times a second. + self._failed_brightness_target: Optional[int] = None self.on_demand_active = False self.on_demand_mode: Optional[str] = None self.on_demand_modes: List[str] = [] # All modes for the on-demand plugin @@ -581,10 +587,22 @@ class DisplayController: Returns: True if Vegas should yield control, False to continue """ + # A Vegas iteration runs for up to max_cycle_duration (240s by + # default) without returning to the main loop, and this is the only + # code of ours it calls while it does. Without servicing here, an + # on-demand request was never even read until the iteration ended -- + # on_demand_active below is only set by that read -- and a saved + # brightness or a schedule boundary waited just as long. + self._service_pending_changes() + # Check for pending on-demand request if self.on_demand_active: return True + # Scheduled off mid-iteration: hand back so the main loop blanks it. + if not self.is_display_active: + return True + # Check for wifi status that needs display if self._check_wifi_status_message(): return True @@ -1004,12 +1022,22 @@ class DisplayController: self.sync_manager.send_frame(follower_frame) def _sleep_with_plugin_updates(self, duration: float, tick_interval: float = 1.0): - """Sleep while continuing to service plugin update schedules.""" + """Sleep while continuing to service plugin update schedules. + + Also services pending changes (see _service_pending_changes), and + returns early when one of them changes what the panel should show -- + an on-demand start or stop, or the display schedule turning the panel + on or off -- so the caller can act on it instead of finishing a dwell + that could be a minute long (sixty seconds while scheduled off). + """ if duration <= 0: return end_time = time.time() + duration - tick_interval = max(0.001, tick_interval) + tick_interval = max(0.001, min(tick_interval, self.PENDING_CHANGES_INTERVAL)) + mode = self.current_display_mode + display_active = self.is_display_active + on_demand = self.on_demand_active while True: remaining = end_time - time.time() @@ -1019,6 +1047,11 @@ class DisplayController: sleep_time = min(tick_interval, remaining) time.sleep(sleep_time) self._tick_plugin_updates() + self._service_pending_changes() + if (self.current_display_mode != mode + or self.is_display_active != display_active + or self.on_demand_active != on_demand): + break def _get_display_duration(self, mode_key): """Seconds to show a mode: the Rotation & Durations page's value for it @@ -1215,6 +1248,83 @@ class DisplayController: #: perceptible, and it cuts the read rate by 30x. ON_DEMAND_POLL_INTERVAL = 0.25 + #: Shortest gap between _service_pending_changes passes. The same floor as + #: the mailbox poll, since that read is the only real cost in the pass: + #: the schedule checks are gated to once per clock minute and the rest is + #: attribute compares. Callers run at frame rate, so between passes the + #: whole cost is one monotonic-clock compare. + PENDING_CHANGES_INTERVAL = ON_DEMAND_POLL_INTERVAL + + def _service_pending_changes(self) -> None: + """Apply changes made elsewhere while the display thread is busy. + + The main loop applies on-demand requests, the display on/off schedule + and brightness (a saved display.hardware.brightness, or a dim-schedule + transition) once per pass -- that is, once per screen. A screen can + dwell a minute and a Vegas iteration four, so those waited just as + long: an on-demand request sat unread for up to 240s, and a brightness + saved and reverted within one screen never reached the panel at all. + + So the places the display thread spends long stretches -- the dwell + sleep, the per-screen render loops and Vegas's interrupt check -- call + this. It is throttled to PENDING_CHANGES_INTERVAL, runs on the display + thread (set_brightness must not be called from the config watcher), + and does what the main loop does, in the same order. Callers decide + what to do about the result from the state it leaves behind + (on_demand_active, current_display_mode, is_display_active). + """ + now = time.monotonic() + last = self._last_pending_service + if last is not None and now - last < self.PENDING_CHANGES_INTERVAL: + return + self._last_pending_service = now + + try: + self._poll_on_demand_requests() + self._check_on_demand_expiration() + self._evaluate_schedule() + self._apply_brightness_target(repaint=True) + except Exception: # pylint: disable=broad-except + # Called from inside Vegas and the render loops; a failure here + # must not take the display loop down with it. + logger.exception("Error servicing pending display changes") + + def _evaluate_schedule(self) -> None: + """Re-check the on/off schedule, letting an on-demand session override it.""" + self._check_schedule() + if self.on_demand_active and not self.is_display_active: + if not self.on_demand_schedule_override: + logger.info("On-demand override keeping display active during scheduled downtime") + self.on_demand_schedule_override = True + self.is_display_active = True + elif not self.on_demand_active and self.on_demand_schedule_override: + self.on_demand_schedule_override = False + + def _apply_brightness_target(self, repaint: bool = False) -> None: + """Push the brightness the config and dim schedule call for, if it changed. + + Args: + repaint: Re-push the current frame afterwards. The panel only shows + a new brightness from the next frame pushed; the main loop is + about to render one, but a dwell or a static screen may not + push again for a minute. + """ + if not self.is_display_active: + return + target = self._check_dim_schedule() + if target == self.current_brightness: + self._failed_brightness_target = None + return + if target == self._failed_brightness_target: + return + if self.display_manager.set_brightness(target): + self.current_brightness = target + self._failed_brightness_target = None + if repaint: + self.display_manager.update_display() + else: + self._failed_brightness_target = target + def _select_startup_plugins(self, discovered_plugins: List[str], on_demand_config: Optional[Dict[str, Any]]) -> List[str]: """Which plugins to load at startup, restoring on-demand state if any. @@ -1845,21 +1955,11 @@ class DisplayController: self._log_memory_stats_if_due() # Check the schedule - self._check_schedule() - if self.on_demand_active and not self.is_display_active: - if not self.on_demand_schedule_override: - logger.info("On-demand override keeping display active during scheduled downtime") - self.on_demand_schedule_override = True - self.is_display_active = True - elif not self.on_demand_active and self.on_demand_schedule_override: - self.on_demand_schedule_override = False + self._evaluate_schedule() - # Check dim schedule and apply brightness (only when display is active) - if self.is_display_active: - target_brightness = self._check_dim_schedule() - if target_brightness != self.current_brightness: - if self.display_manager.set_brightness(target_brightness): - self.current_brightness = target_brightness + # Check dim schedule and apply brightness (only when display + # is active). No repaint: this screen's first frame pushes it. + self._apply_brightness_target() if not self.is_display_active: # Clear display when schedule makes it inactive to ensure blank screen @@ -2010,6 +2110,10 @@ class DisplayController: else: # Vegas was interrupted (live priority), fall through to normal handling logger.debug("Vegas mode interrupted, falling back to normal rotation") + if not self.is_display_active: + # Scheduled off mid-iteration: blank the + # panel now rather than render a screen. + continue except Exception: logger.exception("Vegas mode error") # Fall through to normal rotation on error @@ -2473,8 +2577,8 @@ class DisplayController: self._send_follower_frame(manager_to_display) self._tick_plugin_updates() - self._poll_on_demand_requests() - self._check_on_demand_expiration() + # Throttled: one clock compare between passes. + self._service_pending_changes() # Pace to the frame deadline rather than sleeping a flat # interval on top of the work. display() has already @@ -2492,7 +2596,8 @@ class DisplayController: # update threads and the web UI are not starved of the GIL. time.sleep(_remaining if _remaining > 0 else 0.001) - if self.current_display_mode != active_mode: + if (self.current_display_mode != active_mode + or not self.is_display_active): logger.debug("Mode changed during high-FPS loop, breaking early") break @@ -2570,9 +2675,9 @@ class DisplayController: # Multi-display sync: send follower frame after each render self._send_follower_frame(manager_to_display) - self._poll_on_demand_requests() - self._check_on_demand_expiration() - if self.current_display_mode != active_mode: + self._service_pending_changes() + if (self.current_display_mode != active_mode + or not self.is_display_active): logger.info("Mode changed during display loop from %s to %s, breaking early", active_mode, self.current_display_mode) break @@ -2599,7 +2704,10 @@ class DisplayController: # Removing it again will silently reintroduce both issues. _activate_on_demand # already sets force_change=True and clears the display, so the next loop # iteration renders the new mode immediately. - if self.current_display_mode != active_mode: + # Likewise if the schedule turned the display off + # mid-screen: the next iteration blanks it. + if (self.current_display_mode != active_mode + or not self.is_display_active): continue # Ensure we honour minimum duration when not dynamic and loop ended early @@ -2647,7 +2755,15 @@ class DisplayController: else: # For non-plugin modes, use the original behavior self._sleep_with_plugin_updates(max_duration) - + + # The dwell sleeps above return early when a pending change + # (on-demand started or stopped, display scheduled off) has + # already decided what comes next; rotating now would skip it + # -- an on-demand start would advance past the requested mode. + if (self.current_display_mode != active_mode + or not self.is_display_active): + continue + # Move to next mode if self.on_demand_active: # Guard against empty on_demand_modes to prevent ZeroDivisionError diff --git a/test/test_display_pending_changes.py b/test/test_display_pending_changes.py new file mode 100644 index 00000000..8a4e4684 --- /dev/null +++ b/test/test_display_pending_changes.py @@ -0,0 +1,592 @@ +"""Changes made from the web UI must reach the panel while a screen is showing. + +The main loop applies on-demand requests, the display schedule and brightness +once per pass -- once per screen. Screens are long: a dwell can be a minute and +a Vegas iteration runs for vegas_scroll.max_cycle_duration (240s here). On a +real Pi on 2026-09-23: + + * an on-demand request posted at 10:54:27 was activated at 10:57:24, when the + Vegas iteration it arrived during finally ended -- nothing read the mailbox + in between, because _check_vegas_interrupt only looked at a flag that the + main-loop read sets; + * two brightness saves 12s apart inside one 30s screen never showed at all. + +_service_pending_changes is the fix: a throttled pass the dwell sleep, the +render loops and the Vegas interrupt check all call. These tests drive those +long stretches on a fake clock and check a change lands within one throttle +interval -- and that the throttle holds, since the callers run at frame rate. +""" + +import threading +import types +from datetime import datetime, timedelta +from unittest.mock import MagicMock, patch + +import pytest + +from src.vegas_mode.config import VegasModeConfig +from src.vegas_mode.coordinator import VegasModeCoordinator + +VEGAS_ITERATION_SECONDS = 240 +REQUEST_KEY = 'display_on_demand_request' + + +class FakeClock: + """A clock that only moves when the code under test sleeps. + + Installed as the ``time`` module of display_controller and the Vegas + coordinator only -- patching the real module would also speed up every + background thread in the process. + """ + + def __init__(self, start=10_000.0): + self.t = start + self._events = [] + + def now(self): + return self.t + + def sleep(self, seconds): + # Never zero, so a loop that sleeps "the rest of the frame" still + # advances when its fake frame took no time. + self.t += max(seconds, 0.0005) + while self._events and self._events[0][0] <= self.t: + _, fn = self._events.pop(0) + fn() + + def after(self, seconds, fn): + """Run fn once the clock is `seconds` past now.""" + self._events.append((self.t + seconds, fn)) + self._events.sort(key=lambda e: e[0]) + + def module(self): + return types.SimpleNamespace(time=self.now, monotonic=self.now, + perf_counter=self.now, sleep=self.sleep) + + +@pytest.fixture +def clock(monkeypatch): + c = FakeClock() + fake_time = c.module() + monkeypatch.setattr('src.display_controller.time', fake_time) + monkeypatch.setattr('src.vegas_mode.coordinator.time', fake_time) + return c + + +@pytest.fixture +def controller(test_display_controller, clock): + """A controller at rest: no schedule, full brightness, empty mailbox.""" + c = test_display_controller + c._refresh_config_cache({'display': {'hardware': {'brightness': 90}}}) + c.current_brightness = 90 + c.is_display_active = True + c._check_wifi_status_message = MagicMock(return_value=None) + + c.mailbox = {} # what the web process has written + + def cache_get(key, *args, **kwargs): + if key == REQUEST_KEY: + return c.mailbox.get('request') + return None + + def cache_delete(key): + if key == REQUEST_KEY: + c.mailbox.pop('request', None) + + c.cache_manager.get = MagicMock(side_effect=cache_get) + c.cache_manager.delete = MagicMock(side_effect=cache_delete) + c.cache_manager.set = MagicMock() + c.display_manager.set_brightness = MagicMock(return_value=True) + c.display_manager.update_display = MagicMock() + + # Two loaded plugins: the rotation is on one, on-demand can ask for either. + for mode in ('clock', 'weather'): + c.plugin_display_modes[mode] = [mode] + c.plugin_modes[mode] = MagicMock(spec=[]) + c.mode_to_plugin_id[mode] = mode + c.available_modes = ['clock', 'weather'] + c.current_mode_index = 0 + c.current_display_mode = 'clock' + return c + + +def post_request(controller, request_id='r1', mode='clock'): + controller.mailbox['request'] = { + 'request_id': request_id, 'action': 'start', + 'plugin_id': mode, 'mode': mode, + } + + +def save_brightness(controller, brightness): + """What the config watcher thread does when the web UI saves.""" + controller._controller_config_change( + controller.config, + {'display': {'hardware': {'brightness': brightness}}}) + + +def mailbox_reads(controller): + return sum(1 for call in controller.cache_manager.get.call_args_list + if call.args and call.args[0] == REQUEST_KEY) + + +def vegas_coordinator(controller): + """A coordinator that runs real run_iteration() over a stub renderer. + + Wired to the controller's interrupt checker the way _initialize_vegas_mode + wires it (every 10 frames). + """ + coord = VegasModeCoordinator.__new__(VegasModeCoordinator) + coord.vegas_config = VegasModeConfig.from_config({'display': {'vegas_scroll': { + 'enabled': True, 'max_cycle_duration': VEGAS_ITERATION_SECONDS}}}) + assert coord.vegas_config.continuous_scroll + coord.render_pipeline = MagicMock() + coord.stream_manager = MagicMock() + coord.display_manager = controller.display_manager + coord.stats = {'cycles_completed': 0, 'interruptions': 0} + coord._state_lock = threading.Lock() + coord._is_active = True + coord._is_paused = False + coord._should_stop = False + coord._fps_last_health_log = 0.0 + coord._fps_was_degraded = False + coord._live_priority_check = None + coord._live_priority_active = False + coord._update_callback = None + coord._update_tick_running = False + coord.sync_manager = None + coord._update_static_mode_plugins = lambda: None + coord._check_static_plugin_trigger = lambda: None + coord.frames = 0 + + def run_frame(): + coord.frames += 1 + return True + + coord.run_frame = run_frame + coord.set_interrupt_checker(controller._check_vegas_interrupt, check_interval=10) + return coord + + +def frame_budget(coord): + """Longest a change can wait for the next interrupt check.""" + return coord._interrupt_check_interval * coord.vegas_config.get_frame_interval() + + +class TestOnDemandDuringVegas: + def test_a_request_mid_iteration_ends_the_iteration_promptly(self, controller, clock): + coord = vegas_coordinator(controller) + start = clock.t + posted = {} + + def post(): + posted['at'] = clock.t + post_request(controller, mode='weather') + + clock.after(30, post) + completed = coord.run_iteration() + latency = clock.t - posted['at'] + + assert completed is False, "Vegas ran the whole iteration past the request" + assert controller.on_demand_active + assert controller.current_display_mode == 'weather' + assert latency <= controller.PENDING_CHANGES_INTERVAL + frame_budget(coord) + 0.01, ( + f"request took {latency:.2f}s to interrupt Vegas") + assert clock.t - start < VEGAS_ITERATION_SECONDS + + def test_the_interrupt_checker_answers_within_one_interval(self, controller, clock): + controller._service_pending_changes() # the main loop's last pass + post_request(controller) + waited = 0.0 + step = 0.01 + while not controller._check_vegas_interrupt(): + clock.t += step + waited += step + assert waited <= controller.PENDING_CHANGES_INTERVAL + step, ( + "interrupt checker never saw the request") + assert controller.on_demand_active + + def test_an_empty_mailbox_lets_the_iteration_run_out(self, controller, clock): + coord = vegas_coordinator(controller) + assert coord.run_iteration() is True + assert not controller.on_demand_active + # And the read is throttled: at most one per interval, not per check. + max_reads = VEGAS_ITERATION_SECONDS / controller.PENDING_CHANGES_INTERVAL + 1 + assert 0 < mailbox_reads(controller) <= max_reads + checks = coord.frames // coord._interrupt_check_interval + assert mailbox_reads(controller) < checks / 2 + + +class TestBrightnessIsAppliedMidScreen: + def _applied_at(self, controller, clock, brightness): + """Clock time set_brightness(brightness) was first called, or None.""" + return controller.applied.get(brightness) + + def _record_brightness(self, controller, clock): + controller.applied = {} + + def set_brightness(value): + controller.applied.setdefault(value, clock.t) + return True + + controller.display_manager.set_brightness = MagicMock(side_effect=set_brightness) + + def test_during_a_long_dwell(self, controller, clock): + """The observed failure: two saves 12s apart inside one screen.""" + self._record_brightness(controller, clock) + saves = {} + + def save(value): + saves[value] = clock.t + save_brightness(controller, value) + + clock.after(5, lambda: save(40)) + clock.after(17, lambda: save(90)) + controller._sleep_with_plugin_updates(30) + + interval = controller.PENDING_CHANGES_INTERVAL + assert 40 in controller.applied, "brightness 40 never reached the panel" + assert controller.applied[40] - saves[40] <= interval + 0.01 + assert controller.applied[90] - saves[90] <= interval + 0.01 + assert controller.current_brightness == 90 + + def test_the_frame_is_repushed_so_the_change_shows(self, controller, clock): + clock.after(1, lambda: save_brightness(controller, 40)) + controller._sleep_with_plugin_updates(5) + calls = controller.display_manager.method_calls + names = [c[0] for c in calls] + assert 'set_brightness' in names + assert 'update_display' in names[names.index('set_brightness'):], ( + "nothing pushed a frame after the brightness change") + + def test_during_a_vegas_iteration(self, controller, clock): + self._record_brightness(controller, clock) + coord = vegas_coordinator(controller) + saved = {} + + def save(): + saved['at'] = clock.t + save_brightness(controller, 25) + + clock.after(60, save) + assert coord.run_iteration() is True # brightness does not end the scroll + assert 25 in controller.applied, "brightness waited for the iteration to end" + assert controller.applied[25] - saved['at'] <= ( + controller.PENDING_CHANGES_INTERVAL + frame_budget(coord) + 0.01) + + def test_a_dim_schedule_transition_mid_dwell(self, controller, clock): + self._record_brightness(controller, clock) + controller._refresh_config_cache({ + 'display': {'hardware': {'brightness': 90}}, + 'dim_schedule': {'enabled': True, 'start_time': '20:00', + 'end_time': '07:00', 'dim_brightness': 20}, + }) + wall_start = datetime(2026, 9, 21, 19, 59, 30) + t0 = clock.t + with patch('src.display_controller.datetime') as mock_dt: + mock_dt.strptime = datetime.strptime + mock_dt.now.side_effect = lambda tz=None: wall_start + timedelta(seconds=clock.t - t0) + controller._sleep_with_plugin_updates(60) + + assert 20 in controller.applied, "the dim period started mid-screen and was missed" + dimmed_at = t0 + 30 # 20:00:00 + assert controller.applied[20] - dimmed_at <= controller.PENDING_CHANGES_INTERVAL + 0.01 + + +class TestThrottle: + def test_no_reads_or_brightness_calls_between_passes(self, controller, clock): + controller._service_pending_changes() + reads = mailbox_reads(controller) + save_brightness(controller, 40) + post_request(controller) + for _ in range(500): # a few seconds of frames, all inside one interval + controller._service_pending_changes() + clock.t += controller.PENDING_CHANGES_INTERVAL / 1000 + assert mailbox_reads(controller) == reads + controller.display_manager.set_brightness.assert_not_called() + assert not controller.on_demand_active + + clock.t += controller.PENDING_CHANGES_INTERVAL + controller._service_pending_changes() + # The poll, plus the consume step's re-read of the request it acted on. + assert mailbox_reads(controller) > reads + controller.display_manager.set_brightness.assert_called_once_with(40) + assert controller.on_demand_active + + def test_between_passes_it_does_no_work_at_all(self, controller, clock): + """The high-FPS and Vegas loops call this every frame.""" + controller._service_pending_changes() + controller._poll_on_demand_requests = MagicMock() + controller._check_dim_schedule = MagicMock(return_value=90) + controller._check_schedule = MagicMock() + for _ in range(100): + controller._service_pending_changes() + controller._check_vegas_interrupt() + controller._poll_on_demand_requests.assert_not_called() + controller._check_dim_schedule.assert_not_called() + controller._check_schedule.assert_not_called() + + def test_an_unchanged_brightness_is_not_re_sent(self, controller, clock): + for _ in range(20): + controller._service_pending_changes() + clock.t += controller.PENDING_CHANGES_INTERVAL + controller.display_manager.set_brightness.assert_not_called() + + def test_a_refused_brightness_is_not_retried_every_pass(self, controller, clock): + controller.display_manager.set_brightness = MagicMock(return_value=False) + save_brightness(controller, 40) + for _ in range(20): + controller._service_pending_changes() + clock.t += controller.PENDING_CHANGES_INTERVAL + controller.display_manager.set_brightness.assert_called_once_with(40) + assert controller.current_brightness == 90 + # A different target is still tried. + save_brightness(controller, 50) + controller._service_pending_changes() + controller.display_manager.set_brightness.assert_called_with(50) + + def test_a_refused_brightness_is_tried_again_after_going_back(self, controller, clock): + controller.display_manager.set_brightness = MagicMock(return_value=False) + for value in (40, 90, 40): + save_brightness(controller, value) + controller._service_pending_changes() + clock.t += controller.PENDING_CHANGES_INTERVAL + assert controller.display_manager.set_brightness.call_count == 2 + + +class TestScheduleAndDwells: + def _schedule_until(self, controller, end_time): + controller._refresh_config_cache({ + 'display': {'hardware': {'brightness': 90}}, + 'schedule': {'enabled': True, 'start_time': '07:00', 'end_time': end_time}, + }) + + def test_vegas_hands_back_when_the_display_is_scheduled_off(self, controller, clock): + self._schedule_until(controller, '22:59') + coord = vegas_coordinator(controller) + wall_start = datetime(2026, 9, 21, 22, 59, 0) + t0 = clock.t + with patch('src.display_controller.datetime') as mock_dt: + mock_dt.strptime = datetime.strptime + mock_dt.now.side_effect = lambda tz=None: wall_start + timedelta(seconds=clock.t - t0) + completed = coord.run_iteration() + assert completed is False + assert not controller.is_display_active + off_at = t0 + 60 # 23:00:00 + assert clock.t - off_at <= controller.PENDING_CHANGES_INTERVAL + frame_budget(coord) + 0.01 + + def test_on_demand_wakes_a_scheduled_off_display_within_an_interval(self, controller, clock): + """The display-off branch sleeps 60s at a time between checks.""" + controller.is_display_active = False + controller._evaluate_schedule = MagicMock() # hold the schedule "off" + posted = {} + + def post(): + posted['at'] = clock.t + post_request(controller) + + clock.after(3, post) + start = clock.t + controller._sleep_with_plugin_updates(60) + assert controller.on_demand_active + assert clock.t - start < 60, "the off-schedule sleep ran its full minute" + assert clock.t - posted['at'] <= controller.PENDING_CHANGES_INTERVAL + 0.01 + + def test_the_dwell_returns_early_on_an_on_demand_stop(self, controller, clock): + post_request(controller) + controller._service_pending_changes() + assert controller.on_demand_active + clock.t += controller.PENDING_CHANGES_INTERVAL + controller.mailbox['request'] = {'request_id': 'r2', 'action': 'stop'} + start = clock.t + controller._sleep_with_plugin_updates(30) + assert not controller.on_demand_active + assert clock.t - start <= controller.PENDING_CHANGES_INTERVAL + 0.01 + + def test_a_quiet_dwell_runs_its_full_length(self, controller, clock): + start = clock.t + controller._sleep_with_plugin_updates(30) + assert clock.t - start >= 30 + + +class _Plugin: + """A static (non-scrolling) plugin mode that records what was shown.""" + + needs_high_fps = False + + def __init__(self, mode, plugin_id, shown, results=()): + self.mode = mode + self.plugin_id = plugin_id + self._shown = shown + self._results = iter(results) + + def display(self, force_clear=False): + self._shown.append(self.mode) + if len(self._shown) > 6: + raise KeyboardInterrupt # ends run(); it catches this and cleans up + return next(self._results, True) + + +class TestRunLoopDoesNotSkipTheRequestedMode: + def test_on_demand_during_the_minimum_duration_sleep(self, controller, clock): + """A dwell sleep that ends early on an on-demand start must not rotate. + + The first screen shows once, then reports no content, so run() sleeps + out the rest of its minimum duration. The request lands during that + sleep. Falling through to "move to next mode" advanced the new + on-demand rotation past the mode that was asked for. + """ + c = controller + shown = [] + c.plugin_modes.clear() + c.mode_to_plugin_id.clear() + c.plugin_display_modes.clear() + modes = {'clock': ('clock', [True, False]), 'news': ('news', []), + 'weather': ('weather', []), 'weather_hourly': ('weather', [])} + for mode, (pid, results) in modes.items(): + c.plugin_modes[mode] = _Plugin(mode, pid, shown, results) + c.mode_to_plugin_id[mode] = pid + c.plugin_display_modes.setdefault(pid, []).append(mode) + c.available_modes = list(modes) + c.current_mode_index = 0 + c._cleanup_expired_wifi_status = MagicMock() + c.config.setdefault('display', {})['display_durations'] = {'clock': 30} + c.plugin_manager.plugin_executor.execute_display.side_effect = ( + lambda target, plugin_id, force_clear=False, display_mode=None, **kw: + target.display(force_clear=force_clear)) + clock.after(10, lambda: post_request(c, mode='weather')) + + c.run() + + assert c.on_demand_modes[:2] == ['weather', 'weather_hourly'] + assert shown[:2] == ['clock', 'clock'] + assert shown[2] == 'weather', f"on-demand opened on {shown[2]}, not the requested mode" + + +class _Stop(KeyboardInterrupt): + """Ends run(): it catches KeyboardInterrupt and cleans up.""" + + +class TestRunLoopBlanksWhenVegasHandsBack: + def test_scheduled_off_mid_iteration_blanks_instead_of_rendering(self, controller, clock): + c = controller + shown = [] + c.plugin_modes['clock'] = _Plugin('clock', 'clock', shown) + c.plugin_modes['weather'] = _Plugin('weather', 'weather', shown) + c._cleanup_expired_wifi_status = MagicMock() + c._refresh_config_cache({ + 'display': {'hardware': {'brightness': 90}}, + 'schedule': {'enabled': True, 'start_time': '07:00', 'end_time': '22:59'}, + }) + c.vegas_coordinator = vegas_coordinator(c) + c.vegas_coordinator._pending_config_update = False + + def stop(): + raise _Stop() + + wall_start = datetime(2026, 9, 21, 22, 59, 0) + t0 = clock.t + clock.after(90, stop) + with patch('src.display_controller.datetime') as mock_dt: + mock_dt.strptime = datetime.strptime + mock_dt.now.side_effect = lambda tz=None: wall_start + timedelta(seconds=clock.t - t0) + c.run() + + assert c.vegas_coordinator.frames > 0 # Vegas was running first + assert not c.is_display_active + assert shown == [], f"rendered {shown} after the display was scheduled off" + c.display_manager.clear.assert_called() + + @pytest.mark.parametrize('needs_high_fps', [True, False]) + def test_a_screen_scheduled_off_midway_stops_rendering(self, controller, clock, + needs_high_fps): + c = controller + shown_at = [] + plugin = _Plugin('ticker', 'ticker', []) + plugin.needs_high_fps = needs_high_fps + plugin.display = lambda force_clear=False: shown_at.append(clock.t) or True + c.plugin_modes.clear() + c.mode_to_plugin_id.clear() + c.plugin_display_modes.clear() + c.plugin_modes['ticker'] = plugin + c.mode_to_plugin_id['ticker'] = 'ticker' + c.plugin_display_modes['ticker'] = ['ticker'] + c.available_modes = ['ticker'] + c.current_mode_index = 0 + c._cleanup_expired_wifi_status = MagicMock() + c._refresh_config_cache({ + 'display': {'hardware': {'brightness': 90}, + 'display_durations': {'ticker': 120}}, + 'schedule': {'enabled': True, 'start_time': '07:00', 'end_time': '22:59'}, + }) + c.plugin_manager.plugin_executor.execute_display.side_effect = ( + lambda target, plugin_id, force_clear=False, display_mode=None, **kw: + target.display(force_clear=force_clear)) + + def stop(): + raise _Stop() + + wall_start = datetime(2026, 9, 21, 22, 59, 0) + t0 = clock.t + clock.after(150, stop) + with patch('src.display_controller.datetime') as mock_dt: + mock_dt.strptime = datetime.strptime + mock_dt.now.side_effect = lambda tz=None: wall_start + timedelta(seconds=clock.t - t0) + c.run() + + off_at = t0 + 60 # 23:00:00, halfway through a 120s screen + assert shown_at and shown_at[0] < off_at + assert shown_at[-1] - off_at <= 1.0 + c.PENDING_CHANGES_INTERVAL, ( + f"kept rendering {shown_at[-1] - off_at:.1f}s after the display was scheduled off") + assert not c.is_display_active + + +class TestRunLoopAppliesBrightnessMidScreen: + """The render loops of run() itself, not just the helpers they call.""" + + def _run_screen(self, c, clock, needs_high_fps): + shown = [] + plugin = _Plugin('ticker', 'ticker', shown) + plugin.needs_high_fps = needs_high_fps + plugin.display = lambda force_clear=False: shown.append('ticker') or True + c.plugin_modes.clear() + c.mode_to_plugin_id.clear() + c.plugin_display_modes.clear() + c.plugin_modes['ticker'] = plugin + c.mode_to_plugin_id['ticker'] = 'ticker' + c.plugin_display_modes['ticker'] = ['ticker'] + c.available_modes = ['ticker'] + c.current_mode_index = 0 + c._cleanup_expired_wifi_status = MagicMock() + c.config.setdefault('display', {})['display_durations'] = {'ticker': 60} + c.plugin_manager.plugin_executor.execute_display.side_effect = ( + lambda target, plugin_id, force_clear=False, display_mode=None, **kw: + target.display(force_clear=force_clear)) + applied = {} + + def set_brightness(value): + applied.setdefault(value, clock.t) + return True + + c.display_manager.set_brightness = MagicMock(side_effect=set_brightness) + start = clock.t + clock.after(5, lambda: save_brightness(c, 30)) + + def stop(): + raise _Stop() + + clock.after(20, stop) + c.run() + return start, applied, shown + + def test_a_scrolling_screen(self, controller, clock): + start, applied, shown = self._run_screen(controller, clock, needs_high_fps=True) + assert len(shown) > 100 # really was the high-FPS loop + assert 30 in applied, "brightness waited for the scrolling screen to end" + assert applied[30] - (start + 5) <= controller.PENDING_CHANGES_INTERVAL + 0.02 + + def test_a_static_screen(self, controller, clock): + start, applied, shown = self._run_screen(controller, clock, needs_high_fps=False) + assert len(shown) < 30 # really was the once-a-second loop + assert 30 in applied, "brightness waited for the static screen to end" + # This loop redraws once a second, and services changes after each redraw. + assert applied[30] - (start + 5) <= 1.0 + controller.PENDING_CHANGES_INTERVAL