mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-10 17:16:36 +00:00
Merge origin/main into claude/remove-skins-and-base-classes
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
@@ -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.
|
||||
|
||||
### Removed
|
||||
|
||||
- **The skin system.** Skins never rendered with the current scoreboard
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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'
|
||||
})
|
||||
|
||||
|
||||
+15
-58
@@ -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 "<name>.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,
|
||||
|
||||
+140
-24
@@ -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
|
||||
|
||||
+195
-75
@@ -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 ``<name>.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)
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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"<html>oops</html>", 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"]
|
||||
|
||||
@@ -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/*"
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
"""
|
||||
web_interface/app.py must start with an absolute ``plugins_directory``.
|
||||
|
||||
``project_root`` used to be assigned only in the relative-path branch, so an
|
||||
absolute path in config.json raised NameError at module import (the
|
||||
SchemaManager construction was the first use). Each case imports the real
|
||||
module in a fresh interpreter, since app.py does its setup at import time and
|
||||
the test process may already hold a copy built from the real config.
|
||||
"""
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
|
||||
_IMPORT_APP = textwrap.dedent(
|
||||
"""
|
||||
import json, sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, {root!r})
|
||||
from src.config_manager import ConfigManager
|
||||
ConfigManager.load_config = lambda self: {config!r}
|
||||
import web_interface.app as web_app
|
||||
print(json.dumps({{
|
||||
"plugins_dir": str(web_app.plugins_dir),
|
||||
"project_root": str(web_app.project_root),
|
||||
"schema_project_root": str(web_app.schema_manager.project_root),
|
||||
}}))
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def _import_app_with(plugins_directory):
|
||||
config = {"plugin_system": {"plugins_directory": plugins_directory}}
|
||||
script = _IMPORT_APP.format(root=str(PROJECT_ROOT), config=config)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", script],
|
||||
cwd=str(PROJECT_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
return json.loads(result.stdout.strip().splitlines()[-1])
|
||||
|
||||
|
||||
def test_absolute_plugins_directory_imports(tmp_path):
|
||||
plugins = tmp_path / "my-plugins"
|
||||
plugins.mkdir()
|
||||
|
||||
info = _import_app_with(str(plugins))
|
||||
|
||||
assert Path(info["plugins_dir"]) == plugins
|
||||
assert Path(info["project_root"]) == PROJECT_ROOT
|
||||
assert Path(info["schema_project_root"]) == PROJECT_ROOT
|
||||
|
||||
|
||||
def test_relative_plugins_directory_resolves_under_project_root():
|
||||
info = _import_app_with("plugin-repos")
|
||||
|
||||
assert Path(info["plugins_dir"]) == PROJECT_ROOT / "plugin-repos"
|
||||
assert Path(info["project_root"]) == PROJECT_ROOT
|
||||
@@ -89,12 +89,15 @@ config = config_manager.load_config()
|
||||
plugin_system_config = config.get('plugin_system', {})
|
||||
plugins_dir_name = plugin_system_config.get('plugins_directory', 'plugin-repos')
|
||||
|
||||
# Project root (LEDMatrix directory). Needed below for data/ and assets/ paths
|
||||
# whether or not the plugins directory is absolute.
|
||||
project_root = Path(__file__).parent.parent
|
||||
|
||||
# Resolve plugin directory - handle both absolute and relative paths
|
||||
if os.path.isabs(plugins_dir_name):
|
||||
plugins_dir = Path(plugins_dir_name)
|
||||
else:
|
||||
# If relative, resolve relative to the project root (LEDMatrix directory)
|
||||
project_root = Path(__file__).parent.parent
|
||||
# If relative, resolve relative to the project root
|
||||
plugins_dir = project_root / plugins_dir_name
|
||||
|
||||
plugin_manager = PluginManager(
|
||||
|
||||
Reference in New Issue
Block a user