mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-18 04:49:13 +00:00
Compare commits
3
Commits
bdced206dc
...
eae063700f
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eae063700f | ||
|
|
af96bd5cb0 | ||
|
|
5e5979973e |
@@ -161,9 +161,11 @@ The system supports live, recent, and upcoming game information for multiple spo
|
||||
### LED Matrix Panels
|
||||
(2x in a horizontal chain is recommended)
|
||||
- [Adafruit 64×32](https://www.adafruit.com/product/2278) – designed for 128×32 but works with dynamic scaling on many displays (pixel pitch is user preference)
|
||||
**Warning: Lately the Waveshare Panels have had different variations - only some are compatible with this project. I hope to identify what is different to fix it but so far there is a decent chance you get a mis-matched set of panels if you don't buy them all at once! **
|
||||
- [Waveshare 64×32](https://amzn.to/3Kw55jK) - Does not require E addressable pad
|
||||
- [Waveshare 96×48](https://amzn.to/4bydNcv) – higher resolution, requires soldering the **E addressable pad** on the [Adafruit RGB Bonnet](https://www.adafruit.com/product/3211) to “8” **OR** toggling the DIP switch on the Adafruit Triple LED Matrix Bonnet *(no soldering required!)*
|
||||
> Amazon Affiliate Link – ChuckBuilds receives a small commission on purchases
|
||||
- [Waveshare 96×48](https://amzn.to/4bydNcv) – higher resolution, requires soldering the **E addressable pad** on the [Adafruit RGB Bonnet](https://www.adafruit.com/product/3211) to “8” **OR** toggling the DIP switch on the Adafruit Triple LED Matrix Bonnet *(no soldering required!)*
|
||||
- There are some Panels on Aliexpress that have worked fine for me, shop around! I think Adafruit is probably the "safest" but they do have some limitation on resolution and layout.
|
||||
> Amazon Affiliate Links – ChuckBuilds receives a small commission on purchases
|
||||
|
||||
### Power Supply
|
||||
- [5V 4A DC Power Supply](https://www.adafruit.com/product/658) (good for 2 -3 displays, depending on brightness and pixel density, you'll need higher amperage for more)
|
||||
|
||||
@@ -14,11 +14,12 @@ Key Features:
|
||||
- Memory-efficient data storage
|
||||
"""
|
||||
|
||||
import itertools
|
||||
import time
|
||||
import logging
|
||||
import threading
|
||||
import requests
|
||||
from typing import Dict, Any, Optional, Callable
|
||||
from typing import Dict, Any, Optional, Callable, List
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
import queue
|
||||
@@ -50,8 +51,19 @@ class FetchRequest:
|
||||
max_retries: int = 3
|
||||
priority: int = 1 # Higher number = higher priority
|
||||
callback: Optional[Callable] = None
|
||||
# Callbacks from submitters that JOINED this fetch instead of starting a
|
||||
# duplicate one. The primary `callback` above belongs to whoever created
|
||||
# the request; these belong to everyone who asked for the same cache_key
|
||||
# while it was still in flight.
|
||||
extra_callbacks: List[Callable] = field(default_factory=list)
|
||||
created_at: float = field(default_factory=time.time)
|
||||
status: FetchStatus = FetchStatus.PENDING
|
||||
# Set once the worker has decided this response will be cached, while it
|
||||
# still holds the lock. From that point cancelling is refused: the write
|
||||
# is already authorised, and abandoning it here would put the payload in
|
||||
# the cache with the callbacks suppressed -- joiners waiting forever for a
|
||||
# fetch that did, in fact, succeed.
|
||||
commit_claimed: bool = False
|
||||
result: Optional[Any] = None
|
||||
error: Optional[str] = None
|
||||
|
||||
@@ -74,6 +86,10 @@ class FetchResult:
|
||||
fetch_time: float = 0.0
|
||||
retry_count: int = 0
|
||||
completed_at: float = field(default_factory=time.time) # Timestamp when request completed
|
||||
# The request's final status, recorded so a finished request can still be
|
||||
# reported accurately. Without it a caller can only be told COMPLETED or
|
||||
# FAILED, which turns "you cancelled this" into "this errored".
|
||||
final_status: Optional[FetchStatus] = None
|
||||
|
||||
class BackgroundDataService:
|
||||
"""
|
||||
@@ -98,6 +114,20 @@ class BackgroundDataService:
|
||||
|
||||
# Thread management
|
||||
self.executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="BackgroundData")
|
||||
# cache_key -> request_id for fetches currently in flight. Submitting
|
||||
# the same key twice used to start two identical fetches: request_id
|
||||
# carries a millisecond timestamp, so every submit looked new, and
|
||||
# active_requests is keyed by it rather than by what is being fetched.
|
||||
# On a real board the season-schedule key is requested by both the
|
||||
# Recent and the Upcoming manager, which miss the cache in the same
|
||||
# millisecond and each download and parse the same payload.
|
||||
self._inflight_by_cache_key: Dict[str, str] = {}
|
||||
# request_id was sport_year_milliseconds, which is not unique: two
|
||||
# submits inside the same millisecond produced the SAME id, so one
|
||||
# silently replaced the other in active_requests and completed_requests.
|
||||
# Rare before, but dedupe hands this id back to every joiner as their
|
||||
# handle for get_result(), so it has to be unique. A counter is enough.
|
||||
self._request_seq = itertools.count()
|
||||
self.active_requests: Dict[str, FetchRequest] = {}
|
||||
self.completed_requests: Dict[str, FetchResult] = {}
|
||||
self.request_queue = queue.PriorityQueue()
|
||||
@@ -185,7 +215,9 @@ class BackgroundDataService:
|
||||
if cache_key is None:
|
||||
cache_key = self.get_sport_cache_key(sport)
|
||||
|
||||
request_id = f"{sport}_{year}_{int(time.time() * 1000)}"
|
||||
with self._lock:
|
||||
request_id = (f"{sport}_{year}_{int(time.time() * 1000)}"
|
||||
f"_{next(self._request_seq)}")
|
||||
|
||||
# Check cache first
|
||||
cached_data = self.cache_manager.get(cache_key)
|
||||
@@ -231,7 +263,29 @@ class BackgroundDataService:
|
||||
)
|
||||
|
||||
with self._lock:
|
||||
existing_id = self._inflight_by_cache_key.get(cache_key)
|
||||
existing = self.active_requests.get(existing_id) if existing_id else None
|
||||
if existing_id and existing is None:
|
||||
# Stranded index entry: the request it names is gone. Drop it and
|
||||
# fetch normally. Looking the request up rather than trusting the
|
||||
# id is what stops a stale entry wedging a key forever.
|
||||
del self._inflight_by_cache_key[cache_key]
|
||||
if existing is not None:
|
||||
# Someone is already fetching this key. Ride along rather than
|
||||
# duplicating the download, the parse and the resident copy.
|
||||
if callback:
|
||||
existing.extra_callbacks.append(callback)
|
||||
self.stats['deduplicated_requests'] = (
|
||||
self.stats.get('deduplicated_requests', 0) + 1
|
||||
)
|
||||
logger.info(
|
||||
"Joined in-flight fetch %s for %s (cache_key=%s) instead of "
|
||||
"starting a duplicate", existing_id, sport, cache_key
|
||||
)
|
||||
return existing_id
|
||||
|
||||
self.active_requests[request_id] = request
|
||||
self._inflight_by_cache_key[cache_key] = request_id
|
||||
self.stats['total_requests'] += 1
|
||||
self.stats['cache_misses'] += 1
|
||||
|
||||
@@ -256,7 +310,31 @@ class BackgroundDataService:
|
||||
|
||||
try:
|
||||
with self._lock:
|
||||
request.status = FetchStatus.IN_PROGRESS
|
||||
# A request cancelled while it sat in the executor queue must
|
||||
# stay cancelled. Overwriting the status here undid the cancel
|
||||
# outright: the worker went on to download, cache and call back
|
||||
# for work the caller had already withdrawn.
|
||||
if request.status == FetchStatus.CANCELLED:
|
||||
cancelled_before_start = True
|
||||
else:
|
||||
cancelled_before_start = False
|
||||
request.status = FetchStatus.IN_PROGRESS
|
||||
if cancelled_before_start:
|
||||
logger.info(
|
||||
"Request %s was cancelled before its worker started; "
|
||||
"skipping the fetch entirely", request.id
|
||||
)
|
||||
# Assign before returning: the finally block stores `result`
|
||||
# in completed_requests, so building a fresh one here would
|
||||
# file the untouched placeholder instead of this outcome.
|
||||
result = FetchResult(
|
||||
request_id=request.id,
|
||||
success=False,
|
||||
error="cancelled",
|
||||
fetch_time=time.time() - start_time,
|
||||
retry_count=request.retry_count
|
||||
)
|
||||
return result
|
||||
|
||||
logger.info(f"Starting background fetch for {request.sport} {request.year}")
|
||||
|
||||
@@ -282,6 +360,37 @@ class BackgroundDataService:
|
||||
# Log data validation
|
||||
logger.debug(f"Validated {len(events)} events for {request.sport} {request.year}")
|
||||
|
||||
# A cancelled request must not commit anything. Cancelling
|
||||
# releases the cache_key, so a replacement fetch for the same key
|
||||
# may already be in flight or finished -- writing this response to
|
||||
# the cache now would overwrite fresher data with the response
|
||||
# nobody wanted. The worker has no way to abort the HTTP call, so
|
||||
# this is where the work gets discarded.
|
||||
with self._lock:
|
||||
cancelled = request.status == FetchStatus.CANCELLED
|
||||
if not cancelled:
|
||||
# Claim the commit in the same critical section that read
|
||||
# the status, so a cancel cannot slip in between the check
|
||||
# and the cache write below. The write itself stays outside
|
||||
# the lock: it serialises a multi-megabyte payload to the
|
||||
# SD card, and holding the service lock across that would
|
||||
# stall every submit, status query and cancel behind it.
|
||||
request.commit_claimed = True
|
||||
if cancelled:
|
||||
logger.info(
|
||||
"Discarding response for cancelled request %s; %s may "
|
||||
"already belong to a replacement fetch",
|
||||
request.id, request.cache_key
|
||||
)
|
||||
result = FetchResult(
|
||||
request_id=request.id,
|
||||
success=False,
|
||||
error="cancelled",
|
||||
fetch_time=time.time() - start_time,
|
||||
retry_count=request.retry_count
|
||||
)
|
||||
return result
|
||||
|
||||
# Cache the data
|
||||
self.cache_manager.set(request.cache_key, data)
|
||||
|
||||
@@ -307,7 +416,12 @@ class BackgroundDataService:
|
||||
logger.error(f"Failed to fetch {request.sport} {request.year} data: {error_msg}")
|
||||
|
||||
with self._lock:
|
||||
request.status = FetchStatus.FAILED
|
||||
# Don't relabel a cancelled request. The callback gate in the
|
||||
# finally block only suppresses CANCELLED, so promoting it to
|
||||
# FAILED here delivered an error callback for a fetch nobody
|
||||
# was waiting on any more.
|
||||
if request.status != FetchStatus.CANCELLED:
|
||||
request.status = FetchStatus.FAILED
|
||||
request.error = error_msg
|
||||
|
||||
result = FetchResult(
|
||||
@@ -321,9 +435,26 @@ class BackgroundDataService:
|
||||
finally:
|
||||
# Store result and clean up
|
||||
with self._lock:
|
||||
result.final_status = request.status
|
||||
self.completed_requests[request.id] = result
|
||||
if request.id in self.active_requests:
|
||||
del self.active_requests[request.id]
|
||||
# Stop accepting joiners and take the callback list in the same
|
||||
# critical section. A submitter that arrives after this point
|
||||
# finds no in-flight entry and either hits the cache (written
|
||||
# above, before the result was built) or starts a fresh fetch --
|
||||
# what it must never do is join a fetch whose callbacks have
|
||||
# already run and then never be called.
|
||||
if self._inflight_by_cache_key.get(request.cache_key) == request.id:
|
||||
del self._inflight_by_cache_key[request.cache_key]
|
||||
# A cancelled request delivers nothing: its joiners were told
|
||||
# about a fetch that has been abandoned, and a replacement will
|
||||
# call them via its own request.
|
||||
if request.status == FetchStatus.CANCELLED:
|
||||
callbacks = []
|
||||
else:
|
||||
callbacks = ([request.callback] if request.callback else [])
|
||||
callbacks.extend(request.extra_callbacks)
|
||||
|
||||
# Update statistics
|
||||
if result.success:
|
||||
@@ -340,10 +471,11 @@ class BackgroundDataService:
|
||||
# Periodic cleanup after storing result
|
||||
self._cleanup_completed_requests()
|
||||
|
||||
# Call callback if provided
|
||||
if request.callback:
|
||||
# Call every callback: the original submitter's and any that joined
|
||||
# this fetch. One raising must not stop the others being delivered.
|
||||
for cb in callbacks:
|
||||
try:
|
||||
request.callback(result)
|
||||
cb(result)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in callback for request {request.id}: {e}")
|
||||
# Delivered. Drop both references -- they point at the same
|
||||
@@ -456,6 +588,8 @@ class BackgroundDataService:
|
||||
return self.active_requests[request_id].status
|
||||
elif request_id in self.completed_requests:
|
||||
result = self.completed_requests[request_id]
|
||||
if result.final_status is not None:
|
||||
return result.final_status
|
||||
return FetchStatus.COMPLETED if result.success else FetchStatus.FAILED
|
||||
return None
|
||||
|
||||
@@ -472,8 +606,22 @@ class BackgroundDataService:
|
||||
with self._lock:
|
||||
if request_id in self.active_requests:
|
||||
request = self.active_requests[request_id]
|
||||
if request.commit_claimed:
|
||||
# Too late: the worker holds an authorised commit. Report
|
||||
# the failure rather than half-cancelling a request whose
|
||||
# data is about to land in the cache.
|
||||
logger.debug(
|
||||
"Not cancelling %s: its response is already being "
|
||||
"committed", request_id
|
||||
)
|
||||
return False
|
||||
request.status = FetchStatus.CANCELLED
|
||||
del self.active_requests[request_id]
|
||||
# Cancelling is the other way a request leaves active_requests,
|
||||
# so the in-flight index has to be released here too or the key
|
||||
# stays pointed at a request that no longer exists.
|
||||
if self._inflight_by_cache_key.get(request.cache_key) == request_id:
|
||||
del self._inflight_by_cache_key[request.cache_key]
|
||||
logger.info(f"Cancelled request {request_id}")
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user