mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-14 15:18:04 +00:00
test(logos): cover LogoHelper, and stop bad downloads poisoning the cache
Nothing in test/ referenced logo_helper.py, so its caching, resizing and download-fallback logic was entirely unexercised. Two bugs surfaced. _download_logo wrote response.content to disk with no size limit and no check that the bytes were an image. A logo URL is remote input, so the response chose how much went into the assets directory; worse, an undecodable one stayed there, and because load_logo() only reports the decode failure and returns None, every later call re-read the same corrupt file. The download path never retried, so a single bad response made a logo permanently blank rather than falling back to the placeholder. Cap the response, verify it decodes, and delete it if not, which lets the existing fallback in load_logo_with_download do its job. get_cache_stats() divided by self.cache_size with no guard, so a helper built with cache_size=0 raised ZeroDivisionError from what is only a stats call. 37 tests: size-qualified cache keys, LRU eviction and refresh, the four load_logo_with_download paths, download permissions and timeout, placeholder generation, and the abbreviation normalizer — including a test pinning its deliberate divergence from LogoDownloader.normalize_abbreviation, since logo filenames on existing installs depend on both behaviors staying put. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh
This commit is contained in:
@@ -19,6 +19,10 @@ from src.common.permission_utils import (
|
||||
)
|
||||
|
||||
|
||||
# Well above any real team logo; bounds what a remote URL can write to disk.
|
||||
MAX_LOGO_BYTES = 10 * 1024 * 1024
|
||||
|
||||
|
||||
class LogoHelper:
|
||||
"""
|
||||
Helper class for logo loading, caching, and resizing.
|
||||
@@ -226,7 +230,10 @@ class LogoHelper:
|
||||
return {
|
||||
'cached_logos': len(self._logo_cache),
|
||||
'cache_size_limit': self.cache_size,
|
||||
'cache_usage_percent': (len(self._logo_cache) / self.cache_size) * 100
|
||||
'cache_usage_percent': (
|
||||
(len(self._logo_cache) / self.cache_size) * 100
|
||||
if self.cache_size else 0
|
||||
),
|
||||
}
|
||||
|
||||
def _resize_logo(self, logo: Image.Image, max_width: Optional[int] = None,
|
||||
@@ -258,21 +265,43 @@ class LogoHelper:
|
||||
self._cache_order.append(cache_key)
|
||||
|
||||
def _download_logo(self, url: str, file_path: Path) -> None:
|
||||
"""Download logo from URL."""
|
||||
"""Download logo from URL.
|
||||
|
||||
The response size is capped and the saved file is verified as a
|
||||
decodable image before it is left on disk: a logo URL is remote
|
||||
input, and without this an oversized or malformed response would
|
||||
be cached for every later load_logo() call to trip over.
|
||||
"""
|
||||
# Ensure directory exists with proper permissions
|
||||
ensure_directory_permissions(file_path.parent, get_assets_dir_mode())
|
||||
|
||||
|
||||
# Download with timeout
|
||||
response = self.session.get(url, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
content = response.content
|
||||
if len(content) > MAX_LOGO_BYTES:
|
||||
raise ValueError(
|
||||
f"Logo at {url} is {len(content)} bytes, over the "
|
||||
f"{MAX_LOGO_BYTES}-byte limit; not saved")
|
||||
|
||||
# Save to file
|
||||
with open(file_path, 'wb') as f:
|
||||
f.write(response.content)
|
||||
|
||||
f.write(content)
|
||||
|
||||
# Verify it decodes before leaving it on disk. PIL raises
|
||||
# DecompressionBombError past its own pixel limit; a partial or
|
||||
# non-image response raises UnidentifiedImageError/OSError.
|
||||
try:
|
||||
with Image.open(file_path) as probe:
|
||||
probe.load()
|
||||
except Exception:
|
||||
file_path.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
# Set proper file permissions after saving
|
||||
ensure_file_permissions(file_path, get_assets_file_mode())
|
||||
|
||||
|
||||
self.logger.debug(f"Downloaded logo to {file_path}")
|
||||
|
||||
def _create_placeholder_logo(self, team_abbr: str,
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
"""
|
||||
Tests for src/common/logo_helper.py — logo loading, LRU caching, resizing,
|
||||
and download-with-fallback. Previously untested: nothing in test/ referenced
|
||||
this module at all.
|
||||
|
||||
Real PIL images under tmp_path are used rather than mocked ones, since
|
||||
load_logo() does real Path.exists() and Image.open() calls; only the HTTP
|
||||
session and the permission helpers are patched.
|
||||
|
||||
Regression coverage for two fixed bugs:
|
||||
- _download_logo wrote response.content to disk with no size cap and no
|
||||
check that the bytes decoded as an image, so a hostile or broken URL
|
||||
could leave arbitrary/oversized content cached in the assets directory.
|
||||
- get_cache_stats() divided by self.cache_size unguarded, raising
|
||||
ZeroDivisionError for a helper constructed with cache_size=0.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from PIL import Image
|
||||
|
||||
from src.common.logo_helper import MAX_LOGO_BYTES, LogoHelper
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _no_real_chmod(monkeypatch):
|
||||
# Keep the permission helpers out of the way: their own env detection
|
||||
# is not what these tests are about.
|
||||
monkeypatch.setattr("src.common.logo_helper.ensure_directory_permissions", MagicMock())
|
||||
monkeypatch.setattr("src.common.logo_helper.ensure_file_permissions", MagicMock())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def helper():
|
||||
return LogoHelper(display_width=64, display_height=32,
|
||||
logger=logging.getLogger("test.logo_helper"))
|
||||
|
||||
|
||||
def write_logo(path: Path, size=(20, 20), color=(255, 0, 0), fmt="PNG") -> Path:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
Image.new("RGB", size, color).save(path, format=fmt)
|
||||
return path
|
||||
|
||||
|
||||
def fake_response(content: bytes):
|
||||
response = MagicMock()
|
||||
response.content = content
|
||||
response.raise_for_status = MagicMock()
|
||||
return response
|
||||
|
||||
|
||||
def png_bytes(size=(20, 20), color=(0, 128, 0)) -> bytes:
|
||||
import io
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGB", size, color).save(buf, format="PNG")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
class TestLoadLogo:
|
||||
def test_loads_and_converts_to_rgba(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png")
|
||||
logo = helper.load_logo("PHI", path)
|
||||
assert logo is not None
|
||||
assert logo.mode == "RGBA"
|
||||
|
||||
def test_missing_file_returns_none(self, helper, tmp_path, caplog):
|
||||
with caplog.at_level(logging.WARNING):
|
||||
assert helper.load_logo("NOPE", tmp_path / "missing.png") is None
|
||||
assert "Logo not found" in caplog.text
|
||||
|
||||
def test_second_load_is_served_from_cache(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png")
|
||||
first = helper.load_logo("PHI", path)
|
||||
path.unlink() # cache hit must not touch the filesystem
|
||||
assert helper.load_logo("PHI", path) is first
|
||||
|
||||
def test_cache_key_includes_requested_size(self, helper, tmp_path):
|
||||
# A panel-size change must not hand back a logo sized for the old
|
||||
# dimensions, so the two sizes get separate cache entries.
|
||||
path = write_logo(tmp_path / "PHI.png", size=(100, 100))
|
||||
small = helper.load_logo("PHI", path, max_width=10, max_height=10)
|
||||
large = helper.load_logo("PHI", path, max_width=50, max_height=50)
|
||||
assert small is not large
|
||||
assert small.size != large.size
|
||||
assert len(helper._logo_cache) == 2
|
||||
|
||||
def test_default_size_is_one_and_a_half_display(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png", size=(500, 500))
|
||||
logo = helper.load_logo("PHI", path)
|
||||
assert logo.width <= int(64 * 1.5)
|
||||
assert logo.height <= int(32 * 1.5)
|
||||
|
||||
def test_smaller_image_is_not_upscaled(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png", size=(8, 8))
|
||||
assert helper.load_logo("PHI", path, max_width=64, max_height=64).size == (8, 8)
|
||||
|
||||
def test_larger_image_is_downscaled_preserving_aspect(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png", size=(200, 100))
|
||||
logo = helper.load_logo("PHI", path, max_width=50, max_height=50)
|
||||
assert logo.width <= 50 and logo.height <= 50
|
||||
assert logo.width == 50 and logo.height == 25 # 2:1 preserved
|
||||
|
||||
def test_string_path_accepted(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png")
|
||||
assert helper.load_logo("PHI", str(path)) is not None
|
||||
|
||||
def test_corrupt_file_returns_none(self, helper, tmp_path, caplog):
|
||||
bad = tmp_path / "bad.png"
|
||||
bad.write_bytes(b"not an image")
|
||||
with caplog.at_level(logging.ERROR):
|
||||
assert helper.load_logo("BAD", bad) is None
|
||||
assert "Error loading logo" in caplog.text
|
||||
|
||||
|
||||
class TestCacheManagement:
|
||||
def test_lru_evicts_oldest(self, tmp_path):
|
||||
helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
|
||||
paths = [write_logo(tmp_path / f"T{i}.png") for i in range(3)]
|
||||
for i, path in enumerate(paths):
|
||||
helper.load_logo(f"T{i}", path)
|
||||
assert len(helper._logo_cache) == 2
|
||||
assert not any(k.startswith("T0_") for k in helper._logo_cache)
|
||||
|
||||
def test_cache_hit_refreshes_lru_position(self, tmp_path):
|
||||
helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
|
||||
a, b, c = [write_logo(tmp_path / f"{n}.png") for n in ("A", "B", "C")]
|
||||
helper.load_logo("A", a)
|
||||
helper.load_logo("B", b)
|
||||
helper.load_logo("A", a) # A is now most-recently used
|
||||
helper.load_logo("C", c) # evicts B, not A
|
||||
assert any(k.startswith("A_") for k in helper._logo_cache)
|
||||
assert not any(k.startswith("B_") for k in helper._logo_cache)
|
||||
|
||||
def test_clear_cache_empties_both_structures(self, helper, tmp_path):
|
||||
helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
|
||||
helper.clear_cache()
|
||||
assert helper._logo_cache == {}
|
||||
assert helper._cache_order == []
|
||||
|
||||
def test_cache_stats(self, tmp_path):
|
||||
helper = LogoHelper(64, 32, cache_size=4, logger=MagicMock())
|
||||
helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
|
||||
stats = helper.get_cache_stats()
|
||||
assert stats["cached_logos"] == 1
|
||||
assert stats["cache_size_limit"] == 4
|
||||
assert stats["cache_usage_percent"] == 25
|
||||
|
||||
def test_zero_cache_size_does_not_divide_by_zero(self):
|
||||
# Regression: this raised ZeroDivisionError.
|
||||
stats = LogoHelper(64, 32, cache_size=0, logger=MagicMock()).get_cache_stats()
|
||||
assert stats["cache_usage_percent"] == 0
|
||||
assert stats["cache_size_limit"] == 0
|
||||
|
||||
|
||||
class TestLoadLogoWithDownload:
|
||||
def test_existing_file_skips_download(self, helper, tmp_path):
|
||||
path = write_logo(tmp_path / "PHI.png")
|
||||
helper.session.get = MagicMock()
|
||||
assert helper.load_logo_with_download("PHI", path, "http://x/logo.png") is not None
|
||||
helper.session.get.assert_not_called()
|
||||
|
||||
def test_downloads_then_loads(self, helper, tmp_path):
|
||||
path = tmp_path / "PHI.png"
|
||||
helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
|
||||
logo = helper.load_logo_with_download("PHI", path, "http://x/logo.png")
|
||||
assert logo is not None
|
||||
assert path.exists()
|
||||
helper.session.get.assert_called_once_with("http://x/logo.png", timeout=30)
|
||||
|
||||
def test_download_failure_falls_back_to_placeholder(self, helper, tmp_path):
|
||||
helper.session.get = MagicMock(
|
||||
side_effect=requests.RequestException("connection reset"))
|
||||
logo = helper.load_logo_with_download(
|
||||
"PHI", tmp_path / "PHI.png", "http://x/logo.png",
|
||||
max_width=20, max_height=20)
|
||||
assert logo is not None and logo.size == (20, 20) # placeholder
|
||||
|
||||
def test_http_error_falls_back_to_placeholder(self, helper, tmp_path):
|
||||
response = fake_response(b"")
|
||||
response.raise_for_status.side_effect = requests.HTTPError("404")
|
||||
helper.session.get = MagicMock(return_value=response)
|
||||
logo = helper.load_logo_with_download(
|
||||
"PHI", tmp_path / "PHI.png", "http://x/logo.png",
|
||||
max_width=20, max_height=20)
|
||||
assert logo is not None and logo.size == (20, 20)
|
||||
|
||||
def test_no_url_and_no_file_gives_placeholder(self, helper, tmp_path):
|
||||
logo = helper.load_logo_with_download(
|
||||
"PHI", tmp_path / "missing.png", None, max_width=16, max_height=16)
|
||||
assert logo is not None and logo.size == (16, 16)
|
||||
|
||||
|
||||
class TestDownloadLogo:
|
||||
def test_writes_file_and_sets_permissions(self, helper, tmp_path):
|
||||
path = tmp_path / "assets" / "PHI.png"
|
||||
# Directory creation is ensure_directory_permissions' job, and the
|
||||
# autouse fixture stubs it out — so make the directory here.
|
||||
path.parent.mkdir()
|
||||
helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
|
||||
with patch("src.common.logo_helper.ensure_directory_permissions") as dirs, \
|
||||
patch("src.common.logo_helper.ensure_file_permissions") as files:
|
||||
helper._download_logo("http://x/logo.png", path)
|
||||
assert path.exists()
|
||||
dirs.assert_called_once()
|
||||
files.assert_called_once()
|
||||
assert dirs.call_args[0][0] == path.parent
|
||||
|
||||
def test_oversized_response_is_rejected_without_writing(self, helper, tmp_path):
|
||||
# Regression: an unbounded response.content was written straight to
|
||||
# disk, so a hostile URL chose how many bytes landed in assets/.
|
||||
path = tmp_path / "huge.png"
|
||||
helper.session.get = MagicMock(
|
||||
return_value=fake_response(b"\x00" * (MAX_LOGO_BYTES + 1)))
|
||||
with pytest.raises(ValueError, match="over the"):
|
||||
helper._download_logo("http://x/huge.png", path)
|
||||
assert not path.exists()
|
||||
|
||||
def test_non_image_response_is_deleted_and_raises(self, helper, tmp_path):
|
||||
# Regression: undecodable bytes stayed on disk, so every later
|
||||
# load_logo() call hit the corrupt file instead of re-downloading.
|
||||
path = tmp_path / "bad.png"
|
||||
helper.session.get = MagicMock(return_value=fake_response(b"<html>404</html>"))
|
||||
with pytest.raises(Exception):
|
||||
helper._download_logo("http://x/bad.png", path)
|
||||
assert not path.exists()
|
||||
|
||||
def test_decompression_bomb_is_deleted_and_raises(self, helper, tmp_path, monkeypatch):
|
||||
path = tmp_path / "bomb.png"
|
||||
helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
|
||||
|
||||
class Bomb:
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def load(self):
|
||||
raise Image.DecompressionBombError("too many pixels")
|
||||
|
||||
monkeypatch.setattr("src.common.logo_helper.Image.open", lambda *a, **kw: Bomb())
|
||||
with pytest.raises(Image.DecompressionBombError):
|
||||
helper._download_logo("http://x/bomb.png", path)
|
||||
assert not path.exists()
|
||||
|
||||
def test_bad_download_surfaces_as_placeholder_not_crash(self, helper, tmp_path):
|
||||
# The new guards raise, and load_logo_with_download's existing
|
||||
# broad except turns that into the placeholder path.
|
||||
helper.session.get = MagicMock(return_value=fake_response(b"garbage"))
|
||||
logo = helper.load_logo_with_download(
|
||||
"PHI", tmp_path / "PHI.png", "http://x/bad.png",
|
||||
max_width=12, max_height=12)
|
||||
assert logo is not None and logo.size == (12, 12)
|
||||
|
||||
|
||||
class TestLogoVariations:
|
||||
def test_plain_abbreviation_returns_itself(self, helper):
|
||||
assert helper.get_logo_variations("PHI") == ["PHI"]
|
||||
|
||||
def test_ampersand_expanded(self, helper):
|
||||
assert "TAAND M" in helper.get_logo_variations("TA& M")
|
||||
|
||||
def test_and_contracted(self, helper):
|
||||
assert "T&M" in helper.get_logo_variations("TANDM")
|
||||
|
||||
def test_special_case_appends_known_aliases(self, helper):
|
||||
variations = helper.get_logo_variations("TA&M")
|
||||
assert "TAMU" in variations and "TEXASAM" in variations
|
||||
assert "TAANDM" in variations # the generic & rule still applies
|
||||
|
||||
|
||||
class TestNormalizeAbbreviation:
|
||||
def test_uppercases_and_strips(self, helper):
|
||||
assert helper.normalize_abbreviation(" phi ") == "PHI"
|
||||
|
||||
def test_ampersand_becomes_and(self, helper):
|
||||
assert helper.normalize_abbreviation("TA&M") == "TAANDM"
|
||||
|
||||
def test_internal_spaces_removed(self, helper):
|
||||
assert helper.normalize_abbreviation("New York") == "NEWYORK"
|
||||
|
||||
def test_deliberately_differs_from_logo_downloader(self, helper):
|
||||
# Pinned, not a bug: LogoDownloader.normalize_abbreviation replaces
|
||||
# filesystem-unsafe characters but keeps spaces, and plugins call
|
||||
# that one. Changing either changes which logo filenames resolve on
|
||||
# existing installs. Both docstrings say so explicitly.
|
||||
from src.logo_downloader import LogoDownloader
|
||||
assert helper.normalize_abbreviation("New York") == "NEWYORK"
|
||||
assert LogoDownloader.normalize_abbreviation("New York") == "NEW YORK"
|
||||
|
||||
|
||||
class TestPlaceholderLogo:
|
||||
def test_uses_requested_dimensions(self, helper):
|
||||
assert helper._create_placeholder_logo("PHI", 30, 20).size == (30, 20)
|
||||
|
||||
def test_defaults_to_one_and_a_half_display(self, helper):
|
||||
assert helper._create_placeholder_logo("PHI").size == (96, 48)
|
||||
|
||||
def test_is_rgba(self, helper):
|
||||
assert helper._create_placeholder_logo("PHI", 10, 10).mode == "RGBA"
|
||||
|
||||
def test_invalid_dimensions_return_none(self, helper, caplog):
|
||||
with caplog.at_level(logging.ERROR):
|
||||
assert helper._create_placeholder_logo("PHI", -5, -5) is None
|
||||
assert "Error creating placeholder" in caplog.text
|
||||
|
||||
|
||||
class TestSessionConfiguration:
|
||||
def test_user_agent_and_accept_headers(self, helper):
|
||||
assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
|
||||
assert helper.session.headers["Accept"] == "image/*"
|
||||
Reference in New Issue
Block a user