Compare commits

..
Author SHA1 Message Date
ChuckBuildsandClaude Opus 5 0e0b89388d fix(odds): identify the odds requests to ESPN
The odds fetch used a bare requests.get, so it went out as
python-requests/x.y -- the one agent ESPN is known to reject. Around
2026-08-04 it began 403ing browser strings and bare custom tokens alike;
what it accepts is a token carrying a URL that says who is calling.
Every other ESPN caller in the tree already sends that header
(src/common/api_helper.py, src/base_classes/data_sources.py); this path
was simply missed.

It is the worst one to miss. Odds are fetched per live game from inside
the live update loop, so its failures are the ones that cost the caller
its whole update budget -- the same path the 5s timeout and the cooldown
were added to protect.

Sent via a session rather than per-call, which also reuses the
connection across a slate. Deliberately no retry adapter, unlike
api_helper: retries multiply request_timeout, which is 5s precisely to
stay inside the 30s operation budget.

The existing tests patched the module's requests.get, which this change
bypasses -- test_base_odds_manager was consequently reaching the real
ESPN and taking 404s. Both files now patch the session, and the new
tests pin the agent against api_helper's live value so the two cannot
drift apart the next time ESPN moves the goalposts.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5
2026-08-11 15:39:42 -04:00
13 changed files with 116 additions and 688 deletions
-3
View File
@@ -129,9 +129,6 @@
"plugin_rotation_order": [],
"use_short_date_format": true,
"vegas_scroll": {
"live_in_ticker": false,
"live_weight": 3,
"favorite_live_weight": 5,
"enabled": false,
"scroll_speed": 50,
"separator_width": 32,
+1 -89
View File
@@ -64,98 +64,10 @@ JSON is optional.
| `target_fps` | `125` | Target frame rate |
| `buffer_ahead` | `2` | Number of plugins buffered ahead |
This table is a subset — `display.vegas_scroll` supports 30 keys in
This table is a subset — `display.vegas_scroll` supports 26 keys in
total. See the full list in
[CONFIG_REFERENCE.md](CONFIG_REFERENCE.md#displayvegas_scroll--continuous-scroll-mode).
### Live Content in the Ticker
By default, live content **preempts** Vegas mode: while any plugin reports
live priority, the display controller refuses to run the ticker and shows
that plugin's full-screen display instead. You get a big readable scoreboard,
but the marquee stops entirely for the duration of the game.
Set `live_in_ticker` to keep the ticker running and let live content take
**extra turns inside it** instead:
```json
"vegas_scroll": {
"live_in_ticker": true,
"live_weight": 3,
"favorite_live_weight": 5
}
```
#### Why weights exist
The rotation is otherwise a strict round robin — every plugin appears exactly
once per cycle. With a dozen plugins enabled, a live score comes round once a
lap and can be minutes old by the time you see it. A weight of *N* gives a
plugin *N* slots per cycle.
The slots are placed by **Smooth Weighted Round-Robin**, the same scheduler
the sports plugins use internally to rotate their own games. The important
property is that repeats are *spread through the cycle* rather than clumped:
three appearances in a row followed by a long silence would be worse than not
boosting at all.
Twelve plugins, with a favorite's baseball game and an ordinary live hockey
game (`live_weight: 3`, `favorite_live_weight: 5`):
```
baseball > hockey > weather > clock > baseball
stocks > news > flights > baseball > hockey
calendar > f1 > music > baseball > tides
birds > hockey > baseball
```
18 slots for 12 plugins. Baseball appears 5 times, hockey 3, everything else
once, and no plugin ever appears twice in a row — **including across the seam**
where the cycle loops back on itself. Smooth Weighted Round-Robin schedules the
heaviest item first and usually last as well, so the strip would otherwise show
it twice running at exactly the one join a within-cycle check cannot see. The
trailing repeat is moved into the widest remaining gap. Where a double is
unavoidable — a plugin holding most of the slots has to neighbour itself — the
schedule is left as it is.
#### Where the weight comes from
For each plugin in the rotation, in order:
1. **The plugin's own answer.** If it implements
`get_vegas_priority_weight()` and returns a number, that wins. This is the
only route for favorite-team awareness — the core can see *that* a game is
live, but not *whose*, so a scoreboard has to say so itself.
2. **The core's default.** When the plugin returns `None` (the base-class
default), a plugin where both `has_live_priority()` and `has_live_content()`
are true gets `live_weight`.
3. **Everything else** gets 1.
Because of step 2, **existing plugins need no changes** — any scoreboard with
`live_priority` enabled already gets extra turns. Step 1 is opt-in, for
plugins that want to distinguish a favorite's game from any other live game.
Weights are clamped to 110. A weight of 1 is no boost; a weight below 1 would
drop the plugin from the rotation entirely, which is never what is meant.
#### Things worth knowing
- **Weights are per plugin, not per game.** A scoreboard showing four live
games still occupies one slot at a time, rotating its own games within that
slot using its own `favorite_live_boost`. This controls how often the
*plugin* comes round.
- **The ticker is zero-sum.** Giving baseball 5 slots does not make the cycle
faster; it makes the cycle *longer* and everything else proportionally
rarer. If you want live scores sooner in wall-clock terms, pair this with a
smaller `plugins_per_cycle`.
- **Frequency is not freshness.** Each appearance redraws from the plugin's
current data (`refresh_updated_plugins()` drops cached content when a
plugin's data changes), but how current that data is depends on the
plugin's own `live_update_interval`. Showing a stale score five times a lap
is no better than showing it once.
- **Everything still appears.** A boost never starves another plugin out of
the cycle; low-weight plugins keep their single slot.
### Per-Plugin Configuration
Override Vegas behavior for specific plugins:
+1 -5
View File
@@ -103,8 +103,7 @@ logical image to multiple chained physical panels.
## `display.vegas_scroll` — continuous scroll mode
Read by `src/vegas_mode/config.py` (`VegasScrollConfig.from_config`). See
[ADVANCED_FEATURES.md](ADVANCED_FEATURES.md) for behavior details, including
[live content in the ticker](ADVANCED_FEATURES.md#live-content-in-the-ticker).
[ADVANCED_FEATURES.md](ADVANCED_FEATURES.md) for behavior details.
| Key | Type / default |
|---|---|
@@ -135,9 +134,6 @@ Read by `src/vegas_mode/config.py` (`VegasScrollConfig.from_config`). See
| `max_cycle_duration` | int, `240` |
| `frame_based_scrolling` | bool, `true` — frame-count-based scroll stepping |
| `scroll_delay` | float, `0.02` — seconds between scroll updates (~50 FPS) |
| `live_in_ticker` | bool, `false` — keep scrolling during live games instead of handing the display to a full-screen scoreboard |
| `live_weight` | int, `3` (110) — slots per cycle for a plugin with live content |
| `favorite_live_weight` | int, `5` (110) — slots per cycle when a plugin reports a favorite team is live |
## `sync` — multi-display synchronization
-41
View File
@@ -170,47 +170,6 @@ Default returns `False`.
List of display modes to show during a live takeover. Default returns the
plugin's `display_modes` from its manifest.
#### `get_vegas_priority_weight() -> Optional[int]`
How many slots per Vegas cycle this plugin should get. Default returns
`None`, which defers to the core.
The Vegas ticker is otherwise a strict round robin — every plugin appears
exactly once per cycle — so with a dozen plugins enabled a live score can be
minutes stale by the time it comes round. A weight of *N* gives the plugin
*N* slots per cycle, spread evenly through it rather than clumped.
**You usually do not need this.** When the hook returns `None`, the core
already gives a plugin `vegas_scroll.live_weight` whenever
`has_live_priority()` and `has_live_content()` are both true. Live sports get
extra turns with no code at all.
Implement it only when the plugin knows something the core cannot. The
motivating case is favorite teams — the core can see *that* a game is live,
but not *whose*:
```python
def get_vegas_priority_weight(self):
if not (self.has_live_priority() and self.has_live_content()):
return None # let the core decide
vegas = self.global_config.get('display', {}).get('vegas_scroll', {})
if self._favorite_is_live():
return vegas.get('favorite_live_weight', 5)
return vegas.get('live_weight', 3)
```
The weight is per *plugin*, not per game: a scoreboard showing four live games
still occupies one slot at a time and rotates its own games within it. Values
are clamped to 110 by the caller. An exception here is caught and logged, and
the core then falls back to its own live-content check — so a plugin whose
weight calculation is broken still gets `live_weight` for a game that really
is live, rather than being demoted to 1.
Only consulted when the user has set `vegas_scroll.live_in_ticker`. With the
default (`false`) live content preempts Vegas entirely and there is no ticker
to be weighted within. See
[ADVANCED_FEATURES.md](ADVANCED_FEATURES.md#live-content-in-the-ticker).
### Vegas scroll hooks
Vegas mode shows multiple plugins as a single continuous scroll instead of
+19 -1
View File
@@ -44,6 +44,24 @@ class BaseOddsManager:
self.config_manager = config_manager
self.logger = logging.getLogger(__name__)
self.base_url = "https://sports.core.api.espn.com/v2/sports"
# This path used a bare requests.get, so it identified itself as
# python-requests/x.y -- the one thing ESPN is known to reject. Around
# 2026-08-04 it began 403ing browser strings and bare custom tokens
# alike; what it accepts is a token with a URL that says who is
# calling. Every other ESPN caller in the tree already sends this
# (src/common/api_helper.py, src/base_classes/data_sources.py); the
# odds path was simply missed, and it is the one whose failures cost
# the caller its whole update budget.
#
# Deliberately no retry adapter, unlike api_helper: retries multiply
# request_timeout, which is set to 5s precisely to stay inside that
# budget. One try, then the cooldown below.
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)',
'Accept': 'application/json',
})
# Configuration with defaults
self.update_interval = 3600 # 1 hour default
@@ -144,7 +162,7 @@ class BaseOddsManager:
url = f"{self.base_url}/{sport}/leagues/{espn_league}/events/{event_id}/competitions/{event_id}/odds"
self.logger.info(f"Requesting odds from URL: {url}")
response = requests.get(url, timeout=self.request_timeout)
response = self.session.get(url, timeout=self.request_timeout)
response.raise_for_status()
raw_data = response.json()
+2 -18
View File
@@ -1638,12 +1638,6 @@ class DisplayController:
logger.warning("Error checking live priority for %s: %s", mode_name, e)
return live
def _vegas_keeps_live_in_ticker(self) -> bool:
"""Whether live content should stay in the ticker instead of preempting it."""
coordinator = getattr(self, 'vegas_coordinator', None)
config = getattr(coordinator, 'vegas_config', None)
return bool(getattr(config, 'live_in_ticker', False))
def _check_live_priority(self, advance=False):
"""Return the live-priority mode to display, or None if nothing is live.
@@ -1857,24 +1851,14 @@ class DisplayController:
# Check for live priority content and switch to it immediately.
# advance=True so multiple simultaneously-live games take turns
# (round-robin) instead of pinning to the first plugin.
# Skipped when the ticker is keeping live content: switching
# the rotation underneath Vegas would move current_mode_index
# and stash a resume point for a takeover that never happens.
if (not self.on_demand_active and not wifi_status_data
and not (self._is_vegas_mode_active()
and self._vegas_keeps_live_in_ticker())):
if not self.on_demand_active and not wifi_status_data:
live_priority_mode = self._check_live_priority(advance=True)
self._apply_live_priority(live_priority_mode)
# Vegas scroll mode - continuous ticker across all plugins
# Priority: on-demand > wifi-status > live-priority > vegas > normal rotation
if self._is_vegas_mode_active() and not wifi_status_data:
# Live content normally preempts the ticker entirely. With
# vegas_scroll.live_in_ticker the marquee keeps running and
# the live plugin takes extra turns inside it instead --
# see StreamManager._apply_priority_weights.
live_mode = (None if self._vegas_keeps_live_in_ticker()
else self._check_live_priority())
live_mode = self._check_live_priority()
if not live_mode:
try:
# Run Vegas mode iteration
-42
View File
@@ -555,48 +555,6 @@ class BasePlugin(ABC):
"""
return False
def get_vegas_priority_weight(self) -> Optional[int]:
"""How many slots per Vegas cycle this plugin should get, or None.
The Vegas ticker is otherwise a strict round robin: every plugin
appears exactly once per cycle. With a dozen plugins enabled that puts
minutes between a live score and its next appearance. A weight of N
gives the plugin N slots per cycle, spread evenly through it rather
than clumped together.
Return ``None`` (the default) to let the core decide. It gives a
plugin ``vegas_scroll.live_weight`` when ``has_live_priority()`` and
``has_live_content()`` are both true, and 1 otherwise -- so live sports
already get extra turns without implementing this at all.
Implement it only when the plugin knows something the core cannot. The
motivating case is favorite teams: the core can see *that* a game is
live but not *whose*, so a scoreboard that wants its favorite's game
shown more often than other live games has to say so::
def get_vegas_priority_weight(self):
if not (self.has_live_priority() and self.has_live_content()):
return None # let the core decide
cfg = self.global_config.get('display', {}).get('vegas_scroll', {})
if self._favorite_is_live():
return cfg.get('favorite_live_weight', 5)
return cfg.get('live_weight', 3)
The weight is per *plugin*, not per game. A scoreboard showing four
live games still occupies one slot at a time and rotates its own games
within that slot; this controls how often the plugin itself comes
round.
Raising is safe: the core logs it and falls back to its own
live-content check, so a broken weight calculation costs the plugin
the favorite distinction but not the live boost.
Returns:
Slots per cycle (clamped to 1..10 by the caller), or None to
defer to the core's own live-content weighting.
"""
return None
def get_live_modes(self) -> List[str]:
"""
Get list of display modes that should be used during live priority takeover.
-44
View File
@@ -125,32 +125,6 @@ class VegasModeConfig:
plugin_order: List[str] = field(default_factory=list)
excluded_plugins: Set[str] = field(default_factory=set)
# --- Live content in the ticker -------------------------------------
#
# By default a live game preempts Vegas entirely: the display controller
# refuses to run the ticker while any plugin reports live priority, and you
# get the full-screen scoreboard instead. Set live_in_ticker to keep the
# marquee running and let live content take extra turns within it.
#
# The rotation is otherwise a strict round robin -- every plugin appears
# exactly once per cycle -- so with a dozen plugins enabled a live score
# comes round once a lap and can be minutes old on screen. Weighting lets a
# plugin claim several slots per cycle instead.
#
# Weights are per plugin, not per game: a scoreboard showing four live
# games still occupies one slot at a time, and rotates its own games within
# that slot using its own favorite_live_boost.
live_in_ticker: bool = False
# Slots per cycle for a plugin reporting live content. 1 disables the boost
# and restores the plain round robin.
live_weight: int = 3
# Slots per cycle for a plugin whose live content involves a favorite team.
# Only plugins implementing get_vegas_priority_weight() can claim this --
# the core cannot tell whose game is on, so the plugin reports it.
favorite_live_weight: int = 5
# Performance settings
target_fps: int = 125 # Target frame rate
buffer_ahead: int = 2 # Number of plugins to buffer ahead
@@ -201,12 +175,6 @@ class VegasModeConfig:
overflow_mode=str(vegas_config.get('overflow_mode', 'rotate')),
plugin_order=list(vegas_config.get('plugin_order', [])),
excluded_plugins=set(vegas_config.get('excluded_plugins', [])),
live_in_ticker=bool(vegas_config.get('live_in_ticker', False)),
# Clamped: a weight below 1 would drop the plugin from the rotation
# entirely, and a very large one starves everything else.
live_weight=max(1, min(10, int(vegas_config.get('live_weight', 3)))),
favorite_live_weight=max(
1, min(10, int(vegas_config.get('favorite_live_weight', 5)))),
target_fps=int(vegas_config.get('target_fps', 125)),
buffer_ahead=int(vegas_config.get('buffer_ahead', 2)),
frame_based_scrolling=vegas_config.get('frame_based_scrolling', True),
@@ -236,9 +204,6 @@ class VegasModeConfig:
'lead_in_width': self.lead_in_width,
'plugins_per_cycle': self.plugins_per_cycle,
'max_plugin_width_ratio': self.max_plugin_width_ratio,
'live_in_ticker': self.live_in_ticker,
'live_weight': self.live_weight,
'favorite_live_weight': self.favorite_live_weight,
'overflow_mode': self.overflow_mode,
'plugin_order': self.plugin_order,
'excluded_plugins': list(self.excluded_plugins),
@@ -406,15 +371,6 @@ class VegasModeConfig:
if 'enabled' in vegas_config:
self.enabled = vegas_config['enabled']
if 'live_in_ticker' in vegas_config:
self.live_in_ticker = bool(vegas_config['live_in_ticker'])
# Clamped exactly as from_config does: a weight below 1 would drop the
# plugin from the rotation, and a huge one starves everything else.
if 'live_weight' in vegas_config:
self.live_weight = max(1, min(10, int(vegas_config['live_weight'])))
if 'favorite_live_weight' in vegas_config:
self.favorite_live_weight = max(
1, min(10, int(vegas_config['favorite_live_weight'])))
if 'scroll_speed' in vegas_config:
self.scroll_speed = float(vegas_config['scroll_speed'])
if 'separator_width' in vegas_config:
-6
View File
@@ -497,12 +497,6 @@ class VegasModeCoordinator:
if not self._live_priority_check:
return False
if self.vegas_config.live_in_ticker:
# The ticker keeps live content rather than yielding to it; the
# extra turns are arranged in the rotation itself, so there is
# nothing to pause for.
return False
try:
live_mode = self._live_priority_check()
if live_mode:
-126
View File
@@ -406,8 +406,6 @@ class StreamManager:
)
logger.info("Ordered plugins: %s", ordered_plugins)
ordered_plugins = self._apply_priority_weights(ordered_plugins)
# Atomically update shared state under lock to avoid races with prefetchers
with self._buffer_lock:
self._ordered_plugins = ordered_plugins
@@ -419,130 +417,6 @@ class StreamManager:
logger.info("=" * 60)
def _plugin_weight(self, plugin_id: str) -> int:
"""Slots per cycle for one plugin.
A plugin may answer for itself via get_vegas_priority_weight() -- the
only way favorite-team awareness can reach here, since the core can see
that a game is live but not whose. When it declines (returns None, the
default), live content earns ``live_weight`` and everything else 1.
"""
plugin = None
try:
plugin = self.plugin_manager.plugins.get(plugin_id)
except (AttributeError, TypeError):
return 1
if plugin is None:
return 1
try:
if hasattr(plugin, 'get_vegas_priority_weight'):
declared = plugin.get_vegas_priority_weight()
if declared is not None:
return max(1, min(10, int(declared)))
except Exception:
# Deliberately falls through to the core's own live check rather
# than demoting to 1. The plugin's weight calculation is broken,
# but has_live_priority() and has_live_content() are separate
# methods guarded separately below -- a plugin that genuinely has
# a live game should still get live_weight for it.
logger.exception("[%s] get_vegas_priority_weight() failed", plugin_id)
try:
if (hasattr(plugin, 'has_live_priority')
and hasattr(plugin, 'has_live_content')
and plugin.has_live_priority()
and plugin.has_live_content()):
return self.config.live_weight
except Exception:
logger.exception("[%s] live-content check failed", plugin_id)
return 1
def _apply_priority_weights(self, ordered: List[str]) -> List[str]:
"""Expand the rotation so weighted plugins take several turns per cycle.
Smooth Weighted Round-Robin, the same scheduler the sports plugins use
to rotate their own games: a plugin of weight N appears N times per
cycle, and the repeats are spaced through the cycle rather than
clumped, so a live score is never three-in-a-row followed by a long
silence.
Returns the input unchanged when nothing is weighted, which is both the
common case and the pre-existing behaviour.
"""
if not ordered or not self.config.live_in_ticker:
return ordered
weights = {pid: self._plugin_weight(pid) for pid in ordered}
total = sum(weights.values())
if total <= len(ordered):
return ordered # nothing boosted; plain round robin
current = {pid: 0 for pid in ordered}
schedule: List[str] = []
for _ in range(total):
for pid in ordered:
current[pid] += weights[pid]
picked = max(current, key=lambda p: current[p])
current[picked] -= total
schedule.append(picked)
schedule = self._unclump_seam(schedule)
boosted = {p: w for p, w in weights.items() if w > 1}
logger.info(
"Vegas rotation weighted: %d slots for %d plugins (boosted: %s)",
len(schedule), len(ordered), boosted)
return schedule
@staticmethod
def _unclump_seam(schedule: List[str]) -> List[str]:
"""Stop the heaviest plugin sitting on both ends of the cycle.
Smooth Weighted Round-Robin spaces repeats well *within* a pass, but
it schedules the heaviest item first and often last too. The strip
loops, so those two are neighbours: the one place the marquee shows
the same plugin twice running is the seam between cycles.
Rotating the list cannot fix this. Rotation preserves the cyclic order
exactly, so it only moves where the seam is drawn, not the adjacency
itself. The trailing entry has to be swapped with one from the middle
whose neighbours differ from it, which breaks the pair without
creating another.
Left alone when no such position exists -- a rotation short enough or
lopsided enough to have none is one where the plugin is unavoidably
adjacent to itself anyway.
"""
if len(schedule) < 3 or schedule[0] != schedule[-1]:
return schedule
repeated = schedule[-1]
size = len(schedule)
elsewhere = [i for i, p in enumerate(schedule[:-1]) if p == repeated]
def clearance(j: int) -> int:
"""Cyclic distance from j to the nearest other appearance."""
return min(min((i - j) % size, (j - i) % size) for i in elsewhere)
candidates = [
j for j in range(1, size - 1)
if schedule[j] != repeated
and schedule[j - 1] != repeated
and schedule[j + 1] != repeated
]
if not candidates:
return schedule
# Drop it into the widest gap rather than the first slot that fits.
# Taking the first one undoes the spacing this whole function exists
# to protect: on a 28-slot rotation it moved a repeat from a gap of 7
# to a gap of 2, which is more clumped than the seam ever was.
best = max(candidates, key=clearance) if elsewhere else candidates[0]
schedule = list(schedule)
schedule[best], schedule[-1] = schedule[-1], schedule[best]
return schedule
def _prefetch_content(self, count: int = 1) -> None:
"""
Prefetch content for upcoming plugins.
+4 -2
View File
@@ -8,7 +8,9 @@ is_odds_available's ML-blind truth table, the fixed format_odds_summary
gate (money-line-only odds now format), get_odds_for_games, and
configuration loading.
No real network: src.base_odds_manager.requests.get is always patched.
No real network: requests.Session.get is always patched. The odds path sends
its requests through a session so it can identify itself to ESPN, so patching
the module-level requests.get would no longer intercept anything.
"""
from unittest.mock import MagicMock, patch
@@ -59,7 +61,7 @@ def manager(cache_manager):
@pytest.fixture
def mock_get():
with patch('src.base_odds_manager.requests.get') as m:
with patch('src.base_odds_manager.requests.Session.get') as m:
m.return_value = _make_response({'items': [dict(FULL_ITEM)]})
yield m
+89 -49
View File
@@ -10,10 +10,15 @@ and the update carrying every game's score was killed:
Invisible out of season -- preseason week 1 returns a single game -- and a
Sunday slate is around sixteen.
The request now goes through a session that identifies the caller, so the
tests patch `manager.session.get` rather than the module's `requests.get`.
"""
from unittest.mock import Mock
import requests
from src.base_odds_manager import BaseOddsManager
PLUGIN_BUDGET = 30.0 # PluginExecutor(default_timeout=30.0)
@@ -25,43 +30,83 @@ def _manager(cache=None):
return BaseOddsManager(cache_manager=cache, config_manager=None)
def _timing_out(manager):
"""Point the manager's session at a request that always times out."""
manager.session.get = Mock(side_effect=requests.exceptions.Timeout("x"))
return manager.session.get
def _returning(manager, payload):
resp = Mock()
resp.json.return_value = payload
resp.raise_for_status.return_value = None
manager.session.get = Mock(return_value=resp)
return manager.session.get
class TestRequestTimeout:
def test_leaves_room_in_the_operation_budget(self):
assert _manager().request_timeout < PLUGIN_BUDGET / 2
def test_the_timeout_is_the_one_actually_used(self):
m = _manager()
import src.base_odds_manager as mod
real = mod.requests.get
try:
mod.requests.get = Mock(side_effect=mod.requests.exceptions.Timeout("x"))
m.get_odds("football", "nfl", "401")
assert mod.requests.get.call_args.kwargs["timeout"] == m.request_timeout
finally:
mod.requests.get = real
get = _timing_out(m)
m.get_odds("football", "nfl", "401")
assert get.call_args.kwargs["timeout"] == m.request_timeout
class TestIdentifiesItselfToEspn:
"""ESPN 403s python-requests' default agent, and bare custom tokens.
What it accepts is a token carrying a URL that says who is calling. This
path used a bare requests.get and so sent the default -- the one thing
known to be rejected. Everything else in the tree that talks to ESPN
already sends the header below.
"""
def test_the_user_agent_names_the_project_and_links_to_it(self):
ua = _manager().session.headers["User-Agent"]
assert "python-requests" not in ua
assert "LEDMatrix" in ua
assert "github.com/ChuckBuilds/LEDMatrix" in ua
def test_it_is_the_same_agent_the_rest_of_the_tree_sends(self):
# Compared against the live value rather than a copied literal, so the
# two cannot drift apart the next time ESPN moves the goalposts.
from src.common.api_helper import APIHelper
assert (_manager().session.headers["User-Agent"]
== APIHelper().session.headers["User-Agent"])
def test_the_header_reaches_the_request(self):
m = _manager()
get = _returning(m, {})
m._extract_espn_data = Mock(return_value=None)
m.get_odds("football", "nfl", "401")
# Sent via the session, so it applies without being passed per-call.
assert get.call_count == 1
assert "User-Agent" in m.session.headers
def test_no_retry_adapter_multiplies_the_timeout(self):
# api_helper mounts a retrying adapter; this path must not, or a 5s
# timeout becomes 15s and the budget fix is undone.
m = _manager()
for adapter in m.session.adapters.values():
retries = getattr(adapter, "max_retries", None)
assert getattr(retries, "total", 0) in (0, None), (
"odds session mounts a retrying adapter (total=%r); retries "
"multiply request_timeout" % getattr(retries, "total", None))
class TestSlowEspnCannotKillTheUpdate:
def test_one_failure_stops_the_rest_of_the_slate_hitting_the_network(self):
m = _manager()
import src.base_odds_manager as mod
real = mod.requests.get
calls = {"n": 0}
get = _timing_out(m)
for i in range(16): # a full slate, one game at a time
m.get_odds("football", "nfl", "4018730%02d" % i)
def timeout(*a, **k):
calls["n"] += 1
raise mod.requests.exceptions.Timeout("timed out")
try:
mod.requests.get = timeout
for i in range(16): # a full slate, one game at a time
m.get_odds("football", "nfl", "4018730%02d" % i)
finally:
mod.requests.get = real
assert calls["n"] == 1, (
assert get.call_count == 1, (
"%d games each paid the timeout; the breaker should have stopped "
"after the first" % calls["n"])
"after the first" % get.call_count)
def test_worst_case_slate_stays_inside_the_budget(self):
m = _manager()
@@ -70,53 +115,48 @@ class TestSlowEspnCannotKillTheUpdate:
def test_recovery_is_automatic(self):
m = _manager()
import src.base_odds_manager as mod
real_get, real_monotonic = mod.requests.get, mod.time.monotonic
real_monotonic = mod.time.monotonic
clock = {"t": 1000.0}
try:
mod.time.monotonic = lambda: clock["t"]
mod.requests.get = Mock(
side_effect=mod.requests.exceptions.Timeout("timed out"))
get = _timing_out(m)
m.get_odds("football", "nfl", "401")
assert m._skip_network_until > clock["t"], "breaker did not open"
clock["t"] += 1
before = mod.requests.get.call_count
before = get.call_count
m.get_odds("football", "nfl", "402")
assert mod.requests.get.call_count == before, "should not have retried"
assert get.call_count == before, "should not have retried"
clock["t"] += m._FAILURE_COOLDOWN
m.get_odds("football", "nfl", "403")
assert mod.requests.get.call_count > before, "never retried"
assert get.call_count > before, "never retried"
finally:
mod.requests.get, mod.time.monotonic = real_get, real_monotonic
mod.time.monotonic = real_monotonic
def test_a_healthy_fetch_clears_the_breaker(self):
m = _manager()
m._skip_network_until = 0.0
m._extract_espn_data = Mock(return_value=None)
import src.base_odds_manager as mod
real = mod.requests.get
try:
resp = Mock()
resp.json.return_value = {}
resp.raise_for_status.return_value = None
mod.requests.get = Mock(return_value=resp)
m.get_odds("football", "nfl", "401")
finally:
mod.requests.get = real
_returning(m, {})
m.get_odds("football", "nfl", "401")
assert m._skip_network_until == 0.0
def test_a_403_opens_the_breaker_rather_than_hammering(self):
# raise_for_status raises HTTPError, a RequestException -- so a wrong
# or missing agent backs off instead of 403ing once per game.
m = _manager()
resp = Mock()
resp.raise_for_status.side_effect = requests.exceptions.HTTPError("403")
m.session.get = Mock(return_value=resp)
m.get_odds("football", "nfl", "401")
assert m._skip_network_until > 0.0
def test_the_stale_cache_fallback_still_works(self):
# The failing request must still hand back whatever was cached; only
# the *subsequent* games skip the network.
cache = Mock()
cache.get_with_auto_strategy.side_effect = [None, {"details": "stale"}]
m = BaseOddsManager(cache_manager=cache, config_manager=None)
import src.base_odds_manager as mod
real = mod.requests.get
try:
mod.requests.get = Mock(
side_effect=mod.requests.exceptions.Timeout("timed out"))
assert m.get_odds("football", "nfl", "401") == {"details": "stale"}
finally:
mod.requests.get = real
_timing_out(m)
assert m.get_odds("football", "nfl", "401") == {"details": "stale"}
-262
View File
@@ -1,262 +0,0 @@
"""Tests that live content can take extra turns inside the Vegas ticker.
Vegas was a strict round robin -- every plugin exactly once per cycle -- and
live content did not appear in it at all, because the display controller
refused to run the ticker while anything was live. With a dozen plugins
enabled that left a live score either absent or minutes stale.
Two things change, both off by default. `live_in_ticker` keeps the marquee
running instead of yielding to a full-screen takeover, and the rotation is
expanded by Smooth Weighted Round-Robin so a weighted plugin gets several
slots per cycle, spaced through it rather than clumped.
Weights are per plugin, not per game: a scoreboard showing four live games
still occupies one slot at a time and rotates its own games within it.
"""
from unittest.mock import Mock
import pytest
from src.vegas_mode.config import VegasModeConfig
from src.vegas_mode.stream_manager import StreamManager
class FakePlugin:
"""A plugin that can fail in each place independently.
hook_raises and live_raises are separate because they mean different
things: a broken weight calculation should still leave the core's own
live-content check usable, while a plugin that cannot answer whether it is
live at all has nothing left to fall back on.
"""
def __init__(self, live=False, declared=None, raises=False,
hook_raises=False, live_raises=False):
self._live = live
self._declared = declared
self._hook_raises = hook_raises or raises
self._live_raises = live_raises or raises
self.enabled = True
def has_live_priority(self):
if self._live_raises:
raise RuntimeError("cannot say whether I am live")
return self._live
def has_live_content(self):
return self._live
def get_vegas_priority_weight(self):
if self._hook_raises:
raise RuntimeError("weight calculation blew up")
return self._declared
def _manager(plugins, **cfg):
config = VegasModeConfig(live_in_ticker=cfg.pop('live_in_ticker', True), **cfg)
pm = Mock()
pm.plugins = plugins
sm = StreamManager.__new__(StreamManager)
sm.config = config
sm.plugin_manager = pm
return sm
def _counts(schedule):
return {p: schedule.count(p) for p in set(schedule)}
def _max_gap(schedule, plugin_id):
"""Largest gap between consecutive appearances, wrapping around."""
at = [i for i, p in enumerate(schedule) if p == plugin_id]
if len(at) < 2:
return len(schedule)
gaps = [b - a for a, b in zip(at, at[1:])]
gaps.append(len(schedule) - at[-1] + at[0])
return max(gaps)
class TestWeightsComeFromTheRightPlace:
def test_a_quiet_plugin_gets_one_slot(self):
sm = _manager({'clock': FakePlugin()})
assert sm._plugin_weight('clock') == 1
def test_live_content_earns_the_configured_weight(self):
sm = _manager({'mlb': FakePlugin(live=True)}, live_weight=4)
assert sm._plugin_weight('mlb') == 4
def test_a_plugin_may_answer_for_itself(self):
# The only route for favorite-team awareness: the core can see that a
# game is live, not whose.
sm = _manager({'mlb': FakePlugin(live=True, declared=7)}, live_weight=3)
assert sm._plugin_weight('mlb') == 7
def test_declaring_none_defers_to_the_core(self):
sm = _manager({'mlb': FakePlugin(live=True, declared=None)}, live_weight=3)
assert sm._plugin_weight('mlb') == 3
def test_a_declared_weight_is_clamped(self):
sm = _manager({'a': FakePlugin(declared=99), 'b': FakePlugin(declared=0)})
assert sm._plugin_weight('a') == 10
assert sm._plugin_weight('b') == 1
def test_a_plugin_that_raises_everywhere_weighs_one(self):
sm = _manager({'bad': FakePlugin(raises=True)})
assert sm._plugin_weight('bad') == 1
def test_a_broken_hook_still_earns_the_live_boost(self):
# The hook is only how a plugin asks for *more* than live_weight.
# Losing it should cost the favorite distinction, not the live boost:
# has_live_priority/has_live_content are separate and still work.
sm = _manager({'mlb': FakePlugin(live=True, hook_raises=True)},
live_weight=4)
assert sm._plugin_weight('mlb') == 4
def test_a_broken_hook_on_a_quiet_plugin_weighs_one(self):
sm = _manager({'clock': FakePlugin(live=False, hook_raises=True)},
live_weight=4)
assert sm._plugin_weight('clock') == 1
def test_a_plugin_that_cannot_say_whether_it_is_live_weighs_one(self):
# Nothing left to fall back on, so no boost.
sm = _manager({'mlb': FakePlugin(live=True, live_raises=True)},
live_weight=4)
assert sm._plugin_weight('mlb') == 1
def test_an_unknown_plugin_weighs_one(self):
assert _manager({})._plugin_weight('ghost') == 1
class TestTheSchedule:
def test_nothing_weighted_leaves_the_order_untouched(self):
order = ['weather', 'clock', 'news']
sm = _manager({p: FakePlugin() for p in order})
assert sm._apply_priority_weights(order) == order
def test_off_by_default_the_order_is_untouched(self):
order = ['weather', 'mlb', 'news']
sm = _manager({'weather': FakePlugin(), 'mlb': FakePlugin(live=True),
'news': FakePlugin()}, live_in_ticker=False, live_weight=3)
assert sm._apply_priority_weights(order) == order
def test_a_live_plugin_takes_its_share_of_slots(self):
order = ['weather', 'mlb', 'news', 'clock']
sm = _manager({'weather': FakePlugin(), 'mlb': FakePlugin(live=True),
'news': FakePlugin(), 'clock': FakePlugin()},
live_weight=3)
schedule = sm._apply_priority_weights(order)
counts = _counts(schedule)
assert counts['mlb'] == 3, counts
assert counts['weather'] == counts['news'] == counts['clock'] == 1, counts
assert len(schedule) == 6
def test_every_plugin_still_appears(self):
# A boost must not starve anything out of the cycle.
order = ['a', 'b', 'c', 'd', 'e', 'f']
plugins = {p: FakePlugin() for p in order}
plugins['a'] = FakePlugin(live=True, declared=10)
sm = _manager(plugins)
schedule = sm._apply_priority_weights(order)
assert set(schedule) == set(order), set(order) - set(schedule)
def test_nothing_doubles_across_the_cycle_seam(self):
# The strip loops, so the last slot neighbours the first. Smooth
# Weighted Round-Robin schedules the heaviest item first and often
# last too, which put the one clump the algorithm exists to avoid at
# the one place a within-cycle check cannot see.
order = ['baseball', 'weather', 'geochron', 'flights', 'stocks',
'oftheday', 'youtube', 'stocknews', 'leaderboard',
'countdown', 'odds', 'f1', 'football', 'music']
plugins = {p: FakePlugin() for p in order}
plugins['baseball'] = FakePlugin(live=True, declared=5)
plugins['football'] = FakePlugin(live=True, declared=3)
schedule = _manager(plugins)._apply_priority_weights(order)
n = len(schedule)
doubles = [schedule[i] for i in range(n)
if schedule[i] == schedule[(i + 1) % n]]
assert not doubles, "%r repeats across the seam in %r" % (doubles, schedule)
def test_the_seam_repair_keeps_every_slot(self):
order = ['a', 'b', 'c', 'd', 'e', 'f']
plugins = {p: FakePlugin() for p in order}
plugins['a'] = FakePlugin(live=True, declared=4)
schedule = _manager(plugins)._apply_priority_weights(order)
assert _counts(schedule)['a'] == 4, _counts(schedule)
assert sorted(schedule) == sorted(
['a'] * 4 + ['b', 'c', 'd', 'e', 'f']), schedule
def test_the_repair_uses_the_widest_gap(self):
# Moving the trailing repeat into the first slot that merely fits
# undoes the spacing: on a 28-slot rotation that turned a gap of 7
# into a gap of 2, which is more clumped than the seam ever was.
order = ['a'] + ['p%d' % i for i in range(13)]
plugins = {p: FakePlugin() for p in order}
plugins['a'] = FakePlugin(live=True, declared=4)
schedule = _manager(plugins)._apply_priority_weights(order)
at = [i for i, p in enumerate(schedule) if p == 'a']
gaps = [b - a for a, b in zip(at, at[1:])]
gaps.append(len(schedule) - at[-1] + at[0])
ideal = len(schedule) / len(at)
assert min(gaps) >= ideal / 2, "gaps %r for ideal %.1f" % (gaps, ideal)
def test_an_unavoidable_double_is_left_alone(self):
# Five of seven slots are the same plugin, so it must neighbour
# itself. Better to schedule it than to refuse or loop forever.
order = ['a', 'b', 'c']
plugins = {p: FakePlugin() for p in order}
plugins['a'] = FakePlugin(live=True, declared=5)
schedule = _manager(plugins)._apply_priority_weights(order)
assert _counts(schedule) == {'a': 5, 'b': 1, 'c': 1}, _counts(schedule)
assert set(schedule) == {'a', 'b', 'c'}
def test_a_schedule_too_short_to_repair_is_returned_as_is(self):
sm = _manager({})
assert sm._unclump_seam(['a', 'a']) == ['a', 'a']
assert sm._unclump_seam(['a']) == ['a']
assert sm._unclump_seam([]) == []
def test_a_schedule_with_no_seam_clash_is_untouched(self):
sm = _manager({})
plain = ['a', 'b', 'c', 'a', 'd']
assert sm._unclump_seam(plain) == plain
def test_repeats_are_spread_not_clumped(self):
# The point of Smooth Weighted Round-Robin. Three-in-a-row followed by
# a long silence would be worse than not boosting at all.
order = ['weather', 'mlb', 'news', 'clock', 'stocks', 'f1']
plugins = {p: FakePlugin() for p in order}
plugins['mlb'] = FakePlugin(live=True)
sm = _manager(plugins, live_weight=3)
schedule = sm._apply_priority_weights(order)
assert _counts(schedule)['mlb'] == 3
# Evenly spread over 8 slots means a gap of about 3, never 6.
assert _max_gap(schedule, 'mlb') <= 4, schedule
# And never twice running.
assert not any(a == b == 'mlb' for a, b in zip(schedule, schedule[1:])), schedule
def test_a_favorite_outranks_another_live_game(self):
order = ['weather', 'mlb', 'nhl']
sm = _manager({'weather': FakePlugin(),
'mlb': FakePlugin(live=True, declared=5),
'nhl': FakePlugin(live=True)}, live_weight=2)
counts = _counts(sm._apply_priority_weights(order))
assert counts['mlb'] == 5 and counts['nhl'] == 2 and counts['weather'] == 1, counts
def test_an_empty_rotation_is_harmless(self):
assert _manager({})._apply_priority_weights([]) == []
class TestConfigParsing:
def test_defaults_preserve_todays_behaviour(self):
cfg = VegasModeConfig.from_config({})
assert cfg.live_in_ticker is False
assert cfg.live_weight == 3 and cfg.favorite_live_weight == 5
@pytest.mark.parametrize("given,expected", [(0, 1), (-4, 1), (99, 10), (4, 4)])
def test_weights_are_clamped(self, given, expected):
cfg = VegasModeConfig.from_config(
{'display': {'vegas_scroll': {'live_weight': given}}})
assert cfg.live_weight == expected