mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-02 09:18:06 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
85d321cf33 | ||
|
|
63a233f3ed | ||
|
|
7a9d01342a |
+15
-6
@@ -726,7 +726,11 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
|
||||
|
||||
if command -v timeout >/dev/null 2>&1; then
|
||||
# Use timeout if available (10 minutes = 600 seconds)
|
||||
if timeout 600 python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
|
||||
# --ignore-installed: apt-managed packages (e.g. python3-requests)
|
||||
# ship no pip RECORD file, so upgrading them would otherwise abort
|
||||
# with "uninstall-no-record-file"; this lays the new version down
|
||||
# alongside instead of trying to uninstall the apt copy first.
|
||||
if timeout 600 python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
|
||||
INSTALL_SUCCESS=true
|
||||
else
|
||||
EXIT_CODE=$?
|
||||
@@ -734,7 +738,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
|
||||
echo "✗ Timeout (10 minutes) installing: $line"
|
||||
echo " This package may require building from source, which can be slow on Raspberry Pi."
|
||||
echo " You can try installing it manually later with:"
|
||||
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose '$line'"
|
||||
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose '$line'"
|
||||
else
|
||||
echo "✗ Failed to install: $line (exit code: $EXIT_CODE)"
|
||||
fi
|
||||
@@ -742,7 +746,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
|
||||
else
|
||||
# No timeout command available, install without timeout
|
||||
echo " Note: timeout command not available, installation may take a while..."
|
||||
if python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
|
||||
if python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose "$line" > "$INSTALL_OUTPUT" 2>&1; then
|
||||
INSTALL_SUCCESS=true
|
||||
else
|
||||
EXIT_CODE=$?
|
||||
@@ -794,7 +798,7 @@ if [ -f "$PROJECT_ROOT_DIR/requirements.txt" ]; then
|
||||
echo " 1. Ensure you have enough disk space: df -h"
|
||||
echo " 2. Check available memory: free -h"
|
||||
echo " 3. Try installing failed packages individually with verbose output:"
|
||||
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --verbose <package>"
|
||||
echo " python3 -m pip install --break-system-packages --no-cache-dir --prefer-binary --ignore-installed --verbose <package>"
|
||||
echo " 4. For packages that build from source (like numpy), consider:"
|
||||
echo " - Installing pre-built wheels: python3 -m pip install --only-binary :all: <package>"
|
||||
echo " - Or installing via apt if available: sudo apt install python3-<package>"
|
||||
@@ -816,7 +820,10 @@ echo ""
|
||||
# Install web interface dependencies
|
||||
echo "Installing web interface dependencies..."
|
||||
if [ -f "$PROJECT_ROOT_DIR/web_interface/requirements.txt" ]; then
|
||||
if python3 -m pip install --break-system-packages --prefer-binary -r "$PROJECT_ROOT_DIR/web_interface/requirements.txt"; then
|
||||
# --ignore-installed: apt-managed packages (e.g. python3-requests) ship no
|
||||
# pip RECORD file, so upgrading them to the version pinned here would
|
||||
# otherwise abort the whole install with "uninstall-no-record-file".
|
||||
if python3 -m pip install --break-system-packages --prefer-binary --ignore-installed -r "$PROJECT_ROOT_DIR/web_interface/requirements.txt"; then
|
||||
echo "✓ Web interface dependencies installed"
|
||||
# Create marker file to indicate dependencies are installed
|
||||
touch "$PROJECT_ROOT_DIR/.web_deps_installed"
|
||||
@@ -977,7 +984,9 @@ else
|
||||
else
|
||||
echo "Using pip to install dependencies..."
|
||||
if [ -f "$PROJECT_ROOT_DIR/requirements_web_v2.txt" ]; then
|
||||
python3 -m pip install --break-system-packages --prefer-binary -r requirements_web_v2.txt
|
||||
# --ignore-installed: see the Step 5 web_interface/requirements.txt
|
||||
# install above — same apt/pip RECORD-file conflict applies here.
|
||||
python3 -m pip install --break-system-packages --prefer-binary --ignore-installed -r requirements_web_v2.txt
|
||||
else
|
||||
echo "⚠ requirements_web_v2.txt not found; skipping web dependency install"
|
||||
fi
|
||||
|
||||
@@ -10,6 +10,7 @@ import os
|
||||
import logging
|
||||
import shutil as _shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
@@ -287,3 +288,104 @@ def sudo_remove_directory(path: Path, allowed_bases: Optional[list] = None) -> b
|
||||
logger.error(f"Unexpected error during sudo helper for {path}: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def install_requirements_file(req_file: Path, timeout: int = 300) -> subprocess.CompletedProcess:
|
||||
"""
|
||||
Install a requirements.txt file for a plugin (or the project itself).
|
||||
|
||||
Prefers the vetted sudo wrapper (scripts/fix_perms/safe_pip_install.sh) so
|
||||
packages end up visible to root-run ledmatrix.service, not just to
|
||||
whichever non-root user happens to run the calling process (e.g. the web
|
||||
interface). Falls back to installing with the calling process's own
|
||||
interpreter if the wrapper isn't set up yet (the admin hasn't run
|
||||
scripts/install/configure_web_sudo.sh), so dependency installation still
|
||||
does *something* useful rather than hard-failing.
|
||||
|
||||
Always installs with the interpreter that will actually run the code
|
||||
(``sys.executable`` in the fallback path, the wrapper's ``python3`` in the
|
||||
sudo path) rather than a bare ``pip``/``pip3`` off PATH, which can
|
||||
silently resolve to a different Python installation (e.g. system Python
|
||||
vs. a virtualenv) than the one importing the package at runtime.
|
||||
|
||||
Args:
|
||||
req_file: Path to a requirements.txt file
|
||||
timeout: Subprocess timeout in seconds
|
||||
|
||||
Returns:
|
||||
subprocess.CompletedProcess from the pip (or wrapper) invocation.
|
||||
Never raises on a non-zero exit; callers should check ``returncode``.
|
||||
``stdout`` is prefixed with an explanatory note when the root wrapper
|
||||
was unavailable and the fallback path was used.
|
||||
"""
|
||||
project_root = Path(__file__).resolve().parent.parent.parent
|
||||
wrapper = project_root / "scripts" / "fix_perms" / "safe_pip_install.sh"
|
||||
|
||||
if wrapper.exists():
|
||||
# See sudo_remove_directory / configure_web_sudo.sh for why bash must
|
||||
# be invoked with an explicit, known path rather than relying on the
|
||||
# wrapper's shebang: sudoers matches the exact command line.
|
||||
bash_candidates = []
|
||||
for candidate in ("/usr/bin/bash", "/bin/bash", _shutil.which("bash")):
|
||||
if candidate and candidate not in bash_candidates:
|
||||
bash_candidates.append(candidate)
|
||||
|
||||
result = None
|
||||
for bash_path in bash_candidates:
|
||||
# bash_path and wrapper are fixed, known-good paths, and
|
||||
# safe_pip_install.sh independently re-validates req_file is an
|
||||
# allowed requirements.txt before installing anything as root.
|
||||
result = subprocess.run( # nosec B603 - no shell invoked (list-form argv) # nosemgrep
|
||||
["sudo", "-n", bash_path, str(wrapper), str(req_file)],
|
||||
capture_output=True, text=True, timeout=timeout, cwd=str(project_root)
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return result
|
||||
# Distinguish "sudo rejected this exact command line" (worth
|
||||
# trying the next bash candidate) from "sudo ran it but pip
|
||||
# itself failed" (a real error — stop and surface it).
|
||||
denied = any(
|
||||
phrase in result.stderr
|
||||
for phrase in ("a password is required", "is not allowed to run", "no tty present")
|
||||
)
|
||||
if not denied:
|
||||
logger.warning(
|
||||
"Root pip install failed (rc=%s) for %s: %s",
|
||||
result.returncode, req_file, result.stderr.strip()[:500],
|
||||
)
|
||||
return result
|
||||
|
||||
logger.warning(
|
||||
"Root pip install wrapper denied via sudo for %s; falling back to "
|
||||
"user-level install: %s",
|
||||
req_file, result.stderr.strip()[:500] if result else "no bash candidates found",
|
||||
)
|
||||
note = (
|
||||
f"[Root install unavailable ({(result.stderr.strip() if result else 'sudo denied') or 'sudo denied'}); "
|
||||
"installed for the current process's user only. Packages may not be "
|
||||
"visible to ledmatrix.service if it runs as a different user — "
|
||||
"run scripts/install/configure_web_sudo.sh to fix this.]\n"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"safe_pip_install.sh not found; falling back to user-level install for %s",
|
||||
req_file,
|
||||
)
|
||||
note = (
|
||||
"[safe_pip_install.sh not found; installed for the current process's "
|
||||
"user only. Run scripts/install/configure_web_sudo.sh to enable "
|
||||
"root installs visible to ledmatrix.service.]\n"
|
||||
)
|
||||
|
||||
# sys.executable is this process's own interpreter (not
|
||||
# attacker-influenced), and req_file is a Path built internally by callers
|
||||
# (store_manager.py plugin paths, PROJECT_ROOT/requirements.txt), never
|
||||
# raw external/user input. --ignore-installed matches safe_pip_install.sh:
|
||||
# apt-managed packages (e.g. python3-requests) ship no pip RECORD file, so
|
||||
# upgrading them would otherwise abort with "uninstall-no-record-file".
|
||||
result = subprocess.run( # nosec B603 - no shell invoked (list-form argv) # nosemgrep
|
||||
[sys.executable, "-m", "pip", "install", "--break-system-packages", "--ignore-installed", "-r", str(req_file)],
|
||||
capture_output=True, text=True, timeout=timeout, cwd=str(project_root)
|
||||
)
|
||||
result.stdout = note + (result.stdout or "")
|
||||
return result
|
||||
|
||||
|
||||
@@ -235,16 +235,51 @@ class PluginLoader:
|
||||
return True
|
||||
else:
|
||||
stderr = result.stderr or ""
|
||||
# uninstall-no-record-file means the package is already present at the
|
||||
# system level (e.g. installed via dnf/apt without a pip RECORD file).
|
||||
# pip can't replace it, but it IS installed — write the marker so we
|
||||
# don't retry on every restart.
|
||||
# uninstall-no-record-file means a system-managed copy of a package
|
||||
# (e.g. apt's python3-requests, which ships no pip RECORD file) is in
|
||||
# the way of the version this requirements.txt pins. Retry with
|
||||
# --ignore-installed so pip lays the pinned version down alongside
|
||||
# the system copy instead of trying to replace it — matching the
|
||||
# retry already used by install_dependencies_apt.py / safe_pip_install.sh.
|
||||
# Without this retry, the plugin would silently keep running against
|
||||
# whatever version the system happened to ship, even though the
|
||||
# marker below claims the requirement is satisfied.
|
||||
if "uninstall-no-record-file" in stderr:
|
||||
self.logger.warning(
|
||||
"Dependencies for %s include system-managed packages (no pip RECORD). "
|
||||
"Assuming they are satisfied: %s",
|
||||
"Dependencies for %s conflict with a system-managed package "
|
||||
"(no pip RECORD); retrying with --ignore-installed: %s",
|
||||
plugin_id, stderr.strip()
|
||||
)
|
||||
# Wrapped in its own try/except so a retry timeout is
|
||||
# tolerated the same way as a retry failure, instead of
|
||||
# propagating to the outer handler and returning False
|
||||
# (which would contradict the "assume satisfied" fallback
|
||||
# below).
|
||||
try:
|
||||
# sys.executable is this process's own interpreter (not
|
||||
# attacker-influenced), and requirements_file is a path
|
||||
# built internally by find_plugin_directory, never raw
|
||||
# external input.
|
||||
retry_result = subprocess.run( # nosec B603 - no shell invoked (list-form argv) # nosemgrep
|
||||
[sys.executable, "-m", "pip", "install", "--break-system-packages",
|
||||
"--ignore-installed", "-r", requirements_file],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
check=False
|
||||
)
|
||||
if retry_result.returncode != 0:
|
||||
self.logger.warning(
|
||||
"Retry with --ignore-installed also failed for %s; assuming the "
|
||||
"system-managed version satisfies the requirement: %s",
|
||||
plugin_id, (retry_result.stderr or "").strip()
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.logger.warning(
|
||||
"Retry with --ignore-installed timed out for %s; assuming the "
|
||||
"system-managed version satisfies the requirement",
|
||||
plugin_id
|
||||
)
|
||||
try:
|
||||
with open(marker_file, 'w', encoding='utf-8') as fh:
|
||||
fh.write(current_hash)
|
||||
|
||||
@@ -25,7 +25,7 @@ import logging
|
||||
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from src.common.permission_utils import sudo_remove_directory
|
||||
from src.common.permission_utils import sudo_remove_directory, install_requirements_file
|
||||
|
||||
try:
|
||||
from jsonschema import Draft7Validator, ValidationError
|
||||
@@ -1915,13 +1915,19 @@ class PluginStoreManager:
|
||||
|
||||
try:
|
||||
self.logger.info(f"Installing dependencies for {plugin_path.name}")
|
||||
subprocess.run(
|
||||
['pip3', 'install', '--break-system-packages', '-r', str(requirements_file)],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=300
|
||||
)
|
||||
# Routed through the shared root-visible installer (same one the
|
||||
# web UI's "Reinstall Plugin Deps" tool uses) rather than a bare
|
||||
# `pip`/`pip3` off PATH: a bare pip binary can silently resolve to
|
||||
# a different Python installation than the one that actually runs
|
||||
# ledmatrix.service, so pip reports success while the package
|
||||
# stays invisible to the running plugin (e.g. missing `astral`
|
||||
# for the weather plugin even though "install" succeeded).
|
||||
result = install_requirements_file(requirements_file, timeout=300)
|
||||
if result.returncode != 0:
|
||||
self.logger.error(
|
||||
f"Error installing dependencies for {plugin_path.name}: {result.stderr}"
|
||||
)
|
||||
return False
|
||||
self.logger.info(f"Dependencies installed successfully for {plugin_path.name}")
|
||||
# Write hash marker so plugin_loader skips redundant pip run on next startup
|
||||
try:
|
||||
@@ -1930,10 +1936,7 @@ class PluginStoreManager:
|
||||
except OSError as marker_err:
|
||||
self.logger.debug("Could not write dependency marker for %s: %s", plugin_path.name, marker_err)
|
||||
return True
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
self.logger.error(f"Error installing dependencies: {e.stderr}")
|
||||
return False
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
self.logger.error("Dependency installation timed out")
|
||||
return False
|
||||
|
||||
@@ -279,6 +279,19 @@ class PluginAdapter:
|
||||
# Copy the image to prevent modification
|
||||
img = cached_image.copy()
|
||||
|
||||
# Plugins that build their own ticker image via this shared
|
||||
# ScrollHelper's create_scrolling_image() get a solid-black
|
||||
# leading margin exactly `display_width` columns wide baked in
|
||||
# (scroll_helper.py's "initial gap before first item"). Vegas mode
|
||||
# adds its own leading gap/separator around every item already,
|
||||
# so leaving this in stacks a second, uncontrolled blank margin on
|
||||
# top of vegas_scroll.separator_width — making this plugin's
|
||||
# transitions look inconsistent with plugins that provide content
|
||||
# via get_vegas_content() (which carries no such margin). Strip it
|
||||
# here so every plugin contributes only its real content and the
|
||||
# gap between items is governed solely by separator_width.
|
||||
img = self._strip_scroll_padding(img, scroll_helper, plugin_id)
|
||||
|
||||
# Ensure correct height
|
||||
if img.height != self.display_height:
|
||||
logger.info(
|
||||
@@ -306,6 +319,69 @@ class PluginAdapter:
|
||||
logger.exception("[%s] Error getting scroll_helper content", plugin_id)
|
||||
return None
|
||||
|
||||
def _strip_scroll_padding(
|
||||
self, img: Image.Image, scroll_helper: Any, plugin_id: str
|
||||
) -> Image.Image:
|
||||
"""
|
||||
Crop off a plugin's own leading/trailing blank margins, if present.
|
||||
|
||||
create_scrolling_image() always pads the *start* of its cached image
|
||||
with exactly `scroll_helper.display_width` columns of solid black
|
||||
(0, 0, 0) ("initial gap before first item"). Some ticker-style plugins
|
||||
also pad the *end* of their own cached image (e.g. so their standalone
|
||||
display exits cleanly before looping). Vegas mode already adds its own
|
||||
gap/separator around every item, so either margin left in place stacks
|
||||
an extra, uncontrolled blank stretch on top of `separator_width` —
|
||||
only when running inside Vegas mode does this matter, since the
|
||||
plugin's own standalone display still wants that margin. Detect solid
|
||||
black margins up to `scroll_helper.display_width` wide on each edge and
|
||||
crop them here. Images built via set_scrolling_image() (no such
|
||||
margins) are left untouched.
|
||||
|
||||
Args:
|
||||
img: Captured scroll_helper.cached_image (already copied)
|
||||
scroll_helper: The plugin's ScrollHelper instance
|
||||
plugin_id: Plugin identifier for logging
|
||||
|
||||
Returns:
|
||||
img, cropped on whichever edge(s) had a matching blank margin
|
||||
"""
|
||||
pad_width = getattr(scroll_helper, 'display_width', None)
|
||||
if not isinstance(pad_width, int) or pad_width <= 0 or pad_width >= img.width:
|
||||
return img
|
||||
|
||||
def is_solid_black(strip: Image.Image) -> bool:
|
||||
return strip.convert('RGB').getextrema() == ((0, 0), (0, 0), (0, 0))
|
||||
|
||||
left = pad_width if is_solid_black(img.crop((0, 0, pad_width, img.height))) else 0
|
||||
right = (
|
||||
pad_width
|
||||
if is_solid_black(img.crop((img.width - pad_width, 0, img.width, img.height)))
|
||||
else 0
|
||||
)
|
||||
|
||||
if not left and not right:
|
||||
return img
|
||||
|
||||
# Degenerate case (e.g. an all-black cached image): don't crop past
|
||||
# zero width, just leave the image as-is.
|
||||
if left + right >= img.width:
|
||||
return img
|
||||
|
||||
cropped = img.crop((left, 0, img.width - right, img.height))
|
||||
|
||||
# Both edges matching at once is a much stronger signal of genuine
|
||||
# baked-in padding than a single edge (which has a small chance of
|
||||
# coinciding with real all-black content, e.g. a dark logo touching
|
||||
# one boundary). Log that case at warning level so an unexpected
|
||||
# double-edge crop is easy to spot in the field.
|
||||
log = logger.warning if (left and right) else logger.info
|
||||
log(
|
||||
"[%s] Stripping scroll_helper padding (left=%dpx, right=%dpx): %dpx -> %dpx",
|
||||
plugin_id, left, right, img.width, cropped.width
|
||||
)
|
||||
return cropped
|
||||
|
||||
def _trigger_scroll_content_generation(
|
||||
self, plugin: 'BasePlugin', plugin_id: str, scroll_helper: Any
|
||||
) -> Optional[Image.Image]:
|
||||
|
||||
@@ -4,6 +4,8 @@ Tests for PluginLoader.
|
||||
Tests plugin directory discovery, module loading, and class instantiation.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from src.plugin_system.plugin_loader import PluginLoader
|
||||
@@ -216,7 +218,82 @@ class TestPluginLoader:
|
||||
requirements_file.write_text("package1==1.0.0\n")
|
||||
|
||||
mock_subprocess.return_value = MagicMock(returncode=1)
|
||||
|
||||
|
||||
result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
|
||||
|
||||
|
||||
assert result is False
|
||||
|
||||
@patch('subprocess.run')
|
||||
def test_install_dependencies_retries_with_ignore_installed_on_apt_conflict(
|
||||
self, mock_subprocess, plugin_loader, tmp_plugins_dir
|
||||
):
|
||||
"""An apt-managed package with no pip RECORD file triggers a retry with
|
||||
--ignore-installed rather than silently assuming the old version satisfies
|
||||
the requirement."""
|
||||
plugin_dir = tmp_plugins_dir / "test_plugin"
|
||||
plugin_dir.mkdir()
|
||||
requirements_file = plugin_dir / "requirements.txt"
|
||||
requirements_file.write_text("requests>=2.33.0,<3.0.0\n")
|
||||
|
||||
first_attempt = MagicMock(
|
||||
returncode=1,
|
||||
stderr="ERROR: Cannot uninstall requests 2.32.3\nuninstall-no-record-file"
|
||||
)
|
||||
retry_attempt = MagicMock(returncode=0, stderr="")
|
||||
mock_subprocess.side_effect = [first_attempt, retry_attempt]
|
||||
|
||||
result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
|
||||
|
||||
assert result is True
|
||||
assert mock_subprocess.call_count == 2
|
||||
retry_cmd = mock_subprocess.call_args_list[1][0][0]
|
||||
assert "--ignore-installed" in retry_cmd
|
||||
|
||||
@patch('subprocess.run')
|
||||
def test_install_dependencies_apt_conflict_retry_also_fails(
|
||||
self, mock_subprocess, plugin_loader, tmp_plugins_dir
|
||||
):
|
||||
"""Still tolerates the failure (returns True) if the --ignore-installed
|
||||
retry itself fails, matching the prior soft-fallback behavior."""
|
||||
plugin_dir = tmp_plugins_dir / "test_plugin"
|
||||
plugin_dir.mkdir()
|
||||
requirements_file = plugin_dir / "requirements.txt"
|
||||
requirements_file.write_text("requests>=2.33.0,<3.0.0\n")
|
||||
|
||||
first_attempt = MagicMock(
|
||||
returncode=1,
|
||||
stderr="ERROR: Cannot uninstall requests 2.32.3\nuninstall-no-record-file"
|
||||
)
|
||||
retry_attempt = MagicMock(returncode=1, stderr="some other pip error")
|
||||
mock_subprocess.side_effect = [first_attempt, retry_attempt]
|
||||
|
||||
result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
|
||||
|
||||
assert result is True
|
||||
assert mock_subprocess.call_count == 2
|
||||
|
||||
@patch('subprocess.run')
|
||||
def test_install_dependencies_apt_conflict_retry_times_out(
|
||||
self, mock_subprocess, plugin_loader, tmp_plugins_dir
|
||||
):
|
||||
"""A retry timeout must be tolerated the same way as a retry failure
|
||||
(return True), not propagate to the outer TimeoutExpired handler and
|
||||
return False."""
|
||||
plugin_dir = tmp_plugins_dir / "test_plugin"
|
||||
plugin_dir.mkdir()
|
||||
requirements_file = plugin_dir / "requirements.txt"
|
||||
requirements_file.write_text("requests>=2.33.0,<3.0.0\n")
|
||||
|
||||
first_attempt = MagicMock(
|
||||
returncode=1,
|
||||
stderr="ERROR: Cannot uninstall requests 2.32.3\nuninstall-no-record-file"
|
||||
)
|
||||
mock_subprocess.side_effect = [
|
||||
first_attempt,
|
||||
subprocess.TimeoutExpired(cmd="pip", timeout=300),
|
||||
]
|
||||
|
||||
result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
|
||||
|
||||
assert result is True
|
||||
assert mock_subprocess.call_count == 2
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
"""
|
||||
Tests for src/vegas_mode/plugin_adapter.py
|
||||
|
||||
Covers PluginAdapter._strip_scroll_padding(): the heuristic that crops a
|
||||
plugin's own baked-in leading/trailing blank margins before Vegas mode
|
||||
composites the content, so vegas_scroll.separator_width is the only gap
|
||||
applied between items.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from src.common.scroll_helper import ScrollHelper
|
||||
from src.vegas_mode.plugin_adapter import PluginAdapter
|
||||
|
||||
|
||||
class FakeDisplayManager:
|
||||
width = 64
|
||||
height = 32
|
||||
|
||||
|
||||
class FakePlugin:
|
||||
def __init__(self, scroll_helper):
|
||||
self.scroll_helper = scroll_helper
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def adapter():
|
||||
return PluginAdapter(FakeDisplayManager())
|
||||
|
||||
|
||||
def _solid(width: int, height: int, color: tuple) -> Image.Image:
|
||||
"""Create a solid-color RGB image of the given dimensions."""
|
||||
return Image.new('RGB', (width, height), color)
|
||||
|
||||
|
||||
class TestStripScrollPadding:
|
||||
def test_leading_pad_from_create_scrolling_image_is_stripped(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
item = _solid(40, 32, (200, 50, 50))
|
||||
sh.create_scrolling_image([item], item_gap=10, element_gap=0)
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
assert images[0].width == 40
|
||||
assert images[0].getpixel((0, 0)) == (200, 50, 50)
|
||||
|
||||
def test_leading_and_trailing_pad_both_stripped(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
content_w = 80
|
||||
full = _solid(64 + content_w + 64, 32, (0, 0, 0))
|
||||
full.paste(_solid(content_w, 32, (10, 220, 30)), (64, 0))
|
||||
sh.set_scrolling_image(full)
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
assert images[0].width == content_w
|
||||
assert images[0].getpixel((0, 0)) == (10, 220, 30)
|
||||
assert images[0].getpixel((content_w - 1, 0)) == (10, 220, 30)
|
||||
|
||||
def test_leading_only_pad_stripped_trailing_content_kept(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
content_w = 80
|
||||
full = _solid(64 + content_w, 32, (0, 0, 0))
|
||||
full.paste(_solid(content_w, 32, (5, 5, 250)), (64, 0))
|
||||
sh.set_scrolling_image(full)
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
assert images[0].width == content_w
|
||||
assert images[0].getpixel((0, 0)) == (5, 5, 250)
|
||||
|
||||
def test_trailing_only_pad_stripped_leading_content_kept(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
content_w = 80
|
||||
full = _solid(content_w + 64, 32, (0, 0, 0))
|
||||
full.paste(_solid(content_w, 32, (5, 5, 250)), (0, 0))
|
||||
sh.set_scrolling_image(full)
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
assert images[0].width == content_w
|
||||
assert images[0].getpixel((0, 0)) == (5, 5, 250)
|
||||
|
||||
def test_no_margin_image_left_untouched(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
raw = _solid(150, 32, (5, 5, 5))
|
||||
raw.paste(_solid(50, 32, (123, 45, 67)), (0, 0))
|
||||
sh.set_scrolling_image(raw)
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "no_margin")
|
||||
assert images[0].width == 150
|
||||
|
||||
def test_degenerate_all_black_image_left_untouched(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
sh.set_scrolling_image(_solid(50, 32, (0, 0, 0)))
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "all_black")
|
||||
assert images[0].width == 50
|
||||
|
||||
def test_missing_display_width_attribute_left_untouched(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
item = _solid(40, 32, (200, 50, 50))
|
||||
sh.create_scrolling_image([item], item_gap=10, element_gap=0)
|
||||
original_width = sh.cached_image.width
|
||||
del sh.display_width
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
assert images[0].width == original_width
|
||||
|
||||
def test_pad_width_not_smaller_than_image_left_untouched(self, adapter):
|
||||
sh = ScrollHelper(64, 32)
|
||||
sh.set_scrolling_image(_solid(64, 32, (0, 0, 0)))
|
||||
|
||||
images = adapter._get_scroll_helper_content(FakePlugin(sh), "narrow")
|
||||
assert images[0].width == 64
|
||||
|
||||
def test_both_edges_matching_logs_warning(self, adapter, caplog):
|
||||
sh = ScrollHelper(64, 32)
|
||||
content_w = 80
|
||||
full = _solid(64 + content_w + 64, 32, (0, 0, 0))
|
||||
full.paste(_solid(content_w, 32, (10, 220, 30)), (64, 0))
|
||||
sh.set_scrolling_image(full)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="src.vegas_mode.plugin_adapter"):
|
||||
adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
|
||||
assert any("Stripping scroll_helper padding" in r.message for r in caplog.records)
|
||||
|
||||
def test_single_edge_match_logs_info_not_warning(self, adapter, caplog):
|
||||
sh = ScrollHelper(64, 32)
|
||||
item = _solid(40, 32, (200, 50, 50))
|
||||
sh.create_scrolling_image([item], item_gap=10, element_gap=0)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="src.vegas_mode.plugin_adapter"):
|
||||
adapter._get_scroll_helper_content(FakePlugin(sh), "ticker")
|
||||
|
||||
strip_records = [r for r in caplog.records if "Stripping scroll_helper padding" in r.message]
|
||||
assert len(strip_records) == 1
|
||||
assert strip_records[0].levelno == logging.INFO
|
||||
@@ -26,6 +26,7 @@ from src.web_interface.validators import (
|
||||
validate_file_upload
|
||||
)
|
||||
from src.error_aggregator import get_error_aggregator
|
||||
from src.common.permission_utils import install_requirements_file
|
||||
|
||||
_SUDO = shutil.which('sudo')
|
||||
_JOURNALCTL = shutil.which('journalctl')
|
||||
@@ -50,83 +51,12 @@ def _pip_install_requirements(req_file: Path, timeout: int) -> subprocess.Comple
|
||||
for the current process only if the wrapper isn't set up yet (i.e. the
|
||||
admin hasn't run scripts/install/configure_web_sudo.sh since upgrading),
|
||||
so the button still does *something* useful rather than hard-failing.
|
||||
|
||||
Thin wrapper around the shared implementation in permission_utils so the
|
||||
Plugin Store's own dependency installation (store_manager.py) follows the
|
||||
exact same root-visible install path instead of a divergent one.
|
||||
"""
|
||||
wrapper = PROJECT_ROOT / 'scripts' / 'fix_perms' / 'safe_pip_install.sh'
|
||||
if wrapper.exists():
|
||||
# Must invoke via an explicit `bash <path>` — matching both the
|
||||
# sudoers rule configure_web_sudo.sh provisions ($BASH_PATH
|
||||
# $SAFE_PIP_INSTALL_PATH *) and the existing safe_plugin_rm.sh call
|
||||
# in src/common/permission_utils.py. Calling the script path directly
|
||||
# (relying on its shebang) makes sudo check a different command line
|
||||
# than what's actually allowlisted, so `sudo -n` denies it on any
|
||||
# install that only has the specific rules this script provisions —
|
||||
# it only appeared to work in prior testing because that device also
|
||||
# had a broader, non-standard NOPASSWD: ALL grant.
|
||||
#
|
||||
# $BASH_PATH is resolved once at setup time (configure_web_sudo.sh's
|
||||
# `command -v bash`) and baked into the static sudoers file as a
|
||||
# literal path; sudo requires an exact string match against that, so
|
||||
# if this process's own PATH resolves bash somewhere else, the
|
||||
# sudoers rule won't match here either. Try the standard Debian/
|
||||
# Raspberry Pi OS locations first, then this process's own
|
||||
# resolution, so a divergence in just one of them doesn't break this.
|
||||
bash_candidates = []
|
||||
for candidate in ('/usr/bin/bash', '/bin/bash', shutil.which('bash')):
|
||||
if candidate and candidate not in bash_candidates:
|
||||
bash_candidates.append(candidate)
|
||||
|
||||
result = None
|
||||
for bash_path in bash_candidates:
|
||||
result = subprocess.run(
|
||||
['sudo', '-n', bash_path, str(wrapper), str(req_file)],
|
||||
capture_output=True, text=True, timeout=timeout, cwd=str(PROJECT_ROOT)
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return result
|
||||
# Best-effort distinction between "sudo rejected this exact
|
||||
# command line" (no matching NOPASSWD rule for this bash path —
|
||||
# worth trying the next candidate) and "sudo ran it but the
|
||||
# wrapper/pip itself failed" (a real error — stop and surface it
|
||||
# rather than uselessly retrying other bash paths or doubling up
|
||||
# with a redundant non-root install attempt).
|
||||
denied = any(
|
||||
phrase in result.stderr
|
||||
for phrase in ('a password is required', 'is not allowed to run', 'no tty present')
|
||||
)
|
||||
if not denied:
|
||||
logger.warning(
|
||||
"[Pip Install] Root install failed (rc=%s) for %s: %s",
|
||||
result.returncode, req_file, result.stderr.strip()[:500],
|
||||
)
|
||||
return result
|
||||
|
||||
logger.warning(
|
||||
"[Pip Install] Root wrapper denied via sudo for %s; falling back "
|
||||
"to user-level install: %s",
|
||||
req_file, result.stderr.strip()[:500] if result else 'no bash candidates found',
|
||||
)
|
||||
note = (
|
||||
f"[Root install unavailable ({(result.stderr.strip() if result else 'sudo denied') or 'sudo denied'}); "
|
||||
"installed for the web service's user only. Packages may not be "
|
||||
"visible to ledmatrix.service if it runs as a different user — "
|
||||
"run scripts/install/configure_web_sudo.sh to fix this.]\n"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"[Pip Install] safe_pip_install.sh not found; falling back to user-level install for %s",
|
||||
req_file,
|
||||
)
|
||||
note = (
|
||||
"[safe_pip_install.sh not found; installed for the web service's "
|
||||
"user only. Run scripts/install/configure_web_sudo.sh to enable "
|
||||
"root installs visible to ledmatrix.service.]\n"
|
||||
)
|
||||
result = subprocess.run(
|
||||
[sys.executable, '-m', 'pip', 'install', '--break-system-packages', '-r', str(req_file)],
|
||||
capture_output=True, text=True, timeout=timeout, cwd=str(PROJECT_ROOT)
|
||||
)
|
||||
result.stdout = note + (result.stdout or '')
|
||||
return result
|
||||
return install_requirements_file(req_file, timeout=timeout)
|
||||
|
||||
|
||||
def _scrub_git_remote_url(url: str) -> str:
|
||||
|
||||
Reference in New Issue
Block a user