Merge remote-tracking branch 'origin/main' into HEAD

This commit is contained in:
t
2026-09-07 11:35:44 -04:00
committed by ChuckBuilds
53 changed files with 7006 additions and 163 deletions
+85
View File
@@ -0,0 +1,85 @@
"""Non-finite JSON numbers must be rejected, not raise.
json.loads accepts Infinity/-Infinity/NaN by default (they are not valid JSON,
but Python's parser emits them) and Flask's get_json passes them straight
through. int(float('inf')) raises OverflowError, which is neither ValueError
nor TypeError -- so validation blocks that carefully caught those let it
through and Flask turned it into a 500.
The damage was not the status code. /config/dim-schedule answered with
CONFIG_SAVE_FAILED and suggested "Check file permissions on config directory"
and "Check available disk space" for what was actually an invalid number.
NaN already returned 400 (int(nan) raises ValueError), which is why this only
showed up for the infinities.
"""
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
#: (route, field) that returned 500 before OverflowError was caught. Both
#: infinity signs are exercised: int() raises OverflowError for either, but
#: only one of them was in the original report, and a guard that special-cased
#: the sign would pass a one-sided test.
NON_FINITE_ROUTES = [
('/api/v3/config/dim-schedule', 'dim_brightness'),
('/api/v3/errors/clear', 'max_age_hours'),
('/api/v3/config/main', 'multiplexing'),
('/api/v3/config/main', 'row_address_type'),
]
NON_FINITE_CASES = [
(route, '{"%s": %s}' % (field, literal))
for route, field in NON_FINITE_ROUTES
for literal in ('Infinity', '-Infinity')
]
@pytest.mark.parametrize("route,body", NON_FINITE_CASES)
def test_infinity_is_a_client_error_not_a_server_error(api_v3_client, route, body):
"""Exactly 400, not merely "some 4xx".
Accepting any 4xx would let a 404 pass, so renaming one of these routes
would leave the test green while testing nothing -- the failure mode this
whole file exists to catch.
"""
response = api_v3_client.post(route, data=body, content_type='application/json')
assert response.status_code == 400, (
f"{route} with {body} answered {response.status_code}; expected 400"
)
@pytest.mark.parametrize("route,body", [
('/api/v3/config/dim-schedule', '{"dim_brightness": NaN}'),
('/api/v3/errors/clear', '{"max_age_hours": NaN}'),
])
def test_nan_is_also_a_client_error(api_v3_client, route, body):
"""int(nan) raises ValueError so this path already worked -- pinned so a
refactor that narrows the except tuple cannot quietly break it."""
response = api_v3_client.post(route, data=body, content_type='application/json')
assert response.status_code == 400
def test_a_valid_number_is_accepted(api_v3_client, api_v3_module, monkeypatch):
"""Prove the widened except did not start swallowing ordinary input.
Asserting "not a 400" would not show that: the mocked save path fails for
any input, so the assertion would hold even if validation had rejected the
value. Give load_config a real dict and stub the atomic save, and the
endpoint reaches its success response -- which only happens if 30 passed
validation.
"""
api_v3_module.api_v3.config_manager.load_config.return_value = {}
monkeypatch.setattr(api_v3_module, '_save_config_atomic',
lambda *a, **k: (True, ''))
response = api_v3_client.post(
'/api/v3/config/dim-schedule',
data='{"dim_brightness": 30}',
content_type='application/json',
)
assert response.status_code == 200, response.get_data(as_text=True)[:200]
+450
View File
@@ -0,0 +1,450 @@
"""A second request for a key already being fetched must join, not duplicate.
request_id embeds a millisecond timestamp and active_requests is keyed by it,
so every submit looked new and nothing compared what was actually being
fetched. On a real board the season-schedule cache_key is requested by both
the Recent and the Upcoming manager: they miss the cache in the same
millisecond and each start a full download and parse of the same payload.
Measured on a running board, 138 background fetches in 24 hours arriving in
pairs at identical timestamps -- half of them redundant.
The cost of a duplicate is a second download, a second JSON parse (the
expensive part on a Pi), and a second parsed copy resident at the same time.
Schedules on that board run from 256KB to 20MB. It also consumes a second of
the three executor slots with identical work, which is what makes two large
parses peak simultaneously.
"""
import threading
import time
from unittest.mock import MagicMock, Mock, patch
import pytest
import requests
from src.background_data_service import BackgroundDataService, FetchStatus
PAYLOAD = {"events": [{"id": f"g{i}"} for i in range(20)]}
@pytest.fixture
def cache():
m = MagicMock()
m.get.return_value = None # always a miss: force the fetch path
m.set.return_value = None
return m
@pytest.fixture
def service(cache):
svc = BackgroundDataService(cache, max_workers=3, request_timeout=5)
yield svc
svc.shutdown(wait=False)
def _resp():
r = Mock()
r.json.return_value = PAYLOAD
r.raise_for_status.return_value = None
return r
def _wait(service, req_id, timeout=5):
deadline = time.time() + timeout
while not service.is_request_complete(req_id) and time.time() < deadline:
time.sleep(0.02)
class _BlockingSession:
"""Holds the first fetch open so a second can be submitted mid-flight."""
def __init__(self):
self.calls = 0
self.release = threading.Event()
self.started = threading.Event()
def get(self, *a, **k):
self.calls += 1
self.started.set()
self.release.wait(timeout=5)
return _resp()
def test_a_second_submit_for_the_same_key_does_not_fetch_twice(service):
session = _BlockingSession()
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="nba_2026",
callback=lambda r: None, max_retries=0)
assert session.started.wait(timeout=5)
second = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="nba_2026",
callback=lambda r: None, max_retries=0)
assert second == first, "the joiner should share the in-flight request id"
session.release.set()
_wait(service, first)
assert session.calls == 1, f"the payload was fetched {session.calls} times"
def test_the_joiner_still_gets_its_callback(service):
session = _BlockingSession()
seen = []
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: seen.append("first"), max_retries=0)
assert session.started.wait(timeout=5)
joined = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: seen.append("second"), max_retries=0)
# Assert the coalescing happened, otherwise this passes trivially:
# two independent requests would each fire their own callback and the
# test would say nothing about the joined path.
assert joined == first
session.release.set()
_wait(service, first)
deadline = time.time() + 5
while len(seen) < 2 and time.time() < deadline:
time.sleep(0.02)
assert sorted(seen) == ["first", "second"], (
f"both submitters must be called back, got {seen}")
def test_one_callback_raising_does_not_silence_the_other(service):
session = _BlockingSession()
seen = []
def boom(result):
raise RuntimeError("consumer blew up")
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=boom, max_retries=0)
assert session.started.wait(timeout=5)
joined = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: seen.append("survivor"), max_retries=0)
# Same reason: without coalescing these are separate requests and
# neither callback can affect the other.
assert joined == first
session.release.set()
_wait(service, first)
deadline = time.time() + 5
while not seen and time.time() < deadline:
time.sleep(0.02)
assert seen == ["survivor"]
def test_different_keys_are_not_coalesced(service):
session = _BlockingSession()
with patch.object(service, "session", session):
a = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/a", cache_key="key_a",
callback=lambda r: None, max_retries=0)
assert session.started.wait(timeout=5)
b = service.submit_fetch_request(
sport="nhl", year=2026, url="https://x/b", cache_key="key_b",
callback=lambda r: None, max_retries=0)
assert a != b, "different cache keys must not share a request"
session.release.set()
_wait(service, a)
_wait(service, b)
assert session.calls == 2
def test_a_later_submit_after_completion_fetches_again(service):
"""Dedupe is for concurrent requests only, not a second cache layer."""
with patch.object(service.session, "get", side_effect=[_resp(), _resp()]) as get:
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: None, max_retries=0)
_wait(service, first)
second = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: None, max_retries=0)
_wait(service, second)
assert first != second
assert get.call_count == 2
def test_cancelling_releases_the_key(service):
"""A cancelled request must not wedge its key against future fetches."""
session = _BlockingSession()
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: None, max_retries=0)
assert session.started.wait(timeout=5)
service.cancel_request(first)
assert "k" not in service._inflight_by_cache_key
session.release.set()
def test_a_stranded_index_entry_cannot_wedge_a_key(service):
"""Defensive: the request is looked up, not trusted from the id alone."""
service._inflight_by_cache_key["ghost"] = "no_such_request"
with patch.object(service.session, "get", return_value=_resp()):
req = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="ghost",
callback=lambda r: None, max_retries=0)
_wait(service, req)
assert service.get_result(req).success is True
def test_the_deduplicated_count_is_reported(service):
session = _BlockingSession()
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: None, max_retries=0)
assert session.started.wait(timeout=5)
service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
max_retries=0)
session.release.set()
_wait(service, first)
assert service.get_statistics().get("deduplicated_requests") == 1
def test_a_cancelled_worker_cannot_overwrite_its_replacement(service, cache):
"""Cancelling frees the key, so a replacement may already own it.
The worker cannot abort an HTTP call in flight, so when the cancelled one
finally returns it must discard its response rather than write it. Without
that, the sequence is: cancel A, submit B for the same key, B fetches and
caches fresh data, A returns and overwrites it with the response nobody
wanted -- and calls A's callbacks too.
"""
slow = _BlockingSession()
stale = {"events": [{"id": "STALE"}]}
slow_resp = Mock()
slow_resp.json.return_value = stale
slow_resp.raise_for_status.return_value = None
def blocked_get(*a, **k):
slow.calls += 1
slow.started.set()
slow.release.wait(timeout=5)
return slow_resp
called = []
with patch.object(service.session, "get", side_effect=blocked_get):
first = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: called.append("cancelled_one"), max_retries=0)
assert slow.started.wait(timeout=5)
service.cancel_request(first)
assert "k" not in service._inflight_by_cache_key
# The replacement writes the fresh value while the cancelled fetch is held.
fresh = {"events": [{"id": "FRESH"}]}
fresh_resp = Mock()
fresh_resp.json.return_value = fresh
fresh_resp.raise_for_status.return_value = None
with patch.object(service.session, "get", return_value=fresh_resp):
second = service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s", cache_key="k",
callback=lambda r: called.append("replacement"), max_retries=0)
_wait(service, second)
assert cache.set.call_args[0][1] == fresh, "replacement must own the cache"
# Now let the cancelled fetch finish. It must write nothing and call nobody.
# Wait for the worker to actually finish rather than sleeping: a fixed
# sleep is a race under load, and a slow worker would make this pass for
# the wrong reason. A cancelled request is still filed in
# completed_requests, so that is the signal it has run to completion.
writes_before = cache.set.call_count
slow.release.set()
deadline = time.time() + 5
while first not in service.completed_requests and time.time() < deadline:
time.sleep(0.02)
assert first in service.completed_requests, "cancelled worker never finished"
assert cache.set.call_count == writes_before, (
"the cancelled worker wrote to the cache after its replacement")
assert cache.set.call_args[0][1] == fresh, "stale data overwrote fresh"
assert "cancelled_one" not in called, (
"a cancelled request must not deliver callbacks")
def test_request_ids_are_unique_within_a_millisecond(service):
"""request_id was sport_year_milliseconds, which collides.
Two submits inside the same millisecond produced the SAME id, so one
silently replaced the other in active_requests and completed_requests.
Dedupe hands this id back to every joiner as their handle for
get_result(), so uniqueness is now load-bearing rather than incidental.
"""
# Stub the executor rather than the session: this is about what submit
# hands back, and letting 50 workers loose would outlive the patch and
# make real network calls.
with patch.object(service.executor, "submit"):
ids = [
service.submit_fetch_request(
sport="nba", year=2026, url="https://x/s",
cache_key=f"key_{i}", # distinct keys: no dedupe
callback=lambda r: None, max_retries=0)
for i in range(50)
]
assert len(set(ids)) == len(ids), "request ids collided"
# --- cancellation must be terminal ------------------------------------------
#
# Cancelling used to be advisory: three separate paths wrote request.status
# without checking whether the request had already been cancelled, so a cancel
# could be silently undone and the work it was meant to stop went ahead.
URL = "http://example.invalid/scores"
KEY = "sched_nfl_2025"
class _CountingSession:
"""Records whether an HTTP fetch was ever attempted."""
def __init__(self):
self.calls = 0
def get(self, *a, **k):
self.calls += 1
return _resp()
class _BlockingFailingSession(_CountingSession):
"""Holds the fetch open, then fails it.
Cancelling while the worker is parked inside the HTTP call is the only way
to reach the exception handler as a cancelled request. Cancel it before
the call starts and the worker returns at the pre-start branch instead,
which would leave the except path untested.
"""
def __init__(self):
super().__init__()
self.started = threading.Event()
self.release = threading.Event()
def get(self, *a, **k):
self.calls += 1
self.started.set()
self.release.wait(timeout=5)
raise requests.RequestException("connection reset")
def _fill_every_worker_slot(service, slots=3):
"""Occupy the pool so the next submit is queued rather than started.
This is what makes "cancel before the worker runs" deterministic instead
of a race the test would win only sometimes. Returns the gate that
releases the pool.
"""
gate = threading.Event()
for _ in range(slots):
service.executor.submit(gate.wait, 5)
return gate
def test_cancelling_before_the_worker_starts_stops_the_fetch(service, cache):
"""The queued worker must honour a cancel, not overwrite it with IN_PROGRESS.
Between submit and the worker picking the job up, the request sits in the
executor queue. Cancelling there is the cheapest possible cancel -- nothing
has been downloaded yet -- and it was the one that did not work.
"""
gate = _fill_every_worker_slot(service)
session = _CountingSession()
delivered = []
with patch.object(service, 'session', session):
rid = service.submit_fetch_request(
"nfl", 2025, URL, KEY, max_retries=0, callback=delivered.append
)
assert service.cancel_request(rid) is True
gate.set() # let the queued worker run
_wait(service, rid)
assert session.calls == 0, (
"cancelled before it started, yet the worker still downloaded the payload"
)
assert cache.set.call_count == 0, "a cancelled request wrote to the cache"
assert delivered == [], "a cancelled request invoked its callbacks"
def test_a_cancel_during_the_commit_is_refused(service, cache):
"""Once the worker has claimed the commit, cancelling is too late.
The claim and the cancelled-check happen in one critical section, so a
cancel arriving after it cannot retroactively abandon data already on its
way to the cache. Letting it through stranded every joiner: the payload
landed in the cache but the callbacks were suppressed, so a manager that
joined this fetch waited for a call that never came.
"""
gate = _fill_every_worker_slot(service)
late = {}
delivered = []
def cancel_mid_write(key, data, *a, **k):
late['returned'] = service.cancel_request(late['rid'])
cache.set.side_effect = cancel_mid_write
# Read the payload inside the callback. The service releases result.data
# once every callback has been delivered, so inspecting the FetchResult
# afterwards sees the released object, not what the caller was handed.
def record(result):
delivered.append((result.success, result.data))
with patch.object(service, 'session', _CountingSession()):
late['rid'] = service.submit_fetch_request(
"nfl", 2025, URL, KEY, max_retries=0, callback=record
)
gate.set() # only now can the worker reach the commit
_wait(service, late['rid'])
assert late.get('returned') is False, (
"cancelled a request that had already committed"
)
assert cache.set.call_count == 1, "the commit itself was lost"
assert delivered == [(True, PAYLOAD)], (
"data reached the cache but the callbacks were suppressed -- "
"every joined submitter is left waiting forever"
)
def test_a_failure_after_cancelling_stays_cancelled(service, cache):
"""A cancelled request that then errors must not resurface as FAILED.
The except path overwrote CANCELLED with FAILED, and the callback gate in
the finally block only suppresses callbacks for CANCELLED -- so cancelling
a request that was about to time out delivered a spurious error callback.
"""
session = _BlockingFailingSession()
delivered = []
with patch.object(service, 'session', session):
rid = service.submit_fetch_request(
"nfl", 2025, URL, KEY, max_retries=0, callback=delivered.append
)
# Assert the worker is inside the HTTP call before cancelling,
# otherwise this silently degrades into the pre-start case and the
# exception handler is never exercised.
assert session.started.wait(timeout=5)
assert service.cancel_request(rid) is True
session.release.set()
_wait(service, rid)
assert session.calls == 1, "the fetch never started, so nothing could fail"
assert delivered == [], "a cancelled request delivered a failure callback"
assert service.get_request_status(rid) is FetchStatus.CANCELLED, (
"a cancelled request that then errored was reported as FAILED"
)
+280
View File
@@ -0,0 +1,280 @@
"""A delivered fetch payload must not stay resident on the stored result.
BackgroundDataService kept the fetched body on the FetchResult it filed in
`completed_requests`, which is swept only hourly and capped at 500 entries by
count. For status records that is free; for a season schedule it is not. NCAA
football's 2026 schedule is 946 games, and on a 1GB Pi 3B+ the parsed payload
measured ~90MB -- a tenth of the board's memory, pinned for an hour after the
consumer had already been handed it.
The cache-hit path was the worse of the two. It runs once per update interval
per sport, mints a fresh request_id each time, and hands back whatever the
cache returns -- so a memory-tier miss (the tier is capped at 150 entries)
re-parses the payload from disk into a genuinely new object. Those accumulate
as separate copies rather than shared references, which is the staircase seen
in the field: RSS stepping up ~90MB per sport as seasons loaded and never
coming back down.
Releasing is safe because the payload is written to the cache under the
request's cache_key before the result is built, and that is where consumers
read it from -- the callback is handed the object directly and the plugins use
it only in passing before reading the cache back.
Requests submitted *without* a callback keep their payload: polling
get_result() is then the only way to collect it, so releasing would break that
contract.
And the release must happen after EVERY callback, not after each one. Callers
that joined an in-flight fetch share a single FetchResult, so releasing per
delivery strips the payload out from under everyone still queued -- see
TestJoinersAllGetTheData.
"""
import threading
import time
import pytest
from unittest.mock import MagicMock, Mock, patch
from src.background_data_service import BackgroundDataService
PAYLOAD = {"events": [{"id": f"g{i}"} for i in range(50)]}
@pytest.fixture
def cache():
m = MagicMock()
m.get.return_value = None
m.set.return_value = None
return m
@pytest.fixture
def service(cache):
svc = BackgroundDataService(cache, max_workers=2, request_timeout=5)
yield svc
svc.shutdown(wait=False)
def _wait(service, req_id, timeout=5):
"""Wait for the result to be FILED.
Enough for anything that is true by the time the worker stores the result:
its success flag, its error, the cache write that happened during the
fetch.
"""
deadline = time.time() + timeout
while not service.is_request_complete(req_id) and time.time() < deadline:
time.sleep(0.02)
def _wait_for_release(service, req_id, timeout=5):
"""Wait for the payload to be RELEASED, which is strictly later.
The worker files the result, then runs the callback, then releases. So
is_request_complete() goes true while the callback still has not run --
waiting on it alone leaves a window in which `seen` is empty and the
payload is still resident, and the assertions race the worker. It passes
in practice only because a one-line callback usually beats the 20ms poll.
Release happens after the callback returns, so a released payload also
means the callback has finished: one wait covers both.
"""
deadline = time.time() + timeout
while time.time() < deadline:
result = service.get_result(req_id)
if result is not None and result.data is None:
return
time.sleep(0.02)
raise AssertionError(
f"payload for {req_id} was never released (callback may not have run)")
def _resp():
r = Mock()
r.json.return_value = PAYLOAD
r.raise_for_status.return_value = None
return r
class TestFetchPath:
def test_callback_receives_the_payload_then_it_is_released(self, service, cache):
seen = {}
def callback(result):
# The consumer's one look at the data happens here.
seen['events'] = len(result.data['events'])
with patch.object(service.session, "get", return_value=_resp()):
req_id = service.submit_fetch_request(
sport="ncaa_fb", year=2026, url="https://example.com/s",
cache_key="ncaa_fb_2026", callback=callback, max_retries=0,
)
_wait_for_release(service, req_id)
assert seen['events'] == 50, "callback must still be handed the payload"
stored = service.get_result(req_id)
assert stored is not None
assert stored.success is True
assert stored.data is None, "payload must not stay on the stored result"
def test_nothing_is_lost_the_cache_holds_it(self, service, cache):
with patch.object(service.session, "get", return_value=_resp()):
req_id = service.submit_fetch_request(
sport="ncaa_fb", year=2026, url="https://example.com/s",
cache_key="ncaa_fb_2026", callback=lambda r: None, max_retries=0,
)
_wait(service, req_id)
cache.set.assert_called_once()
key, written = cache.set.call_args[0][:2]
assert key == "ncaa_fb_2026"
assert written == PAYLOAD, "the payload must be persisted before release"
def test_without_a_callback_the_payload_is_kept(self, service, cache):
# Polling get_result() is then the only delivery mechanism.
with patch.object(service.session, "get", return_value=_resp()):
req_id = service.submit_fetch_request(
sport="nfl", year=2026, url="https://example.com/s",
cache_key="nfl_2026", max_retries=0,
)
_wait(service, req_id)
assert service.get_result(req_id).data == PAYLOAD
def test_a_failed_fetch_still_records_its_error(self, service, cache):
with patch.object(service.session, "get", side_effect=Exception("boom")):
req_id = service.submit_fetch_request(
sport="nfl", year=2026, url="https://example.com/s",
cache_key="nfl_2026", callback=lambda r: None, max_retries=0,
)
_wait(service, req_id)
stored = service.get_result(req_id)
assert stored.success is False
assert stored.error is not None
class TestCacheHitPath:
def test_cache_hit_releases_after_the_callback(self, service, cache):
cache.get.return_value = PAYLOAD
seen = {}
req_id = service.submit_fetch_request(
sport="ncaa_fb", year=2026, url="https://example.com/s",
cache_key="ncaa_fb_2026",
callback=lambda r: seen.update(events=len(r.data['events'])),
)
assert seen['events'] == 50
assert service.get_result(req_id).data is None
def test_repeated_cache_hits_do_not_accumulate_payloads(self, service, cache):
# The staircase: one entry per update interval per sport, each one
# potentially a freshly parsed copy after a memory-tier miss.
cache.get.return_value = PAYLOAD
for _ in range(25):
service.submit_fetch_request(
sport="ncaa_fb", year=2026, url="https://example.com/s",
cache_key="ncaa_fb_2026", callback=lambda r: None,
)
retained = [r for r in service.completed_requests.values() if r.data is not None]
assert retained == [], f"{len(retained)} payloads still resident"
def test_cache_hit_without_a_callback_is_unchanged(self, service, cache):
cache.get.return_value = PAYLOAD
req_id = service.submit_fetch_request(
sport="nfl", year=2026, url="https://example.com/s",
cache_key="nfl_2026",
)
assert service.get_result(req_id).data == PAYLOAD
class TestJoinersAllGetTheData:
"""Deduplicated callers share one FetchResult; releasing between them
empties it for the rest.
Not a corner case. A sport's recent, upcoming and live managers all ask for
the same season schedule, so the second and third are joiners on almost
every cycle. Releasing inside the delivery loop handed the payload to
whichever ran first and gave the others `result.data is None`.
The consequence was worse than a quiet degradation, because consumers do
`result.data.get('events')`: they raised AttributeError, the delivery loop
caught it, and the whole failure surfaced as a single
"Error in callback for request ..." line while that manager silently never
received its schedule.
"""
class _BlockingSession:
"""Holds the fetch open so a second submit lands while in flight."""
def __init__(self):
self.release = threading.Event()
self.started = threading.Event()
def get(self, *a, **k):
self.started.set()
self.release.wait(timeout=5)
return _resp()
def test_every_joiner_is_handed_the_payload(self, service):
session = self._BlockingSession()
seen = {}
def record(name):
# Read it the way the sport managers do. `result.data['events']`
# would raise TypeError on None; `.get` raises AttributeError,
# which is the error actually seen in the field.
def cb(result):
seen[name] = result.data.get('events') if result.data else None
return cb
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nhl", year=2026, url="https://x/s", cache_key="nhl_2026",
callback=record("first"), max_retries=0)
assert session.started.wait(timeout=5)
joined = service.submit_fetch_request(
sport="nhl", year=2026, url="https://x/s", cache_key="nhl_2026",
callback=record("second"), max_retries=0)
# Without coalescing these are two independent fetches that each
# own their result, and the test proves nothing about sharing.
assert joined == first, "the joiner should share the in-flight id"
session.release.set()
_wait(service, first)
deadline = time.time() + 5
while len(seen) < 2 and time.time() < deadline:
time.sleep(0.02)
assert set(seen) == {"first", "second"}, f"both must be called, got {seen}"
for name, events in seen.items():
assert events is not None, (
f"{name!r} was handed a released payload: the result was "
f"emptied before every callback had been delivered")
assert len(events) == 50, f"{name!r} got {events!r}"
def test_the_payload_is_still_released_once_they_have_all_had_it(self, service):
"""The memory fix must survive the ordering fix."""
session = self._BlockingSession()
with patch.object(service, "session", session):
first = service.submit_fetch_request(
sport="nhl", year=2026, url="https://x/s", cache_key="nhl_2026",
callback=lambda r: None, max_retries=0)
assert session.started.wait(timeout=5)
service.submit_fetch_request(
sport="nhl", year=2026, url="https://x/s", cache_key="nhl_2026",
callback=lambda r: None, max_retries=0)
session.release.set()
_wait_for_release(service, first)
stored = service.get_result(first)
assert stored is not None and stored.data is None, (
"the payload must still be dropped once every callback has run")
+122
View File
@@ -0,0 +1,122 @@
"""The gate must read the core version from disk, not from its own import.
`from src import __version__` binds whatever the process loaded at start. The
web UI is a long-lived service of its own (ledmatrix-web.service), and updating
the core replaces files on disk without restarting it -- the update route says
so and asks the user to restart, but its prompt named only the *display*
service, so a user who followed it left the web process holding the old number.
The plugin store's gate lives in that web process, so being stale by exactly
one release is the case that bites: every plugin flooring on the release you
just installed is refused, with a message blaming a core version that is
already correct on disk.
Observed on hardware: after updating a rig to 3.3.0 and restarting only the
display service, all eight sports scoreboards were refused with
"supports LEDMatrix >=3.3.0, but this system is running 3.2.0" while
src/__init__.py on that machine read 3.3.0.
"""
import importlib
import sys
import pytest
from src.plugin_system import compatibility
class TestCurrentCoreVersion:
def test_it_reads_the_file_rather_than_the_imported_value(self, monkeypatch, tmp_path):
# Simulate a process whose import predates the update: the module
# object says 3.2.0 while the file on disk says 3.3.0.
import src
monkeypatch.setattr(src, "__version__", "3.2.0")
fake = tmp_path / "__init__.py"
fake.write_text('__version__ = "3.3.0"\n', encoding="utf-8")
monkeypatch.setattr(compatibility, "_VERSION_FILE", fake)
assert compatibility.current_core_version() == "3.3.0"
def test_it_matches_the_real_file_by_default(self):
import src
assert compatibility.current_core_version() == src.__version__
@pytest.mark.parametrize("body", [
"__version__ = '3.4.1'\n",
'__version__="3.4.1"\n',
'"""doc"""\n\n__version__ = "3.4.1" # trailing comment\n',
])
def test_it_tolerates_the_ways_that_line_gets_written(self, monkeypatch, tmp_path, body):
fake = tmp_path / "__init__.py"
fake.write_text(body, encoding="utf-8")
monkeypatch.setattr(compatibility, "_VERSION_FILE", fake)
assert compatibility.current_core_version() == "3.4.1"
def test_a_missing_file_falls_back_to_the_import(self, monkeypatch, tmp_path):
# Never worse than before: an unreadable file returns what the old
# code would have returned.
import src
monkeypatch.setattr(src, "__version__", "3.2.0")
monkeypatch.setattr(compatibility, "_VERSION_FILE", tmp_path / "gone.py")
assert compatibility.current_core_version() == "3.2.0"
def test_a_file_without_the_line_falls_back(self, monkeypatch, tmp_path):
import src
monkeypatch.setattr(src, "__version__", "3.2.0")
fake = tmp_path / "__init__.py"
fake.write_text("# no version here\n", encoding="utf-8")
monkeypatch.setattr(compatibility, "_VERSION_FILE", fake)
assert compatibility.current_core_version() == "3.2.0"
def test_it_never_raises(self, monkeypatch, tmp_path):
# This runs on the install path; an exception here would surface as a
# failed update rather than a version mismatch.
bad = tmp_path / "__init__.py"
bad.write_bytes(b"\xff\xfe\x00 not utf-8 \xff")
monkeypatch.setattr(compatibility, "_VERSION_FILE", bad)
assert isinstance(compatibility.current_core_version(), str)
class TestTheBugItFixes:
def test_a_stale_import_no_longer_refuses_a_compatible_plugin(self, monkeypatch, tmp_path):
"""The exact hardware failure, as a test."""
import src
monkeypatch.setattr(src, "__version__", "3.2.0") # what the process holds
fake = tmp_path / "__init__.py"
fake.write_text('__version__ = "3.3.0"\n', encoding="utf-8") # what is on disk
monkeypatch.setattr(compatibility, "_VERSION_FILE", fake)
manifest = {"name": "Hockey Scoreboard", "min_ledmatrix_version": "3.3.0"}
stale_ok, _ = compatibility.check(manifest, src.__version__)
assert stale_ok is False, "precondition: the stale value is what refused it"
fresh_ok, reason = compatibility.check(
manifest, compatibility.current_core_version())
assert fresh_ok is True, f"the disk version must allow it, got: {reason}"
def test_it_still_refuses_when_the_core_really_is_too_old(self, monkeypatch, tmp_path):
# The gate must not become permissive: a genuinely old core still says no.
fake = tmp_path / "__init__.py"
fake.write_text('__version__ = "3.2.0"\n', encoding="utf-8")
monkeypatch.setattr(compatibility, "_VERSION_FILE", fake)
ok, reason = compatibility.check(
{"name": "Hockey", "min_ledmatrix_version": "3.3.0"},
compatibility.current_core_version())
assert ok is False
assert "3.2.0" in (reason or "")
class TestCallSites:
@pytest.mark.parametrize("module", [
"src.plugin_system.store_manager",
"src.plugin_system.plugin_loader",
])
def test_no_gate_binds_the_version_at_import(self, module):
"""Catch a future call site reintroducing the stale read."""
import inspect
mod = importlib.import_module(module)
source = inspect.getsource(mod)
assert "from src import __version__ as core_version" not in source, (
f"{module} binds __version__ at import; use "
f"compatibility.current_core_version() so a long-lived process "
f"sees a core update.")
@@ -6,6 +6,7 @@ These tests cover the reconcile path that loads/unloads plugins and rebuilds
the dispatch maps on the main thread when the enabled set changes.
"""
import copy
from unittest.mock import MagicMock
@@ -253,3 +254,182 @@ class TestEnabledSetChanged:
{"a": {"enabled": True, "duration": 30}},
{"a": {"enabled": True, "duration": 45}},
) is False
class TestEnabledPluginNotRunning:
"""A plugin that fails validate_config() is enabled but absent, and the
config edit that fixes it is nested inside the plugin's own section -- so
the top-level ``enabled`` comparison never sees it. These cover the second
gate that queues a reconcile in that case.
"""
def test_nested_edit_is_invisible_to_the_enabled_set_check(self, test_display_controller):
"""The original gate: proves why a second one is needed."""
controller = test_display_controller
old = {"hockey-scoreboard": {"enabled": True, "nhl": {"enabled": False}}}
new = {"hockey-scoreboard": {"enabled": True, "nhl": {"enabled": True}}}
# Enabling a league changes no top-level flag.
assert controller._enabled_set_changed(old, new) is False
def test_queues_reconcile_when_enabled_plugin_is_absent(self, test_display_controller):
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {} # failed to load
cfg = {"hockey-scoreboard": {"enabled": True, "nhl": {"enabled": True}}}
assert controller._enabled_plugin_not_running(cfg) is True
def test_quiet_when_every_enabled_plugin_is_running(self, test_display_controller):
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {"hockey-scoreboard": ["nhl"]}
cfg = {"hockey-scoreboard": {"enabled": True}}
assert controller._enabled_plugin_not_running(cfg) is False
def test_disabled_plugin_does_not_queue(self, test_display_controller):
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {}
cfg = {"hockey-scoreboard": {"enabled": False}}
assert controller._enabled_plugin_not_running(cfg) is False
def test_non_plugin_sections_do_not_queue(self, test_display_controller):
"""``schedule``/``display`` carry their own ``enabled`` and are never
in plugin_display_modes -- without the manifest check they would queue
a reconcile, and therefore a filesystem scan, on every config save."""
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {"hockey-scoreboard": ["nhl"]}
cfg = {
"hockey-scoreboard": {"enabled": True},
"schedule": {"enabled": True},
"display": {"enabled": True},
}
assert controller._enabled_plugin_not_running(cfg) is False
def test_non_dict_section_is_ignored(self, test_display_controller):
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {}
assert controller._enabled_plugin_not_running({"hockey-scoreboard": "nonsense"}) is False
def test_no_plugin_manager_is_quiet(self, test_display_controller):
controller = test_display_controller
controller.plugin_manager = None
assert controller._enabled_plugin_not_running({"x": {"enabled": True}}) is False
class TestReconcileQueuedThroughSubscriber:
"""End-to-end through the real config-change subscriber, not the helper.
Without the second gate this is the four-day-outage path: the plugin is
enabled, absent, and the save that enables its league sets no flag.
"""
@staticmethod
def _subscriber(controller):
subs = controller.config_service._subscribers['*']
for cb in subs:
if getattr(cb, '__name__', '') == '_controller_config_change':
return cb
raise AssertionError(f"controller subscriber not found among {subs}")
@staticmethod
def _configs(controller, plugin_section_old, plugin_section_new):
"""Build two full configs differing only inside the plugin section --
the subscriber refreshes its cache from these, so they must be real."""
base = copy.deepcopy(controller.config)
old = copy.deepcopy(base)
new = copy.deepcopy(base)
old["hockey-scoreboard"] = plugin_section_old
new["hockey-scoreboard"] = plugin_section_new
return old, new
def test_nested_edit_queues_reconcile_for_absent_plugin(self, test_display_controller):
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {} # validate_config() said False
controller._pending_plugin_reconcile = False
old, new = self._configs(
controller,
{"enabled": True, "nhl": {"enabled": False}},
{"enabled": True, "nhl": {"enabled": True}},
)
# The original gate is blind to this edit ...
assert controller._enabled_set_changed(old, new) is False
self._subscriber(controller)(old, new)
# ... but the reconcile is queued anyway.
assert controller._pending_plugin_reconcile is True
def test_steady_state_does_not_queue_reconcile(self, test_display_controller):
"""Everything enabled is running: an unrelated edit must not queue a
reconcile, or every config save drags a filesystem scan onto the
render thread."""
controller = test_display_controller
controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
controller.plugin_display_modes = {"hockey-scoreboard": ["nhl"]}
controller._pending_plugin_reconcile = False
old, new = self._configs(
controller,
{"enabled": True, "scroll_speed": 1},
{"enabled": True, "scroll_speed": 2},
)
self._subscriber(controller)(old, new)
assert controller._pending_plugin_reconcile is False
class TestPendingReconcileNotLost:
"""A config change arriving *during* reconcile must not be discarded.
The flag used to be cleared after a successful reconcile. Reconcile has
already read its config by then, so that clear erased a request it never
served and the newest config never reconciled -- the same "my save did
nothing" symptom this path exists to prevent.
"""
def test_request_arriving_during_reconcile_survives(self, test_display_controller):
controller = test_display_controller
controller._pending_plugin_reconcile = True
def reconcile_and_race():
# The watcher thread queues another change while we are mid-flight.
with controller._reconcile_flag_lock:
controller._pending_plugin_reconcile = True
return True
controller._reconcile_enabled_plugins = reconcile_and_race
controller._service_pending_reconcile()
assert controller._pending_plugin_reconcile is True, \
"a config change landing during reconcile was discarded"
def test_flag_cleared_on_a_quiet_success(self, test_display_controller):
controller = test_display_controller
controller._pending_plugin_reconcile = True
controller._reconcile_enabled_plugins = lambda: True
controller._service_pending_reconcile()
assert controller._pending_plugin_reconcile is False
def test_retryable_failure_rearms(self, test_display_controller):
controller = test_display_controller
controller._pending_plugin_reconcile = True
controller._reconcile_enabled_plugins = lambda: False
controller._service_pending_reconcile()
assert controller._pending_plugin_reconcile is True
def test_no_reconcile_when_nothing_pending(self, test_display_controller):
controller = test_display_controller
controller._pending_plugin_reconcile = False
calls = []
controller._reconcile_enabled_plugins = lambda: calls.append(1) or True
controller._service_pending_reconcile()
assert calls == []
+34 -2
View File
@@ -13,6 +13,7 @@ need root and mutate the system, so they are exercised manually instead.
"""
import subprocess
import tempfile
from pathlib import Path
import pytest
@@ -31,6 +32,16 @@ def run_lib(snippet: str, env: dict | None = None) -> subprocess.CompletedProces
)
def _fstype_of(path: object) -> str:
"""Filesystem type backing ``path``, via the same tool the helper uses."""
result = subprocess.run(
["findmnt", "-no", "FSTYPE", "--target", str(path)],
capture_output=True, text=True,
env={"PATH": "/usr/bin:/bin:/usr/sbin:/sbin"},
)
return result.stdout.strip()
def call(fn: str, *args: object, env: dict | None = None) -> str:
joined = " ".join(str(a) for a in args)
result = run_lib(f"{fn} {joined}", env=env)
@@ -195,8 +206,29 @@ class TestOomDetection:
class TestDiskBackedTmpdir:
def test_returns_nothing_when_tmpdir_is_already_disk_backed(self, tmp_path):
# tmp_path is on the regular filesystem, so the default must be kept.
assert call("lm_disk_backed_tmpdir", env={"TMPDIR": str(tmp_path)}) == ""
# Do not assume tmp_path is disk-backed. Debian 13 -- the platform this
# helper exists for -- mounts /tmp as tmpfs, and pytest puts tmp_path
# under /tmp, so this asserted against a *memory*-backed directory and
# failed on the target platform while the helper behaved exactly as
# designed. Search for a directory whose backing store is really disk.
scratch = None
disk_backed = None
for candidate in (tmp_path, Path("/var/tmp"), LIB.parent):
if _fstype_of(candidate) not in ("tmpfs", "ramfs", ""):
if candidate is tmp_path:
disk_backed = candidate
else:
scratch = Path(tempfile.mkdtemp(dir=str(candidate)))
disk_backed = scratch
break
if disk_backed is None:
pytest.skip("no disk-backed directory available to test against")
try:
assert call("lm_disk_backed_tmpdir",
env={"TMPDIR": str(disk_backed)}) == ""
finally:
if scratch is not None:
scratch.rmdir()
def test_redirects_away_from_a_memory_backed_tmpdir(self):
# Debian 13 mounts /tmp as tmpfs, which would otherwise hold the whole
+8 -2
View File
@@ -89,11 +89,17 @@ class TestContextualFormatter:
assert "hello" in out
def test_location_toggle(self):
# Assert on the whole "module.func:lineno" token, not a bare ":42".
# The formatted line starts with an HH:MM:SS timestamp, so a bare
# ":{lineno}" also matches the clock whenever the minute or second
# happens to equal the line number -- about 3% of runs, which is a
# flaky failure with nothing wrong.
record = make_record()
location = f"{record.module}.{record.funcName}:{record.lineno}"
with_loc = ContextualFormatter(include_location=True).format(record)
without = ContextualFormatter(include_location=False).format(record)
assert f":{record.lineno}" in with_loc
assert f":{record.lineno}" not in without
assert location in with_loc
assert location not in without
def test_record_not_mutated_no_double_prefix(self):
# Regression: a record is formatted once PER HANDLER. The formatter
+225 -1
View File
@@ -8,11 +8,27 @@ ensure_logo_directory, and the download_missing_logo function path
"""
import os
import time
import pytest
from pathlib import Path
from unittest.mock import patch, Mock, MagicMock
from src.logo_downloader import LogoDownloader
from PIL import Image
from PIL.PngImagePlugin import PngInfo
from src.logo_downloader import (
PLACEHOLDER_BG,
PLACEHOLDER_MARKER,
PLACEHOLDER_RETRY_SECONDS,
PLACEHOLDER_SIZE,
LogoDownloader,
download_missing_logo,
is_placeholder_logo,
placeholder_age_seconds,
refresh_placeholder_timestamp,
should_attempt_download,
)
# ---------------------------------------------------------------------------
@@ -127,3 +143,211 @@ class TestEnsureLogoDirectory:
with patch("builtins.open", side_effect=mock_open):
result = downloader.ensure_logo_directory(test_dir)
assert result is False
# ---------------------------------------------------------------------------
# Placeholder detection and retry
#
# A failed download used to be cached as a placeholder wearing the real logo's
# filename, and download_missing_logo returned early on "the file exists". One
# transient failure therefore pinned a team to a grey box permanently.
# ---------------------------------------------------------------------------
class TestPlaceholderLogos:
def _placeholder(self, tmp_path, abbrev="COLL"):
downloader = LogoDownloader()
assert downloader.create_placeholder_logo(abbrev, str(tmp_path)) is True
return tmp_path / f"{abbrev}.png"
def test_generated_placeholder_is_recognised(self, tmp_path):
assert is_placeholder_logo(self._placeholder(tmp_path)) is True
def test_real_logo_is_not_a_placeholder(self, tmp_path):
real = tmp_path / "REAL.png"
Image.new("RGBA", (500, 500), (12, 34, 56, 255)).save(real)
assert is_placeholder_logo(real) is False
def test_legacy_unmarked_placeholder_is_recognised(self, tmp_path):
"""Placeholders written before the marker existed must still be caught.
They are already sitting on users' disks; if they were not recognised
those teams would stay grey boxes forever even after this fix.
"""
legacy = tmp_path / "LEGACY.png"
Image.new("RGBA", PLACEHOLDER_SIZE, PLACEHOLDER_BG).save(legacy)
assert is_placeholder_logo(legacy) is True
def test_same_size_but_different_colour_is_not_a_placeholder(self, tmp_path):
real = tmp_path / "SMALL.png"
Image.new("RGBA", PLACEHOLDER_SIZE, (10, 200, 10, 255)).save(real)
assert is_placeholder_logo(real) is False
def test_missing_file_is_not_a_placeholder(self, tmp_path):
assert is_placeholder_logo(tmp_path / "nope.png") is False
def test_existing_real_logo_short_circuits_without_downloading(self, tmp_path):
real = tmp_path / "REAL.png"
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
with patch.object(LogoDownloader, "download_logo") as download:
assert download_missing_logo(
"afl", "1", "REAL", real, logo_url="http://example/x.png") is True
download.assert_not_called()
def _age_placeholder(self, path, seconds):
"""Rewrite a placeholder's marker so it reads as `seconds` old."""
metadata = PngInfo()
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - seconds))
with Image.open(path) as img:
img.copy().save(path, "PNG", pnginfo=metadata)
def test_stale_placeholder_triggers_a_retry(self, tmp_path):
path = self._placeholder(tmp_path)
self._age_placeholder(path, PLACEHOLDER_RETRY_SECONDS + 60)
assert placeholder_age_seconds(path) > PLACEHOLDER_RETRY_SECONDS
with patch.object(LogoDownloader, "download_logo", return_value=True) as download:
assert download_missing_logo(
"afl", "1", "COLL", path,
logo_url="http://example/coll.png") is True
download.assert_called_once()
def test_placeholder_age_survives_an_mtime_touch(self, tmp_path):
"""The age comes from the stamp, not the filesystem.
Anything that rewrites file times -- a backup restore, an rsync, a
permissions fix script -- would otherwise reset the retry clock.
"""
path = self._placeholder(tmp_path)
self._age_placeholder(path, PLACEHOLDER_RETRY_SECONDS + 60)
now = time.time()
os.utime(path, (now, now))
assert placeholder_age_seconds(path) > PLACEHOLDER_RETRY_SECONDS
def test_fresh_placeholder_does_not_retry(self, tmp_path):
"""Rate limiting: a placeholder written seconds ago must not re-download.
Without this the fix would trade a permanent grey box for an ESPN
request on every frame.
"""
path = self._placeholder(tmp_path)
with patch.object(LogoDownloader, "download_logo") as download:
assert download_missing_logo(
"afl", "1", "COLL", path,
logo_url="http://example/coll.png") is True
download.assert_not_called()
class TestDownloadEligibility:
"""One rule, shared by every download site.
The two bulk loops and the single-logo path each had their own idea of what
counted as "already have it", which is how one of them ended up retrying
fresh placeholders and the other skipping stale ones forever.
"""
def _placeholder(self, tmp_path, abbrev="COLL"):
assert LogoDownloader().create_placeholder_logo(abbrev, str(tmp_path))
return tmp_path / f"{abbrev}.png"
def _age(self, path, seconds):
metadata = PngInfo()
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - seconds))
with Image.open(path) as img:
img.copy().save(path, "PNG", pnginfo=metadata)
def test_missing_file_is_eligible(self, tmp_path):
assert should_attempt_download(tmp_path / "nope.png") is True
def test_real_logo_is_not_eligible(self, tmp_path):
real = tmp_path / "REAL.png"
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
assert should_attempt_download(real) is False
def test_force_download_beats_a_real_logo(self, tmp_path):
real = tmp_path / "REAL.png"
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
assert should_attempt_download(real, force_download=True) is True
def test_fresh_placeholder_is_not_eligible(self, tmp_path):
assert should_attempt_download(self._placeholder(tmp_path)) is False
def test_stale_placeholder_is_eligible(self, tmp_path):
path = self._placeholder(tmp_path)
self._age(path, PLACEHOLDER_RETRY_SECONDS + 60)
assert should_attempt_download(path) is True
def test_league_bulk_loop_skips_a_fresh_placeholder(self, tmp_path):
"""A bulk pass honours the same back-off as everything else."""
self._placeholder(tmp_path, "AAA")
downloader = LogoDownloader()
teams = [{"abbreviation": "AAA", "display_name": "A", "logo_url": "http://x/a.png"}]
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
with patch.object(LogoDownloader, "download_logo") as download:
downloader.download_missing_logos_for_league("nfl")
download.assert_not_called()
def test_league_bulk_loop_retries_a_stale_placeholder(self, tmp_path):
path = self._placeholder(tmp_path, "AAA")
self._age(path, PLACEHOLDER_RETRY_SECONDS + 60)
downloader = LogoDownloader()
teams = [{"abbreviation": "AAA", "display_name": "A", "logo_url": "http://x/a.png"}]
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
with patch.object(LogoDownloader, "download_logo", return_value=True) as download:
downloader.download_missing_logos_for_league("nfl")
download.assert_called_once()
def test_ncaa_bulk_loop_retries_a_stale_placeholder(self, tmp_path):
"""This loop skipped placeholders forever; it now shares the rule."""
path = self._placeholder(tmp_path, "AAA")
self._age(path, PLACEHOLDER_RETRY_SECONDS + 60)
downloader = LogoDownloader()
teams = [{"abbreviation": "AAA", "display_name": "A",
"logo_url": "http://x/a.png", "category": "FBS",
"conference": "SEC"}]
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
with patch.object(LogoDownloader, "download_logo", return_value=True) as download:
downloader.download_all_ncaa_football_logos()
download.assert_called_once()
def test_ncaa_bulk_loop_skips_a_fresh_placeholder(self, tmp_path):
self._placeholder(tmp_path, "AAA")
downloader = LogoDownloader()
teams = [{"abbreviation": "AAA", "display_name": "A",
"logo_url": "http://x/a.png", "category": "FBS",
"conference": "SEC"}]
with patch.object(LogoDownloader, "get_logo_directory", return_value=str(tmp_path)):
with patch.object(LogoDownloader, "fetch_teams_data", return_value={"sports": [{}]}):
with patch.object(LogoDownloader, "extract_teams_from_data", return_value=teams):
with patch.object(LogoDownloader, "download_logo") as download:
downloader.download_all_ncaa_football_logos()
download.assert_not_called()
class TestRefreshPlaceholderTimestamp:
def test_restarts_the_back_off(self, tmp_path):
assert LogoDownloader().create_placeholder_logo("COLL", str(tmp_path))
path = tmp_path / "COLL.png"
metadata = PngInfo()
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - (PLACEHOLDER_RETRY_SECONDS + 60)))
with Image.open(path) as img:
img.copy().save(path, "PNG", pnginfo=metadata)
assert should_attempt_download(path) is True
assert refresh_placeholder_timestamp(path) is True
assert should_attempt_download(path) is False
def test_refuses_to_touch_a_real_logo(self, tmp_path):
real = tmp_path / "REAL.png"
Image.new("RGBA", (500, 500), (1, 2, 3, 255)).save(real)
before = real.read_bytes()
assert refresh_placeholder_timestamp(real) is False
assert real.read_bytes() == before
def test_missing_file_is_not_an_error(self, tmp_path):
assert refresh_placeholder_timestamp(tmp_path / "nope.png") is False
+73
View File
@@ -421,3 +421,76 @@ class TestSessionConfiguration:
def test_user_agent_and_accept_headers(self, helper):
assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
assert helper.session.headers["Accept"] == "image/*"
class TestStalePlaceholderHandling:
"""load_logo_with_download must not be fooled by a cached placeholder.
A placeholder wears the real logo's filename, so both the file cache and
the in-memory cache can hold one and look like a hit.
"""
def _placeholder(self, tmp_path, abbrev="COLL"):
from src.logo_downloader import LogoDownloader
assert LogoDownloader().create_placeholder_logo(abbrev, str(tmp_path))
return tmp_path / f"{abbrev}.png"
def _make_stale(self, path):
import time
from PIL.PngImagePlugin import PngInfo
from src.logo_downloader import PLACEHOLDER_MARKER, PLACEHOLDER_RETRY_SECONDS
metadata = PngInfo()
metadata.add_text(PLACEHOLDER_MARKER, str(time.time() - (PLACEHOLDER_RETRY_SECONDS + 60)))
with Image.open(path) as img:
img.copy().save(path, "PNG", pnginfo=metadata)
def test_fresh_placeholder_is_served_without_a_download(self, helper, tmp_path):
path = self._placeholder(tmp_path)
with patch.object(LogoHelper, "_download_logo") as download:
assert helper.load_logo_with_download("COLL", path, "http://x/c.png") is not None
download.assert_not_called()
def test_stale_placeholder_triggers_a_download(self, helper, tmp_path):
path = self._placeholder(tmp_path)
self._make_stale(path)
with patch.object(LogoHelper, "_download_logo") as download:
helper.load_logo_with_download("COLL", path, "http://x/c.png")
download.assert_called_once()
def test_replacement_logo_is_not_masked_by_the_cached_placeholder(self, helper, tmp_path):
"""The bug this guards: load_logo answers from cache before the disk.
Without invalidation the freshly downloaded logo would not appear until
the process restarted.
"""
path = self._placeholder(tmp_path)
first = helper.load_logo_with_download("COLL", path, "http://x/c.png")
assert first is not None
self._make_stale(path)
def fake_download(_self, _url, file_path):
Image.new("RGB", (500, 500), (7, 8, 9)).save(file_path, format="PNG")
with patch.object(LogoHelper, "_download_logo", fake_download):
second = helper.load_logo_with_download("COLL", path, "http://x/c.png")
assert second is not None
from src.logo_downloader import is_placeholder_logo
assert is_placeholder_logo(path) is False
assert second.getpixel((0, 0))[:3] == (7, 8, 9)
def test_failed_retry_restarts_the_back_off(self, helper, tmp_path):
"""Otherwise a stale placeholder means a download attempt per call."""
from src.logo_downloader import should_attempt_download
path = self._placeholder(tmp_path)
self._make_stale(path)
assert should_attempt_download(path) is True
def boom(_self, _url, _file_path):
raise OSError("network down")
with patch.object(LogoHelper, "_download_logo", boom):
helper.load_logo_with_download("COLL", path, "http://x/c.png")
assert should_attempt_download(path) is False
+298 -3
View File
@@ -19,6 +19,7 @@ The rules being pinned here, in priority order:
"""
import json
import subprocess
from pathlib import Path
from unittest.mock import MagicMock
@@ -146,9 +147,18 @@ def store(tmp_path, monkeypatch):
class TestInstallGate:
"""`install_plugin` is the chokepoint: `_reinstall_with_rollback` calls it,
so gating there covers updates too, and a refused update restores the
version the user already had."""
"""`install_plugin` is the chokepoint for every route that re-downloads:
`_reinstall_with_rollback` calls it, so a refused update restores the
version the user already had.
It is not the *only* route in, and saying so here once is cheaper than
rediscovering it. `update_plugin` also has a git branch that pulls in
place and never re-downloads; that one is gated separately by
`_gate_pulled_commit` and pinned in `TestGitPullGate` below. A third
route, `install_from_url` (sideloading from a URL), is gated in that
function and pinned in `TestSideloadGate`. All three refuse on the same
rule.
"""
def _install_with_manifest(self, store, manifest, core_version, monkeypatch):
mgr, plugins_dir = store
@@ -208,6 +218,291 @@ class TestInstallGate:
assert (path / "manifest.json").exists()
class TestSideloadGate:
"""`install_from_url` never looked at the core version.
Sideloading is an explicit act rather than an automatic store update, so
the argument for gating it is different: not "the user did not choose
this", but that the floor states the plugin *cannot run here*. Letting it
through produces the same silent PluginState.ERROR at load that the store
gate exists to prevent, and the user who typed the URL is no better placed
to diagnose it than one who pressed Update.
"""
def _sideload(self, store, manifest, core_version, monkeypatch):
mgr, plugins_dir = store
plugin_id = manifest["id"]
def fake_clone(repo_url, target_path, branches=None):
target_path.mkdir(parents=True, exist_ok=True)
(target_path / "manifest.json").write_text(
json.dumps(manifest), encoding="utf-8")
(target_path / "manager.py").write_text(
"class P: pass\n", encoding="utf-8")
return "main"
monkeypatch.setattr(mgr, "_install_via_git", fake_clone)
monkeypatch.setattr(mgr, "_install_dependencies", lambda *a, **k: True)
import src
monkeypatch.setattr(src, "__version__", core_version)
return mgr.install_from_url("https://example.invalid/plugin"), \
plugins_dir / plugin_id
def test_refuses_a_plugin_that_needs_a_newer_core(self, store, monkeypatch):
manifest = {
"id": "sideload-newer", "name": "Sideload", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "9.9.9",
}
result, path = self._sideload(store, manifest, "3.2.0", monkeypatch)
assert result["success"] is False
assert "9.9.9" in result["error"], result["error"]
assert not path.exists(), (
"a refused sideload must not leave the plugin installed")
def test_allows_a_compatible_plugin(self, store, monkeypatch):
"""The guard against over-refusing: a gate that blocks everything
passes the test above and breaks sideloading entirely."""
manifest = {
"id": "sideload-fine", "name": "Sideload", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "3.0.0",
}
result, path = self._sideload(store, manifest, "3.2.0", monkeypatch)
assert result["success"] is True, result.get("error")
assert (path / "manifest.json").exists()
def test_untrustworthy_core_does_not_block_a_2_0_0_floor(
self, store, monkeypatch):
"""Same rule as the other two routes: a v3.1.0 release reports 1.0.0,
and nearly every published manifest floors at 2.0.0."""
manifest = {
"id": "sideload-floored", "name": "Sideload", "class_name": "P",
"display_modes": ["a"], "versions": [{"ledmatrix_min": "2.0.0"}],
}
result, _ = self._sideload(store, manifest, "1.0.0", monkeypatch)
assert result["success"] is True, result.get("error")
# --------------------------------------------------------------------------
# The git-pull update path — the one route that does not re-download
# --------------------------------------------------------------------------
def _git_available() -> bool:
# OSError, not just a non-zero exit: with no git on PATH subprocess raises
# FileNotFoundError, and this runs at import time -- before skipif can act,
# so the whole module would error out instead of skipping.
try:
return subprocess.run(
['git', '--version'], capture_output=True).returncode == 0
except OSError:
return False
_HAS_GIT = _git_available()
def _git(*args, cwd):
return subprocess.run(['git', *args], cwd=str(cwd),
capture_output=True, text=True, check=True)
@pytest.mark.skipif(not _HAS_GIT, reason='git not available')
class TestGitPullGate:
"""A plugin installed as a git checkout updates by pulling in place, so it
never passes through `install_plugin` and was never gated.
Real git repositories rather than mocks, because the claim under test is
that `git reset --hard` puts the checkout back — a mock of git would only
prove the call was made, which is the easy half.
Only the handful of registry entries with no `plugin_path` reach this path
in practice; monorepo plugins install as archives and update through
`_reinstall_with_rollback`. It is gated anyway because the sunset rule in
`docs/plugin-development/08-shared-sports-code.md` states the core enforces
the floor "at install/update time", and a precondition that is documented
but not true is worse than one that is merely missing.
"""
@pytest.fixture
def checkout(self, tmp_path, monkeypatch):
"""An origin repo holding a plugin, and a clone of it installed as
`plugin-repos/gitplug`, with the store pointed at it."""
from src.plugin_system.store_manager import PluginStoreManager
origin = tmp_path / 'origin'
origin.mkdir()
_git('init', '--initial-branch=main', '--bare', cwd=origin)
seed = tmp_path / 'seed'
_git('clone', str(origin), str(seed), cwd=tmp_path)
_git('config', 'user.email', 'test@example.com', cwd=seed)
_git('config', 'user.name', 'Test', cwd=seed)
(seed / 'manifest.json').write_text(json.dumps({
"id": "gitplug", "name": "Git Plug", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "2.0.0",
}), encoding='utf-8')
(seed / 'manager.py').write_text("class P: pass\n", encoding='utf-8')
_git('add', '-A', cwd=seed)
_git('commit', '-m', 'initial', cwd=seed)
_git('push', '-u', 'origin', 'main', cwd=seed)
plugins_dir = tmp_path / 'plugin-repos'
plugins_dir.mkdir()
work = plugins_dir / 'gitplug'
_git('clone', str(origin), str(work), cwd=tmp_path)
_git('config', 'user.email', 'test@example.com', cwd=work)
_git('config', 'user.name', 'Test', cwd=work)
mgr = PluginStoreManager(plugins_dir=str(plugins_dir))
mgr.logger = MagicMock()
# No registry entry, so update_plugin takes the plain-pull branch
# rather than the remote-mismatch or already-current shortcuts.
monkeypatch.setattr(mgr, 'fetch_registry', lambda *a, **k: None)
monkeypatch.setattr(mgr, 'get_plugin_info', lambda *a, **k: None)
monkeypatch.setattr(mgr, '_install_dependencies', lambda *a, **k: True)
return mgr, seed, work
def _push(self, seed, manifest_text):
(seed / 'manifest.json').write_text(manifest_text, encoding='utf-8')
_git('add', '-A', cwd=seed)
_git('commit', '-m', 'update', cwd=seed)
_git('push', 'origin', 'main', cwd=seed)
def _head(self, work):
return _git('rev-parse', 'HEAD', cwd=work).stdout.strip()
def test_refuses_a_pulled_commit_that_needs_a_newer_core(
self, checkout, monkeypatch):
mgr, seed, work = checkout
before = self._head(work)
self._push(seed, json.dumps({
"id": "gitplug", "name": "Git Plug", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "9.9.9",
}))
import src
monkeypatch.setattr(src, '__version__', '3.2.0')
assert mgr.update_plugin('gitplug') is False
assert self._head(work) == before, (
"a refused update must leave the checkout on the commit it was "
"already running, not on the one it cannot load")
floor = json.loads((work / 'manifest.json').read_text())
assert floor['min_ledmatrix_version'] == '2.0.0', (
"the reset must restore the working tree, not just the ref")
def test_allows_a_pulled_commit_the_core_can_run(
self, checkout, monkeypatch):
"""The guard against over-refusing. A gate that refuses everything
passes the test above and breaks every update."""
mgr, seed, work = checkout
before = self._head(work)
self._push(seed, json.dumps({
"id": "gitplug", "name": "Git Plug", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "3.0.0",
}))
import src
monkeypatch.setattr(src, '__version__', '3.2.0')
assert mgr.update_plugin('gitplug') is True
assert self._head(work) != before
def test_an_unreadable_manifest_after_pull_is_not_a_refusal(
self, checkout, monkeypatch):
"""Rule 1 of this file — refuse only on evidence. A manifest that
will not parse declares no floor, so it is not evidence of anything."""
mgr, seed, work = checkout
self._push(seed, '{ this is not json')
import src
monkeypatch.setattr(src, '__version__', '3.2.0')
assert mgr.update_plugin('gitplug') is True
def test_an_untrustworthy_core_does_not_block_a_2_0_0_floor_on_pull(
self, checkout, monkeypatch):
"""The same regression guard as
`test_untrustworthy_core_does_not_block_installs`, because this path
now shares that rule: a v3.1.0 release reports 1.0.0, and nearly every
published manifest floors at 2.0.0."""
mgr, seed, work = checkout
before = self._head(work)
self._push(seed, json.dumps({
"id": "gitplug", "name": "Git Plug", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "2.0.0",
"description": "changed",
}))
import src
monkeypatch.setattr(src, '__version__', '1.0.0')
assert mgr.update_plugin('gitplug') is True
assert self._head(work) != before
def test_a_failed_stash_stops_the_update_before_pulling(
self, checkout, monkeypatch):
"""Uncommitted work is not collateral for the gate.
The gate's only rollback is `git reset --hard`, which discards
uncommitted tracked edits. update_plugin stashes them first -- but when
that stash fails it used to pull anyway, and a pull touching different
files succeeds on a dirty tree. Refuse, refuse the rollback's rollback,
and the user's edits go with it.
"""
mgr, seed, work = checkout
before = self._head(work)
(work / 'manager.py').write_text(
"class P:\n MINE = 'do not lose this'\n", encoding='utf-8')
self._push(seed, json.dumps({
"id": "gitplug", "name": "Git Plug", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "9.9.9",
}))
real_run = subprocess.run
def fail_stash(cmd, *a, **k):
if isinstance(cmd, list) and 'stash' in cmd:
return subprocess.CompletedProcess(cmd, 1, '', 'stash boom')
return real_run(cmd, *a, **k)
import src
monkeypatch.setattr(src, '__version__', '3.2.0')
monkeypatch.setattr(
'src.plugin_system.store_manager.subprocess.run', fail_stash)
assert mgr.update_plugin('gitplug') is False
assert self._head(work) == before, "must not pull what it cannot undo"
assert 'do not lose this' in (work / 'manager.py').read_text(), (
"the local edit the stash failed to save must still be there")
def test_rollback_reports_the_recovery_command_when_git_fails(
self, checkout, monkeypatch):
"""If the reset itself fails the user is left on a version that cannot
load, so the log line has to carry the command that fixes it —
it is the only thing standing between them and a manual reinstall."""
mgr, seed, work = checkout
self._push(seed, json.dumps({
"id": "gitplug", "name": "Git Plug", "class_name": "P",
"display_modes": ["a"], "min_ledmatrix_version": "9.9.9",
}))
real_run = subprocess.run
def fail_reset(cmd, *a, **k):
if isinstance(cmd, list) and 'reset' in cmd:
return subprocess.CompletedProcess(cmd, 1, '', 'reset boom')
return real_run(cmd, *a, **k)
import src
monkeypatch.setattr(src, '__version__', '3.2.0')
monkeypatch.setattr(
'src.plugin_system.store_manager.subprocess.run', fail_reset)
assert mgr.update_plugin('gitplug') is False
logged = ' '.join(str(c) for c in mgr.logger.error.call_args_list)
assert 'reset --hard' in logged, logged
class TestLoaderAndStoreAgree:
"""Both read the same manifests; a disagreement means one of them is
lying to the user."""
@@ -0,0 +1,63 @@
"""Tests for PluginManager.discovered_plugin_ids().
The config-watcher thread needs the set of discovered plugin ids while the
render thread may be rebuilding plugin_manifests. Iterating that dict directly
can observe a half-populated mapping or raise "dictionary changed size during
iteration", so the accessor snapshots it under the discovery lock.
"""
import tempfile
import threading
from pathlib import Path
import pytest
from src.plugin_system.plugin_manager import PluginManager
@pytest.fixture
def pm():
with tempfile.TemporaryDirectory() as tmp:
yield PluginManager(plugins_dir=str(Path(tmp) / "plugins"))
def test_returns_the_discovered_ids(pm):
pm.plugin_manifests = {"clock-simple": {}, "hockey-scoreboard": {}}
assert pm.discovered_plugin_ids() == {"clock-simple", "hockey-scoreboard"}
def test_empty_when_nothing_discovered(pm):
pm.plugin_manifests = {}
assert pm.discovered_plugin_ids() == set()
def test_is_a_snapshot_not_a_live_view(pm):
"""The caller iterates the result on another thread; it must not alias
the mapping discovery is still writing to."""
pm.plugin_manifests = {"clock-simple": {}}
snapshot = pm.discovered_plugin_ids()
pm.plugin_manifests["hockey-scoreboard"] = {}
assert snapshot == {"clock-simple"}
def test_takes_the_discovery_lock(pm):
"""Guards against the lock being dropped in a later refactor: with the
lock held by another thread the call must block rather than read."""
pm.plugin_manifests = {"clock-simple": {}}
finished = threading.Event()
def call():
pm.discovered_plugin_ids()
finished.set()
pm._discovery_lock.acquire()
try:
# RLock is reentrant per-thread, so use a *different* thread to prove
# the accessor actually waits on it.
t = threading.Thread(target=call, daemon=True)
t.start()
assert not finished.wait(timeout=0.3), "accessor did not take the discovery lock"
finally:
pm._discovery_lock.release()
t.join(timeout=2)
assert finished.is_set()
+166
View File
@@ -0,0 +1,166 @@
"""Plugin state history must not grow without bound.
`PluginStateManager` recorded every state transition in a per-plugin list and
never trimmed it. The only code that removed entries was `clear_state()`, called
solely from `PluginManager.unload_plugin()`, so a plugin that stays loaded --
i.e. normal operation -- never released a single entry.
The list is written on the hot scheduling path. Every update cycle appends
twice: `_reserve_for_update()` sets RUNNING and `_finish()` sets ENABLED back
again. At the default 60-second update interval that is 2,880 entries per
plugin per day, and nothing ever reads the entries -- `get_state_info()` only
takes their `len()`. It is pure dead weight.
Measured against the unpatched class, ten plugins on a 60s interval retain
864,010 transitions after thirty simulated days, for 231 MB of heap. On a 1 GB
Pi that is fatal on its own, and the failure is not a clean OOM: once
MemAvailable falls far enough, fork() starts returning ENOMEM, so sshd accepts
connections and closes them before its banner while the kernel still answers
pings. The board looks like a hardware fault and needs a power cycle.
These tests pin the cap, the retention order, and the one piece of behaviour the
cap must not change: `state_history_count` is surfaced through the web API, so
it has to keep reporting the lifetime total rather than plateauing at the cap.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.plugin_system.plugin_state import ( # noqa: E402
MAX_STATE_HISTORY_PER_PLUGIN,
PluginState,
PluginStateManager,
)
def _cycle_updates(manager, plugin_id, cycles):
"""Drive the real scheduling path: RUNNING on reserve, ENABLED on finish."""
for _ in range(cycles):
manager.set_state(plugin_id, PluginState.RUNNING)
manager.set_state(plugin_id, PluginState.ENABLED)
def test_state_history_is_capped():
"""A day of updates must not retain a day of transitions."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
# One simulated day at the default 60s update interval.
_cycle_updates(manager, "clock", 1440)
history = manager.get_state_history("clock")
assert len(history) <= MAX_STATE_HISTORY_PER_PLUGIN, (
f"history grew to {len(history)} entries; it is never trimmed"
)
def test_state_history_keeps_the_most_recent_transitions():
"""Trimming drops the oldest entries, not the newest."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", MAX_STATE_HISTORY_PER_PLUGIN)
history = manager.get_state_history("clock")
# The scheduling cycle ends on ENABLED, so the newest entry is the
# RUNNING -> ENABLED half of the last cycle.
assert history[-1]["from"] == PluginState.RUNNING.value
assert history[-1]["to"] == PluginState.ENABLED.value
# And the very first ENABLED transition has aged out.
assert history[0]["from"] != PluginState.UNLOADED.value
def test_state_history_count_reports_lifetime_total():
"""The count exposed through the API must not plateau at the cap.
`get_state_info()['state_history_count']` is surfaced by the web UI. Capping
the retained list must not turn it into "entries we happen to still hold".
"""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
total = 1
cycles = MAX_STATE_HISTORY_PER_PLUGIN * 2
_cycle_updates(manager, "clock", cycles)
total += cycles * 2
info = manager.get_state_info("clock")
assert info["state_history_count"] == total
assert len(manager.get_state_history("clock")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_error_transitions_are_capped_too():
"""set_state_with_error() appends to the same list and needs the same cap."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 2):
manager.set_state_with_error(
"clock",
PluginState.ENABLED,
{"reason": "update timeout"},
error=RuntimeError("boom"),
)
assert len(manager.get_state_history("clock")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_history_is_isolated_per_plugin():
"""The cap is per plugin, not shared across the manager."""
manager = PluginStateManager()
for plugin_id in ("clock", "weather"):
manager.set_state(plugin_id, PluginState.ENABLED)
_cycle_updates(manager, plugin_id, 50)
assert len(manager.get_state_history("clock")) == 101
assert len(manager.get_state_history("weather")) == 101
def test_get_state_history_returns_a_copy():
"""Callers must not be able to mutate the manager's internal history."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
history = manager.get_state_history("clock")
history.clear()
assert len(manager.get_state_history("clock")) == 1
def test_get_state_history_entries_are_copies():
"""Copying the outer list is not enough -- the entries are handed out too.
A caller holding a returned transition must not be able to rewrite the
manager's record of what happened.
"""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
entry = manager.get_state_history("clock")[0]
entry["to"] = "tampered"
entry["error"] = "injected"
stored = manager.get_state_history("clock")[0]
assert stored["to"] == PluginState.ENABLED.value
assert stored["error"] is None
def test_clear_state_drops_history():
"""Unloading a plugin still releases everything it accumulated."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", 10)
manager.clear_state("clock")
assert manager.get_state_history("clock") == []
assert manager.get_state_info("clock")["state_history_count"] == 0
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
+209
View File
@@ -0,0 +1,209 @@
"""Retention is bounded by age first and by count second.
The cap added in the parent change is a flat entry count, and an entry count
answers the wrong question. What a reader wants from this history is "the last
couple of hours"; how many transitions that is depends entirely on the
plugin's update interval, which on a real board spans 2s to 3600s. A flat 200
entries is 4.2 days of history for the slowest plugin and 3.3 minutes for the
fastest -- so the plugin churning hardest, the one actually worth looking at,
keeps the least.
Trimming by age makes the retained window comparable whatever the cadence, and
the count then serves only as a memory ceiling for pollers fast enough to
produce thousands of transitions inside that window.
"""
import time
import pytest
from src.plugin_system.plugin_state import (
PluginState,
PluginStateManager,
MAX_STATE_HISTORY_PER_PLUGIN,
STATE_HISTORY_MAX_AGE_SECONDS,
)
class FakeClock:
"""A monotonic clock the test drives, so no test has to sleep."""
def __init__(self):
self.t = 1000.0
def __call__(self):
return self.t
def advance(self, seconds):
self.t += seconds
@pytest.fixture
def clock(monkeypatch):
c = FakeClock()
monkeypatch.setattr("src.plugin_system.plugin_state.time.monotonic", c)
return c
def _cycle(manager, plugin_id, clock, interval, cycles):
"""One update cycle: RUNNING on reserve, ENABLED on finish."""
for _ in range(cycles):
manager.set_state(plugin_id, PluginState.RUNNING)
manager.set_state(plugin_id, PluginState.ENABLED)
clock.advance(interval)
def test_transitions_older_than_the_window_are_dropped(clock):
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=10)
assert len(m.get_state_history("clock")) == 20
# Nothing happens for longer than the window, then one more cycle.
clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
_cycle(m, "clock", clock, interval=60, cycles=1)
assert len(m.get_state_history("clock")) == 2, (
"only the transitions inside the window should survive")
def test_every_plugin_keeps_the_same_WINDOW_not_the_same_COUNT(clock):
"""The point of the age policy, stated as the property that distinguishes it.
Run both plugins for three times the retention window. Under a flat count
cap the slow one would still be holding transitions from hours before the
window, because it never produces enough entries to evict them. Under the
age policy each plugin retains its own last two hours and no more --
different entry counts, same span of time.
"""
window = STATE_HISTORY_MAX_AGE_SECONDS
m = PluginStateManager()
_cycle(m, "slow", clock, interval=60, cycles=(3 * window) // 60)
slow = len(m.get_state_history("slow"))
# Assert the property directly rather than a derived count. The guarantee
# is about the SPAN of retained history, not its age against the current
# clock: trimming happens on append, so a plugin that has gone quiet keeps
# its last window until it writes again. That is intentional -- it is
# bounded either way, and a lazy trim costs nothing on the hot path.
stamps = [stamp for stamp, _ in m._state_history["slow"]]
assert stamps[-1] - stamps[0] <= window, (
f"retained history spans {stamps[-1] - stamps[0]:.0f}s, "
f"window is {window}s")
assert slow < 2 * ((3 * window) // 60), (
f"slow plugin kept {slow} entries -- three windows' worth was retained")
clock.t = 1000.0
_cycle(m, "fast", clock, interval=2, cycles=(3 * window) // 2)
fast = len(m.get_state_history("fast"))
# Different counts, and the fast poller keeps more of them -- under a flat
# count cap these would be equal and the fast one would cover minutes.
assert fast > slow, f"fast={fast} slow={slow}"
def test_the_count_ceiling_still_bounds_a_fast_poller(clock):
"""Age alone would let a 2s plugin hold 7,200 entries."""
m = PluginStateManager()
_cycle(m, "flights", clock, interval=2, cycles=STATE_HISTORY_MAX_AGE_SECONDS)
assert len(m.get_state_history("flights")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_a_burst_inside_the_window_is_capped_not_kept(clock):
"""Transitions with no time between them still cannot grow without bound."""
m = PluginStateManager()
for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 3):
m.set_state("flapping", PluginState.RUNNING) # clock never advances
assert len(m.get_state_history("flapping")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_ageing_out_does_not_disturb_the_lifetime_count(clock):
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=10)
clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
_cycle(m, "clock", clock, interval=60, cycles=1)
assert len(m.get_state_history("clock")) == 2
assert m.get_state_info("clock")["state_history_count"] == 22, (
"the lifetime total must survive trimming, it is the flap signal")
def test_the_surviving_entries_are_the_recent_ones(clock):
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=5)
clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
m.set_state("clock", PluginState.ERROR)
history = m.get_state_history("clock")
assert [h["to"] for h in history] == ["error"]
def test_a_monotonic_clock_is_used_not_the_wall_clock(clock):
"""A DST shift or NTP step must not flush the history.
The trim reads time.monotonic(); the human-readable datetime inside each
transition is for display only.
"""
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=3)
before = len(m.get_state_history("clock"))
import datetime as real_datetime
class ShiftedDatetime(real_datetime.datetime):
@classmethod
def now(cls, tz=None):
return real_datetime.datetime(1999, 1, 1) # clock jumps backwards
import src.plugin_system.plugin_state as ps
original = ps.datetime
ps.datetime = ShiftedDatetime
try:
m.set_state("clock", PluginState.ENABLED)
finally:
ps.datetime = original
assert len(m.get_state_history("clock")) == before + 1, (
"a wall-clock jump must not trim anything")
def test_get_state_info_is_a_consistent_snapshot():
"""An unload running concurrently must not be observed half-done.
Each field used to be read under its own lock, so clear_state() could
interleave: 'state' read before the removal, 'state_history_count' after,
handing a caller a plugin that is ENABLED with zero transitions. The whole
payload is now built in one critical section.
"""
import threading
m = PluginStateManager()
for _ in range(50):
m.set_state("clock", PluginState.RUNNING)
m.set_state("clock", PluginState.ENABLED)
inconsistent = []
stop = threading.Event()
def reader():
while not stop.is_set():
info = m.get_state_info("clock")
# Either fully present or fully cleared -- never a live state with
# a wiped count.
if info["state"] != PluginState.UNLOADED.value and \
info["state_history_count"] == 0:
inconsistent.append(info)
return
def clearer():
for _ in range(200):
for _ in range(20):
m.set_state("clock", PluginState.ENABLED)
m.clear_state("clock")
t = threading.Thread(target=reader, daemon=True)
t.start()
clearer()
stop.set()
t.join(timeout=5)
assert not inconsistent, f"observed a torn snapshot: {inconsistent[:1]}"
+195
View File
@@ -0,0 +1,195 @@
"""The card helpers the eight scoreboards now share.
These bodies lived in eight byte-identical copies. Moving them here means one
fix reaches every scoreboard — and that a mistake does too, which is what this
file guards. Each case below is one the plugins' own code already handled; the
point is that it keeps handling it.
The functions take ``config``/``logger``/``fonts`` as arguments rather than
reading them off an instance, so a plugin keeps its method and delegates the
body. That is what let all eight adopt this with byte-identical renders.
"""
import logging
import json
import os
import pytest
from src.common import sports_card as C
@pytest.fixture
def log():
return logging.getLogger("test_sports_card")
class TestSettingsLookup:
def test_reads_the_scroll_card_block(self):
cfg = {"scroll_card": {"vs_text": "@"}}
assert C.scroll_card_option(cfg, "vs_text", "VS") == "@"
@pytest.mark.parametrize("cfg", [None, {}, {"scroll_card": None},
{"scroll_card": {"vs_text": None}}])
def test_missing_or_null_falls_back(self, cfg):
"""A null in config means "unset", not "empty string"."""
assert C.scroll_card_option(cfg, "vs_text", "VS") == "VS"
def test_upcoming_center_rejects_unknown_modes(self):
for bad in ("sideways", "", None, 7):
assert C.upcoming_center_mode({"scroll_card": {"upcoming_center": bad}}) == "vs"
assert C.upcoming_center_mode({"scroll_card": {"upcoming_center": "DATE_TIME"}}) == "date_time"
class TestColour:
def test_rgb_list_and_hex_both_work(self):
assert C.element_color({"customization": {"score_text": {"text_color": [1, 2, 3]}}},
"score_text") == (1, 2, 3)
assert C.element_color({"customization": {"score_text": {"text_color": "#ff8000"}}},
"score_text") == (255, 128, 0)
@pytest.mark.parametrize("value", ["nope", "#fff", [1, 2], None, ["a", "b", "c"]])
def test_unusable_colour_falls_back(self, value):
cfg = {"customization": {"score_text": {"text_color": value}}}
assert C.element_color(cfg, "score_text", (9, 9, 9)) == (9, 9, 9)
def test_coerce_rgb_clamps_rather_than_rejecting(self):
assert C.coerce_rgb([300, -5, 20], (0, 0, 0)) == (255, 0, 20)
def test_coerce_rgb_refuses_a_three_character_string(self):
""""123" would otherwise iterate into three digits and yield a colour."""
assert C.coerce_rgb("123", (7, 7, 7)) == (7, 7, 7)
def test_font_colour_is_resolved_by_identity(self):
a, b = object(), object()
cfg = {"customization": {"score_text": {"text_color": [4, 5, 6]}}}
assert C.font_color(cfg, {"score": a, "team": b}, a) == (4, 5, 6)
def test_a_shared_face_gives_up_rather_than_guessing(self):
"""One object used for two elements has no single right colour."""
shared = object()
cfg = {"customization": {"score_text": {"text_color": [4, 5, 6]}}}
assert C.font_color(cfg, {"score": shared, "team": shared}, shared) == (255, 255, 255)
class TestFavourites:
GAME = {"home_abbr": "TB", "away_abbr": "NO", "home_score": "21", "away_score": "17"}
def test_win_loss_and_tie(self):
cfg = {"favorite_teams": ["TB"]}
assert C.favorite_result(cfg, self.GAME) == "win"
assert C.favorite_result({"favorite_teams": ["NO"]}, self.GAME) == "loss"
tied = dict(self.GAME, home_score="3", away_score="3")
assert C.favorite_result(cfg, tied) == "tie"
def test_no_verdict_without_exactly_one_favourite_side(self):
assert C.favorite_result({}, self.GAME) is None
assert C.favorite_result({"favorite_teams": ["TB", "NO"]}, self.GAME) is None
assert C.favorite_result({"favorite_teams": ["SEA"]}, self.GAME) is None
def test_unusable_scores_give_no_verdict(self):
bad = dict(self.GAME, home_score="x")
assert C.favorite_result({"favorite_teams": ["TB"]}, bad) is None
def test_nested_payload_shape_is_read_too(self):
game = {"home_team": {"abbrev": "TB", "score": 9},
"away_team": {"abbrev": "NO", "score": 2}}
assert C.side_score(game, "home") == 9
assert C.side_is_favorite(game, "home", {"TB"}) is True
def test_matches_on_id_where_abbreviations_collide(self):
"""NRL keys favourites by ESPN id; abbreviations are not unique there."""
game = {"home_abbr": "SYD", "home_id": "4321"}
assert C.side_is_favorite(game, "home", {"4321"}) is True
def test_game_and_config_favourites_are_both_used(self):
"""Games carry resolved dynamic groups; config catches later edits."""
game = dict(self.GAME, favorite_teams=["NO"], league="nfl")
assert set(C.favorite_teams_for({"nfl": {"favorite_teams": ["TB"]}}, game)) == {"NO", "TB"}
class TestDateAndTime:
def test_date_formats(self, log):
for fmt, want in [("abbrev", "Sep 5"), ("numeric", "9/5"),
("day_first", "5 Sep"), ("numeric_day_first", "5/9")]:
cfg = {"scroll_card": {"date_format": fmt}}
assert C.format_game_date(cfg, log, "9/5") == want
@pytest.mark.parametrize("raw", ["", "garbage", "13/40", "no/slash/here"])
def test_unparseable_dates_pass_through(self, log, raw):
assert C.format_game_date({}, log, raw) == raw.strip()
def test_24h_conversion(self):
cfg = {"scroll_card": {"time_format": "24h"}}
assert C.format_game_time(cfg, "7:30 PM") == "19:30"
assert C.format_game_time(cfg, "12:00 AM") == "00:00"
assert C.format_game_time(cfg, "12:15 PM") == "12:15"
def test_12h_is_left_alone_and_junk_survives(self):
assert C.format_game_time({}, "7:30 PM") == "7:30 PM"
assert C.format_game_time({"scroll_card": {"time_format": "24h"}}, "soon") == "soon"
def test_a_bad_timezone_falls_back_to_utc(self, log):
"""A typo in config should not blank the card."""
from datetime import timezone
assert C.card_tzinfo({"timezone": "Not/AZone"}, log) is timezone.utc
class TestFontSizing:
def test_snaps_to_the_faces_pixel_grid(self):
assert C.crisp_size("4x6-font.ttf", 6) == 7 # 7px grid
assert C.crisp_size("PressStart2P-Regular.ttf", 10) == 8
assert C.crisp_size("PressStart2P-Regular.ttf", 13) == 16
def test_an_unknown_face_is_never_second_guessed(self):
assert C.crisp_size("SomeUserFont.ttf", 11) == 11
def test_aliases_resolve_before_the_grid_lookup(self):
assert C.crisp_size("four_by_six", 6) == C.crisp_size("4x6-font.ttf", 6)
@pytest.mark.parametrize("desired", [0, -3, None])
def test_unusable_sizes_pass_through_without_raising(self, desired):
"""None reached this in the field; football's variant raised TypeError."""
assert C.crisp_size("4x6-font.ttf", desired) == desired
def test_schema_cache_is_keyed_per_schema_not_globally(self, tmp_path):
"""Two plugins declaring different defaults must not share an answer."""
a, b = tmp_path / "a.json", tmp_path / "b.json"
for path, size in ((a, 11), (b, 22)):
path.write_text(json.dumps({"properties": {"customization": {"properties": {
"score_text": {"properties": {"font_size": {"default": size}}}}}}}))
assert C.schema_font_size(str(a), "score_text") == 11
assert C.schema_font_size(str(b), "score_text") == 22
def test_a_missing_schema_is_not_an_error(self, tmp_path):
assert C.schema_font_size(str(tmp_path / "nope.json"), "score_text") is None
def test_a_configured_size_matching_the_schema_default_is_not_a_choice(self, tmp_path):
"""The web UI writes the whole default block on every save, so
font_size == schema default carries no intent and must not pin the
install to an off-grid size forever."""
schema = tmp_path / "s.json"
schema.write_text(json.dumps({"properties": {"customization": {"properties": {
"score_text": {"properties": {"font_size": {"default": 10}}}}}}}))
got = C.resolve_font_size(str(schema), {"font_size": 10}, "score_text", 10,
"PressStart2P-Regular.ttf")
assert got == 8, "a default-valued size should snap to the grid"
def test_a_real_choice_wins(self, tmp_path):
schema = tmp_path / "s.json"
schema.write_text(json.dumps({"properties": {"customization": {"properties": {
"score_text": {"properties": {"font_size": {"default": 10}}}}}}}))
got = C.resolve_font_size(str(schema), {"font_size": 13}, "score_text", 10,
"PressStart2P-Regular.ttf")
assert got == 13, "an explicit size the user chose is not second-guessed"
class TestTables:
def test_every_font_key_maps_to_an_element(self):
assert set(C.ELEMENT_FOR_FONT) == {"score", "time", "team", "status", "detail", "rank"}
def test_result_colours_cover_every_verdict(self):
assert set(C.FAVORITE_RESULT_COLOR_DEFAULTS) == {"win", "loss", "tie"}
def test_month_and_weekday_tables_are_complete(self):
assert len(C.MONTH_ABBR) == 12 and len(C.WEEKDAY_ABBR) == 7
+258
View File
@@ -0,0 +1,258 @@
"""The shared card geometry, exercised against a host that supplies only what
the mixin's contract names.
The point of these is the contract, not the arithmetic. The mixin reaches for
``display_width``, ``fonts``, ``config`` and six ``sports_card`` delegations
through ``self``, and the eight plugins are what actually provide them. A
stub host that provides exactly the documented surface and nothing else is
what catches the mixin quietly growing a dependency the plugins do not have.
"""
import pytest
from PIL import Image, ImageDraw, ImageFont
from src.common.sports_game_renderer import SportsGameRendererMixin
class Host(SportsGameRendererMixin):
"""The documented contract, and not one attribute more."""
def __init__(self, width=128, height=32, config=None, scroll=None):
self.display_width = width
self.display_height = height
self.config = config or {}
self.logger = _Logger()
self._team_rankings_cache = {}
self._scroll = scroll or {}
font = ImageFont.load_default()
self.fonts = {'score': font, 'time': font, 'detail': font}
self.drawn = []
# -- the six sports_card delegations the mixin calls --
def _scroll_card_option(self, key, default=None):
return self._scroll.get(key, default)
def _upcoming_center_mode(self):
return self._scroll.get('upcoming_center', 'vs')
def _vs_text(self):
return self._scroll.get('vs_text', 'VS')
def _element_color(self, element):
return (255, 255, 255)
def _format_game_date(self, raw, game):
return raw
def _format_game_time(self, raw):
return raw
# -- the one hook whose body genuinely varies per plugin --
def _draw_text_with_outline(self, draw, text, position, font,
fill=None, outline_color=(0, 0, 0)):
self.drawn.append((text, position))
class _Logger:
def debug(self, *a, **k):
pass
def _draw():
return ImageDraw.Draw(Image.new("RGB", (256, 64)))
class TestCenterGap:
def test_explicit_center_gap_wins_outright(self):
assert Host(scroll={'center_gap': 31})._center_gap_width() == 31
def test_explicit_zero_restores_edge_to_edge_logos(self):
# 0 is a real setting, not a falsy miss -- the guard is `>= 0`.
assert Host(scroll={'center_gap': 0})._center_gap_width() == 0
def test_otherwise_it_scales_with_card_width_within_the_clamp(self):
h = Host(width=512)
# 512 * 0.28 = 143, clamped to the 40px ceiling.
assert h._center_gap_width() >= h.CENTER_GAP_MIN_PX
def test_the_gap_never_ends_up_narrower_than_the_score(self):
# This is the bug the measurement exists to prevent: a derived gap
# smaller than the rendered score drew the score over the logos.
h = Host(width=64)
assert h._center_gap_width() >= h._score_reserve_width()
def test_a_junk_ratio_falls_back_to_the_floor(self):
h = Host(scroll={'center_gap_ratio': 'wide'})
assert h._center_gap_width() == h.CENTER_GAP_MIN_PX
class TestNonFiniteSettings:
"""inf reaches int() and raises OverflowError, which the old
`except (TypeError, ValueError)` did not catch -- so one bad config value
aborted the entire card render rather than falling back."""
@pytest.mark.parametrize("bad", [float("inf"), float("-inf")])
def test_a_non_finite_center_gap_falls_back(self, bad):
h = Host(scroll={'center_gap': bad})
assert h._center_gap_width() >= h.CENTER_GAP_MIN_PX
@pytest.mark.parametrize("bad", [float("inf"), float("-inf"), float("nan")])
def test_a_non_finite_ratio_falls_back_to_the_floor(self, bad):
h = Host(scroll={'center_gap_ratio': bad})
assert h._center_gap_width() == h.CENTER_GAP_MIN_PX
@pytest.mark.parametrize("bad", [float("inf"), float("-inf")])
def test_non_finite_clamp_bounds_fall_back(self, bad):
h = Host(scroll={'center_gap_min': bad, 'center_gap_max': bad})
assert h._center_gap_width() == h.CENTER_GAP_MIN_PX
@pytest.mark.parametrize("bad", [float("inf"), float("-inf"), "inf", "-inf", "nan"])
def test_a_non_finite_layout_offset_gives_the_default(self, bad):
cfg = {'customization': {'layout': {'score': {'x_offset': bad}}}}
assert Host(config=cfg)._layout_offset('score', 'x_offset', 7) == 7
def test_a_finite_value_is_still_honoured(self):
# The guard must not swallow ordinary settings.
assert Host(scroll={'center_gap': 31})._center_gap_width() == 31
cfg = {'customization': {'layout': {'score': {'x_offset': -3}}}}
assert Host(config=cfg)._layout_offset('score', 'x_offset', 7) == -3
class TestScoreReserve:
def test_it_measures_the_probe_plus_both_gutters(self):
h = Host()
assert h._score_reserve_width() > 2 * h._SCORE_LOGO_GUTTER_PX
def test_a_wider_probe_reserves_more(self):
class Wide(Host):
_SCORE_PROBE = "000-000"
assert Wide()._score_reserve_width() > Host()._score_reserve_width()
def test_an_unmeasurable_font_reserves_nothing_rather_than_raising(self):
h = Host()
h.fonts = {'score': object()}
assert h._score_reserve_width() == 0
class TestLogoSlot:
def test_the_slot_is_what_is_left_after_the_gap(self):
h = Host(width=128, scroll={'center_gap': 40})
assert h._logo_slot_width() == 44
def test_it_is_not_capped_at_the_card_height(self):
# The height cap is what froze logos at 46px on a 128px card.
h = Host(width=512, height=32, scroll={'center_gap': 40})
assert h._logo_slot_width() > h.display_height
def test_a_gap_wider_than_the_card_still_leaves_a_usable_slot(self):
assert Host(width=64, scroll={'center_gap': 200})._logo_slot_width() == 8
def test_the_cache_key_is_scoped_to_the_slot_not_just_the_name(self):
# One cache dict is shared by renderers of different card widths.
narrow = Host(width=64, scroll={'center_gap': 20})._logo_cache_key("NYY")
wide = Host(width=256, scroll={'center_gap': 20})._logo_cache_key("NYY")
assert narrow != wide
class TestLayoutOffset:
def _host(self, value):
return Host(config={'customization': {'layout': {'score': {'x_offset': value}}}})
def test_it_reads_the_same_block_as_the_full_screen_scorebug(self):
assert self._host(5)._layout_offset('score', 'x_offset') == 5
def test_a_string_offset_from_the_web_ui_is_coerced(self):
assert self._host("-3")._layout_offset('score', 'x_offset') == -3
def test_a_bool_is_not_silently_an_offset_of_one(self):
assert self._host(True)._layout_offset('score', 'x_offset', 9) == 9
@pytest.mark.parametrize("cfg", [{}, {'customization': {}},
{'customization': {'layout': {}}}])
def test_a_missing_block_gives_the_default(self, cfg):
assert Host(config=cfg)._layout_offset('score', 'x_offset', 7) == 7
def test_an_unparseable_offset_gives_the_default(self):
assert self._host("left")._layout_offset('score', 'x_offset', 4) == 4
class TestUpcomingCenter:
def test_none_draws_nothing(self):
h = Host(scroll={'upcoming_center': 'none'})
h._draw_upcoming_center(_draw(), {})
assert h.drawn == []
def test_vs_is_the_default_and_never_a_score(self):
# An upcoming game has not started; the extractor's 0-0 is noise.
h = Host()
h._draw_upcoming_center(_draw(), {'home_score': 0, 'away_score': 0})
assert [t for t, _ in h.drawn] == ['VS']
def test_an_empty_vs_string_draws_nothing(self):
h = Host(scroll={'vs_text': ''})
h._draw_upcoming_center(_draw(), {})
assert h.drawn == []
def test_date_time_stacks_both_lines(self):
h = Host(scroll={'upcoming_center': 'date_time'})
h._draw_upcoming_center(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
assert [t for t, _ in h.drawn] == ['Sep 19', '7:00 PM']
def test_hiding_both_lines_draws_nothing(self):
h = Host(scroll={'upcoming_center': 'date_time',
'show_date': False, 'show_time': False})
h._draw_upcoming_center(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
assert h.drawn == []
class TestUpcomingStatus:
def test_time_on_top_and_date_below_by_default(self):
h = Host()
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
assert [t for t, _ in h.drawn] == ['7:00 PM', 'Sep 19']
def test_swap_date_time_reverses_them(self):
h = Host(scroll={'swap_date_time': True})
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
assert [t for t, _ in h.drawn] == ['Sep 19', '7:00 PM']
def test_it_stays_out_of_the_way_when_the_centre_already_has_them(self):
# Otherwise the date and time print twice on the same card.
h = Host(scroll={'upcoming_center': 'date_time'})
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
assert h.drawn == []
def test_the_bottom_line_is_measured_not_a_fixed_offset(self):
# A fixed -7 ran "Sep 19" past the card wherever the detail font is
# 10px rather than 6px.
h = Host(height=64)
h._draw_upcoming_game_status(_draw(), {'game_date': 'Sep 19', 'game_time': '7:00 PM'})
bottom_y = h.drawn[1][1][1]
assert 0 <= bottom_y < h.display_height
class TestRankings:
def test_set_rankings_cache_replaces_the_cache(self):
h = Host()
h.set_rankings_cache({'UGA': 1})
assert h._team_rankings_cache == {'UGA': 1}
class TestContract:
def test_the_mixin_carries_no_state_of_its_own(self):
# Adoption must be one line on the class statement; a mixin with an
# __init__ would force eight constructors to cooperate.
assert '__init__' not in SportsGameRendererMixin.__dict__
def test_a_host_providing_the_documented_surface_needs_nothing_more(self):
# Host defines exactly what the module docstring names. If the mixin
# grows a new self.* dependency, this is what fails.
h = Host()
h._center_gap_width()
h._logo_slot_width()
h._logo_cache_key("X")
h._layout_offset('score', 'x_offset')
h._upcoming_date_and_time({})
h._draw_upcoming_center(_draw(), {})
h._draw_upcoming_game_status(_draw(), {})
h.set_rankings_cache({})
+376
View File
@@ -0,0 +1,376 @@
"""The shared sports.py mixins: their host contract, and _plugin_dir.
Two things are worth testing here and the rest is not. The 45 method bodies
moved verbatim from the plugins, so they are covered by the plugins' own tests
and by 176 byte-identical safety-harness renders. What is genuinely new is:
1. The contract. Every ``self.<CONSTANT>`` a mixin reads must be defined on the
mixin, or a host that does not happen to declare it raises AttributeError at
runtime. Two were missed on the first pass (_QUALITY_CHOICES and
_RANKING_COVERAGE_SECONDS); the eight plugins all declare them, so nothing
failed -- it would only have bitten a ninth. The test derives the list rather
than restating it, so the next omission fails here instead of in the field.
2. ``_plugin_dir``. This is the only line of genuinely new logic in the move. In
sports.py these methods found config_schema.json with ``__file__``; here that
is src/common/, so the plugin directory has to be recovered from the
instance -- and getting it wrong is silent, costing grid-snapped font sizes
(measured at 81% anti-aliased edges) rather than raising.
"""
import ast
import os
import sys
import types
from abc import ABC
import pytest
from src.common import sports_shared
from src.common.sports_shared import (
SportsCoreSharedMixin, SportsLiveSharedMixin, SportsRecentSharedMixin)
MIXINS = (SportsCoreSharedMixin, SportsLiveSharedMixin, SportsRecentSharedMixin)
def _constants_read_by_mixins():
"""Every ALL-CAPS ``self.X`` the mixin bodies read, found by parsing them."""
tree = ast.parse(open(sports_shared.__file__).read())
names = set()
for node in ast.walk(tree):
if (isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id == "self"
and node.attr.upper() == node.attr):
names.add(node.attr)
return names
class TestHostContract:
def test_every_constant_read_is_also_defined(self):
# Otherwise a host that does not declare it raises AttributeError the
# first time the code path runs -- which for these is mid-render.
missing = sorted(
name for name in _constants_read_by_mixins()
if not any(hasattr(m, name) for m in MIXINS))
assert missing == [], (
f"read but never defined on a mixin: {missing}. Give each a default "
f"on SportsCoreSharedMixin and document it in the module docstring.")
@pytest.mark.parametrize("name,expected", [
("_QUALITY_CHOICES", frozenset({"any", "ranked"})),
("_RANKING_COVERAGE_SECONDS", 3600),
("_SCORE_PROBE_TEXT", "00-00"),
("_FONT_DESIGN_HEIGHT", 32),
])
def test_defaults_match_what_the_plugins_ship(self, name, expected):
# The eight plugins declare their own copies, which shadow these. The
# values must still agree, or a ninth plugin inheriting the default
# behaves differently from the eight.
assert getattr(SportsCoreSharedMixin, name) == expected
def test_only_the_recent_mixin_carries_a_constructor(self):
# SportsCore and SportsLive keep their own __init__ -- those differ per
# plugin. SportsRecent.__init__ was one of the 48 byte-identical bodies,
# so it moved with the rest; that is deliberate, not an oversight.
assert "__init__" not in SportsCoreSharedMixin.__dict__
assert "__init__" not in SportsLiveSharedMixin.__dict__
assert "__init__" in SportsRecentSharedMixin.__dict__
def test_the_recent_constructor_still_chains_to_the_host(self):
"""Its zero-arg super() binds to where it is DEFINED, not where it is used.
Moving a body containing bare ``super()`` is the one move that can
change meaning: the compiler closes over __class__ = the defining class,
so after the move that is SportsRecentSharedMixin rather than the
plugin's SportsRecent. It still works only because the mixin is listed
first, leaving the host class next in the MRO -- adopt it in the other
order and the chain silently skips the host's __init__.
"""
calls = []
class Host:
def __init__(self, config, display_manager, cache_manager, logger, sport_key):
calls.append(sport_key)
self.mode_config = {}
class Recent(SportsRecentSharedMixin, Host):
pass
inst = Recent({}, None, None, None, "nhl")
assert calls == ["nhl"], "the host constructor must still run"
assert inst.current_game_index == 0
assert inst.update_interval == 3600
assert inst._zero_clock_timestamps == {}
def test_adopting_the_recent_mixin_second_would_skip_the_host(self):
# The failure mode the ordering above prevents, pinned so nobody
# "tidies" the base list.
calls = []
class Host:
def __init__(self, *a):
calls.append(a)
self.mode_config = {}
class Wrong(Host, SportsRecentSharedMixin):
pass
Wrong({}, None, None, None, "nhl")
# Host.__init__ wins and the mixin's setup never runs at all.
assert not hasattr(Wrong({}, None, None, None, "nhl"), "current_game_index")
class _Host(SportsCoreSharedMixin):
pass
def _write_plugin(tmp_path, name="fakeplug", schema=True):
"""A throwaway package on sys.path, with or without a config_schema.json."""
d = tmp_path / name
d.mkdir()
(d / "__init__.py").write_text("")
(d / "mod.py").write_text("class Leaf:\n pass\n")
if schema:
(d / "config_schema.json").write_text(
'{"properties": {"customization": {"properties": '
'{"score": {"properties": {"font_size": {"default": 16}}}}}}}')
return d
class TestPluginDir:
def test_it_finds_the_directory_holding_config_schema_json(self, tmp_path, monkeypatch):
d = _write_plugin(tmp_path)
monkeypatch.syspath_prepend(str(tmp_path))
mod = __import__("fakeplug.mod", fromlist=["Leaf"])
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
assert host._plugin_dir() == str(d)
def test_a_class_built_by_type_still_resolves(self, tmp_path, monkeypatch):
# SportsCore is an ABC, so type(name, bases, ns) reports __module__ as
# "abc" rather than the plugin -- which is exactly what the plugins'
# own tests build. Walking the MRO is what steps past it.
d = _write_plugin(tmp_path, "abcplug")
monkeypatch.syspath_prepend(str(tmp_path))
mod = __import__("abcplug.mod", fromlist=["Leaf"])
class Base(SportsCoreSharedMixin, mod.Leaf, ABC):
pass
synthetic = type("Probe", (Base,), {})
assert synthetic.__module__ == "abc", "precondition: the trap this guards"
assert synthetic.__new__(synthetic)._plugin_dir() == str(d)
def test_it_returns_none_when_no_schema_is_anywhere_on_the_mro(self, tmp_path, monkeypatch):
d = _write_plugin(tmp_path, "noschema", schema=False)
monkeypatch.syspath_prepend(str(tmp_path))
mod = __import__("noschema.mod", fromlist=["Leaf"])
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
# None rather than a wrong guess: _schema_font_size then caches empty
# and every element keeps its own default.
assert host._plugin_dir() is None
def test_it_never_returns_the_core_module_directory(self):
# The bug this replaced: __file__ pointed at src/common/, so the schema
# was never found and font sizes silently stopped snapping to the grid.
host = _Host()
core_common = os.path.dirname(os.path.abspath(sports_shared.__file__))
assert host._plugin_dir() != core_common
def test_a_module_with_no_file_is_skipped_not_crashed_on(self, monkeypatch):
# Namespace packages and some frozen/dynamic modules have no __file__.
ghost = types.ModuleType("ghost_no_file")
if hasattr(ghost, "__file__"):
del ghost.__file__
monkeypatch.setitem(sys.modules, "ghost_no_file", ghost)
cls = type("H", (SportsCoreSharedMixin,), {"__module__": "ghost_no_file"})
assert cls.__new__(cls)._plugin_dir() is None
class TestSchemaFontSize:
def test_it_reads_the_plugin_schema_not_the_cores(self, tmp_path, monkeypatch):
d = _write_plugin(tmp_path, "sizeplug")
monkeypatch.syspath_prepend(str(tmp_path))
mod = __import__("sizeplug.mod", fromlist=["Leaf"])
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
assert host._schema_font_size("score") == 16
def test_an_unknown_element_is_none(self, tmp_path, monkeypatch):
_write_plugin(tmp_path, "unkplug")
monkeypatch.syspath_prepend(str(tmp_path))
mod = __import__("unkplug.mod", fromlist=["Leaf"])
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
assert host._schema_font_size("nonesuch") is None
def test_an_empty_key_is_none_without_touching_the_disk(self):
assert _Host()._schema_font_size("") is None
def test_a_missing_schema_degrades_to_none_rather_than_raising(self, tmp_path, monkeypatch):
_write_plugin(tmp_path, "bareplug", schema=False)
monkeypatch.syspath_prepend(str(tmp_path))
mod = __import__("bareplug.mod", fromlist=["Leaf"])
host = type("H", (mod.Leaf, SportsCoreSharedMixin), {})()
assert host._schema_font_size("score") is None
class _LiveHost(SportsLiveSharedMixin):
"""The documented contract for the live mixin, and nothing else."""
def __init__(self, no_data_interval=300, stale_game_timeout=600, over=()):
self.no_data_interval = no_data_interval
self.stale_game_timeout = stale_game_timeout
self.game_update_timestamps = {}
self._over = set(over)
class _L:
def __getattr__(self, _n):
return lambda *a, **k: None
self.logger = _L()
def _is_game_really_over(self, game):
return game.get("id") in self._over
class TestLiveMixin:
"""These three moved to the core, so they are tested here.
They were already covered by hockey's and lacrosse's own tests, but those
two plugins disable live mode in their safety-harness fixtures, so the 176
renders never exercise this path. Testing the mixin directly means the
coverage no longer depends on which plugin happens to have a unit test.
"""
def test_a_stale_game_is_dropped_and_forgotten(self):
h = _LiveHost(stale_game_timeout=600)
import time as _t
h.game_update_timestamps["g1"] = {"last_seen": _t.time() - 5000}
games = [{"id": "g1", "home_abbr": "H", "away_abbr": "A"}]
h._detect_stale_games(games)
assert games == []
assert "g1" not in h.game_update_timestamps, "its timestamp must go too"
def test_a_fresh_game_survives(self):
h = _LiveHost(stale_game_timeout=600)
import time as _t
h.game_update_timestamps["g1"] = {"last_seen": _t.time() - 5}
games = [{"id": "g1"}]
h._detect_stale_games(games)
assert len(games) == 1
def test_a_game_never_seen_is_not_treated_as_stale(self):
# last_seen 0 means "no reading", not "seen at the epoch".
h = _LiveHost()
games = [{"id": "g1"}]
h._detect_stale_games(games)
assert len(games) == 1
def test_a_game_with_no_id_is_left_alone(self):
h = _LiveHost()
games = [{"home_abbr": "H"}]
h._detect_stale_games(games)
assert len(games) == 1
def test_a_finished_game_is_dropped_even_when_fresh(self):
h = _LiveHost(over=("g2",))
games = [{"id": "g1"}, {"id": "g2"}]
h._detect_stale_games(games)
assert [g["id"] for g in games] == ["g1"]
def test_removing_several_does_not_skip_any(self):
# It iterates a copy for exactly this reason; mutating the live list
# while looping would step over the element after each removal.
h = _LiveHost(over=("g1", "g2", "g3"))
games = [{"id": "g1"}, {"id": "g2"}, {"id": "g3"}]
h._detect_stale_games(games)
assert games == []
def test_the_idle_interval_escalates_with_the_empty_streak(self):
h = _LiveHost(no_data_interval=60)
h.live_idle_max_interval = 100000
base = h._idle_live_interval()
h._empty_live_streak = 6
short = h._idle_live_interval()
h._empty_live_streak = 24
long = h._idle_live_interval()
assert base < short < long
def test_the_ceiling_bounds_even_the_unescalated_interval(self):
# base > ceiling is a reachable config: the two settings are
# independent integers with no cross-validation. Returning base
# unclamped made the wait SHRINK as the streak grew.
h = _LiveHost(no_data_interval=3600)
h.live_idle_max_interval = 900
h._empty_live_streak = 0
assert h._idle_live_interval() == 900
h._empty_live_streak = 24
assert h._idle_live_interval() == 900
def test_finding_a_live_game_resets_the_streak(self):
h = _LiveHost()
h._note_live_fetch(False)
h._note_live_fetch(False)
assert h._empty_live_streak == 2
h._note_live_fetch(True)
assert h._empty_live_streak == 0
def test_the_streak_starts_from_absent_state(self):
# The host is not required to pre-declare _empty_live_streak.
h = _LiveHost()
assert not hasattr(h, "_empty_live_streak")
h._note_live_fetch(False)
assert h._empty_live_streak == 1
class TestPluginDirIsToldNotDeduced:
"""The regression that shipped: _plugin_dir returned None in production.
The first version walked the MRO for a class whose module sits beside a
config_schema.json. That passes when a test imports the plugin directly --
which is how it was verified -- and returns None under the real plugin
loader, which imports modules by a path that leaves no such entry on the
MRO.
Silent, and expensive: no schema means _schema_font_size returns None for
every element, so a configured size equal to the schema default stops
looking like a default, is treated as a deliberate choice, and skips the
grid snap. 4x6-font.ttf then renders at 6 rather than 7 -- 3px-wide glyphs
instead of 4px. On a 256x64 panel that made the odds, the records and the
date row illegible. A user counted the pixels; no gate here caught it.
"""
def test_a_declared_directory_is_used(self, tmp_path):
d = _write_plugin(tmp_path, "declared")
host = type("H", (SportsCoreSharedMixin,), {"_PLUGIN_DIR": str(d)})()
assert host._plugin_dir() == str(d)
def test_it_works_when_no_module_on_the_mro_helps(self, tmp_path, monkeypatch):
"""The production case: nothing on the MRO sits beside a schema."""
d = _write_plugin(tmp_path, "loaderstyle")
# A class whose module is not importable by name, as the loader produces.
cls = type("Loaded", (SportsCoreSharedMixin,), {"_PLUGIN_DIR": str(d)})
cls.__module__ = "a.module.name.that.is.not.in.sys.modules"
assert cls.__new__(cls)._plugin_dir() == str(d), (
"the declared directory must win when the MRO walk cannot help")
def test_without_it_the_mro_walk_would_have_failed(self):
# Pin the precondition, so this test still means something if the
# fallback is ever changed.
cls = type("Orphan", (SportsCoreSharedMixin,), {})
cls.__module__ = "not.a.real.module"
assert cls.__new__(cls)._plugin_dir() is None
def test_a_declared_directory_without_a_schema_is_not_trusted(self, tmp_path):
# A stale or wrong path must not shadow the fallback.
empty = tmp_path / "noschema"
empty.mkdir()
d = _write_plugin(tmp_path, "realone")
monkey = type("H", (SportsCoreSharedMixin,), {"_PLUGIN_DIR": str(empty)})
assert monkey.__new__(monkey)._plugin_dir() != str(empty)
def test_the_font_size_consequence(self, tmp_path):
"""End to end: a declared directory restores the schema lookup."""
d = _write_plugin(tmp_path, "sizeconseq")
host = type("H", (SportsCoreSharedMixin,), {"_PLUGIN_DIR": str(d)})()
assert host._schema_font_size("score") == 16, (
"without the schema this is None, which is what made a configured "
"size look user-chosen and skipped the grid snap")
+310
View File
@@ -0,0 +1,310 @@
"""What happens to an adopted sports plugin on each core it can meet.
B5 moved the eight scoreboards onto `src.common.sports_scroll` behind a guarded
import, keeping a bundled copy as the fallback. B6 deletes those copies. The
two phases make *different* promises, and only the second one is obvious:
| | bundled copy present | bundled copy removed |
|--------------------|---------------------------------|-----------------------------|
| **pinned old core**| loads -- B5's whole guarantee | ERROR, naming the module |
| **current core** | loads, using core code | loads, using core code |
The top-left cell is the one worth having: nothing else in this suite proves an
adopted plugin still runs on a core that predates the module, and that claim is
the entire basis for having shipped B5 ahead of B6's gate.
The bottom-left cell is the B6 failure mode, and it is asserted through
`PluginManager.load_plugin` rather than a bare import on purpose. The manager
catches the `ModuleNotFoundError`, so nothing propagates to a caller: a test
that expected `pytest.raises` would pass against a core where the module is
merely *broken* rather than absent, and would say nothing about what the user
actually experiences. What they get is a plugin parked in ERROR and one log
line -- which is precisely why B6 needs the install gate rather than trusting
the failure to be noticed.
This covers the load path. The install/update gate -- which is what should stop
a sunset plugin reaching an old core in the first place -- is the other half of
the guarantee and is tested in test_plugin_compatibility_gate.py.
See docs/SPORTS_UNIFICATION.md, "B6 -- why the sunset needs more than a version
floor".
"""
import itertools
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
project_root = Path(__file__).parent.parent
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
from src.plugin_system.plugin_manager import PluginManager
from src.plugin_system.plugin_state import PluginState
_PLUGIN_IDS = itertools.count()
CORE_MODULE = "src.common.sports_scroll"
# Only the leaf. A pre-3.2.0 core still ships `src/common/` -- scroll_helper
# and friends live there -- and it is `sports_scroll.py` alone that is absent.
# Hiding the whole package would be a different, harsher core than any that
# shipped, and it would make the load failure name the package rather than the
# module, which is the thing a reader needs to see.
HIDDEN = (CORE_MODULE,)
class _PinnedOldCore:
"""Make `src.common.sports_scroll` un-importable for the duration.
A meta_path finder rather than a monkeypatched `__import__`: the plugin is
executed by the real loader through `exec_module`, so the block has to live
in the import system itself to be reached.
"""
def __init__(self, *names):
self.names = set(names)
self._saved = {}
def find_spec(self, fullname, path=None, target=None):
if fullname in self.names:
raise ModuleNotFoundError(f"No module named {fullname!r}", name=fullname)
return None
def __enter__(self):
for name in list(sys.modules):
if name in self.names:
self._saved[name] = sys.modules.pop(name)
sys.meta_path.insert(0, self)
return self
def __exit__(self, *exc):
sys.meta_path.remove(self)
sys.modules.update(self._saved)
return False
# The two source shapes, kept as literals rather than copied from a plugin at
# runtime: this file lives in the core repo and must not depend on a plugin
# checkout being present, and pinning the shapes here means a plugin that
# drifts away from one is a visible edit, not a silently weakened test.
# B5, as the eight scoreboards ship today: prefer the core module, fall back to
# the bundled copy. The except clause is narrow on purpose -- a bare
# `except ImportError` would also swallow a failure raised *inside* a core
# module that is present, quietly loading the legacy copy and hiding a broken
# core install.
ADOPTED_WITH_FALLBACK = '''
_USING_CORE_SCROLL = False
try:
from src.common.sports_scroll import SportsScrollDisplay as _Base
_USING_CORE_SCROLL = True
except ModuleNotFoundError as exc:
if exc.name not in {"src", "src.common", "src.common.sports_scroll"}:
raise
_Base = None
if not _USING_CORE_SCROLL:
from scroll_display_legacy import LegacyScrollDisplay as _Base
'''
# B6, once the copies are deleted: there is nothing to fall back to, so the
# guard goes with them. Keeping the try/except while removing the file it
# falls back to would only mislabel the failure -- the plugin would report a
# missing `scroll_display_legacy` and say nothing about the core module that
# is actually absent.
SUNSET_NO_FALLBACK = '''
from src.common.sports_scroll import SportsScrollDisplay as _Base
_USING_CORE_SCROLL = True
'''
PLUGIN_BODY = '''
from src.plugin_system.base_plugin import BasePlugin
class SunsetProbe(BasePlugin):
"""Records which scroll implementation the guarded import selected."""
using_core_scroll = _USING_CORE_SCROLL
scroll_base = _Base
def update(self):
pass
def display(self, force_clear=False):
pass
'''
LEGACY_COPY = '''
class LegacyScrollDisplay:
"""Stands in for the bundled pre-3.2.0 implementation."""
'''
def _write_plugin(plugins_dir: Path, plugin_id: str, *, bundled_copy: bool) -> Path:
path = plugins_dir / plugin_id
path.mkdir(parents=True)
(path / "manifest.json").write_text(
json.dumps({
"id": plugin_id,
"name": "Sunset Probe",
"version": "1.0.0",
"entry_point": "manager.py",
"class_name": "SunsetProbe",
"display_modes": ["sunset_probe"],
}),
encoding="utf-8",
)
shape = ADOPTED_WITH_FALLBACK if bundled_copy else SUNSET_NO_FALLBACK
(path / "manager.py").write_text(shape + PLUGIN_BODY, encoding="utf-8")
if bundled_copy:
(path / "scroll_display_legacy.py").write_text(LEGACY_COPY, encoding="utf-8")
return path
@pytest.fixture
def load(tmp_path):
"""Load a synthetic adopted plugin through the real PluginManager.
Returns a callable taking the two axes of the matrix and handing back the
manager, so the caller can ask it for state and recorded error.
"""
plugins_dir = tmp_path / "plugin-repos"
plugins_dir.mkdir()
def _load(*, core_has_module: bool, bundled_copy: bool, hide=HIDDEN):
# A distinct id per cell, from a counter that spans the whole session.
# The loader names plugin modules after the plugin id and sys.modules
# is process-global, so a per-test counter would hand the second test
# the first test's already-imported module -- which passes or fails on
# the wrong plugin's import.
plugin_id = f"sunset-probe-{next(_PLUGIN_IDS)}"
_write_plugin(plugins_dir, plugin_id, bundled_copy=bundled_copy)
with patch('src.common.permission_utils.ensure_directory_permissions'):
manager = PluginManager(
plugins_dir=str(plugins_dir),
config_manager=MagicMock(),
display_manager=MagicMock(),
cache_manager=MagicMock(),
font_manager=MagicMock(),
)
manager.discover_plugins()
if core_has_module:
ok = manager.load_plugin(plugin_id)
else:
with _PinnedOldCore(*hide):
ok = manager.load_plugin(plugin_id)
return manager, plugin_id, ok
return _load
def _assert_loaded(manager, plugin_id, ok):
assert ok is True, (
f"load_plugin returned False; state is "
f"{manager.state_manager.get_state(plugin_id)}, error "
f"{manager.state_manager.get_error_info(plugin_id)}"
)
assert manager.state_manager.get_state(plugin_id) is not PluginState.ERROR
class TestBundledCopyPresent:
"""B5's shape: the guarded import with the fallback still shipped."""
def test_old_core_falls_back_and_still_loads(self, load):
"""The claim that made it safe to ship B5 before B6's gate."""
manager, plugin_id, ok = load(core_has_module=False, bundled_copy=True)
_assert_loaded(manager, plugin_id, ok)
plugin = manager.plugins[plugin_id]
assert plugin.using_core_scroll is False, (
"the core module was hidden, so the plugin must be on its bundled copy"
)
assert plugin.scroll_base.__name__ == "LegacyScrollDisplay"
def test_current_core_prefers_the_core_module(self, load):
manager, plugin_id, ok = load(core_has_module=True, bundled_copy=True)
_assert_loaded(manager, plugin_id, ok)
plugin = manager.plugins[plugin_id]
assert plugin.using_core_scroll is True, (
"the bundled copy must not win while the core module is importable"
)
assert plugin.scroll_base.__name__ == "SportsScrollDisplay"
def test_a_core_without_the_package_at_all_still_falls_back(self, load):
"""The guard's other accepted names.
Its except clause accepts `src` and `src.common` as well as the module
itself, so those branches exist in all eight shipped plugins. No core
that old is likely still running, but the code claiming to handle it is
real and nothing else exercises it -- an untested branch in a fallback
is exactly the kind that rots unnoticed until the fallback is needed.
"""
manager, plugin_id, ok = load(
core_has_module=False, bundled_copy=True, hide=(CORE_MODULE, "src.common")
)
_assert_loaded(manager, plugin_id, ok)
assert manager.plugins[plugin_id].using_core_scroll is False
class TestBundledCopyRemoved:
"""B6's shape: the copies are gone and only the core module remains."""
def test_current_core_still_loads(self, load):
manager, plugin_id, ok = load(core_has_module=True, bundled_copy=False)
_assert_loaded(manager, plugin_id, ok)
assert manager.plugins[plugin_id].using_core_scroll is True
def test_old_core_errors_and_records_the_missing_module(self, load):
"""The B6 failure mode, as the user meets it.
Not `pytest.raises`: load_plugin catches it, so nothing reaches a
caller. The observable consequences are the ERROR state and the
recorded error -- and the error has to name the module, or whoever
reads the log cannot tell a missing core module from any other
import failure.
"""
manager, plugin_id, ok = load(core_has_module=False, bundled_copy=False)
assert ok is False, "a plugin with no scroll implementation must not load"
assert manager.state_manager.get_state(plugin_id) is PluginState.ERROR
assert plugin_id not in manager.plugins, (
"a plugin that failed to load must not be left registered"
)
info = manager.state_manager.get_error_info(plugin_id)
assert info is not None, "ERROR state recorded no error to explain it"
assert info['error_type'] == 'ModuleNotFoundError', info
assert CORE_MODULE in info['error'], (
f"the recorded error must name the module that was missing, got {info['error']!r}"
)
def test_the_matrix_has_one_failing_cell(load):
"""Guards the shape of the table itself.
Each cell above is asserted on its own, so a change that broke two of them
in compensating ways could leave every individual test passing. This says
the outcome depends on both axes and fails in exactly one combination.
"""
outcomes = {
(core, bundled): load(core_has_module=core, bundled_copy=bundled)[2]
for core in (True, False)
for bundled in (True, False)
}
assert outcomes == {
(True, True): True,
(True, False): True,
(False, True): True,
(False, False): False,
}, outcomes
@@ -0,0 +1,92 @@
"""/api/v3/system/status must report MemAvailable, not just used/total.
"Memory used %" cannot tell a healthy board from one about to fail. Page cache
counts as used and is reclaimable on demand, so a Pi can read 70% used and be
perfectly fine, or read the same and be minutes from trouble. MemAvailable is
the kernel's own estimate of what a new allocation can actually obtain, and it
is the number that tracked the failure on a 1GB Pi 3B+: healthy running sat at
500MB+, the crash happened at 73MB, and by then fork() was failing -- sshd
could not spawn a session and systemd could not respawn the display, while the
kernel carried on answering pings.
psutil.virtual_memory().available is MemAvailable on Linux. total - used is not
a substitute: they diverge exactly when unreclaimable memory (shmem, tmpfs) is
in play, which is when the distinction matters.
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
sys.path.insert(0, str(Path(__file__).parent.parent))
MB = 1024 * 1024
@pytest.fixture
def client():
pytest.importorskip("psutil")
app = Flask(__name__)
app.config["TESTING"] = True
from web_interface.blueprints.api_v3 import api_v3
for attr in ("config_manager", "plugin_manager", "cache_manager"):
setattr(api_v3, attr, MagicMock())
if "api_v3" not in app.blueprints:
app.register_blueprint(api_v3, url_prefix="/api/v3")
return app.test_client()
def _memory(total_mb, used_mb, available_mb):
m = MagicMock()
m.total = total_mb * MB
m.used = used_mb * MB
m.available = available_mb * MB
m.percent = round(used_mb / total_mb * 100, 1)
return m
def _get_status(client, memory):
# The endpoint caches for 10s; bypass so each case is measured fresh.
with patch("web_interface.cache.get_cached", return_value=None), \
patch("psutil.virtual_memory", return_value=memory), \
patch("psutil.cpu_percent", return_value=5.0), \
patch("psutil.boot_time", return_value=0.0):
resp = client.get("/api/v3/system/status")
assert resp.status_code == 200, resp.data
return json.loads(resp.data)["data"]
def test_available_memory_is_reported(client):
data = _get_status(client, _memory(total_mb=905, used_mb=620, available_mb=284))
assert "memory_available_mb" in data
assert data["memory_available_mb"] == pytest.approx(284, abs=0.5)
def test_available_is_not_total_minus_used(client):
# The case the readout exists for: 600MB is "not used", but only 300MB can
# actually be allocated. Reporting used% alone would call this healthy.
data = _get_status(client, _memory(total_mb=1000, used_mb=400, available_mb=300))
derived = data["memory_total_mb"] - data["memory_used_mb"]
assert derived == pytest.approx(600, abs=1)
assert data["memory_available_mb"] == pytest.approx(300, abs=0.5)
assert data["memory_available_mb"] != pytest.approx(derived, abs=1), \
"available must come from MemAvailable, not be derived from used"
def test_existing_memory_fields_are_unchanged(client):
data = _get_status(client, _memory(total_mb=905, used_mb=620, available_mb=284))
assert data["memory_total_mb"] == pytest.approx(905, abs=0.5)
assert data["memory_used_mb"] == pytest.approx(620, abs=0.5)
assert "memory_used_percent" in data
def test_a_nearly_exhausted_board_reports_a_small_number(client):
# 73MB available is what the board actually read when it stopped being able
# to fork. The readout has to surface that rather than round it away.
data = _get_status(client, _memory(total_mb=905, used_mb=800, available_mb=73))
assert data["memory_available_mb"] == pytest.approx(73, abs=0.5)
+99
View File
@@ -0,0 +1,99 @@
"""Frame pacing and FPS health reporting must not depend on the wall clock.
These devices have no RTC, so the system clock jumps by however wrong boot
time was the moment NTP first syncs. The render loop sleeps the *remainder*
of each frame budget:
frame_elapsed = <now> - frame_started
time.sleep(max(0.0, frame_interval - frame_elapsed))
With a wall-clock `now`, a backward jump makes frame_elapsed negative, so
`frame_interval - frame_elapsed` exceeds the whole budget and the render loop
stalls for the size of the correction. A forward jump instead inflates the
p99 and worst-frame numbers the telemetry reports.
"""
import ast
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
COORD = (Path(__file__).resolve().parent.parent
/ "src" / "vegas_mode" / "coordinator.py")
TREE = ast.parse(COORD.read_text(encoding="utf-8"))
def _assignments_of(name):
"""Every `name = <expr>` in the module, as unparsed source."""
out = []
for node in ast.walk(TREE):
if isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name) and target.id == name:
out.append((node.lineno, ast.unparse(node.value)))
return out
def test_per_frame_timestamps_are_monotonic():
for name in ("frame_started", "frame_elapsed"):
assigns = _assignments_of(name)
assert assigns, f"{name} is no longer assigned -- has the loop changed?"
for lineno, expr in assigns:
assert "time.time()" not in expr, (
f"{name} at line {lineno} uses the wall clock ({expr!r}). A "
"backward NTP step makes the per-frame delta negative and the "
"loop then sleeps longer than the whole frame budget.")
assert "time.monotonic()" in expr, (
f"{name} at line {lineno} is {expr!r}, expected monotonic")
def test_the_fps_window_is_monotonic():
for lineno, expr in _assignments_of("current_time"):
assert "time.monotonic()" in expr, (
f"current_time at line {lineno} is {expr!r}; fps is frames divided "
"by this delta, so a clock step would corrupt the rate itself")
def test_health_state_is_not_reset_every_iteration():
"""run_iteration() runs once per cycle -- locals here reset every few seconds.
As locals, `last_fps_health_log = 0.0` made the 300s heartbeat fire on the
first sample of every iteration, and a recovery spanning two iterations was
never reported because was_degraded had already gone back to False.
"""
run_iteration = next(
(n for n in ast.walk(TREE)
if isinstance(n, ast.FunctionDef) and n.name == "run_iteration"), None)
assert run_iteration is not None, "run_iteration() not found"
local_names = {t.id for n in ast.walk(run_iteration)
if isinstance(n, ast.Assign)
for t in n.targets if isinstance(t, ast.Name)}
for leaked in ("last_fps_health_log", "was_degraded"):
assert leaked not in local_names, (
f"{leaked} is a local of run_iteration() again, so it resets every "
"cycle -- the heartbeat degenerates to once per iteration")
body = ast.unparse(run_iteration)
assert "self._fps_last_health_log" in body and "self._fps_was_degraded" in body, (
"the health state should live on the coordinator, across iterations")
def test_start_clears_stale_health_state():
"""A new run must not inherit "was degraded" from the previous one."""
start = next((n for n in ast.walk(TREE)
if isinstance(n, ast.FunctionDef) and n.name == "start"), None)
assert start is not None, "start() not found"
body = ast.unparse(start)
assert "self._fps_last_health_log" in body and "self._fps_was_degraded" in body, (
"start() does not reset the FPS health state")
def test_the_degraded_threshold_is_documented():
"""The 90% band is deliberate; say so where the constant is defined."""
source = COORD.read_text(encoding="utf-8")
idx = source.index("_FPS_HEALTHY_FRACTION = ")
preamble = source[max(0, idx - 700):idx]
assert "90%" in preamble or "0.9" in preamble, (
"the degradation threshold is not explained at its definition, so "
"'below target' reads as a bug rather than a deliberate band")
@@ -229,6 +229,44 @@ class TestSavePluginConfig:
"REAL-KEY-0123456789", "an unrelated edit destroyed the API key"
assert env.fresh_load()[PLUGIN_ID]["city"] == "Dallas"
def test_an_unrelated_edit_does_not_erase_array_item_secrets(self, env):
"""The scalar api_key case above, but for a list of credentials.
remove_empty_secrets recursed into dicts only, so a list went into
deep_merge untouched -- and lists merge by *replacement*. Saving any
unrelated field posted [{"token": ""}, ...] straight over the stored
array and destroyed every token in it at once.
"""
assert self._save(env, {"accounts": [
{"name": "a", "token": "REAL-A"},
{"name": "b", "token": "REAL-B"},
], "city": "Austin"}).status_code == 200
# the user changes the city; both masked tokens ride along blank
assert self._save(env, {"accounts": [
{"name": "a", "token": ""},
{"name": "b", "token": ""},
], "city": "Dallas"}).status_code == 200
merged = env.fresh_load()[PLUGIN_ID]
assert [a.get("token") for a in merged["accounts"]] == \
["REAL-A", "REAL-B"], "an unrelated edit destroyed the array secrets"
assert [a["name"] for a in merged["accounts"]] == ["a", "b"]
assert merged["city"] == "Dallas"
def test_one_array_secret_can_be_changed_without_losing_the_rest(self, env):
assert self._save(env, {"accounts": [
{"name": "a", "token": "REAL-A"},
{"name": "b", "token": "REAL-B"},
]}).status_code == 200
assert self._save(env, {"accounts": [
{"name": "a", "token": ""},
{"name": "b", "token": "NEW-B"},
]}).status_code == 200
merged = env.fresh_load()[PLUGIN_ID]
assert [a.get("token") for a in merged["accounts"]] == ["REAL-A", "NEW-B"]
def test_a_secret_can_still_be_changed(self, env):
"""Dropping blanks must not stop a real new value from being saved."""
self._save(env, {"api_key": "first-key"})
@@ -0,0 +1,45 @@
"""The validation logging ran before separate_secrets, so it logged credentials.
api_v3's plugin-config save logged `Full config: {plugin_config}` at INFO and
`Config that failed: {plugin_config}` at ERROR. Both run *before*
separate_secrets(), so plugin_config still held the values the user just typed
into the form -- API keys and tokens went to the journal in clear text.
"""
import re
from pathlib import Path
import pytest
SOURCE = (Path(__file__).resolve().parents[2]
/ "web_interface" / "blueprints" / "api_v3.py")
#: Objects that still hold submitted secret values at the point these log
#: calls run. Interpolating one whole into a log message leaks credentials.
UNREDACTED = ("plugin_config", "secrets_config", "current_secrets")
def _logging_lines():
for number, line in enumerate(SOURCE.read_text(encoding="utf-8").splitlines(), 1):
stripped = line.strip()
if stripped.startswith("#"):
continue
if re.match(r"logger\.(debug|info|warning|error|critical|exception)\(", stripped):
yield number, stripped
@pytest.mark.parametrize("name", UNREDACTED)
def test_no_log_call_interpolates_a_whole_secret_bearing_object(name):
# {name} or {name['k']} leaks; {list(name.keys())} and {len(name)} do not.
bare = re.compile(r"\{" + re.escape(name) + r"(\[[^\]]*\])*\}")
offenders = [f"{n}: {text}" for n, text in _logging_lines() if bare.search(text)]
assert not offenders, (
f"{name} still holds submitted secrets where these log calls run:\n "
+ "\n ".join(offenders))
def test_the_guard_would_notice_a_reintroduced_leak():
"""Pin the detector itself, so a rewrite cannot silently stop matching."""
bare = re.compile(r"\{" + re.escape("plugin_config") + r"(\[[^\]]*\])*\}")
assert bare.search('logger.info(f"Full config: {plugin_config}")')
assert bare.search("logger.error(f\"{plugin_config['api_key']}\")")
assert not bare.search('logger.info(f"{list(plugin_config.keys())}")')
+65
View File
@@ -17,6 +17,7 @@ from src.web_interface.secret_helpers import (
separate_secrets,
mask_secret_fields,
mask_all_secret_values,
merge_secrets,
remove_empty_secrets,
)
@@ -239,3 +240,67 @@ class TestRemoveEmptySecrets:
def test_keeps_falsey_non_string_values(self):
# 0 and False are neither None nor blank strings — they are kept.
assert remove_empty_secrets({"a": 0, "b": False}) == {"a": 0, "b": False}
class TestArrayItemSecrets:
"""Lists merge by replacement, so a blanked array wipes stored credentials.
remove_empty_secrets recursed into dicts but let a list through untouched,
so [{"token": ""}] went straight into deep_merge and overwrote the stored
list. Saving any unrelated setting destroyed every token in the array.
"""
STORED = {"accounts": [{"name": "a", "token": "REAL-A"},
{"name": "b", "token": "REAL-B"}]}
def test_an_unrelated_save_keeps_every_stored_token(self):
posted = {"accounts": [{"name": "a", "token": ""},
{"name": "b", "token": ""}]}
merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
assert [a["token"] for a in merged["accounts"]] == ["REAL-A", "REAL-B"]
def test_editing_one_entry_leaves_the_others_alone(self):
posted = {"accounts": [{"name": "a", "token": ""},
{"name": "b", "token": "NEW-B"}]}
merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
assert [a["token"] for a in merged["accounts"]] == ["REAL-A", "NEW-B"]
def test_a_new_entry_is_appended(self):
posted = {"accounts": [{"name": "a", "token": ""},
{"name": "b", "token": ""},
{"name": "c", "token": "NEW-C"}]}
merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
assert [a["token"] for a in merged["accounts"]] == \
["REAL-A", "REAL-B", "NEW-C"]
def test_a_list_of_bare_strings_merges_by_index(self):
merged = merge_secrets({"keys": ["K1", "K2", "K3"]},
remove_empty_secrets({"keys": ["", "K2-NEW", ""]}))
assert merged["keys"] == ["K1", "K2-NEW", "K3"]
def test_an_all_blank_list_is_dropped_entirely(self):
posted = {"accounts": [{"token": ""}, {"token": ""}]}
assert "accounts" not in remove_empty_secrets(posted)
def test_plain_dict_secrets_are_unaffected(self):
merged = merge_secrets({"api_key": "OLD", "other": "keep"},
remove_empty_secrets({"api_key": "", "other": "changed"}))
assert merged == {"api_key": "OLD", "other": "changed"}
def test_a_removed_entry_takes_its_secret_with_it(self):
"""The regular config's list is authoritative about how many items
exist, and the secrets list runs parallel to it -- see
ConfigManager._strip_secrets_recursive. So a shorter incoming list
must shorten the stored secrets too, or the two fall out of step."""
posted = {"accounts": [{"name": "a", "token": "NEW-A"}]}
merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
assert [a["token"] for a in merged["accounts"]] == ["NEW-A"]
def test_an_emptied_item_stays_a_dict_not_none(self):
"""None there stops the list looking parallel, and
_strip_secrets_recursive then drops the whole key from the main
config -- deleting the item's non-secret fields as well."""
pruned = remove_empty_secrets(
{"accounts": [{"token": "real"}, {"token": ""}]})
assert pruned["accounts"] == [{"token": "real"}, {}]
assert None not in pruned["accounts"]