Compare commits

...
Author SHA1 Message Date
ChuckandClaude Opus 5.5 8879886e50 fix(web): keep exception messages out of API responses (py/stack-trace-exposure)
CodeQL had ~40 open py/stack-trace-exposure alerts on main. Almost all
flowed through describe_exception(), which returned "TypeName: message"
(redacted, capped); the rest through _run_systemctl_command's str(err),
WiFiManager's `return False, str(e)`, unit_refresh's f-strings and two
str(e)/f"{err}" messages in api_v3/__init__.py.

describe_exception() now returns a reason code -- the type, plus the
errno symbol for an OSError ("OSError:EIO", "PermissionError:EACCES") --
and logs the redacted message itself. That keeps what #538 wanted (a
failing disk still says EIO in the response) without quoting paths,
URLs or library internals, and fixes every call site at once; the
test_no_api_v3_handler_discards_its_exception policy still holds.

Service results: _get_display_service_status returns active/returncode
only, and the on-demand start/stop `service` result keeps
returncode/active/started/status but drops systemctl stdout/stderr
(logged on failure). Nothing in web_interface/static, the templates or
the MQTT bridge reads those fields. The Starlark SIGKILL-restart error
no longer returns systemctl stderr as `details`.

WiFi, unit-refresh, config-save and plugin-removal failures now say
what failed with the reason code and point at the log. display.py is
untouched (draft #773 edits it).

Tests: test_api_v3_no_exception_text.py drives one route per affected
file with a marker in the exception message and asserts it never
reaches the body; all 13 fail on origin/main, and targeted mutations
(drop the service filter, put stderr back, str(e) in WiFiManager,
{e} in unit_refresh, {install_err} in system.py, message back in
describe_exception) each fail at least one. Tests that asserted the old
message-in-details contract now assert the reason code.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 16:41:07 -04:00
ChuckandClaude Opus 5.5 e40bc47d28 chore: prepare the 3.8.2 release (#776)
Bumps src.__version__ to 3.8.2 and turns Unreleased into ## 3.8.2: the
display's malloc arena cap and between-screen malloc_trim (#774), and
src.common.sports_favorites (#775, sports family 6), which the scoreboards
adopt by flooring on 3.8.2. src/common/README.md and
docs/SPORTS_UNIFICATION.md say 3.8.2 for it; the SPORTS_UNIFICATION module
table also said "next release" for the four stage 4 modules, which shipped
in 3.8.0.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 12:50:05 -04:00
ChuckandClaude Opus 5.5 3bdb5bff3b feat(common): sports_favorites -- the reconciled favourite matching (sports family 6) (#775)
* feat(common): sports_favorites -- the reconciled favourite matching (sports family 6)

New hardware-free module src/common/sports_favorites.py, copied from
ledmatrix-plugins claude/family6-reconcile once the nine scoreboards made
_is_favorite_game (seven bodies), _select_games_for_display (two) and
_select_recent_games_for_display (three) one body each. One mixin per class
that carries the methods, so adopting one gives no manager a method it did
not have:

- SportsFavoritesMixin (SportsCore): _is_favorite_game and _favorite_code.
- SportsUpcomingFavoritesMixin: _select_games_for_display.
- SportsRecentFavoritesMixin: _select_recent_games_for_display.

Each side of a game is named by the 3.5.0 _favorite_key seam
(SportsHelpersMixin; the abbreviation by default, nrl overrides it with the
ESPN team id and None for a missing id) and compared with favorite_teams
stripped and upper-cased. The selection methods give each favourite up to the
per-team limit, count a game between two favourites for both, treat only
games with an id as possible duplicates and log their summary at INFO.

- test/test_sports_favorites.py: the plugins' pinned cases for an abbreviation
  host and an id-keyed (nrl-style) host -- case, spaces, ids, the NEW
  collision, the "None" favourite, missing keys; selection order, limits,
  duplicates and the id-less fix, the INFO summary; host contract, one carrier
  per method, and SportsGameRulesMixin reaching the shared body.
- test/test_sports_favorites_parity.py: with LEDMATRIX_PLUGINS, compares each
  body with every plugin copy (drift-report normalisation plus decorators),
  checks no other plugin class carries a copy, and that only nrl overrides
  _favorite_key.
- mypy ratchet, src/common/README.md, CHANGELOG (Unreleased, New modules).
- sports_helpers docstrings: _favorite_key now has a caller and an override.
- docs/SPORTS_UNIFICATION.md: family 6 status and decisions, and the seam
  table. SportsCoreSharedMixin._round_robin_favorites still groups by raw
  abbreviation or _team_in: it is not one of the plugin bodies, so it waits
  for a later family.

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

* docs(sports): family 6 also routes the Upcoming favourites-only filter and three live boosts

ledmatrix-plugins claude/family6-reconcile now sends the Upcoming update()'s
favourites-only pre-filter and the basketball, hockey and lacrosse live
favourite boost through _is_favorite_game, so a lower-case favourite works on
a favourites-only Upcoming board. The module is unchanged (update() is not
promoted); the parity test still passes against the branch. Updates the
pinned row and cell counts and what is left for later families.

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

* docs: cite ledmatrix-plugins #635 for the family 6 reconcile

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 12:33:05 -04:00
ChuckandClaude Opus 5.5 e745ae8060 feat(display): cap malloc arenas in-process and malloc_trim between screens (#774)
* feat(display): cap malloc arenas in-process and malloc_trim between screens

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

* test: malloc_tuning with ctypes mocked; add to the mypy ratchet

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

* docs(changelog): malloc arena cap and malloc_trim between screens

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

* docs(changelog): spacing

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 10:47:29 -04:00
26 changed files with 1407 additions and 117 deletions
+52
View File
@@ -19,6 +19,58 @@ accepts both, but the store flags the old spelling as deprecated
## Unreleased
- Web API error responses no longer carry an exception's message (CodeQL
`py/stack-trace-exposure`). `describe_exception()` now returns a reason
code -- the exception type, plus the errno for an `OSError`
(`OSError:EIO`, `PermissionError:EACCES`) -- and logs the message
instead, so `details` still names the fault without quoting paths, URLs
or library internals. The display service status and the on-demand
start/stop `service` results keep `active`, `returncode` and `started`
but drop systemctl's `stdout`/`stderr`; WiFi, unit-refresh and
config-save failures say what failed and point at the log.
## 3.8.2
The display hands freed memory back to the OS (#774), and sports consolidation
family 6: `src.common.sports_favorites`, which the scoreboards adopt by
flooring on 3.8.2 (#775).
### The display hands freed memory back to the OS
The display process's resident memory climbed in steps for hours while the
data it held stayed flat: glibc keeps what Python frees in per-thread malloc
arenas and returns little of it. `src/malloc_tuning.py` (new, standard library
only, a no-op off Linux/glibc) does two things in-process, so it reaches
devices without re-running the installer:
- **Arena cap at start-up.** `run.py` calls `mallopt(M_ARENA_MAX, 2)` before any
thread exists, the same cap as the unit's `Environment=MALLOC_ARENA_MAX=2`.
Units installed before that line never got it (systemd runs the copy in
`/etc/systemd/system`); a `MALLOC_ARENA_MAX` in the environment still wins.
- **`malloc_trim(0)` between screens**, at most every 5 minutes, from the top of
the render loop where no frame is being drawn. Measured on a Pi 4: 2-11 ms
per call.
On ledpi (Pi 4, 192x48, Vegas on, nine plugins, a unit without
`MALLOC_ARENA_MAX`), alternated main / branch / branch / main arms of 2.5 h:
two hours in, resident memory was 551 MB on main (the second main arm was
already at 651 MB after 1 h 44 min) against 412 and 386 MB with this change,
and the 20-minute frame soaks came out at 0.147-0.165% late against main's
0.151-0.188%.
### New modules
- `src/common/sports_favorites.py` -- sports consolidation family 6, once the
plugins made `_is_favorite_game` (seven bodies), `_select_games_for_display`
(two) and `_select_recent_games_for_display` (three) one each:
`SportsFavoritesMixin` (`SportsCore`: `_is_favorite_game`, `_favorite_code`),
`SportsUpcomingFavoritesMixin` and `SportsRecentFavoritesMixin` (the
favourites-only picks). Each side of a game is named by the 3.5.0
`_favorite_key` seam and compared with `favorite_teams` stripped and
upper-cased; nrl overrides the key with the ESPN team id. Only a game with an
id can be a duplicate. A plugin may inherit the mixins once it floors on
3.8.2, and deletes its copies then. (#775)
## 3.8.1
Smooth scrolling at the slower speeds, and the fixes and performance work
+49 -11
View File
@@ -87,11 +87,12 @@ more. Shared sports code lives in `src/common`:
| `sports_celebration.py` | 3.7.0 | `SportsCelebrationMixin` — draws the score/win takeover; colour helpers |
| `sports_fetch.py` | 3.7.0 | `SportsFetchMixin` — season fetch, live lookback and live-odds decisions |
| `sports_card_wrappers.py` | 3.7.0 | `SportsCardWrappersMixin` — the game renderer's `sports_card` delegations |
| `sports_plugin_host.py` | next release | `SportsPluginHostMixin` — the plugin class's (`manager.py`) identical helpers: Vegas weight, off-thread switch refresh |
| `sports_live_scroll.py` | next release | `SportsLiveScrollMixin` — rebuild a live scroll strip mid-cycle, keeping the marquee's place |
| `sports_display_rules.py` | next release | `SportsCardOptionsMixin`, `SportsGameRulesMixin` — scorebug date options, the no-favourites filter, non-favourite live dwell |
| `sports_font_path.py` | next release | `resolve_font_path` — what the plugins' `_resolve_font_path` copies return |
| `sports_plugin_host.py` | 3.8.0 | `SportsPluginHostMixin` — the plugin class's (`manager.py`) identical helpers: Vegas weight, off-thread switch refresh |
| `sports_live_scroll.py` | 3.8.0 | `SportsLiveScrollMixin` — rebuild a live scroll strip mid-cycle, keeping the marquee's place |
| `sports_display_rules.py` | 3.8.0 | `SportsCardOptionsMixin`, `SportsGameRulesMixin` — scorebug date options, the no-favourites filter, non-favourite live dwell |
| `sports_font_path.py` | 3.8.0 | `resolve_font_path` — what the plugins' `_resolve_font_path` copies return |
| `sports_game_over.py` | 3.8.1 | `SportsGameOverMixin` — `_is_game_really_over`, with the `FINAL_PERIOD` seam (family 5) |
| `sports_favorites.py` | 3.8.2 | `SportsFavoritesMixin`, `SportsUpcomingFavoritesMixin`, `SportsRecentFavoritesMixin` — `_is_favorite_game` and the favourites-only picks, on the `_favorite_key` seam (family 6) |
Each is described in [src/common/README.md](../src/common/README.md).
@@ -103,7 +104,8 @@ modules taken from the plugin copies, each a **new module** rather than growth
on an existing one: a plugin that deletes a method copy and relies on an older
module having gained it fails at runtime with an `AttributeError`, while a
missing module fails at load, where the version checks can see it.
`sports_helpers.py` holds `_favorite_key`, the override point listed below.
`sports_helpers.py` holds `_favorite_key`, the override point listed below;
`sports_favorites.py` is what calls it.
Each promoted module has a parity test that compares its bodies against the
plugin copies when `LEDMATRIX_PLUGINS` points at a checkout
(`test_sports_helpers.py`, `test_sports_stage3_parity.py`), and
@@ -126,7 +128,7 @@ deprecation cycle.
| `_custom_scorebug_layout(game, draw)` | Per-sport overlay on the base layout | no-op |
| `score_phrase(points, team_abbr)` | Celebration wording (`"GOOOOAAALLL!"` vs `"TOUCHDOWN!"`). `points` is the score delta, which sports with variable-value scores use to name the play | `"<abbr> SCORES!"` — only consulted when `CelebrationMixin` is present |
| `win_phrase(team_abbr)` | Win-celebration wording | `"<abbr> WINS!"` — mixin only |
| `_favorite_key(game, side)` | Which view-model field identifies a team for favorites matching | `game["<side>_abbr"]` |
| `_favorite_key(game, side)` | Which view-model field identifies a team for favorites matching. `sports_favorites` compares it, and each `favorite_teams` entry, stripped and upper-cased; a `None` matches nothing | `game["<side>_abbr"]`. nrl returns the ESPN team id, `None` when it is missing |
| `_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 |
@@ -315,6 +317,35 @@ were pixel-identical. `src/common/sports_game_over.py` holds the body;
`test/test_sports_game_over_parity.py` compares it, and each plugin's
`FINAL_PERIOD`, with the plugin copies.
### Family 6: favourite matching (core done; adoption waits for a release)
ledmatrix-plugins `scripts/test_favourite_matching.py` (#634) pinned 204 rows
across the nine plugins first: `_is_favorite_game` on each manager role, the
two selection methods, the real `update()` with favourites-only on and off,
and the INFO summary; the reconcile extends it to 217 (a lower-case and a
padded favourite through `update()`, and the live favourite boost). The reconcile (ledmatrix-plugins
#635) made `_is_favorite_game` one body on `SportsCore`
(afl and soccer's `SportsUpcoming` copies and five `SportsLive` copies, all
redundant, are gone), added `_favorite_code` beside it, and gave nrl a
`_favorite_key` override instead of its own copies. So that a lower-case
favourite works on a favourites-only Upcoming board, the Upcoming `update()`'s
favourites-only pre-filter and the basketball, hockey and lacrosse live boost
now ask `_is_favorite_game` too (a one-line change each; `update()` itself is
family 13). Of 3,897 cells only those the decisions above explain changed:
case and spaces in eight plugins (30-40 each), the id-less duplicate fix (6-8
each), nrl's key (6) and its "None" match (6), and the INFO line in baseball,
football and ufc. The harness renders were byte-identical. `src/common/sports_favorites.py` holds the
bodies, one mixin per carrying class; `test/test_sports_favorites_parity.py`
compares them with the plugin copies and checks that only nrl overrides
`_favorite_key`.
Left for later families: the live screens' favourites-only filter
(`_classify_live_game` and its inline copies) and favourites-first sort still
compare abbreviations exactly, and
`SportsCoreSharedMixin._round_robin_favorites` groups favourites by raw
abbreviation (or by `_team_in` where a plugin has one) instead of through
`_favorite_key`. The result-colour helpers also wait (decision above).
### Why the method changes
Byte-identical promotion has nearly run dry. Measured on ledmatrix-plugins
@@ -400,7 +431,7 @@ release.
|---|---|---|---|
| 4 | Identical sweep | `manager.py`: `_dispatch_switch_refresh`, `_favorite_team_is_live`, `get_vegas_priority_weight`, `_game_involves`, `_favorite_scan_targets`, `_favorite_scan_games`, `_get_total_games_for_manager` (all nine, 1); the live-scroll helpers `_preserving_scroll_position`, `_refresh_live_scroll_managers`, `_live_scroll_managers`, `_note_live_scroll_built`, `_live_scroll_needs_rebuild`, `_live_scroll_fields` (eight, 1). `sports.py`: `_card_option`, `_filtered_or_all`, `_effective_live_duration`, `_recent_date_text` (eight, 1). 58 identical families in all | Nothing to decide; brings `manager.py` into core as a `SportsPluginHostMixin`. `_resolve_font_path` (identical in nine `sports.py` and eight renderers) becomes `sports_font_path.resolve_font_path`, not `font_layout.resolve_asset_path`, which skips the cwd. Core side done; see [Stage 4](#stage-4-the-identical-sweep-core-done-adoption-waits-for-a-release) |
| 5 | Game-over check | `SportsLive._is_game_really_over` (5) | Pure logic, no pixels; one seam, `FINAL_PERIOD`. The pilot for the procedure. Reconciled to one body and promoted as `sports_game_over`; adoption waits for the release that ships it. See [Family 5](#family-5-the-game-over-check-core-done-adoption-waits-for-a-release) |
| 6 | Favourite matching | `_is_favorite_game` (7 across three classes), `_select_games_for_display` (2: nrl), `_select_recent_games_for_display` (3) | Everything that asks "is this a favourite" goes through the 3.5.0 `_favorite_key` seam |
| 6 | Favourite matching | `_is_favorite_game` (7 across three classes), `_select_games_for_display` (2: nrl), `_select_recent_games_for_display` (3) | Everything that asks "is this a favourite" goes through the 3.5.0 `_favorite_key` seam. Reconciled to one body each and promoted as `sports_favorites`; adoption waits for the release that ships it. See [Family 6](#family-6-favourite-matching-core-done-adoption-waits-for-a-release) |
| 7 | Other-games rotation | `_by_importance`, `_other_games_window`, `_advance_other_games_if_due` (2 each: football), `_rotate_other_games_on_display` (2: ufc) | One outlier each; football carries two fixes the other eight lack |
| 8 | Rankings | `_fetch_team_rankings` (3), `_choose_poll` (3), `_load_division_team_ids`, `_passes_other_filters`, `_best_rank`, `_is_ranked_game` (2 each: football) | Needs 7; the rank badge and the "ranked only" filter read it |
| 9 | Live fetch and odds | `_fetch_todays_games` (5), `_fetch_odds` (3), `_attach_odds_to_rotated_games` (3) | The prerequisite for one shared ESPN poller across plugins |
@@ -444,10 +475,17 @@ suspected behaviour that needs a payload or a rig to confirm first.
as `0:00`). A score level at 0:00 is not over: the game stays live through
the break before overtime, and one that really ends tied ends on its final
status. Baseball keeps its postponed/suspended override in `BaseballLive`.
- **6, favourite matching.** NRL keeps matching favourites by team id
(abbreviations collide: NEW, CAN), through `_favorite_key` rather than its
own copies of the selection methods. Six plugins log the recent-games
selection at INFO; baseball, football and ufc do not.
- **6, favourite matching. Decided 2026-10-05, done:** each side of a game is
named by `_favorite_key` (the abbreviation; NRL overrides it with the ESPN
team id, and `None` for a missing id, which fixes a favourite typed "None"
matching every game without one) and compared with `favorite_teams`
stripped and upper-cased, so " bos" matches BOS. NRL's ambiguous "NEW"
still matches nothing and is logged; routing the result-colour helpers
(`side_is_favorite`, which tint both NEW clubs) through `_favorite_key` is
left for a later family. The recent-games selection logs at INFO in all
nine. ufc stays on the shared body, dormant: its favourites are fighters,
which its MMA managers match themselves (a follow-up). Fix ported: only a
game with an id can be a duplicate in the selection methods.
- **7, other-games rotation.** football advances the rotation window under
`_games_lock` (update() and display() both advance it; interleaved, a
window of games is skipped) and fixes a favourites-only pool that recomposed
+2
View File
@@ -36,6 +36,7 @@ src/common/sports_card.py
src/common/sports_card_wrappers.py
src/common/sports_celebration.py
src/common/sports_display_rules.py
src/common/sports_favorites.py
src/common/sports_fetch.py
src/common/sports_font_path.py
src/common/sports_game_over.py
@@ -59,6 +60,7 @@ src/ipc/contract.py
src/ipc/server.py
src/logging_config.py
src/logo_downloader.py
src/malloc_tuning.py
src/matrix_support.py
src/pi5_matrix_support.py
src/plugin_system/__init__.py
+6
View File
@@ -14,6 +14,12 @@ project_dir = os.path.dirname(os.path.abspath(__file__))
if project_dir not in sys.path:
sys.path.insert(0, project_dir)
# Cap glibc's malloc arenas before any thread exists (arenas already made
# stay): the in-process twin of the unit's MALLOC_ARENA_MAX=2, for units
# installed before that line. A no-op off glibc. See src/malloc_tuning.py.
from src import malloc_tuning
malloc_tuning.cap_arenas()
# Under systemd the watchdog clock is already running, and start-up (plugin
# loads, initial updates) takes far longer than the render loop's limit. Widen
# it before anything slow is imported; the render loop narrows it again once
+1 -1
View File
@@ -4,5 +4,5 @@ LEDMatrix Display System
Core source package for the LED Matrix Display project.
"""
__version__ = "3.8.1"
__version__ = "3.8.2"
+14
View File
@@ -44,6 +44,7 @@ Rules for the package:
| [`sports_card_wrappers`](#sports_card_wrappers) | The game renderer's `sports_card` delegations | Yes (scoreboards) | 3.7.0 |
| [`sports_celebration`](#sports_celebration) | Draw a scoreboard's score/win celebration | Yes (scoreboards) | 3.7.0 |
| [`sports_display_rules`](#sports_display_rules) | Which games a scoreboard shows, for how long, and its scorebug date line | Yes (scoreboards) | 3.8.0 |
| [`sports_favorites`](#sports_favorites) | Which games involve a favourite team, and the favourites-only picks | Yes (scoreboards) | 3.8.2 |
| [`sports_fetch`](#sports_fetch) | Scoreboard season fetch, lookback and live-odds decisions | Yes (scoreboards) | 3.7.0 |
| [`sports_font_path`](#sports_font_path) | Find a scoreboard's bundled font whatever the cwd | Yes (scoreboards) | 3.8.0 |
| [`sports_game_over`](#sports_game_over) | Whether a game ESPN still lists as live has ended | Yes (scoreboards) | 3.8.1 |
@@ -280,6 +281,19 @@ list it before `SportsCoreSharedMixin`) and `SportsGameRulesMixin`
`_effective_live_duration()`, the shorter dwell for a non-favourite live
game).
### sports_favorites
[`sports_favorites.py`](sports_favorites.py). Sports family 6, one mixin per
class that carried the methods: `SportsFavoritesMixin` (`SportsCore`:
`_is_favorite_game(game)` and `_favorite_code(value)`),
`SportsUpcomingFavoritesMixin` (`_select_games_for_display`) and
`SportsRecentFavoritesMixin` (`_select_recent_games_for_display`). Each side
of a game is named by `_favorite_key` (from `SportsHelpersMixin`; NRL
overrides it with the team id) and compared with `favorite_teams` stripped and
upper-cased. The selection methods give each favourite up to the per-team
limit, count a game between two favourites for both, and treat only games
with an id as possible duplicates.
### sports_fetch
[`sports_fetch.py`](sports_fetch.py). `SportsFetchMixin`: the `SportsCore`
+263
View File
@@ -0,0 +1,263 @@
"""Which games involve a favourite team, and which of them to show (sports family 6).
The scoreboards' favourite matching, reconciled in ledmatrix-plugins
(family 6) from seven ``_is_favorite_game`` bodies, two
``_select_games_for_display`` and three ``_select_recent_games_for_display``
into one each, and copied here under their existing names:
- ``SportsFavoritesMixin`` (``SportsCore``): ``_is_favorite_game(game)``,
asked by ``SportsCoreSharedMixin._favorites_first``, the switch-mode
favourite boost (``SportsHelpersMixin._next_switch_index``), the
non-favourite live dwell (``SportsGameRulesMixin._effective_live_duration``)
and the plugins' live rotation; and ``_favorite_code(value)``, the
normalisation both sides of every comparison go through.
- ``SportsUpcomingFavoritesMixin`` (``SportsUpcoming``):
``_select_games_for_display``, the favourites-only pick of upcoming games.
- ``SportsRecentFavoritesMixin`` (``SportsRecent``):
``_select_recent_games_for_display``, the same for finished games, most
recent first.
Each mixin carries only what its class already had, so no manager gains a
method it did not have.
THE RULE
--------
Each side of a game is named by ``_favorite_key(game, side)``, the override
point ``SportsHelpersMixin`` (``src.common.sports_helpers``) has carried since
3.5.0: the team abbreviation by default. A sport whose abbreviations are not
unique overrides it -- NRL returns the ESPN team id (and None when the id is
missing), because "NEW" is both Newcastle and New Zealand. That value and every
entry of ``favorite_teams`` are compared as ``_favorite_code`` leaves them:
stripped and upper-cased, a blank or missing value matching nothing. So
" bos" in the config matches BOS.
The selection methods give each favourite team up to the per-team limit
(``upcoming_games_to_show`` / ``recent_games_to_show``); a game between two
favourites counts for both. Only a game with an id can be a duplicate: two
games without one are two games.
A new module rather than more methods on ``sports_shared`` or
``sports_helpers``, for the reason ``sports_helpers`` gives: a missing module
fails at load, where the version checks see it; a missing method fails
mid-update.
WHAT A HOST MUST PROVIDE
------------------------
Derived by walking every ``self.<attr>`` the mixins read; the host-contract
test in ``test/test_sports_favorites.py`` fails if a read is added without
being listed here.
- ``favorite_teams`` -- the resolved favourites list (``_is_favorite_game``).
The selection methods are handed the list instead.
- ``_favorite_key`` -- ``SportsHelpersMixin`` supplies the default.
- ``_favorite_code`` -- from ``SportsFavoritesMixin``, which the Upcoming and
Recent classes inherit through their ``SportsCore``.
- ``logger`` -- the selection methods log each pick at DEBUG and a summary at
INFO.
- ``upcoming_games_to_show`` (Upcoming) and ``recent_games_to_show`` (Recent)
-- the per-team limits.
The methods read the game dict's ``id`` and ``start_time_utc`` (selection),
whatever ``_favorite_key`` reads (``home_abbr`` / ``away_abbr`` by default),
and ``home_abbr`` / ``away_abbr`` again for the DEBUG line; any may be missing.
BASE ORDER
----------
No other mixin defines these methods, so the position in the bases does not
change which body runs; a method on the plugin's own class still wins. The
mixins have no ``__init__`` and no state.
"""
import logging
from datetime import datetime, timezone
from typing import Callable, Dict, List, Optional
class SportsFavoritesMixin:
"""``SportsCore``'s favourite check. See module docstring."""
# The host contract, declared for type checking only.
favorite_teams: List[str]
_favorite_key: Callable[[Dict, str], Optional[str]]
@staticmethod
def _favorite_code(value) -> Optional[str]:
"""``value`` as favourites are compared: stripped and upper-cased.
None for a missing or blank value, which matches nothing.
"""
if value is None:
return None
return str(value).strip().upper() or None
def _is_favorite_game(self, game: Dict) -> bool:
"""Does either side of this game belong to a favourite team?
``_favorite_key`` names each side (the abbreviation; nrl overrides it
with the ESPN team id), and both it and ``favorite_teams`` are compared
as ``_favorite_code`` normalises them, so " bos" matches BOS.
"""
favorites = {self._favorite_code(team) for team in self.favorite_teams or ()}
favorites.discard(None)
return any(
self._favorite_code(self._favorite_key(game, side)) in favorites
for side in ("home", "away")
)
class SportsUpcomingFavoritesMixin:
"""``SportsUpcoming``'s favourites-only pick. See module docstring."""
# The host contract, declared for type checking only.
logger: logging.Logger
upcoming_games_to_show: int
_favorite_key: Callable[[Dict, str], Optional[str]]
_favorite_code: Callable[[object], Optional[str]]
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.
Teams are matched as _is_favorite_game matches them. Only a game with
an id can be a duplicate: two games without one are two games.
"""
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: Dict[Optional[str], int] = {
code: 0 for code in map(self._favorite_code, favorite_teams) if code
}
for game in sorted_games:
game_id = game.get("id")
if game_id is not None and game_id in selected_ids:
continue
home = self._favorite_code(self._favorite_key(game, "home"))
away = self._favorite_code(self._favorite_key(game, "away"))
home_fav = home in team_counts
away_fav = away in team_counts
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)
if game_id is not None:
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 {game.get('away_abbr')}@{game.get('home_abbr')}: "
f"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
class SportsRecentFavoritesMixin:
"""``SportsRecent``'s favourites-only pick. See module docstring."""
# The host contract, declared for type checking only.
logger: logging.Logger
recent_games_to_show: int
_favorite_key: Callable[[Dict, str], Optional[str]]
_favorite_code: Callable[[object], Optional[str]]
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.
Teams are matched as _is_favorite_game matches them. Only a game with
an id can be a duplicate: two games without one are two games.
"""
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: Dict[Optional[str], int] = {
code: 0 for code in map(self._favorite_code, favorite_teams) if code
}
for game in sorted_games:
game_id = game.get("id")
if game_id is not None and game_id in selected_ids:
continue
home = self._favorite_code(self._favorite_key(game, "home"))
away = self._favorite_code(self._favorite_key(game, "away"))
home_fav = home in team_counts
away_fav = away in team_counts
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)
if game_id is not None:
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 {game.get('away_abbr')}@{game.get('home_abbr')}: "
f"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
__all__ = ["SportsFavoritesMixin", "SportsUpcomingFavoritesMixin", "SportsRecentFavoritesMixin"]
+10 -12
View File
@@ -34,8 +34,9 @@ first core release that ships it (see ``CHANGELOG.md``).
``_favorite_key`` is the one method not taken from the plugins: it is the
override point from the since-removed ``src/base_classes`` sports core,
carried here so later phases (shared celebrations and game selection) have a
hardware-free home for the seam. No plugin defines it today and nothing in this module calls it.
carried here as the hardware-free home for the seam. ``sports_favorites``
calls it; nrl overrides it with the ESPN team id. Nothing in this module
calls it.
WHAT A HOST MUST PROVIDE
------------------------
@@ -47,8 +48,8 @@ listed here.
- ``mode_config`` (dict) and ``logger`` -- ``_setting_int``. ``league`` is
read with ``getattr`` for the warning text only.
- ``games_list`` and ``current_game_index`` -- ``_next_switch_index``; plus
``_is_favorite_game`` (called with a game), which stays per-plugin and is
only called when
``_is_favorite_game`` (called with a game; ``sports_favorites`` has the
shared body), only called when
``favorite_rotation_boost`` is above 1. ``favorite_rotation_boost`` itself
defaults to 1 on the mixin.
- ``last_game_switch`` -- ``_reset_dwell_on_reentry``, read with ``getattr``
@@ -216,15 +217,12 @@ class SportsHelpersMixin:
rather than a branch so core never has to learn the string "nrl"::
def _favorite_key(self, game, side):
return str(game.get(f"{side}_id"))
team_id = game.get(f"{side}_id")
return None if team_id is None else str(team_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`` for a missing
abbreviation, which never matches.
Carried from the since-removed ``src/base_classes`` sports core for
later phases; nothing in this module calls it yet.
It returns ``None`` for a missing id rather than ``str(None)``, which
would match a favourite typed "None". A ``None`` never matches.
``sports_favorites`` compares the value stripped and upper-cased.
"""
return game.get(f"{side}_abbr")
+7
View File
@@ -37,6 +37,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed # pylint: disab
import pytz
from src import display_watchdog
from src.malloc_tuning import MallocTrimmer
from src.display_arbiter import (
Arbiter, ArbiterInputs, ArbiterState, FramePolicy,
ScreenPlan, Source, WifiNotice, live_pick, live_takeover, on_demand_bound, rotation_plan,
@@ -4217,6 +4218,7 @@ class DisplayController:
logger.info(f"Initial mode set to: {self.current_display_mode} (index: {self.current_mode_index}, total modes: {len(self.available_modes)})")
self._publish_current_mode_state()
runner = ScreenRunner(_MODULE_CLOCK, _ScreenHost(self), logger)
trimmer = MallocTrimmer()
while True:
# Arms the watchdog after the first frame -- or after the
@@ -4224,6 +4226,11 @@ class DisplayController:
# it from then on.
display_watchdog.watchdog.loop_pass()
# Between screens, nothing being drawn: every few minutes hand
# the memory glibc is holding for freed images back to the OS
# (src/malloc_tuning.py). A clock read when none is due.
trimmer.maybe_trim()
# Apply plugin enable/disable edits saved via the web UI. The
# config-watcher thread only sets the flag; loading/unloading and
# rebuilding available_modes happens here on the render thread so
+123
View File
@@ -0,0 +1,123 @@
"""Keep glibc's malloc from holding on to memory the display has freed.
The display process allocates and frees PIL images and numpy buffers all day
from a dozen threads. glibc gives each allocating thread its own malloc arena
(up to 8 x CPU count) and returns little of what is freed inside them to the
OS, so resident memory climbs for hours while the live data stays flat. Two
in-process remedies, both standard library only (ctypes) and both no-ops off
Linux/glibc:
* :func:`cap_arenas` -- ``mallopt(M_ARENA_MAX, 2)``, the in-process twin of the
unit's ``Environment=MALLOC_ARENA_MAX=2``. Units installed before that line
existed never got it (systemd runs the copy in /etc/systemd/system), so the
process applies it itself. Call it before any other thread starts: arenas
already created stay. A ``MALLOC_ARENA_MAX`` set in the environment wins.
* :class:`MallocTrimmer` -- ``malloc_trim(0)`` at most every few minutes,
called from the render loop between screens, where no frame is being drawn.
glibc 2.8+ releases free pages from the middle of every arena, not only the
top of the main heap.
Without glibc (macOS, Windows, musl, the dev server on any of them) nothing is
loaded and every call returns False.
"""
import ctypes
import logging
import os
import sys
import time
from typing import Any, Callable, Optional
logger = logging.getLogger(__name__)
#: glibc's mallopt() parameter number for the arena cap (malloc.h).
M_ARENA_MAX = -8
#: The arena cap applied when the environment does not set one; the same value
#: as the unit's ``MALLOC_ARENA_MAX``.
DEFAULT_ARENA_MAX = 2
#: Seconds between malloc_trim() calls. A trim takes about 1-20 ms on a Pi 4,
#: so this keeps it far from frame timing while still returning memory long
#: before it piles up.
TRIM_INTERVAL_SECONDS = 300.0
_UNLOADED = object()
_libc: Any = _UNLOADED
def _load_libc() -> Optional[Any]:
"""The process's C library if it is glibc with malloc_trim, else None."""
global _libc
if _libc is _UNLOADED:
_libc = None
if sys.platform.startswith('linux'):
try:
libc = ctypes.CDLL(None)
# gnu_get_libc_version is glibc-only, so musl (which has
# mallopt but no malloc_trim) is left alone as a whole.
if all(hasattr(libc, name) for name in
('gnu_get_libc_version', 'malloc_trim', 'mallopt')):
libc.malloc_trim.argtypes = [ctypes.c_size_t]
libc.malloc_trim.restype = ctypes.c_int
libc.mallopt.argtypes = [ctypes.c_int, ctypes.c_int]
libc.mallopt.restype = ctypes.c_int
_libc = libc
except (OSError, AttributeError, TypeError):
logger.debug("glibc malloc controls unavailable", exc_info=True)
return _libc
def cap_arenas(max_arenas: int = DEFAULT_ARENA_MAX) -> bool:
"""Cap glibc's malloc arenas at ``max_arenas``. True when the cap was set.
Skipped when ``MALLOC_ARENA_MAX`` is in the environment: glibc has read it
already, and an operator who set it chose that value.
"""
if os.environ.get('MALLOC_ARENA_MAX'):
return False
libc = _load_libc()
if libc is None:
return False
try:
return bool(libc.mallopt(M_ARENA_MAX, int(max_arenas)))
except Exception: # pylint: disable=broad-except
logger.debug("mallopt(M_ARENA_MAX) failed", exc_info=True)
return False
class MallocTrimmer:
"""Calls ``malloc_trim(0)`` at most once per ``interval`` seconds.
:meth:`maybe_trim` is meant for an idle point of the render loop; it costs
one clock read when no trim is due. The first trim comes one interval
after construction, so start-up's allocations have settled.
"""
def __init__(self, interval: float = TRIM_INTERVAL_SECONDS,
clock: Callable[[], float] = time.monotonic) -> None:
self._interval = interval
self._clock = clock
self._libc = _load_libc()
self._next = clock() + interval
@property
def available(self) -> bool:
return self._libc is not None
def maybe_trim(self) -> bool:
"""Trim if one is due. True when malloc_trim ran and released memory."""
if self._libc is None:
return False
now = self._clock()
if now < self._next:
return False
self._next = now + self._interval
try:
released = bool(self._libc.malloc_trim(0))
except Exception: # pylint: disable=broad-except
logger.debug("malloc_trim failed; not trying again", exc_info=True)
self._libc = None
return False
logger.debug("malloc_trim(0) took %.1f ms, released=%s",
(self._clock() - now) * 1000.0, released)
return released
+15 -11
View File
@@ -4,6 +4,7 @@ Centralized error handling for web interface.
Provides helpers for consistent error responses across API endpoints.
"""
import errno
from typing import Any, Optional
from flask import jsonify
@@ -20,10 +21,9 @@ logger = get_logger(__name__)
_MAX_DETAIL_LENGTH = 400
def describe_exception(exc: BaseException,
max_length: int = _MAX_DETAIL_LENGTH) -> str:
def describe_exception(exc: BaseException) -> str:
"""
One-line, safe-to-return description of an exception.
Machine-readable reason code for an exception, safe to return over HTTP.
The generic "an error occurred; see logs for details" tells a user nothing
and, when the failure is bad enough, the logs are unreachable too: a device
@@ -31,20 +31,24 @@ def describe_exception(exc: BaseException,
*including* the log viewer, because journalctl could not be executed. The
underlying `[Errno 5] Input/output error` named the fault immediately.
Returns "TypeName: message", credentials redacted and length capped. The
type alone is worth carrying -- a bare PermissionError says more than any
generic sentence.
So the type and errno still go back -- "OSError:EIO", "PermissionError:
EACCES", "TimeoutExpired" -- but never the exception's message, which can
quote paths, URLs, credentials or a library's internals (CodeQL
py/stack-trace-exposure). The message is logged here instead, so every
reason code a client sees has its full text in the log.
Args:
exc: The exception to describe
max_length: Truncate beyond this many characters
Returns:
A single-line description, never empty
"TypeName" or "TypeName:ERRNO", never empty
"""
message = str(exc).strip()
text = f"{type(exc).__name__}: {message}" if message else type(exc).__name__
return redact_text(text, max_length)
code = type(exc).__name__
exc_errno = getattr(exc, 'errno', None)
if isinstance(exc_errno, int) and exc_errno in errno.errorcode:
code = f"{code}:{errno.errorcode[exc_errno]}"
logger.warning("Error reported to the client as %s: %s", code, redact_text(str(exc)))
return code
def redact_text(text: str, max_length: int = _MAX_DETAIL_LENGTH) -> str:
+9 -9
View File
@@ -1369,7 +1369,7 @@ class WiFiManager:
self.enable_ap_mode(force=True)
except Exception as ap_error: # nosec B110 - last-resort; do not re-raise, but log for debugging
logger.error("Last-resort AP mode enable failed in recovery path: %s", ap_error, exc_info=True)
return False, str(e)
return False, f"Connection failed ({type(e).__name__}); see logs for details"
def _failsafe_ap(self, enabled_msg: str, failed_msg: str) -> Tuple[bool, str]:
"""Force the setup AP up after a connect that left no working network,
@@ -1585,7 +1585,7 @@ class WiFiManager:
except Exception as e:
logger.error(f"Error connecting with nmcli: {e}")
self._show_led_message("Connection error", duration=5)
return False, str(e)
return False, f"Connection failed ({type(e).__name__}); see logs for details"
# 802.11 caps an SSID at 32 octets. Control characters cannot appear in a
# real one, and a leading "-" would be read by nmcli as an option rather
@@ -1725,7 +1725,7 @@ class WiFiManager:
return False, "nmcli is required to disconnect from WiFi"
except Exception as e:
logger.error(f"Error disconnecting from WiFi: {e}")
return False, str(e)
return False, f"Disconnect failed ({type(e).__name__}); see logs for details"
def _ensure_wifi_radio_enabled(self, max_retries: int = 3) -> bool:
"""
@@ -2004,7 +2004,7 @@ class WiFiManager:
return False, "No WiFi tools available (nmcli, hostapd, or dnsmasq required)"
except Exception as e:
logger.error(f"Error in enable_ap_mode: {e}")
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
def _mark_forced(self) -> None:
"""Record that AP mode was forced on, so the periodic check leaves it
@@ -2099,10 +2099,10 @@ class WiFiManager:
return True, "AP mode enabled"
except Exception as e:
logger.error(f"Error starting AP services: {e}")
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
except Exception as e:
logger.error(f"Error enabling AP mode: {e}")
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
def _enable_ap_mode_nmcli_hotspot(self) -> Tuple[bool, str]:
"""
@@ -2227,7 +2227,7 @@ class WiFiManager:
logger.error(f"Error starting AP mode with nmcli: {e}")
self._remove_nm_dnsmasq_captive_conf()
self._show_led_message("Setup mode error", duration=5)
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
def _get_ap_status_nmcli(self) -> Dict:
"""
@@ -2409,10 +2409,10 @@ class WiFiManager:
return True, "AP mode disabled"
except Exception as e:
logger.error(f"Error stopping AP services: {e}")
return False, str(e)
return False, f"Could not disable AP mode ({type(e).__name__}); see logs for details"
except Exception as e:
logger.error(f"Error disabling AP mode: {e}")
return False, str(e)
return False, f"Could not disable AP mode ({type(e).__name__}); see logs for details"
def _create_hostapd_config(self):
"""Create hostapd configuration file"""
+2 -1
View File
@@ -147,7 +147,8 @@ class TestOneBadConfigSectionDoesNotBlankTheList:
side_effect=RuntimeError("disk is gone"))
resp = api_v3_client.get('/api/v3/display/modes')
assert resp.status_code == 500
assert 'disk is gone' in resp.get_json()['details']
assert resp.get_json()['details'] == 'RuntimeError'
assert 'disk is gone' not in json.dumps(resp.get_json())
def test_credentials_in_the_exception_are_redacted(self, api_v3_module, api_v3_client):
"""describe_exception is what makes returning detail safe."""
+147
View File
@@ -0,0 +1,147 @@
"""No API response carries an exception's message (CodeQL py/stack-trace-exposure).
One representative route per file that had open alerts. Each forces a failure
whose message holds a marker and asserts the marker is nowhere in the body:
the message goes to the log, the client gets a fixed message plus a reason
code (describe_exception: the type, and the errno for an OSError).
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
LEAK = "LEAKED-/home/pi/secret token=abc123"
API = "web_interface.blueprints.api_v3"
def _assert_no_leak(response):
body = response.get_data(as_text=True)
assert "LEAKED" not in body, body
assert "abc123" not in body, body
return json.loads(body)
def test_display_service_status_drops_systemctl_output(api_v3_module, api_v3_client,
monkeypatch):
"""display.py: the on-demand routes return the service status verbatim."""
api_v3_module.api_v3.cache_manager.get.return_value = None
monkeypatch.setattr(f"{API}.display.display_state.read_state", lambda: None)
with patch(f"{API}.subprocess.run", side_effect=OSError(13, LEAK)):
body = _assert_no_leak(api_v3_client.get("/api/v3/display/on-demand/status"))
assert body["data"]["service"] == {"active": False, "returncode": -1}
@pytest.mark.parametrize("helper", ["_ensure_display_service_running",
"_stop_display_service"])
def test_service_results_keep_returncode_but_not_output(api_v3_module, helper):
"""display.py start/stop: returncode/active/started stay, stdout/stderr go."""
failed = MagicMock(returncode=1, stdout=LEAK, stderr=LEAK)
with patch(f"{API}.subprocess.run", return_value=failed):
result = getattr(api_v3_module, helper)()
assert "LEAKED" not in json.dumps(result)
assert result["returncode"] == 1 and result["active"] is False
assert "stdout" not in result and "stderr" not in result
def test_wifi_connect_failure(api_v3_client):
"""wifi.py: a raising connect, and the attempt /wifi/status reports after."""
with patch("src.wifi_manager.WiFiManager") as cls:
cls.return_value._is_ap_mode_active.return_value = False
cls.return_value.connect_to_network.side_effect = RuntimeError(LEAK)
body = _assert_no_leak(api_v3_client.post(
"/api/v3/wifi/connect", json={"ssid": "HomeNet", "password": "pw"}))
assert body["details"] == "RuntimeError"
cls.return_value.get_wifi_status.return_value = MagicMock(
connected=False, ssid=None, ip_address=None, signal=0, ap_mode_active=False)
cls.return_value.config = {}
status = _assert_no_leak(api_v3_client.get("/api/v3/wifi/status"))
assert status["data"]["last_connect_attempt"]["message"] == (
"Failed to connect to network (RuntimeError)")
def test_wifi_manager_messages_carry_no_exception_text():
"""src/wifi_manager.py: its (success, message) is what the wifi routes return."""
from src.wifi_manager import WiFiManager
manager = WiFiManager.__new__(WiFiManager) # no __init__: no host access
manager.get_wifi_status = MagicMock(side_effect=OSError(5, LEAK))
success, message = manager.disconnect_from_network()
assert success is False
assert "LEAKED" not in message and "OSError" in message
def test_system_action_exception(api_v3_client):
"""system.py: execute_system_action's catch-all."""
with patch("subprocess.run", side_effect=OSError(5, LEAK)):
body = _assert_no_leak(api_v3_client.post(
"/api/v3/system/action", json={"action": "stop_display"}))
assert body["details"] == "OSError:EIO"
def test_calendar_registration_failure(api_v3_client, tmp_path, monkeypatch):
"""plugin_calendar.py: the auth script could not be run."""
plugin_dir = tmp_path / "calendar"
plugin_dir.mkdir()
(plugin_dir / "credentials.json").write_text("{}", encoding="utf-8")
(plugin_dir / "calendar_registration.py").write_text("", encoding="utf-8")
monkeypatch.setattr(f"{API}._calendar_plugin_dir", lambda: plugin_dir)
with patch(f"{API}.subprocess.run", side_effect=OSError(13, LEAK)):
body = _assert_no_leak(api_v3_client.post(
"/api/v3/plugins/calendar/authenticate", json={"code": "x"}))
assert "EACCES" in body["message"]
def test_health_failure(api_v3_client, monkeypatch):
"""misc.py: get_health's catch-all."""
def boom():
raise RuntimeError(LEAK)
monkeypatch.setattr(f"{API}.misc._get_display_service_status", boom)
body = _assert_no_leak(api_v3_client.get("/api/v3/health"))
assert body["details"] == "RuntimeError"
def test_config_route_failure(api_v3_module, api_v3_client):
"""error_handler.py: create_error_response, as config.py's routes use it."""
api_v3_module.api_v3.config_manager.load_config.side_effect = RuntimeError(LEAK)
body = _assert_no_leak(api_v3_client.get("/api/v3/config/schedule"))
assert body["details"] == "RuntimeError"
def test_plugin_route_failure(api_v3_module, api_v3_client):
"""plugins.py: an unhandled error in a plugin route."""
api_v3_module.api_v3.plugin_catalog.get_all_plugin_info.side_effect = RuntimeError(LEAK)
body = _assert_no_leak(api_v3_client.get("/api/v3/plugins/installed"))
assert body["details"] == "RuntimeError"
def test_starlark_route_failure(api_v3_client):
"""starlark.py: one of its catch-alls."""
with patch(f"{API}._get_starlark_plugin", side_effect=RuntimeError(LEAK)):
body = _assert_no_leak(api_v3_client.get("/api/v3/starlark/status"))
assert body["details"] == "RuntimeError"
def test_unit_refresh_failure(monkeypatch):
"""system.py git_pull: perform_core_update appends unit_refresh's message."""
from web_interface import unit_refresh
def boom(*_a, **_k):
raise RuntimeError(LEAK)
monkeypatch.setattr(unit_refresh, "stale_units", boom)
result = unit_refresh.refresh_after_update()
assert result["status"] == unit_refresh.FAILED
assert "LEAKED" not in result["message"]
def test_install_base_requirements_failure(api_v3_client):
"""system.py: a pip install that could not start, in the action's output."""
with patch(f"{API}.system._pip_install_requirements", side_effect=OSError(5, LEAK)):
body = _assert_no_leak(api_v3_client.post(
"/api/v3/system/action", json={"action": "install_base_requirements"}))
assert "Failed: OSError:EIO" in body["output"]
+4 -3
View File
@@ -95,9 +95,10 @@ class TestRefreshPluginStore:
RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
body = api_v3_client.post(self.URL, json={}).get_json()
assert "Traceback" not in str(body)
# `details` is describe_exception output: one line, type-named,
# credential-redacted. It may quote the message, but never a stack.
assert body["details"].startswith("RuntimeError:")
# `details` is describe_exception output: the type, never the
# message or a stack.
assert body["details"] == "RuntimeError"
assert "secret.py" not in str(body)
assert "\n" not in body["details"]
+206
View File
@@ -0,0 +1,206 @@
"""src/malloc_tuning.py: glibc arena cap and periodic malloc_trim, ctypes mocked."""
import ctypes
from pathlib import Path
from unittest import mock
import pytest
from src import malloc_tuning as mt
class FakeLibc:
"""Stands in for ctypes.CDLL(None) on glibc: records calls."""
def __init__(self, trim_result=1, glibc=True):
self.trims = []
self.mallopts = []
self._trim_result = trim_result
if glibc:
self.gnu_get_libc_version = lambda: b'2.41'
self.malloc_trim = mock.Mock(side_effect=self._trim)
self.mallopt = mock.Mock(side_effect=self._mallopt)
def _trim(self, pad):
self.trims.append(pad)
if isinstance(self._trim_result, Exception):
raise self._trim_result
return self._trim_result
def _mallopt(self, param, value):
self.mallopts.append((param, value))
return 1
@pytest.fixture(autouse=True)
def fresh_libc(monkeypatch):
"""Each test loads the C library itself; nothing real is called."""
monkeypatch.setattr(mt, '_libc', mt._UNLOADED)
monkeypatch.delenv('MALLOC_ARENA_MAX', raising=False)
yield
def _on_glibc(monkeypatch, libc):
monkeypatch.setattr(mt.sys, 'platform', 'linux')
cdll = mock.Mock(return_value=libc)
monkeypatch.setattr(mt.ctypes, 'CDLL', cdll)
return cdll
class Clock:
def __init__(self, t=1000.0):
self.t = t
def __call__(self):
return self.t
# -- loading ----------------------------------------------------------------
@pytest.mark.parametrize('platform', ['win32', 'darwin', 'freebsd14'])
def test_not_linux_loads_nothing(monkeypatch, platform):
monkeypatch.setattr(mt.sys, 'platform', platform)
cdll = mock.Mock(side_effect=AssertionError('must not load'))
monkeypatch.setattr(mt.ctypes, 'CDLL', cdll)
assert mt._load_libc() is None
assert mt.cap_arenas() is False
trimmer = mt.MallocTrimmer(interval=0)
assert not trimmer.available
assert trimmer.maybe_trim() is False
cdll.assert_not_called()
def test_linux_without_glibc_is_a_noop(monkeypatch):
"""musl: no gnu_get_libc_version (and no malloc_trim) -- nothing is called."""
libc = FakeLibc(glibc=False)
del libc.malloc_trim
_on_glibc(monkeypatch, libc)
assert mt._load_libc() is None
assert mt.cap_arenas() is False
assert mt.MallocTrimmer(interval=0).maybe_trim() is False
assert libc.mallopts == []
def test_cdll_failure_is_a_noop(monkeypatch):
monkeypatch.setattr(mt.sys, 'platform', 'linux')
monkeypatch.setattr(mt.ctypes, 'CDLL', mock.Mock(side_effect=OSError('no libc')))
assert mt._load_libc() is None
assert mt.cap_arenas() is False
def test_loads_once(monkeypatch):
cdll = _on_glibc(monkeypatch, FakeLibc())
mt._load_libc()
mt._load_libc()
mt.MallocTrimmer()
assert cdll.call_count == 1
def test_declares_c_signatures(monkeypatch):
libc = FakeLibc()
_on_glibc(monkeypatch, libc)
mt._load_libc()
assert libc.malloc_trim.argtypes == [ctypes.c_size_t]
assert libc.mallopt.argtypes == [ctypes.c_int, ctypes.c_int]
# -- cap_arenas ---------------------------------------------------------------
def test_cap_arenas_calls_mallopt(monkeypatch):
libc = FakeLibc()
_on_glibc(monkeypatch, libc)
assert mt.cap_arenas() is True
assert libc.mallopts == [(mt.M_ARENA_MAX, 2)]
assert mt.M_ARENA_MAX == -8 # glibc's malloc.h
def test_cap_arenas_defers_to_the_environment(monkeypatch):
libc = FakeLibc()
_on_glibc(monkeypatch, libc)
monkeypatch.setenv('MALLOC_ARENA_MAX', '4')
assert mt.cap_arenas() is False
assert libc.mallopts == []
def test_cap_arenas_swallows_errors(monkeypatch):
libc = FakeLibc()
libc.mallopt = mock.Mock(side_effect=RuntimeError('boom'))
_on_glibc(monkeypatch, libc)
assert mt.cap_arenas() is False
def test_cap_arenas_matches_the_unit():
"""The in-process default is the value the unit's MALLOC_ARENA_MAX carries."""
unit = (Path(__file__).resolve().parent.parent / 'systemd' / 'ledmatrix.service').read_text()
assert f'Environment=MALLOC_ARENA_MAX={mt.DEFAULT_ARENA_MAX}\n' in unit
# -- MallocTrimmer ------------------------------------------------------------
def test_trim_waits_one_interval_then_rate_limits(monkeypatch):
libc = FakeLibc()
_on_glibc(monkeypatch, libc)
clock = Clock()
trimmer = mt.MallocTrimmer(interval=300, clock=clock)
assert trimmer.available
assert trimmer.maybe_trim() is False # start-up: not yet
clock.t += 299.9
assert trimmer.maybe_trim() is False
clock.t += 0.1
assert trimmer.maybe_trim() is True
assert libc.trims == [0]
clock.t += 100
assert trimmer.maybe_trim() is False # rate-limited
clock.t += 200
assert trimmer.maybe_trim() is True
assert libc.trims == [0, 0]
def test_trim_reports_nothing_released(monkeypatch):
libc = FakeLibc(trim_result=0)
_on_glibc(monkeypatch, libc)
clock = Clock()
trimmer = mt.MallocTrimmer(interval=10, clock=clock)
clock.t += 10
assert trimmer.maybe_trim() is False
assert libc.trims == [0]
def test_trim_failure_disables_trimming(monkeypatch):
libc = FakeLibc(trim_result=RuntimeError('boom'))
_on_glibc(monkeypatch, libc)
clock = Clock()
trimmer = mt.MallocTrimmer(interval=10, clock=clock)
clock.t += 10
assert trimmer.maybe_trim() is False
clock.t += 10
assert trimmer.maybe_trim() is False
assert libc.trims == [0] # not retried
assert not trimmer.available
# -- wiring -------------------------------------------------------------------
def test_run_py_caps_arenas_before_threads():
"""run.py applies the cap before the watchdog or the controller import."""
src = (Path(__file__).resolve().parent.parent / 'run.py').read_text()
cap = src.index('malloc_tuning.cap_arenas()')
assert cap < src.index('display_watchdog.watchdog.begin_startup()')
assert cap < src.index('from src.display_controller import main')
def test_render_loop_trims_between_screens():
src = (Path(__file__).resolve().parent.parent / 'src' / 'display_controller.py').read_text()
loop = src.index('display_watchdog.watchdog.loop_pass()')
trim = src.index('trimmer.maybe_trim()')
assert loop < trim < src.index('outcome = runner.run(plan, manager_to_display)')
@pytest.mark.skipif(not mt.sys.platform.startswith('linux'), reason='glibc only')
def test_real_libc_on_linux():
"""On a real Linux C library the calls go through without raising."""
if mt._load_libc() is None:
pytest.skip('not glibc')
trimmer = mt.MallocTrimmer(interval=0)
assert trimmer.available
assert trimmer.maybe_trim() in (True, False)
assert trimmer.available # did not fail and disable itself
+280
View File
@@ -0,0 +1,280 @@
"""src.common.sports_favorites: behaviour, the _favorite_key seam, host contract.
The cases follow ledmatrix-plugins' ``scripts/test_favourite_matching.py``
(the tables the family 6 reconcile was checked against), with the favourites
given as each plugin's resolver hands them over: as typed for the
abbreviation sports, as ESPN team ids for an NRL-style host that overrides
``_favorite_key``.
"""
import ast
import logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
from src.common import sports_favorites
from src.common.sports_favorites import (
SportsFavoritesMixin,
SportsRecentFavoritesMixin,
SportsUpcomingFavoritesMixin,
)
from src.common.sports_helpers import SportsHelpersMixin
LOG = logging.getLogger("test_sports_favorites")
def _id_key(self, game, side):
"""NRL's override: the ESPN team id, None when it is missing."""
team_id = game.get(f"{side}_id")
return None if team_id is None else str(team_id)
def host(favorites, by_id=False, limit=3):
"""A manager stand-in: the three mixins over SportsHelpersMixin's default key."""
attrs = {"_favorite_key": _id_key} if by_id else {}
cls = type("Host", (SportsUpcomingFavoritesMixin, SportsRecentFavoritesMixin,
SportsFavoritesMixin, SportsHelpersMixin), attrs)
h = cls()
h.logger = LOG
h.favorite_teams = favorites
h.upcoming_games_to_show = h.recent_games_to_show = limit
return h
TEAM = {"1": "AAA", "2": "BBB", "3": "CCC", "4": "DDD", "41": "NEW", "42": "NEW"}
def match(home, away, **extra):
g = {"home_id": home, "home_abbr": TEAM[home], "away_id": away, "away_abbr": TEAM[away]}
g.update(extra)
return g
GAMES = {
"AAA home v BBB": match("1", "2"),
"BBB home v AAA": match("2", "1"),
"CCC v DDD": match("3", "4"),
"Knights (NEW 41) v CCC": match("41", "3"),
"Warriors (NEW 42) v CCC": match("42", "3"),
"AAA v BBB, no ids": {"home_abbr": "AAA", "away_abbr": "BBB"},
"ids 1 v 2, no abbrs": {"home_id": "1", "away_id": "2"},
"AAA v BBB, int ids": match("1", "2", home_id=1, away_id=2),
"lower-case abbrs": {"home_abbr": "aaa ", "away_abbr": "bbb"},
"empty game": {},
}
#: label -> favorite_teams as the resolver hands it over.
FAVORITES = {
"none": [],
"AAA": ["AAA"],
"aaa": ["aaa"],
"' AAA '": [" AAA "],
"1": ["1"],
"int 1": [1],
"AAA,CCC": ["AAA", "CCC"],
"NEW": ["NEW"],
"41": ["41"],
"'None'": ["None"],
"blank": ["", " "],
}
#: (favourites, game) -> answer with the abbreviation key, then the id key.
EXPECTED_IS_FAVORITE = {
"AAA home v BBB": {"AAA": "Y.", "aaa": "Y.", "' AAA '": "Y.", "1": ".Y", "int 1": ".Y",
"AAA,CCC": "Y."},
"BBB home v AAA": {"AAA": "Y.", "aaa": "Y.", "' AAA '": "Y.", "1": ".Y", "int 1": ".Y",
"AAA,CCC": "Y."},
"CCC v DDD": {"AAA,CCC": "Y."},
"Knights (NEW 41) v CCC": {"AAA,CCC": "Y.", "NEW": "Y.", "41": ".Y"},
"Warriors (NEW 42) v CCC": {"AAA,CCC": "Y.", "NEW": "Y."},
"AAA v BBB, no ids": {"AAA": "Y.", "aaa": "Y.", "' AAA '": "Y.", "AAA,CCC": "Y."},
"ids 1 v 2, no abbrs": {"1": ".Y", "int 1": ".Y"},
"AAA v BBB, int ids": {"AAA": "Y.", "aaa": "Y.", "' AAA '": "Y.", "1": ".Y",
"int 1": ".Y", "AAA,CCC": "Y."},
"lower-case abbrs": {"AAA": "Y.", "aaa": "Y.", "' AAA '": "Y.", "AAA,CCC": "Y."},
"empty game": {},
}
@pytest.mark.parametrize("game_label", sorted(GAMES))
@pytest.mark.parametrize("fav_label", sorted(FAVORITES))
def test_is_favorite_game(fav_label, game_label):
want = EXPECTED_IS_FAVORITE[game_label].get(fav_label, "..")
got = "".join("Y" if host(FAVORITES[fav_label], by_id)._is_favorite_game(dict(GAMES[game_label]))
else "." for by_id in (False, True))
assert got == want
class TestFavoriteCode:
@pytest.mark.parametrize("value, code", [
("bos", "BOS"), (" BOS ", "BOS"), ("BOS", "BOS"), (41, "41"),
("", None), (" ", None), (None, None),
])
def test_normalises(self, value, code):
assert SportsFavoritesMixin._favorite_code(value) == code
def test_a_missing_id_is_not_the_string_none(self):
"""str(None) would match a favourite typed "None"; None matches nothing."""
h = host(["None"], by_id=True)
assert h._is_favorite_game({"home_abbr": "AAA", "away_abbr": "BBB"}) is False
assert h._is_favorite_game({}) is False
def test_a_none_favorites_list_matches_nothing(self):
assert host(None)._is_favorite_game(dict(GAMES["AAA home v BBB"])) is False
# ---------------------------------------------------------------------------
# Selection. A shuffled slate: two games share id s2, two have no id, s9 has
# no start time. Hours from now; Recent gets the same slate in the past.
# ---------------------------------------------------------------------------
NOW = datetime(2026, 10, 5, 15, tzinfo=timezone.utc)
SLATE = (("s5", "41", "2", 5), ("s1", "1", "2", 1), ("s3", "4", "3", 3),
("s2", "3", "1", 2), ("s7", "2", "3", 7), ("s4", "1", "4", 4),
("s6", "42", "4", 6), ("s2", "1", "4", 8), ("s9", "1", "3", None),
(None, "3", "1", 9), (None, "2", "1", 10))
def slate(recent):
sign = -1 if recent else 1
games = []
for gid, home, away, hours in SLATE:
g = match(home, away, id=gid)
if hours is not None:
g["start_time_utc"] = NOW + timedelta(hours=sign * hours)
games.append(g)
return games
def pick(favorites, limit, recent, by_id=False):
h = host(favorites, by_id, limit)
method = h._select_recent_games_for_display if recent else h._select_games_for_display
return ",".join(g["id"] or "~" for g in method(slate(recent), favorites)) or "none"
ALL = "s1,s2,s3,s4,s5,s6,s7,s2,~,~,s9"
#: (favourites, per-team limit) -> picked ids, the same for Upcoming and Recent.
EXPECTED_SELECT = {
(("AAA",), 1): "s1", (("AAA",), 2): "s1,s2", (("AAA",), 5): "s1,s2,s4,~,~",
(("aaa",), 5): "s1,s2,s4,~,~", ((" AAA ",), 2): "s1,s2",
(("AAA", "CCC"), 1): "s1,s2", (("AAA", "CCC"), 2): "s1,s2,s3",
(("AAA", "CCC"), 5): "s1,s2,s3,s4,s7,~,~,s9",
(("AAA", "ZZZ"), 2): "s1,s2", (("NEW",), 2): "s5,s6",
(("ZZZ",), 2): "none", ((), 1): ALL,
}
@pytest.mark.parametrize("recent", [False, True], ids=["upcoming", "recent"])
@pytest.mark.parametrize("favorites, limit", sorted(EXPECTED_SELECT))
def test_select(favorites, limit, recent):
assert pick(list(favorites), limit, recent) == EXPECTED_SELECT[(favorites, limit)]
class TestSelectByTeamId:
"""The NRL-style host: the key is the team id, so NEW is two teams."""
@pytest.mark.parametrize("recent", [False, True])
def test_one_club_of_a_shared_abbreviation(self, recent):
assert pick(["41"], 2, recent, by_id=True) == "s5"
@pytest.mark.parametrize("recent", [False, True])
def test_an_unresolved_abbreviation_matches_nothing(self, recent):
assert pick(["NEW"], 2, recent, by_id=True) == "none"
def test_ids_select_like_abbreviations(self):
assert pick(["1"], 5, False, by_id=True) == "s1,s2,s4,~,~"
class TestSelectionRules:
def test_a_game_between_two_favourites_counts_for_both(self):
h = host(["AAA", "BBB"], limit=1)
picked = h._select_games_for_display(slate(False), ["AAA", "BBB"])
assert [g["id"] for g in picked] == ["s1"]
def test_games_without_an_id_are_never_duplicates(self):
games = [match("1", "2"), match("1", "3")]
assert len(host(["AAA"])._select_games_for_display(games, ["AAA"])) == 2
def test_a_reused_id_is_a_duplicate(self):
games = [match("1", "2", id="x"), match("1", "3", id="x")]
assert len(host(["AAA"])._select_games_for_display(games, ["AAA"])) == 1
def test_upcoming_is_soonest_first_and_recent_newest_first(self):
assert pick(["CCC"], 5, False) == "s2,s3,s7,~,s9"
assert pick(["CCC"], 5, True) == "s2,s3,s7,~,s9"
def test_the_handed_list_is_used_not_favorite_teams(self):
h = host(["CCC"], limit=1)
assert [g["id"] for g in h._select_games_for_display(slate(False), ["AAA"])] == ["s1"]
@pytest.mark.parametrize("recent", [False, True])
def test_the_summary_is_logged_at_info(self, recent, caplog):
with caplog.at_level(logging.INFO, logger=LOG.name):
pick(["AAA"], 1, recent)
name = "_select_recent_games_for_display" if recent else "_select_games_for_display"
assert [r.levelno for r in caplog.records if r.funcName == name
and r.levelno >= logging.INFO] == [logging.INFO]
# ---------------------------------------------------------------------------
# Carriers and host contract
# ---------------------------------------------------------------------------
MIXINS = {
"SportsFavoritesMixin": ["_favorite_code", "_is_favorite_game"],
"SportsUpcomingFavoritesMixin": ["_select_games_for_display"],
"SportsRecentFavoritesMixin": ["_select_recent_games_for_display"],
}
def _classes():
tree = ast.parse(Path(sports_favorites.__file__).read_text(encoding="utf-8"))
return {n.name: n for n in tree.body if isinstance(n, ast.ClassDef)}
def _self_reads(cls):
return {node.attr for node in ast.walk(cls)
if isinstance(node, ast.Attribute) and isinstance(node.ctx, ast.Load)
and isinstance(node.value, ast.Name) and node.value.id == "self"}
class TestHostContract:
def test_each_mixin_carries_only_its_class_methods(self):
"""So adopting one gives no manager a method it did not have."""
for name, methods in MIXINS.items():
mixin = getattr(sports_favorites, name)
assert sorted(n for n in vars(mixin) if not n.startswith("__")) == methods
def test_every_host_read_is_documented(self):
reads = set().union(*(_self_reads(c) for c in _classes().values()))
undocumented = sorted(n for n in reads if f"``{n}``" not in sports_favorites.__doc__)
assert undocumented == [], f"read but not in the host contract: {undocumented}"
def test_the_key_comes_from_sports_helpers(self):
"""The seam stays where 3.5.0 put it; this module only calls it."""
assert "_favorite_key" in vars(SportsHelpersMixin)
assert all("_favorite_key" not in vars(getattr(sports_favorites, n)) for n in MIXINS)
def test_no_other_shared_mixin_defines_these(self):
from src.common import sports_display_rules, sports_shared
others = [sports_shared.SportsCoreSharedMixin, sports_shared.SportsRecentSharedMixin,
sports_shared.SportsLiveSharedMixin, SportsHelpersMixin,
sports_display_rules.SportsGameRulesMixin]
for methods in MIXINS.values():
for name in methods:
assert not any(name in vars(o) for o in others), name
def test_the_shared_callers_reach_it(self):
"""_favorites_first and the live dwell ask _is_favorite_game; one body answers."""
from src.common.sports_display_rules import SportsGameRulesMixin
class Host(SportsGameRulesMixin, SportsFavoritesMixin, SportsHelpersMixin):
favorite_teams = ["aaa"]
game_display_duration = 15
non_favorite_live_game_duration = 5
assert Host()._effective_live_duration(dict(GAMES["AAA home v BBB"])) == 15
assert Host()._effective_live_duration(dict(GAMES["CCC v DDD"])) == 5
+135
View File
@@ -0,0 +1,135 @@
"""sports_favorites still matches every plugin copy, and only nrl overrides the key.
``src.common.sports_favorites`` was copied from the scoreboards once family 6
had made each method one body in all nine: ``SportsCore._favorite_code`` and
``_is_favorite_game``, ``SportsUpcoming._select_games_for_display`` and
``SportsRecent._select_recent_games_for_display``. The plugins delete their
copies once they floor on the release that ships this module. Until each has, a
copy that changes on its own is a fix one side has and the other lacks.
Point LEDMATRIX_PLUGINS at a ledmatrix-plugins checkout and each method is
compared with every plugin copy using ``scripts/sports_drift_report.py``'s own
normalisation (the AST with docstrings and annotations dropped), plus the
decorators. A copy that is gone counts as adopted when the plugin's
``sports.py`` names the module. The owner's decision that only nrl overrides
``_favorite_key`` (with the team id) is checked too; that override stays in
the plugin after adoption. Without the variable this skips: core CI has no
plugins checkout.
"""
import ast
import importlib.util
import os
from pathlib import Path
import pytest
from src.common import sports_favorites
REPO = Path(__file__).resolve().parents[1]
SPORTS = ("afl", "baseball", "basketball", "football", "hockey", "lacrosse",
"nrl", "soccer", "ufc")
#: plugin class -> (our mixin, the methods it carries)
CARRIERS = {
"SportsCore": ("SportsFavoritesMixin", ("_favorite_code", "_is_favorite_game")),
"SportsUpcoming": ("SportsUpcomingFavoritesMixin", ("_select_games_for_display",)),
"SportsRecent": ("SportsRecentFavoritesMixin", ("_select_recent_games_for_display",)),
}
#: The owner's decision (docs/SPORTS_UNIFICATION.md, family 6): the sports
#: that name a team by something other than its abbreviation.
OVERRIDES_FAVORITE_KEY = {"nrl"}
def _drift_report():
"""scripts/sports_drift_report.py, loaded by path (scripts/ is no package)."""
spec = importlib.util.spec_from_file_location(
"sports_drift_report", REPO / "scripts" / "sports_drift_report.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
DRIFT = _drift_report()
def _plugins_root():
root = DRIFT.resolve_plugins_dir(os.environ.get("LEDMATRIX_PLUGINS"))
if root is None:
pytest.skip("set LEDMATRIX_PLUGINS to a ledmatrix-plugins checkout to "
"compare this module against the plugin copies")
return root
def _class(tree, name):
return next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name)
def _method(cls, name):
return next((n for n in cls.body
if isinstance(n, ast.FunctionDef) and n.name == name), None)
def _fingerprint(node):
return (DRIFT._digest(node, DRIFT._Canonical()),
tuple(ast.unparse(d) for d in node.decorator_list))
def _ours(mixin):
tree = ast.parse(Path(sports_favorites.__file__).read_text(encoding="utf-8"))
return _class(tree, mixin)
def _plugin_tree(root, sport):
source = (root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")
return source, ast.parse(source)
CASES = [(sport, cls, name) for sport in SPORTS
for cls, (_, names) in CARRIERS.items() for name in names]
@pytest.mark.parametrize("sport, cls, name", CASES)
def test_every_remaining_plugin_copy_matches(sport, cls, name):
source, tree = _plugin_tree(_plugins_root(), sport)
copy = _method(_class(tree, cls), name)
if copy is None:
assert sports_favorites.__name__ in source, (
f"{sport}: no {name} on {cls} and no {sports_favorites.__name__} import")
else:
ours = _method(_ours(CARRIERS[cls][0]), name)
assert _fingerprint(copy) == _fingerprint(ours), (
f"{cls}.{name} in {sport} differs from sports_favorites. "
f"Port the change to both, or stop treating it as shared.")
@pytest.mark.parametrize("sport", SPORTS)
def test_no_other_plugin_class_carries_a_copy(sport):
"""A copy on another class (afl's old SportsUpcoming._is_favorite_game) would shadow the shared one."""
_, tree = _plugin_tree(_plugins_root(), sport)
shared = {name: cls for cls, (_, names) in CARRIERS.items() for name in names}
strays = [f"{node.name}.{name}" for node in tree.body if isinstance(node, ast.ClassDef)
for name, home in shared.items()
if node.name != home and _method(node, name) is not None]
assert strays == []
def test_only_the_decided_sports_override_the_key():
root = _plugins_root()
overriding = {sport for sport in SPORTS
if any(_method(node, "_favorite_key") is not None
for node in _plugin_tree(root, sport)[1].body
if isinstance(node, ast.ClassDef))}
assert overriding == OVERRIDES_FAVORITE_KEY
def test_the_drift_report_still_calls_them_identical():
root = _plugins_root()
families = DRIFT.build(root, ("sports.py",))
rows = {(r["file"], r["family"]): r
for r in (DRIFT.summarise(k, v) for k, v in families.items())}
for _, names in CARRIERS.values():
for name in names:
row = rows.get(("sports.py", name))
assert row is None or row["worst_class_variants"] == 1, name
+31 -27
View File
@@ -11,26 +11,32 @@ than even logging it.
import pytest
from src.web_interface.error_handler import describe_exception
from src.web_interface.error_handler import describe_exception, redact_text
class TestDescribeException:
def test_names_the_type_and_message(self):
detail = describe_exception(OSError(5, "Input/output error", "systemctl"))
assert detail == "OSError: [Errno 5] Input/output error: 'systemctl'"
"""describe_exception is a reason code: type and errno, never the message.
def test_the_reported_failure_is_legible(self):
# The whole point: this string is the diagnosis.
assert "Input/output error" in describe_exception(
OSError(5, "Input/output error", "systemctl"))
The message can quote paths, URLs or credentials (CodeQL
py/stack-trace-exposure), so it goes to the log; the code still names the
fault, as "[Errno 5]" did.
"""
def test_an_oserror_names_its_errno(self):
assert describe_exception(
OSError(5, "Input/output error", "systemctl")) == "OSError:EIO"
def test_a_bare_exception_still_names_its_type(self):
# A PermissionError with no message still says more than "unknown".
assert describe_exception(PermissionError()) == "PermissionError"
assert describe_exception(Exception()) == "Exception"
def test_message_is_kept_when_present(self):
assert describe_exception(ValueError("bad port")) == "ValueError: bad port"
def test_the_message_never_reaches_the_code(self):
assert describe_exception(ValueError("bad port /etc/secret")) == "ValueError"
def test_the_message_is_logged_instead(self, caplog):
describe_exception(RuntimeError("disk on fire token=abc123"))
assert "disk on fire" in caplog.text
assert "abc123" not in caplog.text
class TestCredentialRedaction:
@@ -56,47 +62,44 @@ class TestCredentialRedaction:
("authorization: barecredential", "barecredential"),
])
def test_credentials_never_reach_the_response(self, secret_text, leaked):
detail = describe_exception(RuntimeError(secret_text))
detail = redact_text(secret_text)
assert leaked not in detail
assert "<redacted>" in detail
def test_the_parameter_name_survives_redaction(self):
# Knowing *which* credential was involved is part of the diagnosis.
detail = describe_exception(RuntimeError("https://x/y?api_key=SEC123"))
detail = redact_text("https://x/y?api_key=SEC123")
assert "api_key" in detail
def test_unknown_schemes_keep_their_name(self):
for scheme in ("ApiKey", "Negotiate", "NTLM", "AWS4-HMAC-SHA256"):
detail = describe_exception(
RuntimeError("Authorization: %s SECRETVALUE" % scheme))
detail = redact_text("Authorization: %s SECRETVALUE" % scheme)
assert scheme in detail, detail
assert "SECRETVALUE" not in detail, detail
def test_auth_scheme_and_username_survive(self):
# Which kind of credential, and whose, without the credential itself.
assert "Bearer" in describe_exception(
RuntimeError("Authorization: Bearer eyJ.SECRET.sig"))
assert "user" in describe_exception(
RuntimeError("https://user:hunter2@example.com"))
assert "Bearer" in redact_text("Authorization: Bearer eyJ.SECRET.sig")
assert "user" in redact_text("https://user:hunter2@example.com")
def test_non_secret_context_is_preserved(self):
detail = describe_exception(RuntimeError("https://api.x.com/v1?city=Tampa"))
detail = redact_text("https://api.x.com/v1?city=Tampa")
assert "city=Tampa" in detail
assert "<redacted>" not in detail
class TestBounds:
def test_long_messages_are_truncated(self):
detail = describe_exception(ValueError("x" * 5000))
detail = redact_text("x" * 5000)
assert len(detail) <= 400
def test_newlines_are_collapsed_to_one_line(self):
detail = describe_exception(ValueError("line one\nline two\tthree"))
detail = redact_text("line one\nline two\tthree")
assert "\n" not in detail and "\t" not in detail
assert detail == "ValueError: line one line two three"
assert detail == "line one line two three"
def test_custom_length_is_honoured(self):
assert len(describe_exception(ValueError("y" * 500), max_length=50)) <= 50
assert len(redact_text("y" * 500, max_length=50)) <= 50
class TestHandlersCarryDetail:
@@ -319,10 +322,10 @@ class TestHandlersCarryDetail:
assert resp.status_code == 405, "a wrong method must stay a 405"
assert resp.get_json()["error_code"] == "METHOD_NOT_ALLOWED"
# A genuine server fault still reports as one, with its detail.
# A genuine server fault still reports as one, with its reason code.
resp = client.get("/boom")
assert resp.status_code == 500
assert "Input/output error" in resp.get_json()["details"]
assert resp.get_json()["details"] == "OSError:EIO"
def test_global_handler_reports_the_underlying_error(self):
from flask import Flask, jsonify
@@ -345,4 +348,5 @@ class TestHandlersCarryDetail:
client = app.test_client()
body = client.get("/boom").get_json()
assert body["error_code"] == "UNKNOWN_ERROR"
assert "Input/output error" in body["details"]
assert body["details"] == "OSError:EIO"
assert "Input/output error" not in str(body)
@@ -114,12 +114,11 @@ def test_the_answer_is_what_the_catch_all_returned(client, caplog, method, url,
assert records[-1].exc_info[1] is FORCED
def test_credentials_are_redacted_from_the_detail(client):
def test_the_exception_message_never_reaches_the_detail(client):
body = client.get("/api/v3/plugins/installed").get_json()
for secret in ("SECRET123", "pw1", "K1"):
assert secret not in body["details"]
assert "<redacted>" in body["details"]
assert body["details"].startswith("RuntimeError: forced failure")
for secret in ("SECRET123", "pw1", "K1", "forced failure"):
assert secret not in str(body)
assert body["details"] == "RuntimeError"
def _raise_415():
@@ -218,7 +217,7 @@ class TestPluginActionStep1:
encoding="utf-8")
return d
def test_the_script_error_reaches_the_response(self, plugin_dir, monkeypatch):
def test_the_script_error_is_reported_by_type(self, plugin_dir, monkeypatch):
from unittest.mock import MagicMock
manager = MagicMock()
manager.get_plugin_directory.return_value = str(plugin_dir)
@@ -232,5 +231,6 @@ class TestPluginActionStep1:
assert resp.status_code == 500
body = resp.get_json()
assert body["details"] == "RuntimeError: the auth script failed"
assert body["details"] == "RuntimeError"
assert "the auth script failed" not in str(body)
assert body["message"] == 'An error occurred; see logs for details'
@@ -946,21 +946,27 @@ class TestTheStoreReportsWhyItIsEmpty:
class TestACrashCarriesItsDetail:
"""Seventeen Starlark handlers answered 5xx with no detail at all."""
"""Seventeen Starlark handlers answered 5xx with no detail at all.
def test_browse_returns_the_exception_detail(self, client):
The detail is a reason code (the exception type), not the exception's
message, which stays in the log (CodeQL py/stack-trace-exposure).
"""
def test_browse_returns_the_reason_code(self, client):
with patch('web_interface.blueprints.api_v3._get_tronbyte_repository_class',
side_effect=ImportError("No module named 'yaml'")):
body = client.get('/api/v3/starlark/repository/browse').get_json()
assert 'yaml' in body.get('details', ''), body
assert body.get('details') == 'ImportError', body
assert 'yaml' not in str(body), body
def test_status_returns_the_exception_detail(self, client):
def test_status_returns_the_reason_code(self, client):
with patch('web_interface.blueprints.api_v3._get_starlark_plugin',
side_effect=RuntimeError("plugin manager is not attached")):
body = client.get('/api/v3/starlark/status').get_json()
assert 'plugin manager is not attached' in body.get('details', ''), body
assert body.get('details') == 'RuntimeError', body
assert 'plugin manager is not attached' not in str(body), body
class TestTheListingIsNotCappedAtOneThousand:
+26 -21
View File
@@ -222,7 +222,7 @@ def _save_config_atomic(config_manager, config_data, create_backup=True):
config_manager.save_config(config_data)
return True, None
except Exception as e:
return False, str(e)
return False, f"Failed to save configuration ({describe_exception(e)})"
def _coerce_to_bool(value):
"""
Coerce a form value to a proper Python boolean.
@@ -246,7 +246,11 @@ def _coerce_to_bool(value):
return value.lower() in ('true', 'on', '1', 'yes')
return False
def _get_display_service_status():
"""Return status information about the ledmatrix service."""
"""Return status information about the ledmatrix service.
active/returncode only: this goes back in API responses, and systemctl's
output (or an exception's text) is logged rather than returned.
"""
try:
result = subprocess.run(
['systemctl', 'is-active', 'ledmatrix'],
@@ -254,26 +258,18 @@ def _get_display_service_status():
text=True,
timeout=3
)
if result.stderr.strip():
logger.debug('systemctl is-active ledmatrix: %s', result.stderr.strip())
return {
'active': result.stdout.strip() == 'active',
'returncode': result.returncode,
'stdout': result.stdout.strip(),
'stderr': result.stderr.strip()
}
except subprocess.TimeoutExpired:
return {
'active': False,
'returncode': -1,
'stdout': '',
'stderr': 'timeout'
}
except Exception as err:
return {
'active': False,
'returncode': -1,
'stdout': '',
'stderr': str(err)
}
logger.warning('systemctl is-active ledmatrix timed out')
return {'active': False, 'returncode': -1}
except Exception:
logger.warning('Could not query ledmatrix.service status', exc_info=True)
return {'active': False, 'returncode': -1}
def _run_systemctl_command(args):
"""Run a systemctl command safely."""
try:
@@ -295,18 +291,26 @@ def _run_systemctl_command(args):
'stderr': 'timeout'
}
except Exception as err:
logger.warning('%s failed', ' '.join(args), exc_info=True)
return {
'returncode': -1,
'stdout': '',
'stderr': str(err)
'stderr': describe_exception(err)
}
def _public_service_result(result):
"""A _run_systemctl_command result fit for a response: no stdout/stderr."""
if result.get('returncode') != 0:
logger.error('systemctl exited %s: %s', result.get('returncode'),
(result.get('stderr') or '').strip())
return {k: v for k, v in result.items() if k not in ('stdout', 'stderr')}
def _ensure_display_service_running():
"""Ensure the ledmatrix display service is running."""
status = _get_display_service_status()
if status.get('active'):
status['started'] = False
return status
result = _run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix.service'])
result = _public_service_result(
_run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix.service']))
service_status = _get_display_service_status()
result['started'] = result.get('returncode') == 0
result['active'] = service_status.get('active')
@@ -314,7 +318,8 @@ def _ensure_display_service_running():
return result
def _stop_display_service():
"""Stop the ledmatrix display service."""
result = _run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix.service'])
result = _public_service_result(
_run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix.service']))
status = _get_display_service_status()
result['active'] = status.get('active')
result['status'] = status
@@ -712,7 +717,7 @@ def _do_transactional_uninstall(plugin_id, preserve_config):
success = api_v3.plugin_store_manager.uninstall_plugin(plugin_id)
except Exception as remove_err:
_rollback()
return False, f"Failed to remove plugin {plugin_id}: {remove_err}"
return False, f"Failed to remove plugin {plugin_id} ({describe_exception(remove_err)})"
if not success:
_rollback()
@@ -983,7 +983,6 @@ def stop_pixlet_editor():
'status': 'error',
'message': 'Editor force-stopped, but the display could not be '
'restarted automatically - start it manually.',
'details': (result.get('stderr') or '').strip(),
'data': {'running': False}}), 500
return jsonify({'status': 'success',
'message': 'Editor force-stopped; the display has been '
+3 -4
View File
@@ -689,7 +689,7 @@ def execute_system_action():
logger.warning("install_base_requirements timed out for %s", label)
except OSError as install_err:
all_ok = False
outputs.append(f"== {label} ==\nFailed: {install_err}")
outputs.append(f"== {label} ==\nFailed: {describe_exception(install_err)}")
logger.warning("install_base_requirements errored for %s: %s", label, install_err)
return jsonify({
'status': 'success' if all_ok else 'error',
@@ -784,11 +784,10 @@ def execute_system_action():
return jsonify({'status': 'error', 'message': 'Command timed out', 'returncode': -1, 'stderr': 'timeout'})
except Exception as e:
logger.error("execute_system_action failed: %s", e, exc_info=True)
detail = describe_exception(e)
resp = {
'status': 'error',
'message': _sudo_hint_for(detail) or 'Action failed; see logs for details',
'details': detail,
'message': _sudo_hint_for(str(e)) or 'Action failed; see logs for details',
'details': describe_exception(e),
}
return jsonify(resp), 500
@api_v3.route('/system/git-info', methods=['GET'])
+2 -2
View File
@@ -67,7 +67,7 @@ def _run_background_connect(ssid, password):
payload = _connect_result_payload(ssid, success, message)
except Exception as e:
logger.error("Background WiFi connect failed", exc_info=True)
payload = {'status': 'error', 'message': describe_exception(e)}
payload = {'status': 'error', 'message': f'Failed to connect to network ({describe_exception(e)})'}
_record_connect_result(ssid, payload)
@@ -276,7 +276,7 @@ def connect_wifi():
try:
success, message = wifi_manager.connect_to_network(ssid, password)
except Exception as e:
_record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)})
_record_connect_result(ssid, {'status': 'error', 'message': f'Failed to connect to network ({describe_exception(e)})'})
raise
payload = _connect_result_payload(ssid, success, message)
_record_connect_result(ssid, payload)
+2 -2
View File
@@ -86,7 +86,7 @@ def refresh_after_update(run=None, systemd_dir=None, helper_source=None, helper_
except Exception as e: # a broken template must not fail the update itself
if type(e).__name__ != 'UnitsUnreadable':
logger.warning("Could not compare the installed systemd units with the new templates: %s", e)
return _result(FAILED, f'The service settings could not be checked: {e}.')
return _result(FAILED, 'The service settings could not be checked; see logs for details.')
# Units installed mode 0600 (install_service.sh run on its own, before
# it set 0644): only root can compare them, so let the helper decide.
stale = []
@@ -110,7 +110,7 @@ def refresh_after_update(run=None, systemd_dir=None, helper_source=None, helper_
timeout=TIMEOUT_SECONDS)
except (subprocess.SubprocessError, OSError) as e:
logger.warning("Refreshing the systemd units failed: %s", e)
return _result(FAILED, f'Updating the service settings ({names}) failed: {e}.', stale)
return _result(FAILED, f'Updating the service settings ({names}) failed; see logs for details.', stale)
if result.returncode == 0:
# The helper says what it did: "units refreshed: a b" or "units: up to date".
done = next((line.split(':', 1)[1].split() for line in (result.stdout or '').splitlines()