mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-01 16:58:06 +00:00
feat(sports): promote the nine universal methods into the base classes
Phase B1b of docs/SPORTS_UNIFICATION.md. Every method here is present in
all nine bundled plugin sports.py copies and absent from core, so this is
reuse of code the fleet already agreed on — not new behavior. The
promotions are inert until B5: the plugins' own overrides still run.
SportsCore: cleanup, _get_layout_offset, _load_custom_font_from_element_config
SportsUpcoming: _select_games_for_display
SportsRecent: _get_zero_clock_duration, _clear_zero_clock_tracking,
_select_recent_games_for_display
SportsLive: _is_game_really_over, _detect_stale_games
Where the copies disagreed, the canonical form was chosen on evidence and
the genuine per-sport differences became seams rather than branches:
- _favorite_key(game, side) -- NRL matches favorites on team id because its
abbreviations are ambiguous (NEW is both Newcastle Knights and New
Zealand Warriors). Default is the abbreviation; NRL overrides. Core never
learns the string nrl.
- FINAL_PERIOD / CLOCK_COUNTS_DOWN -- hockey ends in P3, and soccer/afl/nrl
clocks count UP, so 0:00 means kickoff, not expiry.
- _config_schema_path() / _font_root() -- plugin-supplied locations, never
derived from this module's __file__.
BEHAVIOR CHANGE (baseball, ufc): the rejected variant coerced a missing or
non-str clock to the literal 0:00 and then declared the game over at
period >= 4. MLB has no game clock and period is the inning, so live games
were being evicted from the 5th inning onward; UFC likewise. The promoted
variant skips the clock check when the clock is unusable -- it fails safe
(keeps showing the game) instead of failing destructive.
Also fixes a regression from the package move in e591cec: the bodies were
byte-identical but __file__ gained a directory, so _resolve_project_path's
parents[2] silently began resolving to <root>/src instead of the repo root.
Both it and _font_root now derive from a single _INSTALL_ROOT constant, so
a future move needs one line changed rather than two hand-counted depths.
Tests assert the resolved values, not the index.
The font loader takes baseball's body (BDF memo cache + native-strike
retry) under hockey's Optional signature -- the older lineage is the
correct one here, and basketball's positional str default breaks on an
explicit None. It resolves through _font_root rather than the cwd, so it
does not reintroduce the bug just fixed for FontManager, and delegates to
FontManager for the alias table and BDF header parse instead of shipping
second copies. cleanup gained the two new font caches and still leaves
background_service alone -- it is a process-wide singleton.
Verified: 111 new tests (48 core + 59 modes + 4 install-root regression);
characterization + skin suites still exactly 94, unchanged.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FgbA8SMutQQpXkMG8LMmC4
This commit is contained in:
Claude
@@ -95,6 +95,29 @@ deprecation cycle.
|
||||
| `_custom_scorebug_layout(game, draw)` | Per-sport overlay on the base layout | no-op |
|
||||
| `render_skin_card(game, size)` | Skin-system entry point | built-in fallback |
|
||||
| `score_phrase(game)` | Celebration wording (`"GOAL"` vs `"TOUCHDOWN"`) | `"SCORE"` — only consulted when `CelebrationMixin` is present |
|
||||
| `_favorite_key(game, side)` | Which view-model field identifies a team for favorites matching | `game["<side>_abbr"]` |
|
||||
| `_config_schema_path()` | Plugin's `config_schema.json` — returning it routes `_get_layout_offset` through the `src.element_style` resolver (and gives it the defaults to compare against) | `None`, i.e. the classic inline `customization.layout` read |
|
||||
| `_font_root()` | Directory to resolve `assets/fonts` against | core install root |
|
||||
|
||||
Two class attributes serve the same purpose for values that are per-sport
|
||||
constants rather than behavior:
|
||||
|
||||
| Attribute | Meaning | Default |
|
||||
|---|---|---|
|
||||
| `FINAL_PERIOD` | Period at/after which a zero clock can mean "over" | `4` (hockey overrides to `3`) |
|
||||
| `CLOCK_COUNTS_DOWN` | Whether `0:00` means "expired" | `True` (soccer/afl/nrl override to `False` — their clocks count up, so `0:00` is kickoff) |
|
||||
|
||||
### Why these are seams and not branches
|
||||
|
||||
`_favorite_key` exists because NRL abbreviations are **not unique** — "NEW" is both
|
||||
Newcastle Knights and New Zealand Warriors, "CAN" both Canberra and Canterbury —
|
||||
so NRL matches favorites on team ID. Flattening every plugin to abbreviations
|
||||
would silently select the wrong club for NRL users. The base declares the seam,
|
||||
NRL fills it, and core never learns the string `"nrl"`.
|
||||
|
||||
`CLOCK_COUNTS_DOWN` exists for the same reason in the opposite direction: a
|
||||
soccer clock reading `0:00` means the match has not kicked off, so running the
|
||||
clock-expiry branch there would evict live games.
|
||||
|
||||
## Phases
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ import time
|
||||
from abc import ABC, abstractmethod
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import pytz
|
||||
import requests
|
||||
@@ -32,6 +32,77 @@ try:
|
||||
except ImportError:
|
||||
OddsManager = None
|
||||
|
||||
# The core install root, resolved from this module's own location so nothing
|
||||
# here depends on the process working directory.
|
||||
#
|
||||
# The index is the number of directories between this file and the root:
|
||||
# src/base_classes/sports/core.py -> sports -> base_classes -> src -> root.
|
||||
# It is defined ONCE, here, because hand-counting it at each use site is
|
||||
# exactly what broke when this module moved from src/base_classes/sports.py
|
||||
# into the package (the old depth of 2 silently started resolving to src/).
|
||||
# If this module ever moves again, this is the single line to update.
|
||||
_INSTALL_ROOT = Path(__file__).resolve().parents[3]
|
||||
|
||||
# Shared element-style resolver. Core always ships src.element_style, so the
|
||||
# guard is never taken here — it is kept because this module's methods are
|
||||
# back-copied verbatim into the plugins' bundled `sports.py`, where older
|
||||
# cores genuinely lack the module (the plugins fall back to the classic
|
||||
# inline config read below).
|
||||
try:
|
||||
from src.element_style import ElementStyleResolver, defaults_from_schema_file
|
||||
STYLE_AVAILABLE = True
|
||||
except ImportError: # pragma: no cover - core always ships element_style
|
||||
STYLE_AVAILABLE = False
|
||||
|
||||
|
||||
# --- Font catalog delegation -------------------------------------------------
|
||||
# The plugin copies each carried their own alias table mapping font *family*
|
||||
# names ("press_start") to filenames. That table is a duplicate of the core
|
||||
# FontManager's `common_fonts` catalog, so resolve through the catalog instead
|
||||
# and let the two never drift apart. Same lazy module-level shape as
|
||||
# skin_runtime._get_font_manager / base_plugin._fallback_font_manager: a
|
||||
# SportsCore host has no plugin_manager to borrow a FontManager from.
|
||||
|
||||
_shared_font_catalog: Optional[Any] = None
|
||||
|
||||
|
||||
def _font_catalog() -> Any:
|
||||
"""Shared read-only FontManager used purely as a font-name catalog."""
|
||||
global _shared_font_catalog
|
||||
if _shared_font_catalog is None:
|
||||
from src.font_manager import FontManager
|
||||
_shared_font_catalog = FontManager({})
|
||||
return _shared_font_catalog
|
||||
|
||||
|
||||
def _resolve_font_family_alias(font_name: str) -> str:
|
||||
"""Resolve a font family alias to its filename, leaving filenames as-is.
|
||||
|
||||
A config `font` value may be either a family name from the FontManager
|
||||
catalog ("press_start", "four_by_six", "five_by_seven") or a literal
|
||||
filename; the former resolve here, the latter pass through unchanged.
|
||||
"""
|
||||
try:
|
||||
catalogued = _font_catalog().common_fonts.get(font_name)
|
||||
except Exception:
|
||||
return font_name
|
||||
return os.path.basename(catalogued) if catalogued else font_name
|
||||
|
||||
|
||||
def _read_bdf_native_size(bdf_path: str) -> Optional[int]:
|
||||
"""A BDF file's one true pixel size, delegated to FontManager.
|
||||
|
||||
Deliberately NOT reimplemented here: the plugin copies grew a variant
|
||||
that scans the whole file and returns a partially collected value when
|
||||
parsing raises, where FontManager's stops at the first STARTCHAR and
|
||||
returns None on error.
|
||||
"""
|
||||
try:
|
||||
from src.font_manager import FontManager
|
||||
return FontManager._read_bdf_native_size(bdf_path)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
class SportsCore(ABC):
|
||||
# Which ScoreboardSkin render method this class's display path maps to.
|
||||
@@ -97,6 +168,13 @@ class SportsCore(ABC):
|
||||
|
||||
self._logo_cache = {}
|
||||
|
||||
# Font caches for _load_custom_font_from_element_config: per-frame
|
||||
# callers (font-ladder walks) resolve the same (name, size) over and
|
||||
# over, and the BDF strike size means re-reading a file header.
|
||||
# Both are released in cleanup().
|
||||
self._font_cache: Dict[Tuple[str, int], Any] = {}
|
||||
self._bdf_native_size_cache: Dict[str, Optional[int]] = {}
|
||||
|
||||
# Set up headers
|
||||
self.headers = {
|
||||
'User-Agent': 'LEDMatrix/1.0 (https://github.com/yourusername/LEDMatrix; contact@example.com)',
|
||||
@@ -172,8 +250,7 @@ class SportsCore(ABC):
|
||||
"""Convert relative paths to absolute ones rooted at the project directory."""
|
||||
if path.is_absolute():
|
||||
return path
|
||||
project_root = Path(__file__).resolve().parents[2]
|
||||
return (project_root / path).resolve()
|
||||
return (_INSTALL_ROOT / path).resolve()
|
||||
|
||||
def _get_logo_directory_fallbacks(self, configured_dir: Path) -> List[Path]:
|
||||
"""Return fallback directories to try when the configured directory is not writable."""
|
||||
@@ -753,3 +830,261 @@ class SportsCore(ABC):
|
||||
|
||||
def _custom_scorebug_layout(self, game: dict, draw_overlay: ImageDraw.ImageDraw):
|
||||
pass
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Promoted from the plugin copies (see docs/SPORTS_UNIFICATION.md).
|
||||
# Everything below is present in all nine bundled `sports.py` copies;
|
||||
# the canonical form of each is documented on the method.
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _favorite_key(self, game: Dict, side: str) -> Optional[str]:
|
||||
"""Override point: which view-model field identifies a team when
|
||||
matching against ``favorite_teams``.
|
||||
|
||||
``side`` is ``"home"`` or ``"away"``. The default is the team
|
||||
abbreviation — what eight of the nine scoreboards match on, and what
|
||||
users type into their favorites list.
|
||||
|
||||
NRL overrides this to the team **id**, because NRL abbreviations are
|
||||
not unique: "NEW" is both Newcastle Knights and New Zealand Warriors,
|
||||
"CAN" both Canberra Raiders and Canterbury Bulldogs. Matching those by
|
||||
abbreviation selects the wrong club. It is a seam rather than a branch
|
||||
precisely so core never has to learn the string "nrl"::
|
||||
|
||||
def _favorite_key(self, game, side):
|
||||
return str(game.get(f"{side}_id"))
|
||||
|
||||
An override that stringifies should note that a missing id becomes the
|
||||
literal ``"None"``, which would spuriously match a favorites list
|
||||
containing that string. The default returns ``None``, which never
|
||||
matches.
|
||||
"""
|
||||
return game.get(f"{side}_abbr")
|
||||
|
||||
def _config_schema_path(self) -> Optional[str]:
|
||||
"""Override point: the plugin's ``config_schema.json``, used as the
|
||||
reference for style-resolver defaults.
|
||||
|
||||
None (the default) means no schema is available — layout offsets are
|
||||
then read with the classic inline config lookup, which is exactly what
|
||||
a plugin that never shipped resolver support does today. A plugin
|
||||
opting into :mod:`src.element_style` returns its own schema path::
|
||||
|
||||
def _config_schema_path(self):
|
||||
return os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||||
'config_schema.json')
|
||||
|
||||
It must not be derived from this module's ``__file__``: after
|
||||
promotion that resolves inside ``src/base_classes/sports/``, where no
|
||||
``config_schema.json`` exists.
|
||||
"""
|
||||
return None
|
||||
|
||||
def _font_root(self) -> str:
|
||||
"""Override point: the directory ``assets/fonts`` resolves against.
|
||||
|
||||
Defaults to the core install root, derived from this module's own
|
||||
location — the same strategy as ``FontManager._resolve_asset_path``.
|
||||
A plugin bundling its own fonts overrides this to return its plugin
|
||||
directory. Never a cwd-relative path: fonts must load no matter where
|
||||
the process was started from (e.g. the plugin safety harness on CI).
|
||||
"""
|
||||
return str(_INSTALL_ROOT)
|
||||
|
||||
def _resolve_font_path(self, font_name: str) -> str:
|
||||
"""Locate a font file by filename, independently of the process cwd.
|
||||
|
||||
Tries ``assets/fonts/<name>`` relative to the cwd first (preserving
|
||||
behavior for a process started from an install root), then under
|
||||
:meth:`_font_root`. Returns the cwd-relative path unchanged when the
|
||||
file is nowhere to be found, so callers log the familiar path.
|
||||
"""
|
||||
relative = os.path.join('assets', 'fonts', font_name)
|
||||
if os.path.exists(relative):
|
||||
return relative
|
||||
candidate = os.path.join(self._font_root(), relative)
|
||||
if os.path.exists(candidate):
|
||||
return candidate
|
||||
return relative
|
||||
|
||||
def _get_layout_offset(self, element: str, axis: str, default: int = 0) -> int:
|
||||
"""
|
||||
Get layout offset for a specific element and axis.
|
||||
|
||||
Args:
|
||||
element: Element name (e.g., 'home_logo', 'score', 'status_text')
|
||||
axis: 'x_offset' or 'y_offset' (or 'away_x_offset', 'home_x_offset' for records)
|
||||
default: Default value if not configured (default: 0)
|
||||
|
||||
Returns:
|
||||
Offset value from config or default (always returns int)
|
||||
"""
|
||||
schema_path = self._config_schema_path() if STYLE_AVAILABLE else None
|
||||
if schema_path:
|
||||
# Shared resolver (rebuilt if the config dict was swapped out,
|
||||
# matching the classic path's read-config-on-every-call semantics).
|
||||
# Note it is stricter than the classic read below: a boolean offset
|
||||
# degrades to the default instead of counting as 1/0, which is the
|
||||
# more correct reading of a pixel offset.
|
||||
resolver = getattr(self, '_style_resolver_cached', None)
|
||||
if resolver is None or resolver._config is not self.config:
|
||||
resolver = ElementStyleResolver(
|
||||
self.config, defaults_from_schema_file(schema_path))
|
||||
self._style_resolver_cached = resolver
|
||||
return resolver.offset_value(element, axis, default)
|
||||
try:
|
||||
layout_config = self.config.get('customization', {}).get('layout', {})
|
||||
element_config = layout_config.get(element, {})
|
||||
offset_value = element_config.get(axis, default)
|
||||
|
||||
# Ensure we return an integer (handle float/string from config)
|
||||
if isinstance(offset_value, (int, float)):
|
||||
return int(offset_value)
|
||||
elif isinstance(offset_value, str):
|
||||
# Try to convert string to int
|
||||
try:
|
||||
return int(float(offset_value))
|
||||
except (ValueError, TypeError):
|
||||
self.logger.warning(
|
||||
f"Invalid layout offset value for {element}.{axis}: '{offset_value}', using default {default}"
|
||||
)
|
||||
return default
|
||||
else:
|
||||
return default
|
||||
except Exception as e:
|
||||
# Gracefully handle any config access errors
|
||||
self.logger.debug(f"Error reading layout offset for {element}.{axis}: {e}, using default {default}")
|
||||
return default
|
||||
|
||||
def _load_custom_font_from_element_config(
|
||||
self,
|
||||
element_config: Dict[str, Any],
|
||||
default_size: int = 8,
|
||||
default_font: Optional[str] = None,
|
||||
) -> ImageFont.FreeTypeFont:
|
||||
"""
|
||||
Load a custom font from an element configuration dictionary.
|
||||
|
||||
Args:
|
||||
element_config: Configuration dict for a single element containing 'font' and 'font_size' keys
|
||||
default_size: Default font size if not specified in config
|
||||
default_font: Default font filename when not specified in config (e.g. '4x6-font.ttf' for odds)
|
||||
|
||||
Returns:
|
||||
PIL ImageFont object
|
||||
"""
|
||||
base_default = default_font or "PressStart2P-Regular.ttf"
|
||||
font_name = element_config.get('font', base_default)
|
||||
font_size = int(element_config.get('font_size', default_size)) # Ensure integer for PIL
|
||||
|
||||
# Resolve family aliases (e.g. "press_start") to real filenames, then
|
||||
# locate the file against _font_root() rather than the cwd.
|
||||
resolved_name = _resolve_font_family_alias(font_name)
|
||||
font_path = self._resolve_font_path(resolved_name)
|
||||
|
||||
# Memoized: per-frame callers (font-ladder walks) resolve the same
|
||||
# (name, size) repeatedly -- return the previously loaded face.
|
||||
cache_key = (resolved_name, font_size)
|
||||
cached_font = self._font_cache.get(cache_key)
|
||||
if cached_font is not None:
|
||||
return cached_font
|
||||
|
||||
# Try to load the font
|
||||
try:
|
||||
if os.path.exists(font_path):
|
||||
# Try loading as TTF first (works for both TTF and some BDF files with PIL)
|
||||
if font_path.lower().endswith('.ttf'):
|
||||
font = ImageFont.truetype(font_path, font_size)
|
||||
self.logger.debug(f"Loaded font: {font_name} at size {font_size}")
|
||||
self._font_cache[cache_key] = font
|
||||
return font
|
||||
elif font_path.lower().endswith('.bdf'):
|
||||
# BDF fonts are fixed-size bitmaps, not scalable outlines --
|
||||
# FreeType only accepts the exact pixel size baked into the
|
||||
# file (its "strike") and raises "invalid pixel size" for
|
||||
# anything else. Try the requested size first (in case it
|
||||
# happens to match), then fall back to the file's real
|
||||
# native size, so a BDF font can still be selected via
|
||||
# font_size-driven configs without the caller needing to
|
||||
# know its exact strike size.
|
||||
#
|
||||
# This retry is the OLDER lineage's behavior and it is the
|
||||
# correct one: the newer copies call truetype() on a BDF at
|
||||
# any size (which simply fails) or refuse BDF outright.
|
||||
try:
|
||||
font = ImageFont.truetype(font_path, font_size)
|
||||
self.logger.debug(f"Loaded BDF font: {font_name} at size {font_size}")
|
||||
self._font_cache[cache_key] = font
|
||||
return font
|
||||
except OSError:
|
||||
if font_path in self._bdf_native_size_cache:
|
||||
native_size = self._bdf_native_size_cache[font_path]
|
||||
else:
|
||||
native_size = _read_bdf_native_size(font_path)
|
||||
self._bdf_native_size_cache[font_path] = native_size
|
||||
if native_size and native_size != font_size:
|
||||
try:
|
||||
font = ImageFont.truetype(font_path, native_size)
|
||||
self.logger.debug(
|
||||
f"Loaded BDF font: {font_name} at its native size {native_size} "
|
||||
f"(requested {font_size} isn't a valid strike for this file)"
|
||||
)
|
||||
self._font_cache[cache_key] = font
|
||||
return font
|
||||
except Exception as retry_exc:
|
||||
self.logger.debug(
|
||||
f"BDF font {font_name} also failed to load at native "
|
||||
f"size {native_size}: {retry_exc}"
|
||||
)
|
||||
self.logger.warning(f"Could not load BDF font {font_name} with PIL, using default")
|
||||
# Fall through to default
|
||||
else:
|
||||
self.logger.warning(f"Unknown font file type: {font_name}, using default")
|
||||
else:
|
||||
self.logger.warning(f"Font file not found: {font_path}, using default")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error loading font {font_name}: {e}, using default")
|
||||
|
||||
# Fall back to default font. Cached under the requested (name, size)
|
||||
# key too, so a misconfigured or missing font pays the disk cost once
|
||||
# instead of on every frame of a font-ladder walk.
|
||||
default_font_path = self._resolve_font_path(
|
||||
_resolve_font_family_alias(base_default))
|
||||
try:
|
||||
if os.path.exists(default_font_path):
|
||||
font = ImageFont.truetype(default_font_path, font_size)
|
||||
else:
|
||||
self.logger.warning("Default font not found, using PIL default")
|
||||
font = ImageFont.load_default()
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error loading default font: {e}")
|
||||
font = ImageFont.load_default()
|
||||
self._font_cache[cache_key] = font
|
||||
return font
|
||||
|
||||
def cleanup(self):
|
||||
"""Clean up resources when plugin is unloaded."""
|
||||
# Close HTTP session
|
||||
if hasattr(self, 'session') and self.session:
|
||||
try:
|
||||
self.session.close()
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Error closing session: {e}")
|
||||
|
||||
# Clear caches
|
||||
if hasattr(self, '_logo_cache'):
|
||||
self._logo_cache.clear()
|
||||
# Font caches hold PIL faces; without this they are an unbounded
|
||||
# per-instance leak across enable/disable cycles.
|
||||
if hasattr(self, '_font_cache'):
|
||||
self._font_cache.clear()
|
||||
if hasattr(self, '_bdf_native_size_cache'):
|
||||
self._bdf_native_size_cache.clear()
|
||||
|
||||
# NOTE: self.background_service is deliberately NOT shut down here.
|
||||
# get_background_service() returns a PROCESS-WIDE singleton shared by
|
||||
# every scoreboard; shutting it down from one unloading plugin would
|
||||
# stop background fetching for all the others. Whoever owns the
|
||||
# process owns its lifecycle. Do not "fix" this.
|
||||
|
||||
self.logger.info(f"{self.__class__.__name__} cleanup completed")
|
||||
|
||||
@@ -7,7 +7,7 @@ import logging
|
||||
import time
|
||||
from abc import abstractmethod
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
@@ -34,6 +34,71 @@ class SportsUpcoming(SportsCore):
|
||||
self.last_game_switch = 0
|
||||
self.game_display_duration = 15 # Display each upcoming game for 15 seconds
|
||||
|
||||
def _select_games_for_display(
|
||||
self, processed_games: List[Dict], favorite_teams: List[str]
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Single-pass game selection with proper deduplication and counting.
|
||||
|
||||
When a game involves two favorite teams, it counts toward BOTH teams' limits.
|
||||
This prevents unexpected game counts from the multi-pass algorithm.
|
||||
|
||||
Team identity goes through the ``_favorite_key`` override point rather
|
||||
than reading ``home_abbr``/``away_abbr`` directly, because abbreviations
|
||||
are not unique in every league (NRL matches on team ID instead).
|
||||
"""
|
||||
sorted_games = sorted(
|
||||
processed_games,
|
||||
key=lambda g: g.get("start_time_utc")
|
||||
or datetime.max.replace(tzinfo=timezone.utc),
|
||||
)
|
||||
|
||||
if not favorite_teams:
|
||||
return sorted_games
|
||||
|
||||
selected_games = []
|
||||
selected_ids = set()
|
||||
team_counts = {team: 0 for team in favorite_teams}
|
||||
|
||||
for game in sorted_games:
|
||||
game_id = game.get("id")
|
||||
if game_id in selected_ids:
|
||||
continue
|
||||
|
||||
home = self._favorite_key(game, "home")
|
||||
away = self._favorite_key(game, "away")
|
||||
|
||||
home_fav = home in favorite_teams
|
||||
away_fav = away in favorite_teams
|
||||
|
||||
if not home_fav and not away_fav:
|
||||
continue
|
||||
|
||||
home_needs = home_fav and team_counts[home] < self.upcoming_games_to_show
|
||||
away_needs = away_fav and team_counts[away] < self.upcoming_games_to_show
|
||||
|
||||
if home_needs or away_needs:
|
||||
selected_games.append(game)
|
||||
selected_ids.add(game_id)
|
||||
if home_fav:
|
||||
team_counts[home] += 1
|
||||
if away_fav:
|
||||
team_counts[away] += 1
|
||||
|
||||
self.logger.debug(
|
||||
f"Selected game {away}@{home}: team_counts={team_counts}"
|
||||
)
|
||||
|
||||
if all(c >= self.upcoming_games_to_show for c in team_counts.values()):
|
||||
self.logger.debug("All favorite teams satisfied, stopping selection")
|
||||
break
|
||||
|
||||
self.logger.info(
|
||||
f"Selected {len(selected_games)} games for {len(favorite_teams)} "
|
||||
f"favorite teams: {team_counts}"
|
||||
)
|
||||
return selected_games
|
||||
|
||||
def update(self):
|
||||
"""Update upcoming games data."""
|
||||
if not self.is_enabled: return
|
||||
@@ -369,6 +434,96 @@ class SportsRecent(SportsCore):
|
||||
self.update_interval = self.mode_config.get("recent_update_interval", 3600) # Check for recent games every hour
|
||||
self.last_game_switch = 0
|
||||
self.game_display_duration = 15 # Display each recent game for 15 seconds
|
||||
# Tracks when each game was first seen with an expired clock, keyed by
|
||||
# game id. Promoted alongside the zero-clock helpers below; without it
|
||||
# the first _get_zero_clock_duration() call raises AttributeError.
|
||||
self._zero_clock_timestamps: Dict[str, float] = {} # Track games at 0:00
|
||||
|
||||
# -- Zero-clock tracking ------------------------------------------------
|
||||
# Byte-identical in all nine plugin copies. Note that afl/nrl/soccer define
|
||||
# these but never call them — their clocks count up, so 0:00 means kickoff
|
||||
# rather than expiry (see CLOCK_COUNTS_DOWN on SportsLive). That makes the
|
||||
# pair a future `CountdownClockMixin` candidate so it stops appearing in the
|
||||
# MRO of sports that cannot use it — B2 work, not now.
|
||||
|
||||
def _get_zero_clock_duration(self, game_id: str) -> float:
|
||||
"""Track how long a game has been at 0:00 clock."""
|
||||
current_time = time.time()
|
||||
if game_id not in self._zero_clock_timestamps:
|
||||
self._zero_clock_timestamps[game_id] = current_time
|
||||
return 0.0
|
||||
return current_time - self._zero_clock_timestamps[game_id]
|
||||
|
||||
def _clear_zero_clock_tracking(self, game_id: str) -> None:
|
||||
"""Clear tracking when game clock moves away from 0:00 or game ends."""
|
||||
if game_id in self._zero_clock_timestamps:
|
||||
del self._zero_clock_timestamps[game_id]
|
||||
|
||||
def _select_recent_games_for_display(
|
||||
self, processed_games: List[Dict], favorite_teams: List[str]
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Single-pass game selection for recent games with proper deduplication.
|
||||
|
||||
When a game involves two favorite teams, it counts toward BOTH teams' limits.
|
||||
Games are sorted by most recent first.
|
||||
|
||||
Team identity goes through the ``_favorite_key`` override point rather
|
||||
than reading ``home_abbr``/``away_abbr`` directly, because abbreviations
|
||||
are not unique in every league (NRL matches on team ID instead).
|
||||
"""
|
||||
sorted_games = sorted(
|
||||
processed_games,
|
||||
key=lambda g: g.get("start_time_utc")
|
||||
or datetime.min.replace(tzinfo=timezone.utc),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
if not favorite_teams:
|
||||
return sorted_games
|
||||
|
||||
selected_games = []
|
||||
selected_ids = set()
|
||||
team_counts = {team: 0 for team in favorite_teams}
|
||||
|
||||
for game in sorted_games:
|
||||
game_id = game.get("id")
|
||||
if game_id in selected_ids:
|
||||
continue
|
||||
|
||||
home = self._favorite_key(game, "home")
|
||||
away = self._favorite_key(game, "away")
|
||||
|
||||
home_fav = home in favorite_teams
|
||||
away_fav = away in favorite_teams
|
||||
|
||||
if not home_fav and not away_fav:
|
||||
continue
|
||||
|
||||
home_needs = home_fav and team_counts[home] < self.recent_games_to_show
|
||||
away_needs = away_fav and team_counts[away] < self.recent_games_to_show
|
||||
|
||||
if home_needs or away_needs:
|
||||
selected_games.append(game)
|
||||
selected_ids.add(game_id)
|
||||
if home_fav:
|
||||
team_counts[home] += 1
|
||||
if away_fav:
|
||||
team_counts[away] += 1
|
||||
|
||||
self.logger.debug(
|
||||
f"Selected recent game {away}@{home}: team_counts={team_counts}"
|
||||
)
|
||||
|
||||
if all(c >= self.recent_games_to_show for c in team_counts.values()):
|
||||
self.logger.debug("All favorite teams satisfied, stopping selection")
|
||||
break
|
||||
|
||||
self.logger.info(
|
||||
f"Selected {len(selected_games)} recent games for {len(favorite_teams)} "
|
||||
f"favorite teams: {team_counts}"
|
||||
)
|
||||
return selected_games
|
||||
|
||||
def update(self):
|
||||
"""Update recent games data."""
|
||||
@@ -659,6 +814,17 @@ class SportsRecent(SportsCore):
|
||||
return False
|
||||
|
||||
class SportsLive(SportsCore):
|
||||
# Per-sport constants for the "is this live game actually over?" check.
|
||||
# These are values, not behavior, so they are class attributes rather than
|
||||
# override points (see docs/SPORTS_UNIFICATION.md "Override points").
|
||||
#
|
||||
# FINAL_PERIOD: the period at/after which an expired clock can mean "over".
|
||||
# 4 for four-quarter sports; hockey overrides to 3.
|
||||
# CLOCK_COUNTS_DOWN: whether "0:00" means the clock expired. False for
|
||||
# sports whose clock counts up (soccer/afl/nrl), where 0:00 is kickoff —
|
||||
# running the expiry branch there would evict games that just started.
|
||||
FINAL_PERIOD = 4
|
||||
CLOCK_COUNTS_DOWN = True
|
||||
|
||||
def __init__(self, config: Dict[str, Any], display_manager: DisplayManager, cache_manager: CacheManager, logger: logging.Logger, sport_key: str):
|
||||
super().__init__(config, display_manager, cache_manager, logger, sport_key)
|
||||
@@ -676,11 +842,107 @@ class SportsLive(SportsCore):
|
||||
self.count_log_interval = 5 # Only log count data every 5 seconds
|
||||
# Initialize test_mode - defaults to False (live mode)
|
||||
self.test_mode = self.mode_config.get("test_mode", False)
|
||||
# Freshness bookkeeping for _detect_stale_games():
|
||||
# {game_id: {"clock": ts, "score": ts, "last_seen": ts}}
|
||||
self.game_update_timestamps = {}
|
||||
self.stale_game_timeout = self.mode_config.get("stale_game_timeout", 300) # 5 minutes default
|
||||
|
||||
@abstractmethod
|
||||
def _test_mode_update(self) -> None:
|
||||
return
|
||||
|
||||
def _is_game_really_over(self, game: Dict) -> bool:
|
||||
"""Check if a game appears to be over even if API says it's live.
|
||||
|
||||
Two independent signals:
|
||||
1. ``period_text`` says "final" — universal across every sport.
|
||||
2. The clock has expired at/after :attr:`FINAL_PERIOD` — only meaningful
|
||||
where :attr:`CLOCK_COUNTS_DOWN` is true.
|
||||
|
||||
Fails *safe*: anything ambiguous returns False and the game keeps being
|
||||
displayed. The only caller, :meth:`_detect_stale_games`, removes games
|
||||
on a True, so a false positive silently drops a live game.
|
||||
"""
|
||||
game_str = f"{game.get('away_abbr')}@{game.get('home_abbr')}"
|
||||
|
||||
# `period_text` may be present-but-None; `or ""` keeps that from raising
|
||||
# AttributeError — the caller has no try/except around this call.
|
||||
period_text = (game.get("period_text") or "").lower()
|
||||
if "final" in period_text:
|
||||
self.logger.debug(
|
||||
f"_is_game_really_over({game_str}): "
|
||||
f"returning True - 'final' in period_text='{period_text}'"
|
||||
)
|
||||
return True
|
||||
|
||||
if not self.CLOCK_COUNTS_DOWN:
|
||||
# Count-up clock: 0:00 means the match has not started.
|
||||
self.logger.debug(
|
||||
f"_is_game_really_over({game_str}): returning False "
|
||||
f"(count-up clock, period_text='{period_text}')"
|
||||
)
|
||||
return False
|
||||
|
||||
raw_clock = game.get("clock")
|
||||
period = game.get("period", 0)
|
||||
|
||||
# Only check clock-based finish if we have a valid clock string. A
|
||||
# missing or non-string clock is NOT coerced to "0:00": sports without a
|
||||
# game clock (e.g. baseball, where `period` is the inning) would
|
||||
# otherwise be declared over from the FINAL_PERIOD-th period onward.
|
||||
if isinstance(raw_clock, str) and raw_clock.strip() and period >= self.FINAL_PERIOD:
|
||||
clock = raw_clock
|
||||
clock_normalized = clock.replace(":", "").strip()
|
||||
if clock_normalized in ("000", "00") or clock in ("0:00", ":00"):
|
||||
self.logger.debug(
|
||||
f"_is_game_really_over({game_str}): "
|
||||
f"returning True - clock at 0:00 (clock='{clock}', period={period})"
|
||||
)
|
||||
return True
|
||||
|
||||
self.logger.debug(
|
||||
f"_is_game_really_over({game_str}): returning False"
|
||||
)
|
||||
return False
|
||||
|
||||
def _detect_stale_games(self, games: List[Dict]) -> None:
|
||||
"""Remove games that appear stale or haven't updated.
|
||||
|
||||
Mutates ``games`` **in place** and returns None. Removal is by value
|
||||
(``list.remove`` uses ``dict.__eq__``), so two structurally-equal game
|
||||
dicts in the same list would drop the first occurrence.
|
||||
"""
|
||||
current_time = time.time()
|
||||
|
||||
for game in games[:]: # Copy list to iterate safely
|
||||
game_id = game.get("id")
|
||||
if not game_id:
|
||||
continue
|
||||
|
||||
# Check if game data is stale
|
||||
timestamps = self.game_update_timestamps.get(game_id, {})
|
||||
last_seen = timestamps.get("last_seen", 0)
|
||||
|
||||
if last_seen > 0 and current_time - last_seen > self.stale_game_timeout:
|
||||
self.logger.warning(
|
||||
f"Removing stale game {game.get('away_abbr')}@{game.get('home_abbr')} "
|
||||
f"(last seen {int(current_time - last_seen)}s ago)"
|
||||
)
|
||||
games.remove(game)
|
||||
if game_id in self.game_update_timestamps:
|
||||
del self.game_update_timestamps[game_id]
|
||||
continue
|
||||
|
||||
# Also check if game appears to be over
|
||||
if self._is_game_really_over(game):
|
||||
self.logger.debug(
|
||||
f"Removing game that appears over: {game.get('away_abbr')}@{game.get('home_abbr')} "
|
||||
f"(clock={game.get('clock')}, period={game.get('period')}, period_text={game.get('period_text')})"
|
||||
)
|
||||
games.remove(game)
|
||||
if game_id in self.game_update_timestamps:
|
||||
del self.game_update_timestamps[game_id]
|
||||
|
||||
def update(self):
|
||||
"""Update live game data and handle game switching."""
|
||||
if not self.is_enabled:
|
||||
|
||||
@@ -0,0 +1,558 @@
|
||||
"""Tests for the methods promoted onto SportsCore from the nine bundled
|
||||
plugin copies of ``sports.py`` (phase B1 of docs/SPORTS_UNIFICATION.md).
|
||||
|
||||
Three methods and their seams land here:
|
||||
|
||||
- ``cleanup()`` — byte-identical in all nine copies. The tests pin the
|
||||
ordering (session close, then caches, then the completion log) and the
|
||||
deliberate *omission*: the process-wide background service must never be
|
||||
shut down by one unloading plugin.
|
||||
- ``_get_layout_offset()`` — football's resolver-backed variant, with the
|
||||
classic inline config read as the fallback used by every plugin that
|
||||
doesn't hand core a ``_config_schema_path()``.
|
||||
- ``_load_custom_font_from_element_config()`` — baseball's body (the only
|
||||
copy that handles BDF strikes correctly) under hockey's wider signature,
|
||||
resolving font files through the ``_font_root()`` seam instead of the
|
||||
process cwd.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from PIL import Image, ImageFont
|
||||
|
||||
# src.base_classes.sports transitively imports the hardware matrix driver;
|
||||
# stub it so these tests can import the sports base classes off-device.
|
||||
sys.modules.setdefault("rgbmatrix", MagicMock())
|
||||
|
||||
from src.base_classes.sports import SportsCore
|
||||
|
||||
LOGGER = logging.getLogger("test_sports_core_promotions")
|
||||
|
||||
CORE_ROOT = Path(__file__).resolve().parents[1]
|
||||
FONTS_DIR = CORE_ROOT / "assets" / "fonts"
|
||||
TTF_NAME = "PressStart2P-Regular.ttf"
|
||||
BDF_NAME = "5x7.bdf" # a BDF whose only valid strike is 7px
|
||||
BDF_NATIVE_SIZE = 7
|
||||
|
||||
|
||||
class _StubSports(SportsCore):
|
||||
"""Minimal concrete SportsCore — the abstract methods are never called
|
||||
by anything under test here."""
|
||||
|
||||
def _fetch_data(self):
|
||||
return None
|
||||
|
||||
def _extract_game_details(self, game_event):
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def build(monkeypatch, tmp_path):
|
||||
"""Factory for real SportsCore instances: logo dir redirected to tmp and
|
||||
the process-wide background service replaced with a MagicMock so the
|
||||
tests can assert nothing ever calls it."""
|
||||
monkeypatch.setattr(
|
||||
SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path)
|
||||
monkeypatch.setattr(
|
||||
"src.base_classes.sports.core.get_background_service",
|
||||
lambda *args, **kwargs: MagicMock())
|
||||
|
||||
def _build(config=None, cls=_StubSports):
|
||||
display_manager = MagicMock()
|
||||
display_manager.matrix.width = 128
|
||||
display_manager.matrix.height = 32
|
||||
display_manager.width = 128
|
||||
display_manager.height = 32
|
||||
display_manager.image = Image.new("RGB", (128, 32))
|
||||
cache_manager = MagicMock()
|
||||
cache_manager.cache_dir = str(tmp_path)
|
||||
return cls(config if config is not None else {"timezone": "UTC"},
|
||||
display_manager, cache_manager, LOGGER, "nhl")
|
||||
|
||||
return _build
|
||||
|
||||
|
||||
def probe(config=None):
|
||||
"""Unbound-call stand-in for hosts we don't need a full instance for
|
||||
(same pattern as make_probe in test_sports_base_characterization)."""
|
||||
host = MagicMock()
|
||||
host.logger = LOGGER
|
||||
host.config = config if config is not None else {}
|
||||
host._font_cache = {}
|
||||
host._bdf_native_size_cache = {}
|
||||
host._config_schema_path.return_value = None
|
||||
host._font_root.side_effect = lambda: SportsCore._font_root(host)
|
||||
host._resolve_font_path.side_effect = (
|
||||
lambda name: SportsCore._resolve_font_path(host, name))
|
||||
return host
|
||||
|
||||
|
||||
def offset(host, element, axis, default=0):
|
||||
return SportsCore._get_layout_offset(host, element, axis, default)
|
||||
|
||||
|
||||
def load_font(host, *args, **kwargs):
|
||||
return SportsCore._load_custom_font_from_element_config(host, *args, **kwargs)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. cleanup()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCleanup:
|
||||
def test_closes_session_and_clears_all_caches(self, build):
|
||||
manager = build()
|
||||
session = MagicMock()
|
||||
manager.session = session
|
||||
manager._logo_cache["TB"] = object()
|
||||
manager._font_cache[("PressStart2P-Regular.ttf", 8)] = object()
|
||||
manager._bdf_native_size_cache["assets/fonts/5x7.bdf"] = 7
|
||||
|
||||
manager.cleanup()
|
||||
|
||||
session.close.assert_called_once_with()
|
||||
assert manager._logo_cache == {}
|
||||
# Promoted alongside the font loader: these hold PIL faces and are
|
||||
# an unbounded leak across enable/disable cycles if never released.
|
||||
assert manager._font_cache == {}
|
||||
assert manager._bdf_native_size_cache == {}
|
||||
|
||||
def test_second_cleanup_is_a_noop(self, build):
|
||||
manager = build()
|
||||
manager.session = MagicMock()
|
||||
manager._logo_cache["TB"] = object()
|
||||
manager._font_cache[("x", 8)] = object()
|
||||
|
||||
manager.cleanup()
|
||||
manager.cleanup() # must not raise on already-released state
|
||||
|
||||
assert manager._logo_cache == {}
|
||||
assert manager._font_cache == {}
|
||||
assert manager._bdf_native_size_cache == {}
|
||||
assert manager.session.close.call_count == 2
|
||||
|
||||
def test_does_not_shut_down_the_shared_background_service(self, build):
|
||||
# get_background_service() hands out a PROCESS-WIDE singleton shared
|
||||
# by every scoreboard. One plugin unloading must not stop background
|
||||
# fetching for the other eight — cleanup() touches it not at all.
|
||||
manager = build()
|
||||
service = manager.background_service
|
||||
manager.session = MagicMock()
|
||||
|
||||
manager.cleanup()
|
||||
|
||||
assert service.shutdown.called is False
|
||||
assert service.stop.called is False
|
||||
assert service.method_calls == [], (
|
||||
"cleanup() called into the shared background service: "
|
||||
f"{service.method_calls}")
|
||||
|
||||
def test_completion_is_logged_even_when_session_close_raises(self, build, caplog):
|
||||
manager = build()
|
||||
manager.session = MagicMock()
|
||||
manager.session.close.side_effect = RuntimeError("socket already gone")
|
||||
manager._logo_cache["TB"] = object()
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger=LOGGER.name):
|
||||
manager.cleanup()
|
||||
|
||||
messages = [r.message for r in caplog.records]
|
||||
assert any("Error closing session" in m for m in messages)
|
||||
# Ordering is load-bearing: the caches still get cleared and the
|
||||
# completion log still fires after a failed close.
|
||||
assert manager._logo_cache == {}
|
||||
assert any("cleanup completed" in m for m in messages)
|
||||
|
||||
def test_tolerates_missing_attributes(self):
|
||||
# The hasattr guards exist so a partially constructed instance (an
|
||||
# __init__ that raised) can still be cleaned up.
|
||||
host = MagicMock(spec=["logger"])
|
||||
host.logger = LOGGER
|
||||
SportsCore.cleanup(host)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. _get_layout_offset() + the _config_schema_path() seam
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def layout_config(element, axis, value):
|
||||
return {"customization": {"layout": {element: {axis: value}}}}
|
||||
|
||||
|
||||
class TestLayoutOffsetClassicPath:
|
||||
"""The default path: _config_schema_path() returns None, so offsets come
|
||||
from the inline customization.layout read every plugin ships today."""
|
||||
|
||||
def test_config_schema_path_defaults_to_none(self, build):
|
||||
manager = build()
|
||||
assert manager._config_schema_path() is None
|
||||
|
||||
def test_reads_configured_int(self):
|
||||
host = probe(layout_config("home_logo", "x_offset", 5))
|
||||
assert offset(host, "home_logo", "x_offset") == 5
|
||||
|
||||
def test_float_is_truncated_to_int(self):
|
||||
host = probe(layout_config("score", "y_offset", 2.9))
|
||||
result = offset(host, "score", "y_offset")
|
||||
assert result == 2 and isinstance(result, int)
|
||||
|
||||
def test_numeric_string_is_coerced(self):
|
||||
host = probe(layout_config("score", "x_offset", "-3.5"))
|
||||
assert offset(host, "score", "x_offset") == -3
|
||||
|
||||
def test_unconfigured_element_and_axis_use_default(self):
|
||||
host = probe(layout_config("score", "x_offset", 5))
|
||||
assert offset(host, "status_text", "x_offset", 7) == 7
|
||||
assert offset(host, "score", "y_offset", -1) == -1
|
||||
assert offset(probe(), "score", "x_offset", 4) == 4
|
||||
|
||||
def test_non_numeric_string_degrades_to_default(self):
|
||||
host = probe(layout_config("score", "x_offset", "left"))
|
||||
assert offset(host, "score", "x_offset", 3) == 3
|
||||
|
||||
def test_unsupported_type_degrades_to_default(self):
|
||||
host = probe(layout_config("score", "x_offset", {"nested": 1}))
|
||||
assert offset(host, "score", "x_offset", 2) == 2
|
||||
host = probe(layout_config("score", "x_offset", None))
|
||||
assert offset(host, "score", "x_offset", 2) == 2
|
||||
|
||||
def test_broken_config_object_degrades_to_default(self):
|
||||
host = probe()
|
||||
host.config = "not a dict"
|
||||
assert offset(host, "score", "x_offset", 6) == 6
|
||||
|
||||
def test_boolean_counts_as_one(self):
|
||||
# PINNED AS-IS: the classic read predates the resolver and treats a
|
||||
# bool as its int value (True -> 1). See the resolver test below for
|
||||
# the stricter, more correct handling.
|
||||
host = probe(layout_config("score", "x_offset", True))
|
||||
assert offset(host, "score", "x_offset", 4) == 1
|
||||
|
||||
|
||||
class TestLayoutOffsetResolverPath:
|
||||
"""When a plugin supplies its config_schema.json, offsets resolve through
|
||||
src.element_style instead."""
|
||||
|
||||
@pytest.fixture
|
||||
def schema_path(self, tmp_path):
|
||||
path = tmp_path / "config_schema.json"
|
||||
path.write_text(json.dumps({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"customization": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"layout": {"type": "object", "properties": {}},
|
||||
},
|
||||
},
|
||||
},
|
||||
}))
|
||||
return str(path)
|
||||
|
||||
def host(self, schema_path, config):
|
||||
host = probe(config)
|
||||
host._config_schema_path.return_value = schema_path
|
||||
del host._style_resolver_cached # MagicMock auto-attrs otherwise
|
||||
host._style_resolver_cached = None
|
||||
return host
|
||||
|
||||
def test_reads_configured_offsets(self, schema_path):
|
||||
host = self.host(schema_path, layout_config("home_logo", "x_offset", 5))
|
||||
assert offset(host, "home_logo", "x_offset") == 5
|
||||
|
||||
def test_numeric_string_is_coerced(self, schema_path):
|
||||
host = self.host(schema_path, layout_config("score", "x_offset", "-3.5"))
|
||||
assert offset(host, "score", "x_offset") == -3
|
||||
|
||||
def test_missing_value_uses_default(self, schema_path):
|
||||
host = self.host(schema_path, layout_config("score", "x_offset", 5))
|
||||
assert offset(host, "score", "y_offset", 9) == 9
|
||||
|
||||
def test_bad_input_degrades_to_default(self, schema_path):
|
||||
host = self.host(schema_path, layout_config("score", "x_offset", "left"))
|
||||
assert offset(host, "score", "x_offset", 3) == 3
|
||||
host = self.host(schema_path, {"customization": {"layout": "nope"}})
|
||||
assert offset(host, "score", "x_offset", 3) == 3
|
||||
|
||||
def test_boolean_is_rejected_unlike_the_classic_path(self, schema_path):
|
||||
# The intended behavior difference: a bool is not a pixel offset, so
|
||||
# the resolver returns the default where the classic read returns 1.
|
||||
host = self.host(schema_path, layout_config("score", "x_offset", True))
|
||||
assert offset(host, "score", "x_offset", 4) == 4
|
||||
|
||||
def test_resolver_is_cached_and_rebuilt_when_config_is_swapped(self, schema_path):
|
||||
host = self.host(schema_path, layout_config("score", "x_offset", 5))
|
||||
assert offset(host, "score", "x_offset") == 5
|
||||
first = host._style_resolver_cached
|
||||
assert offset(host, "score", "x_offset") == 5
|
||||
assert host._style_resolver_cached is first
|
||||
|
||||
# on_config_change swaps the dict object; the resolver must follow.
|
||||
host.config = layout_config("score", "x_offset", 11)
|
||||
assert offset(host, "score", "x_offset") == 11
|
||||
assert host._style_resolver_cached is not first
|
||||
|
||||
def test_missing_schema_file_still_resolves_offsets(self, tmp_path):
|
||||
# Offsets don't depend on schema defaults, so an unreadable schema
|
||||
# must not cost the plugin its layout customization.
|
||||
host = self.host(str(tmp_path / "absent.json"),
|
||||
layout_config("score", "x_offset", 5))
|
||||
assert offset(host, "score", "x_offset") == 5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. _load_custom_font_from_element_config() + the _font_root() seam
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestFontRootSeam:
|
||||
def test_default_font_root_is_the_core_install_root(self, build):
|
||||
manager = build()
|
||||
assert Path(manager._font_root()) == CORE_ROOT
|
||||
assert (Path(manager._font_root()) / "assets" / "fonts").is_dir()
|
||||
|
||||
def test_resolve_font_path_honors_an_overridden_root(self, tmp_path):
|
||||
fonts = tmp_path / "assets" / "fonts"
|
||||
fonts.mkdir(parents=True)
|
||||
(fonts / "Bundled.ttf").write_bytes(b"not really a font")
|
||||
host = probe()
|
||||
host._font_root.side_effect = lambda: str(tmp_path)
|
||||
assert SportsCore._resolve_font_path(host, "Bundled.ttf") == str(
|
||||
fonts / "Bundled.ttf")
|
||||
|
||||
def test_unknown_font_returns_the_familiar_relative_path(self):
|
||||
host = probe()
|
||||
assert SportsCore._resolve_font_path(host, "Nope.ttf") == os.path.join(
|
||||
"assets", "fonts", "Nope.ttf")
|
||||
|
||||
|
||||
class TestFontLoaderSignature:
|
||||
"""Hockey's signature is the only safe superset: basketball's positional
|
||||
``default_font: str`` blows up on an explicit None."""
|
||||
|
||||
def test_two_arg_call(self):
|
||||
font = load_font(probe(), {"font": TTF_NAME, "font_size": 10})
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
assert font.size == 10
|
||||
|
||||
def test_default_size_is_used_when_config_omits_it(self):
|
||||
assert load_font(probe(), {}, 12).size == 12
|
||||
|
||||
def test_three_positional_args(self):
|
||||
font = load_font(probe(), {}, 6, "4x6-font.ttf")
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
assert font.size == 6
|
||||
assert font.path.endswith("4x6-font.ttf")
|
||||
|
||||
def test_explicit_default_font_none(self):
|
||||
# The regression this signature guards: os.path.join(..., None).
|
||||
font = load_font(probe(), {"font_size": 9}, default_font=None)
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
assert font.path.endswith(TTF_NAME)
|
||||
|
||||
def test_config_font_wins_over_default_font(self):
|
||||
font = load_font(probe(), {"font": TTF_NAME}, 8, "4x6-font.ttf")
|
||||
assert font.path.endswith(TTF_NAME)
|
||||
|
||||
def test_string_font_size_is_coerced(self):
|
||||
assert load_font(probe(), {"font": TTF_NAME, "font_size": "11"}).size == 11
|
||||
|
||||
|
||||
class TestFontLoaderBehavior:
|
||||
def test_family_alias_resolves_through_the_font_manager_catalog(self):
|
||||
# "press_start" is a FontManager catalog family, not a filename; the
|
||||
# promoted loader must not carry its own duplicate alias table.
|
||||
host = probe()
|
||||
font = load_font(host, {"font": "press_start", "font_size": 8})
|
||||
assert font.path.endswith(TTF_NAME)
|
||||
assert ("PressStart2P-Regular.ttf", 8) in host._font_cache
|
||||
|
||||
def test_memo_cache_returns_the_same_face(self):
|
||||
host = probe()
|
||||
first = load_font(host, {"font": TTF_NAME, "font_size": 8})
|
||||
second = load_font(host, {"font": TTF_NAME, "font_size": 8})
|
||||
assert first is second
|
||||
assert len(host._font_cache) == 1
|
||||
# A different size is a different face.
|
||||
assert load_font(host, {"font": TTF_NAME, "font_size": 9}) is not first
|
||||
assert len(host._font_cache) == 2
|
||||
|
||||
def test_bdf_loads_at_its_native_strike_when_the_request_misses(self):
|
||||
# BDF is a fixed-size bitmap format: FreeType raises "invalid pixel
|
||||
# size" for anything but the file's own strike. Baseball's retry is
|
||||
# the only copy that gets this right.
|
||||
host = probe()
|
||||
font = load_font(host, {"font": BDF_NAME, "font_size": 8})
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
assert font.size == BDF_NATIVE_SIZE
|
||||
assert font.path.endswith(BDF_NAME)
|
||||
assert set(host._bdf_native_size_cache.values()) == {BDF_NATIVE_SIZE}
|
||||
# The retried face is memoized under the REQUESTED size.
|
||||
assert host._font_cache[(BDF_NAME, 8)] is font
|
||||
|
||||
def test_bdf_at_its_native_size_needs_no_retry(self):
|
||||
host = probe()
|
||||
font = load_font(host, {"font": BDF_NAME, "font_size": BDF_NATIVE_SIZE})
|
||||
assert font.size == BDF_NATIVE_SIZE
|
||||
assert host._bdf_native_size_cache == {}
|
||||
|
||||
def test_bdf_strike_lookup_is_memoized(self, monkeypatch):
|
||||
calls = []
|
||||
real = SportsCore.__module__
|
||||
|
||||
def counting(path):
|
||||
calls.append(path)
|
||||
from src.font_manager import FontManager
|
||||
return FontManager._read_bdf_native_size(path)
|
||||
|
||||
monkeypatch.setattr(f"{real}._read_bdf_native_size", counting)
|
||||
host = probe()
|
||||
load_font(host, {"font": BDF_NAME, "font_size": 8})
|
||||
host._font_cache.clear() # force the load path again
|
||||
load_font(host, {"font": BDF_NAME, "font_size": 8})
|
||||
assert len(calls) == 1
|
||||
|
||||
def test_missing_font_falls_back_and_caches_the_fallback(self, caplog):
|
||||
host = probe()
|
||||
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
|
||||
font = load_font(host, {"font": "DoesNotExist.ttf", "font_size": 8})
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
assert font.path.endswith(TTF_NAME)
|
||||
assert any("Font file not found" in r.message for r in caplog.records)
|
||||
# Cached under the requested name so a misconfiguration costs one
|
||||
# disk probe, not one per frame.
|
||||
assert host._font_cache[("DoesNotExist.ttf", 8)] is font
|
||||
|
||||
def test_unknown_extension_falls_back(self):
|
||||
host = probe()
|
||||
font = load_font(host, {"font": "AUTHORS", "font_size": 8})
|
||||
assert font.path.endswith(TTF_NAME)
|
||||
|
||||
def test_fallback_honors_the_supplied_default_font(self):
|
||||
host = probe()
|
||||
font = load_font(host, {"font": "DoesNotExist.ttf"}, 6, "4x6-font.ttf")
|
||||
assert font.path.endswith("4x6-font.ttf")
|
||||
|
||||
|
||||
class TestFontLoaderCwdIndependence:
|
||||
"""The bug the _font_root() seam exists to prevent: every plugin copy
|
||||
joins 'assets/fonts' onto the process cwd, so a process started anywhere
|
||||
else silently degrades to PIL's default bitmap face (the same defect
|
||||
already fixed in FontManager — see CHANGELOG Unreleased/Fixed)."""
|
||||
|
||||
@pytest.mark.parametrize("font_name,expected_size",
|
||||
[(TTF_NAME, 8), (BDF_NAME, BDF_NATIVE_SIZE)])
|
||||
def test_fonts_load_from_an_unrelated_cwd(self, monkeypatch, font_name,
|
||||
expected_size):
|
||||
monkeypatch.chdir("/")
|
||||
host = probe()
|
||||
font = load_font(host, {"font": font_name, "font_size": 8})
|
||||
assert isinstance(font, ImageFont.FreeTypeFont), (
|
||||
f"{font_name} degraded to PIL's default face when the process "
|
||||
"runs outside the install root")
|
||||
assert font.size == expected_size
|
||||
assert Path(font.path) == FONTS_DIR / font_name
|
||||
|
||||
def test_fallback_font_also_survives_an_unrelated_cwd(self, monkeypatch):
|
||||
monkeypatch.chdir("/")
|
||||
font = load_font(probe(), {"font": "DoesNotExist.ttf", "font_size": 8})
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
assert Path(font.path) == FONTS_DIR / TTF_NAME
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Seam guard rails
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestPromotedSeamsExist:
|
||||
@pytest.mark.parametrize("name", [
|
||||
"cleanup", "_get_layout_offset", "_load_custom_font_from_element_config",
|
||||
"_config_schema_path", "_font_root", "_resolve_font_path",
|
||||
])
|
||||
def test_method_is_callable_on_the_base_class(self, name):
|
||||
assert callable(getattr(SportsCore, name, None)), (
|
||||
f"SportsCore.{name} is part of the promoted plugin-facing seam "
|
||||
"(docs/SPORTS_UNIFICATION.md) — plugins probe for it with "
|
||||
"hasattr before delegating.")
|
||||
|
||||
def test_no_sport_names_leaked_into_core(self):
|
||||
"""core.py must never branch on which sport it is (prose and skin-id
|
||||
examples in docstrings are fine — executable code is not)."""
|
||||
tree = ast.parse((CORE_ROOT / "src" / "base_classes" / "sports"
|
||||
/ "core.py").read_text())
|
||||
docstrings = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef,
|
||||
ast.AsyncFunctionDef)):
|
||||
first = node.body[0] if node.body else None
|
||||
if (isinstance(first, ast.Expr)
|
||||
and isinstance(first.value, ast.Constant)
|
||||
and isinstance(first.value.value, str)):
|
||||
docstrings.add(id(first.value))
|
||||
|
||||
tokens = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||||
if id(node) not in docstrings:
|
||||
tokens.append(node.value)
|
||||
elif isinstance(node, ast.Name):
|
||||
tokens.append(node.id)
|
||||
elif isinstance(node, ast.Attribute):
|
||||
tokens.append(node.attr)
|
||||
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef,
|
||||
ast.ClassDef)):
|
||||
tokens.append(node.name)
|
||||
|
||||
haystack = " ".join(tokens).lower()
|
||||
for sport in ("afl", "nrl", "hockey", "baseball", "basketball",
|
||||
"football", "lacrosse", "soccer", "ufc"):
|
||||
assert sport not in haystack, (
|
||||
f"core.py code mentions '{sport}' — core must never learn "
|
||||
"sport names; add an override point instead.")
|
||||
|
||||
|
||||
class TestInstallRootResolution:
|
||||
"""Guards the depth bug the sports.py -> package move introduced.
|
||||
|
||||
The move was byte-identical in every class body, but `__file__` gained a
|
||||
directory, so `Path(__file__).resolve().parents[2]` silently changed from
|
||||
the repo root to `<root>/src`. Textual identity is not semantic identity
|
||||
when code measures its own location: these tests assert the resolved
|
||||
values, not the index.
|
||||
"""
|
||||
|
||||
def test_install_root_is_the_repo_root(self):
|
||||
from src.base_classes.sports.core import _INSTALL_ROOT
|
||||
|
||||
# The repo root is the directory that actually holds src/ and assets/.
|
||||
assert (_INSTALL_ROOT / "src").is_dir()
|
||||
assert (_INSTALL_ROOT / "src" / "base_classes" / "sports").is_dir()
|
||||
assert _INSTALL_ROOT.name != "src", (
|
||||
"_INSTALL_ROOT resolved to src/ — the parents[] depth is off by "
|
||||
"one, which is exactly the regression the package move caused.")
|
||||
|
||||
def test_resolve_project_path_roots_at_repo_not_src(self):
|
||||
from src.base_classes.sports.core import SportsCore, _INSTALL_ROOT
|
||||
|
||||
resolved = SportsCore._resolve_project_path(None, Path("assets/fonts"))
|
||||
assert resolved == _INSTALL_ROOT / "assets" / "fonts"
|
||||
assert "src" not in resolved.relative_to(_INSTALL_ROOT).parts
|
||||
|
||||
def test_absolute_paths_pass_through_unchanged(self):
|
||||
from src.base_classes.sports.core import SportsCore
|
||||
|
||||
absolute = Path("/tmp/some/logo/dir")
|
||||
assert SportsCore._resolve_project_path(None, absolute) == absolute
|
||||
|
||||
def test_font_root_and_project_path_share_one_anchor(self):
|
||||
"""Both consumers must derive from the same constant, so a future
|
||||
move needs exactly one line changed rather than two."""
|
||||
from src.base_classes.sports.core import SportsCore, _INSTALL_ROOT
|
||||
|
||||
assert SportsCore._font_root(None) == str(_INSTALL_ROOT)
|
||||
@@ -0,0 +1,565 @@
|
||||
"""Tests for the methods promoted onto SportsUpcoming / SportsRecent /
|
||||
SportsLive from the nine plugin copies (phase B1 of the sports unification;
|
||||
see docs/SPORTS_UNIFICATION.md).
|
||||
|
||||
Covered promotions:
|
||||
- SportsRecent: `_get_zero_clock_duration` / `_clear_zero_clock_tracking`
|
||||
(+ the `_zero_clock_timestamps` initializer).
|
||||
- SportsLive: `_is_game_really_over` / `_detect_stale_games`
|
||||
(+ `game_update_timestamps` / `stale_game_timeout`, and the
|
||||
`FINAL_PERIOD` / `CLOCK_COUNTS_DOWN` class attributes).
|
||||
- SportsUpcoming: `_select_games_for_display`.
|
||||
- SportsRecent: `_select_recent_games_for_display`.
|
||||
|
||||
The live pair is the risk centre: `_detect_stale_games` is the only caller
|
||||
that *removes* games, so `_is_game_really_over` returning a false positive
|
||||
silently drops a live game from the display. The canonical form deliberately
|
||||
declines to treat a missing clock as 0:00 — the plugin variant that did so
|
||||
dropped clockless sports (baseball) from the FINAL_PERIOD-th period onward.
|
||||
That regression is pinned by
|
||||
`test_missing_clock_at_late_period_is_not_over`.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from freezegun import freeze_time
|
||||
from PIL import Image
|
||||
|
||||
# src.base_classes.sports transitively imports the hardware matrix driver;
|
||||
# stub it so these tests can import the sports base classes off-device.
|
||||
sys.modules.setdefault("rgbmatrix", MagicMock())
|
||||
|
||||
from src.base_classes.hockey import Hockey, HockeyLive
|
||||
from src.base_classes.sports import (
|
||||
SportsCore,
|
||||
SportsLive,
|
||||
SportsRecent,
|
||||
SportsUpcoming,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Harnesses
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _UpcomingHarness(Hockey, SportsUpcoming):
|
||||
"""Cheapest concrete SportsUpcoming: hockey extractor + cache-fed data.
|
||||
|
||||
`_favorite_key` is inherited from SportsCore — these harnesses
|
||||
deliberately do NOT define it, so the selection tests exercise the real
|
||||
seam rather than a local stand-in.
|
||||
"""
|
||||
|
||||
def _fetch_data(self):
|
||||
return None
|
||||
|
||||
|
||||
class _RecentHarness(Hockey, SportsRecent):
|
||||
def _fetch_data(self):
|
||||
return None
|
||||
|
||||
|
||||
class _LiveHarness(HockeyLive):
|
||||
def _fetch_data(self):
|
||||
return None
|
||||
|
||||
|
||||
class _ThreePeriodLiveHarness(_LiveHarness):
|
||||
"""Hockey-shaped: regulation ends after period 3."""
|
||||
|
||||
FINAL_PERIOD = 3
|
||||
|
||||
|
||||
class _CountUpLiveHarness(_LiveHarness):
|
||||
"""Soccer/AFL/NRL-shaped: the clock counts up, so 0:00 is kickoff."""
|
||||
|
||||
CLOCK_COUNTS_DOWN = False
|
||||
|
||||
|
||||
class _IdFavoriteUpcomingHarness(_UpcomingHarness):
|
||||
"""NRL-shaped: abbreviations are ambiguous, so favorites match on team id."""
|
||||
|
||||
def _favorite_key(self, game, side):
|
||||
team_id = game.get(f"{side}_id")
|
||||
return str(team_id) if team_id is not None else None
|
||||
|
||||
|
||||
class _IdFavoriteRecentHarness(_RecentHarness):
|
||||
def _favorite_key(self, game, side):
|
||||
team_id = game.get(f"{side}_id")
|
||||
return str(team_id) if team_id is not None else None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def build_manager(monkeypatch, tmp_path):
|
||||
"""Factory for concrete sports managers: mocked display/cache managers,
|
||||
logo dir redirected to tmp, background service stubbed, and the requests
|
||||
session rigged to prove nothing hits the network."""
|
||||
monkeypatch.setattr(
|
||||
SportsCore, "_initialize_logo_dir", lambda self, configured: tmp_path)
|
||||
monkeypatch.setattr(
|
||||
"src.base_classes.sports.core.get_background_service",
|
||||
lambda *args, **kwargs: MagicMock())
|
||||
|
||||
def build(cls, **mode_cfg):
|
||||
config = {
|
||||
"timezone": "UTC",
|
||||
"display": {},
|
||||
"nhl_scoreboard": {"enabled": True, **mode_cfg},
|
||||
}
|
||||
display_manager = MagicMock()
|
||||
display_manager.matrix.width = 128
|
||||
display_manager.matrix.height = 32
|
||||
display_manager.width = 128
|
||||
display_manager.height = 32
|
||||
display_manager.image = Image.new("RGB", (128, 32))
|
||||
cache_manager = MagicMock()
|
||||
cache_manager.get.return_value = None
|
||||
cache_manager.cache_dir = str(tmp_path)
|
||||
manager = cls(config, display_manager, cache_manager,
|
||||
logging.getLogger("test_sports_modes_promotions"),
|
||||
"nhl")
|
||||
manager.session = MagicMock()
|
||||
manager.session.get.side_effect = requests.exceptions.ConnectionError(
|
||||
"promotion tests are offline")
|
||||
return manager
|
||||
|
||||
return build
|
||||
|
||||
|
||||
def game(game_id="1", home="BOS", away="TOR", start=None, home_id=None,
|
||||
away_id=None, **extra):
|
||||
g = {
|
||||
"id": game_id,
|
||||
"home_abbr": home,
|
||||
"away_abbr": away,
|
||||
"home_id": home_id,
|
||||
"away_id": away_id,
|
||||
"start_time_utc": start,
|
||||
}
|
||||
g.update(extra)
|
||||
return g
|
||||
|
||||
|
||||
def at(day, hour=12):
|
||||
return datetime(2026, 1, day, hour, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _ids(games):
|
||||
return [g["id"] for g in games]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier 1 — zero-clock tracking (SportsRecent)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestZeroClockTracking:
|
||||
def test_initializer_present_and_empty(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
assert manager._zero_clock_timestamps == {}
|
||||
|
||||
def test_first_call_returns_zero_and_starts_tracking(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
assert manager._get_zero_clock_duration("g1") == 0.0
|
||||
assert "g1" in manager._zero_clock_timestamps
|
||||
|
||||
def test_subsequent_call_returns_elapsed_seconds(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
with freeze_time("2026-01-20 12:00:00") as frozen:
|
||||
assert manager._get_zero_clock_duration("g1") == 0.0
|
||||
frozen.tick(45)
|
||||
assert manager._get_zero_clock_duration("g1") == pytest.approx(45.0)
|
||||
frozen.tick(15)
|
||||
assert manager._get_zero_clock_duration("g1") == pytest.approx(60.0)
|
||||
|
||||
def test_tracking_is_per_game(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
with freeze_time("2026-01-20 12:00:00") as frozen:
|
||||
manager._get_zero_clock_duration("g1")
|
||||
frozen.tick(30)
|
||||
assert manager._get_zero_clock_duration("g2") == 0.0
|
||||
assert manager._get_zero_clock_duration("g1") == pytest.approx(30.0)
|
||||
|
||||
def test_clear_resets_tracking(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
with freeze_time("2026-01-20 12:00:00") as frozen:
|
||||
manager._get_zero_clock_duration("g1")
|
||||
frozen.tick(30)
|
||||
manager._clear_zero_clock_tracking("g1")
|
||||
assert "g1" not in manager._zero_clock_timestamps
|
||||
# Restarts from zero after clearing.
|
||||
assert manager._get_zero_clock_duration("g1") == 0.0
|
||||
|
||||
def test_clear_unknown_game_is_a_noop(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
manager._clear_zero_clock_tracking("never-seen") # must not raise
|
||||
assert manager._zero_clock_timestamps == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier 2a — _is_game_really_over (SportsLive)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestIsGameReallyOver:
|
||||
def test_missing_clock_at_late_period_is_not_over(self, build_manager):
|
||||
"""THE baseball regression: no `clock` key at all, period 7.
|
||||
|
||||
The rejected variant coerced a missing clock to the literal "0:00" and
|
||||
declared the game over — dropping every MLB game from the 5th inning
|
||||
onward, since baseball has no game clock and `period` is the inning.
|
||||
A missing clock must fail safe.
|
||||
"""
|
||||
manager = build_manager(_LiveHarness)
|
||||
g = game(period=7, period_text="Top 7th")
|
||||
assert "clock" not in g
|
||||
assert manager._is_game_really_over(g) is False
|
||||
|
||||
def test_none_clock_at_late_period_is_not_over(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock=None, period=7, period_text="Top 7th")) is False
|
||||
|
||||
def test_non_string_clock_at_late_period_is_not_over(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock=0, period=9, period_text="Bot 9th")) is False
|
||||
|
||||
def test_blank_clock_at_late_period_is_not_over(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock=" ", period=5, period_text="5th")) is False
|
||||
|
||||
@pytest.mark.parametrize("period_text", ["Final", "final", "Final/OT",
|
||||
"FINAL", "Final - SO"])
|
||||
def test_final_period_text_is_over(self, build_manager, period_text):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="12:00", period=2, period_text=period_text)) is True
|
||||
|
||||
def test_none_period_text_does_not_raise(self, build_manager):
|
||||
"""All nine plugin copies called `.lower()` on `game.get("period_text", "")`,
|
||||
which is None when the key is present-but-None; `_detect_stale_games`
|
||||
has no try/except around the call."""
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="12:00", period=2, period_text=None)) is False
|
||||
|
||||
def test_missing_period_text_does_not_raise(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(game(clock="12:00", period=2)) is False
|
||||
|
||||
@pytest.mark.parametrize("clock", ["0:00", ":00", "00", "000", " 0:00 "])
|
||||
def test_expired_clock_at_final_period_is_over(self, build_manager, clock):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock=clock, period=4, period_text="Q4")) is True
|
||||
|
||||
def test_expired_clock_after_final_period_is_over(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="0:00", period=5, period_text="OT")) is True
|
||||
|
||||
def test_expired_clock_before_final_period_is_not_over(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="0:00", period=3, period_text="Q3")) is False
|
||||
|
||||
@pytest.mark.parametrize("clock", [":40", "0:40", "1:00"])
|
||||
def test_running_clock_is_not_over(self, build_manager, clock):
|
||||
"""Sub-minute clocks like ':40' are legitimate, not expired."""
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock=clock, period=4, period_text="Q4")) is False
|
||||
|
||||
def test_defaults_are_four_period_countdown(self):
|
||||
assert SportsLive.FINAL_PERIOD == 4
|
||||
assert SportsLive.CLOCK_COUNTS_DOWN is True
|
||||
|
||||
def test_final_period_override_three(self, build_manager):
|
||||
"""Hockey-shaped subclass: regulation ends after period 3."""
|
||||
manager = build_manager(_ThreePeriodLiveHarness)
|
||||
assert manager.FINAL_PERIOD == 3
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="0:00", period=3, period_text="P3")) is True
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="0:00", period=2, period_text="P2")) is False
|
||||
# And the unmodified default still requires period 4.
|
||||
assert build_manager(_LiveHarness)._is_game_really_over(
|
||||
game(clock="0:00", period=3, period_text="P3")) is False
|
||||
|
||||
def test_count_up_clock_never_expires(self, build_manager):
|
||||
"""Soccer/AFL/NRL: 0:00 means kickoff, so the clock branch must not run."""
|
||||
manager = build_manager(_CountUpLiveHarness)
|
||||
assert manager.CLOCK_COUNTS_DOWN is False
|
||||
for period in (1, 2, 4, 9):
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="0:00", period=period, period_text="1st Half")) is False
|
||||
|
||||
def test_count_up_clock_still_honors_final_text(self, build_manager):
|
||||
manager = build_manager(_CountUpLiveHarness)
|
||||
assert manager._is_game_really_over(
|
||||
game(clock="0:00", period=2, period_text="Final")) is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier 2b — _detect_stale_games (SportsLive)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestDetectStaleGames:
|
||||
def test_initializer_defaults(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
assert manager.game_update_timestamps == {}
|
||||
assert manager.stale_game_timeout == 300
|
||||
|
||||
def test_stale_timeout_is_configurable(self, build_manager):
|
||||
manager = build_manager(_LiveHarness, stale_game_timeout=42)
|
||||
assert manager.stale_game_timeout == 42
|
||||
|
||||
def test_mutates_caller_list_in_place_and_returns_none(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
fresh = game("1", period_text="P2", clock="10:00", period=2)
|
||||
over = game("2", period_text="Final", clock="0:00", period=3)
|
||||
games = [fresh, over]
|
||||
original = games
|
||||
|
||||
result = manager._detect_stale_games(games)
|
||||
|
||||
assert result is None
|
||||
assert games is original # same object, mutated in place
|
||||
assert _ids(games) == ["1"]
|
||||
|
||||
def test_evicts_only_past_timeout_games(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
with freeze_time("2026-01-20 12:00:00"):
|
||||
now = time.time()
|
||||
manager.game_update_timestamps = {
|
||||
"1": {"last_seen": now - 10}, # fresh
|
||||
"2": {"last_seen": now - 299}, # just inside the timeout
|
||||
"3": {"last_seen": now - 301}, # past the timeout
|
||||
}
|
||||
games = [game("1", period_text="P1", clock="10:00", period=1),
|
||||
game("2", period_text="P1", clock="10:00", period=1),
|
||||
game("3", period_text="P1", clock="10:00", period=1)]
|
||||
manager._detect_stale_games(games)
|
||||
|
||||
assert _ids(games) == ["1", "2"]
|
||||
assert "3" not in manager.game_update_timestamps
|
||||
assert set(manager.game_update_timestamps) == {"1", "2"}
|
||||
|
||||
def test_unknown_last_seen_is_never_stale(self, build_manager):
|
||||
"""last_seen == 0 (or no entry) means 'never recorded', not 'ancient'."""
|
||||
manager = build_manager(_LiveHarness)
|
||||
games = [game("1", period_text="P1", clock="10:00", period=1),
|
||||
game("2", period_text="P1", clock="10:00", period=1)]
|
||||
manager.game_update_timestamps = {"1": {"last_seen": 0}}
|
||||
manager._detect_stale_games(games)
|
||||
assert _ids(games) == ["1", "2"]
|
||||
|
||||
def test_removes_games_that_are_really_over(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
manager.game_update_timestamps = {"2": {"last_seen": 0}}
|
||||
games = [game("1", period_text="Q2", clock="5:00", period=2),
|
||||
game("2", period_text="Final", clock="0:00", period=4)]
|
||||
manager._detect_stale_games(games)
|
||||
assert _ids(games) == ["1"]
|
||||
assert "2" not in manager.game_update_timestamps
|
||||
|
||||
def test_keeps_clockless_late_game(self, build_manager):
|
||||
"""The end-to-end form of the baseball regression: a clockless game in
|
||||
the 7th must survive the removal path."""
|
||||
manager = build_manager(_LiveHarness)
|
||||
games = [game("mlb-1", period=7, period_text="Top 7th")]
|
||||
manager._detect_stale_games(games)
|
||||
assert _ids(games) == ["mlb-1"]
|
||||
|
||||
def test_games_without_id_are_skipped(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
no_id = {"home_abbr": "BOS", "away_abbr": "TOR",
|
||||
"period_text": "Final", "clock": "0:00", "period": 4}
|
||||
games = [no_id]
|
||||
manager._detect_stale_games(games)
|
||||
# `continue` fires before the "really over" check, so it stays.
|
||||
assert games == [no_id]
|
||||
|
||||
def test_empty_list_is_tolerated(self, build_manager):
|
||||
manager = build_manager(_LiveHarness)
|
||||
games = []
|
||||
assert manager._detect_stale_games(games) is None
|
||||
assert games == []
|
||||
|
||||
def test_removal_is_by_value_not_identity(self, build_manager):
|
||||
"""Sharp edge worth pinning: `list.remove` compares with `dict.__eq__`,
|
||||
so two structurally-equal dicts drop the FIRST occurrence."""
|
||||
manager = build_manager(_LiveHarness)
|
||||
first = game("1", period_text="Final", clock="0:00", period=4)
|
||||
twin = dict(first)
|
||||
games = [first, twin]
|
||||
manager._detect_stale_games(games)
|
||||
# Both entries are removed here (two iterations, two removals), but the
|
||||
# first removal deletes `first`, not the dict being iterated.
|
||||
assert games == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier 3a — _select_games_for_display (SportsUpcoming)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSelectGamesForDisplay:
|
||||
def test_no_favorites_returns_all_sorted_ascending(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness)
|
||||
games = [game("late", start=at(20)), game("early", start=at(10)),
|
||||
game("mid", start=at(15))]
|
||||
assert _ids(manager._select_games_for_display(games, [])) == [
|
||||
"early", "mid", "late"]
|
||||
|
||||
def test_missing_start_time_sorts_last(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness)
|
||||
games = [game("none", start=None), game("early", start=at(10))]
|
||||
assert _ids(manager._select_games_for_display(games, [])) == [
|
||||
"early", "none"]
|
||||
|
||||
def test_filters_to_favorite_teams(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness)
|
||||
games = [game("1", home="BOS", away="TOR", start=at(10)),
|
||||
game("2", home="NYR", away="PIT", start=at(11)),
|
||||
game("3", home="MTL", away="BOS", start=at(12))]
|
||||
assert _ids(manager._select_games_for_display(games, ["BOS"])) == ["1", "3"]
|
||||
|
||||
def test_respects_upcoming_games_to_show_per_team(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness, upcoming_games_to_show=2)
|
||||
games = [game(str(i), home="BOS", away="TOR", start=at(10 + i))
|
||||
for i in range(5)]
|
||||
assert _ids(manager._select_games_for_display(games, ["BOS"])) == ["0", "1"]
|
||||
|
||||
def test_game_between_two_favorites_counts_for_both(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness, upcoming_games_to_show=1)
|
||||
games = [game("shared", home="BOS", away="TOR", start=at(10)),
|
||||
game("bos2", home="BOS", away="NYR", start=at(11)),
|
||||
game("tor2", home="TOR", away="PIT", start=at(12))]
|
||||
# "shared" fills both BOS's and TOR's single slot, so nothing else fits.
|
||||
assert _ids(manager._select_games_for_display(
|
||||
games, ["BOS", "TOR"])) == ["shared"]
|
||||
|
||||
def test_deduplicates_by_game_id(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness, upcoming_games_to_show=5)
|
||||
g = game("dupe", home="BOS", away="TOR", start=at(10))
|
||||
assert _ids(manager._select_games_for_display(
|
||||
[g, dict(g)], ["BOS"])) == ["dupe"]
|
||||
|
||||
def test_non_favorite_games_excluded(self, build_manager):
|
||||
manager = build_manager(_UpcomingHarness)
|
||||
games = [game("1", home="NYR", away="PIT", start=at(10))]
|
||||
assert manager._select_games_for_display(games, ["BOS"]) == []
|
||||
|
||||
def test_favorite_key_seam_supports_id_matching(self, build_manager):
|
||||
"""The NRL case: two clubs share the abbreviation 'NEW', so favorites
|
||||
must be matched on team id. Only the seam changes — the promoted
|
||||
method is identical."""
|
||||
games = [
|
||||
game("knights", home="NEW", away="SYD", home_id=1, away_id=2,
|
||||
start=at(10)),
|
||||
game("warriors", home="NEW", away="SYD", home_id=99, away_id=2,
|
||||
start=at(11)),
|
||||
]
|
||||
abbr_manager = build_manager(_UpcomingHarness)
|
||||
# Abbreviation matching cannot tell the two "NEW" clubs apart.
|
||||
assert _ids(abbr_manager._select_games_for_display(games, ["NEW"])) == [
|
||||
"knights", "warriors"]
|
||||
|
||||
id_manager = build_manager(_IdFavoriteUpcomingHarness)
|
||||
assert _ids(id_manager._select_games_for_display(games, ["99"])) == [
|
||||
"warriors"]
|
||||
|
||||
def test_favorite_key_none_never_matches(self, build_manager):
|
||||
"""`_favorite_key` returning None (missing id) must not match, even
|
||||
against a favorites list holding the string 'None'."""
|
||||
manager = build_manager(_IdFavoriteUpcomingHarness)
|
||||
games = [game("1", home="BOS", away="TOR", start=at(10))] # no ids
|
||||
assert manager._select_games_for_display(games, ["None"]) == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tier 3b — _select_recent_games_for_display (SportsRecent)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSelectRecentGamesForDisplay:
|
||||
def test_no_favorites_returns_all_sorted_descending(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
games = [game("early", start=at(10)), game("late", start=at(20)),
|
||||
game("mid", start=at(15))]
|
||||
assert _ids(manager._select_recent_games_for_display(games, [])) == [
|
||||
"late", "mid", "early"]
|
||||
|
||||
def test_missing_start_time_sorts_last(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
games = [game("none", start=None), game("late", start=at(20))]
|
||||
assert _ids(manager._select_recent_games_for_display(games, [])) == [
|
||||
"late", "none"]
|
||||
|
||||
def test_respects_recent_games_to_show_per_team(self, build_manager):
|
||||
manager = build_manager(_RecentHarness, recent_games_to_show=2)
|
||||
games = [game(str(i), home="BOS", away="TOR", start=at(10 + i))
|
||||
for i in range(5)]
|
||||
# Most recent first.
|
||||
assert _ids(manager._select_recent_games_for_display(
|
||||
games, ["BOS"])) == ["4", "3"]
|
||||
|
||||
def test_game_between_two_favorites_counts_for_both(self, build_manager):
|
||||
manager = build_manager(_RecentHarness, recent_games_to_show=1)
|
||||
games = [game("shared", home="BOS", away="TOR", start=at(20)),
|
||||
game("bos2", home="BOS", away="NYR", start=at(19)),
|
||||
game("tor2", home="TOR", away="PIT", start=at(18))]
|
||||
assert _ids(manager._select_recent_games_for_display(
|
||||
games, ["BOS", "TOR"])) == ["shared"]
|
||||
|
||||
def test_deduplicates_by_game_id(self, build_manager):
|
||||
manager = build_manager(_RecentHarness, recent_games_to_show=5)
|
||||
g = game("dupe", home="BOS", away="TOR", start=at(10))
|
||||
assert _ids(manager._select_recent_games_for_display(
|
||||
[g, dict(g)], ["BOS"])) == ["dupe"]
|
||||
|
||||
def test_non_favorite_games_excluded(self, build_manager):
|
||||
manager = build_manager(_RecentHarness)
|
||||
games = [game("1", home="NYR", away="PIT", start=at(10))]
|
||||
assert manager._select_recent_games_for_display(games, ["BOS"]) == []
|
||||
|
||||
def test_favorite_key_seam_supports_id_matching(self, build_manager):
|
||||
games = [
|
||||
game("knights", home="NEW", away="SYD", home_id=1, away_id=2,
|
||||
start=at(10)),
|
||||
game("warriors", home="NEW", away="SYD", home_id=99, away_id=2,
|
||||
start=at(11)),
|
||||
]
|
||||
id_manager = build_manager(_IdFavoriteRecentHarness)
|
||||
assert _ids(id_manager._select_recent_games_for_display(
|
||||
games, ["99"])) == ["warriors"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Seam / inertness guards
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestPromotionShape:
|
||||
def test_promoted_methods_live_on_the_right_classes(self):
|
||||
assert hasattr(SportsRecent, "_get_zero_clock_duration")
|
||||
assert hasattr(SportsRecent, "_clear_zero_clock_tracking")
|
||||
assert hasattr(SportsRecent, "_select_recent_games_for_display")
|
||||
assert hasattr(SportsUpcoming, "_select_games_for_display")
|
||||
assert hasattr(SportsLive, "_is_game_really_over")
|
||||
assert hasattr(SportsLive, "_detect_stale_games")
|
||||
|
||||
def test_modes_does_not_define_the_favorite_key_seam(self):
|
||||
"""`_favorite_key` is a SportsCore override point. modes.py must call
|
||||
it, never define it — this test fails if the seam is added in the
|
||||
wrong file."""
|
||||
import src.base_classes.sports.modes as modes
|
||||
|
||||
for cls in (SportsUpcoming, SportsRecent, SportsLive):
|
||||
assert "_favorite_key" not in vars(cls)
|
||||
assert "_favorite_key" not in modes.__dict__
|
||||
Reference in New Issue
Block a user