mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-10 09:06:36 +00:00
Merge remote-tracking branch 'origin/main' into HEAD
# Conflicts: # CHANGELOG.md
This commit is contained in:
Chuck
@@ -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
|
||||
|
||||
@@ -0,0 +1,283 @@
|
||||
"""The one BDF loader and the one BDF rasterizer (src/common/bdf_font.py).
|
||||
|
||||
DisplayManager, the plugin test harness (VisualTestDisplayManager), FontManager
|
||||
and element_style used to carry their own copies of both. Plugin golden images
|
||||
depend on the exact pixels, so the rasterizer is checked against a frozen copy
|
||||
of the per-pixel loop DisplayManager._draw_bdf_text ran before it was shared
|
||||
(``_reference_draw`` below) for every bundled BDF font, at native and off-strike
|
||||
sizes, clipped and unclipped. Pixels are compared as raw bytes, never PNG
|
||||
hashes, so a Pillow upgrade can't fake or mask a difference.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from pathlib import Path
|
||||
|
||||
import freetype
|
||||
import pytest
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
os.environ.setdefault("EMULATOR", "true")
|
||||
|
||||
from src.common import bdf_font
|
||||
from src.common.bdf_font import draw_bdf_text, load_bdf_face, read_bdf_native_size
|
||||
|
||||
FONTS_DIR = Path(__file__).resolve().parent.parent / "assets" / "fonts"
|
||||
BDF_FONTS = sorted(p.name for p in FONTS_DIR.glob("*.bdf"))
|
||||
|
||||
STRINGS = [
|
||||
"Hello, World!", "0123456789", "12:34 PM", "!\"#$%&'()*+,-./:;<=>?@[\\]^_`{|}~",
|
||||
"AaBbGgJjQqYy|", "", " ", "café üñ 72°F —€",
|
||||
]
|
||||
|
||||
|
||||
def _reference_draw(draw, width, height, text, x, y, face, color):
|
||||
"""DisplayManager._draw_bdf_text as it was before the shared rasterizer.
|
||||
|
||||
Frozen on purpose: it is the definition of the panel's output that the
|
||||
fast path must reproduce. Do not "fix" it.
|
||||
"""
|
||||
try:
|
||||
ascender_px = face.size.ascender >> 6
|
||||
except Exception:
|
||||
ascender_px = 0
|
||||
baseline_y = y + ascender_px
|
||||
for char in text:
|
||||
face.load_char(char)
|
||||
bitmap = face.glyph.bitmap
|
||||
glyph_left = face.glyph.bitmap_left
|
||||
glyph_top = face.glyph.bitmap_top
|
||||
for i in range(bitmap.rows):
|
||||
for j in range(bitmap.width):
|
||||
byte_index = i * bitmap.pitch + (j // 8)
|
||||
if byte_index < len(bitmap.buffer):
|
||||
byte = bitmap.buffer[byte_index]
|
||||
if byte & (1 << (7 - (j % 8))):
|
||||
pixel_x = x + glyph_left + j
|
||||
pixel_y = baseline_y - glyph_top + i
|
||||
if 0 <= pixel_x < width and 0 <= pixel_y < height:
|
||||
draw.point((pixel_x, pixel_y), fill=color)
|
||||
x += face.glyph.advance.x >> 6
|
||||
|
||||
|
||||
def _pair(size=(64, 32), mode="RGB", draw_mode=None):
|
||||
a = Image.new(mode, size)
|
||||
b = Image.new(mode, size)
|
||||
return a, ImageDraw.Draw(a, draw_mode), b, ImageDraw.Draw(b, draw_mode)
|
||||
|
||||
|
||||
def _assert_same(expected, actual, what):
|
||||
assert expected.tobytes() == actual.tobytes(), what
|
||||
|
||||
|
||||
def _sizes(name):
|
||||
native = read_bdf_native_size(str(FONTS_DIR / name))
|
||||
return sorted({native, native + 3, max(1, native - 2)})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- rasterizer
|
||||
|
||||
@pytest.mark.parametrize("name", BDF_FONTS)
|
||||
def test_every_bundled_font_matches_the_reference_raster(name):
|
||||
path = str(FONTS_DIR / name)
|
||||
w, h = 96, 24
|
||||
cases = [((0, 0), (255, 255, 255), text) for text in STRINGS]
|
||||
# Clipped on every edge, in a colour that isn't all-or-nothing per channel.
|
||||
cases += [(xy, (12, 200, 77), text)
|
||||
for xy in ((-4, -3), (w - 11, h - 5), (w // 2, -9))
|
||||
for text in STRINGS[:2]]
|
||||
for size in _sizes(name):
|
||||
face, _ = load_bdf_face(path, size)
|
||||
for (x, y), color, text in cases:
|
||||
ref, rdraw, new, ndraw = _pair((w, h))
|
||||
_reference_draw(rdraw, w, h, text, x, y, face, color)
|
||||
draw_bdf_text(ndraw, text, x, y, face, color)
|
||||
_assert_same(ref, new, (name, size, x, y, color, text))
|
||||
|
||||
|
||||
def test_returns_the_pen_position():
|
||||
face, _ = load_bdf_face(str(FONTS_DIR / "5x7.bdf"), 7)
|
||||
_, _, _, draw = _pair()
|
||||
assert draw_bdf_text(draw, "", 3, 0, face) == 3
|
||||
assert draw_bdf_text(draw, "12:34", 3, 0, face) == 3 + 5 * 5
|
||||
|
||||
|
||||
def test_clip_smaller_than_the_canvas_matches_the_reference():
|
||||
# DisplayManager clips to its logical size, which is the canvas size in
|
||||
# practice but not by construction; honour the clip as the loop did.
|
||||
face, _ = load_bdf_face(str(FONTS_DIR / "6x10.bdf"), 10)
|
||||
for clip in ((20, 7), (1, 1), (0, 0), (200, 200)):
|
||||
ref, rdraw, new, ndraw = _pair()
|
||||
_reference_draw(rdraw, clip[0], clip[1], "Mixed 123", -2, -1, face, (1, 2, 3))
|
||||
draw_bdf_text(ndraw, "Mixed 123", -2, -1, face, (1, 2, 3), clip=clip)
|
||||
_assert_same(ref, new, clip)
|
||||
|
||||
|
||||
def test_blending_draw_blends_exactly_like_the_reference():
|
||||
# ImageDraw.Draw(rgb, "RGBA") blends a translucent colour; a mask fill
|
||||
# would not, so this path must fall back to points.
|
||||
face, _ = load_bdf_face(str(FONTS_DIR / "7x13B.bdf"), 13)
|
||||
ref, rdraw, new, ndraw = _pair(draw_mode="RGBA")
|
||||
for img in (ref, new):
|
||||
img.paste((40, 80, 120), (0, 0, *img.size))
|
||||
_reference_draw(rdraw, 64, 32, "Blend", 1, 1, face, (255, 0, 0, 128))
|
||||
draw_bdf_text(ndraw, "Blend", 1, 1, face, (255, 0, 0, 128))
|
||||
_assert_same(ref, new, "blend")
|
||||
colours = {c for _, c in new.getcolors()}
|
||||
assert (255, 0, 0) not in colours and len(colours) == 2, colours # blended
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode,color", [("L", 200), ("P", (255, 0, 0)), ("RGBA", (9, 8, 7, 255)), ("1", 1)])
|
||||
def test_other_canvas_modes_match_the_reference(mode, color):
|
||||
face, _ = load_bdf_face(str(FONTS_DIR / "5x8.bdf"), 8)
|
||||
ref, rdraw, new, ndraw = _pair(mode=mode)
|
||||
_reference_draw(rdraw, 64, 32, "Mode 42", 2, 2, face, color)
|
||||
draw_bdf_text(ndraw, "Mode 42", 2, 2, face, color)
|
||||
_assert_same(ref, new, mode)
|
||||
|
||||
|
||||
def test_a_non_mono_face_reads_the_same_bits_as_the_reference():
|
||||
# A plugin can hand draw_text a freetype.Face of a TTF: 8-bit gray glyphs
|
||||
# whose pitch is not ceil(width/8). The loop read them as packed bits, and
|
||||
# so must the fast path -- same (odd) pixels, not "better" ones.
|
||||
face = freetype.Face(str(FONTS_DIR / "PressStart2P-Regular.ttf"))
|
||||
face.set_char_size(8 * 64, 8 * 64, 72, 72)
|
||||
ref, rdraw, new, ndraw = _pair()
|
||||
_reference_draw(rdraw, 64, 32, "Gray", 0, 0, face, (255, 255, 255))
|
||||
draw_bdf_text(ndraw, "Gray", 0, 0, face, (255, 255, 255))
|
||||
_assert_same(ref, new, "gray face")
|
||||
|
||||
|
||||
def test_errors_propagate_after_earlier_glyphs_are_drawn():
|
||||
face, _ = load_bdf_face(str(FONTS_DIR / "5x7.bdf"), 7)
|
||||
_, _, img, draw = _pair()
|
||||
with pytest.raises(Exception):
|
||||
draw_bdf_text(draw, "A", 0, 0, face, "not-a-colour")
|
||||
# Blank glyphs never touch the colour, as before.
|
||||
draw_bdf_text(draw, " ", 0, 0, face, "not-a-colour")
|
||||
assert img.getbbox() is None
|
||||
with pytest.raises(Exception):
|
||||
draw_bdf_text(draw, "A", 0, 0, object())
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- callers
|
||||
|
||||
def _dm_stub(img):
|
||||
from src.display_manager import DisplayManager
|
||||
stub = types.SimpleNamespace(width=img.width, height=img.height,
|
||||
draw=ImageDraw.Draw(img), calendar_font=None)
|
||||
return DisplayManager, stub
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["5x7.bdf", "tom-thumb.bdf", "9x18B.bdf", "MatrixChunky8X.bdf"])
|
||||
def test_display_manager_and_test_harness_draw_the_reference_pixels(name):
|
||||
from src.plugin_system.testing.visual_display_manager import VisualTestDisplayManager
|
||||
|
||||
face, _ = load_bdf_face(str(FONTS_DIR / name), 20) # off-strike for all four
|
||||
for text in STRINGS:
|
||||
ref, rdraw, dm_img, _ = _pair((64, 32))
|
||||
_reference_draw(rdraw, 64, 32, text, -1, 3, face, (255, 128, 0))
|
||||
|
||||
DisplayManager, stub = _dm_stub(dm_img)
|
||||
DisplayManager._draw_bdf_text(stub, text, -1, 3, (255, 128, 0), face)
|
||||
_assert_same(ref, dm_img, ("DisplayManager", name, text))
|
||||
|
||||
vt = VisualTestDisplayManager(64, 32)
|
||||
vt.draw_text(text, -1, 3, (255, 128, 0), font=face)
|
||||
_assert_same(ref, vt.image, ("VisualTestDisplayManager", name, text))
|
||||
|
||||
|
||||
def test_harness_calendar_font_is_the_panels():
|
||||
# The harness built a bare freetype.Face with no size: ascender 0, so
|
||||
# every calendar_font line drew a baseline too high, and its
|
||||
# get_font_height() returned 0.
|
||||
from src.plugin_system.testing.visual_display_manager import VisualTestDisplayManager
|
||||
|
||||
vt = VisualTestDisplayManager(64, 32)
|
||||
panel_face, _ = load_bdf_face(str(FONTS_DIR / "5x7.bdf"), 7)
|
||||
assert vt.calendar_font is panel_face
|
||||
assert vt.get_font_height(vt.calendar_font) == panel_face.size.height >> 6 > 0
|
||||
|
||||
|
||||
# -------------------------------------------------------------------- loader
|
||||
|
||||
def test_off_strike_size_loads_the_native_strike():
|
||||
path = str(FONTS_DIR / "5x7.bdf")
|
||||
face, realised = load_bdf_face(path, 10)
|
||||
assert isinstance(face, freetype.Face)
|
||||
assert realised == 7 and face.size.y_ppem == 7
|
||||
assert load_bdf_face(path, 7)[1] == 7
|
||||
|
||||
|
||||
def test_faces_are_cached_and_shared_by_every_loader():
|
||||
from src.element_style import _load_bdf
|
||||
from src.font_manager import FontManager
|
||||
|
||||
path = str(FONTS_DIR / "6x10.bdf")
|
||||
face, realised = load_bdf_face(path, 12)
|
||||
assert load_bdf_face(path, 12)[0] is face
|
||||
assert _load_bdf(path, 12) == (face, realised)
|
||||
assert FontManager({})._load_bdf_font(path, 12) is face
|
||||
|
||||
|
||||
def test_touched_file_is_reloaded(tmp_path):
|
||||
# The cache key carries the file's mtime, so a font re-uploaded under the
|
||||
# same name is not served from a stale face.
|
||||
target = tmp_path / "f.bdf"
|
||||
target.write_bytes((FONTS_DIR / "5x7.bdf").read_bytes())
|
||||
first, _ = load_bdf_face(str(target), 7)
|
||||
assert load_bdf_face(str(target), 7)[0] is first
|
||||
os.utime(target, ns=(10**9, 10**9))
|
||||
assert load_bdf_face(str(target), 7)[0] is not first
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == "win32",
|
||||
reason="FreeType holds the font file open; Windows refuses to overwrite it")
|
||||
def test_replaced_file_is_reloaded(tmp_path):
|
||||
target = tmp_path / "f.bdf"
|
||||
target.write_bytes((FONTS_DIR / "5x7.bdf").read_bytes())
|
||||
first, _ = load_bdf_face(str(target), 7)
|
||||
target.write_bytes((FONTS_DIR / "6x10.bdf").read_bytes())
|
||||
os.utime(target, ns=(1, 1))
|
||||
second, realised = load_bdf_face(str(target), 7)
|
||||
assert second is not first and realised == 10
|
||||
|
||||
|
||||
def test_unloadable_file_raises(tmp_path):
|
||||
with pytest.raises(Exception):
|
||||
load_bdf_face(str(tmp_path / "missing.bdf"), 7)
|
||||
bad = tmp_path / "bad.bdf"
|
||||
bad.write_text("not a font")
|
||||
with pytest.raises(Exception):
|
||||
load_bdf_face(str(bad), 7)
|
||||
|
||||
|
||||
def test_cache_is_bounded(monkeypatch):
|
||||
monkeypatch.setattr(bdf_font, "_FACE_CACHE_MAX", 2)
|
||||
bdf_font.clear_face_cache()
|
||||
path = str(FONTS_DIR / "5x7.bdf")
|
||||
for size in (7, 8, 9):
|
||||
load_bdf_face(path, size)
|
||||
assert len(bdf_font._face_cache) == 2
|
||||
bdf_font.clear_face_cache()
|
||||
|
||||
|
||||
def test_each_thread_gets_its_own_face():
|
||||
# FreeType forbids two threads using one face at once: load_char rewrites
|
||||
# the face's glyph slot. Within a thread the face is shared.
|
||||
import threading
|
||||
path = str(FONTS_DIR / "5x7.bdf")
|
||||
here, _ = load_bdf_face(path, 7)
|
||||
assert load_bdf_face(path, 7)[0] is here
|
||||
other = []
|
||||
worker = threading.Thread(target=lambda: other.append(load_bdf_face(path, 7)[0]))
|
||||
worker.start()
|
||||
worker.join()
|
||||
assert other and other[0] is not here
|
||||
|
||||
|
||||
def test_native_size_prefers_pixel_size_over_point_size():
|
||||
# 6x13.bdf is defined at 75dpi: SIZE says 12 (points), PIXEL_SIZE 13.
|
||||
assert read_bdf_native_size(str(FONTS_DIR / "6x13.bdf")) == 13
|
||||
assert read_bdf_native_size(str(FONTS_DIR / "nope.bdf")) is None
|
||||
@@ -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
|
||||
@@ -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:<redacted>@example.com/api"),
|
||||
("HTTPS://USER:HUNTER2@EXAMPLE.COM",
|
||||
"HTTPS://USER:<redacted>@EXAMPLE.COM"),
|
||||
("git+ssh://deploy:hunter2@host/repo",
|
||||
"git+ssh://deploy:<redacted>@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:<redacted>@host"),
|
||||
("+.-http://user:hunter2@host", "+.-http://user:<redacted>@host"),
|
||||
("a1+http://user:hunter2@host", "a1+http://user:<redacted>@host"),
|
||||
("see a://u:first@b and c://v:second@d",
|
||||
"see a://u:<redacted>@b and c://v:<redacted>@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 <redacted>"),
|
||||
("Proxy-Authorization: Basic dXNlcg==", "Proxy-Authorization: Basic <redacted>"),
|
||||
("authorization: barecredential", "authorization: <redacted>"),
|
||||
# Whitespace and an opening quote around the value, in either order.
|
||||
('authorization=" Bearer tok"', 'authorization=" Bearer <redacted>"'),
|
||||
("authorization: ' tok'", "authorization: ' <redacted>'"),
|
||||
("authorization:\n\tBearer tok", "authorization:\n\tBearer <redacted>"),
|
||||
])
|
||||
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:<redacted>@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 <redacted>"
|
||||
assert elapsed < _TIME_LIMIT
|
||||
@@ -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"<raises {type(exc).__name__}>"
|
||||
|
||||
|
||||
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 <side>_abbr / <side>_id keys;
|
||||
# the card also reads <side>_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)
|
||||
@@ -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.
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}")
|
||||
|
||||
@@ -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
|
||||
@@ -1,9 +1,14 @@
|
||||
"""Tests for the web interface's in-memory cache helpers."""
|
||||
import sys
|
||||
import threading
|
||||
from typing import Iterator
|
||||
|
||||
import pytest
|
||||
|
||||
from web_interface.cache import delete_cached, get_cached, invalidate_cache, set_cached
|
||||
from web_interface import cache as cache_module
|
||||
from web_interface.cache import (
|
||||
TTLCache, delete_cached, get_cached, invalidate_cache, set_cached,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@@ -44,3 +49,142 @@ def test_invalidate_cache_pattern() -> None:
|
||||
invalidate_cache('fonts')
|
||||
assert get_cached('fonts_catalog') is None
|
||||
assert get_cached('plugins_list') == 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Expiry. set_cached used to accept ttl_seconds and ignore it; only the TTL a
|
||||
# reader passed to get_cached counted, and get_cached defaulted to 60s.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _Clock:
|
||||
def __init__(self) -> None:
|
||||
self.now = 1000.0
|
||||
|
||||
def __call__(self) -> float:
|
||||
return self.now
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clock() -> _Clock:
|
||||
return _Clock()
|
||||
|
||||
|
||||
def test_entry_expires_after_its_ttl(clock: _Clock) -> None:
|
||||
c = TTLCache(clock=clock)
|
||||
c.set('k', 'v', ttl=10)
|
||||
clock.now += 9.9
|
||||
assert c.get('k') == 'v'
|
||||
clock.now += 0.1
|
||||
assert c.get('k') is None
|
||||
|
||||
|
||||
def test_default_ttl_applies_when_none_given(clock: _Clock) -> None:
|
||||
c = TTLCache(default_ttl=5, clock=clock)
|
||||
c.set('k', 'v')
|
||||
clock.now += 4.9
|
||||
assert c.get('k') == 'v'
|
||||
clock.now += 0.1
|
||||
assert c.get('k') is None
|
||||
|
||||
|
||||
def test_reader_max_age_can_only_shorten(clock: _Clock) -> None:
|
||||
c = TTLCache(clock=clock)
|
||||
c.set('k', 'v', ttl=10)
|
||||
clock.now += 5
|
||||
assert c.get('k', max_age=6) == 'v'
|
||||
assert c.get('k', max_age=5) is None
|
||||
clock.now += 5
|
||||
assert c.get('k', max_age=60) is None, "a reader extended a 10s entry"
|
||||
|
||||
|
||||
def test_set_cached_ttl_is_honoured(monkeypatch: pytest.MonkeyPatch, clock: _Clock) -> None:
|
||||
monkeypatch.setattr(cache_module, '_default_cache', TTLCache(clock=clock))
|
||||
set_cached('short', 1, ttl_seconds=2)
|
||||
set_cached('long', 2, ttl_seconds=300)
|
||||
clock.now += 2
|
||||
assert get_cached('short') is None, "set_cached ignored its ttl_seconds"
|
||||
clock.now += 100 # past the old implicit 60s read default
|
||||
assert get_cached('long') == 2
|
||||
|
||||
|
||||
def test_get_cached_ttl_still_bounds_the_read(monkeypatch: pytest.MonkeyPatch, clock: _Clock) -> None:
|
||||
"""The existing callers pass the TTL on both sides; that keeps working."""
|
||||
monkeypatch.setattr(cache_module, '_default_cache', TTLCache(clock=clock))
|
||||
set_cached('system_status', {'cpu': 1}, ttl_seconds=10)
|
||||
clock.now += 9
|
||||
assert get_cached('system_status', ttl_seconds=10) == {'cpu': 1}
|
||||
clock.now += 1
|
||||
assert get_cached('system_status', ttl_seconds=10) is None
|
||||
|
||||
|
||||
def test_peek_returns_the_last_value_after_expiry(clock: _Clock) -> None:
|
||||
c = TTLCache(clock=clock)
|
||||
assert c.peek('k', 'fallback') == 'fallback'
|
||||
c.set('k', True, ttl=1)
|
||||
clock.now += 5
|
||||
assert c.get('k') is None
|
||||
assert c.peek('k', False) is True
|
||||
|
||||
|
||||
def test_falsy_values_are_cached(clock: _Clock) -> None:
|
||||
c = TTLCache(clock=clock)
|
||||
c.set('k', False, ttl=10)
|
||||
assert c.get('k', default='miss') is False
|
||||
|
||||
|
||||
def test_clear_pattern_on_instance() -> None:
|
||||
c = TTLCache()
|
||||
c.set('fonts_catalog', 1)
|
||||
c.set('system_status', 2)
|
||||
c.clear('fonts')
|
||||
assert c.peek('fonts_catalog') is None
|
||||
assert c.get('system_status') == 2
|
||||
c.clear()
|
||||
assert c.peek('system_status') is None
|
||||
|
||||
|
||||
def test_concurrent_expiry_reads_and_writes_do_not_raise() -> None:
|
||||
"""The old dicts deleted expired keys inside get; two threads reading the
|
||||
same expired key (or one reading while another invalidated) could raise
|
||||
KeyError, which the endpoints turned into a 500."""
|
||||
c = TTLCache()
|
||||
keys = [f'k{n}' for n in range(8)]
|
||||
errors = []
|
||||
stop = threading.Event()
|
||||
|
||||
def reader() -> None:
|
||||
try:
|
||||
while not stop.is_set():
|
||||
for key in keys:
|
||||
c.get(key, max_age=0) # always expired for this reader
|
||||
c.get(key)
|
||||
c.peek(key)
|
||||
c.clear('k1')
|
||||
except Exception as exc: # pragma: no cover - the failure being tested
|
||||
errors.append(exc)
|
||||
|
||||
def writer() -> None:
|
||||
try:
|
||||
for i in range(20000):
|
||||
key = keys[i % len(keys)]
|
||||
c.set(key, i, ttl=0 if i % 2 else 60)
|
||||
if i % 7 == 0:
|
||||
c.delete(key)
|
||||
except Exception as exc: # pragma: no cover
|
||||
errors.append(exc)
|
||||
|
||||
# Switch threads as often as possible so an unlocked check-then-act
|
||||
# actually gets interleaved within the test's run time.
|
||||
old_interval = sys.getswitchinterval()
|
||||
sys.setswitchinterval(1e-6)
|
||||
try:
|
||||
readers = [threading.Thread(target=reader) for _ in range(4)]
|
||||
for t in readers:
|
||||
t.start()
|
||||
writer()
|
||||
stop.set()
|
||||
for t in readers:
|
||||
t.join()
|
||||
finally:
|
||||
sys.setswitchinterval(old_interval)
|
||||
assert errors == []
|
||||
|
||||
@@ -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))
|
||||
)
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
"""The web process logs the way the display process does.
|
||||
|
||||
web_interface/app.py used to call its own setup (web_interface/logging_config.py)
|
||||
which replaced the root handlers with a plain stdout formatter. Under systemd
|
||||
every line then reached the journal as PRIORITY=6, so
|
||||
|
||||
journalctl -p err -u ledmatrix-web
|
||||
|
||||
showed nothing while the web interface was logging errors. It also logged
|
||||
every request at INFO, including what the UI polls: the journal on a Pi showed
|
||||
``GET /api/v3/errors/summary - 200`` every minute per open tab.
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
|
||||
from web_interface import request_logging
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The real app, imported the way systemd runs it
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CHILD = textwrap.dedent("""
|
||||
import logging
|
||||
import web_interface.app as web_app
|
||||
|
||||
# Startup reconciliation may try to reinstall plugins; not this test's job.
|
||||
web_app._reconciliation_started = True
|
||||
client = web_app.app.test_client()
|
||||
client.get('/api/v3/errors/summary')
|
||||
client.get('/favicon.ico')
|
||||
client.get('/api/v3/no-such-endpoint')
|
||||
logging.getLogger('web_interface.probe').error('probe error line')
|
||||
logging.getLogger('web_interface.probe').info('probe info line')
|
||||
""")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def journal_output(tmp_path_factory):
|
||||
"""Run the child with stdout as a file systemd would call the journal.
|
||||
|
||||
systemd sets JOURNAL_STREAM to the dev:ino of the stream it captures;
|
||||
src.logging_config only adds priorities when stdout really is that stream,
|
||||
so hand the child a file and name that file's dev:ino.
|
||||
"""
|
||||
out_path = tmp_path_factory.mktemp("journal") / "stdout.txt"
|
||||
with open(out_path, "wb") as out:
|
||||
st = os.fstat(out.fileno())
|
||||
env = dict(os.environ)
|
||||
env.update({
|
||||
"JOURNAL_STREAM": f"{st.st_dev}:{st.st_ino}",
|
||||
"PYTHONUTF8": "1",
|
||||
"EMULATOR": "true",
|
||||
"PYTHONPATH": str(PROJECT_ROOT),
|
||||
})
|
||||
env.pop("LEDMATRIX_DEBUG", None)
|
||||
env.pop("LEDMATRIX_JSON_LOGGING", None)
|
||||
proc = subprocess.run(
|
||||
[sys.executable, "-c", _CHILD], cwd=str(PROJECT_ROOT), env=env,
|
||||
stdout=out, stderr=subprocess.PIPE, timeout=180,
|
||||
)
|
||||
text = out_path.read_text(encoding="utf-8", errors="replace")
|
||||
assert proc.returncode == 0, proc.stderr.decode(errors="replace")[-4000:]
|
||||
return text.splitlines()
|
||||
|
||||
|
||||
def test_error_reaches_the_journal_as_err(journal_output):
|
||||
lines = [l for l in journal_output if "probe error line" in l]
|
||||
assert lines, "\n".join(journal_output[-40:])
|
||||
assert lines[0].startswith("<3>"), lines[0]
|
||||
# Same readable shape as the display service (and what the log viewer strips).
|
||||
assert " - ERROR - web_interface.probe - probe error line" in lines[0]
|
||||
|
||||
|
||||
def test_info_reaches_the_journal_as_info(journal_output):
|
||||
lines = [l for l in journal_output if "probe info line" in l]
|
||||
assert lines and lines[0].startswith("<6>"), journal_output[-40:]
|
||||
|
||||
|
||||
def test_polling_gets_are_not_logged_at_info(journal_output):
|
||||
for path in ("/api/v3/errors/summary", "/favicon.ico"):
|
||||
assert not [l for l in journal_output if f"GET {path} " in l], (
|
||||
f"a successful GET {path} was logged by default")
|
||||
|
||||
|
||||
def test_failed_request_is_still_logged(journal_output):
|
||||
lines = [l for l in journal_output if "GET /api/v3/no-such-endpoint - 404" in l]
|
||||
assert lines and lines[0].startswith("<4>"), journal_output[-40:]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The level policy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("method,status,level", [
|
||||
("GET", 200, logging.DEBUG),
|
||||
("GET", 304, logging.DEBUG),
|
||||
("HEAD", 200, logging.DEBUG),
|
||||
("OPTIONS", 204, logging.DEBUG),
|
||||
("get", 200, logging.DEBUG),
|
||||
("POST", 200, logging.INFO),
|
||||
("PUT", 204, logging.INFO),
|
||||
("DELETE", 200, logging.INFO),
|
||||
("PATCH", 302, logging.INFO),
|
||||
("GET", 404, logging.WARNING),
|
||||
("POST", 400, logging.WARNING),
|
||||
("GET", 500, logging.ERROR),
|
||||
("POST", 503, logging.ERROR),
|
||||
])
|
||||
def test_request_log_level(method, status, level):
|
||||
assert request_logging.request_log_level(method, status) == level
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tiny_app():
|
||||
app = Flask(__name__)
|
||||
request_logging.init_app(app)
|
||||
|
||||
@app.route("/poll")
|
||||
def poll():
|
||||
return "ok"
|
||||
|
||||
@app.route("/save", methods=["POST"])
|
||||
def save():
|
||||
return "saved"
|
||||
|
||||
@app.route("/boom")
|
||||
def boom():
|
||||
return "no", 500
|
||||
|
||||
return app.test_client()
|
||||
|
||||
|
||||
def test_hooks_log_each_request_once_at_its_level(tiny_app, caplog):
|
||||
caplog.set_level(logging.DEBUG, logger="web_interface.api")
|
||||
tiny_app.get("/poll")
|
||||
tiny_app.post("/save")
|
||||
tiny_app.get("/boom")
|
||||
tiny_app.get("/missing")
|
||||
got = [(r.levelno, r.getMessage().split(" (")[0]) for r in caplog.records
|
||||
if r.name == "web_interface.api"]
|
||||
assert got == [
|
||||
(logging.DEBUG, "GET /poll - 200"),
|
||||
(logging.INFO, "POST /save - 200"),
|
||||
(logging.ERROR, "GET /boom - 500"),
|
||||
(logging.WARNING, "GET /missing - 404"),
|
||||
]
|
||||
|
||||
|
||||
def test_duration_is_rounded(tiny_app, caplog):
|
||||
caplog.set_level(logging.DEBUG, logger="web_interface.api")
|
||||
tiny_app.post("/save")
|
||||
msg = caplog.records[-1].getMessage()
|
||||
duration = msg.rsplit("(", 1)[1]
|
||||
assert duration.endswith("ms)") and len(duration.split(".")[1]) == len("0ms)"), msg
|
||||
|
||||
|
||||
def test_success_response_timing_uses_the_same_clock():
|
||||
# request_logging stamps request.start_time from perf_counter; a reader
|
||||
# subtracting it from time.time() reported ~1.8e12 ms (found on a Pi).
|
||||
from src.web_interface.api_helpers import success_response
|
||||
app = Flask(__name__)
|
||||
request_logging.init_app(app)
|
||||
|
||||
@app.route('/timed')
|
||||
def timed():
|
||||
return success_response(data={}, metadata={})
|
||||
|
||||
body = app.test_client().get('/timed').get_json()
|
||||
assert 0 <= body['metadata']['response_time_ms'] < 10_000
|
||||
Reference in New Issue
Block a user