"""Focused offline network-widget cache tests; providers and Kodi are synthetic.""" import ast import json import os import sys import tempfile import time import types import unittest from pathlib import Path from unittest.mock import patch HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import stubs stubs.install() import xbmc import xbmcaddon import wcs.metadata from wcs.library2 import store from wcs.metadata import TMDbCache from wcs.skin import refresher def isolated_fetch(requests): """Load the actual owned function without importing the large interactive TMDb module.""" path = Path(stubs.ADDON_DIR) / 'wcs/metadata/TMDbClient.py' definition = next(node for node in ast.parse(path.read_text()).body if isinstance(node, ast.FunctionDef) and node.name == 'fetch_tmdb_items') namespace = {'requests': requests, 'get_tmdb_api_key': lambda addon: addon.getSetting('tmdb_api_key') or missing_key()} exec(compile(ast.Module(body=[definition], type_ignores=[]), str(path), 'exec'), namespace) return namespace['fetch_tmdb_items'] def missing_key(): raise ValueError('TMDb key missing') class RequestFailure(Exception): pass class SkinCachePerformanceTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory(prefix='skin-cache-performance-') self.addCleanup(self.directory.cleanup) self.addCleanup(patch.stopall) patch.object(refresher, 'PROFILE', self.directory.name).start() patch.object(refresher, 'CACHE_DIR', os.path.join(self.directory.name, 'skin_cache')).start() patch.object(store, '_override_path', os.path.join(self.directory.name, 'library.db')).start() patch.object(TMDbCache, '_purged', False).start() patch.object(xbmcaddon.Addon, 'settings', {'tmdb_api_key': 'synthetic', 'sc_auth_token': 'synthetic'}).start() self.addon = xbmcaddon.Addon() self.changed, self.calls = [], [] revision = types.ModuleType('wcs.skin.refresh') revision.changed = lambda names: self.changed.append(list(names)) patch.dict(sys.modules, {'wcs.skin.refresh': revision}).start() self.payloads = {} self.fail_paths = set() requests = types.SimpleNamespace(get=self.http, exceptions=types.SimpleNamespace(RequestException=RequestFailure)) self.fetch = isolated_fetch(requests) provider = types.ModuleType('wcs.metadata.TMDbClient') provider.fetch_tmdb_items = self.fetch patch.dict(sys.modules, {'wcs.metadata.TMDbClient': provider}).start() patch.object(wcs.metadata, 'TMDbClient', provider, create=True).start() self.kraska = {'menu': []} sc = types.ModuleType('wcs.streamcinema.SCClient') sc._api_get = lambda *args, **kwargs: self.kraska sc_package = types.ModuleType('wcs.streamcinema') sc_package.SCClient = sc patch.dict(sys.modules, {'wcs.streamcinema': sc_package, 'wcs.streamcinema.SCClient': sc}).start() def http(self, url, params, timeout): endpoint = url.split('/3/', 1)[1] self.calls.append((endpoint, dict(params))) if endpoint in self.fail_paths: raise RequestFailure('synthetic failure') payload = self.payloads.get(endpoint, {'results': []}) return types.SimpleNamespace(raise_for_status=lambda: None, json=lambda: payload) def read(self, name): with open(refresher.cache_path(name), encoding='utf-8') as handle: return json.load(handle) def raw(self, name): return Path(refresher.cache_path(name)).read_bytes() def cinemas(self, *movies): Path(self.directory.name, 'mycinema_channels.json').write_text(json.dumps( [{'media_list': [{'id': identity} for identity in movies]}])) def row(self, identity, **extra): return dict({'id': identity, 'title': 'Movie {}'.format(identity), 'poster_path': '/poster.jpg', 'release_date': '2026-01-01', 'overview': 'Plot', 'vote_average': 8}, **extra) def age_movie_cache(self, identity, seconds): key = TMDbCache._make_key('movie/{}/recommendations'.format(identity), {'language': 'cs-CZ'}) with store.tx() as conn: conn.execute('UPDATE meta SET fetched_at=? WHERE key=?', (int(time.time() - seconds), key)) def test_unique_atomic_writes_keep_previous_visible_until_replace_and_only_named_revision(self): name = 'widget_tmdb_popular_movies' old, new = [{'title': 'Old'}], [{'title': 'New'}] self.assertTrue(refresher.write_cache(name, old)) self.changed.clear() replace = os.replace temporary_paths = [] def atomic(source, target): temporary_paths.append(source) self.assertNotEqual(source, target + '.tmp') self.assertEqual(self.read(name)['items'], old) self.assertEqual(json.loads(Path(source).read_text())['items'], new) replace(source, target) with patch.object(refresher.os, 'replace', atomic): self.assertTrue(refresher.write_cache(name, new)) self.assertEqual(self.changed, [[name]]) self.assertEqual(self.read(name)['items'], new) self.assertTrue(all(not os.path.exists(path) for path in temporary_paths)) def test_noop_renews_cache_without_revision_and_empty_success_is_published(self): name = 'widget_mycinema_recommended' refresher.write_cache(name, [{'title': 'Old'}]) self.changed.clear() self.assertFalse(refresher.write_cache(name, [{'title': 'Old'}])) self.assertEqual(self.changed, []) self.assertTrue(refresher.write_cache(name, [])) self.assertEqual(self.changed, [[name]]) self.changed.clear() self.assertFalse(refresher.write_cache(name, [])) self.assertEqual(self.changed, []) self.assertEqual(self.read(name)['items'], []) def test_unique_temporary_names_and_write_failure_preserves_last_snapshot(self): name = 'widget_tmdb_popular_movies' refresher.write_cache(name, [{'title': 'Old'}]) before = self.raw(name) self.changed.clear() paths = [] def fail(source, target): paths.append(source) raise OSError('synthetic write failure') for _attempt in range(2): with patch.object(refresher.os, 'replace', fail), self.assertRaises(OSError): refresher.write_cache(name, []) self.assertEqual(len(set(paths)), 2) self.assertEqual(self.raw(name), before) self.assertEqual(self.changed, []) self.assertTrue(all(not os.path.exists(path) for path in paths)) def test_nonlist_cache_payload_rejected_before_mutation(self): with self.assertRaises(ValueError): refresher.write_cache('widget_tmdb_popular_movies', None) self.assertEqual(self.changed, []) def test_tmdb_refresh_empty_success_and_failure_keeps_old_cache(self): name = 'widget_tmdb_popular_movies' self.payloads['movie/popular'] = {'results': [self.row(1)]} self.assertTrue(refresher.refresh(self.addon, name)) before = self.raw(name) self.changed.clear() self.fail_paths.add('movie/popular') self.assertFalse(refresher.refresh(self.addon, name)) self.assertEqual(self.raw(name), before) self.assertEqual(self.changed, []) self.fail_paths.clear() self.payloads['movie/popular'] = {'results': []} self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual(self.read(name)['items'], []) self.assertEqual(self.changed, [[name]]) def test_strict_is_optin_and_valid_empty_results_and_parts_are_supported(self): self.fail_paths.add('movie/popular') self.assertEqual(self.fetch('movie', 'movie/popular', {}, self.addon), []) with self.assertRaises(RequestFailure): self.fetch('movie', 'movie/popular', {}, self.addon, strict=True) self.fail_paths.clear() self.assertEqual(self.fetch('movie', 'movie/popular', {}, self.addon, strict=True), []) self.payloads['collection/1'] = {'parts': [self.row(1)]} self.assertEqual(self.fetch('collection', 'collection/1', {}, self.addon, strict=True), [self.row(1)]) def test_strict_rejects_malformed_and_error_payloads_without_empty_publication(self): name = 'widget_tmdb_popular_movies' refresher.write_cache(name, [{'title': 'Old'}]) before = self.raw(name) self.changed.clear() for payload in ({}, [], None, {'results': None}, {'results': {}}, {'results': [None]}, {'results': [{}]}, {'results': [], 'success': False}): with self.subTest(payload=payload): self.payloads['movie/popular'] = payload with self.assertRaises(ValueError): self.fetch('movie', 'movie/popular', {}, self.addon, strict=True) self.assertFalse(refresher.refresh(self.addon, name)) self.assertEqual(self.raw(name), before) self.assertEqual(self.changed, []) def test_tmdb_params_remain_compatible_and_unmodified(self): params = {'page': 2, 'language': 'en-US'} self.fetch('movie', 'movie/popular', params, self.addon, strict=True) self.assertEqual(params, {'page': 2, 'language': 'en-US'}) self.assertEqual(self.calls[-1][1], {'page': 2, 'language': 'en-US', 'api_key': 'synthetic'}) xbmcaddon.Addon.settings.pop('tmdb_api_key') with self.assertRaises(ValueError): self.fetch('movie', 'movie/popular', {}, self.addon) def test_mycinema_deduplicates_sources_and_votes_and_warm_cache_avoids_provider(self): Path(self.directory.name, 'mycinema_channels.json').write_text(json.dumps([ {'media_list': [{'id': 1}, {'id': '1'}, {'id': 2}]}, {'media_list': [{'id': 1}]}])) self.payloads['movie/1/recommendations'] = {'results': [self.row(20), self.row(20), self.row(21), self.row(2)]} self.payloads['movie/2/recommendations'] = {'results': [self.row(20), self.row(22, vote_average=9)]} name = 'widget_mycinema_recommended' self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual([endpoint for endpoint, _params in self.calls], ['movie/1/recommendations', 'movie/2/recommendations']) self.assertEqual([row['tmdb_id'] for row in self.read(name)['items']], [20, 22, 21]) self.calls.clear() self.changed.clear() self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual(self.calls, []) self.assertEqual(self.changed, []) def test_mycinema_empty_per_movie_response_is_cached_and_last_movie_removal_publishes_empty(self): self.cinemas(1) self.assertTrue(refresher.refresh(self.addon, 'widget_mycinema_recommended')) self.calls.clear() self.assertTrue(refresher.refresh(self.addon, 'widget_mycinema_recommended')) self.assertEqual(self.calls, []) self.payloads['movie/2/recommendations'] = {'results': [self.row(20)]} self.cinemas(2) self.assertTrue(refresher.refresh(self.addon, 'widget_mycinema_recommended')) self.cinemas() xbmcaddon.Addon.settings.pop('tmdb_api_key') self.calls.clear() self.changed.clear() self.assertTrue(refresher.refresh(self.addon, 'widget_mycinema_recommended')) self.assertEqual(self.read('widget_mycinema_recommended')['items'], []) self.assertEqual(self.changed, [['widget_mycinema_recommended']]) self.assertEqual(self.calls, []) def test_recommendation_ttl_is_24h_and_failure_retains_stale_generic_and_widget_cache(self): self.cinemas(1) self.payloads['movie/1/recommendations'] = {'results': [self.row(20)]} name = 'widget_mycinema_recommended' self.assertTrue(refresher.refresh(self.addon, name)) before = self.raw(name) self.age_movie_cache(1, 86401) self.fail_paths.add('movie/1/recommendations') self.changed.clear() self.calls.clear() self.assertFalse(refresher.refresh(self.addon, name)) self.assertEqual(len(self.calls), 1) self.assertEqual(self.raw(name), before) self.assertEqual(self.changed, []) endpoint, params = 'movie/1/recommendations', {'language': 'cs-CZ'} self.assertIsNone(TMDbCache.get_from_cache(endpoint, params, expire_seconds=86400)) self.assertEqual(TMDbCache.get_from_cache(endpoint, params, expire_seconds=TMDbCache.KEEP_SECONDS), [self.row(20)]) self.fail_paths.clear() self.payloads[endpoint] = {'results': []} self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual(self.read(name)['items'], []) def test_partial_recommendations_failure_retries_only_uncached_movie_without_publishing_partial(self): self.cinemas(1, 2) name = 'widget_mycinema_recommended' refresher.write_cache(name, [{'title': 'Old'}]) before = self.raw(name) self.payloads['movie/1/recommendations'] = {'results': [self.row(20)]} self.fail_paths.add('movie/2/recommendations') self.assertFalse(refresher.refresh(self.addon, name)) self.assertEqual(self.raw(name), before) self.calls.clear() self.fail_paths.clear() self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual([endpoint for endpoint, _params in self.calls], ['movie/2/recommendations']) def test_malformed_cinema_file_is_failure_and_preserves_widget_cache(self): name = 'widget_mycinema_recommended' refresher.write_cache(name, [{'title': 'Old'}]) before = self.raw(name) Path(self.directory.name, 'mycinema_channels.json').write_text('{invalid') self.assertFalse(refresher.refresh(self.addon, name)) self.assertEqual(self.raw(name), before) def test_kraska_success_empty_and_failed_transport_are_distinct(self): name = 'widget_kraska_root' self.kraska = {'menu': [{'type': 'dir', 'title': 'Movies', 'url': '/movies', 'art': {'icon': 'cover'}}]} self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual(self.read(name)['items'], [{'title': 'Movies', 'path': '/movies', 'icon': 'cover'}]) before = self.raw(name) self.changed.clear() for payload in (None, {}, {'menu': None}, {'menu': [None]}): self.kraska = payload self.assertFalse(refresher.refresh(self.addon, name)) self.assertEqual(self.raw(name), before) self.kraska = {'menu': []} self.assertTrue(refresher.refresh(self.addon, name)) self.assertEqual(self.read(name)['items'], []) self.assertEqual(self.changed, [[name]]) def test_active_rows_reads_all_sixty_slots_filters_and_deduplicates(self): labels = [] def value(label): labels.append(label) return {'Skin.String(mil.cont.6.type)': 'vinyl_songs', 'Skin.String(mil.vinyl.6.type)': 'vinyl_songs', 'Skin.String(mil.tmdb.1.type)': 'widget_tmdb_popular_movies', 'Skin.String(mil.mytv.1.type)': 'widget_mytv_channels', 'Skin.String(mil.my_movies.1.type)': 'my_movies_all'}.get(label, '') with patch.object(xbmc, 'getInfoLabel', value): self.assertEqual(refresher.active_rows(), {'vinyl_songs', 'widget_tmdb_popular_movies'}) self.assertEqual(len(labels), 60) def test_refresh_due_only_active_due_rows_and_reports_boolean_results(self): active = {'widget_tmdb_popular_movies', 'vinyl_songs'} with patch.object(refresher, 'active_rows', return_value=active), patch.object(refresher, 'cache_age', return_value=None), patch.object( refresher, 'refresh', side_effect=lambda addon, name: name == 'vinyl_songs') as refresh: self.assertEqual(refresher.refresh_due(), {'vinyl_songs': True, 'widget_tmdb_popular_movies': False}) self.assertEqual({call.args[1] for call in refresh.call_args_list}, active) with patch.object(refresher, 'active_rows', return_value=active), patch.object(refresher, 'cache_age', return_value=10), patch.object( refresher, 'refresh') as refresh: self.assertEqual(refresher.refresh_due(), {}) refresh.assert_not_called() def test_refresh_due_video_gates_routine_work_and_force_names_remain_compatible(self): active = {'vinyl_songs', 'widget_tmdb_popular_movies'} with patch.object(refresher, 'active_rows', return_value=active), patch.object(refresher, 'cache_age', return_value=None), patch.object( xbmc, 'getCondVisibility', return_value=True), patch.object(refresher, 'refresh', return_value=True) as refresh: self.assertEqual(refresher.refresh_due(), {}) refresh.assert_not_called() self.assertEqual(refresher.refresh_due(force=['vinyl']), {'vinyl_songs': True}) self.assertEqual(refresher.refresh_due(force=['widget_mycinema_recommended']), {'widget_mycinema_recommended': True}) def test_named_vinyl_refresh_and_age_use_exact_account_source(self): from wcs.vinyl import client, widgets with patch.object(widgets, 'refresh', return_value=True) as refresh: self.assertTrue(refresher.refresh(self.addon, 'vinyl_favorite_songs')) refresh.assert_called_once_with(sources=['favorite_songs']) with patch.object(client, 'PROFILE', self.directory.name): xbmcaddon.Addon.settings.update(vinyl_token='synthetic', vinyl_user_id='11') self.assertEqual(refresher.cache_path('vinyl_songs'), client.cache_path('songs')) self.assertIsNone(refresher.cache_age('vinyl_songs')) path = Path(client.cache_path('songs')) path.parent.mkdir(parents=True, exist_ok=True) path.write_text('{"items": []}') os.utime(path, (time.time() - 10, time.time() - 10)) self.assertGreaterEqual(refresher.cache_age('vinyl_songs'), 10) self.assertFalse(refresher.refresh(self.addon, 'unknown')) if __name__ == '__main__': unittest.main()