diff --git a/test/test_api_helper.py b/test/test_api_helper.py new file mode 100644 index 00000000..b3c24e99 --- /dev/null +++ b/test/test_api_helper.py @@ -0,0 +1,275 @@ +""" +Tests for src/common/api_helper.py (APIHelper). + +Covers rate limiting, cached GETs, ESPN URL/cache-key construction, +session header defaults and per-call merging, the retry adapter, and the +fixed clear_cache() behavior (real CacheManager surface: clear_cache / +delete / list_cache_files, with safe no-ops elsewhere). + +No real network: helper.session.get/post are always replaced with mocks. +""" + +import types +from unittest.mock import MagicMock, Mock + +import pytest +import requests +from freezegun import freeze_time + +import src.common.api_helper as api_helper_module +from src.common.api_helper import APIHelper + + +def _make_response(payload): + response = MagicMock() + response.json.return_value = payload + response.raise_for_status.return_value = None + return response + + +@pytest.fixture +def cache(): + cache = MagicMock() + cache.get.return_value = None + return cache + + +@pytest.fixture +def helper(cache): + helper = APIHelper(cache_manager=cache) + # Default min interval is 1.0s and would really sleep between requests. + helper.set_rate_limit(0) + return helper + + +# --------------------------------------------------------------------------- +# Rate limiting +# --------------------------------------------------------------------------- + +class TestRateLimiting: + def test_sleeps_for_remaining_interval(self, helper, monkeypatch): + fake_time = MagicMock() + fake_time.time.side_effect = [102.0, 105.0] + monkeypatch.setattr(api_helper_module, 'time', fake_time) + + helper.set_rate_limit(5) + helper._last_request_time = 100.0 + helper._enforce_rate_limit() + + # 2s elapsed of a 5s interval -> sleep the remaining 3s. + fake_time.sleep.assert_called_once() + assert fake_time.sleep.call_args[0][0] == pytest.approx(3.0) + assert helper._last_request_time == 105.0 + + def test_no_sleep_when_interval_elapsed(self, helper, monkeypatch): + fake_time = MagicMock() + fake_time.time.side_effect = [200.0, 201.0] + monkeypatch.setattr(api_helper_module, 'time', fake_time) + + helper.set_rate_limit(5) + helper._last_request_time = 100.0 + helper._enforce_rate_limit() + + fake_time.sleep.assert_not_called() + assert helper._last_request_time == 201.0 + + +# --------------------------------------------------------------------------- +# get() +# --------------------------------------------------------------------------- + +class TestGet: + def test_cache_hit_skips_request_and_rate_limit(self, helper, cache): + cache.get.return_value = {'cached': True} + helper.session.get = Mock() + rate_spy = Mock() + helper._enforce_rate_limit = rate_spy + + result = helper.get('https://example.com/api', cache_key='k') + + assert result == {'cached': True} + helper.session.get.assert_not_called() + rate_spy.assert_not_called() + + def test_cache_miss_fetches_and_caches_without_ttl(self, helper, cache): + cache.get.return_value = None + helper.session.get = Mock(return_value=_make_response({'a': 1})) + + result = helper.get('https://example.com/api', cache_key='k', + cache_ttl=999) + + assert result == {'a': 1} + # Pin the ttl-dropped contract: CacheManager.set is called with + # (key, data) only — the cache_ttl argument is discarded. + cache.set.assert_called_once_with('k', {'a': 1}) + + def test_request_exception_returns_none_and_caches_nothing( + self, helper, cache): + helper.session.get = Mock( + side_effect=requests.exceptions.RequestException('boom')) + + result = helper.get('https://example.com/api', cache_key='k') + + assert result is None + cache.set.assert_not_called() + + def test_timeout_zero_falls_back_to_default(self, helper): + # Quirk pin: `timeout or self.default_timeout` treats an explicit + # timeout=0 as falsy, so the default (30) is used instead. + helper.session.get = Mock(return_value=_make_response({})) + + helper.get('https://example.com/api', timeout=0) + + assert helper.session.get.call_args.kwargs['timeout'] == 30 + + def test_per_call_headers_merge_over_session_headers(self, helper): + helper.session.get = Mock(return_value=_make_response({})) + + helper.get('https://example.com/api', headers={'X-Custom': 'yes'}) + + sent = helper.session.get.call_args.kwargs['headers'] + # Merged, not replaced: session defaults survive alongside the + # per-call header. + assert sent['X-Custom'] == 'yes' + assert sent['User-Agent'] == ( + 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)') + assert sent['Accept'] == 'application/json' + # The session's own headers are not polluted by the per-call ones. + assert 'X-Custom' not in helper.session.headers + + +# --------------------------------------------------------------------------- +# ESPN helpers +# --------------------------------------------------------------------------- + +class TestEspnHelpers: + @freeze_time('2026-08-07') + def test_fetch_espn_scoreboard_url_params_and_cache_key(self, helper): + helper.get = Mock(return_value={'ok': 1}) + + result = helper.fetch_espn_scoreboard('football', 'nfl') + + assert result == {'ok': 1} + helper.get.assert_called_once_with( + 'https://site.api.espn.com/apis/site/v2/sports/football/nfl/scoreboard', + params={'dates': '20260807', 'limit': 1000}, + cache_key='espn_football_nfl_20260807', + cache_ttl=300, + ) + + def test_fetch_espn_scoreboard_explicit_date(self, helper): + helper.get = Mock(return_value=None) + + helper.fetch_espn_scoreboard('basketball', 'nba', date='20250115') + + kwargs = helper.get.call_args.kwargs + assert kwargs['params'] == {'dates': '20250115', 'limit': 1000} + assert kwargs['cache_key'] == 'espn_basketball_nba_20250115' + + def test_fetch_espn_standings_url_and_cache_key(self, helper): + helper.get = Mock(return_value={'ok': 1}) + + helper.fetch_espn_standings('football', 'nfl') + + helper.get.assert_called_once_with( + 'https://site.api.espn.com/apis/site/v2/sports/football/nfl/standings', + cache_key='espn_standings_football_nfl', + cache_ttl=3600, + ) + + def test_fetch_espn_rankings_url_and_cache_key(self, helper): + helper.get = Mock(return_value={'ok': 1}) + + helper.fetch_espn_rankings('football', 'college-football') + + helper.get.assert_called_once_with( + 'https://site.api.espn.com/apis/site/v2/sports/football/college-football/rankings', + cache_key='espn_rankings_football_college-football', + cache_ttl=3600, + ) + + +# --------------------------------------------------------------------------- +# Session setup +# --------------------------------------------------------------------------- + +class TestSessionSetup: + def test_user_agent_exact(self, helper): + # Regression guard: ESPN began 403ing other user agents; this exact + # string must be sent on every request. + assert helper.session.headers['User-Agent'] == ( + 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)') + + def test_retry_adapter_configuration(self): + helper = APIHelper(cache_manager=None, max_retries=7) + + retries = helper.session.get_adapter('https://x').max_retries + assert retries.total == 7 + assert {429, 500, 502, 503, 504} <= set(retries.status_forcelist) + + +# --------------------------------------------------------------------------- +# clear_cache (fixed behavior: real CacheManager surface) +# --------------------------------------------------------------------------- + +class TestClearCache: + def test_no_pattern_uses_clear_cache_method(self): + manager = types.SimpleNamespace(clear_cache=Mock()) + helper = APIHelper(cache_manager=manager) + helper.set_rate_limit(0) + + helper.clear_cache() + + manager.clear_cache.assert_called_once_with() + + def test_no_pattern_falls_back_to_clear(self): + manager = types.SimpleNamespace(clear=Mock()) + helper = APIHelper(cache_manager=manager) + helper.set_rate_limit(0) + + helper.clear_cache() + + manager.clear.assert_called_once_with() + + def test_no_pattern_manager_without_any_clear_is_noop(self): + helper = APIHelper(cache_manager=object()) + helper.set_rate_limit(0) + + helper.clear_cache() # must not raise + + def test_pattern_deletes_only_matching_keys(self): + manager = types.SimpleNamespace( + list_cache_files=Mock(return_value=[ + {'key': 'espn_nfl_x'}, + {'key': 'other'}, + ]), + delete=Mock(), + ) + helper = APIHelper(cache_manager=manager) + helper.set_rate_limit(0) + + helper.clear_cache(pattern='espn') + + manager.delete.assert_called_once_with('espn_nfl_x') + + def test_pattern_manager_without_list_cache_files_is_noop(self): + helper = APIHelper(cache_manager=object()) + helper.set_rate_limit(0) + + helper.clear_cache(pattern='espn') # must not raise + + +# --------------------------------------------------------------------------- +# No cache manager +# --------------------------------------------------------------------------- + +class TestNoCacheManager: + def test_all_cache_operations_safe_without_manager(self): + helper = APIHelper(cache_manager=None) + helper.set_rate_limit(0) + + assert helper.get_cache('k') is None + assert helper._get_from_cache('k') is None + assert helper.set_cache('k', {'a': 1}) is None + assert helper.clear_cache() is None + assert helper.clear_cache(pattern='espn') is None diff --git a/test/test_base_odds_manager.py b/test/test_base_odds_manager.py new file mode 100644 index 00000000..b08183ad --- /dev/null +++ b/test/test_base_odds_manager.py @@ -0,0 +1,359 @@ +""" +Tests for src/base_odds_manager.py (BaseOddsManager). + +Covers get_odds validation/caching/URL construction, the null-safe +_extract_espn_data fix (ESPN sends explicit JSON nulls for absent sides), +the no_odds sentinel, stale-cache fallback on request failure, +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. +""" + +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from src.base_odds_manager import BaseOddsManager + + +FULL_ITEM = { + 'details': 'DAL -3.5', + 'overUnder': 47.5, + 'spread': -3.5, + 'homeTeamOdds': {'moneyLine': -150, 'current': {'pointSpread': {'value': -3.5}}}, + 'awayTeamOdds': {'moneyLine': 130, 'current': {'pointSpread': {'value': 3.5}}}, +} + +FULL_EXTRACTED = { + 'details': 'DAL -3.5', + 'over_under': 47.5, + 'spread': -3.5, + 'home_team_odds': {'money_line': -150, 'spread_odds': -3.5}, + 'away_team_odds': {'money_line': 130, 'spread_odds': 3.5}, +} + + +def _make_response(payload): + response = MagicMock() + response.json.return_value = payload + response.raise_for_status.return_value = None + return response + + +@pytest.fixture +def cache_manager(): + cm = MagicMock() + # A bare MagicMock returns truthy Mocks from every call, so every + # get_odds() would look like a cache hit. Explicitly wire a miss. + cm.get_with_auto_strategy.return_value = None + return cm + + +@pytest.fixture +def manager(cache_manager): + return BaseOddsManager(cache_manager) + + +@pytest.fixture +def mock_get(): + with patch('src.base_odds_manager.requests.get') as m: + m.return_value = _make_response({'items': [dict(FULL_ITEM)]}) + yield m + + +# --------------------------------------------------------------------------- +# get_odds +# --------------------------------------------------------------------------- + +class TestGetOdds: + def test_none_sport_raises(self, manager): + with pytest.raises(ValueError): + manager.get_odds(None, 'nfl', '1') + + def test_none_league_raises(self, manager): + with pytest.raises(ValueError): + manager.get_odds('football', None, '1') + + def test_cache_key_and_url(self, manager, cache_manager, mock_get): + manager.get_odds('football', 'nfl', '401') + + cache_manager.get_with_auto_strategy.assert_called_once_with( + 'odds_espn_football_nfl_401') + url = mock_get.call_args[0][0] + # Event id appears twice: /events//competitions//odds + assert '/events/401/competitions/401/odds' in url + assert url == ('https://sports.core.api.espn.com/v2/sports/football/' + 'leagues/nfl/events/401/competitions/401/odds') + assert mock_get.call_args.kwargs['timeout'] == 30 + + def test_ncaa_fb_maps_to_college_football(self, manager, mock_get): + manager.get_odds('football', 'ncaa_fb', '401') + + url = mock_get.call_args[0][0] + assert '/leagues/college-football/' in url + + def test_unknown_league_passes_through(self, manager, mock_get): + manager.get_odds('football', 'xfl', '401') + + assert '/leagues/xfl/' in mock_get.call_args[0][0] + + def test_cache_hit_skips_http(self, manager, cache_manager, mock_get): + cache_manager.get_with_auto_strategy.return_value = {'spread': -3.0} + + result = manager.get_odds('football', 'nfl', '401') + + assert result == {'spread': -3.0} + mock_get.assert_not_called() + + def test_cached_no_odds_sentinel_returned_verbatim( + self, manager, cache_manager, mock_get): + cache_manager.get_with_auto_strategy.return_value = {'no_odds': True} + + result = manager.get_odds('football', 'nfl', '401') + + assert result == {'no_odds': True} + mock_get.assert_not_called() + assert manager.is_odds_available(result) is False + + def test_success_caches_extracted_data_with_interval_ttl( + self, manager, cache_manager, mock_get): + result = manager.get_odds('football', 'nfl', '401', + update_interval_seconds=100) + + assert result == FULL_EXTRACTED + cache_manager.set.assert_called_once_with( + 'odds_espn_football_nfl_401', FULL_EXTRACTED, ttl=100) + + def test_no_odds_caches_sentinel(self, manager, cache_manager, mock_get): + mock_get.return_value = _make_response({'count': 0, 'items': []}) + + result = manager.get_odds('football', 'nfl', '401') + + assert result is None + cache_manager.set.assert_called_once_with( + 'odds_espn_football_nfl_401', {'no_odds': True}, ttl=3600) + + def test_zero_interval_falls_back_to_default( + self, manager, cache_manager, mock_get): + # Quirk pin: `update_interval_seconds or self.update_interval` + # treats an explicit 0 as falsy, so the 3600 default wins. + manager.get_odds('football', 'nfl', '401', update_interval_seconds=0) + + assert cache_manager.set.call_args.kwargs['ttl'] == 3600 + + def test_request_exception_falls_back_to_stale_cache( + self, manager, cache_manager, mock_get): + cache_manager.get_with_auto_strategy.side_effect = [ + None, {'stale': True}] + mock_get.side_effect = requests.exceptions.RequestException('boom') + + result = manager.get_odds('football', 'nfl', '401') + + assert result == {'stale': True} + assert cache_manager.get_with_auto_strategy.call_count == 2 + + +# --------------------------------------------------------------------------- +# _extract_espn_data +# --------------------------------------------------------------------------- + +class TestExtractEspnData: + def test_full_item_extracts_all_fields(self, manager): + result = manager._extract_espn_data({'items': [dict(FULL_ITEM)]}) + assert result == FULL_EXTRACTED + + def test_explicit_nulls_do_not_raise(self, manager): + # Post-fix: ESPN sends explicit JSON nulls for absent sides + # ("homeTeamOdds": null, "current": null); extraction must not + # raise and yields None fields. + payload = {'items': [{ + 'homeTeamOdds': None, + 'awayTeamOdds': {'moneyLine': 150, 'current': None}, + }]} + + result = manager._extract_espn_data(payload) + + assert result is not None + assert result['home_team_odds']['money_line'] is None + assert result['home_team_odds']['spread_odds'] is None + assert result['away_team_odds']['money_line'] == 150 + assert result['away_team_odds']['spread_odds'] is None + + def test_valid_empty_response_returns_none(self, manager): + assert manager._extract_espn_data({'count': 0, 'items': []}) is None + + def test_unexpected_structure_returns_none(self, manager): + assert manager._extract_espn_data({'unexpected': True}) is None + + def test_item_without_odds_fields_cached_as_data_not_sentinel( + self, manager, cache_manager, mock_get): + # Characterization pin: an item with no odds fields still extracts + # to a truthy dict of all-None values, so get_odds caches it as + # real data (NOT the no_odds sentinel) — but is_odds_available + # correctly reports False for it. + mock_get.return_value = _make_response({'items': [{}]}) + + result = manager.get_odds('football', 'nfl', '401') + + assert result == { + 'details': None, + 'over_under': None, + 'spread': None, + 'home_team_odds': {'money_line': None, 'spread_odds': None}, + 'away_team_odds': {'money_line': None, 'spread_odds': None}, + } + cache_manager.set.assert_called_once_with( + 'odds_espn_football_nfl_401', result, ttl=3600) + assert manager.is_odds_available(result) is False + + +# --------------------------------------------------------------------------- +# is_odds_available +# --------------------------------------------------------------------------- + +class TestIsOddsAvailable: + def test_none_is_false(self, manager): + assert manager.is_odds_available(None) is False + + def test_empty_dict_is_false(self, manager): + assert manager.is_odds_available({}) is False + + def test_no_odds_sentinel_is_false(self, manager): + assert manager.is_odds_available({'no_odds': True}) is False + + def test_spread_is_true(self, manager): + assert manager.is_odds_available({'spread': -3.5}) is True + + def test_over_under_is_true(self, manager): + assert manager.is_odds_available({'over_under': 47.5}) is True + + def test_nested_home_spread_odds_is_true(self, manager): + assert manager.is_odds_available( + {'home_team_odds': {'spread_odds': -3.5}}) is True + + def test_nested_away_spread_odds_is_true(self, manager): + assert manager.is_odds_available( + {'away_team_odds': {'spread_odds': 3.5}}) is True + + def test_moneyline_only_is_false(self, manager): + # Pinned ML-blind contract: is_odds_available ignores money lines + # (its callers decide whether to render an odds widget). Note that + # format_odds_summary deliberately uses a DIFFERENT gate — it will + # still format money-line-only odds (see TestFormatOddsSummary). + ml_only = { + 'home_team_odds': {'money_line': -120}, + 'away_team_odds': {'money_line': 100}, + } + assert manager.is_odds_available(ml_only) is False + + +# --------------------------------------------------------------------------- +# format_odds_summary (fixed gate: empty / no_odds only) +# --------------------------------------------------------------------------- + +class TestFormatOddsSummary: + def test_moneyline_only_formats(self, manager): + result = manager.format_odds_summary({ + 'home_team_odds': {'money_line': -120}, + 'away_team_odds': {'money_line': 100}, + }) + assert result == 'Home ML: -120 | Away ML: 100' + + def test_full_data_formats_all_parts(self, manager): + result = manager.format_odds_summary(FULL_EXTRACTED) + assert result == 'Spread: -3.5 | O/U: 47.5 | Home ML: -150 | Away ML: 130' + + def test_none_is_no_odds(self, manager): + assert manager.format_odds_summary(None) == 'No odds available' + + def test_empty_dict_is_no_odds(self, manager): + assert manager.format_odds_summary({}) == 'No odds available' + + def test_no_odds_sentinel_is_no_odds(self, manager): + assert manager.format_odds_summary( + {'no_odds': True}) == 'No odds available' + + +# --------------------------------------------------------------------------- +# get_odds_for_games +# --------------------------------------------------------------------------- + +class TestGetOddsForGames: + def test_missing_fields_get_none_odds_without_http(self, manager, mock_get): + games = [ + {'sport': 'football'}, + {'league': 'nfl'}, + {'id': '9'}, + {}, + ] + + result = manager.get_odds_for_games(games) + + assert all(g['odds'] is None for g in result) + mock_get.assert_not_called() + + def test_per_game_exception_continues_loop(self, manager, monkeypatch): + def fake_get_odds(sport, league, event_id, + update_interval_seconds=None): + if event_id == 'bad': + raise RuntimeError('boom') + return {'spread': -1.0} + + monkeypatch.setattr(manager, 'get_odds', fake_get_odds) + games = [ + {'sport': 'football', 'league': 'nfl', 'id': 'bad'}, + {'sport': 'football', 'league': 'nfl', 'id': 'ok'}, + ] + + result = manager.get_odds_for_games(games) + + assert len(result) == 2 + assert result[0]['odds'] is None + assert result[1]['odds'] == {'spread': -1.0} + + def test_input_dicts_mutated_in_place_and_returned(self, manager, mock_get): + # Pin: get_odds_for_games mutates the caller's game dicts in place + # and returns the same objects, not copies. + game = {'sport': 'football', 'league': 'nfl', 'id': '401'} + + result = manager.get_odds_for_games([game]) + + assert result[0] is game + assert game['odds'] == FULL_EXTRACTED + + +# --------------------------------------------------------------------------- +# _load_configuration +# --------------------------------------------------------------------------- + +class TestLoadConfiguration: + def test_loads_values_from_config(self, cache_manager): + config_manager = MagicMock() + config_manager.get_config.return_value = { + 'base_odds_manager': { + 'update_interval': 100, + 'timeout': 5, + 'cache_ttl': 42, + } + } + + manager = BaseOddsManager(cache_manager, config_manager=config_manager) + + assert manager.update_interval == 100 + # Key/attr mismatch pin: the config key is 'timeout' but the + # attribute is request_timeout. + assert manager.request_timeout == 5 + assert manager.cache_ttl == 42 + + def test_get_config_raising_keeps_defaults(self, cache_manager): + config_manager = MagicMock() + config_manager.get_config.side_effect = RuntimeError('boom') + + manager = BaseOddsManager(cache_manager, config_manager=config_manager) + + assert manager.update_interval == 3600 + assert manager.request_timeout == 30 + assert manager.cache_ttl == 1800 diff --git a/test/test_config_helper.py b/test/test_config_helper.py new file mode 100644 index 00000000..c8e9ae85 --- /dev/null +++ b/test/test_config_helper.py @@ -0,0 +1,243 @@ +""" +Tests for src/common/config_helper.py — pins the ConfigHelper contract. + +Covers: load/save round trips (missing/malformed files return {} rather +than raising, non-ASCII preserved via ensure_ascii=False, top-level JSON +lists returned as-is), dot-notation get/set including the silent-failure +contract when an intermediate key holds a non-dict, merge_configs deep +semantics with NO aliasing of the base config (the fixed bug — the old +shallow copy let mutations of the merged result leak into base's nested +dicts), simplified schema validation including the caught-TypeError path +when a schema 'type' is given as a string, plugin config key conventions +('{plugin_id}_config', enabled defaults True), and required-key checks +where a key present with value None counts as present. +""" + +import json + +import pytest + +from src.common.config_helper import ConfigHelper + + +@pytest.fixture +def helper(): + return ConfigHelper() + + +class TestLoadConfig: + def test_missing_file_returns_empty_dict(self, helper, tmp_path): + assert helper.load_config(tmp_path / "nope.json") == {} + + def test_malformed_json_returns_empty_dict(self, helper, tmp_path): + path = tmp_path / "bad.json" + path.write_text("{ this is not json", encoding="utf-8") + assert helper.load_config(path) == {} + + def test_top_level_list_returned_as_is(self, helper, tmp_path): + # load_config does not enforce a dict shape: a JSON list comes + # straight back. Pinned as a characterization of current behavior. + path = tmp_path / "list.json" + path.write_text("[1, 2, 3]", encoding="utf-8") + assert helper.load_config(path) == [1, 2, 3] + + +class TestSaveConfig: + def test_round_trip(self, helper, tmp_path): + path = tmp_path / "config.json" + config = {'display': {'hardware': {'rows': 32}}, 'timezone': 'UTC'} + assert helper.save_config(config, path) is True + assert helper.load_config(path) == config + + def test_creates_parent_directories(self, helper, tmp_path): + path = tmp_path / "deep" / "nested" / "config.json" + assert helper.save_config({'a': 1}, path) is True + assert path.exists() + assert helper.load_config(path) == {'a': 1} + + def test_non_ascii_survives_round_trip(self, helper, tmp_path): + path = tmp_path / "config.json" + config = {'city': 'Zürich', 'note': 'météo ☀'} + assert helper.save_config(config, path) is True + assert helper.load_config(path) == config + # ensure_ascii=False: characters are written raw, not \u-escaped + assert 'Zürich' in path.read_text(encoding='utf-8') + + def test_directory_path_returns_false_not_raise(self, helper, tmp_path): + assert helper.save_config({'a': 1}, tmp_path) is False + + +class TestGetConfigValue: + def test_dot_notation_hit(self, helper): + config = {'display': {'hardware': {'rows': 32}}} + assert helper.get_config_value(config, 'display.hardware.rows') == 32 + + def test_missing_returns_default(self, helper): + sentinel = object() + assert helper.get_config_value({}, 'display.rows', default=sentinel) is sentinel + + def test_intermediate_non_dict_returns_default(self, helper): + config = {'display': 'not-a-dict'} + assert helper.get_config_value(config, 'display.hardware.rows', default=64) == 64 + + def test_required_missing_raises_keyerror(self, helper): + with pytest.raises(KeyError): + helper.get_config_value({}, 'display.rows', required=True) + + +class TestSetConfigValue: + def test_sets_top_level(self, helper): + config = {} + helper.set_config_value(config, 'timezone', 'UTC') + assert config == {'timezone': 'UTC'} + + def test_auto_creates_intermediates(self, helper): + config = {} + helper.set_config_value(config, 'display.hardware.rows', 32) + assert config == {'display': {'hardware': {'rows': 32}}} + + def test_silent_failure_on_non_dict_intermediate(self, helper): + # 'a' exists but holds an int; the assignment attempt raises + # TypeError internally, which set_config_value swallows and logs. + # The config is left unchanged — pinned silent-failure contract. + config = {'a': 5} + helper.set_config_value(config, 'a.b', 1) + assert config == {'a': 5} + + +class TestMergeConfigs: + def test_nested_dicts_merge_recursively(self, helper): + base = {'display': {'rows': 32, 'cols': 64}, 'timezone': 'UTC'} + override = {'display': {'cols': 128, 'brightness': 90}} + merged = helper.merge_configs(base, override) + assert merged == { + 'display': {'rows': 32, 'cols': 128, 'brightness': 90}, + 'timezone': 'UTC', + } + + def test_scalar_override_wins_over_dict(self, helper): + merged = helper.merge_configs({'display': {'rows': 32}}, {'display': 7}) + assert merged['display'] == 7 + + def test_dict_override_wins_over_scalar(self, helper): + merged = helper.merge_configs({'display': 7}, {'display': {'rows': 32}}) + assert merged['display'] == {'rows': 32} + + def test_no_aliasing_of_base(self, helper): + # Post-fix: merge deep-copies base, so mutating the result never + # leaks back into the caller's base config. + base = {'display': {'x': 1}} + merged = helper.merge_configs(base, {}) + assert merged['display'] is not base['display'] + merged['display']['x'] = 99 + assert base['display']['x'] == 1 + + def test_inputs_unchanged(self, helper): + base = {'a': {'b': 1}} + override = {'a': {'c': 2}} + helper.merge_configs(base, override) + assert base == {'a': {'b': 1}} + assert override == {'a': {'c': 2}} + + +class TestValidateConfig: + def test_no_schema_dict_is_valid(self, helper): + assert helper.validate_config({'a': 1}) is True + + def test_no_schema_list_is_invalid(self, helper): + assert helper.validate_config([1, 2]) is False + + def test_required_key_missing_is_invalid(self, helper): + schema = {'rows': {'required': True, 'type': int}} + assert helper.validate_config({}, schema) is False + + def test_optional_key_missing_is_valid(self, helper): + schema = {'rows': {'required': False, 'type': int}} + assert helper.validate_config({}, schema) is True + + def test_wrong_type_is_invalid(self, helper): + schema = {'rows': {'type': int}} + assert helper.validate_config({'rows': 'thirty-two'}, schema) is False + assert helper.validate_config({'rows': 32}, schema) is True + + def test_allowed_values_violation_is_invalid(self, helper): + schema = {'mode': {'allowed_values': ['clock', 'weather']}} + assert helper.validate_config({'mode': 'stocks'}, schema) is False + assert helper.validate_config({'mode': 'clock'}, schema) is True + + def test_string_type_in_schema_is_invalid_via_typeerror(self, helper): + # 'type' given as the STRING "int" makes isinstance() raise + # TypeError; validate_config catches it and returns False rather + # than raising. Pinned characterization. + schema = {'rows': {'type': 'int'}} + assert helper.validate_config({'rows': 32}, schema) is False + + +class TestPluginConfigHelpers: + def test_get_plugin_config_uses_suffixed_key(self, helper): + plugin_cfg = {'enabled': True, 'display_duration': 30} + assert helper.get_plugin_config({'clock_config': plugin_cfg}, 'clock') == plugin_cfg + + def test_get_plugin_config_bare_id_key_not_found(self, helper): + # Only '{plugin_id}_config' is consulted — a bare 'clock' section + # is invisible to this helper. Pinned key contract. + assert helper.get_plugin_config({'clock': {'enabled': True}}, 'clock') == {} + + def test_create_default_config_wraps_in_suffixed_key(self, helper): + defaults = {'enabled': True} + assert helper.create_default_config('clock', defaults) == {'clock_config': defaults} + + def test_is_plugin_enabled_defaults_true_for_unknown(self, helper): + assert helper.is_plugin_enabled({}, 'clock') is True + + def test_is_plugin_enabled_false_when_disabled(self, helper): + config = {'clock_config': {'enabled': False}} + assert helper.is_plugin_enabled(config, 'clock') is False + + def test_is_plugin_enabled_ignores_bare_id_key(self, helper): + # Disabled under the wrong key -> still reported enabled (default). + config = {'clock': {'enabled': False}} + assert helper.is_plugin_enabled(config, 'clock') is True + + +class TestSportsAndDisplayHelpers: + def test_get_display_config(self, helper): + display = {'hardware': {'rows': 32}} + assert helper.get_display_config({'display': display}) == display + assert helper.get_display_config({}) == {} + + def test_get_sports_config_uses_scoreboard_suffix(self, helper): + sport_cfg = {'favorite_teams': ['TB']} + config = {'football_scoreboard': sport_cfg} + assert helper.get_sports_config(config, 'football') == sport_cfg + assert helper.get_sports_config(config, 'hockey') == {} + + def test_get_favorite_teams(self, helper): + config = {'football_scoreboard': {'favorite_teams': ['TB', 'DAL']}} + assert helper.get_favorite_teams(config, 'football') == ['TB', 'DAL'] + assert helper.get_favorite_teams({}, 'football') == [] + + def test_get_display_modes(self, helper): + modes = {'live': True, 'recent': False} + config = {'football_scoreboard': {'display_modes': modes}} + assert helper.get_display_modes(config, 'football') == modes + assert helper.get_display_modes({}, 'football') == {} + + +class TestValidateRequiredKeys: + def test_returns_missing_subset(self, helper): + config = {'a': 1, 'c': {'d': 2}} + missing = helper.validate_required_keys(config, ['a', 'b', 'c.d', 'c.e']) + assert missing == ['b', 'c.e'] + + def test_dot_notation_present(self, helper): + config = {'display': {'hardware': {'rows': 32}}} + assert helper.validate_required_keys(config, ['display.hardware.rows']) == [] + + def test_empty_requirements(self, helper): + assert helper.validate_required_keys({'a': 1}, []) == [] + + def test_present_with_none_counts_as_present(self, helper): + # _has_key checks key membership, not truthiness — a key set to + # None is NOT reported missing. Pinned semantics. + assert helper.validate_required_keys({'a': None}, ['a']) == [] diff --git a/test/test_display_helper.py b/test/test_display_helper.py new file mode 100644 index 00000000..b8c9be63 --- /dev/null +++ b/test/test_display_helper.py @@ -0,0 +1,307 @@ +"""Tests for src/common/display_helper.py (DisplayHelper). + +Pure-PIL tests, no hardware or mocks required. Pixel assertions rely on +getbbox()/getpixel() rather than exact text pixel counts, because the +default-font metrics vary across Pillow versions. + +These tests pin the FIXED behaviors on this branch: +- draw_error_message / draw_no_data_message return a rendered image + (they previously crashed with AttributeError), +- draw_scorebug_layout draws period/status/clock as one combined top + line (previously overprinted at the same y), +- draw_ticker_layout draws at x=0 (previously started at + x=display_width, i.e. entirely off-canvas -> blank frames). +""" + +from PIL import Image, ImageDraw, ImageFont + +from src.common.display_helper import DisplayHelper + + +def default_font(): + return ImageFont.load_default() + + +def make_helper(width=128, height=32): + return DisplayHelper(width, height) + + +class TestCreateBaseImage: + def test_default_is_black_rgb_display_sized(self): + helper = make_helper() + img = helper.create_base_image() + assert img.size == (128, 32) + assert img.mode == 'RGB' + assert img.getpixel((0, 0)) == (0, 0, 0) + assert img.getpixel((127, 31)) == (0, 0, 0) + # Entirely black -> no bounding box in luminance + assert img.convert('L').getbbox() is None + + def test_custom_background_color(self): + helper = make_helper() + img = helper.create_base_image(background_color=(10, 20, 30)) + assert img.getpixel((0, 0)) == (10, 20, 30) + assert img.getpixel((64, 16)) == (10, 20, 30) + + def test_mode_rgba_is_honored(self): + helper = make_helper() + img = helper.create_base_image(mode='RGBA') + assert img.mode == 'RGBA' + assert img.size == (128, 32) + + +class TestCreateOverlay: + def test_overlay_is_transparent_rgba(self): + helper = make_helper() + overlay = helper.create_overlay() + assert overlay.mode == 'RGBA' + assert overlay.size == (128, 32) + assert overlay.getpixel((0, 0)) == (0, 0, 0, 0) + assert overlay.getpixel((127, 31)) == (0, 0, 0, 0) + + +class TestCompositeImages: + def test_rgb_inputs_are_upconverted_and_result_is_rgba(self): + helper = make_helper() + base = Image.new('RGB', (128, 32), (0, 0, 0)) + overlay = Image.new('RGB', (128, 32), (255, 0, 0)) + result = helper.composite_images(base, overlay) + assert result.mode == 'RGBA' + assert result.size == base.size + # RGB->RGBA conversion yields a fully opaque overlay + assert result.getpixel((0, 0)) == (255, 0, 0, 255) + + def test_transparent_overlay_leaves_base_visible(self): + helper = make_helper() + base = Image.new('RGB', (128, 32), (5, 6, 7)) + overlay = helper.create_overlay() + result = helper.composite_images(base, overlay) + assert result.mode == 'RGBA' + assert result.getpixel((64, 16)) == (5, 6, 7, 255) + + +class TestScorebugLayout: + def test_full_game_data_renders(self): + helper = make_helper() + font = default_font() + fonts = {'time': font, 'status': font, 'score': font, 'team': font} + game_data = { + 'home_score': 3, 'away_score': 2, + 'home_abbr': 'NYY', 'away_abbr': 'BOS', + 'status_text': 'LIVE', 'period_text': 'T9', 'clock': '2:30', + } + img = helper.draw_scorebug_layout(game_data, fonts) + assert img.mode == 'RGB' + assert img.size == (128, 32) + assert img.convert('L').getbbox() is not None + + def test_empty_game_data_uses_defaults_without_raising(self): + helper = make_helper() + font = default_font() + fonts = {'time': font, 'status': font, 'score': font, 'team': font} + img = helper.draw_scorebug_layout({}, fonts) + assert img.mode == 'RGB' + assert img.size == (128, 32) + # Defaults '0'/'HOME'/'AWAY' actually render something + assert img.convert('L').getbbox() is not None + + def test_empty_fonts_dict_falls_back_to_default_font(self): + # Pin: fonts={} must not raise — PIL falls back to the default + # font when font=None is passed through. + helper = make_helper() + img = helper.draw_scorebug_layout( + {'status_text': 'FINAL', 'period_text': 'Q4', 'clock': '0:00'}, {}) + assert img.size == (128, 32) + assert img.convert('L').getbbox() is not None + + def test_top_line_is_one_combined_centered_draw(self): + # FIXED behavior: period/status/clock are joined into a single + # top line drawn once at y=1 instead of three overprinted draws. + helper = make_helper() + calls = [] + original = helper._draw_centered_text + + def spy(draw, text, font, y_position): + calls.append({'text': text, 'y_position': y_position}) + original(draw, text, font, y_position) + + helper._draw_centered_text = spy + font = default_font() + fonts = {'time': font, 'status': font, 'score': font, 'team': font} + helper.draw_scorebug_layout( + {'period_text': 'Q4', 'status_text': 'LIVE', 'clock': '2:30'}, + fonts) + + top_calls = [c for c in calls if c['y_position'] == 1] + assert len(top_calls) == 1 + text = top_calls[0]['text'] + assert 'Q4' in text + assert 'LIVE' in text + assert '2:30' in text + + def test_no_top_line_when_all_parts_empty(self): + helper = make_helper() + calls = [] + original = helper._draw_centered_text + + def spy(draw, text, font, y_position): + calls.append(y_position) + original(draw, text, font, y_position) + + helper._draw_centered_text = spy + font = default_font() + helper.draw_scorebug_layout({}, {'score': font, 'team': font}) + assert 1 not in calls # no combined top line drawn + + def test_logo_positions_bleed_off_edges(self): + # Home logo pastes at x = width - logo.width + 10 (right edge, + # bleeding off-screen right); away at x = -10 (bleeding left). + helper = make_helper() + home_logo = Image.new('RGBA', (20, 20), (0, 0, 255, 255)) # blue + away_logo = Image.new('RGBA', (20, 20), (255, 0, 0, 255)) # red + # Empty abbrs/status so text can't land on the probed pixels. + game_data = {'home_abbr': '', 'away_abbr': ''} + font = default_font() + img = helper.draw_scorebug_layout(game_data, {'score': font}, + home_logo=home_logo, + away_logo=away_logo) + # center_y = 16; logos span y 6..25 -> probe y=16 at both edges. + assert img.getpixel((0, 16)) == (255, 0, 0) # away (left edge) + assert img.getpixel((127, 16)) == (0, 0, 255) # home (right edge) + # And the off-screen parts are truly clipped: image is still 128 wide + assert img.size == (128, 32) + + +class TestTickerLayout: + def test_frame_is_not_blank(self): + # FIXED behavior: text now starts at x=0. Previously it was drawn + # at x=display_width, entirely off-canvas, so frames were blank. + helper = make_helper() + img = helper.draw_ticker_layout('HELLO WORLD', default_font()) + assert img.size == (128, 32) + assert img.mode == 'RGB' + assert img.convert('L').getbbox() is not None + + def test_text_starts_at_left_edge(self): + helper = make_helper() + img = helper.draw_ticker_layout('HELLO', default_font()) + bbox = img.convert('L').getbbox() + assert bbox is not None + # Text is positioned at x=0 (outline extends 1px left, clipped), + # so ink begins hugging the left edge. Allow a couple of pixels of + # slack for font-dependent left-side bearing. + assert bbox[0] <= 2 + + def test_scroll_speed_does_not_affect_frame(self): + # Pin: scroll_speed is accepted for API compatibility only. + helper = make_helper() + font = default_font() + img1 = helper.draw_ticker_layout('SCROLLING', font, scroll_speed=1) + img5 = helper.draw_ticker_layout('SCROLLING', font, scroll_speed=5) + assert img1.tobytes() == img5.tobytes() + + def test_custom_colors(self): + helper = make_helper() + img = helper.draw_ticker_layout('X', default_font(), + background_color=(0, 0, 40), + text_color=(0, 255, 0)) + assert img.getpixel((127, 0)) == (0, 0, 40) # background corner + colors = {img.getpixel((x, y)) + for x in range(img.width) for y in range(img.height)} + # Text color appears somewhere (anti-aliasing may blend it, so + # check for a green-dominant pixel rather than the exact color). + assert any(g > 150 and r < 100 for (r, g, b) in colors) + + +class TestCenteredText: + def test_renders_centered_text_on_background(self): + helper = make_helper() + img = helper.draw_centered_text('HI', default_font(), + background_color=(0, 0, 60), + text_color=(255, 255, 0)) + assert img.size == (128, 32) + assert img.convert('L').getbbox() is not None + # Corners stay pure background + assert img.getpixel((0, 0)) == (0, 0, 60) + assert img.getpixel((127, 0)) == (0, 0, 60) + assert img.getpixel((0, 31)) == (0, 0, 60) + assert img.getpixel((127, 31)) == (0, 0, 60) + + +class TestErrorAndNoDataMessages: + def test_draw_error_message_returns_rendered_image(self): + # FIXED behavior: used to crash with AttributeError; now returns + # a rendered image on a dark red background. + helper = make_helper() + img = helper.draw_error_message('Boom') + assert img.size == (128, 32) + assert img.mode == 'RGB' + assert img.convert('L').getbbox() is not None + assert img.getpixel((0, 0)) == (50, 0, 0) # dark red background + + def test_draw_error_message_default_text(self): + helper = make_helper() + img = helper.draw_error_message() + assert img.size == (128, 32) + assert img.getpixel((127, 31)) == (50, 0, 0) + + def test_draw_no_data_message_returns_rendered_image(self): + helper = make_helper() + img = helper.draw_no_data_message() + assert img.size == (128, 32) + assert img.mode == 'RGB' + assert img.convert('L').getbbox() is not None + assert img.getpixel((0, 0)) == (0, 0, 0) # black background + + +class TestDrawTextWithOutline: + def test_fill_color_appears_in_output(self): + helper = make_helper() + img = Image.new('RGB', (40, 20), (0, 0, 255)) + draw = ImageDraw.Draw(img) + helper._draw_text_with_outline(draw, 'X', (5, 2), default_font(), + fill=(255, 0, 0)) + pixels = {img.getpixel((x, y)) + for x in range(img.width) for y in range(img.height)} + # Anti-aliased fonts blend edge pixels, so look for red-dominant + # (fill) and near-black (outline) pixels rather than exact colors. + assert any(r > 150 and g < 50 for (r, g, b) in pixels) # fill + assert any(max(p) < 80 for p in pixels) # outline + + def test_default_fill_is_white(self): + helper = make_helper() + img = Image.new('RGB', (40, 20), (0, 0, 255)) + draw = ImageDraw.Draw(img) + helper._draw_text_with_outline(draw, 'X', (5, 2), default_font()) + pixels = {img.getpixel((x, y)) + for x in range(img.width) for y in range(img.height)} + # White-dominant pixel present (exact white may be anti-aliased) + assert any(r > 200 and g > 200 for (r, g, b) in pixels) + + +class TestOrientationAndDimensions: + def test_landscape_display(self): + helper = DisplayHelper(128, 32) + assert helper.is_landscape() is True + assert helper.is_portrait() is False + + def test_portrait_display(self): + helper = DisplayHelper(32, 128) + assert helper.is_portrait() is True + assert helper.is_landscape() is False + + def test_square_display_is_neither(self): + # Pin: a square display is neither portrait nor landscape. + helper = DisplayHelper(64, 64) + assert helper.is_portrait() is False + assert helper.is_landscape() is False + + def test_get_center_position(self): + assert DisplayHelper(128, 32).get_center_position() == (64, 16) + + def test_get_center_position_floors_odd_dimensions(self): + assert DisplayHelper(65, 33).get_center_position() == (32, 16) + + def test_get_display_dimensions(self): + assert DisplayHelper(128, 32).get_display_dimensions() == (128, 32) + assert DisplayHelper(64, 64).get_display_dimensions() == (64, 64) diff --git a/test/test_dynamic_team_resolver.py b/test/test_dynamic_team_resolver.py new file mode 100644 index 00000000..c562d0d9 --- /dev/null +++ b/test/test_dynamic_team_resolver.py @@ -0,0 +1,259 @@ +""" +Tests for src/dynamic_team_resolver.py (DynamicTeamResolver). + +Covers dynamic team expansion (AP_TOP_5/10/25), order-preserving dedup, +unknown dynamic-name dropping, rankings parsing, the fixed genuinely +class-shared rankings cache (fetch and clear_cache write through +DynamicTeamResolver._rankings_cache / _cache_timestamp), TTL expiry, +network-failure resilience, and the resolve_dynamic_teams module function. + +No real network: src.dynamic_team_resolver.requests.get is always patched. +""" + +import types +from unittest.mock import MagicMock, patch + +import pytest +import requests + +import src.dynamic_team_resolver as dtr_module +from src.dynamic_team_resolver import DynamicTeamResolver, resolve_dynamic_teams + + +TOP_TEAMS = ['UGA', 'MICH', 'OSU', 'TEX', 'ALA', 'ORE', 'PSU', 'ND', 'FSU', 'OU'] + + +def _rankings_payload(teams=None): + teams = TOP_TEAMS if teams is None else teams + return { + 'rankings': [{ + 'name': 'AP Top 25', + 'ranks': [ + {'current': i + 1, 'team': {'abbreviation': abbr}} + for i, abbr in enumerate(teams) + ], + }] + } + + +def _make_response(payload): + response = MagicMock() + response.json.return_value = payload + response.raise_for_status.return_value = None + return response + + +@pytest.fixture(autouse=True) +def reset_class_cache(): + """Reset the CLASS-level shared cache between tests.""" + DynamicTeamResolver._rankings_cache = {} + DynamicTeamResolver._cache_timestamp = 0 + yield + DynamicTeamResolver._rankings_cache = {} + DynamicTeamResolver._cache_timestamp = 0 + + +@pytest.fixture +def mock_get(): + with patch('src.dynamic_team_resolver.requests.get') as m: + m.return_value = _make_response(_rankings_payload()) + yield m + + +@pytest.fixture +def resolver(): + return DynamicTeamResolver() + + +# --------------------------------------------------------------------------- +# resolve_teams basics +# --------------------------------------------------------------------------- + +class TestResolveTeamsBasics: + def test_empty_list_returns_empty_no_http(self, resolver, mock_get): + assert resolver.resolve_teams([]) == [] + mock_get.assert_not_called() + + def test_no_dynamic_names_passthrough_no_http(self, resolver, mock_get): + assert resolver.resolve_teams(['UGA', 'AUB', 'LSU']) == [ + 'UGA', 'AUB', 'LSU'] + mock_get.assert_not_called() + + def test_expansion_inserted_in_place_order_preserved( + self, resolver, mock_get): + result = resolver.resolve_teams(['UGA', 'AP_TOP_5', 'AUB']) + + # UGA is also ranked #1, so dedup keeps its first occurrence; the + # top-5 expansion lands where AP_TOP_5 appeared, AUB stays after. + assert result == ['UGA', 'MICH', 'OSU', 'TEX', 'ALA', 'AUB'] + + def test_order_preserving_dedup(self, resolver, mock_get): + result = resolver.resolve_teams(['UGA', 'AP_TOP_5', 'UGA']) + + assert result == ['UGA', 'MICH', 'OSU', 'TEX', 'ALA'] + assert result.count('UGA') == 1 + assert result[0] == 'UGA' + + +# --------------------------------------------------------------------------- +# AP_TOP_N slicing +# --------------------------------------------------------------------------- + +class TestSlicing: + def test_top_n_counts_and_order(self, resolver, mock_get): + teams_25 = [f'T{i:02d}' for i in range(1, 26)] + mock_get.return_value = _make_response(_rankings_payload(teams_25)) + + top5 = resolver.resolve_teams(['AP_TOP_5']) + top10 = resolver.resolve_teams(['AP_TOP_10']) + top25 = resolver.resolve_teams(['AP_TOP_25']) + + assert top5 == teams_25[:5] + assert top10 == teams_25[:10] + assert top25 == teams_25 + + +# --------------------------------------------------------------------------- +# Unknown dynamic-looking names +# --------------------------------------------------------------------------- + +class TestUnknownDynamicNames: + def test_unknown_dynamic_looking_names_dropped(self, resolver, mock_get): + result = resolver.resolve_teams( + ['AP_TOP_100', 'TOP_10', 'RANKED_ALL', 'PLAYOFF_TEAMS']) + + assert result == [] + mock_get.assert_not_called() + + def test_top_substring_hazard(self, resolver, mock_get): + # Hazard pin: _is_potential_dynamic_team matches the substring + # 'TOP_' anywhere in the (upper-cased) name, so a team literally + # named 'TOP_GUN' is dropped as an unknown dynamic team too. + assert resolver.resolve_teams(['TOP_GUN']) == [] + + +# --------------------------------------------------------------------------- +# Rankings parsing +# --------------------------------------------------------------------------- + +class TestRankingsParsing: + def test_drops_zero_rank_and_empty_abbreviation_sorts_ascending( + self, resolver, mock_get): + payload = { + 'rankings': [{ + 'name': 'AP Top 25', + 'ranks': [ + {'current': 3, 'team': {'abbreviation': 'C3'}}, + {'current': 1, 'team': {'abbreviation': 'A1'}}, + {'current': 0, 'team': {'abbreviation': 'ZERO'}}, + {'current': 4, 'team': {'abbreviation': ''}}, + {'current': 2, 'team': {'abbreviation': 'B2'}}, + ], + }] + } + mock_get.return_value = _make_response(payload) + + rankings = resolver._fetch_ncaa_fb_rankings() + + assert list(rankings.keys()) == ['A1', 'B2', 'C3'] + assert list(rankings.values()) == [1, 2, 3] + + def test_empty_rankings_returns_empty_and_caches_nothing( + self, resolver, mock_get): + mock_get.return_value = _make_response({'rankings': []}) + + assert resolver._fetch_ncaa_fb_rankings() == {} + # Nothing was cached, so the next call hits HTTP again. + assert resolver._fetch_ncaa_fb_rankings() == {} + assert mock_get.call_count == 2 + + +# --------------------------------------------------------------------------- +# Shared class cache (fixed behavior) +# --------------------------------------------------------------------------- + +class TestSharedCache: + def test_cache_shared_across_instances(self, mock_get): + resolver1 = DynamicTeamResolver() + resolver1.resolve_teams(['AP_TOP_5']) + assert mock_get.call_count == 1 + + resolver2 = DynamicTeamResolver() + result = resolver2.resolve_teams(['AP_TOP_5']) + + # Post-fix: the class-level cache serves the second instance with + # ZERO additional HTTP calls. + assert result == TOP_TEAMS[:5] + assert mock_get.call_count == 1 + + def test_ttl_expiry_refetches(self, resolver, mock_get, monkeypatch): + resolver.resolve_teams(['AP_TOP_5']) + assert mock_get.call_count == 1 + + stamp = DynamicTeamResolver._cache_timestamp + monkeypatch.setattr( + dtr_module, 'time', types.SimpleNamespace(time=lambda: stamp + 3601)) + + resolver.resolve_teams(['AP_TOP_5']) + assert mock_get.call_count == 2 + + def test_clear_cache_through_one_instance_affects_all(self, mock_get): + resolver1 = DynamicTeamResolver() + resolver1.resolve_teams(['AP_TOP_5']) + assert mock_get.call_count == 1 + + resolver2 = DynamicTeamResolver() + resolver2.clear_cache() + + # Post-fix: clear_cache writes through the class, so resolver1 + # must refetch even though resolver2 did the clearing. + resolver1.resolve_teams(['AP_TOP_5']) + assert mock_get.call_count == 2 + + def test_module_function_benefits_from_class_cache(self, mock_get): + # resolve_dynamic_teams constructs a fresh resolver per call, but + # the class-shared cache means only the first call hits HTTP. + first = resolve_dynamic_teams(['AP_TOP_5']) + second = resolve_dynamic_teams(['AP_TOP_5']) + + assert first == second == TOP_TEAMS[:5] + assert mock_get.call_count == 1 + + +# --------------------------------------------------------------------------- +# Failure handling +# --------------------------------------------------------------------------- + +class TestFailureHandling: + def test_network_failure_drops_dynamic_keeps_static_caches_nothing( + self, resolver, mock_get): + mock_get.side_effect = [ + requests.exceptions.RequestException('boom'), + _make_response(_rankings_payload()), + ] + + result = resolver.resolve_teams(['UGA', 'AP_TOP_5']) + + # Dynamic name silently dropped, static name kept, nothing raises. + assert result == ['UGA'] + + # Nothing was cached on failure: a subsequent call refetches and + # succeeds. + result = resolver.resolve_teams(['UGA', 'AP_TOP_5']) + assert result == ['UGA', 'MICH', 'OSU', 'TEX', 'ALA'] + assert mock_get.call_count == 2 + + +# --------------------------------------------------------------------------- +# sport argument +# --------------------------------------------------------------------------- + +class TestSportArgument: + def test_sport_arg_ignored_for_expansion(self, resolver, mock_get): + # Pin: the sport argument is effectively ignored — each pattern + # carries its own sport ('ncaa_fb'), so passing sport='nfl' still + # expands from the college-football rankings. + result = resolver.resolve_teams(['AP_TOP_5'], sport='nfl') + + assert result == TOP_TEAMS[:5] + assert mock_get.call_count == 1 diff --git a/test/test_logging_config.py b/test/test_logging_config.py new file mode 100644 index 00000000..e0fe7fff --- /dev/null +++ b/test/test_logging_config.py @@ -0,0 +1,274 @@ +""" +Tests for src/logging_config.py — the formatters, adapter, and setup used +by every logger in the system (BasePlugin uses get_logger, not stdlib +logging.getLogger). + +Includes regression guards for two fixed bugs: ContextualFormatter used to +mutate record.msg in place (double-prefixing with two handlers), and +log_error hardcoded exc_info=True so passing it explicitly raised +TypeError. +""" + +import json +import logging +import sys + +import pytest + +from src.logging_config import ( + ContextualFormatter, + PluginLoggerAdapter, + StructuredFormatter, + get_logger, + log_debug, + log_error, + log_info, + log_warning, + log_with_context, + setup_logging, +) + + +def make_record(msg="hello", level=logging.INFO, **extra): + record = logging.LogRecord( + name="test.logger", level=level, pathname=__file__, lineno=42, + msg=msg, args=(), exc_info=None) + for key, value in extra.items(): + setattr(record, key, value) + return record + + +class TestStructuredFormatter: + def test_emits_valid_json_with_base_keys(self): + out = json.loads(StructuredFormatter().format(make_record())) + assert set(out) == { + "timestamp", "level", "logger", "message", + "module", "function", "line", + } + assert out["level"] == "INFO" + assert out["message"] == "hello" + assert out["logger"] == "test.logger" + + def test_optional_keys_only_when_present(self): + record = make_record(context={"k": "v"}, plugin_id="clock", + operation_id="op-1") + out = json.loads(StructuredFormatter().format(record)) + assert out["context"] == {"k": "v"} + assert out["plugin_id"] == "clock" + assert out["operation_id"] == "op-1" + + def test_exception_key_when_exc_info_present(self): + try: + raise ValueError("kaboom") + except ValueError: + record = logging.LogRecord( + name="t", level=logging.ERROR, pathname=__file__, lineno=1, + msg="failed", args=(), exc_info=sys.exc_info()) + out = json.loads(StructuredFormatter().format(record)) + assert "kaboom" in out["exception"] + + def test_percent_args_formatted_into_message(self): + record = logging.LogRecord( + name="t", level=logging.INFO, pathname=__file__, lineno=1, + msg="count=%d", args=(7,), exc_info=None) + out = json.loads(StructuredFormatter().format(record)) + assert out["message"] == "count=7" + + +class TestContextualFormatter: + def test_context_prefix_prepended(self): + record = make_record(plugin_id="clock", operation_id="op-1", + context={"k": "v"}) + out = ContextualFormatter().format(record) + assert "[Plugin: clock] [Op: op-1] [k: v] hello" in out + + def test_include_context_false_leaves_message_bare(self): + record = make_record(plugin_id="clock") + out = ContextualFormatter(include_context=False).format(record) + assert "[Plugin:" not in out + assert "hello" in out + + def test_location_toggle(self): + record = make_record() + with_loc = ContextualFormatter(include_location=True).format(record) + without = ContextualFormatter(include_location=False).format(record) + assert f":{record.lineno}" in with_loc + assert f":{record.lineno}" not in without + + def test_record_not_mutated_no_double_prefix(self): + # Regression: a record is formatted once PER HANDLER. The formatter + # must not mutate record.msg, or the second handler's format call + # prepends the prefix again. + record = make_record(plugin_id="clock") + formatter = ContextualFormatter() + first = formatter.format(record) + second = formatter.format(record) + assert record.msg == "hello" # untouched + assert first.count("[Plugin: clock]") == 1 + assert second.count("[Plugin: clock]") == 1 + + def test_percent_args_still_format_after_copy(self): + record = logging.LogRecord( + name="t", level=logging.INFO, pathname=__file__, lineno=1, + msg="count=%d", args=(7,), exc_info=None) + record.plugin_id = "clock" + out = ContextualFormatter().format(record) + assert "[Plugin: clock] count=7" in out + + def test_exception_renders_through_two_handlers(self): + try: + raise ValueError("kaboom") + except ValueError: + record = logging.LogRecord( + name="t", level=logging.ERROR, pathname=__file__, lineno=1, + msg="failed", args=(), exc_info=sys.exc_info()) + record.plugin_id = "clock" + formatter = ContextualFormatter() + assert "kaboom" in formatter.format(record) + assert "kaboom" in formatter.format(record) # second handler's pass + + +class TestPluginLoggerAdapter: + def _capture(self, adapter): + records = [] + handler = logging.Handler() + handler.emit = records.append + adapter.logger.addHandler(handler) + adapter.logger.setLevel(logging.DEBUG) + return records + + def test_stamps_plugin_id_on_every_record(self): + adapter = get_logger("test.adapter1", plugin_id="clock") + records = self._capture(adapter) + adapter.info("x") + assert records[0].plugin_id == "clock" + + def test_explicit_extra_plugin_id_wins(self): + adapter = get_logger("test.adapter2", plugin_id="clock") + records = self._capture(adapter) + adapter.info("x", extra={"plugin_id": "other"}) + assert records[0].plugin_id == "other" + + def test_unrelated_extra_keys_preserved(self): + adapter = get_logger("test.adapter3", plugin_id="clock") + records = self._capture(adapter) + adapter.info("x", extra={"custom": 1}) + assert records[0].plugin_id == "clock" + assert records[0].custom == 1 + + +class TestGetLogger: + def test_plain_logger_without_plugin_id(self): + logger = get_logger("test.plain") + assert isinstance(logger, logging.Logger) + assert logger.name == "test.plain" + + def test_adapter_with_plugin_id(self): + adapter = get_logger("test.wrapped", plugin_id="clock") + assert isinstance(adapter, PluginLoggerAdapter) + assert adapter.logger.name == "test.wrapped" + + +class TestSetupLogging: + # conftest's autouse reset_logging restores root handlers after each test. + + def test_installs_single_stdout_handler(self): + setup_logging() + root = logging.getLogger() + assert len(root.handlers) == 1 + assert isinstance(root.handlers[0], logging.StreamHandler) + + def test_repeat_calls_do_not_accumulate_handlers(self): + setup_logging() + setup_logging() + assert len(logging.getLogger().handlers) == 1 + + def test_json_format_selects_structured_formatter(self): + setup_logging(format_type="json") + assert isinstance( + logging.getLogger().handlers[0].formatter, StructuredFormatter) + + def test_readable_format_selects_contextual_formatter(self): + setup_logging(format_type="readable") + assert isinstance( + logging.getLogger().handlers[0].formatter, ContextualFormatter) + + def test_log_file_adds_file_handler(self, tmp_path): + log_file = tmp_path / "test.log" + setup_logging(log_file=str(log_file)) + root = logging.getLogger() + file_handlers = [h for h in root.handlers + if isinstance(h, logging.FileHandler)] + assert len(file_handlers) == 1 + for h in file_handlers: + h.close() + + def test_unwritable_log_file_warns_and_keeps_console(self, tmp_path, capsys): + bad_path = tmp_path / "no-such-dir" / "test.log" + setup_logging(log_file=str(bad_path)) # must not raise + assert len(logging.getLogger().handlers) == 1 # console only + assert "Could not set up file logging" in capsys.readouterr().err + + def test_debug_env_true_enables_debug(self, monkeypatch): + monkeypatch.setenv("LEDMATRIX_DEBUG", "TRUE") + setup_logging() + assert logging.getLogger().level == logging.DEBUG + + def test_debug_env_other_values_stay_info(self, monkeypatch): + # Pinned: only the literal (case-insensitive) "true" enables debug; + # "1" does not. + monkeypatch.setenv("LEDMATRIX_DEBUG", "1") + setup_logging() + assert logging.getLogger().level == logging.INFO + + def test_explicit_level_wins_over_env(self, monkeypatch): + monkeypatch.setenv("LEDMATRIX_DEBUG", "true") + setup_logging(level=logging.WARNING) + assert logging.getLogger().level == logging.WARNING + + +class TestLogWithContext: + def _capture(self, name): + logger = logging.getLogger(name) + records = [] + handler = logging.Handler() + handler.emit = records.append + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + return logger, records + + def test_context_attrs_stamped(self): + logger, records = self._capture("test.lwc1") + log_with_context(logger, logging.INFO, "msg", + context={"k": "v"}, plugin_id="clock", + operation_id="op-1") + record = records[0] + assert record.context == {"k": "v"} + assert record.plugin_id == "clock" + assert record.operation_id == "op-1" + + def test_wrappers_use_their_levels(self): + logger, records = self._capture("test.lwc2") + log_debug(logger, "d") + log_info(logger, "i") + log_warning(logger, "w") + assert [r.levelno for r in records] == [ + logging.DEBUG, logging.INFO, logging.WARNING] + + def test_log_error_defaults_exc_info_true(self): + logger, records = self._capture("test.lwc3") + try: + raise ValueError("kaboom") + except ValueError: + log_error(logger, "failed") + assert records[0].levelno == logging.ERROR + assert records[0].exc_info is not None + + def test_log_error_accepts_explicit_exc_info(self): + # Regression: the old hardcoded exc_info=True raised + # "got multiple values for keyword argument 'exc_info'". + logger, records = self._capture("test.lwc4") + log_error(logger, "failed", exc_info=False) + # Falsy exc_info is stored verbatim on the record; the contract is + # simply "no traceback attached". + assert not records[0].exc_info diff --git a/test/test_saved_repositories.py b/test/test_saved_repositories.py new file mode 100644 index 00000000..66a069cf --- /dev/null +++ b/test/test_saved_repositories.py @@ -0,0 +1,223 @@ +""" +Tests for src/plugin_system/saved_repositories.py — pins the +SavedRepositoriesManager contract. + +Covers: the three accepted on-disk load shapes (bare list, wrapped +{"repositories": [...]}, anything else -> []) and that saves always write +the bare-list form; add/remove/has round trips through a fresh manager; +URL normalization post-fix (_clean_url strips only a TRAILING '.git' after +trailing slashes — the old unanchored .replace('.git', '') mangled URLs +like my.github.io); name derivation and registry-vs-single type +classification (the ledmatrix-plugins check is lowercased, the +plugins.json check is case-sensitive); and the post-fix rollback of the +in-memory list when _save_repositories() fails, so memory never diverges +from disk. +""" + +import json + +from src.plugin_system.saved_repositories import SavedRepositoriesManager + + +def make_manager(path): + return SavedRepositoriesManager(config_path=str(path)) + + +class TestLoading: + def test_missing_file_empty_and_not_created(self, tmp_path): + path = tmp_path / "repos.json" + manager = make_manager(path) + assert manager.get_all() == [] + assert not path.exists() + + def test_bare_list_shape(self, tmp_path): + path = tmp_path / "repos.json" + entries = [{'url': 'https://github.com/u/r', 'name': 'r', 'type': 'single'}] + path.write_text(json.dumps(entries)) + assert make_manager(path).get_all() == entries + + def test_wrapped_dict_shape(self, tmp_path): + path = tmp_path / "repos.json" + entries = [{'url': 'https://github.com/u/r', 'name': 'r', 'type': 'single'}] + path.write_text(json.dumps({'repositories': entries})) + assert make_manager(path).get_all() == entries + + def test_other_shape_yields_empty(self, tmp_path): + path = tmp_path / "repos.json" + path.write_text(json.dumps({'x': 1})) + assert make_manager(path).get_all() == [] + + def test_malformed_json_yields_empty_no_raise(self, tmp_path): + path = tmp_path / "repos.json" + path.write_text("not json {{") + assert make_manager(path).get_all() == [] + + +class TestSaveFormat: + def test_save_always_writes_bare_list(self, tmp_path): + # Even when loaded from the wrapped {"repositories": [...]} form, + # the next save normalizes the file to a bare JSON list. + path = tmp_path / "repos.json" + entries = [{'url': 'https://github.com/u/r', 'name': 'r', 'type': 'single'}] + path.write_text(json.dumps({'repositories': entries})) + manager = make_manager(path) + assert manager.add("https://github.com/u/r2") is True + on_disk = json.loads(path.read_text()) + assert isinstance(on_disk, list) + assert len(on_disk) == 2 + + +class TestAdd: + def test_round_trip_creates_parents_and_reloads(self, tmp_path): + path = tmp_path / "sub" / "repos.json" + manager = make_manager(path) + assert manager.add("https://github.com/user/repo") is True + assert path.exists() + entry = manager.get_all()[0] + assert entry['url'] == "https://github.com/user/repo" + assert entry['name'] == "repo" + assert entry['type'] == "single" + # A fresh manager on the same path sees the persisted entry. + fresh = make_manager(path) + assert fresh.get_all() == [entry] + + def test_duplicate_returns_false_file_unchanged(self, tmp_path): + path = tmp_path / "repos.json" + manager = make_manager(path) + assert manager.add("https://github.com/user/repo") is True + before = path.read_text() + assert manager.add("https://github.com/user/repo") is False + assert path.read_text() == before + assert len(manager.get_all()) == 1 + + def test_trailing_git_and_slash_stripped(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + assert manager.add("https://github.com/user/repo.git/") is True + assert manager.get_all()[0]['url'] == "https://github.com/user/repo" + + def test_interior_dot_git_not_mangled(self, tmp_path): + # Regression for the old unanchored .replace('.git', ''): a URL + # merely CONTAINING '.git' must be stored verbatim. + manager = make_manager(tmp_path / "repos.json") + url = "https://github.com/user/my.github.io" + assert manager.add(url) is True + assert manager.get_all()[0]['url'] == url + + +class TestNameExtraction: + def test_name_derived_from_last_path_segment(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + manager.add("https://github.com/user/football-scoreboard") + assert manager.get_all()[0]['name'] == "football-scoreboard" + + def test_explicit_name_preserved(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + manager.add("https://github.com/user/repo", name="My Repo") + assert manager.get_all()[0]['name'] == "My Repo" + + def test_url_without_slash_uses_whole_url(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + manager.add("standalone") + assert manager.get_all()[0]['name'] == "standalone" + + +class TestTypeClassification: + def _type_of(self, tmp_path, url): + manager = make_manager(tmp_path / "repos.json") + assert manager.add(url) is True + return manager.get_all()[0]['type'] + + def test_plugins_json_url_is_registry(self, tmp_path): + url = "https://raw.githubusercontent.com/x/main/plugins.json" + assert self._type_of(tmp_path, url) == "registry" + + def test_ledmatrix_plugins_check_is_case_insensitive(self, tmp_path): + url = "https://github.com/ChuckBuilds/LEDMATRIX-PLUGINS" + assert self._type_of(tmp_path, url) == "registry" + + def test_plugins_json_check_is_case_sensitive(self, tmp_path): + # Only the 'ledmatrix-plugins' check is lowercased; the + # 'plugins.json' substring check is case-sensitive. Pinned. + url = "https://example.com/PLUGINS.JSON" + assert self._type_of(tmp_path, url) == "single" + + def test_plain_repo_is_single(self, tmp_path): + assert self._type_of(tmp_path, "https://github.com/user/repo") == "single" + + def test_get_registry_repositories_filters(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + manager.add("https://github.com/user/repo") + manager.add("https://raw.githubusercontent.com/x/main/plugins.json") + registries = manager.get_registry_repositories() + assert len(registries) == 1 + assert registries[0]['type'] == "registry" + + +class TestRemove: + def test_remove_present_persists(self, tmp_path): + path = tmp_path / "repos.json" + manager = make_manager(path) + manager.add("https://github.com/user/repo") + assert manager.remove("https://github.com/user/repo") is True + assert manager.get_all() == [] + assert make_manager(path).get_all() == [] + + def test_remove_absent_false_no_write(self, tmp_path): + path = tmp_path / "repos.json" + manager = make_manager(path) + manager.add("https://github.com/user/repo") + before = path.read_text() + assert manager.remove("https://github.com/user/other") is False + assert path.read_text() == before + + def test_remove_with_dirty_url_matches_clean_stored(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + manager.add("https://github.com/user/repo") + assert manager.remove("https://github.com/user/repo.git/") is True + assert manager.get_all() == [] + + +class TestHas: + def test_has_applies_url_cleaning(self, tmp_path): + manager = make_manager(tmp_path / "repos.json") + manager.add("https://x/y") + assert manager.has("https://x/y.git/") is True + assert manager.has("https://x/z") is False + + +class TestSaveFailureRollback: + def test_add_rolls_back_on_save_failure(self, tmp_path, monkeypatch): + # Post-fix: a failed save must not leave a phantom in-memory entry. + path = tmp_path / "repos.json" + manager = make_manager(path) + monkeypatch.setattr(manager, "_save_repositories", lambda: False) + assert manager.add("https://github.com/user/repo") is False + assert manager.get_all() == [] + assert not path.exists() + + def test_remove_rolls_back_on_save_failure(self, tmp_path, monkeypatch): + path = tmp_path / "repos.json" + manager = make_manager(path) + manager.add("https://github.com/user/repo") # real save + monkeypatch.setattr(manager, "_save_repositories", lambda: False) + assert manager.remove("https://github.com/user/repo") is False + assert manager.get_all() == [ + {'url': 'https://github.com/user/repo', 'name': 'repo', 'type': 'single'} + ] + # Disk still has the entry too — memory and disk stay in sync. + assert len(json.loads(path.read_text())) == 1 + + +class TestGetAllCopy: + def test_get_all_is_shallow_copy(self, tmp_path): + # Characterization: get_all() copies the LIST but not the entry + # dicts, so mutating a returned entry mutates internal state. + # Appending to the returned list, however, does not. Do not "fix" + # without auditing callers that rely on list-copy semantics. + manager = make_manager(tmp_path / "repos.json") + manager.add("https://github.com/user/repo") + returned = manager.get_all() + returned.append({'url': 'x'}) + assert len(manager.get_all()) == 1 # list itself is copied + manager.get_all()[0]['name'] = 'hacked' + assert manager.get_all()[0]['name'] == 'hacked' # dicts are shared diff --git a/test/test_skin_runtime_cache.py b/test/test_skin_runtime_cache.py new file mode 100644 index 00000000..460e1902 --- /dev/null +++ b/test/test_skin_runtime_cache.py @@ -0,0 +1,377 @@ +"""Gap tests for src/skin_system/skin_runtime.py: the discovery cache, +module namespacing internals, API gating edge cases, and targeting. + +test/test_skin_system.py already covers discovery validation, load_skin +basics, and build_context — nothing here duplicates those. + +NOTE: every test uses a UNIQUE skin id. load_skin caches the entry +module in sys.modules per skin id and never re-executes it, so reusing +an id across tests would silently serve another test's module. +""" + +import builtins +import json +import os +import sys +import time +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +# skin_runtime -> skin_base can transitively reach hardware modules via +# sports imports in sibling tests' processes; stub the matrix driver +# before importing, matching test_skin_system.py. +sys.modules.setdefault("rgbmatrix", MagicMock()) + +from src.skin_system import skin_runtime +from src.skin_system.skin_base import SKIN_API_VERSION, ScoreboardSkin + + +DEFAULT_BODY = ( + "from src.skin_system.skin_base import ScoreboardSkin\n" + "class {cls}(ScoreboardSkin):\n" + " def render_live(self, ctx, game):\n" + " return True\n" +) + + +@pytest.fixture(autouse=True) +def _clean_runtime_state(): + """Clear the discovery cache and any skin modules this test creates.""" + skin_runtime._discovery_cache.clear() + before = {k for k in sys.modules if k.startswith("_skin_")} + yield + skin_runtime._discovery_cache.clear() + created = [k for k in sys.modules + if k.startswith("_skin_") and k not in before] + for k in created: + sys.modules.pop(k, None) + + +def make_skin(skins_dir: Path, skin_id: str, *, + api_version: str = SKIN_API_VERSION, + class_name: str = "TestSkin", + body: str = None, + extra_files: dict = None, + entry_point: str = None, + manifest_id: str = None, + manifest_extra: dict = None, + write_entry: bool = True) -> Path: + """Write a skin package directory and return its path.""" + skin_dir = skins_dir / skin_id + skin_dir.mkdir(parents=True, exist_ok=True) + manifest = { + "id": manifest_id or skin_id, + "name": skin_id, + "version": "1.0.0", + "skin_api_version": api_version, + "class_name": class_name, + } + if entry_point: + manifest["entry_point"] = entry_point + manifest.update(manifest_extra or {}) + (skin_dir / "skin.json").write_text(json.dumps(manifest)) + if write_entry: + entry_name = entry_point or "skin.py" + (skin_dir / entry_name).write_text( + body if body is not None else DEFAULT_BODY.format(cls=class_name)) + for name, content in (extra_files or {}).items(): + (skin_dir / name).write_text(content) + return skin_dir + + +def counting_read_manifest(monkeypatch): + """Wrap skin_runtime._read_manifest with a call counter.""" + original = skin_runtime._read_manifest + counter = {"count": 0} + + def wrapper(skin_dir): + counter["count"] += 1 + return original(skin_dir) + + monkeypatch.setattr(skin_runtime, "_read_manifest", wrapper) + return counter + + +def bump_mtime(path: Path, offset: float = 100.0): + """Set a distinct, strictly later mtime so the fingerprint changes.""" + t = time.time() + offset + os.utime(path, (t, t)) + + +# --------------------------------------------------------------------------- +# A. Discovery cache +# --------------------------------------------------------------------------- + +class TestDiscoveryCache: + def test_second_call_serves_cache(self, tmp_path, monkeypatch): + make_skin(tmp_path, "t01-cache-hit") + counter = counting_read_manifest(monkeypatch) + first = skin_runtime.discover_skins(tmp_path) + count_after_first = counter["count"] + assert count_after_first >= 1 + second = skin_runtime.discover_skins(tmp_path) + assert counter["count"] == count_after_first # no re-read + assert second == first + assert "t01-cache-hit" in second + + def test_manifest_edit_invalidates_without_force_refresh(self, tmp_path): + skin_dir = make_skin(tmp_path, "t02-edit") + skins = skin_runtime.discover_skins(tmp_path) + assert skins["t02-edit"]["name"] == "t02-edit" + + manifest_path = skin_dir / "skin.json" + manifest = json.loads(manifest_path.read_text()) + manifest["name"] = "renamed" + manifest_path.write_text(json.dumps(manifest)) + bump_mtime(manifest_path) + + skins = skin_runtime.discover_skins(tmp_path) # no force_refresh + assert skins["t02-edit"]["name"] == "renamed" + + def test_new_skin_dir_invalidates(self, tmp_path): + make_skin(tmp_path, "t03-first") + assert set(skin_runtime.discover_skins(tmp_path)) == {"t03-first"} + + new_dir = make_skin(tmp_path, "t03-second") + bump_mtime(new_dir / "skin.json") + bump_mtime(tmp_path) + + skins = skin_runtime.discover_skins(tmp_path) # no force_refresh + assert set(skins) == {"t03-first", "t03-second"} + + def test_py_file_change_does_not_invalidate(self, tmp_path, monkeypatch): + # PIN: the fingerprint only globs */skin.json — editing a skin's + # .py file alone does NOT invalidate the cache; the cached + # manifests are still served (a code change needs a restart). + skin_dir = make_skin(tmp_path, "t04-pyedit") + counter = counting_read_manifest(monkeypatch) + skin_runtime.discover_skins(tmp_path) + count_after_first = counter["count"] + + (skin_dir / "skin.py").write_text("# rewritten\n" + + DEFAULT_BODY.format(cls="TestSkin")) + bump_mtime(skin_dir / "skin.py") + + skins = skin_runtime.discover_skins(tmp_path) + assert counter["count"] == count_after_first # cache still served + assert "t04-pyedit" in skins + + def test_force_refresh_rereads_with_unchanged_fingerprint(self, tmp_path, + monkeypatch): + make_skin(tmp_path, "t05-force") + counter = counting_read_manifest(monkeypatch) + skin_runtime.discover_skins(tmp_path) + count_after_first = counter["count"] + skin_runtime.discover_skins(tmp_path, force_refresh=True) + assert counter["count"] > count_after_first + + def test_result_mapping_is_copy_but_manifests_shared(self, tmp_path): + make_skin(tmp_path, "t06-copy") + result = skin_runtime.discover_skins(tmp_path) + + # Mutating the returned mapping does not poison the cache... + del result["t06-copy"] + again = skin_runtime.discover_skins(tmp_path) # cache hit + assert "t06-copy" in again + + # ...but the inner manifest dicts ARE shared with the cache (pin). + again["t06-copy"]["name"] = "mutated-inner" + third = skin_runtime.discover_skins(tmp_path) # cache hit + assert third["t06-copy"]["name"] == "mutated-inner" + + def test_missing_directory_returns_empty_and_caches_nothing(self, tmp_path): + missing = tmp_path / "not-yet" + assert skin_runtime.discover_skins(missing) == {} + assert str(missing) not in skin_runtime._discovery_cache + + # Creating the directory later is picked up without force_refresh. + make_skin(missing, "t07-late") + skins = skin_runtime.discover_skins(missing) + assert "t07-late" in skins + + def test_hidden_underscore_and_plain_file_entries_skipped(self, tmp_path): + make_skin(tmp_path, ".hidden-skin") + make_skin(tmp_path, "_private-skin") + (tmp_path / "stray-file").write_text("not a directory") + make_skin(tmp_path, "t08-good") + skins = skin_runtime.discover_skins(tmp_path, force_refresh=True) + assert set(skins) == {"t08-good"} + + def test_manifest_id_mismatch_keys_by_manifest_id(self, tmp_path): + make_skin(tmp_path, "t09-dirname", manifest_id="t09-manifest-id") + skins = skin_runtime.discover_skins(tmp_path, force_refresh=True) + assert "t09-manifest-id" in skins + assert "t09-dirname" not in skins + assert skins["t09-manifest-id"]["_skin_dir"].endswith("t09-dirname") + + def test_falsy_required_field_drops_skin(self, tmp_path): + make_skin(tmp_path, "t10-empty-class", class_name="") + skins = skin_runtime.discover_skins(tmp_path, force_refresh=True) + assert skins == {} + + +# --------------------------------------------------------------------------- +# B. Module namespacing (_load_skin_module via load_skin) +# --------------------------------------------------------------------------- + +BODY_WITH_HELPERS = ( + "import helpers\n" + "from src.skin_system.skin_base import ScoreboardSkin\n" + "class TestSkin(ScoreboardSkin):\n" + " pass\n" +) + + +class TestModuleNamespacing: + def test_namespaced_sys_modules_keys(self, tmp_path): + make_skin(tmp_path, "t11-ns", body=BODY_WITH_HELPERS, + extra_files={"helpers.py": "VALUE = 11\n"}) + skin = skin_runtime.load_skin("t11-ns", skins_dir=tmp_path) + assert skin is not None + assert "_skin_t11-ns_skin" in sys.modules + assert "_skin_t11-ns_helpers" in sys.modules + + def test_preseeded_bare_name_restored(self, tmp_path, monkeypatch): + sentinel = object() + monkeypatch.setitem(sys.modules, "helpers", sentinel) + make_skin(tmp_path, "t12a-restore", body=BODY_WITH_HELPERS, + extra_files={"helpers.py": "VALUE = 'a'\n"}) + skin = skin_runtime.load_skin("t12a-restore", skins_dir=tmp_path) + assert skin is not None + assert sys.modules["helpers"] is sentinel + + def test_absent_bare_name_stays_absent(self, tmp_path): + saved = sys.modules.pop("helpers", None) + try: + assert "helpers" not in sys.modules + make_skin(tmp_path, "t12b-absent", body=BODY_WITH_HELPERS, + extra_files={"helpers.py": "VALUE = 'b'\n"}) + skin = skin_runtime.load_skin("t12b-absent", skins_dir=tmp_path) + assert skin is not None + assert "helpers" not in sys.modules + finally: + if saved is not None: + sys.modules["helpers"] = saved + + def test_stdlib_shadowing_sibling_leaves_real_module_intact(self, tmp_path): + real_json = sys.modules["json"] + make_skin(tmp_path, "t12c-json", + extra_files={"json.py": "SKIN_LOCAL = True\n"}) + skin = skin_runtime.load_skin("t12c-json", skins_dir=tmp_path) + assert skin is not None + assert sys.modules["json"] is real_json + assert not hasattr(sys.modules["json"], "SKIN_LOCAL") + assert json.loads('{"ok": 1}') == {"ok": 1} # stdlib still works + # The skin's copy lives only under its namespaced alias. + assert getattr(sys.modules["_skin_t12c-json_json"], "SKIN_LOCAL") is True + + def test_entry_module_executed_once_across_loads(self, tmp_path, + monkeypatch): + executions = [] + monkeypatch.setattr(builtins, "_t13_skin_executions", executions, + raising=False) + body = ( + "import builtins\n" + "builtins._t13_skin_executions.append(1)\n" + "from src.skin_system.skin_base import ScoreboardSkin\n" + "class TestSkin(ScoreboardSkin):\n" + " pass\n" + ) + make_skin(tmp_path, "t13-cached", body=body) + for _ in range(3): + skin = skin_runtime.load_skin("t13-cached", skins_dir=tmp_path) + assert skin is not None + assert len(executions) == 1 # module executed exactly once + + def test_sibling_import_failure_returns_none_and_restores_bare( + self, tmp_path, monkeypatch): + sentinel = object() + monkeypatch.setitem(sys.modules, "helpers", sentinel) + make_skin(tmp_path, "t14-sibfail", body=BODY_WITH_HELPERS, + extra_files={"helpers.py": "raise RuntimeError('sibling boom')\n"}) + assert skin_runtime.load_skin("t14-sibfail", skins_dir=tmp_path) is None + assert sys.modules["helpers"] is sentinel + + def test_missing_entry_point_file(self, tmp_path): + make_skin(tmp_path, "t15-noentry", write_entry=False) + assert skin_runtime.load_skin("t15-noentry", skins_dir=tmp_path) is None + + def test_custom_entry_point(self, tmp_path): + make_skin(tmp_path, "t16-custom", entry_point="render.py") + skin = skin_runtime.load_skin("t16-custom", skins_dir=tmp_path) + assert isinstance(skin, ScoreboardSkin) + assert "_skin_t16-custom_render" in sys.modules + assert "_skin_t16-custom_skin" not in sys.modules + + def test_class_name_pointing_at_unrelated_class(self, tmp_path): + body = "class NotASkin:\n pass\n" + make_skin(tmp_path, "t17a-wrongclass", body=body, + class_name="NotASkin") + assert skin_runtime.load_skin("t17a-wrongclass", + skins_dir=tmp_path) is None + + def test_class_name_pointing_at_instance(self, tmp_path): + body = ( + "from src.skin_system.skin_base import ScoreboardSkin\n" + "class MySkin(ScoreboardSkin):\n" + " pass\n" + "obj = MySkin({}, {})\n" + ) + make_skin(tmp_path, "t17b-instance", body=body, class_name="obj") + assert skin_runtime.load_skin("t17b-instance", + skins_dir=tmp_path) is None + + def test_constructor_raising_returns_none(self, tmp_path): + body = ( + "from src.skin_system.skin_base import ScoreboardSkin\n" + "class TestSkin(ScoreboardSkin):\n" + " def __init__(self, manifest, options):\n" + " raise ValueError('ctor boom')\n" + ) + make_skin(tmp_path, "t18-ctor", body=body) + assert skin_runtime.load_skin("t18-ctor", skins_dir=tmp_path) is None + + +# --------------------------------------------------------------------------- +# C. API gate + targeting +# --------------------------------------------------------------------------- + +class TestApiGateAndTargeting: + def test_same_major_higher_minor_loads(self, tmp_path): + make_skin(tmp_path, "t19-minor", api_version="1.9.0") + skin = skin_runtime.load_skin("t19-minor", skins_dir=tmp_path) + assert isinstance(skin, ScoreboardSkin) + + def test_malformed_api_version_refused(self, tmp_path): + make_skin(tmp_path, "t20-malformed", api_version="abc") + assert skin_runtime.load_skin("t20-malformed", + skins_dir=tmp_path) is None + + @pytest.mark.parametrize("manifest,sport,sport_key,expected", [ + # No targets key at all -> matches everything + ({"id": "x"}, "baseball", "mlb", True), + ({"id": "x"}, None, None, True), + # Empty targets dict -> matches everything + ({"id": "x", "targets": {}}, "hockey", None, True), + # sports family match + ({"id": "x", "targets": {"sports": ["baseball"]}}, + "baseball", None, True), + # sport_keys exact match + ({"id": "x", "targets": {"sport_keys": ["milb"]}}, + None, "milb", True), + # OR semantics: sport_keys matches even though sports excludes it + ({"id": "x", "targets": {"sports": ["hockey"], + "sport_keys": ["milb"]}}, + "baseball", "milb", True), + # Neither matches + ({"id": "x", "targets": {"sports": ["hockey"]}}, + "baseball", None, False), + ({"id": "x", "targets": {"sports": ["hockey"], + "sport_keys": ["nhl"]}}, + "baseball", "milb", False), + ]) + def test_skin_matches_target(self, manifest, sport, sport_key, expected): + assert skin_runtime.skin_matches_target( + manifest, sport, sport_key) is expected diff --git a/test/test_sports_capabilities.py b/test/test_sports_capabilities.py index f43a2b61..a3cd0f3f 100644 --- a/test/test_sports_capabilities.py +++ b/test/test_sports_capabilities.py @@ -878,3 +878,200 @@ class TestCapabilityExports: def test_rotation_strategy_base_requires_a_schedule(self): with pytest.raises(NotImplementedError): RotationStrategy().schedule([game("a")]) + + +# --------------------------------------------------------------------------- +# Celebrations: rendering + previously untested edges +# --------------------------------------------------------------------------- + +from PIL import Image, ImageDraw, ImageFont # noqa: E402 + + +class _RenderableLive(_FakeLive): + """A _FakeLive that can actually execute _draw_celebration_layout: + real fonts, a display manager holding a real PIL image, and the two + SportsCore drawing seams the mixin calls.""" + + def __init__(self, mode_config=None, favorite_teams=None, + width=128, height=32, with_matrix=True): + super().__init__(mode_config=mode_config, favorite_teams=favorite_teams) + font = ImageFont.load_default() + self.fonts = {"time": font, "status": font, "score": font} + self.display_width = width + self.display_height = height + dm = MagicMock() + if with_matrix: + dm.matrix.width = width + dm.matrix.height = height + else: + dm.matrix = None + dm.image = Image.new("RGB", (width, height)) + self.display_manager = dm + self.logo_calls = [] + + def _load_and_resize_logo(self, team_id, abbr, path, url): + self.logo_calls.append(abbr) + logo = Image.new("RGBA", (10, 10), (0, 200, 0, 255)) + return logo + + def _draw_text_with_outline(self, draw, text, position, font, fill=(255, 255, 255)): + draw.text(position, str(text), font=font, fill=fill) + + +class _RenderableCelebrating(CelebrationMixin, _RenderableLive): + pass + + +def _armed(manager, *, kind="score", side="home", started_ago=0.0): + manager._start_celebration( + game("g1", home_score=7, away_score=3), kind, + scored_side=side, team_abbr="HOM", away_score=3, home_score=7, + points=7, + ) + manager.active_celebration["started_at"] = time.time() - started_ago + return manager.active_celebration + + +class TestDrawCelebrationLayout: + """The takeover render path, executed for real (previously always + mocked out).""" + + def test_renders_and_hands_frame_to_display_manager(self): + manager = _RenderableCelebrating() + celebration = _armed(manager) + manager._draw_celebration_layout(celebration) + # The final frame was assigned and pushed. + assert isinstance(manager.display_manager.image, Image.Image) + assert manager.display_manager.image.mode == "RGB" + assert manager.display_manager.image.size == (128, 32) + manager.display_manager.update_display.assert_called_once() + assert manager.display_manager.image.convert("L").getbbox() is not None + + def test_force_clear_clears_display_first(self): + manager = _RenderableCelebrating() + celebration = _armed(manager) + manager._draw_celebration_layout(celebration, force_clear=True) + manager.display_manager.clear.assert_called_once() + + def test_flash_background_within_first_window(self): + # elapsed < 1.2 with int(elapsed/0.2) even -> flash color backdrop. + manager = _RenderableCelebrating() + celebration = _armed(manager, started_ago=0.05) + manager._draw_celebration_layout(celebration) + flash = manager.display_manager.image + # After the flash window: plain black backdrop. + celebration["started_at"] = time.time() - 5 + manager._draw_celebration_layout(celebration) + steady = manager.display_manager.image + # Corner pixels (away from logos/text) show the two backgrounds. + assert flash.getpixel((64, 30)) != steady.getpixel((64, 30)) or \ + flash.getpixel((3, 0)) != steady.getpixel((3, 0)) + + def test_matrix_dims_fallback_to_display_attrs(self): + manager = _RenderableCelebrating(width=96, height=48, with_matrix=False) + celebration = _armed(manager) + manager._draw_celebration_layout(celebration) + assert manager.display_manager.image.size == (96, 48) + + def test_highlight_color_alternates_with_elapsed(self): + manager = _RenderableCelebrating() + celebration = _armed(manager) + # int(elapsed*4) % 2 == 0 -> yellow; == 1 -> orange. Force each phase + # and diff the frames. + celebration["started_at"] = time.time() - 2.0 # 8 -> even + manager._draw_celebration_layout(celebration) + even = manager.display_manager.image.tobytes() + celebration["started_at"] = time.time() - 2.25 # 9 -> odd + manager._draw_celebration_layout(celebration) + odd = manager.display_manager.image.tobytes() + assert even != odd + + def test_logo_failure_still_renders_text(self): + manager = _RenderableCelebrating() + + def boom(*a, **k): + raise RuntimeError("disk gone") + + manager._load_and_resize_logo = boom + celebration = _armed(manager, started_ago=5) # steady background + manager._draw_celebration_layout(celebration) # must not raise + assert manager.display_manager.image.convert("L").getbbox() is not None + manager.display_manager.update_display.assert_called_once() + + +class TestCelebrationEdges: + def test_should_celebrate_for_three_way_branch(self, celebrating): + g = game("g1", home="FAV", away="OPP") + favored = celebrating(favorites=["FAV"]) + assert favored._should_celebrate_for(g, "home") is True # favorite + assert favored._should_celebrate_for(g, "away") is False # opponent + favored.celebrate_opponent_scores = True + assert favored._should_celebrate_for(g, "away") is True # opted in + unconfigured = celebrating(favorites=[]) + assert unconfigured._should_celebrate_for(g, "away") is True # no favs + + def test_active_celebration_boundary_is_strict(self, celebrating): + manager = celebrating(mode_config={"celebration_duration": 3}) + manager.active_celebration = {"started_at": time.time() - 3.0} + # elapsed == duration -> strictly-less-than comparison says done. + assert manager.has_active_celebration() is False + manager.active_celebration = None + assert manager.has_active_celebration() is False + + @pytest.mark.parametrize("value,expected", [ + ({"value": None}, None), # int(float(None)) TypeError -> caught + ({"value": "abc"}, None), + ({"other": 1}, 0), # neither key -> default 0 + ([3], None), # list -> TypeError -> caught + ("-4", None), # regex fallback finds digits -> 4? No: + ]) + def test_score_to_int_edges(self, value, expected): + result = CelebrationMixin._score_to_int(value) + if value == "-4": + # int(float("-4")) parses directly: -4. + assert result == -4 + else: + assert result == expected + + def test_both_teams_scoring_prefers_away(self, celebrating): + manager = celebrating(favorites=[]) + manager._check_for_score(game("g1", home_score=0, away_score=0)) + manager._check_for_score(game("g1", home_score=7, away_score=3)) + assert manager.active_celebration["scored_side"] == "away" + + def test_away_not_celebratable_falls_through_to_home(self, celebrating): + manager = celebrating(favorites=["HOM"]) # away is the opponent + manager._check_for_score(game("g1", home_score=0, away_score=0)) + manager._check_for_score(game("g1", home_score=7, away_score=3)) + assert manager.active_celebration["scored_side"] == "home" + + def test_coalesce_expired_celebration_fires_fresh(self, celebrating): + manager = celebrating(cls=_Coalescing, + mode_config={"celebration_duration": 1}) + manager._check_for_score(game("g1")) + manager._check_for_score(game("g1", home_score=6)) + first = manager.active_celebration + assert first is not None + first["started_at"] = time.time() - 2 # expired + manager._check_for_score(game("g1", home_score=7)) + # A new celebration replaced the expired one (coalescing only + # suppresses while one is actively on screen). + assert manager.active_celebration is not first + assert manager.active_celebration["home_score"] == 7 + + def test_disabled_win_check_preserves_baseline(self, celebrating): + manager = celebrating(favorites=["HOM"]) + manager._check_for_score(game("g1")) + assert "g1" in manager._score_baselines + manager.celebration_enabled = False + manager._check_for_win(game("g1", home_score=7)) + # Early return BEFORE consuming the baseline: re-enabling later can + # still fire for this game. + assert "g1" in manager._score_baselines + + def test_prune_drops_baselines_for_idless_live_games(self, celebrating): + manager = celebrating() + manager._score_baselines = {"g1": {"away": 0, "home": 0}} + manager.prune_score_baselines([{"no_id_here": True}]) + # live ids collapse to {None}; g1 is not live -> dropped. + assert manager._score_baselines == {} diff --git a/test/test_startup_validator.py b/test/test_startup_validator.py new file mode 100644 index 00000000..43cc2f38 --- /dev/null +++ b/test/test_startup_validator.py @@ -0,0 +1,293 @@ +""" +Tests for src/startup_validator.py — pins the StartupValidator contract. + +Covers: required-key/config error reporting (errors never propagate out of +validate_all), the load_config/get_config accessor split, cache-directory +error-vs-warning downgrade behavior, plugin discovery/manifest checks with +reserved config keys skipped, idempotent validate_all (fresh error/warning +lists each run — the pre-fix behavior duplicated messages), and the +exception classification precedence in raise_on_errors (config > cache > +plugin > fallback ConfigError). +""" + +import copy +import os +from unittest.mock import MagicMock + +import pytest + +from src.exceptions import CacheError, ConfigError, PluginError +from src.startup_validator import StartupValidator + +GOOD_CONFIG = { + 'display': {'hardware': {'rows': 32, 'cols': 64}}, + 'timezone': 'UTC', +} + + +def make_config_manager(config): + """Config manager whose load_config() and get_config() return `config`.""" + mgr = MagicMock() + mgr.load_config.return_value = copy.deepcopy(config) + mgr.get_config.return_value = copy.deepcopy(config) + return mgr + + +@pytest.fixture +def good_cache(monkeypatch, tmp_path): + """Patch CacheManager so cache validation sees an existing writable dir. + + _validate_cache_directory does `from src.cache_manager import CacheManager` + at call time, so patching the attribute on the module is picked up. + """ + mock_cls = MagicMock() + mock_cls.return_value.get_cache_dir.return_value = str(tmp_path) + monkeypatch.setattr("src.cache_manager.CacheManager", mock_cls) + return tmp_path + + +class TestValidateConfig: + """Configuration validation via load_config().""" + + def test_happy_path(self, good_cache): + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert errors == [] + assert warnings == [] + + def test_missing_required_keys(self, good_cache): + validator = StartupValidator(make_config_manager({})) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + assert "Missing required configuration key: display" in errors + assert "Missing required configuration key: timezone" in errors + + def test_config_error_does_not_propagate(self, good_cache): + mgr = make_config_manager(GOOD_CONFIG) + mgr.load_config.side_effect = ConfigError("bad json") + validator = StartupValidator(mgr) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + config_errors = [e for e in errors if e.startswith("Configuration error:")] + assert len(config_errors) == 1 + assert "bad json" in config_errors[0] + + def test_unexpected_error_does_not_propagate(self, good_cache): + mgr = make_config_manager(GOOD_CONFIG) + mgr.load_config.side_effect = RuntimeError("kapow") + validator = StartupValidator(mgr) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + unexpected = [e for e in errors + if e.startswith("Unexpected error validating configuration:")] + assert len(unexpected) == 1 + assert "kapow" in unexpected[0] + + def test_accessor_split_get_config_failure_is_warning_only(self, good_cache): + # _validate_config uses load_config(); _validate_display_config uses + # get_config(). A broken get_config must degrade to a warning, not + # crash or produce a config error. + mgr = make_config_manager(GOOD_CONFIG) + mgr.get_config.side_effect = RuntimeError("accessor broken") + validator = StartupValidator(mgr) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert errors == [] + assert any(w.startswith("Could not validate display configuration:") + for w in warnings) + assert mgr.load_config.called + assert mgr.get_config.called + + +class TestDisplayConfig: + """Display hardware validation via get_config().""" + + def test_missing_hardware_section_is_error(self, good_cache): + config = {'display': {'runtime': {'gpio_slowdown': 2}}, 'timezone': 'UTC'} + validator = StartupValidator(make_config_manager(config)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + assert "Display hardware configuration is missing" in errors + + def test_missing_rows_cols_are_warnings_not_errors(self, good_cache): + config = {'display': {'hardware': {'brightness': 90}}, 'timezone': 'UTC'} + validator = StartupValidator(make_config_manager(config)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert errors == [] + assert "Display hardware setting 'rows' not specified, using default" in warnings + assert "Display hardware setting 'cols' not specified, using default" in warnings + + +class TestCacheDirectory: + """Cache directory validation error/warning split.""" + + def _patch_cache_dir(self, monkeypatch, cache_dir): + mock_cls = MagicMock() + mock_cls.return_value.get_cache_dir.return_value = cache_dir + monkeypatch.setattr("src.cache_manager.CacheManager", mock_cls) + + def test_nonexistent_cache_dir_is_error(self, monkeypatch, tmp_path): + missing = str(tmp_path / "does_not_exist") + self._patch_cache_dir(monkeypatch, missing) + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + assert any("does not exist" in e and missing in e for e in errors) + + def test_writable_cache_dir_no_errors(self, monkeypatch, tmp_path): + self._patch_cache_dir(monkeypatch, str(tmp_path)) + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert not any('cache' in e.lower() for e in errors) + + def test_unwritable_cache_dir_is_error(self, monkeypatch, tmp_path): + # Root (common in CI) can write anywhere, so chmod tricks don't + # work — force os.access to deny writes for the cache dir only. + cache_dir = str(tmp_path) + self._patch_cache_dir(monkeypatch, cache_dir) + real_access = os.access + + def fake_access(path, mode): + if str(path) == cache_dir and mode == os.W_OK: + return False + return real_access(path, mode) + + monkeypatch.setattr(os, "access", fake_access) + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + assert any("is not writable" in e for e in errors) + + def test_none_cache_dir_is_warning_not_error(self, monkeypatch): + self._patch_cache_dir(monkeypatch, None) + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert errors == [] + assert "Cache directory not available - caching will be disabled" in warnings + + def test_cache_manager_constructor_failure_is_warning(self, monkeypatch): + mock_cls = MagicMock(side_effect=RuntimeError("no disk")) + monkeypatch.setattr("src.cache_manager.CacheManager", mock_cls) + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert errors == [] + assert any(w.startswith("Could not validate cache directory:") for w in warnings) + + +class TestPlugins: + """Plugin validation with a plugin manager present.""" + + def _config_with_plugins(self): + return { + 'display': {'hardware': {'rows': 32, 'cols': 64}}, + 'schedule': {'enabled': True}, # reserved key that LOOKS enabled + 'timezone': 'UTC', + 'plugin_system': {}, + 'known': {'enabled': True}, + 'ghost': {'enabled': True}, + } + + def test_ghost_plugin_warns_and_reserved_keys_skipped(self, good_cache, tmp_path): + pm = MagicMock() + pm.discover_plugins.return_value = ['known'] + known_dir = tmp_path / "known" + known_dir.mkdir() + (known_dir / "manifest.json").write_text("{}") + pm.get_plugin_directory.return_value = str(known_dir) + + validator = StartupValidator(make_config_manager(self._config_with_plugins()), pm) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert "Plugin 'ghost' is enabled but not found in plugins directory" in warnings + # Reserved sections are never treated as plugins, even when they + # contain an 'enabled' flag (schedule above). + for reserved in ('display', 'schedule', 'timezone', 'plugin_system'): + assert not any(f"'{reserved}'" in w for w in warnings) + + def test_enabled_plugin_missing_manifest_is_error(self, good_cache, tmp_path): + pm = MagicMock() + pm.discover_plugins.return_value = ['known'] + plugin_dir = tmp_path / "known" + plugin_dir.mkdir() # exists, but no manifest.json inside + pm.get_plugin_directory.return_value = str(plugin_dir) + + config = dict(GOOD_CONFIG, known={'enabled': True}) + validator = StartupValidator(make_config_manager(config), pm) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is False + assert "Plugin 'known' manifest.json not found" in errors + + def test_disabled_plugin_not_checked_for_manifest(self, good_cache, tmp_path): + pm = MagicMock() + pm.discover_plugins.return_value = ['known'] + pm.get_plugin_directory.return_value = str(tmp_path / "nowhere") + + config = dict(GOOD_CONFIG, known={'enabled': False}) + validator = StartupValidator(make_config_manager(config), pm) + is_valid, errors, warnings = validator.validate_all() + assert is_valid is True + assert errors == [] + + +class TestIdempotence: + """validate_all() resets error/warning state each run (the fixed bug).""" + + def test_repeated_runs_do_not_accumulate(self, good_cache): + validator = StartupValidator(make_config_manager({})) + first = validator.validate_all() + second = validator.validate_all() + assert first == second + assert len(second[1]) == len(first[1]) + assert len(second[2]) == len(first[2]) + + +class TestRaiseOnErrors: + """Exception classification and precedence in raise_on_errors().""" + + def _validator(self, errors): + validator = StartupValidator(make_config_manager(GOOD_CONFIG)) + validator.errors = list(errors) + return validator + + def test_no_errors_returns_none(self): + assert self._validator([]).raise_on_errors() is None + + def test_config_error(self): + msg = "Missing required configuration key: display" + with pytest.raises(ConfigError) as excinfo: + self._validator([msg]).raise_on_errors() + assert excinfo.value.message == "Configuration validation failed" + assert msg in excinfo.value.context['errors'] + + def test_cache_error(self): + msg = "Cache directory does not exist: /nope" + with pytest.raises(CacheError) as excinfo: + self._validator([msg]).raise_on_errors() + assert msg in excinfo.value.context['errors'] + + def test_plugin_error(self): + msg = "Plugin 'known' manifest.json not found" + with pytest.raises(PluginError) as excinfo: + self._validator([msg]).raise_on_errors() + assert msg in excinfo.value.context['errors'] + + def test_unclassified_error_falls_back_to_config_error(self): + msg = "Something entirely else went wrong" + with pytest.raises(ConfigError) as excinfo: + self._validator([msg]).raise_on_errors() + assert excinfo.value.message == "Startup validation failed" + assert msg in excinfo.value.context['errors'] + + def test_precedence_config_beats_cache(self): + # A message matching both 'config' and 'cache' substrings raises + # ConfigError because config classification is checked first. + msg = "config problem touching the cache layer" + with pytest.raises(ConfigError) as excinfo: + self._validator([msg]).raise_on_errors() + assert excinfo.value.message == "Configuration validation failed" + assert msg in excinfo.value.context['errors']