mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-16 03:49:13 +00:00
Merge origin/main into chore/remove-ipc-mailboxes
Conflicts were additive only: CHANGELOG.md (this PR's Unreleased section now sits above main's 3.8.x release notes) and test/js/run_all.js (both new unit suites kept). display.py and api_v3/__init__.py merged cleanly with the stack-trace-exposure fix (#778). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
+2
-1
@@ -33,7 +33,8 @@ const UNIT = ['unit/test_list_filter.js', 'unit/test_render_cards.js',
|
||||
'unit/test_overview_reconciliation_poll.js',
|
||||
'unit/test_display_partial_ids.js',
|
||||
'unit/test_general_web_login_token.js',
|
||||
'unit/test_on_demand_starting.js'];
|
||||
'unit/test_on_demand_starting.js',
|
||||
'unit/test_general_display_autostart.js'];
|
||||
const DOM = ['dom/test_installed_dom.js', 'dom/test_store_dom.js', 'dom/test_no_double_fetch.js',
|
||||
'dom/test_tools_sections.js', 'dom/test_cache_page.js',
|
||||
'dom/test_durations_page.js', 'dom/test_operation_history_page.js',
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
// The General tab's "Start Display on Boot" checkbox (setDisplayAutostart in
|
||||
// static/v3/js/pages/general.js) posts enable_autostart / disable_autostart
|
||||
// to /api/v3/system/action and puts the box back when that is refused --
|
||||
// including the route's HTTP-200 {status: "error"} answer for a failed
|
||||
// systemctl (e.g. the sudo rule is missing).
|
||||
//
|
||||
// Fake fetch, no jsdom and no server needed.
|
||||
|
||||
const path = require('path');
|
||||
const { pathToFileURL } = require('url');
|
||||
|
||||
const JS = path.resolve(__dirname, '../../../web_interface/static/v3/js');
|
||||
const load = f => import(pathToFileURL(path.join(JS, f)).href);
|
||||
|
||||
let pass = 0, fail = 0;
|
||||
const ok = (label, cond, extra) => cond
|
||||
? (pass++, console.log(' ok ' + label))
|
||||
: (fail++, console.log(' FAIL ' + label + (extra !== undefined ? ' ' + JSON.stringify(extra) : '')));
|
||||
|
||||
async function run(checked, answer) {
|
||||
const { createApi } = await load('core/api.js');
|
||||
const general = await load('pages/general.js');
|
||||
const sent = [];
|
||||
const notes = [];
|
||||
const fetch = (url, init) => {
|
||||
sent.push([url, JSON.parse(init.body)]);
|
||||
if (answer === 'network') return Promise.reject(new TypeError('Failed to fetch'));
|
||||
return Promise.resolve({
|
||||
ok: answer.ok, status: answer.ok ? 200 : 500,
|
||||
headers: { get: () => null },
|
||||
text: () => Promise.resolve(JSON.stringify(answer.body)),
|
||||
});
|
||||
};
|
||||
const ctx = { root: {}, state: {}, signal: { aborted: false },
|
||||
api: createApi({ fetch }), notify: (m, t) => notes.push([m, t]) };
|
||||
const box = { checked, disabled: false };
|
||||
await general.setDisplayAutostart(ctx, box);
|
||||
return { box, sent, notes };
|
||||
}
|
||||
|
||||
(async () => {
|
||||
console.log('\n── General tab: Start Display on Boot ──');
|
||||
|
||||
{
|
||||
const t = await run(false, { ok: true, body: { status: 'success', message: 'ok' } });
|
||||
ok('unchecking posts disable_autostart',
|
||||
t.sent.length === 1 && t.sent[0][0] === '/api/v3/system/action' && t.sent[0][1].action === 'disable_autostart', t.sent);
|
||||
ok('...stays unchecked and re-enabled', t.box.checked === false && t.box.disabled === false);
|
||||
ok('...and reports success', t.notes.some(([, type]) => type === 'success'), t.notes);
|
||||
}
|
||||
|
||||
{
|
||||
const t = await run(true, { ok: true, body: { status: 'success' } });
|
||||
ok('checking posts enable_autostart', t.sent[0][1].action === 'enable_autostart', t.sent);
|
||||
}
|
||||
|
||||
{
|
||||
const t = await run(false, { ok: true, body: { status: 'error', message: 'sudo: a password is required' } });
|
||||
ok('a failed systemctl (HTTP 200, status error) puts the box back', t.box.checked === true && !t.box.disabled);
|
||||
ok('...and shows the route\'s message',
|
||||
t.notes.some(([m, type]) => type === 'error' && /sudo/.test(m)), t.notes);
|
||||
}
|
||||
|
||||
{
|
||||
const t = await run(true, { ok: false, body: { status: 'error', message: 'boom' } });
|
||||
ok('an HTTP error puts the box back', t.box.checked === false && !t.box.disabled);
|
||||
}
|
||||
|
||||
{
|
||||
const t = await run(true, 'network');
|
||||
ok('a network failure puts the box back', t.box.checked === false && !t.box.disabled);
|
||||
ok('...and reports it', t.notes.some(([, type]) => type === 'error'), t.notes);
|
||||
}
|
||||
|
||||
console.log(`\n${pass} passed, ${fail} failed\n`);
|
||||
process.exit(fail ? 1 : 0);
|
||||
})().catch(e => { console.log('HARNESS ERROR: ' + e.stack); process.exit(1); });
|
||||
@@ -392,6 +392,15 @@ class TestGetEspnScoreboard:
|
||||
"20261004", cache_manager=cache)
|
||||
assert cache.records == {}
|
||||
|
||||
@pytest.mark.parametrize("payload", [{"events": None}, {}, {"events": "x"}, []])
|
||||
def test_an_answer_without_an_events_list_is_returned_not_cached(self, service, payload):
|
||||
espn = Espn()
|
||||
espn.get = lambda url, **kwargs: _response(payload, url, None)
|
||||
cache = RecordCache()
|
||||
assert get_espn_scoreboard(espn, "football", "nfl", "20261004",
|
||||
cache_manager=cache) == payload
|
||||
assert cache.records == {}
|
||||
|
||||
def test_no_cache_manager_just_fetches(self, service):
|
||||
espn = Espn()
|
||||
get_espn_scoreboard(espn, "football", "nfl", "20261004")
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""The General tab's "Start Display on Boot" toggle reads systemd's state.
|
||||
|
||||
ledmatrix.service enabled -> checked, disabled -> unchecked, unreadable (no
|
||||
systemctl, unit not installed) -> no toggle. The toggle itself posts the
|
||||
existing enable_autostart / disable_autostart system actions, whose sudo
|
||||
grants test_systemctl_sudoers_alignment.py already checks.
|
||||
"""
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
PROJECT_ROOT = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from test.test_web_settings_ui import client # noqa: F401,E402
|
||||
from web_interface.blueprints import pages_v3 # noqa: E402
|
||||
|
||||
|
||||
REAL_RUN = subprocess.run
|
||||
UNIT_QUERY = ['systemctl', 'is-enabled', 'ledmatrix.service']
|
||||
|
||||
|
||||
def _fake_systemctl(monkeypatch, answer):
|
||||
"""Answer the unit query with `answer` (stdout, or an exception to
|
||||
raise); every other command (the update channel's git) runs for real."""
|
||||
calls = []
|
||||
|
||||
def run(cmd, *args, **kwargs):
|
||||
if cmd != UNIT_QUERY:
|
||||
return REAL_RUN(cmd, *args, **kwargs)
|
||||
calls.append(cmd)
|
||||
if isinstance(answer, Exception):
|
||||
raise answer
|
||||
return subprocess.CompletedProcess(cmd, 0, stdout=answer, stderr='')
|
||||
monkeypatch.setattr(pages_v3.subprocess, 'run', run)
|
||||
return calls
|
||||
|
||||
|
||||
def _toggle(body):
|
||||
start = body.find('data-action="display-autostart"')
|
||||
return None if start < 0 else body[start:body.find('>', start)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('stdout, checked', [('enabled\n', True), ('disabled\n', False)])
|
||||
def test_toggle_reflects_the_unit(client, monkeypatch, stdout, checked):
|
||||
calls = _fake_systemctl(monkeypatch, stdout)
|
||||
tag = _toggle(client.get('/v3/partials/general').get_data(as_text=True))
|
||||
assert calls == [UNIT_QUERY]
|
||||
assert tag is not None
|
||||
assert ('checked' in tag) is checked
|
||||
|
||||
|
||||
@pytest.mark.parametrize('outcome', [FileNotFoundError(), 'not-found\n'])
|
||||
def test_toggle_hidden_when_state_unreadable(client, monkeypatch, outcome):
|
||||
_fake_systemctl(monkeypatch, outcome)
|
||||
body = client.get('/v3/partials/general').get_data(as_text=True)
|
||||
assert _toggle(body) is None
|
||||
assert 'name="plugins_directory"' in body # rest of the tab still renders
|
||||
@@ -0,0 +1,399 @@
|
||||
"""src.common.sports_live_fetch: the live poll's request and the odds queue.
|
||||
|
||||
The cases follow ledmatrix-plugins' ``scripts/test_live_fetch_odds.py`` (the
|
||||
tables the family 9 reconcile was checked against); the expected values are
|
||||
its columns, which agree across all nine scoreboards. The host is the mixin
|
||||
over core's ``SportsFetchMixin`` (``_needs_previous_day``), with a session
|
||||
answering like ESPN did on 2026-10-12 (a single date is 200, a range 400) and
|
||||
a cache that ages entries on a test clock. Nothing reaches the network. The
|
||||
odds endpoint is held on an event, so "update() never waits" is shown by the
|
||||
call returning while the request is still open.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from src.common import espn_dates, sports_live_fetch
|
||||
from src.common import fetch_service as fs
|
||||
from src.common.fetch_service import FetchService
|
||||
from src.common.sports_fetch import SportsFetchMixin
|
||||
from src.common.sports_live_fetch import SportsLiveFetchMixin
|
||||
|
||||
LOG = logging.getLogger("test_sports_live_fetch")
|
||||
SITE = "https://site.api.espn.com/apis/site/v2/sports/"
|
||||
|
||||
|
||||
def _response(url, answer):
|
||||
resp = requests.Response()
|
||||
resp.url = url
|
||||
resp.status_code = answer if isinstance(answer, int) else 200
|
||||
resp._content = answer if isinstance(answer, bytes) else json.dumps(
|
||||
{} if isinstance(answer, int) else answer).encode()
|
||||
return resp
|
||||
|
||||
|
||||
class Espn(requests.Session):
|
||||
"""Two games a day for a single date; a range is 400, as ESPN answers."""
|
||||
|
||||
def __init__(self, answer=None):
|
||||
super().__init__()
|
||||
self.answer, self.asked = answer, []
|
||||
|
||||
def get(self, url, params=None, **_kwargs):
|
||||
dates = (params or {}).get("dates")
|
||||
self.asked.append((url[len(SITE):], dates))
|
||||
if dates and "-" in dates:
|
||||
return _response(url, 400)
|
||||
if isinstance(self.answer, type):
|
||||
raise self.answer("refused by the test")
|
||||
if self.answer is not None:
|
||||
return _response(url, self.answer)
|
||||
return _response(url, {"events": [{"id": dates[4:] + s} for s in "ab"]})
|
||||
|
||||
|
||||
class Cache:
|
||||
"""``get(key, max_age)`` against entries stamped on ``clock``."""
|
||||
|
||||
def __init__(self, clock):
|
||||
self.clock, self.store, self.reads = clock, {}, []
|
||||
|
||||
def get(self, key, max_age=None):
|
||||
self.reads.append((key, max_age))
|
||||
if key not in self.store:
|
||||
return None
|
||||
data, stamp = self.store[key]
|
||||
return data if max_age is None or self.clock() - stamp <= max_age else None
|
||||
|
||||
def set(self, key, data, ttl=None):
|
||||
self.store[key] = (data, self.clock())
|
||||
|
||||
|
||||
class Clock:
|
||||
def __init__(self):
|
||||
self.now = 100_000.0
|
||||
|
||||
def __call__(self):
|
||||
return self.now
|
||||
|
||||
|
||||
class Frozen(datetime):
|
||||
at = datetime(2026, 10, 12, 10, 0)
|
||||
|
||||
@classmethod
|
||||
def now(cls, tz=None):
|
||||
return tz.localize(cls.at) if tz is not None else cls.at
|
||||
|
||||
|
||||
class Host(SportsLiveFetchMixin, SportsFetchMixin):
|
||||
"""A scoreboard SportsCore as far as the family reaches."""
|
||||
|
||||
sport, league, sport_key = "basketball", "nba", "nba"
|
||||
|
||||
def __init__(self, session=None, cache=None, odds=None, show_odds=True):
|
||||
self.session, self.headers, self.logger = session or Espn(), {}, LOG
|
||||
self.cache_manager = cache
|
||||
self.odds_manager, self.show_odds = odds, show_odds
|
||||
self.mode_config = {}
|
||||
self._games_lock = threading.RLock()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def offline(monkeypatch):
|
||||
clock = Clock()
|
||||
monkeypatch.setattr(fs, "_service", FetchService({"rate_limits": {}}))
|
||||
monkeypatch.setattr(espn_dates, "_ranges_rejected_until", 0.0)
|
||||
monkeypatch.setattr(sports_live_fetch, "datetime", Frozen)
|
||||
monkeypatch.setattr(time, "time", clock)
|
||||
Frozen.at = datetime(2026, 10, 12, 10, 0)
|
||||
return clock
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The live poll's request
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("hour, asked, key, ids", [
|
||||
(10, [("20261012",)], "espn_scoreboard_basketball_nba_20261012", ["1012a", "1012b"]),
|
||||
(2, [("20261011-20261012",), ("20261011",), ("20261012",)],
|
||||
"espn_scoreboard_basketball_nba_20261011-20261012",
|
||||
["1011a", "1011b", "1012a", "1012b"]),
|
||||
])
|
||||
def test_what_the_poll_asks_and_keeps(offline, hour, asked, key, ids):
|
||||
Frozen.at = datetime(2026, 10, 12, hour, 0)
|
||||
espn, cache = Espn(), Cache(offline)
|
||||
data = Host(espn, cache)._fetch_todays_games()
|
||||
assert [(d,) for _, d in espn.asked] == asked
|
||||
assert {path for path, _ in espn.asked} == {"basketball/nba/scoreboard"}
|
||||
assert cache.reads == [(key, 1)]
|
||||
assert list(cache.store) == [key]
|
||||
assert [e["id"] for e in data["events"]] == ids
|
||||
|
||||
|
||||
def test_college_basketball_is_asked_by_date(offline):
|
||||
espn, host = Espn(), Host()
|
||||
host.session, host.league = espn, "mens-college-basketball"
|
||||
host._fetch_todays_games()
|
||||
assert espn.asked == [("basketball/mens-college-basketball/scoreboard", "20261012")]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("seed_age, later, requests_made", [
|
||||
(None, 20, [1, 1]), # every poll asks ESPN; nothing is held 30 s
|
||||
(None, 0.5, [1, 0]), # a copy from this second is the same answer
|
||||
(0, None, [0]), # another reader's copy from this second
|
||||
(2, None, [1]), # ... and one two seconds old is asked again
|
||||
])
|
||||
def test_every_poll_is_fresh(offline, seed_age, later, requests_made):
|
||||
espn, cache = Espn(), Cache(offline)
|
||||
if seed_age is not None:
|
||||
cache.store["espn_scoreboard_basketball_nba_20261012"] = (
|
||||
{"events": [{"id": "seed"}]}, offline.now - seed_age)
|
||||
host, made = Host(espn, cache), []
|
||||
for step in ([0] if later is None else [0, later]):
|
||||
offline.now += step
|
||||
before = len(espn.asked)
|
||||
host._fetch_todays_games()
|
||||
made.append(len(espn.asked) - before)
|
||||
assert made == requests_made
|
||||
|
||||
|
||||
def test_the_old_per_plugin_key_is_not_read(offline):
|
||||
cache = Cache(offline)
|
||||
cache.store["nba_scoreboard_current"] = ({"events": [{"id": "old"}]}, offline.now)
|
||||
data = Host(Espn(), cache)._fetch_todays_games()
|
||||
assert [e["id"] for e in data["events"]] == ["1012a", "1012b"]
|
||||
assert "nba_scoreboard_current" not in {k for k, _ in cache.reads}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("answer, result, cached", [
|
||||
({}, {"events": []}, False),
|
||||
({"events": None}, {"events": []}, False),
|
||||
({"events": []}, {"events": []}, True),
|
||||
(404, None, False),
|
||||
(500, None, False),
|
||||
(requests.ConnectionError, None, False),
|
||||
(requests.Timeout, None, False),
|
||||
(b"<html>", None, False),
|
||||
])
|
||||
def test_what_a_failed_or_malformed_answer_becomes(offline, answer, result, cached):
|
||||
cache = Cache(offline)
|
||||
assert Host(Espn(answer), cache)._fetch_todays_games() == result
|
||||
assert bool(cache.store) is cached
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Odds
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class Odds:
|
||||
"""An odds endpoint held open until ``release()``, counting what is in
|
||||
flight and how many odds workers are alive while it answers."""
|
||||
|
||||
def __init__(self, answer="odds", held=True):
|
||||
self.answer, self.calls, self.gate = answer, [], threading.Event()
|
||||
if not held:
|
||||
self.gate.set()
|
||||
self.lock, self.active, self.most_active, self.most_workers = threading.Lock(), 0, 0, 0
|
||||
|
||||
def get_odds(self, sport, league, event_id, update_interval_seconds=None):
|
||||
workers = sum(t.name == "nba-odds" and t.is_alive() for t in threading.enumerate())
|
||||
with self.lock:
|
||||
self.calls.append((event_id, update_interval_seconds))
|
||||
self.active += 1
|
||||
self.most_active = max(self.most_active, self.active)
|
||||
self.most_workers = max(self.most_workers, workers)
|
||||
self.gate.wait(10)
|
||||
with self.lock:
|
||||
self.active -= 1
|
||||
if isinstance(self.answer, type):
|
||||
raise self.answer("odds endpoint failed")
|
||||
return {"details": f"{event_id} -3.5"} if self.answer else None
|
||||
|
||||
def release(self):
|
||||
self.gate.set()
|
||||
|
||||
|
||||
def settle():
|
||||
for thread in threading.enumerate():
|
||||
if thread.name == "nba-odds":
|
||||
thread.join(10)
|
||||
|
||||
|
||||
def game(gid="g1", **flags):
|
||||
return {"id": gid, **flags}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flags, interval", [({"is_live": True}, 60), ({"is_upcoming": True}, 3600),
|
||||
({"is_final": True}, 3600)])
|
||||
def test_the_call_never_waits_and_the_line_lands_later(flags, interval):
|
||||
odds = Odds()
|
||||
host, g = Host(odds=odds), game(**flags)
|
||||
host._fetch_odds(g) # returns while the request is held open
|
||||
assert "odds" not in g
|
||||
odds.release()
|
||||
settle()
|
||||
assert g["odds"] == {"details": "g1 -3.5"}
|
||||
assert odds.calls == [("g1", interval)]
|
||||
|
||||
|
||||
def test_the_intervals_come_from_the_mode_config():
|
||||
odds = Odds(held=False)
|
||||
host = Host(odds=odds)
|
||||
host.mode_config = {"live_odds_update_interval": 30, "odds_update_interval": 900}
|
||||
host._fetch_odds(game("l", is_live=True))
|
||||
settle()
|
||||
host._fetch_odds(game("u"))
|
||||
settle()
|
||||
assert odds.calls == [("l", 30), ("u", 900)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("answer", [None, RuntimeError])
|
||||
def test_no_line_or_a_failed_request_leaves_the_game_bare(caplog, answer):
|
||||
odds = Odds(answer, held=False)
|
||||
host, g = Host(odds=odds), game()
|
||||
with caplog.at_level(logging.DEBUG, logger=LOG.name):
|
||||
host._fetch_odds(g)
|
||||
settle()
|
||||
assert "odds" not in g and host._odds_worker_busy is False
|
||||
assert [r for r in caplog.records if r.levelno >= logging.WARNING] == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", [
|
||||
Host(odds=Odds(held=False), show_odds=False),
|
||||
Host(odds=None),
|
||||
], ids=["show_odds off", "no odds manager"])
|
||||
def test_nothing_is_asked(caplog, host):
|
||||
with caplog.at_level(logging.DEBUG, logger=LOG.name):
|
||||
host._fetch_odds(game())
|
||||
assert not [t for t in threading.enumerate() if t.name == "nba-odds"]
|
||||
assert [r.levelname for r in caplog.records] in ([], ["DEBUG"])
|
||||
|
||||
|
||||
def test_a_game_without_an_id_is_not_asked():
|
||||
odds = Odds(held=False)
|
||||
Host(odds=odds)._fetch_odds({"is_live": True})
|
||||
settle()
|
||||
assert odds.calls == []
|
||||
|
||||
|
||||
def test_one_batch_per_manager_and_every_dict_gets_the_answer():
|
||||
odds = Odds()
|
||||
host = Host(odds=odds)
|
||||
first, again, other = game("g1"), game("g1"), game("g2")
|
||||
with host._games_lock: # queued before the worker takes one
|
||||
for g in (first, again, other): # update() rebuilds the dicts
|
||||
host._fetch_odds(g)
|
||||
odds.release()
|
||||
settle()
|
||||
assert odds.calls == [("g1", 3600), ("g2", 3600)]
|
||||
assert odds.most_workers == 1
|
||||
assert first["odds"] == again["odds"] == {"details": "g1 -3.5"}
|
||||
|
||||
|
||||
def test_a_thread_that_will_not_start_is_tried_again(monkeypatch):
|
||||
class NoThread:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def start(self):
|
||||
raise RuntimeError("can't start new thread")
|
||||
|
||||
odds = Odds(held=False)
|
||||
host = Host(odds=odds)
|
||||
monkeypatch.setattr(sports_live_fetch.threading, "Thread", NoThread)
|
||||
host._fetch_odds(game("g1"))
|
||||
assert host._odds_worker_busy is False
|
||||
monkeypatch.undo()
|
||||
host._fetch_odds(game("g2"))
|
||||
settle()
|
||||
assert [c[0] for c in odds.calls] == ["g1", "g2"]
|
||||
|
||||
|
||||
def test_update_and_display_on_two_threads():
|
||||
"""Two updaters queueing fresh dicts while a reader walks the games under
|
||||
the lock, as display() does: no exception, one worker, one request at a
|
||||
time, and every dict ends with its line."""
|
||||
odds = Odds(held=False)
|
||||
odds.get_odds = _slowed(odds.get_odds)
|
||||
host, shown, raised = Host(odds=odds), [], []
|
||||
|
||||
def updater():
|
||||
try:
|
||||
for n in range(30):
|
||||
g = game(f"g{n % 4}")
|
||||
with host._games_lock:
|
||||
shown.append(g)
|
||||
host._fetch_odds(g)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raised.append(exc)
|
||||
|
||||
def display():
|
||||
try:
|
||||
for _ in range(300):
|
||||
with host._games_lock:
|
||||
[g.get("odds") for g in shown]
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raised.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=updater) for _ in range(2)]
|
||||
threads.append(threading.Thread(target=display))
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join(30)
|
||||
settle()
|
||||
assert raised == []
|
||||
assert odds.most_active == 1 and odds.most_workers == 1
|
||||
assert all(g.get("odds") for g in shown)
|
||||
|
||||
|
||||
def _slowed(get_odds):
|
||||
def slow(**kwargs):
|
||||
time.sleep(0.005)
|
||||
return get_odds(**kwargs)
|
||||
return slow
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Carrier and host contract
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
MEMBERS = ["_drain_odds_queue", "_fetch_odds", "_fetch_todays_games"]
|
||||
|
||||
|
||||
def _self_reads():
|
||||
tree = ast.parse(Path(sports_live_fetch.__file__).read_text(encoding="utf-8"))
|
||||
reads = {node.attr for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Attribute) and isinstance(node.ctx, ast.Load)
|
||||
and isinstance(node.value, ast.Name) and node.value.id == "self"}
|
||||
gets = {node.args[1].value for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "getattr"
|
||||
and len(node.args) >= 2 and isinstance(node.args[1], ast.Constant)}
|
||||
return reads | gets
|
||||
|
||||
|
||||
class TestHostContract:
|
||||
def test_the_mixin_carries_only_the_family(self):
|
||||
assert sorted(n for n in vars(SportsLiveFetchMixin) if not n.startswith("__")) == MEMBERS
|
||||
|
||||
def test_every_host_read_is_documented(self):
|
||||
undocumented = sorted(n for n in _self_reads() - set(MEMBERS)
|
||||
if f"``{n}``" not in sports_live_fetch.__doc__)
|
||||
assert undocumented == [], f"read but not in the host contract: {undocumented}"
|
||||
|
||||
def test_no_other_shared_mixin_defines_these(self):
|
||||
from src.common import (sports_display_rules, sports_favorites, sports_helpers,
|
||||
sports_rankings, sports_rotation, sports_shared)
|
||||
others = [sports_shared.SportsCoreSharedMixin, sports_shared.SportsRecentSharedMixin,
|
||||
sports_shared.SportsLiveSharedMixin, sports_helpers.SportsHelpersMixin,
|
||||
sports_favorites.SportsFavoritesMixin, sports_rotation.SportsRotationMixin,
|
||||
sports_rankings.SportsRankingsMixin, SportsFetchMixin,
|
||||
sports_display_rules.SportsGameRulesMixin]
|
||||
assert [m for m in MEMBERS if any(m in vars(o) for o in others)] == []
|
||||
@@ -0,0 +1,104 @@
|
||||
"""sports_live_fetch still matches every plugin copy.
|
||||
|
||||
``src.common.sports_live_fetch`` was copied from the scoreboards once family 9
|
||||
had made each method one body in all nine ``SportsCore`` classes:
|
||||
``_fetch_todays_games``, ``_fetch_odds`` and ``_drain_odds_queue``. The plugins delete their copies once they floor on the release that
|
||||
ships this module. Until each has, a copy that changes on its own is a fix one
|
||||
side has and the other lacks.
|
||||
|
||||
Point LEDMATRIX_PLUGINS at a ledmatrix-plugins checkout and each method is
|
||||
compared with every plugin copy using ``scripts/sports_drift_report.py``'s own
|
||||
normalisation (the AST with docstrings and annotations dropped), plus the
|
||||
decorators. A copy that is gone counts as adopted when the plugin's ``sports.py`` names the module.
|
||||
Without the variable this skips: core CI has no plugins checkout.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import importlib.util
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.common import sports_live_fetch
|
||||
|
||||
REPO = Path(__file__).resolve().parents[1]
|
||||
SPORTS = ("afl", "baseball", "basketball", "football", "hockey", "lacrosse",
|
||||
"nrl", "soccer", "ufc")
|
||||
CARRIER, MIXIN = "SportsCore", "SportsLiveFetchMixin"
|
||||
METHODS = ("_fetch_todays_games", "_fetch_odds", "_drain_odds_queue")
|
||||
|
||||
|
||||
def _drift_report():
|
||||
"""scripts/sports_drift_report.py, loaded by path (scripts/ is no package)."""
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"sports_drift_report", REPO / "scripts" / "sports_drift_report.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
DRIFT = _drift_report()
|
||||
|
||||
|
||||
def _plugins_root():
|
||||
root = DRIFT.resolve_plugins_dir(os.environ.get("LEDMATRIX_PLUGINS"))
|
||||
if root is None:
|
||||
pytest.skip("set LEDMATRIX_PLUGINS to a ledmatrix-plugins checkout to "
|
||||
"compare this module against the plugin copies")
|
||||
return root
|
||||
|
||||
|
||||
def _class(tree, name):
|
||||
return next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name)
|
||||
|
||||
|
||||
def _method(cls, name):
|
||||
return next((n for n in cls.body
|
||||
if isinstance(n, ast.FunctionDef) and n.name == name), None)
|
||||
|
||||
|
||||
|
||||
def _fingerprint(node):
|
||||
return (DRIFT._digest(node, DRIFT._Canonical()),
|
||||
tuple(ast.unparse(d) for d in node.decorator_list))
|
||||
|
||||
|
||||
def _plugin_tree(root, sport):
|
||||
source = (root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")
|
||||
return source, ast.parse(source)
|
||||
|
||||
|
||||
OURS = _class(ast.parse(Path(sports_live_fetch.__file__).read_text(encoding="utf-8")), MIXIN)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sport, name", [(s, n) for s in SPORTS for n in METHODS])
|
||||
def test_every_remaining_plugin_copy_matches(sport, name):
|
||||
source, tree = _plugin_tree(_plugins_root(), sport)
|
||||
copy = _method(_class(tree, CARRIER), name)
|
||||
if copy is None:
|
||||
assert sports_live_fetch.__name__ in source, (
|
||||
f"{sport}: no {name} on {CARRIER} and no {sports_live_fetch.__name__} import")
|
||||
else:
|
||||
assert _fingerprint(copy) == _fingerprint(_method(OURS, name)), (
|
||||
f"{CARRIER}.{name} in {sport} differs from sports_live_fetch. "
|
||||
f"Port the change to both, or stop treating it as shared.")
|
||||
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sport", SPORTS)
|
||||
def test_no_other_plugin_class_carries_a_copy(sport):
|
||||
"""A copy on another class would shadow the shared one for that class."""
|
||||
_, tree = _plugin_tree(_plugins_root(), sport)
|
||||
strays = [f"{node.name}.{name}" for node in tree.body if isinstance(node, ast.ClassDef)
|
||||
for name in METHODS if node.name != CARRIER and _method(node, name) is not None]
|
||||
assert strays == []
|
||||
|
||||
|
||||
def test_the_drift_report_calls_them_identical():
|
||||
families = DRIFT.build(_plugins_root(), ("sports.py",))
|
||||
rows = {(r["file"], r["family"]): r
|
||||
for r in (DRIFT.summarise(k, v) for k, v in families.items())}
|
||||
for name in METHODS:
|
||||
row = rows.get(("sports.py", name))
|
||||
assert row is None or row["worst_class_variants"] == 1, name
|
||||
@@ -0,0 +1,395 @@
|
||||
"""src.common.sports_rankings: the poll, the tables, the season, the filter.
|
||||
|
||||
The cases follow ledmatrix-plugins' ``scripts/test_rankings.py`` (the tables
|
||||
the family 8 reconcile was checked against); the expected values are its
|
||||
columns, which agree across all nine scoreboards for everything this module
|
||||
does. The host is the mixin over core's ``SportsCoreSharedMixin``
|
||||
(``_league_has_rankings``, ``_game_divisions``) and ``SportsRotationMixin``
|
||||
(``_rankings_loaded``), with a data source and an HTTP session that answer
|
||||
from fixtures shaped like ESPN's payloads. Nothing reaches the network.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from src.common import sports_rankings
|
||||
from src.common.sports_favorites import SportsFavoritesMixin
|
||||
from src.common.sports_helpers import SportsHelpersMixin
|
||||
from src.common.sports_rankings import SportsRankingsMixin, choose_top_division_poll
|
||||
from src.common.sports_rotation import SportsRotationMixin
|
||||
from src.common.sports_shared import SportsCoreSharedMixin
|
||||
|
||||
LOG = logging.getLogger("test_sports_rankings")
|
||||
CORE_API = "https://sports.core.api.espn.com/v2/sports/"
|
||||
|
||||
UGA, ND, OSU, TEX, VAN, SDSU = ("61", "UGA"), ("87", "ND"), ("194", "OSU"), \
|
||||
("251", "TEX"), ("238", "VAN"), ("21", "SDSU")
|
||||
SDST, NDSU, NEWARK, FRST, DEN = ("2571", "SDST"), ("2449", "NDSU"), \
|
||||
("9194", "OSU"), ("9001", "FRST"), ("2172", "DEN")
|
||||
FBS_IDS, FCS_IDS = (UGA, ND, OSU, TEX, VAN, SDSU), (SDST, NDSU)
|
||||
|
||||
|
||||
def poll(name, kind, *teams, **extra):
|
||||
return {"name": name, "shortName": name, "type": kind, **extra, "ranks": [
|
||||
{"current": n, "previous": n, "team": {"id": tid, "abbreviation": abbr}}
|
||||
for n, (tid, abbr) in enumerate(teams, 1)]}
|
||||
|
||||
|
||||
AP = poll("AP Top 25", "ap", UGA, ND, OSU, SDSU)
|
||||
COACHES = poll("AFCA Coaches Poll", "usa", ND, UGA, TEX)
|
||||
CFP = poll("College Football Playoff Rankings", "cfp", OSU, UGA)
|
||||
FCS = poll("FCS Coaches Poll", "fcs", NDSU, SDST)
|
||||
D2 = poll("AFCA Division II Coaches Poll", "afca", FRST)
|
||||
SEEDS = poll("NCAA Men's Hockey Tournament Seedings", "tournament", DEN)
|
||||
USCHO = poll("USCHO Men's Poll", "USCHOMENSPOLL", DEN, ND)
|
||||
AP_TABLE = {"ND": 2, "OSU": 3, "SDSU": 4, "UGA": 1}
|
||||
AP_IDS = {194: 3, 21: 4, 61: 1, 87: 2}
|
||||
|
||||
|
||||
def at(text):
|
||||
return datetime.fromisoformat(text).replace(tzinfo=timezone.utc).timestamp()
|
||||
|
||||
|
||||
class Clock:
|
||||
def __init__(self):
|
||||
self.now = 100_000.0
|
||||
|
||||
def __call__(self):
|
||||
return self.now
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def clock(monkeypatch):
|
||||
c = Clock()
|
||||
monkeypatch.setattr(sports_rankings.time, "time", c)
|
||||
monkeypatch.setattr(sports_rankings.time, "monotonic", c)
|
||||
return c
|
||||
|
||||
|
||||
class Rosters:
|
||||
"""The core API's division rosters, published up to ``last`` season."""
|
||||
|
||||
def __init__(self, last=2026):
|
||||
self.last, self.urls = last, []
|
||||
|
||||
def get(self, url, params=None, timeout=None):
|
||||
self.urls.append(url[len(CORE_API):])
|
||||
season, group = url.split("/seasons/")[1].split("/")[0], url.split("/groups/")[1][:2]
|
||||
ids = {"80": FBS_IDS, "81": FCS_IDS}[group] if int(season) <= self.last else ()
|
||||
resp = MagicMock()
|
||||
resp.json.return_value = {"items": [{"$ref": f"{CORE_API}x/teams/{t}?lang=en"}
|
||||
for t, _ in ids]}
|
||||
return resp
|
||||
|
||||
|
||||
class Host(SportsRankingsMixin, SportsRotationMixin, SportsCoreSharedMixin,
|
||||
SportsFavoritesMixin, SportsHelpersMixin):
|
||||
"""A scoreboard SportsCore as far as the rankings reach."""
|
||||
|
||||
sport = "football"
|
||||
|
||||
def __init__(self, league="college-football", payload=None, quality="any",
|
||||
divisions=(), rosters=None):
|
||||
self.logger = LOG
|
||||
self.league = league
|
||||
self.data_source = MagicMock()
|
||||
self.data_source.fetch_standings.return_value = (
|
||||
{"rankings": [AP, COACHES]} if payload is None else payload)
|
||||
self.session = rosters or Rosters()
|
||||
self.cache_manager = MagicMock()
|
||||
self.cache_manager.get.return_value = None
|
||||
self._team_rankings_cache, self._ranked_team_ids = {}, {}
|
||||
self._rankings_cache_timestamp, self._rankings_cache_duration = 0, 3600
|
||||
self._division_team_ids, self._division_loaded_at = None, 0.0
|
||||
self.other_games_min_quality = quality
|
||||
self.other_games_divisions = list(divisions)
|
||||
|
||||
|
||||
def tables(host):
|
||||
return host._team_rankings_cache, host._ranked_team_ids
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Which poll
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
POLL_SETS = [
|
||||
([AP, COACHES], "AP Top 25"),
|
||||
([COACHES, AP], "AFCA Coaches Poll"),
|
||||
([CFP, AP, COACHES], "College Football Playoff Rankings"),
|
||||
([FCS, D2, AP, COACHES], "AP Top 25"),
|
||||
([FCS, D2], None),
|
||||
([poll("Coaches Poll", "fcs", NDSU), AP], "AP Top 25"), # by type alone
|
||||
([SEEDS, USCHO], "USCHO Men's Poll"),
|
||||
([], None),
|
||||
(None, None),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blocks, expected", POLL_SETS)
|
||||
def test_the_poll_chosen(blocks, expected):
|
||||
assert Host()._choose_poll(blocks).get("name") == expected
|
||||
assert choose_top_division_poll(blocks).get("name") == expected
|
||||
|
||||
|
||||
def test_an_untyped_unnamed_block_counts():
|
||||
"""An exclude list: a block ESPN invents is a poll unless it says otherwise."""
|
||||
assert choose_top_division_poll([{"ranks": []}]) == {"ranks": []}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The tables
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("league, asked", [
|
||||
("college-football", True), ("mens-college-basketball", True),
|
||||
("usa.ncaa.m.1", True), ("nfl", False), ("nba", False), ("eng.1", False)])
|
||||
def test_only_a_league_with_a_poll_is_asked(clock, league, asked):
|
||||
host = Host(league)
|
||||
host._fetch_team_rankings()
|
||||
assert host.data_source.fetch_standings.called is asked
|
||||
assert tables(host) == ((AP_TABLE, AP_IDS) if asked else ({}, {}))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload, abbr, ids", [
|
||||
({}, {}, {}),
|
||||
({"rankings": []}, {}, {}),
|
||||
({"rankings": None}, {}, {}),
|
||||
({"rankings": [{"name": "AP Top 25", "type": "ap"}]}, {}, {}),
|
||||
({"rankings": [{"name": "AP Top 25", "type": "ap", "ranks": [
|
||||
{"current": 0, "team": {"id": "61", "abbreviation": "UGA"}},
|
||||
{"current": -1, "team": {"id": "87", "abbreviation": "ND"}},
|
||||
{"team": {"id": "194", "abbreviation": "OSU"}},
|
||||
{"current": 4, "team": {"id": "251", "abbreviation": "TEX"}}]}]},
|
||||
{"TEX": 4}, {251: 4}),
|
||||
({"rankings": [{"name": "AP Top 25", "type": "ap", "ranks": [
|
||||
{"current": 1},
|
||||
{"current": 2, "team": {"id": "87"}},
|
||||
{"current": 3, "team": {"id": "abc", "abbreviation": "OSU"}},
|
||||
{"current": 4, "team": {"abbreviation": "TEX"}}]}]},
|
||||
{"OSU": 3, "TEX": 4}, {87: 2}),
|
||||
({"rankings": [{"name": "AP Top 25", "type": "ap", "ranks": [ # #789: one bad entry
|
||||
{"current": None, "team": {"id": "61", "abbreviation": "UGA"}},
|
||||
{"current": "n/a", "team": {"id": "87", "abbreviation": "ND"}},
|
||||
{"current": 3, "team": {"id": "194", "abbreviation": "OSU"}}]}]},
|
||||
{"OSU": 3}, {194: 3}),
|
||||
])
|
||||
def test_payload_shapes(clock, payload, abbr, ids):
|
||||
host = Host(payload=payload)
|
||||
host._fetch_team_rankings()
|
||||
assert tables(host) == (abbr, ids)
|
||||
assert host._rankings_cache_timestamp == clock.now # an empty answer is cached too
|
||||
|
||||
|
||||
def test_a_failed_fetch_is_not_cached(clock):
|
||||
host = Host(payload={"rankings": {"name": "AP Top 25"}}) # not a list
|
||||
assert host._fetch_team_rankings() == {}
|
||||
assert host._rankings_cache_timestamp == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [{"rankings": [AP, COACHES]}, {"rankings": []}])
|
||||
def test_rankings_are_fetched_hourly(clock, payload):
|
||||
host = Host(payload=payload)
|
||||
calls = []
|
||||
for step in (0, 60, 3599, 3601):
|
||||
clock.now = 100_000.0 + step
|
||||
before = host.data_source.fetch_standings.call_count
|
||||
host._fetch_team_rankings()
|
||||
calls.append(host.data_source.fetch_standings.call_count - before)
|
||||
assert calls == [1, 0, 0, 1]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# A poll from a season that has ended
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("end, now, kept", [
|
||||
("2026-04-08T06:59Z", "2026-10-12T12:00", False), # college basketball in October
|
||||
("2026-04-08T06:59Z", "2026-04-08T06:58", True),
|
||||
("2026-04-08T06:59Z", "2026-04-08T06:59", True), # at endDate: not yet over
|
||||
("2026-04-08T06:59Z", "2026-04-08T07:00", False),
|
||||
("2027-01-28T07:59Z", "2027-01-10T12:00", True), # bowl week
|
||||
("2027-04-08T06:59Z", "2026-11-03T12:00", True), # the new season's first poll
|
||||
("2026-04-08T06:59:00Z", "2026-10-12T12:00", False),
|
||||
("April", "2026-10-12T12:00", True), # unreadable: current
|
||||
(None, "2026-10-12T12:00", True), # no season block
|
||||
])
|
||||
def test_a_finished_seasons_poll_loads_nothing(clock, end, now, kept):
|
||||
block = dict(AP, season={"year": 2026, "endDate": end}) if end else AP
|
||||
host = Host(payload={"rankings": [block, COACHES]})
|
||||
clock.now = at(now)
|
||||
host._fetch_team_rankings()
|
||||
assert tables(host) == ((AP_TABLE, AP_IDS) if kept else ({}, {}))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Division rosters
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class Frozen(datetime):
|
||||
at = datetime(2026, 10, 12, 12)
|
||||
|
||||
@classmethod
|
||||
def now(cls, tz=None):
|
||||
return cls.at
|
||||
|
||||
|
||||
@pytest.mark.parametrize("today, season, sizes", [
|
||||
("2026-01-15", 2025, (6, 2)), ("2026-06-30", 2025, (6, 2)),
|
||||
("2026-07-01", 2026, (6, 2)), ("2026-10-12", 2026, (6, 2)),
|
||||
("2026-12-31", 2026, (6, 2)), ("2027-01-15", 2026, (6, 2))])
|
||||
def test_the_rosters_are_for_the_season_year(clock, monkeypatch, today, season, sizes):
|
||||
monkeypatch.setattr(sports_rankings, "datetime", Frozen)
|
||||
Frozen.at = datetime.fromisoformat(today)
|
||||
host = Host()
|
||||
rosters = host._load_division_team_ids()
|
||||
assert host.session.urls == [
|
||||
f"football/leagues/college-football/seasons/{season}/types/2/groups/{g}/teams"
|
||||
for g in (80, 81)]
|
||||
assert (len(rosters["fbs"]), len(rosters["fcs"])) == sizes
|
||||
|
||||
|
||||
def test_no_rosters_outside_college_football(clock):
|
||||
host = Host("mens-college-basketball")
|
||||
assert host._load_division_team_ids() == {}
|
||||
assert host.session.urls == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("last, cached, calls, sizes", [
|
||||
(2026, None, [2, 0, 0, 2], (6, 2)), # a day
|
||||
(0, None, [2, 0, 2, 2], (0, 0)), # empty: retried after 10 minutes
|
||||
(2026, "cache", [0, 0, 0, 0], (2, 1)), # the cache manager's copy
|
||||
])
|
||||
def test_roster_caching(clock, last, cached, calls, sizes):
|
||||
host = Host(rosters=Rosters(last))
|
||||
if cached:
|
||||
host.cache_manager.get.side_effect = lambda key, *a, **k: (
|
||||
["61", "87"] if key.endswith("_80") else ["2571"])
|
||||
made = []
|
||||
for step in (0, 60, 601, 86_401):
|
||||
clock.now = 100_000.0 + step
|
||||
before = len(host.session.urls)
|
||||
host._load_division_team_ids()
|
||||
made.append(len(host.session.urls) - before)
|
||||
rosters = host._division_team_ids
|
||||
assert made == calls
|
||||
assert (len(rosters["fbs"]), len(rosters["fcs"])) == sizes
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Ranked games and the filter
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
GAMES = {
|
||||
"UGA v TEX": (UGA, TEX),
|
||||
"TEX v VAN": (TEX, VAN),
|
||||
"OSU v VAN": (OSU, VAN),
|
||||
"Newark 'OSU' v VAN": (NEWARK, VAN),
|
||||
"SDSU v SDST": (SDSU, SDST),
|
||||
"SDST v NDSU": (SDST, NDSU),
|
||||
"'OSU' v VAN, no ids": ((None, "OSU"), (None, "VAN")),
|
||||
"61 v 238, no abbrs": (("61", None), ("238", None)),
|
||||
}
|
||||
|
||||
|
||||
def game(label):
|
||||
g = {"id": "g"}
|
||||
for side, (tid, abbr) in zip(("home", "away"), GAMES[label]):
|
||||
if tid is not None:
|
||||
g[f"{side}_id"] = tid
|
||||
if abbr is not None:
|
||||
g[f"{side}_abbr"] = abbr
|
||||
return g
|
||||
|
||||
|
||||
def loaded(**kwargs):
|
||||
host = Host(**kwargs)
|
||||
host._fetch_team_rankings()
|
||||
return host
|
||||
|
||||
|
||||
@pytest.mark.parametrize("label, best", [
|
||||
("UGA v TEX", 1), ("TEX v VAN", 99), ("OSU v VAN", 3),
|
||||
("Newark 'OSU' v VAN", 99), # same abbreviation, another school: by id
|
||||
("SDSU v SDST", 4), ("SDST v NDSU", 99),
|
||||
("'OSU' v VAN, no ids", 3), # no ids: by abbreviation
|
||||
("61 v 238, no abbrs", 1)])
|
||||
def test_ranked_and_best_rank(clock, label, best):
|
||||
host = loaded()
|
||||
assert host._best_rank(game(label)) == best
|
||||
assert host._is_ranked_game(game(label)) is (best != 99)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("home, away, best", [
|
||||
(NEWARK, (None, "VAN"), 99), # Newark's id is unranked; "OSU" must not rank it
|
||||
(("61", None), (None, "VAN"), 1), # UGA's id still counts with no id on the other side
|
||||
])
|
||||
def test_each_side_is_ranked_by_its_own_id(clock, home, away, best):
|
||||
"""#789: one side's missing id no longer sends both sides to abbreviations."""
|
||||
g = {f"{side}_{key}": value for side, team in (("home", home), ("away", away))
|
||||
for key, value in zip(("id", "abbr"), team) if value is not None}
|
||||
assert loaded()._best_rank(g) == best
|
||||
|
||||
|
||||
#: filter -> (quality, divisions, poll loaded?, kept: one Y/. per GAMES entry)
|
||||
FILTERS = {
|
||||
"any": ("any", (), True, "YYYYYYYY"),
|
||||
"ranked": ("ranked", (), True, "Y.Y.Y.YY"),
|
||||
"ranked, no poll loaded": ("ranked", (), False, "YYYYYYYY"),
|
||||
"any, fbs": ("any", ("fbs",), True, "YYYYY.YY"),
|
||||
"ranked, fbs": ("ranked", ("fbs",), True, "Y.Y.Y.YY"),
|
||||
"any, fcs": ("any", ("fcs",), True, "....YYY."),
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", FILTERS)
|
||||
def test_passes_other_filters(clock, monkeypatch, name):
|
||||
quality, divisions, with_poll, kept = FILTERS[name]
|
||||
monkeypatch.setattr(sports_rankings, "datetime", Frozen)
|
||||
Frozen.at = datetime(2026, 10, 12, 12)
|
||||
host = loaded(payload={"rankings": [AP, COACHES] if with_poll else []},
|
||||
quality=quality, divisions=divisions)
|
||||
assert "".join("Y" if host._passes_other_filters(game(g)) else "." for g in GAMES) == kept
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Carrier and host contract
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
MEMBERS = ["_DIVISION_CACHE_TTL", "_DIVISION_GROUPS_BY_LEAGUE", "_DIVISION_RETRY_SECONDS",
|
||||
"_NON_TOP_POLL_NAMES", "_NON_TOP_POLL_TYPES", "_best_rank", "_choose_poll",
|
||||
"_fetch_team_rankings", "_is_ranked_game", "_load_division_team_ids",
|
||||
"_passes_other_filters", "_poll_season_ended", "_ranked_positions"]
|
||||
|
||||
|
||||
def _self_reads():
|
||||
tree = ast.parse(Path(sports_rankings.__file__).read_text(encoding="utf-8"))
|
||||
return {node.attr for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Attribute) and isinstance(node.ctx, ast.Load)
|
||||
and isinstance(node.value, ast.Name) and node.value.id == "self"}
|
||||
|
||||
|
||||
class TestHostContract:
|
||||
def test_the_mixin_carries_only_the_family(self):
|
||||
"""Every scoreboard's SportsCore has all of them, so adopting it adds none."""
|
||||
assert sorted(n for n in vars(SportsRankingsMixin)
|
||||
if not n.startswith("__")) == MEMBERS
|
||||
|
||||
def test_every_host_read_is_documented(self):
|
||||
own = set(MEMBERS)
|
||||
undocumented = sorted(n for n in _self_reads() - own
|
||||
if f"``{n}``" not in sports_rankings.__doc__)
|
||||
assert undocumented == [], f"read but not in the host contract: {undocumented}"
|
||||
|
||||
def test_no_other_shared_mixin_defines_these(self):
|
||||
from src.common import sports_display_rules, sports_shared
|
||||
others = [sports_shared.SportsCoreSharedMixin, sports_shared.SportsRecentSharedMixin,
|
||||
sports_shared.SportsLiveSharedMixin, SportsHelpersMixin,
|
||||
SportsFavoritesMixin, SportsRotationMixin,
|
||||
sports_display_rules.SportsGameRulesMixin]
|
||||
assert [m for m in MEMBERS if any(m in vars(o) for o in others)] == []
|
||||
@@ -0,0 +1,132 @@
|
||||
"""sports_rankings still matches every plugin copy.
|
||||
|
||||
``src.common.sports_rankings`` was copied from the scoreboards once family 8
|
||||
had made each method one body in all nine ``SportsCore`` classes:
|
||||
``_choose_poll``, ``_fetch_team_rankings``, ``_poll_season_ended``,
|
||||
``_load_division_team_ids``, ``_ranked_positions``, ``_is_ranked_game``,
|
||||
``_best_rank`` and ``_passes_other_filters``, with the class constants they
|
||||
read. The plugins delete their copies once they floor on the release that
|
||||
ships this module. Until each has, a copy that changes on its own is a fix one
|
||||
side has and the other lacks.
|
||||
|
||||
Point LEDMATRIX_PLUGINS at a ledmatrix-plugins checkout and each method is
|
||||
compared with every plugin copy using ``scripts/sports_drift_report.py``'s own
|
||||
normalisation (the AST with docstrings and annotations dropped), plus the
|
||||
decorators, and each constant a plugin still defines by its expression. A copy that is
|
||||
gone counts as adopted when the plugin's ``sports.py`` names the module.
|
||||
Without the variable this skips: core CI has no plugins checkout.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import importlib.util
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.common import sports_rankings
|
||||
|
||||
REPO = Path(__file__).resolve().parents[1]
|
||||
SPORTS = ("afl", "baseball", "basketball", "football", "hockey", "lacrosse",
|
||||
"nrl", "soccer", "ufc")
|
||||
CARRIER, MIXIN = "SportsCore", "SportsRankingsMixin"
|
||||
METHODS = ("_choose_poll", "_fetch_team_rankings", "_poll_season_ended",
|
||||
"_load_division_team_ids", "_ranked_positions", "_is_ranked_game",
|
||||
"_best_rank", "_passes_other_filters")
|
||||
CONSTANTS = ("_NON_TOP_POLL_TYPES", "_NON_TOP_POLL_NAMES", "_DIVISION_GROUPS_BY_LEAGUE",
|
||||
"_DIVISION_CACHE_TTL", "_DIVISION_RETRY_SECONDS")
|
||||
|
||||
|
||||
def _drift_report():
|
||||
"""scripts/sports_drift_report.py, loaded by path (scripts/ is no package)."""
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"sports_drift_report", REPO / "scripts" / "sports_drift_report.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
DRIFT = _drift_report()
|
||||
|
||||
|
||||
def _plugins_root():
|
||||
root = DRIFT.resolve_plugins_dir(os.environ.get("LEDMATRIX_PLUGINS"))
|
||||
if root is None:
|
||||
pytest.skip("set LEDMATRIX_PLUGINS to a ledmatrix-plugins checkout to "
|
||||
"compare this module against the plugin copies")
|
||||
return root
|
||||
|
||||
|
||||
def _class(tree, name):
|
||||
return next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name)
|
||||
|
||||
|
||||
def _method(cls, name):
|
||||
return next((n for n in cls.body
|
||||
if isinstance(n, ast.FunctionDef) and n.name == name), None)
|
||||
|
||||
|
||||
def _constant(cls, name):
|
||||
"""The value a class body assigns to ``name`` (plain or annotated), or None."""
|
||||
for node in cls.body:
|
||||
if isinstance(node, ast.Assign) and any(
|
||||
isinstance(t, ast.Name) and t.id == name for t in node.targets):
|
||||
return node.value
|
||||
if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) \
|
||||
and node.target.id == name:
|
||||
return node.value
|
||||
return None
|
||||
|
||||
|
||||
def _fingerprint(node):
|
||||
return (DRIFT._digest(node, DRIFT._Canonical()),
|
||||
tuple(ast.unparse(d) for d in node.decorator_list))
|
||||
|
||||
|
||||
def _plugin_tree(root, sport):
|
||||
source = (root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")
|
||||
return source, ast.parse(source)
|
||||
|
||||
|
||||
OURS = _class(ast.parse(Path(sports_rankings.__file__).read_text(encoding="utf-8")), MIXIN)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sport, name", [(s, n) for s in SPORTS for n in METHODS])
|
||||
def test_every_remaining_plugin_copy_matches(sport, name):
|
||||
source, tree = _plugin_tree(_plugins_root(), sport)
|
||||
copy = _method(_class(tree, CARRIER), name)
|
||||
if copy is None:
|
||||
assert sports_rankings.__name__ in source, (
|
||||
f"{sport}: no {name} on {CARRIER} and no {sports_rankings.__name__} import")
|
||||
else:
|
||||
assert _fingerprint(copy) == _fingerprint(_method(OURS, name)), (
|
||||
f"{CARRIER}.{name} in {sport} differs from sports_rankings. "
|
||||
f"Port the change to both, or stop treating it as shared.")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sport", SPORTS)
|
||||
def test_the_plugin_constants_match(sport):
|
||||
_, tree = _plugin_tree(_plugins_root(), sport)
|
||||
core = _class(tree, CARRIER)
|
||||
for name in CONSTANTS:
|
||||
value = _constant(core, name)
|
||||
if value is not None:
|
||||
assert ast.dump(value) == ast.dump(_constant(OURS, name)), f"{sport}: {name}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sport", SPORTS)
|
||||
def test_no_other_plugin_class_carries_a_copy(sport):
|
||||
"""A copy on another class would shadow the shared one for that class."""
|
||||
_, tree = _plugin_tree(_plugins_root(), sport)
|
||||
strays = [f"{node.name}.{name}" for node in tree.body if isinstance(node, ast.ClassDef)
|
||||
for name in METHODS if node.name != CARRIER and _method(node, name) is not None]
|
||||
assert strays == []
|
||||
|
||||
|
||||
def test_the_drift_report_calls_them_identical():
|
||||
families = DRIFT.build(_plugins_root(), ("sports.py",))
|
||||
rows = {(r["file"], r["family"]): r
|
||||
for r in (DRIFT.summarise(k, v) for k, v in families.items())}
|
||||
for name in METHODS:
|
||||
row = rows.get(("sports.py", name))
|
||||
assert row is None or row["worst_class_variants"] == 1, name
|
||||
Reference in New Issue
Block a user