diff --git a/CHANGELOG.md b/CHANGELOG.md index 67b4d1aa..995f3e88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -91,6 +91,23 @@ floor on the release that ships them): - `src.common.api_helper`: `USER_AGENT`, `DEFAULT_HTTP_HEADERS` (read-only). - `src.logo_downloader`: `fetch_logo`, `save_png_atomically`, `shared_downloader`. +- `src.common.sports_card.unshare_element_fonts` takes an optional third + argument, `element_for_font` (default: the module's `ELEMENT_FOR_FONT`, so + existing calls are unchanged). + +### Sports twins + +- The `SportsCoreSharedMixin` helpers that behave identically to their + `sports_card` twins (`_card_option`, `_vs_text`, `_format_game_time`, + `_coerce_rgb`, `_crisp_size`, `_unshare_element_fonts`, the colour/month/ + weekday/font-grid tables) are now thin wrappers over the `sports_card` + functions, and `_format_game_date` / `_schema_font_size` share its + formatting body and schema parser. No method was removed or renamed and + nothing renders differently: `test/test_sports_twins.py` checks each pair + against the same inputs, and the old and new mixin agree on every input + there. The pairs that do differ -- favourite-result colours on nested + payloads, the weekday's timezone, the element-name map, per-mode colours -- + are left as they are and pinned in that test. ### Logo downloads @@ -130,6 +147,13 @@ floor on the release that ships them): when the count is only known to the display service. - The Logs tab has a **Plugin errors** panel: per-plugin counts, repeating errors and a Clear button. +- Credential redaction in exception text (`src/redaction.py`) takes time + proportional to the text, not its square. Two patterns were quadratic: URL + `user:password@`, on a long unbroken run of letters or digits (a hex digest, + an ID), and `Authorization:` followed by a long run of whitespace. Either + used to stall every thread of the display service for up to seconds each + time the snapshot was published: about 0.5s for 20k characters of hex, 8s + for 20k spaces. What gets redacted is unchanged. ### Removed diff --git a/docs/PLUGIN_DEPENDENCY_GUIDE.md b/docs/PLUGIN_DEPENDENCY_GUIDE.md index 5396b0c6..b5b321e0 100644 --- a/docs/PLUGIN_DEPENDENCY_GUIDE.md +++ b/docs/PLUGIN_DEPENDENCY_GUIDE.md @@ -157,5 +157,6 @@ For more, see the [Plugin Dependency Troubleshooting Guide](PLUGIN_DEPENDENCY_TR - Store installs: `src/plugin_system/store_manager.py` (`_install_dependencies`) - Root install helper: `src/common/permission_utils.py` (`install_requirements_file`), `scripts/fix_perms/safe_pip_install.sh` - Load-time installs: `src/plugin_system/plugin_loader.py` (`install_dependencies`) -- Sudo rules: `scripts/install/configure_web_sudo.sh` +- Sudo rules: `scripts/install/lib_sudoers.sh` (written by `first_time_install.sh` + and `scripts/install/configure_web_sudo.sh`) - Manual installer: `scripts/install_plugin_dependencies.sh` diff --git a/docs/SCROLL_PERFORMANCE.md b/docs/SCROLL_PERFORMANCE.md index 82e9f912..b588adfc 100644 --- a/docs/SCROLL_PERFORMANCE.md +++ b/docs/SCROLL_PERFORMANCE.md @@ -303,6 +303,76 @@ journalctl -u ledmatrix --since "-5min" --no-pager | grep -iE "px/s|px/frame" If a plugin logs its scroll config **twice** with different modes, the second line is what is running. +--- + +## A tear across the middle on fast scrolls + +**Symptom:** while text scrolls, the top and bottom halves of the panel look +shifted sideways against each other along a horizontal line at mid-height, and +the shift grows with scroll speed. It shows most in Vegas mode at high speed. + +**It is the panel's scan, not the software.** The measured panel, like most +64-row panels, is multiplexed 1:32 (some panels of the same size scan +differently, so check yours): it lights two rows at a time, one from each half +(row 0 with row 32, row 1 with row 33, …), stepping down both halves together +once per refresh. So row 31, +the last row of the top half, lights almost a whole refresh period after row 32 +right below it. Your eye follows moving text, and moving content that lights at +different times lands in different places, so the two rows meet with an offset +of roughly + +``` +offset ≈ scroll speed × refresh period +``` + +Each frame already reaches the panel whole (`SwapOnVSync` swaps complete frames +between refreshes), so there is nothing to fix in the render path; the shift is +created inside a single refresh. Other panel heights show it too, at the point +where their two scan halves meet. + +On the 2×128×64 chain above, which refreshes at about 130 Hz flat out +(7.7 ms per pass): + +| scroll speed | offset at the midline | +|---|---| +| 50 px/s (Vegas default) | ~0.4 px | +| 100 px/s | ~0.8 px | +| 150 px/s | ~1.2 px, plainly visible | + +### What changes it + +Only a shorter scan period (a faster refresh) or a slower scroll. Measure what +the panel actually achieves first. The library prints the rate with a carriage +return and no newline, so read it from the raw journal: + +```bash +# set display.hardware.show_refresh_rate to true (web UI, Display tab), restart, then: +journalctl -u ledmatrix --since "-1min" --no-pager -o cat --all | grep -a -oE "[0-9.]+Hz" | tail -5 +``` + +Turn it off again afterwards. Measured on that panel (Pi 4, single chain), +changing one setting at a time from `pwm_bits: 7`, `gpio_slowdown: 3`: + +| change | refresh, uncapped | notes | +|---|---|---| +| none | ~130 Hz | the ceiling for this wiring | +| `pwm_bits: 6` | ~138 Hz | barely faster, and half the colour depth | +| `gpio_slowdown: 2` | ~130 Hz | no faster, **and visible glitching**; keep 3 | +| `limit_refresh_rate_hz: 0` | ~130 Hz | Vegas dropped from 100 to 72–95 fps as the refresh thread took more CPU | + +None of these helps much, because the time goes into shifting each row's pixels +out: a 2×128 chain pushes 256 pixels per row down one output. What does help is +**fewer pixels per output**. On a bonnet with more than one output (the +`regular` and `classic` mappings have 3; `adafruit-hat` has 1), put each panel +on its own output and set `parallel` to the number of outputs used and +`chain_length` to the panels per output, for example `parallel: 2`, +`chain_length: 1` for two panels. Each refresh then shifts half the data, which +should roughly double the refresh rate and halve the offset. That is a cable +change, so measure again afterwards. + +Short of rewiring, keep fast scrolls moderate: at the default 50 px/s the +offset is under half a pixel. + ## Rebuilding the binding ```bash diff --git a/first_time_install.sh b/first_time_install.sh index 98fab2fd..8b8ca6dc 100755 --- a/first_time_install.sh +++ b/first_time_install.sh @@ -502,6 +502,41 @@ print_rgbmatrix_build_failure() { fi } +# Set WEB_SERVICE_USER to the account ledmatrix-web.service runs as, or "root" +# when it cannot tell. Steps 3.1 and 11 choose plugin-directory ownership from +# it. The logic was pasted three times, identically, and is kept verbatim here. +# Note: install_web_service.sh and install_service.sh no longer contain the +# "User=root" / "User=${ACTUAL_USER}" strings grepped for below (the units come +# from systemd/*.service templates with User=__USER__), so until Step 8 has +# installed the unit this yields "root". +detect_web_service_user() { + WEB_SERVICE_USER="root" + if [ -f "/etc/systemd/system/ledmatrix-web.service" ]; then + # Check actual installed service file (most accurate) + WEB_SERVICE_USER=$(grep "^User=" /etc/systemd/system/ledmatrix-web.service | cut -d'=' -f2 || echo "root") + elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh" ]; then + # Check install_web_service.sh (used by first_time_install.sh) + if grep -q "User=root" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then + WEB_SERVICE_USER="root" + elif grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then + WEB_SERVICE_USER="$ACTUAL_USER" + fi + elif [ -f "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" ]; then + # Check template file (may have placeholder) + WEB_SERVICE_USER=$(grep "^User=" "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" | cut -d'=' -f2 || echo "root") + # If template has placeholder, check install script + if [ "$WEB_SERVICE_USER" = "__USER__" ] || [ -z "$WEB_SERVICE_USER" ]; then + # Check install_service.sh to see what user it uses + if [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then + WEB_SERVICE_USER="$ACTUAL_USER" + fi + fi + elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then + # Web service will be installed by install_service.sh as ACTUAL_USER + WEB_SERVICE_USER="$ACTUAL_USER" + fi +} + echo "" echo "This script will perform the following steps:" echo "1. Check prerequisites (network, disk, memory) and install system dependencies" @@ -699,32 +734,7 @@ else fi # Determine ownership based on web service user - # Check if web service file exists and what user it runs as - WEB_SERVICE_USER="root" - if [ -f "/etc/systemd/system/ledmatrix-web.service" ]; then - # Check actual installed service file (most accurate) - WEB_SERVICE_USER=$(grep "^User=" /etc/systemd/system/ledmatrix-web.service | cut -d'=' -f2 || echo "root") - elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh" ]; then - # Check install_web_service.sh (used by first_time_install.sh) - if grep -q "User=root" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then - WEB_SERVICE_USER="root" - elif grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" - fi - elif [ -f "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" ]; then - # Check template file (may have placeholder) - WEB_SERVICE_USER=$(grep "^User=" "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" | cut -d'=' -f2 || echo "root") - # If template has placeholder, check install script - if [ "$WEB_SERVICE_USER" = "__USER__" ] || [ -z "$WEB_SERVICE_USER" ]; then - # Check install_service.sh to see what user it uses - if [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" - fi - fi - elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then - # Web service will be installed by install_service.sh as ACTUAL_USER - WEB_SERVICE_USER="$ACTUAL_USER" - fi + detect_web_service_user # If web service runs as ACTUAL_USER (not root), set ownership to ACTUAL_USER # so the web service can change permissions. Root service can still access via group (775). @@ -758,32 +768,7 @@ if [ ! -d "$PLUGIN_REPOS_DIR" ]; then fi # Determine ownership based on web service user -# Check if web service file exists and what user it runs as -WEB_SERVICE_USER="root" -if [ -f "/etc/systemd/system/ledmatrix-web.service" ]; then - # Check actual installed service file (most accurate) - WEB_SERVICE_USER=$(grep "^User=" /etc/systemd/system/ledmatrix-web.service | cut -d'=' -f2 || echo "root") -elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh" ]; then - # Check install_web_service.sh (used by first_time_install.sh) - if grep -q "User=root" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then - WEB_SERVICE_USER="root" - elif grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" - fi -elif [ -f "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" ]; then - # Check template file (may have placeholder) - WEB_SERVICE_USER=$(grep "^User=" "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" | cut -d'=' -f2 || echo "root") - # If template has placeholder, check install script - if [ "$WEB_SERVICE_USER" = "__USER__" ] || [ -z "$WEB_SERVICE_USER" ]; then - # Check install_service.sh to see what user it uses - if [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" - fi - fi -elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then - # Web service will be installed by install_service.sh as ACTUAL_USER - WEB_SERVICE_USER="$ACTUAL_USER" -fi +detect_web_service_user # If web service runs as ACTUAL_USER (not root), set ownership to ACTUAL_USER # so the web service can change permissions. Root service can still access via group (775). @@ -1516,51 +1501,31 @@ POWEROFF_PATH=$(which poweroff) BASH_PATH=$(which bash) JOURNALCTL_PATH=$(which journalctl 2>/dev/null || true) -# Create sudoers content -cat > "$SUDOERS_TMP" << EOF -# LED Matrix Web Interface passwordless sudo configuration -# This allows the web interface user to run specific commands without a password - -# Allow $ACTUAL_USER to run specific commands without a password for the LED Matrix web interface -$ACTUAL_USER ALL=(ALL) NOPASSWD: $REBOOT_PATH -$ACTUAL_USER ALL=(ALL) NOPASSWD: $POWEROFF_PATH -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH enable ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH disable ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix-web.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix-web.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web.service -$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/safe_plugin_rm.sh * -# Install a requirements.txt as root via vetted helper, so packages are visible -# to root-run ledmatrix.service (not just the web interface's own user). -$ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/safe_pip_install.sh * -EOF -if [ -n "$JOURNALCTL_PATH" ]; then - cat >> "$SUDOERS_TMP" << EOF -# NOEXEC, because these rules end in a wildcard and journalctl starts a pager -# when its output is a terminal. From that pager (less) a "!sh" is a root -# shell -- the standard journalctl escalation. The web interface always passes -# --no-pager, so nothing here needs it, but the rule cannot require a flag that -# sits in the middle of the command line. NOEXEC stops the command executing -# another program at all, which closes the hole without depending on wildcard -# matching subtleties. -$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix.service * -$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix * -$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix * -EOF +# The rules themselves live in scripts/install/lib_sudoers.sh, shared with +# scripts/install/configure_web_sudo.sh so the two cannot drift apart again. +# If it is missing (a damaged checkout), keep whatever is already installed +# rather than failing the whole install; the gate below skips the install. +SUDOERS_VALID=1 +SUDOERS_LIB="$PROJECT_ROOT_DIR/scripts/install/lib_sudoers.sh" +if [ -f "$SUDOERS_LIB" ]; then + # shellcheck source=scripts/install/lib_sudoers.sh + . "$SUDOERS_LIB" + web_sudoers_rules "$ACTUAL_USER" "$PROJECT_ROOT_DIR" "$SYSTEMCTL_PATH" "$BASH_PATH" \ + "$REBOOT_PATH" "$POWEROFF_PATH" "$JOURNALCTL_PATH" > "$SUDOERS_TMP" +else + SUDOERS_VALID=0 + echo "⚠ $SUDOERS_LIB not found; cannot generate the sudoers rules." >&2 + echo "⚠ Leaving $SUDOERS_FILE unchanged. The web interface cannot control" >&2 + echo " the display service until this is fixed." >&2 fi # Never install rules we have not parsed. A malformed drop-in in # /etc/sudoers.d makes sudo refuse every command for every user, which on a # headless Pi leaves no way in at all. If the rules do not parse, say so and # keep whatever is already installed. -SUDOERS_VALID=1 -if command -v visudo >/dev/null 2>&1; then +if [ "$SUDOERS_VALID" = "0" ]; then + : # nothing was generated; already reported above +elif command -v visudo >/dev/null 2>&1; then if ! visudo -c -f "$SUDOERS_TMP" >/dev/null 2>&1; then SUDOERS_VALID=0 echo "⚠ The generated sudoers rules did not parse:" >&2 @@ -1690,28 +1655,8 @@ fi # Re-apply plugin directory permissions based on web service user echo "Re-applying plugin directory permissions..." -# Determine web service user (check installed service, install scripts, or template) -WEB_SERVICE_USER="root" -if [ -f "/etc/systemd/system/ledmatrix-web.service" ]; then - # Check actual installed service file (most accurate) - WEB_SERVICE_USER=$(grep "^User=" /etc/systemd/system/ledmatrix-web.service | cut -d'=' -f2 || echo "root") -elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh" ]; then - # Check install_web_service.sh (used by first_time_install.sh) - if grep -q "User=root" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then - WEB_SERVICE_USER="root" - elif grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_web_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" - fi -elif [ -f "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" ]; then - WEB_SERVICE_USER=$(grep "^User=" "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service" | cut -d'=' -f2 || echo "root") - if [ "$WEB_SERVICE_USER" = "__USER__" ] || [ -z "$WEB_SERVICE_USER" ]; then - if [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" - fi - fi -elif [ -f "$PROJECT_ROOT_DIR/scripts/install/install_service.sh" ] && grep -q "User=\${ACTUAL_USER}" "$PROJECT_ROOT_DIR/scripts/install/install_service.sh"; then - WEB_SERVICE_USER="$ACTUAL_USER" -fi +# Determine ownership based on web service user +detect_web_service_user # Set ownership based on web service user if [ "$WEB_SERVICE_USER" = "$ACTUAL_USER" ] || [ "$WEB_SERVICE_USER" != "root" ]; then diff --git a/scripts/install/configure_web_sudo.sh b/scripts/install/configure_web_sudo.sh index 762327e9..91ba1d3b 100755 --- a/scripts/install/configure_web_sudo.sh +++ b/scripts/install/configure_web_sudo.sh @@ -59,6 +59,16 @@ if [ ! -f "$SAFE_PIP_INSTALL_PATH" ]; then exit 1 fi +# The rules are shared with first_time_install.sh (Step 10) so the two cannot +# drift apart; add or remove a grant in lib_sudoers.sh, not here. +SUDOERS_LIB="$PROJECT_DIR/lib_sudoers.sh" +if [ ! -f "$SUDOERS_LIB" ]; then + echo "Error: Sudoers rules library not found: $SUDOERS_LIB" >&2 + exit 1 +fi +# shellcheck source=scripts/install/lib_sudoers.sh +. "$SUDOERS_LIB" + echo "Command paths:" echo " Python: $PYTHON_PATH" echo " Systemctl: $SYSTEMCTL_PATH" @@ -72,56 +82,8 @@ echo " Safe pip install: $SAFE_PIP_INSTALL_PATH" # Create a temporary sudoers file TEMP_SUDOERS="/tmp/ledmatrix_web_sudoers_$$" -{ - echo "# LED Matrix Web Interface passwordless sudo configuration" - echo "# This allows the web interface user to run specific commands without a password" - echo "" - echo "# Allow $WEB_USER to run specific commands without a password for the LED Matrix web interface" - - # Optional: reboot/poweroff (non-critical — skip if not found) - if [ -n "$REBOOT_PATH" ]; then - echo "$WEB_USER ALL=(ALL) NOPASSWD: $REBOOT_PATH" - fi - if [ -n "$POWEROFF_PATH" ]; then - echo "$WEB_USER ALL=(ALL) NOPASSWD: $POWEROFF_PATH" - fi - - # Required: systemctl - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH enable ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH disable ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix-web.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix-web.service" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web.service" - - # Optional: journalctl (non-critical — skip if not found) - # - # NOEXEC, matching first_time_install.sh. These rules end in a wildcard and - # journalctl starts a pager, so without it the caller can reach a shell: - # less runs "!command" as the user the pager belongs to, which here is - # root. NOEXEC stops the granted command executing anything of its own. - if [ -n "$JOURNALCTL_PATH" ]; then - echo "$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix.service *" - echo "$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix *" - echo "$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix *" - fi - - echo "" - echo "# Allow web user to remove plugin directories via vetted helper script" - echo "# The helper validates that the target path resolves inside plugin-repos/ or plugins/" - echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $SAFE_RM_PATH *" - echo "" - echo "# Allow web user to install a plugin's requirements.txt as root via vetted" - echo "# helper script, so packages are visible to root-run ledmatrix.service" - echo "# (not just the web interface's own user). The helper validates the target" - echo "# is requirements.txt at the project root or under plugin-repos/ or plugins/." - echo "$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $SAFE_PIP_INSTALL_PATH *" -} > "$TEMP_SUDOERS" +web_sudoers_rules "$WEB_USER" "$PROJECT_ROOT" "$SYSTEMCTL_PATH" "$BASH_PATH" \ + "$REBOOT_PATH" "$POWEROFF_PATH" "$JOURNALCTL_PATH" > "$TEMP_SUDOERS" # Never offer to install rules we have not parsed. A malformed drop-in in # /etc/sudoers.d makes sudo refuse every command for every user. diff --git a/scripts/install/lib_sudoers.sh b/scripts/install/lib_sudoers.sh new file mode 100755 index 00000000..40fdd51c --- /dev/null +++ b/scripts/install/lib_sudoers.sh @@ -0,0 +1,76 @@ +#!/bin/bash +# +# The web interface's passwordless-sudo allow-list, /etc/sudoers.d/ledmatrix_web. +# +# Sourced by first_time_install.sh (Step 10) and +# scripts/install/configure_web_sudo.sh. Both used to carry their own copy of +# these rules, and the copies drifted: one granted safe_pip_install.sh and the +# other did not. Each caller still owns its own validate (visudo -c) / install / +# confirm flow; this file only prints the rules. +# +# Add or remove a grant here and nowhere else. + +# web_sudoers_rules WEB_USER PROJECT_ROOT SYSTEMCTL_PATH BASH_PATH REBOOT_PATH POWEROFF_PATH JOURNALCTL_PATH +# +# Print the ledmatrix_web sudoers rules to stdout. +# +# SYSTEMCTL_PATH and BASH_PATH are required, and the caller must make sure they +# are not empty: `visudo -c` does not catch every such rule (with an empty +# BASH_PATH the helper rules still parse, granting the script itself). +# first_time_install.sh stops on a failed `which`; configure_web_sudo.sh checks +# them before calling this. +# REBOOT_PATH, POWEROFF_PATH and JOURNALCTL_PATH are optional: pass "" and +# their rules are left out. +web_sudoers_rules() { + local WEB_USER="${1:-}" + local PROJECT_ROOT="${2:-}" + local SYSTEMCTL_PATH="${3:-}" + local BASH_PATH="${4:-}" + local REBOOT_PATH="${5:-}" + local POWEROFF_PATH="${6:-}" + local JOURNALCTL_PATH="${7:-}" + + cat << EOF +# LED Matrix Web Interface passwordless sudo configuration +# This allows the web interface user to run specific commands without a password + +# Allow $WEB_USER to run specific commands without a password for the LED Matrix web interface +EOF + if [ -n "$REBOOT_PATH" ]; then + printf '%s\n' "$WEB_USER ALL=(ALL) NOPASSWD: $REBOOT_PATH" + fi + if [ -n "$POWEROFF_PATH" ]; then + printf '%s\n' "$WEB_USER ALL=(ALL) NOPASSWD: $POWEROFF_PATH" + fi + cat << EOF +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH enable ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH disable ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH status ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH is-active ledmatrix.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start ledmatrix-web.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop ledmatrix-web.service +$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web.service +$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT/scripts/fix_perms/safe_plugin_rm.sh * +# Install a requirements.txt as root via vetted helper, so packages are visible +# to root-run ledmatrix.service (not just the web interface's own user). +$WEB_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT/scripts/fix_perms/safe_pip_install.sh * +EOF + if [ -n "$JOURNALCTL_PATH" ]; then + cat << EOF +# NOEXEC, because these rules end in a wildcard and journalctl starts a pager +# when its output is a terminal. From that pager (less) a "!sh" is a root +# shell -- the standard journalctl escalation. The web interface always passes +# --no-pager, so nothing here needs it, but the rule cannot require a flag that +# sits in the middle of the command line. NOEXEC stops the command executing +# another program at all, which closes the hole without depending on wildcard +# matching subtleties. +$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix.service * +$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix * +$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix * +EOF + fi +} diff --git a/src/background_data_service.py b/src/background_data_service.py index 22989a56..a41aef45 100644 --- a/src/background_data_service.py +++ b/src/background_data_service.py @@ -26,6 +26,7 @@ from enum import Enum from concurrent.futures import ThreadPoolExecutor import pytz from src.cache_manager import CacheManager +from src.common.json_body import response_json from src.common.espn_dates import ( RANGE_RETRY_SECONDS, _note_range_rejected, @@ -389,7 +390,7 @@ class BackgroundDataService: response.raise_for_status() else: response.raise_for_status() - data = response.json() + data = response_json(response) # Validate data structure if not isinstance(data, dict): diff --git a/src/cache/disk_cache.py b/src/cache/disk_cache.py index 87c3d7a9..fb776b56 100644 --- a/src/cache/disk_cache.py +++ b/src/cache/disk_cache.py @@ -7,6 +7,7 @@ Handles persistent disk-based caching with atomic writes and error recovery. import json import math import os +import re import stat import time import tempfile @@ -98,6 +99,40 @@ def _replace_nonfinite(obj: Any) -> Any: # deleted. Both halves are covered by test/test_cache_nonfinite_floats.py. +#: Enough of a record to hold its header: ``{"timestamp":,"ttl":,``. +_HEAD_BYTES = 256 + +#: A record written with its header first (CacheManager.set does). Anything +#: else -- older files with "data" first, records from other writers -- does not +#: match and is parsed in full, as before. +_HEAD_RE = re.compile( + rb'\A\s*\{\s*"timestamp"\s*:\s*(-?[0-9][0-9.eE+-]*)\s*' + rb'(?:,\s*"ttl"\s*:\s*(-?[0-9][0-9.eE+-]*))?\s*[,}]' +) + + +def _stale_from_head(head: bytes, max_age: Optional[int], now: float) -> bool: + """True when a record's header alone shows it has expired. + + Mirrors the expiry rule in DiskCache.get: a per-entry ttl wins over the + caller's max_age, and no limit at all means never stale. False whenever the + header cannot be read, so the full parse decides as it always did. + """ + match = _HEAD_RE.match(head) + if not match: + return False + try: + timestamp = float(match.group(1)) + limit = max_age + if match.group(2) is not None: + ttl = float(match.group(2)) + if ttl >= 0: + limit = ttl + except ValueError: + return False + return limit is not None and (now - timestamp) > limit + + if orjson is not None: # Encoding the cache record dominated the background fetch worker: on a # Pi 4, stdlib json.dumps runs ~12ms per MB and holds the GIL for all of @@ -266,6 +301,14 @@ class DiskCache: try: with self._lock: with open(cache_path, 'rb') as f: + # Decide staleness from the header before paying for the + # parse. A stale read is the common case for the biggest + # records (a season schedule is re-fetched when its cache + # expires), and parsing 53MB to throw it away held the GIL + # for ~1.8s -- a visible freeze on the panel. + if _stale_from_head(f.read(_HEAD_BYTES), max_age, time.time()): + return None + f.seek(0) record = _loads(f.read()) # Determine record timestamp (prefer embedded, else file mtime) diff --git a/src/cache_manager.py b/src/cache_manager.py index 4d80c986..d064a37f 100644 --- a/src/cache_manager.py +++ b/src/cache_manager.py @@ -522,8 +522,9 @@ class CacheManager: def update_cache(self, data_type: str, data: Dict[str, Any]) -> bool: """Update cache with new data.""" cache_data = { + # Header first; see DiskCache's stale check. + 'timestamp': time.time(), 'data': data, - 'timestamp': time.time() } return self.save_cache(data_type, cache_data) @@ -556,12 +557,15 @@ class CacheManager: from the key and is only a fallback for entries that did not say. Omit it to keep that inferred behaviour. """ - cache_data = { - 'data': data, - 'timestamp': time.time() - } + # timestamp and ttl before data, so they are the first bytes on disk: + # DiskCache.get reads them from the head of the file and can call a + # record stale without parsing it. That matters for the big ones -- a + # whole MLB season is 53MB and ~1.8s of orjson.loads with the GIL held, + # paid in full only to learn the record had expired. + cache_data: Dict[str, Any] = {'timestamp': time.time()} if ttl is not None: cache_data['ttl'] = ttl + cache_data['data'] = data self.save_cache(key, cache_data) @deprecated("3.7.0") diff --git a/src/common/espn_dates.py b/src/common/espn_dates.py index 9be705d1..bda7e86a 100644 --- a/src/common/espn_dates.py +++ b/src/common/espn_dates.py @@ -39,6 +39,14 @@ from datetime import date, timedelta from functools import partial from typing import Any, Dict, List, Optional, Tuple +try: + from src.common.json_body import response_json +except ImportError: + # Plugins bundle copies of this module for older cores, which predate + # json_body; the stdlib parse is what those cores always used. + def response_json(response: Any) -> Any: + return response.json() + # Above this, ESPN returns a truncated list instead of an error. See module # docstring: 500 is the largest value measured to return complete data. ESPN_MAX_LIMIT = 500 @@ -194,7 +202,7 @@ def _fetch_one_chunk( timeout=timeout, ) response.raise_for_status() - return response.json() + return response_json(response) except Exception as exc: # noqa: BLE001 - see docstring if logger: logger.warning("ESPN chunk %s failed, skipping it: %s", chunk, exc) @@ -371,4 +379,4 @@ def fetch_espn_scoreboard( if data is not None: return data response.raise_for_status() - return response.json() + return response_json(response) diff --git a/src/common/json_body.py b/src/common/json_body.py new file mode 100644 index 00000000..fb89301d --- /dev/null +++ b/src/common/json_body.py @@ -0,0 +1,29 @@ +"""Parse an HTTP response body as JSON, with orjson when it is installed. + +``requests``' ``response.json()`` uses the stdlib parser. For the payloads the +sports plugins fetch -- a season schedule is tens of MB -- that runs ~1.7x +slower than orjson on a Pi 4 (3.1s against 1.8s for the 53MB MLB season), and +both hold the GIL for the whole parse, which freezes the display for as long. +Nothing else changes: the result is the same Python objects. +""" + +from __future__ import annotations + +from typing import Any + +try: + import orjson +except ImportError: # optional dependency; see docs/SCROLL_PERFORMANCE.md + orjson = None + + +def response_json(response: Any) -> Any: + """``response.json()``, parsed by orjson when available.""" + body = getattr(response, "content", None) + if orjson is None or not isinstance(body, (bytes, bytearray)): + return response.json() + try: + return orjson.loads(body) + except orjson.JSONDecodeError: + # Let requests raise its usual error, with its usual message. + return response.json() diff --git a/src/common/sports_card.py b/src/common/sports_card.py index b2b7d563..2c5c88fb 100644 --- a/src/common/sports_card.py +++ b/src/common/sports_card.py @@ -339,6 +339,18 @@ def format_game_date(config: Optional[Dict[str, Any]], logger, date_text: str, if not raw: return "" fmt = str(scroll_card_option(config, "date_format", "abbrev") or "abbrev") + return _format_date_as(fmt, raw, lambda: weekday_for(config, logger, game)) + + +def _format_date_as(fmt: str, raw: str, weekday, months=MONTH_ABBR) -> str: + """Render a stripped, non-empty "M/D" *raw* in style *fmt*. + + The body both date formatters share. They differ in which setting names the + style and in which zone the weekday is taken from (see + ``SportsCoreSharedMixin._format_game_date``), so those arrive as arguments: + *weekday* is a zero-argument callable, only called for the "weekday" style. + *months* lets the mixin keep reading its (overridable) ``_MONTH_ABBR``. + """ if fmt == "numeric": return raw parts = raw.replace("-", "/").split("/") @@ -347,14 +359,14 @@ def format_game_date(config: Optional[Dict[str, Any]], logger, date_text: str, month, day = int(parts[0]), int(parts[1]) if not 1 <= month <= 12: return raw - name = MONTH_ABBR[month - 1] + name = months[month - 1] if fmt == "numeric_day_first": return f"{day}/{month}" if fmt == "day_first": return f"{day} {name}" if fmt == "weekday": - weekday = weekday_for(config, logger, game) - return f"{weekday} {name} {day}" if weekday else f"{name} {day}" + day_name = weekday() + return f"{day_name} {name} {day}" if day_name else f"{name} {day}" return f"{name} {day}" @@ -388,6 +400,29 @@ def format_game_time(config: Optional[Dict[str, Any]], time_text: str) -> str: _SCHEMA_FONT_SIZE_CACHE: Dict[str, Dict[str, int]] = {} +def _read_schema_font_sizes(schema_path: str) -> Dict[str, int]: + """``{element: font_size default}`` from a config_schema.json. Raises. + + The parse both schema-default lookups share. Each keeps its own cache -- + this function per schema path, ``SportsCoreSharedMixin._schema_font_size`` + per class -- because the lifetimes differ: a class is rebuilt when the + display service reloads a plugin, a module-level path cache is not. One + cache would change when a reloaded plugin sees an edited schema. + """ + import json + with open(schema_path) as fh: + schema = json.load(fh) + props = (schema.get('properties', {}) + .get('customization', {}) + .get('properties', {})) + sizes: Dict[str, int] = {} + for key, spec in props.items(): + size = spec.get('properties', {}).get('font_size', {}).get('default') + if size is not None: + sizes[key] = int(size) + return sizes + + def schema_font_size(schema_path: str, element_key) -> Optional[int]: """The font_size this plugin's config_schema.json declares, or None. @@ -399,18 +434,8 @@ def schema_font_size(schema_path: str, element_key) -> Optional[int]: return None cache = _SCHEMA_FONT_SIZE_CACHE.get(schema_path) if cache is None: - cache = {} try: - import json - with open(schema_path) as fh: - schema = json.load(fh) - props = (schema.get('properties', {}) - .get('customization', {}) - .get('properties', {})) - for key, spec in props.items(): - size = spec.get('properties', {}).get('font_size', {}).get('default') - if size is not None: - cache[key] = int(size) + cache = _read_schema_font_sizes(schema_path) except Exception as exc: # See sports_shared._schema_font_size: an unreadable schema # silently disables the pixel-grid snap for every element. @@ -444,7 +469,7 @@ def resolve_font_size(schema_path: str, element_config, element_key, return crisp_size(font_name, default_size, aliases, grid_table) -def unshare_element_fonts(logger, fonts): +def unshare_element_fonts(logger, fonts, element_for_font=None): """Give each colourable element its own face object. The colour a draw gets is resolved from the face it was handed, and @@ -459,13 +484,21 @@ def unshare_element_fonts(logger, fonts): the ability to tell two elements apart does. Faces that cannot be rebuilt (a BDF loaded through freetype.Face, anything without a usable path) are left shared, and their draws stay white as before. + + *element_for_font* names the font keys to consider, in order (the first + holder of a face keeps it); it defaults to this module's + :data:`ELEMENT_FOR_FONT`. ``SportsCoreSharedMixin`` passes its own map, + which names different keys -- see ``resolve_font_color`` for why the two + vocabularies are kept apart. """ try: from src.common.font_layout import load_truetype as _load except ImportError: # pragma: no cover return fonts + if element_for_font is None: + element_for_font = ELEMENT_FOR_FONT seen = {} - for key in ELEMENT_FOR_FONT: + for key in element_for_font: font = fonts.get(key) if font is None: continue diff --git a/src/common/sports_shared.py b/src/common/sports_shared.py index 8ac40a3a..e9112ec6 100644 --- a/src/common/sports_shared.py +++ b/src/common/sports_shared.py @@ -64,14 +64,27 @@ live here. Only ``_SCORE_PROBE_TEXT`` varies -- afl and basketball reach three d a side and override it, the same two that override ``_SCORE_PROBE`` on ``SportsGameRendererMixin``. -DELIBERATELY NOT MERGED WITH sports_card ----------------------------------------- -Fourteen of these have same-named twins in ``src/common/sports_card.py``, which -the scoreboards' ``game_renderer.py`` already uses. They are NOT wired together -here. Only five are provably equivalent by source comparison; the other nine -differ in ways inspection cannot settle, and a wrong guess silently changes what -every scoreboard draws. Merging them needs differential testing against both -implementations, and is left for its own change. +TWINS IN sports_card +-------------------- +Many of these have same-named twins in ``src/common/sports_card.py``, which the +scoreboards' ``game_renderer.py`` uses. ``test/test_sports_twins.py`` calls +each pair with the same inputs (the plugins' fixture games in every payload +shape, plus edge cases) and splits them in two: + +- Identical: ``_card_option``, ``_vs_text``, ``_format_game_time``, + ``_coerce_rgb``, ``_crisp_size``, ``_unshare_element_fonts`` (given the same + element map) and the constant tables. These are now thin wrappers over the + ``sports_card`` function; ``_format_game_date`` and ``_schema_font_size`` + share its body/parser while keeping their own setting, zone and cache. + ``_resolve_font_size`` agrees too but keeps its body, because it dispatches + through the overridable ``_schema_font_size``/``_crisp_size``. +- Different, and pinned as they are: ``_side_is_favorite`` / + ``_favorite_result`` / ``_recent_score_color`` (flat keys and the host's + favourites only), ``_weekday_for`` (the plugin's resolved zone, not + ``config["timezone"]``), ``_font_color`` / ``_ELEMENT_FOR_FONT`` (another + element vocabulary), ``_element_color`` (passes ``SKIN_MODE``). Each shows + up in one display mode only, so which side is right is a product decision; + the test that pins it names the difference. """ from __future__ import annotations @@ -87,6 +100,7 @@ import pytz from src.common.espn_dates import fetch_espn_scoreboard import requests from PIL import Image, ImageDraw, ImageFont +from src.common import sports_card as _card from src.common.font_layout import load_truetype logger = logging.getLogger(__name__) @@ -171,19 +185,17 @@ class SportsCoreSharedMixin: _ELEMENT_FOR_FONT: ClassVar[Dict[str, str]] = { "score": "score_text", "time": "period_text", "team": "team_text", "detail": "detail_text", "status": "status_text"} + # The tables below are sports_card's (and font_layout's) values. The dicts + # are copies, so a caller that mutates one module's table -- or a subclass + # that replaces it -- does not reach into the other. #: Default tint for a favourite team's finished game. - FAVORITE_RESULT_COLOR_DEFAULTS: ClassVar[Dict[str, Tuple[int, int, int]]] = { - "win": (0, 255, 0), "loss": (255, 0, 0), "tie": (255, 200, 0)} - _MONTH_ABBR: ClassVar[Tuple[str, ...]] = ( - "Jan", "Feb", "Mar", "Apr", "May", "Jun", - "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") - _WEEKDAY_ABBR: ClassVar[Tuple[str, ...]] = ( - "Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") + FAVORITE_RESULT_COLOR_DEFAULTS: ClassVar[Dict[str, Tuple[int, int, int]]] = dict( + _card.FAVORITE_RESULT_COLOR_DEFAULTS) + _MONTH_ABBR: ClassVar[Tuple[str, ...]] = _card.MONTH_ABBR + _WEEKDAY_ABBR: ClassVar[Tuple[str, ...]] = _card.WEEKDAY_ABBR #: Bitmap fonts snap to their native pixel grid. - _FONT_PIXEL_GRID: ClassVar[Dict[str, int]] = { - "PressStart2P-Regular.ttf": 8, "4x6-font.ttf": 7} - _FONT_NAME_ALIASES: ClassVar[Dict[str, str]] = { - "press_start": "PressStart2P-Regular.ttf", "four_by_six": "4x6-font.ttf"} + _FONT_PIXEL_GRID: ClassVar[Dict[str, int]] = dict(_card.FONT_PIXEL_GRID) + _FONT_NAME_ALIASES: ClassVar[Dict[str, str]] = dict(_card.FONT_NAME_ALIASES) #: Accepted values for the other-games quality filter. _QUALITY_CHOICES: ClassVar[frozenset] = frozenset({"any", "ranked"}) #: How long to stay quiet between ranking-coverage warnings. @@ -213,13 +225,11 @@ class SportsCoreSharedMixin: """Snap *desired* to the nearest size *font_file* renders crisply at. A face with no known grid is returned unchanged, so a user-supplied - font is never second-guessed. + font is never second-guessed. The class's own tables are passed, so a + host that declares extra faces keeps them. """ - font_file = cls._FONT_NAME_ALIASES.get(font_file, font_file) - grid = cls._FONT_PIXEL_GRID.get(font_file) - if not grid or not desired or desired <= 0: - return desired - return max(grid, int(round(float(desired) / grid)) * grid) + return _card.crisp_size(font_file, desired, + cls._FONT_NAME_ALIASES, cls._FONT_PIXEL_GRID) #: Absolute path of this plugin's directory, declared by the plugin #: itself. The mixin cannot work it out -- see _plugin_dir. @@ -281,23 +291,19 @@ class SportsCoreSharedMixin: """The font_size this plugin's config_schema.json declares, or None.""" if not element_key: return None + # Cached per class, not in sports_card's per-path cache: the display + # service rebuilds the class when it reloads a plugin, and that is + # what makes an edited schema take effect. Both caches parse through + # sports_card._read_schema_font_sizes. cache = getattr(self.__class__, '_SCHEMA_FONT_SIZES', None) if cache is None: cache = {} try: - import json directory = self._plugin_dir() if directory is None: raise FileNotFoundError("no config_schema.json on the MRO") - with open(os.path.join(directory, 'config_schema.json')) as fh: - schema = json.load(fh) - props = (schema.get('properties', {}) - .get('customization', {}) - .get('properties', {})) - for key, spec in props.items(): - size = spec.get('properties', {}).get('font_size', {}).get('default') - if size is not None: - cache[key] = int(size) + cache = _card._read_schema_font_sizes( + os.path.join(directory, 'config_schema.json')) except Exception as exc: # Say so. An unreadable schema is not cosmetic: every element's # configured size then stops matching "the schema default", is @@ -339,10 +345,7 @@ class SportsCoreSharedMixin: def _card_option(self, key: str, default: Any = None) -> Any: """Read one key from the scroll_card config block.""" - block = (self.config or {}).get("scroll_card") - if isinstance(block, dict) and block.get(key) is not None: - return block.get(key) - return default + return _card.scroll_card_option(self.config, key, default) def _switch_upcoming_center(self) -> str: """Middle of the full-screen upcoming scorebug: 'vs', 'date_time' or 'none'.""" @@ -354,7 +357,7 @@ class SportsCoreSharedMixin: def _vs_text(self) -> str: """Separator drawn between the teams -- "VS", "@", "at", anything.""" - return str(self._card_option("vs_text", "VS")) + return _card.vs_text(self.config) def _switch_date_format(self) -> str: """Date style for the full-screen scorebug. @@ -374,28 +377,19 @@ class SportsCoreSharedMixin: return fmt def _format_game_date(self, date_text: str, game: Optional[Dict] = None) -> str: - """Format an upcoming date per scroll_card.switch_date_format.""" + """Format an upcoming date per scroll_card.switch_date_format. + + The formatting is sports_card's. What differs from the card's + ``format_game_date`` is passed in: the setting (``switch_date_format``, + see :meth:`_switch_date_format`) and the weekday, which comes from + :meth:`_weekday_for` and so from this plugin's resolved timezone. + """ raw = str(date_text or "").strip() if not raw: return raw - fmt = self._switch_date_format() - if fmt == "numeric": - return raw - parts = raw.replace("-", "/").split("/") - if not (len(parts) >= 2 and parts[0].strip().isdigit() and parts[1].strip().isdigit()): - return raw - month, day = int(parts[0]), int(parts[1]) - if not 1 <= month <= 12: - return raw - name = self._MONTH_ABBR[month - 1] - if fmt == "numeric_day_first": - return f"{day}/{month}" - if fmt == "day_first": - return f"{day} {name}" - if fmt == "weekday": - weekday = self._weekday_for(game) - return f"{weekday} {name} {day}" if weekday else f"{name} {day}" - return f"{name} {day}" + return _card._format_date_as(self._switch_date_format(), raw, + lambda: self._weekday_for(game), + self._MONTH_ABBR) def _weekday_for(self, game: Optional[Dict]) -> str: """Weekday abbreviation from the game's start time, or ''.""" @@ -413,22 +407,7 @@ class SportsCoreSharedMixin: def _format_game_time(self, time_text: str) -> str: """Return the time as-is (12h) or converted to 24h.""" - raw = str(time_text or "").strip() - if not raw or str(self._card_option("time_format", "12h")) != "24h": - return raw - cleaned = raw.upper().replace(" ", "") - meridiem = "AM" if cleaned.endswith("AM") else "PM" if cleaned.endswith("PM") else "" - if not meridiem: - return raw - try: - hh, _, mm = cleaned[:-2].partition(":") - hour, minute = int(hh), int(mm or 0) - except ValueError: - return raw - if not (0 <= hour <= 12 and 0 <= minute <= 59): - return raw - hour = hour % 12 + (12 if meridiem == "PM" else 0) - return f"{hour:02d}:{minute:02d}" + return _card.format_game_time(self.config, time_text) def _scorebug_font(self, draw, text: str, width: int): """The face this scorebug draws its date and time in. @@ -560,15 +539,7 @@ class SportsCoreSharedMixin: @staticmethod def _coerce_rgb(value, fallback): """Turn a configured [R, G, B] list into a clamped (r, g, b) tuple.""" - # Checked before unpacking: a 3-character string ("123") would otherwise - # iterate into three digits and yield a colour rather than the fallback. - if not isinstance(value, (list, tuple)) or len(value) != 3: - return fallback - try: - r, g, b = (max(0, min(255, int(channel))) for channel in value) - except (TypeError, ValueError): - return fallback - return (r, g, b) + return _card.coerce_rgb(value, fallback) @staticmethod def _side_is_favorite(game: Dict, side: str, favorites: set) -> bool: @@ -849,28 +820,12 @@ class SportsCoreSharedMixin: the ability to tell two elements apart does. Faces that cannot be rebuilt (a BDF loaded through freetype.Face, anything without a usable path) are left shared, and their draws stay white as before. + + The body is sports_card's; this class's own element map is passed, so + the keys considered are the ones this class colours by. """ - try: - from src.common.font_layout import load_truetype as _load - except ImportError: # pragma: no cover - return fonts - seen = {} - for key in self._ELEMENT_FOR_FONT: - font = fonts.get(key) - if font is None: - continue - if id(font) not in seen: - seen[id(font)] = key - continue - path, size = getattr(font, "path", None), getattr(font, "size", None) - if not path or not size: - continue - try: - fonts[key] = _load(path, size) - except (OSError, ValueError, TypeError): - self.logger.debug( - "Could not un-share the %s face; it keeps the default colour", key) - return fonts + return _card.unshare_element_fonts(self.logger, fonts, + self._ELEMENT_FOR_FONT) def _font_color(self, font, default: Tuple[int, int, int] = (255, 255, 255)): """Colour for whichever element owns this face. diff --git a/src/redaction.py b/src/redaction.py index b84bf1b1..a1294309 100644 --- a/src/redaction.py +++ b/src/redaction.py @@ -24,8 +24,13 @@ _REDACT_CREDENTIAL = re.compile( # silently leak the ones nobody thought of. Not covered by the generic pattern # above, whose value part stops at whitespace and so would keep the credential # once a space follows the scheme. +# +# The opening quote and the whitespace after it are one optional unit. Written +# `\s*["\']?\s*`, a whitespace run with no quote in it could be split between +# the two `\s*` in every possible way, and a header with no credential after +# it tried them all: quadratic, 8s for 20k spaces. _REDACT_AUTH_HEADER = re.compile( - r'((?:proxy-)?authorization["\']?\s*[=:]\s*["\']?\s*' + r'((?:proxy-)?authorization["\']?\s*[=:]\s*(?:["\']\s*)?' r'(?:[A-Za-z][\w.+-]*[ \t]+)?)' # optional scheme name, kept r'([^\s,"\'<>}]+)', # the credential, redacted re.IGNORECASE, @@ -34,8 +39,16 @@ _REDACT_AUTH_HEADER = re.compile( # Credentials embedded in a URL: https://user:password@host. requests quotes # the full URL in its exceptions, so this is a realistic leak. The username is # kept -- it identifies which account failed without being the secret. -_REDACT_URL_USERINFO = re.compile(r'([a-z][a-z0-9+.-]*://[^/\s:@]+:)([^/\s@]+)(@)', - re.IGNORECASE) +# +# A match may only start where a run of scheme characters starts. Unanchored, +# `[a-z][a-z0-9+.-]*://` was tried from every letter of a long run (a hex +# digest, an ID, a blob of response body), each attempt reading to the end of +# the run: quadratic, 1.6s for 20k characters, all of it holding the GIL. +# Leading digits and `+.-` sit inside group 1 so the substitution puts them +# back; the scheme proper still has to start with a letter. +_REDACT_URL_USERINFO = re.compile( + r'((? str: diff --git a/test/test_auto_update_verify.py b/test/test_auto_update_verify.py index c346091e..e491d42b 100644 --- a/test/test_auto_update_verify.py +++ b/test/test_auto_update_verify.py @@ -320,5 +320,6 @@ def test_units_installers_and_updater_agree(): assert (f'systemd/{unit}', f'/etc/systemd/system/{unit}') in StartupValidator._UNITS # Triggering takes no privilege any more; no sudoers rule should linger. - for sudoers in ('scripts/install/configure_web_sudo.sh', 'first_time_install.sh'): + for sudoers in ('scripts/install/configure_web_sudo.sh', 'first_time_install.sh', + 'scripts/install/lib_sudoers.sh'): assert not re.search(r'NOPASSWD:.*update-verify', (ROOT / sudoers).read_text(encoding='utf-8')), sudoers diff --git a/test/test_cache_stale_header.py b/test/test_cache_stale_header.py new file mode 100644 index 00000000..026b982d --- /dev/null +++ b/test/test_cache_stale_header.py @@ -0,0 +1,118 @@ +"""A stale cache record is recognised from its header, without parsing it. + +The sports plugins cache whole season schedules -- 53MB for MLB, 18MB for NHL. +When one expired, DiskCache.get parsed all of it (~1.8s of orjson.loads on a +Pi 4, GIL held, the whole display frozen) only to find the timestamp too old +and throw the result away. CacheManager.set now writes timestamp and ttl ahead +of the data, and DiskCache.get reads them from the first bytes of the file. +""" + +import json +import time +from types import SimpleNamespace + +import pytest + +from src.cache import disk_cache as disk_cache_module +from src.cache.disk_cache import DiskCache, _stale_from_head +from src.common import json_body + + +@pytest.fixture +def disk(tmp_path): + return DiskCache(cache_dir=str(tmp_path)) + + +@pytest.fixture +def parses(monkeypatch): + """Count full parses of cache files.""" + calls = [] + real = disk_cache_module._loads + + def counting(raw): + calls.append(len(raw)) + return real(raw) + + monkeypatch.setattr(disk_cache_module, "_loads", counting) + return calls + + +def _header_first(age=0.0, ttl=None, events=100): + record = {"timestamp": time.time() - age} + if ttl is not None: + record["ttl"] = ttl + record["data"] = {"events": [{"id": n, "name": "x" * 50} for n in range(events)]} + return record + + +def test_cache_manager_writes_the_header_first(monkeypatch): + from src.cache_manager import CacheManager + written = {} + manager = CacheManager.__new__(CacheManager) + monkeypatch.setattr(manager, "save_cache", + lambda key, record: written.update({key: record}), + raising=False) + CacheManager.set(manager, "k", {"events": []}, ttl=60) + assert list(written["k"]) == ["timestamp", "ttl", "data"] + CacheManager.set(manager, "k", {"events": []}) + assert list(written["k"]) == ["timestamp", "data"] + + +def test_a_stale_record_is_not_parsed(disk, parses): + disk.set("season", _header_first(age=600)) + assert disk.get("season", max_age=300) is None + assert parses == [] + + +def test_a_fresh_record_is_parsed_and_returned(disk, parses): + disk.set("season", _header_first(age=10)) + record = disk.get("season", max_age=300) + assert record["data"]["events"][0]["id"] == 0 + assert len(parses) == 1 + + +def test_the_entry_ttl_wins_over_max_age(disk, parses): + disk.set("long", _header_first(age=600, ttl=3600)) + assert disk.get("long", max_age=300) is not None # ttl says fresh + disk.set("short", _header_first(age=60, ttl=30)) + parses.clear() + assert disk.get("short", max_age=300) is None # ttl says stale + assert parses == [] + + +def test_no_limit_means_never_stale(disk): + disk.set("forever", _header_first(age=10 ** 7)) + assert disk.get("forever", max_age=None) is not None + + +def test_older_files_with_data_first_still_work(disk, parses): + # Records written before the header moved: parsed in full, as before. + disk.set("legacy_fresh", {"data": {"v": 1}, "timestamp": time.time()}) + disk.set("legacy_stale", {"data": {"v": 1}, "timestamp": time.time() - 600}) + assert disk.get("legacy_fresh", max_age=300)["data"] == {"v": 1} + assert disk.get("legacy_stale", max_age=300) is None + assert len(parses) == 2 + + +@pytest.mark.parametrize("head, stale", [ + (b'{"timestamp":100.0,"data":{}}', True), + (b'{"timestamp": 100.0, "ttl": 1000, "data": {}}', False), # stdlib spacing + (b'{"timestamp":1e2,"ttl":5,"data":1}', True), + (b'{"timestamp":100.0}', True), + (b'{"data":{},"timestamp":100.0}', False), # unknown layout + (b'{"timestamp":"100.0","data":{}}', False), # string: parse it + (b'', False), +]) +def test_reading_the_header(head, stale): + assert _stale_from_head(head, 300, now=1000.0) is stale + + +def test_response_json_prefers_orjson_and_falls_back(): + payload = {"events": [1, 2, 3]} + response = SimpleNamespace(content=json.dumps(payload).encode(), + json=lambda: pytest.fail("used the slow path")) + if json_body.orjson is None: + pytest.skip("orjson not installed") + assert json_body.response_json(response) == payload + # A response object without bytes content (a test double) still works. + assert json_body.response_json(SimpleNamespace(json=lambda: payload)) == payload diff --git a/test/test_redaction.py b/test/test_redaction.py new file mode 100644 index 00000000..5f59d696 --- /dev/null +++ b/test/test_redaction.py @@ -0,0 +1,110 @@ +"""redact_credentials must stay linear in the length of its input. + +Regressions under test, both quadratic regexes in src/redaction.py: + +- The URL-userinfo pattern (`scheme://user:password@`) could start a match at + every letter of a run of scheme characters, and each attempt read to the end + of the run looking for `://`: 1.6s for a 20k-character run. +- The Authorization-header pattern had two `\\s*` separated only by an + optional quote, so a header followed by whitespace and no credential tried + every split of that whitespace between them: 8s for 20k spaces. + +The display service redacts every message, stack trace and context value it +publishes in the error snapshot, and re.sub holds the GIL throughout, so an +exception quoting a hex digest or a long ID stalled the render loop with it. +test_error_snapshot_cross_process.py's snapshot-size test spent 140s here. + +The fixed patterns have to redact exactly what the old ones did. +""" + +import time + +import pytest + +from src.redaction import redact_credentials + +# Each timed input took seconds before the fix and takes about a millisecond +# after it; the bound leaves CI plenty of headroom while still failing on a +# quadratic pattern. +_TIME_LIMIT = 1.0 + + +def _timed(text): + start = time.perf_counter() + result = redact_credentials(text) + return result, time.perf_counter() - start + + +class TestUrlUserinfo: + @pytest.mark.parametrize("text,expected", [ + ("401 for https://user:hunter2@example.com/api", + "401 for https://user:@example.com/api"), + ("HTTPS://USER:HUNTER2@EXAMPLE.COM", + "HTTPS://USER:@EXAMPLE.COM"), + ("git+ssh://deploy:hunter2@host/repo", + "git+ssh://deploy:@host/repo"), + # The scheme starts after digits or +.- in the same run. Those + # characters must survive, and the password must still go. + ("1http://user:hunter2@host", "1http://user:@host"), + ("+.-http://user:hunter2@host", "+.-http://user:@host"), + ("a1+http://user:hunter2@host", "a1+http://user:@host"), + ("see a://u:first@b and c://v:second@d", + "see a://u:@b and c://v:@d"), + ]) + def test_password_is_redacted_and_the_rest_kept(self, text, expected): + assert redact_credentials(text) == expected + + def test_a_url_without_a_password_is_untouched(self): + text = "GET https://user@example.com/path failed" + assert redact_credentials(text) == text + + +class TestAuthorizationHeader: + @pytest.mark.parametrize("text,expected", [ + ("Authorization: Bearer eyJ.SECRET.sig", "Authorization: Bearer "), + ("Proxy-Authorization: Basic dXNlcg==", "Proxy-Authorization: Basic "), + ("authorization: barecredential", "authorization: "), + # Whitespace and an opening quote around the value, in either order. + ('authorization=" Bearer tok"', 'authorization=" Bearer "'), + ("authorization: ' tok'", "authorization: ' '"), + ("authorization:\n\tBearer tok", "authorization:\n\tBearer "), + ]) + def test_credential_is_redacted_and_the_rest_kept(self, text, expected): + assert redact_credentials(text) == expected + + @pytest.mark.parametrize("text", ["authorization: ", "authorization: , next"]) + def test_a_header_without_a_credential_is_untouched(self, text): + assert redact_credentials(text) == text + + +class TestLinearTime: + @pytest.mark.parametrize("unit", ["x", "0123456789abcdef", "1a", "a+", "1"]) + def test_long_scheme_character_runs(self, unit): + text = (unit * 50_000)[:50_000] + result, elapsed = _timed(text) + assert result == text + assert elapsed < _TIME_LIMIT, f"{elapsed:.2f}s to redact {len(text)} chars of {unit!r}" + + def test_a_credential_after_a_long_run_is_still_found(self): + run = "ab12" * 10_000 + result, elapsed = _timed(f"{run} https://user:hunter2@example.com") + assert result == f"{run} https://user:@example.com" + assert elapsed < _TIME_LIMIT + + @pytest.mark.parametrize("header,whitespace", [ + ("authorization:", " "), + ("Proxy-Authorization:", "\t"), + ("authorization=", "\n"), + ]) + def test_a_header_followed_by_long_whitespace(self, header, whitespace): + text = header + whitespace * 20_000 + "," + result, elapsed = _timed(text) + assert result == text + assert elapsed < _TIME_LIMIT, ( + f"{elapsed:.2f}s to redact {header!r} and {len(text) - len(header)} more chars") + + def test_a_credential_after_long_whitespace_is_still_found(self): + gap = " " * 20_000 + result, elapsed = _timed(f"authorization:{gap}Bearer tok") + assert result == f"authorization:{gap}Bearer " + assert elapsed < _TIME_LIMIT diff --git a/test/test_sports_twins.py b/test/test_sports_twins.py new file mode 100644 index 00000000..4202468d --- /dev/null +++ b/test/test_sports_twins.py @@ -0,0 +1,672 @@ +"""The twins: ``SportsCoreSharedMixin`` methods vs ``sports_card`` functions. + +Every scoreboard draws the same game twice over: switch mode through its +``sports.py`` (``SportsCoreSharedMixin``, ``self._recent_score_color(...)``) +and scroll/Vegas mode through its ``game_renderer.py`` +(``sports_card``, ``_card.recent_score_color(...)``). The two modules grew +same-named helpers independently, so this file calls each pair with the same +inputs and says which ones agree. + +Two kinds of test live here, and the difference matters: + +* ``TestIdentical`` -- pairs that agree on every input below. Most mixin + methods in this set are now thin wrappers over the ``sports_card`` function, + so the check is also what keeps a future "fix" to one side from quietly + becoming a divergence (a mixin body re-grown, a wrapper given different + arguments). +* ``TestPinnedDivergence`` -- pairs that do NOT agree. Their current behaviour + is pinned on purpose, with the minimal input that shows the difference. A + divergence here is user-visible (a colour, a weekday) in one display mode, and + which side is right is an owner decision, not a refactor. When that decision + is made, the test that pins it is the one to edit, deliberately. + +The game corpus is the plugins' own harness fixtures +(``plugins/*/test/fixtures/mock.json`` in ledmatrix-plugins), reduced to the +keys these helpers read and embedded below, in the three payload shapes the +helpers are handed: flat (what ``_extract_game_details_common`` builds -- the +switch-mode input), flat plus nested (what the renderers' +``_normalize_game_payload`` hands the scroll card), and nested only. Point +``LEDMATRIX_PLUGINS`` at a ledmatrix-plugins checkout to add every event in +those fixtures to the corpus. +""" + +import itertools +import json +import logging +import os +from datetime import datetime, timezone +from pathlib import Path +from zoneinfo import ZoneInfo + +import pytest + +from src.common import sports_card as C +from src.common.font_layout import load_truetype, resolve_asset_path +from src.common.sports_shared import SportsCoreSharedMixin + +LOG = logging.getLogger("test_sports_twins") + +# --------------------------------------------------------------------------- +# Corpus +# --------------------------------------------------------------------------- + +#: (plugin, start, home abbr, home id, home score, away abbr, away id, +#: away score, state) -- one row per distinct event in the eight scoreboards' +#: test/fixtures/mock.json. +FIXTURE_EVENTS = [ + ("afl", "2026-07-10T09:40Z", "COLL", "17", "89", "NMFC", "5", "85", "post"), + ("afl", "2026-07-11T03:15Z", "STK", "18", "0", "PORT", "7", "0", "pre"), + ("afl", "2026-07-10T11:30Z", "FRE", "1", "54", "SYD", "4", "48", "in"), + ("baseball", "2026-07-09T02:10Z", "LAD", "19", "5", "SF", "26", "3", "post"), + ("baseball", "2026-07-10T10:05Z", "NYY", "10", "4", "BOS", "2", "3", "in"), + ("baseball", "2026-07-11T23:10Z", "NYM", "21", "0", "ATL", "15", "0", "pre"), + ("basketball", "2026-01-14T00:30Z", "BOS", "2", "112", "NY", "18", "104", "post"), + ("basketball", "2026-01-15T03:30Z", "LAL", "13", "78", "GS", "9", "72", "in"), + ("basketball", "2026-01-16T02:00Z", "DEN", "7", "0", "DAL", "6", "0", "pre"), + ("football", "2026-01-14T01:15Z", "KC", "12", "27", "BUF", "2", "24", "post"), + ("football", "2026-01-15T10:30Z", "PHI", "21", "17", "DAL", "6", "14", "in"), + ("football", "2026-01-18T23:30Z", "DET", "8", "0", "GB", "9", "0", "pre"), + ("hockey", "2026-01-14T00:00Z", "BOS", "1", "4", "TOR", "21", "2", "post"), + ("hockey", "2026-01-15T10:30Z", "TB", "20", "3", "DAL", "9", "2", "in"), + ("hockey", "2026-01-16T00:00Z", "CHI", "4", "0", "NYR", "13", "0", "pre"), + ("lacrosse", "2026-04-14T18:00Z", "DUKE", "150", "14", "SYR", "183", "11", "post"), + ("lacrosse", "2026-04-15T10:30Z", "JHU", "2305", "8", "UVA", "258", "7", "in"), + ("lacrosse", "2026-04-16T22:00Z", "COR", "172", "0", "PSU", "213", "0", "pre"), + ("nrl", "2026-07-10T09:00Z", "BRI", "16", "18", "PEN", "18", "12", "in"), + ("nrl", "2026-07-09T09:00Z", "MEL", "12", "24", "SYD", "20", "10", "post"), + ("nrl", "2026-07-12T09:00Z", "PAR", "14", "0", "PEN", "18", "0", "pre"), + ("soccer", "2026-01-14T20:00Z", "ARS", "359", "2", "CHE", "363", "1", "post"), + ("soccer", "2026-01-15T11:30Z", "LIV", "364", "1", "MNC", "382", "1", "in"), + ("soccer", "2026-01-16T20:00Z", "TOT", "367", "0", "MAN", "360", "0", "pre"), +] + +_LEAGUE = {"afl": "afl", "baseball": "mlb", "basketball": "nba", "football": "nfl", + "hockey": "nhl", "lacrosse": "ncaa_mens_lacrosse", "nrl": "nrl", + "soccer": "eng.1"} + + +def _extra_fixture_events(): + """Every event in a ledmatrix-plugins checkout, when one is named.""" + raw = os.environ.get("LEDMATRIX_PLUGINS") + if not raw: + return [] + root = Path(raw) + if (root / "plugins").is_dir(): + root = root / "plugins" + rows = [] + for path in sorted(root.glob("*-scoreboard/test/fixtures/mock.json")): + plugin = path.parts[-4].replace("-scoreboard", "") + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + for block in data.values(): + for ev in (block.get("events") or []) if isinstance(block, dict) else []: + try: + comp = ev["competitions"][0] + sides = {c["homeAway"]: c for c in comp["competitors"]} + rows.append((plugin, ev["date"], + sides["home"]["team"]["abbreviation"], + sides["home"]["team"]["id"], sides["home"].get("score"), + sides["away"]["team"]["abbreviation"], + sides["away"]["team"]["id"], sides["away"].get("score"), + "")) + except (KeyError, IndexError, TypeError): + continue + return rows + + +def _flat(row): + plugin, start, ha, hid, hs, aa, aid, as_, _state = row + return { + "league": _LEAGUE.get(plugin, plugin), + "home_abbr": ha, "home_id": hid, "home_score": hs, + "away_abbr": aa, "away_id": aid, "away_score": as_, + "start_time_utc": datetime.fromisoformat(start.replace("Z", "+00:00")), + } + + +def _with_nested(game): + """The scroll card's input: flat keys kept, nested team dicts added.""" + out = dict(game) + for side in ("home", "away"): + out[f"{side}_team"] = {"abbrev": game.get(f"{side}_abbr"), + "id": game.get(f"{side}_id"), + "score": game.get(f"{side}_score")} + return out + + +def _nested_only(game): + out = {k: v for k, v in _with_nested(game).items() + if not k.startswith(("home_abbr", "home_id", "home_score", + "away_abbr", "away_id", "away_score"))} + return out + + +_ROWS = FIXTURE_EVENTS + _extra_fixture_events() +FLAT_GAMES = [_flat(r) for r in _ROWS] +EDGE_FLAT_GAMES = [ + # NRL: abbreviations are not unique, ids are. + {"league": "nrl", "home_abbr": "NEW", "home_id": "4", "home_score": "20", + "away_abbr": "NEW", "away_id": "12", "away_score": "10"}, + {"league": "nfl", "home_abbr": "KC", "away_abbr": "BUF"}, + {"league": "nfl", "home_abbr": "KC", "away_abbr": "BUF", "home_score": "", "away_score": ""}, + {"league": "nfl", "home_abbr": "KC", "away_abbr": "BUF", "home_score": "-", "away_score": "-"}, + {"league": "nfl", "home_abbr": "KC", "away_abbr": "BUF", "home_score": "5.0", + "away_score": "2.0"}, + {"league": "nfl", "home_abbr": "KC", "home_id": 12, "away_abbr": "BUF", "away_id": 2, + "home_score": 3, "away_score": 3}, + {"league": "nfl", "home_abbr": None, "away_abbr": "BUF", "home_score": "1", + "away_score": "2"}, + {"league": "nfl", "home_abbr": " kc ", "away_abbr": "BUF", "home_score": "1", + "away_score": "2"}, +] +ALL_FLAT = FLAT_GAMES + EDGE_FLAT_GAMES +SCROLL_SHAPED = [_with_nested(g) for g in ALL_FLAT] + + +def _favorite_choices(game): + """Every way a favourites list can relate to this game.""" + out = [[], ["AP_TOP_25"], ["NOBODY"]] + for side in ("home", "away"): + for key in ("abbr", "id"): + value = game.get(f"{side}_{key}") + if value is not None: + out.append([str(value)]) + out.append([" " + str(value).lower() + " "]) + if game.get("home_abbr") and game.get("away_abbr"): + out.append([game["home_abbr"], game["away_abbr"]]) + return out + + +# --------------------------------------------------------------------------- +# Hosts +# --------------------------------------------------------------------------- + +class _Host(SportsCoreSharedMixin): + """The mixin with just the state these helpers read.""" + + def __init__(self, config=None, favorites=None, tz=timezone.utc, fonts=None): + self.config = config + self.favorite_teams = favorites + self.logger = LOG + self.fonts = fonts or {} + self._tz = tz + + def _get_timezone(self): + return self._tz + + +#: The map seven of the eight scoreboards' sports.py declare over the mixin's +#: default (football is the one that inherits the default). +PLUGIN_ELEMENT_FOR_FONT = { + "odds": "odds_text", "score": "score_text", "time": "period_text", + "team": "team_name", "status": "status_text", "detail": "detail_text", + "rank": "rank_text", +} + + +def _call(fn, *args): + """Result or the exception type, so a raise on one side is a difference.""" + try: + return fn(*args) + except Exception as exc: # noqa: BLE001 - the type is the result here + return f"" + + +def _mismatches(pairs): + return [(label, a, b) for label, a, b in pairs if a != b] + + +RESULT_COLOURS = [ + {"enabled": True}, + {"enabled": True, "win_color": [1, 2, 3], "loss_color": "123", + "tie_color": [999, -1, "7"]}, + {"enabled": False}, + {}, +] + +SCROLL_CARD_CONFIGS = [ + None, {}, {"scroll_card": None}, {"scroll_card": {}}, {"scroll_card": "notadict"}, + {"scroll_card": {"vs_text": "@", "date_format": "weekday", "time_format": "24h", + "switch_date_format": "inherit"}}, + {"scroll_card": {"vs_text": None, "date_format": "day_first", "time_format": "12h", + "switch_date_format": "inherit"}}, + {"scroll_card": {"vs_text": 7, "date_format": "numeric", "switch_date_format": "inherit"}}, + {"scroll_card": {"date_format": "numeric_day_first", "time_format": "24h", + "switch_date_format": "inherit"}}, + {"scroll_card": {"date_format": "abbrev", "switch_date_format": "inherit"}}, + {"scroll_card": {"date_format": "bogus", "switch_date_format": "inherit"}}, +] +TIMES = ["7:30 PM", "12:00 AM", "12:05pm", "7 PM", "13:00 PM", "TBD", "", None, + "7:61 PM", "x:30 PM", " 9:05 am ", "12:00 PM", "0:15 AM"] +DATES = ["9/19", "09-19", "13/19", "Sep 19", "", None, "9/19/2026", " 1/2 ", "0/5"] +STARTS = [datetime(2026, 9, 19, 23, 30, tzinfo=timezone.utc), "2026-09-19T23:30Z", + "2026-09-20T02:00:00+00:00", "garbage", None, "", datetime(2026, 1, 1)] +TIMEZONES = ["America/New_York", "Australia/Sydney", "UTC", "Not/AZone", None] + +PS = resolve_asset_path("assets/fonts/PressStart2P-Regular.ttf") +F46 = resolve_asset_path("assets/fonts/4x6-font.ttf") + + +def _font_sets(): + a, b, c = load_truetype(PS, 8), load_truetype(PS, 16), load_truetype(F46, 7) + keys = ("odds", "score", "time", "team", "status", "detail", "rank") + return { + "distinct": {"score": a, "time": b, "team": c, "status": load_truetype(PS, 8), + "detail": load_truetype(F46, 7), "rank": load_truetype(F46, 14), + "odds": load_truetype(F46, 7)}, + "score+time share": {"score": a, "time": a, "team": c}, + "team+rank share": {"score": a, "time": b, "team": c, "rank": c}, + "odds+score share": {"score": a, "odds": a, "time": b}, + "all share": {k: a for k in keys}, + } + + +def _partition(fonts): + """Which keys still share one face object -- what unsharing decides.""" + groups = {} + for key, font in fonts.items(): + groups.setdefault(id(font), []).append(key) + return sorted(sorted(keys) for keys in groups.values()) + + +def _faces(fonts): + return {k: (getattr(f, "path", None), getattr(f, "size", None)) for k, f in fonts.items()} + + +def _schema_dir(tmp_path, name, text): + d = tmp_path / name + d.mkdir() + if text is not None: + (d / "config_schema.json").write_text(text) + return d + + +SCHEMAS = { + "good": json.dumps({"properties": {"customization": {"properties": { + "score_text": {"properties": {"font_size": {"default": 10}}}, + "period_text": {"properties": {"font_size": {"default": 8}}}, + "detail_text": {"properties": {"font_size": {"default": "6"}}}, + "team_name": {"properties": {"font": {"default": "x"}}}}}}}), + "bad_json": "{not json", + "bad_default": json.dumps({"properties": {"customization": {"properties": { + "score_text": {"properties": {"font_size": {"default": "big"}}}}}}}), + "missing": None, +} + + +# --------------------------------------------------------------------------- +# Identical pairs +# --------------------------------------------------------------------------- + +class TestIdentical: + """Pairs that agree on every input. Keep it that way.""" + + def test_scroll_card_option(self): + pairs = [] + for i, cfg in enumerate(SCROLL_CARD_CONFIGS): + host = _Host(cfg) + for key, default in itertools.product( + ("vs_text", "date_format", "time_format", "missing"), (None, "D", 0)): + pairs.append((f"cfg{i} {key} {default!r}", + _call(host._card_option, key, default), + _call(C.scroll_card_option, cfg, key, default))) + assert not _mismatches(pairs) + + def test_vs_text(self): + pairs = [(f"cfg{i}", _call(_Host(cfg)._vs_text), _call(C.vs_text, cfg)) + for i, cfg in enumerate(SCROLL_CARD_CONFIGS)] + assert not _mismatches(pairs) + + def test_format_game_time(self): + pairs = [(f"cfg{i} {t!r}", _call(_Host(cfg)._format_game_time, t), + _call(C.format_game_time, cfg, t)) + for (i, cfg), t in itertools.product(enumerate(SCROLL_CARD_CONFIGS), TIMES)] + assert not _mismatches(pairs) + + def test_coerce_rgb(self): + values = [[1, 2, 3], (300, -4, "5"), "123", [1, 2], [1, 2, 3, 4], None, 42, + {"r": 1, "g": 2, "b": 3}, ["a", 1, 2], [1.9, 2, 3], [None, 1, 2]] + pairs = [(repr(v), _call(_Host._coerce_rgb, v, (4, 5, 6)), + _call(C.coerce_rgb, v, (4, 5, 6))) for v in values] + assert not _mismatches(pairs) + + def test_crisp_size(self): + names = ["PressStart2P-Regular.ttf", "4x6-font.ttf", "press_start", "four_by_six", + "5by7.regular.ttf", "user.ttf", None] + sizes = [None, 0, -3, 1, 4, 6, 7, 8, 9, 10, 11, 12, 13, 14, 16, 20, 7.5, "8"] + pairs = [(f"{n} {s!r}", _call(_Host._crisp_size, n, s), _call(C.crisp_size, n, s)) + for n, s in itertools.product(names, sizes)] + assert not _mismatches(pairs) + + def test_crisp_size_honours_a_hosts_own_tables(self): + """A class that declares extra faces keeps them through the wrapper.""" + cls = type("Extra", (_Host,), {"_FONT_PIXEL_GRID": {"extra.ttf": 5}, + "_FONT_NAME_ALIASES": {"x": "extra.ttf"}}) + assert cls._crisp_size("x", 12) == C.crisp_size("x", 12, {"x": "extra.ttf"}, + {"extra.ttf": 5}) == 10 + + def test_constant_tables(self): + assert SportsCoreSharedMixin.FAVORITE_RESULT_COLOR_DEFAULTS == \ + C.FAVORITE_RESULT_COLOR_DEFAULTS + assert SportsCoreSharedMixin._MONTH_ABBR == C.MONTH_ABBR + assert SportsCoreSharedMixin._WEEKDAY_ABBR == C.WEEKDAY_ABBR + assert SportsCoreSharedMixin._FONT_PIXEL_GRID == C.FONT_PIXEL_GRID + assert SportsCoreSharedMixin._FONT_NAME_ALIASES == C.FONT_NAME_ALIASES + + def test_constant_dicts_are_not_aliased(self): + # Equal, but separate objects: a caller mutating one table (tests do) + # must not reach into the other module. + assert SportsCoreSharedMixin.FAVORITE_RESULT_COLOR_DEFAULTS is not \ + C.FAVORITE_RESULT_COLOR_DEFAULTS + assert SportsCoreSharedMixin._FONT_PIXEL_GRID is not C.FONT_PIXEL_GRID + assert SportsCoreSharedMixin._FONT_NAME_ALIASES is not C.FONT_NAME_ALIASES + + @pytest.mark.parametrize("schema", sorted(SCHEMAS)) + def test_schema_font_size_and_resolve_font_size(self, tmp_path, schema): + d = _schema_dir(tmp_path, schema, SCHEMAS[schema]) + host = type("H_" + schema, (_Host,), {"_PLUGIN_DIR": str(d)})() + path = str(d / "config_schema.json") + pairs = [] + for key in ("score_text", "period_text", "detail_text", "team_name", "nope", "", None): + pairs.append((f"schema {key!r}", _call(host._schema_font_size, key), + _call(C.schema_font_size, path, key))) + for ec, name, size in itertools.product( + (None, {}, {"font_size": 10}, {"font_size": "10"}, {"font_size": 11}, + {"font_size": "big"}, {"font_size": None}, {"font_size": 8.7}), + ("PressStart2P-Regular.ttf", "4x6-font.ttf", "press_start", "user.ttf"), + (6, 8, 10, None)): + pairs.append((f"resolve {key!r} {ec} {name} {size}", + _call(host._resolve_font_size, ec, key, size, name), + _call(C.resolve_font_size, path, ec, key, size, name))) + assert not _mismatches(pairs) + + @pytest.mark.parametrize("element_map", ["mixin default", "plugin sports.py"]) + def test_unshare_element_fonts_given_the_same_map(self, element_map): + """Same map in, same faces out. (The maps themselves differ; pinned below.)""" + mapping = (SportsCoreSharedMixin._ELEMENT_FOR_FONT if element_map == "mixin default" + else PLUGIN_ELEMENT_FOR_FONT) + host = type("H", (_Host,), {"_ELEMENT_FOR_FONT": mapping})() + for name, fonts in _font_sets().items(): + mine, theirs = dict(fonts), dict(fonts) + host._unshare_element_fonts(mine) + C.unshare_element_fonts(LOG, theirs, mapping) + assert _partition(mine) == _partition(theirs), name + assert _faces(mine) == _faces(theirs), name + + def test_unshare_element_fonts_default_map_is_unchanged(self): + """Omitting the new argument keeps the card's own map.""" + for name, fonts in _font_sets().items(): + default, explicit = dict(fonts), dict(fonts) + C.unshare_element_fonts(LOG, default) + C.unshare_element_fonts(LOG, explicit, C.ELEMENT_FOR_FONT) + assert _partition(default) == _partition(explicit), name + + def test_format_game_date_when_both_read_the_same_setting_and_zone(self): + """With ``switch_date_format: inherit`` the scorebug reads the card's + ``date_format``; given the same zone the two then format identically.""" + pairs = [] + for (i, cfg), tzname in itertools.product(enumerate(SCROLL_CARD_CONFIGS[5:]), + TIMEZONES): + conf = dict(cfg, timezone=tzname) if tzname else dict(cfg) + host = _Host(conf, tz=C.card_tzinfo(conf, LOG)) + for d, start in itertools.product(DATES, STARTS): + game = {"start_time_utc": start} if start is not None else {} + pairs.append((f"cfg{i} tz={tzname} {d!r} {start!r}", + _call(host._format_game_date, d, game), + _call(C.format_game_date, conf, LOG, d, game))) + pairs.append((f"weekday cfg{i} tz={tzname} {start!r}", + _call(host._weekday_for, game), + _call(C.weekday_for, conf, LOG, game))) + assert not _mismatches(pairs) + + def test_format_game_date_honours_a_hosts_month_table(self): + """The scoreboards redeclare _MONTH_ABBR; the shared body must read it.""" + cls = type("Months", (_Host,), {"_MONTH_ABBR": tuple(f"M{i}" for i in range(1, 13))}) + host = cls({"scroll_card": {"switch_date_format": "abbrev"}}) + assert host._format_game_date("9/19") == "M9 19" + + def test_favorite_result_on_the_games_the_scoreboards_build(self): + """Production shape: the extractor stamps ``favorite_teams`` (the + manager's resolved list) on every game, and the manager holds the same + list. On those games -- flat for switch mode, flat plus nested for the + scroll card -- the two sides agree on every result and every colour.""" + pairs = [] + for game in ALL_FLAT: + for favs in _favorite_choices(game): + stamped = dict(game, favorite_teams=list(favs)) + for colours in RESULT_COLOURS: + cfg = {"customization": {"favorite_result_colors": colours}} + host = _Host(cfg, favorites=list(favs)) + pairs.append((f"{game} {favs}", + _call(host._favorite_result, stamped), + _call(C.favorite_result, cfg, _with_nested(stamped)))) + pairs.append((f"{game} {favs} {colours}", + _call(host._recent_score_color, stamped, (9, 9, 9)), + _call(C.recent_score_color, cfg, LOG, + _with_nested(stamped), (9, 9, 9)))) + assert not _mismatches(pairs) + # And the corpus is not vacuous: every verdict actually occurs. + verdicts = {a for _, a, _ in pairs if isinstance(a, str) or a is None} + assert {"win", "loss", "tie", None} <= verdicts + + def test_side_is_favorite_on_flat_games(self): + pairs = [] + for game in ALL_FLAT: + for favs in _favorite_choices(game): + fav_set = {str(f).strip().upper() for f in favs if str(f).strip()} + for side in ("home", "away"): + pairs.append((f"{game} {side} {fav_set}", + _call(_Host._side_is_favorite, game, side, fav_set), + _call(C.side_is_favorite, game, side, fav_set))) + assert not _mismatches(pairs) + + def test_nrl_collision_is_resolved_the_same_way_on_both_sides(self): + """The _favorite_key seam: NRL's "NEW" is two clubs. Neither helper + calls the seam; both match abbreviation OR id, so an id favourite picks + one club and an abbreviation favourite picks both (no verdict).""" + game = EDGE_FLAT_GAMES[0] + for favs, expected in ((["4"], "win"), (["12"], "loss"), (["NEW"], None)): + host = _Host({}, favorites=favs) + assert host._favorite_result(game) == expected + assert C.favorite_result({"favorite_teams": favs}, game) == expected + + def test_an_ambiguous_nrl_abbreviation_tints_on_both_sides(self): + """Agreed -- and at odds with NRL's own favourite rule. + + NRL's resolver logs a shared abbreviation ("NEW") as an error and + passes it through unchanged, and its _is_favorite_game matches ids + only, so selection never treats "NEW" as a favourite. Both colour + helpers match on abbreviation too, so both modes tint a Knights result + (and a Warriors one) for a user who typed "NEW". Not a twin + divergence; a seam neither helper consults. + """ + on = {"customization": {"favorite_result_colors": {"enabled": True}}} + game = {"league": "3", "home_abbr": "NEW", "home_id": "4", "home_score": "20", + "away_abbr": "MEL", "away_id": "12", "away_score": "10", + "favorite_teams": ["NEW"]} + cfg = dict(on, favorite_teams=["NEW"]) + assert _Host(cfg, favorites=["NEW"])._recent_score_color(game, (9, 9, 9)) \ + == C.recent_score_color(cfg, LOG, game, (9, 9, 9)) == (0, 255, 0) + + +# --------------------------------------------------------------------------- +# Pinned divergences -- owner decision pending. Edit deliberately. +# --------------------------------------------------------------------------- + +class TestPinnedDivergence: + """Each test pins one difference between the twins as it stands today. + + None of these is changed by the consolidation that made the identical pairs + wrappers: each one is a colour, a weekday or a font face that one display + mode shows differently from the other, so choosing a side is a product + decision. If you are here because one of these failed, you changed which + side wins -- make sure that was the decision, then update the pin. + """ + + NESTED_WIN = {"league": "nhl", + "home_team": {"abbrev": "TB", "score": "4"}, + "away_team": {"abbrev": "BOS", "score": "1"}} + + def test_side_is_favorite_nested_payload(self): + # DIVERGENCE: the mixin reads only the flat _abbr / _id keys; + # the card also reads _team.{abbrev,abbreviation,id}. Unreachable + # from the scoreboards' own extractors (always flat), reachable from a + # nested-only payload. + assert _Host._side_is_favorite(self.NESTED_WIN, "home", {"TB"}) is False + assert C.side_is_favorite(self.NESTED_WIN, "home", {"TB"}) is True + + def test_favorite_result_nested_payload(self): + # DIVERGENCE: follows from the one above, plus score source: the mixin + # reads home_score/away_score only; the card prefers the nested score. + host = _Host({}, favorites=["TB"]) + game = dict(self.NESTED_WIN, favorite_teams=["TB"]) + assert host._favorite_result(game) is None + assert C.favorite_result({}, game) == "win" + + def test_favorite_result_when_nested_and_flat_scores_disagree(self): + # DIVERGENCE: same game, two score sources. The mixin uses the flat + # score, the card the nested one. The renderers' normaliser only fills + # a nested score that is missing, so this needs a payload that already + # carried both. + game = {"home_abbr": "TB", "away_abbr": "BOS", "home_score": "1", + "away_score": "4", "home_team": {"abbrev": "TB", "score": "4"}, + "away_team": {"abbrev": "BOS", "score": "1"}, "favorite_teams": ["TB"]} + assert _Host({}, favorites=["TB"])._favorite_result(game) == "loss" + assert C.favorite_result({}, game) == "win" + + def test_favorite_result_favourite_sources(self): + # DIVERGENCE: where the favourites come from. The mixin reads only + # self.favorite_teams (the manager's list, resolved at construction); + # the card reads the game's stamped favorite_teams plus the config's + # league block (or root). All eight scoreboards stamp the game, so in + # production both see the same list -- this pins the hand-built case. + game = {"league": "mlb", "home_abbr": "ATL", "away_abbr": "NYM", + "home_score": "5", "away_score": "2"} + on = {"customization": {"favorite_result_colors": {"enabled": True}}} + # Host favourites only, nothing stamped, nothing in config: + assert _Host(on, favorites=["ATL"])._favorite_result(game) == "win" + assert C.favorite_result(on, game) is None + # Config league block only, host list empty: + cfg = dict(on, mlb={"favorite_teams": ["ATL"]}) + assert _Host(cfg, favorites=[])._favorite_result(game) is None + assert C.favorite_result(cfg, game) == "win" + # Stamped on the game only, host list empty: + stamped = dict(game, favorite_teams=["ATL"]) + assert _Host(on, favorites=[])._favorite_result(stamped) is None + assert C.favorite_result(on, stamped) == "win" + # ...which is what reaches the colour: + assert _Host(on, favorites=["ATL"])._recent_score_color(game, (9, 9, 9)) == (0, 255, 0) + assert C.recent_score_color(on, LOG, game, (9, 9, 9)) == (9, 9, 9) + + def test_weekday_zone_source(self): + # DIVERGENCE, user-visible: the scorebug asks the plugin's + # _get_timezone() (plugin setting -> global setting -> system zone); + # the card reads only config["timezone"] and falls back to UTC. The + # scoreboards' schemas default that key to "", and the scroll display + # hands the renderer the plugin config, so a board that sets only the + # global zone gets UTC weekdays in scroll mode: an evening kickoff in + # New York is labelled with the next day. + game = {"start_time_utc": "2026-09-20T00:30:00+00:00"} # Sat 20:30 EDT + host = _Host({}, tz=ZoneInfo("America/New_York")) + assert host._weekday_for(game) == "Sat" + assert C.weekday_for({}, LOG, game) == "Sun" + cfg = {"scroll_card": {"date_format": "weekday", "switch_date_format": "inherit"}} + host = _Host(cfg, tz=ZoneInfo("America/New_York")) + assert host._format_game_date("9/19", game) == "Sat Sep 19" + assert C.format_game_date(cfg, LOG, "9/19", game) == "Sun Sep 19" + + def test_weekday_out_of_range_start(self): + # DIVERGENCE: the mixin catches OverflowError from astimezone() and + # drops the weekday; the card lets it escape to its caller. + game = {"start_time_utc": "9999-12-31T23:59:00+00:00"} + sydney = {"timezone": "Australia/Sydney"} + assert _Host(sydney, tz=ZoneInfo("Australia/Sydney"))._weekday_for(game) == "" + with pytest.raises(OverflowError): + C.weekday_for(sydney, LOG, game) + + def test_date_format_setting(self): + # DIVERGENCE BY DESIGN (documented on _switch_date_format): the scorebug + # reads scroll_card.switch_date_format (default "numeric", the "9/19" + # it has always drawn); the card reads scroll_card.date_format (default + # "abbrev"). "inherit" opts the scorebug into the card's setting. + assert _Host({})._format_game_date("9/19") == "9/19" + assert C.format_game_date({}, LOG, "9/19") == "Sep 19" + + def test_upcoming_centre_setting(self): + # DIVERGENCE BY DESIGN: not a same-named twin, but the same question. + # switch_upcoming_center defaults to "date_time"; the card's + # upcoming_center to "vs". "inherit" opts the scorebug in. + assert _Host({})._switch_upcoming_center() == "date_time" + assert C.upcoming_center_mode({}) == "vs" + cfg = {"scroll_card": {"switch_upcoming_center": "inherit"}} + assert _Host(cfg)._switch_upcoming_center() == C.upcoming_center_mode(cfg) == "vs" + + def test_element_for_font_maps(self): + # DIVERGENCE: the element vocabulary. The mixin default says team_text + # and has no rank/odds; the card says team_name and has rank but no + # odds. Seven scoreboards override the mixin map in sports.py with + # PLUGIN_ELEMENT_FOR_FONT (team_name, rank, odds); football inherits + # the default, and its schema declares team_name, not team_text. + assert SportsCoreSharedMixin._ELEMENT_FOR_FONT == { + "score": "score_text", "time": "period_text", "team": "team_text", + "detail": "detail_text", "status": "status_text"} + assert C.ELEMENT_FOR_FONT == { + "score": "score_text", "time": "period_text", "team": "team_name", + "status": "status_text", "detail": "detail_text", "rank": "rank_text"} + + def test_font_color_team_element(self): + # DIVERGENCE (consequence of the maps): a colour set on team_name + # reaches the card's team face but not the mixin-default one. + team = load_truetype(F46, 7) + fonts = {"score": load_truetype(PS, 8), "team": team} + cfg = {"customization": {"team_name": {"text_color": [1, 1, 1]}}} + assert _Host(cfg, fonts=fonts)._font_color(team, (7, 7, 7)) == (7, 7, 7) + assert C.font_color(cfg, fonts, team, (7, 7, 7)) == (1, 1, 1) + plugin_host = type("P", (_Host,), {"_ELEMENT_FOR_FONT": PLUGIN_ELEMENT_FOR_FONT}) + assert plugin_host(cfg, fonts=fonts)._font_color(team, (7, 7, 7)) == (1, 1, 1) + + def test_unshare_element_fonts_odds_face(self): + # DIVERGENCE (consequence of the maps): the scoreboards' sports.py map + # includes "odds", so switch mode gives the odds face its own object; + # the card's map has no "odds", so scroll mode leaves it sharing the + # score's face (and _card.font_color then colours it as score_text). + shared = load_truetype(PS, 8) + mine = {"score": shared, "odds": shared} + theirs = dict(mine) + type("P", (_Host,), {"_ELEMENT_FOR_FONT": PLUGIN_ELEMENT_FOR_FONT})() \ + ._unshare_element_fonts(mine) + C.unshare_element_fonts(LOG, theirs) + assert mine["odds"] is not mine["score"] + assert theirs["odds"] is theirs["score"] + + def test_schema_default_cache_lifetimes(self, tmp_path): + # DELIBERATE, and the reason there are still two caches: the mixin + # caches per class, the card per schema path. The display service + # builds new classes when it reloads a plugin, so switch mode picks up + # an edited schema then; the card's module-level cache does not. One + # shared cache would change what switch mode does after a reload. + d = _schema_dir(tmp_path, "reload", SCHEMAS["good"]) + path = str(d / "config_schema.json") + first = type("First", (_Host,), {"_PLUGIN_DIR": str(d)})() + assert first._schema_font_size("score_text") == 10 + assert C.schema_font_size(path, "score_text") == 10 + (d / "config_schema.json").write_text(SCHEMAS["good"].replace("10", "16")) + reloaded = type("Reloaded", (_Host,), {"_PLUGIN_DIR": str(d)})() + assert reloaded._schema_font_size("score_text") == 16 + assert first._schema_font_size("score_text") == 10 + assert C.schema_font_size(path, "score_text") == 10 + + def test_element_color_mode(self): + # DIVERGENCE at the call site, not in a body: both resolve through + # src.element_style, but the mixin passes the instance's SKIN_MODE + # ("live"/"recent"/"upcoming", set by all eight scoreboards) and the + # renderers call _card.element_color with no mode, so a per-mode colour + # override applies in switch mode only. + cfg = {"customization": {"score_text": {"text_color": [255, 0, 0]}, + "modes": {"recent": {"score_text": {"text_color": [0, 0, 255]}}}}} + host = _Host(cfg) + host.SKIN_MODE = "recent" + assert host._element_color("score_text") == (0, 0, 255) + assert C.element_color(cfg, "score_text") == (255, 0, 0) diff --git a/test/test_sudo_allowlist_covers_calls.py b/test/test_sudo_allowlist_covers_calls.py index bec2486b..8f3b88f7 100644 --- a/test/test_sudo_allowlist_covers_calls.py +++ b/test/test_sudo_allowlist_covers_calls.py @@ -37,6 +37,9 @@ ROOT = Path(__file__).resolve().parent.parent INSTALLERS = ( ROOT / "first_time_install.sh", ROOT / "scripts" / "install" / "configure_wifi_permissions.sh", + # The ledmatrix_web rules, which first_time_install.sh and + # configure_web_sudo.sh both take from here. + ROOT / "scripts" / "install" / "lib_sudoers.sh", ) #: Commands this change grants, each fully literal in the source. diff --git a/test/test_sudoers_is_validated.py b/test/test_sudoers_is_validated.py index a29c5779..a65bbd0e 100644 --- a/test/test_sudoers_is_validated.py +++ b/test/test_sudoers_is_validated.py @@ -62,33 +62,135 @@ def test_configure_web_sudo_validates_before_installing(): assert validate < install, "the rules must be checked before they are installed" -def _render_first_time_sudoers(project_root, user): - """Run the installer's own sudoers heredoc with realistic values.""" +def test_a_missing_rules_library_installs_nothing(): + """If lib_sudoers.sh is missing, nothing is generated -- and an empty file + would pass `visudo -c` -- so that branch must set the flag the install is + gated on.""" body = _read(FIRST_TIME) - start = body.index("# Create sudoers content") + missing = body.index('if [ -f "$SUDOERS_LIB" ]; then') + flagged = body.index("SUDOERS_VALID=0", missing) + validate = body.index('visudo -c -f "$SUDOERS_TMP"') + install = body.index('cp "$SUDOERS_TMP" "$SUDOERS_FILE"') + gate = body.rindex('if [ "$SUDOERS_VALID" = "0" ]; then', 0, install) + assert missing < flagged < validate < gate < install + + +def _step10_generation(body): + """first_time_install.sh's own Step 10 code that writes $SUDOERS_TMP.""" + start = body.index("# The rules themselves live in scripts/install/lib_sudoers.sh") end = body.index("# Never install rules we have not parsed.") - block = body[start:end] - out = os.path.join(project_root, "rendered") + return body[start:end] + + +def _run_step10_generation(project_root, user, out): + """Run the installer's Step 10 generation with realistic values. + + Returns the SUDOERS_VALID it leaves behind.""" script = "\n".join( [ - "set -euo pipefail", + "set -Eeuo pipefail", f"ACTUAL_USER={user}", - f"PROJECT_ROOT_DIR={project_root}", - 'SUDOERS_TMP="$(mktemp)"', - "PYTHON_PATH=$(which python3)", + f"PROJECT_ROOT_DIR='{project_root}'", + f"SUDOERS_TMP='{out}'", + "SUDOERS_FILE=/etc/sudoers.d/ledmatrix_web", "SYSTEMCTL_PATH=/usr/bin/systemctl", "REBOOT_PATH=/usr/sbin/reboot", "POWEROFF_PATH=/usr/sbin/poweroff", "BASH_PATH=$(which bash)", "JOURNALCTL_PATH=/usr/bin/journalctl", - block, - f'cp "$SUDOERS_TMP" {out}', + _step10_generation(_read(FIRST_TIME)), + 'printf %s "$SUDOERS_VALID"', ] ) - subprocess.run(["bash", "-c", script], check=True) + return subprocess.run( + ["bash", "-c", script], check=True, capture_output=True, text=True + ).stdout + + +def _render_first_time_sudoers(tmp, user): + """The rules first_time_install.sh generates, via the shared library.""" + out = os.path.join(tmp, "rendered") + assert _run_step10_generation(REPO_ROOT, user, out) == "1" return out +def _run_step10(tmp, project_root, visudo_ok, existing=None): + """Run all of Step 10 against a sudoers file in `tmp`, never /etc. + + systemctl, reboot, poweroff, journalctl and visudo are stubs, so the + outcome does not depend on the machine running the test.""" + body = _read(FIRST_TIME) + step = body[body.index('CURRENT_STEP="Configure passwordless sudo access"'): + body.index('CURRENT_STEP="Configure WiFi management permissions"')] + target = os.path.join(tmp, "ledmatrix_web") + real = 'SUDOERS_FILE="/etc/sudoers.d/ledmatrix_web"' + assert step.count(real) == 1 + step = step.replace(real, f"SUDOERS_FILE='{target}'") + stubs = os.path.join(tmp, "stubs") + os.mkdir(stubs) + for name, code in (("systemctl", 0), ("reboot", 0), ("poweroff", 0), + ("journalctl", 0), ("visudo", 0 if visudo_ok else 1)): + path = os.path.join(stubs, name) + with open(path, "w", encoding="utf-8") as handle: + handle.write(f"#!/bin/sh\nexit {code}\n") + os.chmod(path, 0o755) + if existing is not None: + with open(target, "w", encoding="utf-8") as handle: + handle.write(existing) + env = dict(os.environ, TMPDIR=tmp, + PATH=os.pathsep.join([stubs, os.path.dirname(sys.executable), + "/usr/bin", "/bin"])) + script = "\n".join(["set -Eeuo pipefail", "ACTUAL_USER=ledmatrix", + f"PROJECT_ROOT_DIR='{project_root}'", step]) + result = subprocess.run(["bash", "-c", script], env=env, + capture_output=True, text=True) + assert result.returncode == 0, result.stdout + result.stderr + return target, stubs, result + + +_POSIX_STEP10 = pytest.mark.skipif( + sys.platform == "win32" or shutil.which("which") is None, + reason="needs a POSIX bash and which") + + +@_POSIX_STEP10 +def test_step10_installs_the_generated_rules(): + with tempfile.TemporaryDirectory() as tmp: + target, stubs, _ = _run_step10(tmp, REPO_ROOT, visudo_ok=True) + assert oct(os.stat(target).st_mode & 0o777) == "0o440" + with open(target, encoding="utf-8") as handle: + installed = handle.read() + lib = os.path.join(REPO_ROOT, "scripts", "install", "lib_sudoers.sh") + expected = subprocess.run( + ["bash", "-c", '. "$1"; web_sudoers_rules ledmatrix "$2" "$3/systemctl" ' + '"$(command -v bash)" "$3/reboot" "$3/poweroff" "$3/journalctl"', + "_", lib, REPO_ROOT, stubs], + check=True, capture_output=True, text=True, + env=dict(os.environ, PATH=os.pathsep.join([stubs, "/usr/bin", "/bin"])), + ).stdout + assert installed == expected + assert not [f for f in os.listdir(tmp) if f.startswith("ledmatrix_web_sudoers.")] + + +@_POSIX_STEP10 +def test_step10_without_the_library_keeps_the_existing_file(): + with tempfile.TemporaryDirectory() as tmp: + target, _, result = _run_step10(tmp, tmp, visudo_ok=True, existing="keep\n") + with open(target, encoding="utf-8") as handle: + assert handle.read() == "keep\n" + assert "lib_sudoers.sh not found" in result.stderr + assert "Passwordless sudo access configured" not in result.stdout + + +@_POSIX_STEP10 +def test_step10_keeps_the_existing_file_when_the_rules_do_not_parse(): + with tempfile.TemporaryDirectory() as tmp: + target, _, result = _run_step10(tmp, REPO_ROOT, visudo_ok=False, existing="keep\n") + with open(target, encoding="utf-8") as handle: + assert handle.read() == "keep\n" + assert "did not parse" in result.stderr + + @pytest.mark.skipif(sys.platform == "win32", reason="visudo is POSIX only") @pytest.mark.skipif(VISUDO is None, reason="visudo not installed") def test_the_rules_the_installer_emits_actually_parse(): diff --git a/test/test_sudoers_noexec_on_pagers.py b/test/test_sudoers_noexec_on_pagers.py index ca657785..1f3983cf 100644 --- a/test/test_sudoers_noexec_on_pagers.py +++ b/test/test_sudoers_noexec_on_pagers.py @@ -30,10 +30,14 @@ ROOT = Path(__file__).resolve().parent.parent INSTALLERS = ( ROOT / "first_time_install.sh", ROOT / "scripts" / "install" / "configure_wifi_permissions.sh", - # Writes the same journalctl grants as first_time_install.sh. It was - # missing here, and because of that this suite passed while three - # ungranted wildcard rules sat in it. + # Used to write its own copy of the journalctl grants. It was missing + # here, and because of that this suite passed while three untagged + # wildcard rules sat in it. Both it and first_time_install.sh now take + # their rules from lib_sudoers.sh; they stay listed so a rule written + # directly into either one is still checked. ROOT / "scripts" / "install" / "configure_web_sudo.sh", + # The ledmatrix_web rules, shared by both installers. + ROOT / "scripts" / "install" / "lib_sudoers.sh", ) #: Commands that will start another program of their own accord -- a pager, an diff --git a/test/test_web_sudoers_installers_agree.py b/test/test_web_sudoers_installers_agree.py index 7a6d10d7..248a5520 100644 --- a/test/test_web_sudoers_installers_agree.py +++ b/test/test_web_sudoers_installers_agree.py @@ -1,53 +1,189 @@ -"""The two installers that write /etc/sudoers.d/ledmatrix_web must agree. +"""One generator writes /etc/sudoers.d/ledmatrix_web, and both installers use it. -first_time_install.sh (Step 10, a heredoc) and scripts/install/configure_web_sudo.sh -(a block of echo lines) each generate the web user's sudo allow-list. They -drifted: configure_web_sudo.sh granted scripts/fix_perms/safe_pip_install.sh -but first_time_install.sh did not, so on a device set up only by the first-time +first_time_install.sh (Step 10) and scripts/install/configure_web_sudo.sh each +used to carry their own copy of the web user's sudo allow-list -- a heredoc in +one, a block of echo lines in the other -- and the copies drifted: +configure_web_sudo.sh granted scripts/fix_perms/safe_pip_install.sh but +first_time_install.sh did not, so on a device set up only by the first-time installer permission_utils.install_requirements_file could not use the root wrapper and fell back to a user-level install that root-run ledmatrix.service may not see (and the auto-update rollback reported its reinstall as failed). -This compares the granted command sets after normalising the spellings that -differ between the files but expand identically at install time: -$WEB_USER/$ACTUAL_USER, $PROJECT_ROOT/$PROJECT_ROOT_DIR, and the helper-path -variables configure_web_sudo.sh defines ($SAFE_RM_PATH, ...). +The rules now live once, in web_sudoers_rules() in +scripts/install/lib_sudoers.sh. What keeps them from drifting again: + +* neither installer writes a rule line of its own, and each writes the + generator's output to the very file it then validates and installs; +* each passes its variables to the generator in the right positions -- checked + by running the installer's own call line with distinct values; +* the generator's grants are pinned to an explicit list below, so dropping, + adding or re-pathing a grant is a deliberate edit to this file. It also checks that every fix_perms helper granted via sudo is hardened to -root:root in both scripts -- and, in first_time_install.sh, after Step 11's +root:root in both installers -- and, in first_time_install.sh, after Step 11's project-wide chown to the user, which would otherwise undo it. """ import re +import shutil +import subprocess +import sys from pathlib import Path +import pytest + ROOT = Path(__file__).resolve().parent.parent FIRST_TIME = ROOT / "first_time_install.sh" CONFIGURE = ROOT / "scripts" / "install" / "configure_web_sudo.sh" +LIB = ROOT / "scripts" / "install" / "lib_sudoers.sh" -#: Grants that intentionally exist in only one installer, as normalised -#: commands. There are none today; add one here with a reason rather than -#: loosening the comparison. -ONLY_IN_FIRST_TIME = frozenset() -ONLY_IN_CONFIGURE = frozenset() +#: Every grant web_sudoers_rules() writes, as (tags, command) with the +#: generator's own variable names. Changing the allow-list means changing this. +EXPECTED_GRANTS = frozenset({ + ("NOPASSWD:", "$REBOOT_PATH"), + ("NOPASSWD:", "$POWEROFF_PATH"), + ("NOPASSWD:", "$SYSTEMCTL_PATH start ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH stop ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH restart ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH enable ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH disable ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH status ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH is-active ledmatrix"), + ("NOPASSWD:", "$SYSTEMCTL_PATH is-active ledmatrix.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH start ledmatrix-web.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH stop ledmatrix-web.service"), + ("NOPASSWD:", "$SYSTEMCTL_PATH restart ledmatrix-web.service"), + ("NOPASSWD:", "$BASH_PATH $PROJECT_ROOT/scripts/fix_perms/safe_plugin_rm.sh *"), + ("NOPASSWD:", "$BASH_PATH $PROJECT_ROOT/scripts/fix_perms/safe_pip_install.sh *"), + ("NOPASSWD:NOEXEC:", "$JOURNALCTL_PATH -u ledmatrix.service *"), + ("NOPASSWD:NOEXEC:", "$JOURNALCTL_PATH -u ledmatrix *"), + ("NOPASSWD:NOEXEC:", "$JOURNALCTL_PATH -t ledmatrix *"), +}) + +#: The call each installer makes: its own names for the generator's arguments, +#: in order, and the file it writes the rules to. +CALLERS = { + FIRST_TIME: (("$ACTUAL_USER", "$PROJECT_ROOT_DIR", "$SYSTEMCTL_PATH", "$BASH_PATH", + "$REBOOT_PATH", "$POWEROFF_PATH", "$JOURNALCTL_PATH"), "$SUDOERS_TMP"), + CONFIGURE: (("$WEB_USER", "$PROJECT_ROOT", "$SYSTEMCTL_PATH", "$BASH_PATH", + "$REBOOT_PATH", "$POWEROFF_PATH", "$JOURNALCTL_PATH"), "$TEMP_SUDOERS"), +} + +RULE = re.compile(r'(\S+) ALL=\(ALL\) (NOPASSWD:(?:NOEXEC:)?)\s*(.*?)"?$') def _text(path): - return path.read_text(encoding="utf-8", errors="replace") + return path.read_text(encoding="utf-8", errors="replace").replace("\r\n", "\n") -def _web_sudoers_section(path): - """The part of the script that writes the ledmatrix_web allow-list. +def _generator_grants(): + """{(tags, command)} for every rule line in lib_sudoers.sh.""" + grants = set() + for line in _text(LIB).splitlines(): + m = RULE.search(line.strip()) + if m and m.group(1).endswith("$WEB_USER"): + grants.add((m.group(2), " ".join(m.group(3).split()))) + return grants - first_time_install.sh also writes other files later (WiFi permissions are - delegated to a separate script, but keep this robust against future - additions), so restrict it to Step 10. - """ + +def _call(path): + """The installer's web_sudoers_rules statement, continuation lines joined.""" text = _text(path) - if path == FIRST_TIME: - start = text.index('CURRENT_STEP="Configure passwordless sudo access"') - end = text.index('CURRENT_STEP="Configure WiFi management permissions"') - return text[start:end] - return text + calls = re.findall(r"^[ \t]*web_sudoers_rules\b(?:[^\n]*\\\n)*[^\n]*$", text, re.M) + assert len(calls) == 1, f"{path.name}: expected one web_sudoers_rules call, found {calls}" + return calls[0] + + +def test_generator_grants_exactly_the_expected_rules(): + grants = _generator_grants() + assert grants == EXPECTED_GRANTS, ( + f"lib_sudoers.sh grants changed:\n added: {sorted(grants - EXPECTED_GRANTS)}\n" + f" removed: {sorted(EXPECTED_GRANTS - grants)}") + + +@pytest.mark.parametrize("installer", [FIRST_TIME, CONFIGURE], ids=lambda p: p.name) +def test_installer_writes_no_rules_of_its_own(installer): + """A rule added to one installer only is how they drifted last time.""" + own = [line for line in _text(installer).splitlines() + if "NOPASSWD" in line and not line.lstrip().startswith("#")] + assert not own, f"{installer.name} writes sudoers rules itself: {own}" + + +@pytest.mark.parametrize("installer", [FIRST_TIME, CONFIGURE], ids=lambda p: p.name) +def test_installer_sources_the_generator_and_writes_what_it_validates(installer): + text = _text(installer) + assert "lib_sudoers.sh" in text, f"{installer.name} does not source lib_sudoers.sh" + args, target = CALLERS[installer] + call = _call(installer) + words = call.replace("\\\n", " ").split() + assert words[0] == "web_sudoers_rules" + assert tuple(w.strip('"') for w in words[1:8]) == args, ( + f"{installer.name} passes the generator's arguments out of order: {call}") + assert words[8:] == [">", f'"{target}"'], call + # ...and that file is the one it runs visudo on. + assert f'visudo -c -f "{target}"' in text + + +@pytest.mark.skipif(sys.platform == "win32" or shutil.which("bash") is None, + reason="needs a POSIX bash") +@pytest.mark.parametrize("installer", [FIRST_TIME, CONFIGURE], ids=lambda p: p.name) +def test_installer_call_renders_the_expected_rules(installer, tmp_path): + """Run the installer's own call line, with a distinct value per argument.""" + args, target = CALLERS[installer] + values = { + args[0]: "webuser", args[1]: "/srv/led root", args[2]: "/x/systemctl", + args[3]: "/x/bash", args[4]: "/x/reboot", args[5]: "/x/poweroff", + args[6]: "/x/journalctl", target: str(tmp_path / "out"), + } + assigns = "\n".join(f"{name[1:]}='{value}'" for name, value in values.items()) + script = f"set -euo pipefail\n. '{LIB}'\n{assigns}\n{_call(installer)}\n" + subprocess.run(["bash", "-c", script], check=True) + rendered = set() + for line in (tmp_path / "out").read_text(encoding="utf-8").splitlines(): + m = RULE.match(line) + if m: + assert m.group(1) == "webuser", line + rendered.add((m.group(2), m.group(3))) + subst = {"$SYSTEMCTL_PATH": "/x/systemctl", "$BASH_PATH": "/x/bash", + "$REBOOT_PATH": "/x/reboot", "$POWEROFF_PATH": "/x/poweroff", + "$JOURNALCTL_PATH": "/x/journalctl", "$PROJECT_ROOT": "/srv/led root"} + expected = set() + for tags, command in EXPECTED_GRANTS: + for var, value in subst.items(): + command = command.replace(var, value) + expected.add((tags, command)) + assert rendered == expected + + +@pytest.mark.skipif(sys.platform == "win32" or shutil.which("bash") is None, + reason="needs a POSIX bash") +def test_optional_tools_are_left_out_when_absent(tmp_path): + """configure_web_sudo.sh passes "" for a missing reboot/poweroff/journalctl. + + An empty path would otherwise leave `user ALL=(ALL) NOPASSWD: ` behind, + which visudo rejects, and the whole file would not be installed. + """ + out = subprocess.run( + ["bash", "-c", f". '{LIB}'; web_sudoers_rules u /p /bin/systemctl /bin/bash '' '' ''"], + check=True, capture_output=True, text=True).stdout + rules = [line for line in out.splitlines() if RULE.match(line)] + assert len(rules) == len(EXPECTED_GRANTS) - 5 + assert not [r for r in rules if r.rstrip().endswith("NOPASSWD:")] + assert "journalctl" not in out + + +def test_pip_install_helper_is_granted(): + wanted = ("NOPASSWD:", "$BASH_PATH $PROJECT_ROOT/scripts/fix_perms/safe_pip_install.sh *") + assert wanted in _generator_grants() + + +def _granted_helpers(): + helpers = set() + for _, command in _generator_grants(): + m = re.search(r"scripts/fix_perms/([\w.-]+\.sh)", command) + if m: + helpers.add(m.group(1)) + assert helpers, "no fix_perms helper grant found; the parser matched nothing" + return helpers def _variables(text): @@ -60,57 +196,9 @@ def _normalise(command, variables): for _ in range(3): # helper paths reference $PROJECT_ROOT command = re.sub(r"\$\{?([A-Z][A-Z0-9_]*)\}?", lambda m: variables.get(m.group(1), m.group(0)), command) - command = command.replace("$PROJECT_ROOT_DIR", "$PROJECT_ROOT") return " ".join(command.split()) -def _grants(path): - """{(tags, command)} for every ledmatrix_web rule the script writes.""" - section = _web_sudoers_section(path) - variables = _variables(_text(path)) - grants = set() - for line in section.splitlines(): - m = re.search(r'\$(?:WEB_USER|ACTUAL_USER) ALL=\(ALL\) (NOPASSWD:(?:NOEXEC:)?)\s*(.*)$', - line) - if not m: - continue - command = m.group(2).rstrip().rstrip('"').rstrip() - grants.add((m.group(1), _normalise(command, variables))) - return grants - - -def test_both_installers_generate_rules(): - # Guards against the parser silently matching nothing in either file. - assert len(_grants(FIRST_TIME)) >= 15 - assert len(_grants(CONFIGURE)) >= 15 - - -def test_installers_grant_the_same_commands(): - first = _grants(FIRST_TIME) - configure = _grants(CONFIGURE) - only_first = {c for c in first - configure if c[1] not in ONLY_IN_FIRST_TIME} - only_configure = {c for c in configure - first if c[1] not in ONLY_IN_CONFIGURE} - assert not only_first and not only_configure, ( - "ledmatrix_web sudoers drift between installers:\n" - f" only in first_time_install.sh: {sorted(only_first)}\n" - f" only in configure_web_sudo.sh: {sorted(only_configure)}") - - -def test_pip_install_helper_is_granted(): - wanted = ("NOPASSWD:", "$BASH_PATH $PROJECT_ROOT/scripts/fix_perms/safe_pip_install.sh *") - assert wanted in _grants(FIRST_TIME) - assert wanted in _grants(CONFIGURE) - - -def _granted_helpers(): - helpers = set() - for _, command in _grants(FIRST_TIME) | _grants(CONFIGURE): - m = re.search(r"scripts/fix_perms/([\w.-]+\.sh)", command) - if m: - helpers.add(m.group(1)) - return helpers - - def test_every_granted_helper_is_hardened_in_configure_web_sudo(): text = _text(CONFIGURE) variables = _variables(text) @@ -138,9 +226,8 @@ def test_no_grant_runs_a_file_the_web_user_can_edit(): rule for it lets the web user rewrite the file and run it as root. The grants for display_controller.py, start_display.sh and stop_display.sh were exactly that, and nothing ever ran them through sudo.""" - for installer in (FIRST_TIME, CONFIGURE): - for _, command in _grants(installer): - for token in command.split(): - if token.startswith("$PROJECT_ROOT/"): - assert token.startswith("$PROJECT_ROOT/scripts/fix_perms/"), ( - f"{installer.name} grants root on a user-owned file: {command}") + for _, command in _generator_grants(): + for token in command.split(): + if token.startswith("$PROJECT_ROOT/"): + assert token.startswith("$PROJECT_ROOT/scripts/fix_perms/"), ( + f"lib_sudoers.sh grants root on a user-owned file: {command}") diff --git a/test/web_interface/test_api_v3_names_resolve.py b/test/web_interface/test_api_v3_names_resolve.py new file mode 100644 index 00000000..6b6e6c1c --- /dev/null +++ b/test/web_interface/test_api_v3_names_resolve.py @@ -0,0 +1,25 @@ +"""Names two rarely-run api_v3 paths call must exist. + +Both slipped through because nothing exercised them: the Pixlet editor's +stop route only restarts the display after a SIGKILL, and the Starlark +device-location resolver only builds a cache manager when the web app has +not set one. Either raised NameError when it finally ran. +""" + +from unittest.mock import patch + +from test._api_v3_test_helpers import api_v3_module # noqa: F401 + + +def test_the_editor_stop_route_can_restart_the_display(): + from web_interface.blueprints.api_v3 import starlark + + assert callable(starlark._run_systemctl_command) + + +def test_the_device_location_resolver_builds_without_a_cache_manager(api_v3_module): + pkg = api_v3_module + with patch.object(pkg.api_v3, 'cache_manager', None, create=True), \ + patch.object(pkg, '_starlark_device_location', None): + resolver = pkg._get_starlark_device_location() + assert resolver.cache_manager is None diff --git a/test/web_interface/test_systemctl_sudoers_alignment.py b/test/web_interface/test_systemctl_sudoers_alignment.py index 0013eccf..113b788d 100644 --- a/test/web_interface/test_systemctl_sudoers_alignment.py +++ b/test/web_interface/test_systemctl_sudoers_alignment.py @@ -1,5 +1,6 @@ """Guards that every privileged systemctl call the web interface makes is -covered by a passwordless-sudo grant in configure_web_sudo.sh. +covered by a passwordless-sudo grant in scripts/install/lib_sudoers.sh, which +both first_time_install.sh and configure_web_sudo.sh write the rules from. The web interface runs headless (no TTY), so any `sudo` call that is not matched by a NOPASSWD rule in /etc/sudoers.d/ledmatrix_web falls back to a @@ -25,7 +26,7 @@ API_V3_PKG = PROJECT_ROOT / "web_interface" / "blueprints" / "api_v3" def _api_v3_source() -> str: return "\n".join(p.read_text() for p in sorted(API_V3_PKG.glob("*.py"))) -SUDOERS_SCRIPT = PROJECT_ROOT / "scripts" / "install" / "configure_web_sudo.sh" +SUDOERS_SCRIPT = PROJECT_ROOT / "scripts" / "install" / "lib_sudoers.sh" def _sudo_systemctl_calls(source: str) -> set[tuple[str, str]]: @@ -64,7 +65,7 @@ def test_every_sudo_systemctl_call_is_granted() -> None: uncovered = {c for c in calls if c not in rules} assert not uncovered, ( "These sudo systemctl calls have no matching NOPASSWD grant in " - "configure_web_sudo.sh; they will fail headless with " + "lib_sudoers.sh; they will fail headless with " "'sudo: a terminal is required to read the password': " + ", ".join(f"systemctl {v} {u}" for v, u in sorted(uncovered)) ) diff --git a/web_interface/blueprints/api_v3/__init__.py b/web_interface/blueprints/api_v3/__init__.py index 6d26a980..29a9f4e4 100644 --- a/web_interface/blueprints/api_v3/__init__.py +++ b/web_interface/blueprints/api_v3/__init__.py @@ -1712,7 +1712,7 @@ def _get_starlark_device_location() -> DeviceLocationResolver: global _starlark_device_location if _starlark_device_location is None: _starlark_device_location = DeviceLocationResolver( - getattr(api_v3, 'cache_manager', None) or _ensure_cache_manager(), logger) + getattr(api_v3, 'cache_manager', None), logger) return _starlark_device_location diff --git a/web_interface/blueprints/api_v3/starlark.py b/web_interface/blueprints/api_v3/starlark.py index 856de4a4..08d592c7 100644 --- a/web_interface/blueprints/api_v3/starlark.py +++ b/web_interface/blueprints/api_v3/starlark.py @@ -11,6 +11,7 @@ from web_interface.blueprints.api_v3 import ( _PIXLET_EDITOR_DEFAULT_TIMEOUT, _PIXLET_EDITOR_MAX_TIMEOUT, _PIXLET_EDITOR_SCRIPT, _PIXLET_EDITOR_STATE, _clear_pixlet_editor_state, _find_pixlet_binary, _install_star_file, _pixlet_editor_alive, + _run_systemctl_command, _pixlet_editor_status, _read_pixlet_editor_state, _STARLARK_APPS_DIR, _standalone_render_starlark_app, _starlark_github_token, _starlark_manifest_lock,