rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-04 06:15:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): font zip cache, monotonic timers, resolver back-off, and other core/common fixes (#654)

Browse Source 
* fix(core): font zip cache, monotonic timers, resolver back-off, and other core/common fixes

- font_manager: a .zip font URL is served as its extracted font after a
  restart (the cached-file check returned the archive first); downloads
  use requests with a 30s timeout into a temp file + os.replace.
- api_helper / sync_manager: rate-limit and heartbeat/leader timeouts use
  time.monotonic(); last_request_time and the status file's ts stay
  wall-clock. set_on_new_cycle docstring no longer claims core uses it.
- logo_helper: the placeholder uses the same scaled box as a real logo.
- permission_utils: one _sudo_bash_candidates() helper (with the sudoers
  exact-argv rationale) shared by sudo_remove_directory, which now retries
  the next bash path on a sudo refusal, and install_requirements_file.
- dynamic_team_resolver: failed/empty fetch backs off 5 min; duplicate
  INFO log and contradictory docstring example fixed.
- element_style: scale default looked up through element aliases.
- background_data_service: cache-hit callback runs outside the lock.
- config_arrays: union-aware type check (["array","null"]); stale
  dotToNested() reference removed.
- auto_update_setup: non-dict auto_update reads as off; temp result file
  unlinked when the write fails.
- exceptions: constructors copy the caller's context dict.
- logging_config: StructuredFormatter json.dumps(default=str).
- error_aggregator: removed unused export_path/export_to_file/_auto_export.
- Docstrings: validate_file_upload max_size_mb, raise_on_errors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(sync): retry the status-file rename like the other atomic writers

On Windows os.replace can fail with "Access is denied" while a scanner
briefly holds the target open; config_manager_atomic._replace already
retries that (and re-raises at once on other platforms). The sync status
writer called os.replace directly, which made
test_concurrent_writers_each_use_their_own_temp_file flaky on Windows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-28 10:40:16 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 6f45ff5e63
commit f6c0fe55d9
28 changed files with 596 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+12
View File
@@ -19,6 +19,18 @@ accepts both, but the store flags the old spelling as deprecated
## Unreleased
- Core services and `src.common` fixes:
- A plugin font declared as a `.zip` URL is served as the font extracted from it after a restart, instead of registering the archive itself. Font downloads time out after 30s and land in the cache only once complete, so an interrupted download is retried rather than served forever.
- `APIHelper`'s rate limit and the display-sync heartbeat/leader timeouts measure elapsed time with `time.monotonic()`. A wall-clock step (NTP correcting a Pi with no RTC) could stall API requests for as long as the step or fake a sync timeout. `get_request_stats()['last_request_time']` is still wall-clock time.
- `LogoHelper.load_logo_with_download()` sizes its placeholder to the scaled logo box, like a real logo (only differs when `scale` isn't 1).
- The AP Top 25 resolver remembers a failed or empty rankings fetch for 5 minutes, so an ESPN outage no longer costs every scoreboard update a 30s timeout. Its duplicate INFO log line is gone.
- `sudo_remove_directory()` tries each bash path the sudoers rule might name, as `install_requirements_file()` already did.
- An element's saved layout `scale` equal to its schema default is no longer treated as a user choice when the default is declared under an alias (`score` for `score_text`).
- `BackgroundDataService` runs a cache-hit callback outside its lock, as the fetch path does.
- Plugin config saves recombine position-keyed inputs for nullable array fields (`"type": ["array", "null"]`).
- A hand-edited non-object `auto_update` value reads as off instead of raising at startup, and a failed result write no longer leaves a temp file behind.
- `CacheError`/`ConfigError`/`PluginError`/`DisplayError` no longer write their key into the caller's `context` dict; the JSON log formatter stringifies values it can't encode instead of dropping the record.
- Removed `ErrorAggregator`'s unused JSON export (`export_path`, `export_to_file()`); nothing called it. Docstring fixes in `validate_file_upload`, `StartupValidator.raise_on_errors`, `DisplaySyncManager.set_on_new_cycle`, `dynamic_team_resolver` and `config_arrays`.
- Display thread-safety and consistency fixes:
- `DisplayManager.defer_update()` from a plugin's update thread no longer loses queued updates while the render thread processes the queue; the queue is locked, and the queued callables still run outside the lock.
- BDF fonts: `FontManager.get_font()` and `element_style.load_font()` no longer hand one `freetype.Face` to every thread. BDF faces come from `load_bdf_face`, which already caches them per thread; TrueType fonts are cached as before. `element_style`'s font cache is locked (a concurrent eviction could raise `KeyError`).
src/auto_update_setup.py
+25 -9
View File
@@ -70,7 +70,14 @@ def _read(path):
def is_enabled(config):
return bool((config.get('auto_update') or {}).get('enabled', False))
# Only the {"enabled": true} object turns this on. A hand-edited
# non-dict (e.g. "auto_update": true) raised AttributeError here and
# aborted startup setup; the web UI's save replaces such a value with {}
# (disabled), so read it the same way.
section = config.get('auto_update')
if not isinstance(section, dict):
return False
return bool(section.get('enabled', False))
class UpdateHelperSetup:
@@ -201,17 +208,26 @@ class UpdateHelperSetup:
try:
self.result_file.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=str(self.result_file.parent), prefix='.auto_update_setup_')
with os.fdopen(fd, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2)
os.chmod(tmp, 0o644)
if self._web_ids and hasattr(os, 'chown'):
# Only root can give the file away; the result is readable
# (0644) either way, so a failed chown must not lose it.
try:
with os.fdopen(fd, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2)
os.chmod(tmp, 0o644)
if self._web_ids and hasattr(os, 'chown'):
# Only root can give the file away; the result is readable
# (0644) either way, so a failed chown must not lose it.
try:
os.chown(tmp, *self._web_ids)
except OSError:
pass
os.replace(tmp, self.result_file)
except BaseException:
# Don't leave a .auto_update_setup_* file behind in the
# project dir every time the write fails.
try:
os.chown(tmp, *self._web_ids)
os.unlink(tmp)
except OSError:
pass
os.replace(tmp, self.result_file)
raise
except OSError as e:
logger.warning("Could not record automatic update setup result: %s", e)
return result
src/background_data_service.py
+12 -8
View File
@@ -247,15 +247,19 @@ class BackgroundDataService:
# same object the dict holds.
self.completed_requests[request_id] = result
if callback:
try:
callback(result)
except Exception as e:
logger.error(f"Error in callback for request {request_id}: {e}")
self._release_payload(result)
# The callback runs outside the lock, as on the worker path: it is
# plugin code, and holding the service lock through it blocked
# every worker's result bookkeeping (and any other thread's
# submit) for as long as the callback took.
if callback:
try:
callback(result)
except Exception as e:
logger.error(f"Error in callback for request {request_id}: {e}")
self._release_payload(result)
logger.debug(f"Cache hit for {sport} {year} data")
return request_id
logger.debug(f"Cache hit for {sport} {year} data")
return request_id
# limit above 500 makes an ESPN *scoreboard* return a truncated list
# (src/common/espn_dates.py). Other endpoints need more: /teams has 762
src/common/api_helper.py
+17 -9
View File
@@ -84,7 +84,11 @@ class APIHelper:
self.session.headers.update({**DEFAULT_HTTP_HEADERS, 'Connection': 'keep-alive'})
# Rate limiting
self._last_request_time = 0
self._last_request_time = 0 # wall clock, reported by get_request_stats()
# The interval is measured on time.monotonic(): a wall-clock step
# back (NTP correcting a Pi with no RTC) made time_since_last
# negative and the "remaining interval" sleep as long as the step.
self._last_request_monotonic: Optional[float] = None
self._min_request_interval = 1.0 # Minimum seconds between requests
def get(self, url: str, params: Optional[Dict] = None,
@@ -333,13 +337,14 @@ class APIHelper:
def _enforce_rate_limit(self) -> None:
"""Enforce rate limiting between requests."""
current_time = time.time()
time_since_last = current_time - self._last_request_time
if time_since_last < self._min_request_interval:
sleep_time = self._min_request_interval - time_since_last
time.sleep(sleep_time)
if self._last_request_monotonic is not None:
time_since_last = time.monotonic() - self._last_request_monotonic
if time_since_last < self._min_request_interval:
sleep_time = self._min_request_interval - time_since_last
time.sleep(sleep_time)
self._last_request_monotonic = time.monotonic()
self._last_request_time = time.time()
def set_rate_limit(self, min_interval: float) -> None:
@@ -362,5 +367,8 @@ class APIHelper:
return {
'min_request_interval': self._min_request_interval,
'last_request_time': self._last_request_time,
'time_since_last_request': time.time() - self._last_request_time
'time_since_last_request': (
time.monotonic() - self._last_request_monotonic
if self._last_request_monotonic is not None
else time.time() - self._last_request_time),
}
src/common/logo_helper.py
+26 -14
View File
@@ -8,7 +8,7 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins.
import logging
import time
from pathlib import Path
from typing import Dict, List, Optional, Union
from typing import Dict, List, Optional, Tuple, Union
import requests
from PIL import Image, ImageDraw
@@ -125,17 +125,7 @@ class LogoHelper:
# Resolve the effective target size BEFORE the cache lookup so the
# key is size-qualified — a panel-size change must not return a
# logo resized for the old dimensions.
if max_width is None:
max_width = int(self.display_width * DEFAULT_LOGO_BOX_FACTOR)
if max_height is None:
max_height = int(self.display_height * DEFAULT_LOGO_BOX_FACTOR)
# Imported here: src.element_style imports src.common (for bdf_font),
# whose __init__ imports this module.
from src.element_style import coerce_scale
scale = coerce_scale(scale, 1.0)
if scale != 1.0:
max_width = max(1, int(round(max_width * scale)))
max_height = max(1, int(round(max_height * scale)))
max_width, max_height, scale = self._scaled_box(max_width, max_height, scale)
# The key carries the scaled box, so two elements scaled differently
# cannot be served each other's image.
cache_key = f"{team_abbr}_{logo_path}_{max_width}x{max_height}"
@@ -236,8 +226,30 @@ class LogoHelper:
# exists to prevent.
self._refresh_stale_placeholder(logo_path)
# Create placeholder if all else fails
return self._create_placeholder_logo(team_abbr, max_width, max_height)
# Create placeholder if all else fails. Sized to the same scaled box
# a real logo gets, so a scaled element doesn't jump in size while
# its logo is missing.
box_width, box_height, _ = self._scaled_box(max_width, max_height, scale)
return self._create_placeholder_logo(team_abbr, box_width, box_height)
def _scaled_box(self, max_width: Optional[int], max_height: Optional[int],
scale: float) -> Tuple[int, int, float]:
"""The logo box after defaults and the user's scale are applied.
Returns ``(width, height, coerced_scale)``.
"""
if max_width is None:
max_width = int(self.display_width * DEFAULT_LOGO_BOX_FACTOR)
if max_height is None:
max_height = int(self.display_height * DEFAULT_LOGO_BOX_FACTOR)
# Imported here: src.element_style imports src.common (for bdf_font),
# whose __init__ imports this module.
from src.element_style import coerce_scale
scale = coerce_scale(scale, 1.0)
if scale != 1.0:
max_width = max(1, int(round(max_width * scale)))
max_height = max(1, int(round(max_height * scale)))
return max_width, max_height, scale
def _invalidate_cached_logo(self, team_abbr: str, logo_path: Path) -> None:
"""Drop every cached size of one logo after its file changed on disk."""
src/common/permission_utils.py
+41 -24
View File
@@ -290,6 +290,26 @@ def get_cache_dir_mode() -> int:
return 0o2775 # rwxrwsr-x (setgid + group writable)
def _sudo_bash_candidates() -> list:
"""Bash paths to try, in order, when running a vetted helper via sudo.
sudoers matches the exact argv, so ``sudo -n <bash> <helper> ...`` only
works if <bash> is the same path configure_web_sudo.sh wrote into the
rule -- whatever ``command -v bash`` said on the machine that ran it.
On merged-/usr systems /usr/bin/bash and /bin/bash are the same file but
different strings to sudo, and the web user's PATH can differ from the
installer's, so no single guess is reliable. Callers try each in turn and
move on only when sudo refused the command line (SUDO_REFUSAL_PHRASES).
The helper is invoked through bash rather than its shebang for the same
reason: the rule names bash, not the script.
"""
candidates = []
for candidate in ("/usr/bin/bash", "/bin/bash", _shutil.which("bash")):
if candidate and candidate not in candidates:
candidates.append(candidate)
return candidates
def sudo_remove_directory(path: Path, allowed_bases: Optional[list] = None) -> bool:
"""
Remove a directory using sudo as a last resort.
@@ -350,22 +370,25 @@ def sudo_remove_directory(path: Path, allowed_bases: Optional[list] = None) -> b
logger.error(f"Safe removal helper not found: {helper_script}")
return False
bash_path = _shutil.which('bash') or '/bin/bash'
try:
result = subprocess.run(
['sudo', '-n', bash_path, str(helper_script), str(resolved)],
capture_output=True,
text=True,
timeout=30
)
if result.returncode == 0 and not resolved.exists():
logger.info(f"Successfully removed {path} via sudo helper")
return True
else:
stderr = result.stderr.strip()
logger.error(f"sudo helper failed for {path}: {stderr}")
return False
for bash_path in _sudo_bash_candidates():
result = subprocess.run(
['sudo', '-n', bash_path, str(helper_script), str(resolved)],
capture_output=True,
text=True,
timeout=30
)
if result.returncode == 0 and not resolved.exists():
logger.info(f"Successfully removed {path} via sudo helper")
return True
# Only a refused command line is worth another bash path; if the
# helper itself ran and failed, a retry would just repeat it.
if result.returncode == 0 or not any(
phrase in (result.stderr or '') for phrase in SUDO_REFUSAL_PHRASES):
break
stderr = (result.stderr or '').strip()
logger.error(f"sudo helper failed for {path}: {stderr}")
return False
except subprocess.TimeoutExpired:
logger.error(f"sudo helper timed out for {path}")
return False
@@ -417,16 +440,10 @@ def install_requirements_file(req_file: Path, timeout: int = 300) -> subprocess.
wrapper = project_root / "scripts" / "fix_perms" / "safe_pip_install.sh"
if wrapper.exists():
# See sudo_remove_directory / configure_web_sudo.sh for why bash must
# be invoked with an explicit, known path rather than relying on the
# wrapper's shebang: sudoers matches the exact command line.
bash_candidates = []
for candidate in ("/usr/bin/bash", "/bin/bash", _shutil.which("bash")):
if candidate and candidate not in bash_candidates:
bash_candidates.append(candidate)
# See _sudo_bash_candidates for why bash is invoked by explicit path
# and why there is more than one to try.
result = None
for bash_path in bash_candidates:
for bash_path in _sudo_bash_candidates():
# bash_path and wrapper are fixed, known-good paths, and
# safe_pip_install.sh independently re-validates req_file is an
# allowed requirements.txt before installing anything as root.
src/common/sync_manager.py
+21 -10
View File
@@ -32,6 +32,7 @@ from typing import Callable, Optional
import numpy as np
from PIL import Image
from src.config_manager_atomic import _replace
from src.display_geometry import DEFAULT_CHAIN_LENGTH, DEFAULT_COLS, DEFAULT_ROWS
# Raw-frame wire format: 8-byte magic + 4-byte header + raw RGB pixels
@@ -127,6 +128,9 @@ class DisplaySyncManager:
self._peer_ip: Optional[str] = None
self._peer_compatible: bool = False
self._peer_chain: int = 0
# time.monotonic() readings, like _last_leader_frame_time: these only
# feed the timeout watchdogs, and a wall-clock step (NTP correcting a
# Pi with no RTC) would otherwise fake or mask a timeout.
self._last_heartbeat_time: float = 0.0
self._leader_width: int = 0 # set by display_controller after init
self._oversized_frame_warned: bool = False
@@ -206,7 +210,7 @@ class DisplaySyncManager:
self._handle_hello(msg, sender_ip)
elif t == "hb":
if self._peer_ip == sender_ip:
self._last_heartbeat_time = time.time()
self._last_heartbeat_time = time.monotonic()
except socket.timeout:
continue
except Exception as exc:
@@ -229,7 +233,7 @@ class DisplaySyncManager:
self._peer_ip = sender_ip
self._peer_compatible = compatible
self._peer_chain = peer_chain
self._last_heartbeat_time = time.time()
self._last_heartbeat_time = time.monotonic()
prev_state = self._leader_state
if compatible:
@@ -273,7 +277,7 @@ class DisplaySyncManager:
while self._running:
time.sleep(1.0)
if self._leader_state == LeaderState.CONNECTED:
if time.time() - self._last_heartbeat_time > PEER_TIMEOUT:
if time.monotonic() - self._last_heartbeat_time > PEER_TIMEOUT:
self.logger.info(
"Sync: follower heartbeat timeout — peer disconnected"
)
@@ -501,7 +505,7 @@ class DisplaySyncManager:
"""Note that the leader at ``sender_ip`` just sent something, and
switch from standalone to follower mode if not already following.
Returns True if this call made the switch."""
self._last_leader_frame_time = time.time()
self._last_leader_frame_time = time.monotonic()
self._leader_ip = sender_ip
if self._follower_state != FollowerState.STANDALONE:
return False
@@ -621,11 +625,13 @@ class DisplaySyncManager:
heartbeat = json.dumps({"t": "hb"}).encode("utf-8")
dest = ("<broadcast>", self.port)
last_hello = 0.0
last_hb = 0.0
# -inf, not 0.0: monotonic time starts near boot, so "now - 0.0" can
# be under the interval and would delay the first announcement.
last_hello = float("-inf")
last_hb = float("-inf")
while self._running:
now = time.time()
now = time.monotonic()
if now - last_hello >= HELLO_INTERVAL:
try:
self._send_sock.sendto(hello, dest)
@@ -644,7 +650,7 @@ class DisplaySyncManager:
while self._running:
time.sleep(1.0)
if self._follower_state == FollowerState.FOLLOWER:
if time.time() - self._last_leader_frame_time > LEADER_TIMEOUT:
if time.monotonic() - self._last_leader_frame_time > LEADER_TIMEOUT:
self.logger.info(
"Sync: leader frame timeout — returning to standalone mode"
)
@@ -670,7 +676,10 @@ class DisplaySyncManager:
def set_on_new_cycle(self, callback: Callable[[], None]) -> None:
"""Follower: register a callback fired when the leader starts a new scroll cycle.
Used to trigger a local start_new_cycle() so both Pis rebuild from same fresh data.
Nothing in core registers one: display_controller follows the leader
through set_on_scroll_image() and the scroll position instead of
rebuilding locally. The hook stays for callers that want the signal.
"""
self._on_new_cycle = callback
@@ -734,7 +743,9 @@ class DisplaySyncManager:
# mkstemp makes it owner-only; the web UI may run as a
# different user from the display service.
os.chmod(tmp, 0o644)
os.replace(tmp, STATUS_FILE)
# _replace: on Windows a rename can briefly fail with
# "Access is denied" while a scanner holds the target open.
_replace(tmp, STATUS_FILE)
tmp = None
except Exception as exc:
self.logger.debug("Sync: status file write error: %s", exc)
src/dynamic_team_resolver.py
+17 -4
View File
@@ -13,7 +13,8 @@ Supported dynamic teams:
Usage:
resolver = DynamicTeamResolver()
resolved_teams = resolver.resolve_teams(["UGA", "AP_TOP_25", "AUB"])
# Returns: ["UGA", "UGA", "AUB", "MICH", "OSU", ...] (AP_TOP_25 teams)
# Returns: ["UGA", "MICH", "OSU", ..., "AUB"] -- AP_TOP_25 expanded in
# place, and UGA (also ranked) kept once, at its first position
"""
import logging
@@ -37,6 +38,11 @@ class DynamicTeamResolver:
_rankings_cache: Dict[str, List[str]] = {}
_cache_timestamp: float = 0
_cache_duration: int = 3600 # 1 hour cache
# A failed or empty fetch is remembered briefly too: during an ESPN
# outage every resolve would otherwise wait out request_timeout (30s)
# again, on each scoreboard's update.
_failure_timestamp: float = 0
_failure_backoff: int = 300 # 5 minutes
# Supported dynamic team patterns
DYNAMIC_PATTERNS = {
@@ -70,8 +76,8 @@ class DynamicTeamResolver:
if team in self.DYNAMIC_PATTERNS:
# Resolve dynamic team
dynamic_teams = self._resolve_dynamic_team(team, sport)
# _resolve_dynamic_team already logs the result.
resolved_teams.extend(dynamic_teams)
self.logger.info(f"Resolved {team} to {len(dynamic_teams)} teams: {dynamic_teams[:5]}{'...' if len(dynamic_teams) > 5 else ''}")
elif self._is_potential_dynamic_team(team):
# Unknown dynamic team, skip it
self.logger.warning(f"Unknown dynamic team '{team}' - skipping")
@@ -138,7 +144,11 @@ class DynamicTeamResolver:
if (self._rankings_cache and
current_time - self._cache_timestamp < self._cache_duration):
return self._rankings_cache
# A recent attempt failed: don't pay the request timeout again yet.
if current_time - self._failure_timestamp < self._failure_backoff:
return {}
try:
self.logger.info("Fetching fresh NCAA Football rankings from ESPN API")
rankings_url = "https://site.api.espn.com/apis/site/v2/sports/football/college-football/rankings"
@@ -185,7 +195,9 @@ class DynamicTeamResolver:
except Exception as e:
self.logger.error(f"Error fetching NCAA Football rankings: {e}")
# On the class, for the same reason as the rankings cache above.
DynamicTeamResolver._failure_timestamp = current_time
return {}
def get_available_dynamic_teams(self) -> List[str]:
@@ -229,6 +241,7 @@ class DynamicTeamResolver:
shadow the shared cache for this instance."""
DynamicTeamResolver._rankings_cache = {}
DynamicTeamResolver._cache_timestamp = 0
DynamicTeamResolver._failure_timestamp = 0
self.logger.info("Cleared dynamic team rankings cache")
src/element_style.py
+4 -3
View File
@@ -1499,11 +1499,12 @@ class ElementStyleResolver:
mode_config, 'align')
# scale is geometry, so it lives with the offsets rather than in the
# element block -- a logo has a scale and no font.
layout_defaults = self._defaults.get('layout', {})
# The default is looked up through the aliases too, like the value:
# an exact-key lookup missed a default filed under another name, so
# a configured value equal to it counted as a user choice.
scale = self._forced(
self._layout_element(self._customization(), element_key),
layout_defaults.get(element_key, {})
if isinstance(layout_defaults, dict) else {},
_lookup_element(self._defaults.get('layout', {}), element_key),
self._layout_element(self._mode_block(mode), element_key),
'scale')
src/error_aggregator.py
+3 -40
View File
@@ -6,7 +6,8 @@ for the LEDMatrix system. Enables automatic bug detection by tracking
error frequency, patterns, and context.
This is a local-only implementation with no external dependencies.
Errors are stored in memory with optional JSON export.
Errors are stored in memory; ErrorSnapshotPublisher shares a summary with the
web process through the cache.
"""
import math
@@ -18,7 +19,6 @@ import uuid
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Any, Callable, Tuple
import logging
@@ -105,7 +105,6 @@ class ErrorAggregator:
max_records: int = 1000,
pattern_threshold: int = 5,
pattern_window_minutes: int = 60,
export_path: Optional[Path] = None
):
"""
Initialize the error aggregator.
@@ -114,20 +113,18 @@ class ErrorAggregator:
max_records: Maximum number of error records to keep in memory
pattern_threshold: Number of occurrences to detect a pattern
pattern_window_minutes: Time window for pattern detection
export_path: Optional path for JSON export (auto-export on pattern detection)
"""
self.logger = logging.getLogger(__name__)
self.max_records = max_records
self.pattern_threshold = pattern_threshold
self.pattern_window = timedelta(minutes=pattern_window_minutes)
self.export_path = export_path
self._records: List[ErrorRecord] = []
self._error_counts: Dict[str, int] = defaultdict(int)
self._plugin_error_counts: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
self._patterns: Dict[str, ErrorPattern] = {}
self._pattern_callbacks: List[Callable[[ErrorPattern], None]] = []
self._lock = threading.RLock() # RLock allows nested acquisition for export_to_file
self._lock = threading.RLock() # RLock: build_snapshot and pattern callbacks re-enter
# Track session start for relative timing
self._session_start = datetime.now()
@@ -248,10 +245,6 @@ class ErrorAggregator:
callback(pattern)
except Exception as e:
self.logger.error(f"Pattern callback failed: {e}")
# Auto-export if path configured
if self.export_path:
self._auto_export()
else:
# Update existing pattern
self._patterns[pattern_key].count = count
@@ -424,33 +417,6 @@ class ErrorAggregator:
# turned into a string here rather than failing the write.
return json.loads(json.dumps(summary, default=str))
def export_to_file(self, filepath: Path) -> None:
"""
Export error data to JSON file.
Args:
filepath: Path to export file
"""
with self._lock:
data = {
"exported_at": datetime.now().isoformat(),
"summary": self.get_error_summary(),
"all_records": [r.to_dict() for r in self._records]
}
filepath.parent.mkdir(parents=True, exist_ok=True)
filepath.write_text(json.dumps(data, indent=2))
self.logger.info(f"Exported error data to {filepath}")
def _auto_export(self) -> None:
"""Auto-export on pattern detection (if export_path configured)."""
if self.export_path:
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filepath = self.export_path / f"errors_{timestamp}.json"
self.export_to_file(filepath)
except Exception as e:
self.logger.error(f"Auto-export failed: {e}")
# Global singleton instance
_error_aggregator: Optional[ErrorAggregator] = None
@@ -461,7 +427,6 @@ def get_error_aggregator(
max_records: int = 1000,
pattern_threshold: int = 5,
pattern_window_minutes: int = 60,
export_path: Optional[Path] = None
) -> ErrorAggregator:
"""
Get or create the global error aggregator instance.
@@ -470,7 +435,6 @@ def get_error_aggregator(
max_records: Maximum records to keep (only used on first call)
pattern_threshold: Pattern detection threshold (only used on first call)
pattern_window_minutes: Pattern detection window (only used on first call)
export_path: Export path for auto-export (only used on first call)
Returns:
The global ErrorAggregator instance
@@ -483,7 +447,6 @@ def get_error_aggregator(
max_records=max_records,
pattern_threshold=pattern_threshold,
pattern_window_minutes=pattern_window_minutes,
export_path=export_path
)
return _error_aggregator
src/exceptions.py
+12 -4
View File
@@ -42,7 +42,9 @@ class CacheError(LEDMatrixError):
context: Optional context dictionary
"""
if cache_key:
context = context or {}
# Copy so the caller's dict isn't mutated (a reused context
# dict would otherwise collect every error's keys).
context = dict(context or {})
context['cache_key'] = cache_key
super().__init__(message, context)
self.cache_key = cache_key
@@ -62,7 +64,9 @@ class ConfigError(LEDMatrixError):
context: Optional context dictionary
"""
if config_path or field:
context = context or {}
# Copy so the caller's dict isn't mutated (a reused context
# dict would otherwise collect every error's keys).
context = dict(context or {})
if config_path:
context['config_path'] = config_path
if field:
@@ -85,7 +89,9 @@ class PluginError(LEDMatrixError):
context: Optional context dictionary
"""
if plugin_id:
context = context or {}
# Copy so the caller's dict isn't mutated (a reused context
# dict would otherwise collect every error's keys).
context = dict(context or {})
context['plugin_id'] = plugin_id
super().__init__(message, context)
self.plugin_id = plugin_id
@@ -104,7 +110,9 @@ class DisplayError(LEDMatrixError):
context: Optional context dictionary
"""
if display_mode:
context = context or {}
# Copy so the caller's dict isn't mutated (a reused context
# dict would otherwise collect every error's keys).
context = dict(context or {})
context['display_mode'] = display_mode
super().__init__(message, context)
self.display_mode = display_mode
src/font_manager.py
+57 -16
View File
@@ -28,10 +28,10 @@ for accurate width/height calculations.
import os
import logging
import freetype
import requests
import json
import hashlib
import urllib.parse
import urllib.request
import zipfile
import tempfile
import time
@@ -50,6 +50,9 @@ from src.deprecation import deprecated
logger = logging.getLogger(__name__)
# Seconds before a stalled font download gives up (connect and per-read).
_FONT_DOWNLOAD_TIMEOUT = 30
class FontManager:
"""
Comprehensive font management supporting TTF and BDF fonts with caching,
@@ -320,31 +323,59 @@ class FontManager:
extension = self._get_font_extension(url)
cache_filename = f"{family}_{url_hash}{extension}"
cache_path = self.temp_font_dir / cache_filename
is_zip = url.endswith('.zip')
extract_dir = self.temp_font_dir / f"{family}_{url_hash}"
# Check if already downloaded
if cache_path.exists():
# Check if already downloaded. For a zip the font is the file
# extracted from it, so look there first -- returning the cached
# .zip itself would register the archive as the font after a
# restart.
if is_zip:
extracted = self._find_extracted_font(extract_dir)
if extracted:
logger.info(f"Using cached font: {extracted}")
return extracted
elif cache_path.exists():
logger.info(f"Using cached font: {cache_path}")
return str(cache_path)
# Download font — restrict to http/https to prevent file:// reads
parsed = urllib.parse.urlparse(url)
if parsed.scheme not in ('http', 'https'):
raise ValueError(f"Font URL must use http or https, got: {parsed.scheme!r}")
logger.info(f"Downloading font from {url}")
urllib.request.urlretrieve(url, cache_path) # nosec B310 - scheme validated above
if not cache_path.exists():
# Download font — restrict to http/https to prevent file:// reads
parsed = urllib.parse.urlparse(url)
if parsed.scheme not in ('http', 'https'):
raise ValueError(f"Font URL must use http or https, got: {parsed.scheme!r}")
logger.info(f"Downloading font from {url}")
# Download to a temp file and rename into place, with a
# timeout: writing straight to cache_path left a truncated
# file after a stalled/interrupted download, and the exists()
# check above then served it forever.
fd, tmp_name = tempfile.mkstemp(dir=self.temp_font_dir, suffix='.part')
try:
with os.fdopen(fd, 'wb') as tmp_file:
response = requests.get(url, timeout=_FONT_DOWNLOAD_TIMEOUT, stream=True)
response.raise_for_status()
for chunk in response.iter_content(chunk_size=65536):
if chunk:
tmp_file.write(chunk)
os.replace(tmp_name, cache_path)
except BaseException:
try:
os.unlink(tmp_name)
except OSError:
pass
raise
# Handle zip files
if url.endswith('.zip'):
extract_dir = self.temp_font_dir / f"{family}_{url_hash}"
if is_zip:
extract_dir.mkdir(exist_ok=True)
with zipfile.ZipFile(cache_path, 'r') as zip_ref:
zip_ref.extractall(extract_dir)
# Find the actual font file
for file in extract_dir.iterdir():
if file.suffix.lower() in ['.ttf', '.otf', '.bdf']:
return str(file)
extracted = self._find_extracted_font(extract_dir)
if extracted:
return extracted
return str(cache_path)
@@ -352,6 +383,16 @@ class FontManager:
logger.error(f"Error downloading font from {url}: {e}")
return None
@staticmethod
def _find_extracted_font(extract_dir: Path) -> Optional[str]:
"""Return the first font file in a zip's extract dir, if any."""
if not extract_dir.is_dir():
return None
for file in extract_dir.iterdir():
if file.suffix.lower() in ['.ttf', '.otf', '.bdf']:
return str(file)
return None
def _get_font_extension(self, url: str) -> str:
"""Extract font file extension from URL."""
if '.ttf' in url.lower():
src/logging_config.py
+4 -1
View File
@@ -43,7 +43,10 @@ class StructuredFormatter(logging.Formatter):
if hasattr(record, 'operation_id'):
log_data['operation_id'] = record.operation_id
return json.dumps(log_data)
# default=str: record.context / extras can hold datetimes, Paths,
# exceptions etc.; without it one such value raised TypeError and
# the whole record was dropped by the handler's error path.
return json.dumps(log_data, default=str)
class ContextualFormatter(logging.Formatter):
src/startup_validator.py
+14 -5
View File
@@ -318,12 +318,21 @@ class StartupValidator:
def raise_on_errors(self) -> None:
"""
Raise exceptions if validation errors exist.
Raise one exception if validation errors exist; return None if not.
Nothing in core calls this (see the module docstring). Errors are
grouped by a keyword in their message, not by which check produced
them, and only the first non-empty group is raised, in the order
config > cache > plugin: a "plugin ... config" message counts as a
config error, and cache/plugin errors are not reported while a
config error exists. The raised exception's ``context['errors']``
holds that group's messages only.
Raises:
ConfigError: If configuration validation fails
CacheError: If cache validation fails
PluginError: If plugin validation fails
ConfigError: If any message mentions config/configuration, or if
none matches any group
CacheError: If a message mentions cache (and none config)
PluginError: If a message mentions plugin (and none of the above)
"""
if not self.errors:
return
src/web_interface/config_arrays.py
+22 -7
View File
@@ -2,13 +2,29 @@
A list reaches a plugin-config save keyed by position more often than as a
list. The settings form posts one field per element (``feeds.custom_feeds.0.name``),
which ``_set_nested_value`` stores as ``{"0": {"name": ...}}``, and the JSON
path's dotToNested() in the browser builds the same dict. Validation expects
an array there, so the save converts them first.
which ``_set_nested_value`` stores as ``{"0": {"name": ...}}``, and a JSON
save built by flattening then re-nesting dotted keys carries the same dict.
Validation expects an array there, so the save converts them first.
"""
from typing import Any, Dict
def _schema_type_is(prop: Any, wanted: str) -> bool:
"""Whether a schema property is of ``wanted`` type, unions included.
Mirrors ``_schema_type_is`` in ``web_interface/blueprints/api_v3`` (kept
here so src/ doesn't import the Flask blueprint). A union such as
``["array", "null"]`` -- the per-element style overrides, where null means
"inherit" -- is still an array for recombining position-keyed inputs.
"""
if not isinstance(prop, dict):
return False
declared = prop.get('type')
if isinstance(declared, list):
return wanted in declared
return declared == wanted
def _is_index_dict(value: Any) -> bool:
"""True for a dict keyed only by list positions ("0", "1", ...), or empty."""
return isinstance(value, dict) and all(str(k).isdigit() for k in value)
@@ -35,10 +51,9 @@ def coerce_array_shapes(config: Dict[str, Any], schema_props: Dict[str, Any],
for key, prop_schema in schema_props.items():
if key not in config or not isinstance(prop_schema, dict):
continue
prop_type = prop_schema.get('type')
value = config[key]
if prop_type == 'array':
if _schema_type_is(prop_schema, 'array'):
if _is_index_dict(value):
value = config[key] = [value[k] for k in sorted(value, key=lambda k: int(str(k)))]
if not isinstance(value, list):
@@ -50,11 +65,11 @@ def coerce_array_shapes(config: Dict[str, Any], schema_props: Dict[str, Any],
and isinstance(default, list) and len(default) >= min_items):
value = config[key] = list(default)
items_schema = prop_schema.get('items')
if (isinstance(items_schema, dict) and items_schema.get('type') == 'object'
if (_schema_type_is(items_schema, 'object')
and 'properties' in items_schema):
for element in value:
coerce_array_shapes(element, items_schema['properties'],
short_lists_take_default)
elif prop_type == 'object' and 'properties' in prop_schema:
elif _schema_type_is(prop_schema, 'object') and 'properties' in prop_schema:
coerce_array_shapes(value, prop_schema['properties'], short_lists_take_default)
src/web_interface/validators.py
+4 -2
View File
@@ -8,11 +8,13 @@ from pathlib import Path
def validate_file_upload(filename: str, max_size_mb: int = 10,
allowed_extensions: Optional[List[str]] = None) -> Tuple[bool, Optional[str]]:
"""
Validate file upload parameters.
Validate an upload's filename (not its contents or size).
Args:
filename: Name of the file
max_size_mb: Maximum file size in MB
max_size_mb: Unused. Kept so existing callers keep working; this
function only sees the filename, so callers must enforce the
size limit themselves on the uploaded stream.
allowed_extensions: List of allowed file extensions (e.g., ['.ttf', '.otf'])
Returns:
test/test_api_helper.py
+28 -6
View File
@@ -50,29 +50,51 @@ def helper(cache):
class TestRateLimiting:
def test_sleeps_for_remaining_interval(self, helper, monkeypatch):
fake_time = MagicMock()
fake_time.time.side_effect = [102.0, 105.0]
fake_time.monotonic.side_effect = [102.0, 105.0]
fake_time.time.return_value = 5000.0
monkeypatch.setattr(api_helper_module, 'time', fake_time)
helper.set_rate_limit(5)
helper._last_request_time = 100.0
helper._last_request_monotonic = 100.0
helper._enforce_rate_limit()
# 2s elapsed of a 5s interval -> sleep the remaining 3s.
fake_time.sleep.assert_called_once()
assert fake_time.sleep.call_args[0][0] == pytest.approx(3.0)
assert helper._last_request_time == 105.0
assert helper._last_request_monotonic == 105.0
assert helper._last_request_time == 5000.0
def test_no_sleep_when_interval_elapsed(self, helper, monkeypatch):
fake_time = MagicMock()
fake_time.time.side_effect = [200.0, 201.0]
fake_time.monotonic.side_effect = [200.0, 201.0]
monkeypatch.setattr(api_helper_module, 'time', fake_time)
helper.set_rate_limit(5)
helper._last_request_time = 100.0
helper._last_request_monotonic = 100.0
helper._enforce_rate_limit()
fake_time.sleep.assert_not_called()
assert helper._last_request_time == 201.0
assert helper._last_request_monotonic == 201.0
def test_wall_clock_step_back_does_not_stall_requests(self, helper, monkeypatch):
# NTP stepping the wall clock back an hour between two requests must
# not turn into an hour-long "remaining interval" sleep.
wall = {"now": 10_000.0}
mono = {"now": 50.0}
sleeps = []
fake_time = MagicMock()
fake_time.time.side_effect = lambda: wall["now"]
fake_time.monotonic.side_effect = lambda: mono["now"]
fake_time.sleep.side_effect = sleeps.append
monkeypatch.setattr(api_helper_module, 'time', fake_time)
helper.set_rate_limit(5)
helper._enforce_rate_limit()
wall["now"] -= 3600
mono["now"] += 10
helper._enforce_rate_limit()
assert all(s <= 5 for s in sleeps)
# ---------------------------------------------------------------------------
test/test_auto_update_setup.py
+19
View File
@@ -75,6 +75,25 @@ def test_does_nothing_while_updates_are_off(tmp_path):
assert not (root / aus.RESULT_REL).exists()
@pytest.mark.parametrize('value', [True, 'yes', ['enabled']])
def test_a_non_object_auto_update_value_reads_as_off(value):
"""A hand-edited "auto_update": true raised AttributeError and aborted
setup; the web UI's save treats a non-object as {} (off), so this does too."""
assert aus.is_enabled({'auto_update': value}) is False
def test_a_failed_result_write_leaves_no_temp_file(tmp_path, monkeypatch):
root, etc = project(tmp_path)
def broken_dump(*args, **kwargs):
raise OSError('disk full')
monkeypatch.setattr(aus.json, 'dump', broken_dump)
setup(root, etc, FakeSystemctl())._report('failed', 'x')
leftovers = [p.name for p in (root / aus.RESULT_REL).parent.iterdir()
if p.name.startswith('.auto_update_setup_')]
assert leftovers == []
def test_installs_both_units_for_the_web_user(tmp_path):
root, etc = project(tmp_path)
systemctl = FakeSystemctl()
test/test_background_data_service.py
+27
View File
@@ -93,6 +93,33 @@ class TestCacheHit:
stats = service.get_statistics()
assert stats["cached_hits"] == 1
def test_cache_hit_callback_runs_outside_the_service_lock(
self, service, mock_cache_manager):
"""The worker path calls plugin callbacks after releasing the lock;
the cache-hit path held it, so a slow callback stalled every other
thread's submit and result bookkeeping."""
import threading
mock_cache_manager.get.return_value = {"events": []}
seen = {}
def callback(result):
# The result is already filed, as before.
seen["filed"] = service.get_result(result.request_id) is result
other = {}
def try_lock():
other["got"] = service._lock.acquire(timeout=1)
if other["got"]:
service._lock.release()
t = threading.Thread(target=try_lock)
t.start()
t.join()
seen["lock_free"] = other["got"]
service.submit_fetch_request(sport="nba", year=2024, url="https://x.com",
cache_key="k", callback=callback)
assert seen == {"filed": True, "lock_free": True}
# ---------------------------------------------------------------------------
# Actual fetch path (mocked HTTP)
test/test_dynamic_team_resolver.py
+33 -6
View File
@@ -48,9 +48,11 @@ def reset_class_cache():
"""Reset the CLASS-level shared cache between tests."""
DynamicTeamResolver._rankings_cache = {}
DynamicTeamResolver._cache_timestamp = 0
DynamicTeamResolver._failure_timestamp = 0
yield
DynamicTeamResolver._rankings_cache = {}
DynamicTeamResolver._cache_timestamp = 0
DynamicTeamResolver._failure_timestamp = 0
@pytest.fixture
@@ -158,12 +160,20 @@ class TestRankingsParsing:
assert list(rankings.keys()) == ['A1', 'B2', 'C3']
assert list(rankings.values()) == [1, 2, 3]
def test_empty_rankings_returns_empty_and_caches_nothing(
self, resolver, mock_get):
def test_empty_rankings_returns_empty_and_backs_off(
self, resolver, mock_get, monkeypatch):
mock_get.return_value = _make_response({'rankings': []})
assert resolver._fetch_ncaa_fb_rankings() == {}
# Nothing was cached, so the next call hits HTTP again.
# No rankings were cached, but the failure is: the next call inside
# the back-off window doesn't hit HTTP again.
assert resolver._fetch_ncaa_fb_rankings() == {}
assert mock_get.call_count == 1
assert DynamicTeamResolver._rankings_cache == {}
stamp = DynamicTeamResolver._failure_timestamp
monkeypatch.setattr(
dtr_module, 'time', types.SimpleNamespace(time=lambda: stamp + 301))
assert resolver._fetch_ncaa_fb_rankings() == {}
assert mock_get.call_count == 2
@@ -226,7 +236,7 @@ class TestSharedCache:
class TestFailureHandling:
def test_network_failure_drops_dynamic_keeps_static_caches_nothing(
self, resolver, mock_get):
self, resolver, mock_get, monkeypatch):
mock_get.side_effect = [
requests.exceptions.RequestException('boom'),
_make_response(_rankings_payload()),
@@ -237,12 +247,29 @@ class TestFailureHandling:
# Dynamic name silently dropped, static name kept, nothing raises.
assert result == ['UGA']
# Nothing was cached on failure: a subsequent call refetches and
# succeeds.
# Inside the back-off window the outage isn't retried: each resolve
# would otherwise wait out the full request timeout again.
assert resolver.resolve_teams(['UGA', 'AP_TOP_5']) == ['UGA']
assert mock_get.call_count == 1
# No rankings were cached on failure: once the window passes, the
# next call refetches and succeeds.
stamp = DynamicTeamResolver._failure_timestamp
monkeypatch.setattr(
dtr_module, 'time', types.SimpleNamespace(time=lambda: stamp + 301))
result = resolver.resolve_teams(['UGA', 'AP_TOP_5'])
assert result == ['UGA', 'MICH', 'OSU', 'TEX', 'ALA']
assert mock_get.call_count == 2
def test_clear_cache_forgets_a_failure(self, resolver, mock_get):
mock_get.side_effect = [
requests.exceptions.RequestException('boom'),
_make_response(_rankings_payload()),
]
assert resolver.resolve_teams(['AP_TOP_5']) == []
resolver.clear_cache()
assert resolver.resolve_teams(['AP_TOP_5']) == TOP_TEAMS[:5]
# ---------------------------------------------------------------------------
# sport argument
test/test_element_style.py
+8
View File
@@ -1115,6 +1115,14 @@ class TestAliasedLookup:
).style("score_text")
assert st.scale == 2.0
def test_scale_default_is_found_through_an_alias(self):
"""The saved value equals the default filed under the bare-noun
layout key, so it is the save flow's write-in, not a choice."""
defaults = {"customization": {"layout": {"score": {"scale": 1.5}}}}
config = {"customization": {"layout": {"score": {"scale": 1.5}}}}
st = ElementStyleResolver(config, defaults).style("score_text")
assert st.scale == 1.0
HANDWRITTEN = {
"type": "object",
"properties": {
test/test_error_handling.py
+15
View File
@@ -31,3 +31,18 @@ class TestCustomExceptions:
# DisplayError includes context in string representation
assert "Display not found" in str(error)
assert error.context.get('display_mode') == 'adafruit'
def test_callers_context_dict_is_not_mutated(self):
"""A caller reusing one context dict across raises used to have
every error's own key written into it."""
cases = [
(CacheError, {'cache_key': 'k'}),
(ConfigError, {'config_path': 'c.json', 'field': 'f'}),
(PluginError, {'plugin_id': 'weather'}),
(DisplayError, {'display_mode': 'm'}),
]
for cls, kwargs in cases:
shared = {'attempt': 1}
error = cls("boom", context=shared, **kwargs)
assert shared == {'attempt': 1}, cls.__name__
assert error.context == {'attempt': 1, **kwargs}, cls.__name__
test/test_font_manager.py
+65
View File
@@ -8,13 +8,18 @@ test here asserts observable behavior: returned font types, cache identity,
fallback selection, and BDF native-size reading.
"""
import hashlib
import io
import json
import shutil
import zipfile
from unittest.mock import MagicMock
import freetype
import pytest
from PIL import ImageFont
import src.font_manager as fm_module
from src.common.font_layout import resolve_asset_path
from src.font_manager import FontManager
@@ -188,3 +193,63 @@ class TestPluginFonts:
assert fm.register_plugin_fonts("my-plugin", self.MANIFEST)
assert fm.font_catalog["my-plugin::bundled"] == str(plugin_dir / "fonts" / "Bundled.ttf")
class TestDownloadFont:
"""_download_font: plugin fonts declared by URL, cached in temp_font_dir."""
URL = "https://fonts.example/pack.zip"
@staticmethod
def _zip_bytes():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("MyFont.ttf", b"not really a font")
return buf.getvalue()
@staticmethod
def _response(chunks):
response = MagicMock()
response.raise_for_status.return_value = None
response.iter_content.side_effect = lambda chunk_size: iter(chunks)
return response
def test_a_zip_is_served_as_its_extracted_font_after_a_restart(self, fm, tmp_path):
# The state a previous run leaves: the .zip and its extracted font.
# The cache check used to find the .zip first and register the
# archive itself as the font.
fm.temp_font_dir = tmp_path
url_hash = hashlib.sha256(self.URL.encode()).hexdigest()[:16]
zip_path = tmp_path / f"pack_{url_hash}.zip"
zip_path.write_bytes(self._zip_bytes())
extract_dir = tmp_path / f"pack_{url_hash}"
with zipfile.ZipFile(zip_path) as zf:
zf.extractall(extract_dir)
path = fm._download_font(self.URL, {"family": "pack"})
assert path == str(extract_dir / "MyFont.ttf")
def test_download_has_a_timeout_and_lands_atomically(self, fm, tmp_path, monkeypatch):
fm.temp_font_dir = tmp_path
get = MagicMock(return_value=self._response([self._zip_bytes()]))
monkeypatch.setattr(fm_module.requests, "get", get)
path = fm._download_font(self.URL, {"family": "pack"})
assert path is not None and path.endswith("MyFont.ttf")
assert get.call_args.kwargs.get("timeout")
assert not list(tmp_path.glob("*.part"))
def test_an_interrupted_download_leaves_nothing_to_be_served(self, fm, tmp_path, monkeypatch):
fm.temp_font_dir = tmp_path
def chunks():
yield b"partial"
raise OSError("connection reset")
response = self._response([])
response.iter_content.side_effect = lambda chunk_size: chunks()
monkeypatch.setattr(fm_module.requests, "get", MagicMock(return_value=response))
assert fm._download_font("https://fonts.example/Font.ttf", {"family": "f"}) is None
assert list(tmp_path.iterdir()) == []
test/test_logging_config.py
+11
View File
@@ -57,6 +57,17 @@ class TestStructuredFormatter:
assert out["plugin_id"] == "clock"
assert out["operation_id"] == "op-1"
def test_non_json_context_values_are_stringified(self):
# A datetime/Path in the context used to raise TypeError from
# json.dumps, and the handler dropped the whole record.
from datetime import datetime
from pathlib import Path
record = make_record(context={"at": datetime(2026, 1, 2, 3, 4, 5),
"path": Path("a/b")})
out = json.loads(StructuredFormatter().format(record))
assert out["context"]["at"] == "2026-01-02 03:04:05"
assert out["context"]["path"] == str(Path("a/b"))
def test_exception_key_when_exc_info_present(self):
try:
raise ValueError("kaboom")
test/test_logo_helper.py
+8
View File
@@ -235,6 +235,14 @@ class TestLoadLogoWithDownload:
"PHI", tmp_path / "missing.png", None, max_width=16, max_height=16)
assert logo is not None and logo.size == (16, 16)
def test_placeholder_uses_the_scaled_box(self, helper, tmp_path):
# A real logo at scale 2 is fitted into 32x32; the stand-in for a
# missing one must be the same size, not the unscaled 16x16.
logo = helper.load_logo_with_download(
"PHI", tmp_path / "missing.png", None, max_width=16, max_height=16,
scale=2.0)
assert logo is not None and logo.size == (32, 32)
class TestDownloadLogo:
def test_writes_file_and_sets_permissions(self, helper, tmp_path):
test/test_permission_utils.py
+37
View File
@@ -19,6 +19,7 @@ from src.common.permission_utils import (
_redact_url_credentials,
ensure_shared_group_ownership,
install_requirements_file,
sudo_remove_directory,
)
@@ -46,6 +47,42 @@ class TestRedactUrlCredentials:
assert _redact_url_credentials(text) == text
class TestSudoRemoveDirectory:
"""sudoers matches the exact argv, so the bash path the rule names has
to be found by trying each candidate, as install_requirements_file does."""
def _target(self, tmp_path):
target = tmp_path / "some-plugin"
target.mkdir()
return target
@patch('src.common.permission_utils.subprocess.run')
def test_tries_the_next_bash_path_when_sudo_refuses(self, mock_run, tmp_path):
target = self._target(tmp_path)
def fake_run(argv, **kwargs):
if mock_run.call_count == 1:
return MagicMock(returncode=1, stdout="",
stderr="sudo: a password is required")
target.rmdir()
return MagicMock(returncode=0, stdout="", stderr="")
mock_run.side_effect = fake_run
assert sudo_remove_directory(target, allowed_bases=[tmp_path]) is True
assert mock_run.call_count == 2
first, second = (c.args[0][2] for c in mock_run.call_args_list)
assert first != second
@patch('src.common.permission_utils.subprocess.run')
def test_stops_when_the_helper_itself_fails(self, mock_run, tmp_path):
target = self._target(tmp_path)
mock_run.return_value = MagicMock(returncode=1, stdout="",
stderr="refusing: not a plugin dir")
assert sudo_remove_directory(target, allowed_bases=[tmp_path]) is False
assert mock_run.call_count == 1
class TestInstallRequirementsFileRedaction:
@patch('src.common.permission_utils.subprocess.run')
def test_wrapper_path_redacts_stderr_and_stdout(self, mock_run, tmp_path):
test/test_sync_manager.py
+38 -3
View File
@@ -120,7 +120,7 @@ def raise_n_then_stop(mgr, exc, count):
return _side_effect
def fake_clock(monkeypatch, *, time_fn=None, sleep_fn=None):
def fake_clock(monkeypatch, *, time_fn=None, sleep_fn=None, monotonic_fn=None):
"""Swap sync_manager's own `time` reference for a private stand-in.
sync_manager.time IS the stdlib module, so patching attributes on it
@@ -129,9 +129,14 @@ def fake_clock(monkeypatch, *, time_fn=None, sleep_fn=None):
hard-to-trace source of cross-test flakiness. Rebinding the module's
reference keeps the patch scoped to the code under test. Anything not
overridden falls through to the real functions.
``time_fn`` drives both clocks unless ``monotonic_fn`` is given: the
timers read time.monotonic(), and a test that only needs "a frozen
clock" shouldn't care which one.
"""
monkeypatch.setattr(sync_manager, "time", SimpleNamespace(
time=time_fn or time.time,
monotonic=monotonic_fn or time_fn or time.monotonic,
sleep=sleep_fn or time.sleep,
))
@@ -310,6 +315,34 @@ class TestWatchdogs:
assert mgr._leader_state is LeaderState.CONNECTED
assert mgr._peer_ip == "10.0.0.1"
def test_wall_clock_jump_does_not_time_out_the_peer(self, monkeypatch):
# A Pi has no RTC: NTP can step the wall clock by hours after the
# peer connected. Only elapsed (monotonic) time counts toward the
# heartbeat timeout.
mgr = make_manager(role=SyncRole.LEADER)
mgr._leader_state = LeaderState.CONNECTED
mgr._peer_ip = "10.0.0.1"
mgr._last_heartbeat_time = 100.0
fake_clock(monkeypatch,
time_fn=lambda: 100.0 + 3600,
monotonic_fn=lambda: 101.0,
sleep_fn=lambda _: setattr(mgr, "_running", False))
mgr._running = True
mgr._leader_watchdog()
assert mgr._leader_state is LeaderState.CONNECTED
def test_wall_clock_jump_does_not_drop_the_leader(self, monkeypatch):
mgr = make_manager(role=SyncRole.FOLLOWER)
mgr._follower_state = FollowerState.FOLLOWER
mgr._last_leader_frame_time = 100.0
fake_clock(monkeypatch,
time_fn=lambda: 100.0 + 3600,
monotonic_fn=lambda: 101.0,
sleep_fn=lambda _: setattr(mgr, "_running", False))
mgr._running = True
mgr._follower_watchdog()
assert mgr._follower_state is FollowerState.FOLLOWER
def test_leader_watchdog_ignores_disconnected_state(self, monkeypatch):
mgr = make_manager(role=SyncRole.LEADER)
mgr._leader_state = LeaderState.INCOMPATIBLE
@@ -496,7 +529,8 @@ class TestFollowerRecvLoop:
mgr = make_manager(role=SyncRole.FOLLOWER)
sleeps = MagicMock()
with patch.object(sync_manager, "time",
SimpleNamespace(time=time.time, sleep=sleeps)):
SimpleNamespace(time=time.time, monotonic=time.monotonic,
sleep=sleeps)):
self._drive(mgr, b"12345")
assert mgr.get_latest_frame() is None
sleeps.assert_not_called()
@@ -507,7 +541,8 @@ class TestFollowerRecvLoop:
mgr = make_manager(role=SyncRole.FOLLOWER)
sleeps = MagicMock()
with patch.object(sync_manager, "time",
SimpleNamespace(time=time.time, sleep=sleeps)):
SimpleNamespace(time=time.time, monotonic=time.monotonic,
sleep=sleeps)):
self._drive(mgr, json.dumps(payload).encode())
assert mgr.get_latest_scroll_x() is None
sleeps.assert_not_called()
test/web_interface/test_config_arrays.py
+16
View File
@@ -30,6 +30,22 @@ def test_element_types_are_left_to_normalization():
assert config["color"] == ["1", "2", "3"]
def test_a_nullable_array_union_is_still_an_array():
"""["array", "null"] is how the per-mode style overrides are typed (null
means inherit); their indexed colour inputs must still recombine."""
config = {"text_color": {"0": 1, "1": 2, "2": 3}}
coerce_array_shapes(config, {"text_color": {"type": ["array", "null"]}})
assert config["text_color"] == [1, 2, 3]
def test_a_nullable_object_union_is_walked_into():
config = {"live": {"tags": {"0": "a"}}}
coerce_array_shapes(config, {"live": {
"type": ["object", "null"],
"properties": {"tags": {"type": "array"}}}})
assert config["live"]["tags"] == ["a"]
def test_nested_objects_and_array_items_are_walked():
schema = {"feeds": {"type": "object", "properties": {
"custom_feeds": {"type": "array", "items": {"type": "object", "properties": {