# -*- coding: utf-8 -*- """Focused offline operation-count, snapshot and menu-invalidation regressions.""" import importlib import json import os import sys import tempfile import unittest from datetime import date, timedelta from pathlib import Path from unittest.mock import patch sys.path.insert(0, os.path.dirname(__file__)) import stubs stubs.install() import xbmc import xbmcaddon import xbmcgui import xbmcplugin from wcs.library2 import listings, menu, meta, next_up, rows, store, tmdb from wcs.skin import refresh, row_cache TODAY = date(2026, 10, 4) class RowPerformanceTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix='milionar-row-performance-') self.addCleanup(self.tmp.cleanup) previous_db = store.db_path() self.addCleanup(store.set_path, previous_db) store.set_path(os.path.join(self.tmp.name, 'library.db')) settings, strings = dict(xbmcaddon.Addon.settings), dict(xbmcaddon.Addon.strings) self.addCleanup(self.restore, xbmcaddon.Addon.settings, settings) self.addCleanup(self.restore, xbmcaddon.Addon.strings, strings) xbmcaddon.Addon.settings.clear() xbmcaddon.Addon.strings.update({32400: 'Resume {}', 32404: 'Next {}', 32403: 'Source', 32405: 'Watched', 32406: 'Unwatched', 32408: 'Add', 32409: 'Remove'}) for table in (xbmcplugin.ITEMS, xbmcplugin.ENDED, xbmcplugin.CONTENT, xbmcplugin.SORTS): table.clear() self.day, self.language = TODAY, 'cs' fake_date = type('D', (), {'today': staticmethod(lambda: self.day)}) self.enter(patch.object(rows, 'date', fake_date)) self.enter(patch.object(listings, 'date', fake_date)) self.enter(patch.object(xbmc, 'getLanguage', side_effect=lambda *args, **kwargs: self.language)) self.enter(patch.object(tmdb, '_fetch', side_effect=AssertionError('unexpected provider call'))) self.requests = self.enter(patch.object(refresh, 'request')) def enter(self, context): result = context.__enter__() self.addCleanup(context.__exit__, None, None, None) return result @staticmethod def restore(target, source): target.clear() target.update(source) def movies(self, count=100): for i in range(1, count + 1): store.set_in_list('movie', i, True, {'id': i, 'title': 'Film {:03d}'.format(i)}) meta.put(meta.key_movie(i), {'title': 'Film {:03d}'.format(i), 'poster_path': '/p.jpg', 'genres': [{'id': i, 'name': 'Genre {:03d}'.format(i)}]}, 86400) def show(self, i=1000, episode_count=120): store.set_in_list('tv', i, True, {'id': i, 'name': 'Show {}'.format(i)}) meta.put(meta.key_tv(i), {'name': 'Show {}'.format(i), 'status': 'Returning Series', 'seasons': [{'season_number': 1}], 'poster_path': '/p.jpg', 'genres': [{'id': i, 'name': 'Genre {}'.format(i)}]}, 86400) meta.put(meta.key_season(i, 1), {'episodes': [ {'episode_number': e, 'name': 'Ep {}'.format(e), 'air_date': (TODAY + timedelta(days=e - episode_count)).isoformat()} for e in range(1, episode_count + 1)]}, 86400) def build(self, row='my_movies_all', **params): return rows.build(row, dict(params, row=row)) def snapshot_files(self): return list(Path(self.tmp.name, 'row_snapshots').glob('*.json')) @staticmethod def signature(items): return [(p, i.label, i.art, i.props, i.tag.values, i.context, f) for p, i, f in items] def test_100_movies_limit_10_is_bulk_and_constructs_only_10_offscreen_items(self): self.movies() with patch.object(store, 'connect', wraps=store.connect) as reads, \ patch.object(meta, 'get_many', wraps=meta.get_many) as metadata, \ patch.object(xbmcgui, 'ListItem', wraps=xbmcgui.ListItem) as constructed: items = self.build(limit='10') self.assertEqual(len(items), 10) self.assertEqual(constructed.call_count, 10) self.assertTrue(all(call.kwargs['offscreen'] for call in constructed.call_args_list)) self.assertEqual(metadata.call_count, 1) self.assertEqual(reads.call_count, 5) # 3 data reads + 2 persistent revision checks self.requests.assert_not_called() def test_warm_snapshot_survives_module_reload_and_reads_only_the_revision(self): self.movies() cold = self.build(limit='10', click='play', sort='random') importlib.reload(row_cache) # no in-memory cache can satisfy the next invocation with patch.object(rows.Data, 'titles', side_effect=AssertionError('warm title read')), \ patch.object(meta, 'get_many', side_effect=AssertionError('warm metadata read')), \ patch.object(store, 'get_movie_states', side_effect=AssertionError('warm state read')), \ patch.object(menu.Menu, 'build', side_effect=AssertionError('warm menu rebuild')), \ patch.object(store, 'connect', wraps=store.connect) as reads: warm = self.build(limit='10', click='play', sort='random') self.assertEqual(self.signature(cold), self.signature(warm)) self.assertIsNot(cold[0][1], warm[0][1]) self.assertEqual(reads.call_count, 1) def test_committed_mutation_rebuilds_menu_without_changing_selection_url(self): self.movies(1) before = self.build(click='play')[0] self.assertIn('Watched', [label for label, _cmd in before[1].context]) store.mark_watched('movie', 1) after = self.build(click='play')[0] self.assertEqual(before[0], after[0]) self.assertIn('Unwatched', [label for label, _cmd in after[1].context]) self.assertNotIn('Watched', [label for label, _cmd in after[1].context]) self.assertEqual(after[1].tag.values['Playcount'], 1) def test_db_date_language_profile_and_config_invalidate(self): self.movies(1) self.build() with patch.object(rows, '_build', wraps=rows._build) as rebuild: self.build() self.assertEqual(rebuild.call_count, 0) self.day += timedelta(days=1) self.build() self.language = 'en' self.build() self.build(style='landscape') original_info = xbmcaddon.Addon.getAddonInfo with patch.object(xbmcaddon.Addon, 'getAddonInfo', lambda addon, key: 'another-profile/' if key == 'profile' else original_info(addon, key)): self.build() self.assertEqual(rebuild.call_count, 4) store.set_path(os.path.join(self.tmp.name, 'other.db')) self.assertEqual(self.build(), []) self.assertEqual(rebuild.call_count, 5) def test_account_scope_cannot_reuse_another_accounts_snapshot(self): scope = {'db': store.db_path(), 'profile': 'profile', 'date': TODAY.isoformat(), 'language': 'cs', 'account': '1'} first = row_cache.key('my_movies_all', {'limit': '10'}, scope) self.assertTrue(row_cache.write(first, [])) self.assertEqual(row_cache.read(first), []) second = row_cache.key('my_movies_all', {'limit': '10'}, dict(scope, account='2')) self.assertIsNone(row_cache.read(second)) def test_safe_settings_invalidate_without_a_revision_change_and_never_read_credentials(self): self.movies(1) self.build() versions = refresh.revisions(['my_movies_all']) setting = xbmcaddon.Addon.getSetting allowed = set(rows.ROW_SETTINGS) | {'skin_low_power'} def safe_read(addon, name): self.assertIn(name, allowed) return setting(addon, name) with patch.object(xbmcaddon.Addon, 'getSetting', safe_read), \ patch.object(rows, '_build', wraps=rows._build) as rebuild: self.build() self.assertEqual(rebuild.call_count, 0) for name in rows.ROW_SETTINGS: xbmcaddon.Addon.settings[name] = '1' if name == 'default_movie_action' else 'true' self.build() self.assertEqual(rebuild.call_count, len(rows.ROW_SETTINGS)) self.build() self.assertEqual(rebuild.call_count, len(rows.ROW_SETTINGS)) from wcs import provider_keys before_ai = provider_keys.flag('ai') with patch.object(provider_keys, 'flag', return_value=not before_ai): self.build() self.assertEqual(rebuild.call_count, len(rows.ROW_SETTINGS) + 1) self.assertEqual(refresh.revisions(['my_movies_all']), versions) for path in self.snapshot_files(): config = json.loads(path.read_text())['identity']['scope']['config'] self.assertEqual(set(config), set(rows.ROW_SETTINGS)) def test_limits_clamp_negative_and_preserve_zero_and_valid_normal_limits(self): self.movies(3) for value in (-10, '-1', 0, '0'): with self.subTest(value=value), patch.object(xbmcgui, 'ListItem', wraps=xbmcgui.ListItem) as made: self.assertEqual(self.build(limit=value), []) made.assert_not_called() self.assertEqual(len(self.build(limit='100')), 3) self.assertEqual(rows._limit({'limit': 'invalid'}), 20) xbmcaddon.Addon.settings['skin_low_power'] = 'true' self.assertEqual(rows._limit({'limit': '-1'}), 0) def test_midnight_rebuilds_upcoming_episode_as_aired(self): self.show(episode_count=2) meta.put(meta.key_season(1000, 1), {'episodes': [ {'episode_number': 1, 'air_date': TODAY.isoformat()}, {'episode_number': 2, 'air_date': (TODAY + timedelta(days=1)).isoformat()}]}, 86400) store.mark_watched('tv', 1000, 1, 1) first = self.build('my_series_next')[0] self.assertIn('mupcoming', first[0]) self.day += timedelta(days=1) second = self.build('my_series_next')[0] self.assertNotIn('mupcoming', second[0]) self.assertEqual(second[1].tag.values['Episode'], 2) self.assertTrue(second[1].context) def test_genres_and_search_limit_before_item_construction(self): self.movies(30) for row, params in (('my_movies_genres', {}), ('search_my_movies', {'q': 'film'})): with self.subTest(row=row), patch.object(xbmcgui, 'ListItem', wraps=xbmcgui.ListItem) as made: self.assertEqual(len(self.build(row, limit='3', **params)), 3) self.assertEqual(made.call_count, 3) for i in range(1000, 1005): self.show(i, episode_count=2) for row, params in (('my_series_genres', {}), ('search_my_series', {'q': 'show'})): with self.subTest(row=row), patch.object(xbmcgui, 'ListItem', wraps=xbmcgui.ListItem) as made: self.assertEqual(len(self.build(row, limit='2', **params)), 2) self.assertEqual(made.call_count, 2) def test_series_snapshot_is_compact_and_reuses_counts_and_episode_menus(self): self.show() first = self.build('my_series_all') payload = json.loads(self.snapshot_files()[0].read_text()) card = payload['items'][0]['card'] self.assertEqual(card['counts'], [120, 0]) self.assertFalse({'show', 'states', 'order', 'state'} & card.keys()) with patch.object(next_up, 'episode_order', side_effect=AssertionError('warm episode rebuild')): self.assertEqual(self.signature(first), self.signature(self.build('my_series_all'))) store.save_resume('tv', 1000, 300, 2400, 1, 2) episode = self.build('cont_series') self.assertEqual(episode[0][1].tag.values['ResumePoint'], (300.0, 2400.0)) self.assertEqual(self.signature(episode), self.signature(self.build('cont_series'))) def test_low_power_caps_before_construction_and_changes_artwork_only_at_render(self): self.movies(30) normal = self.build(limit='20') self.assertEqual(len(normal), 20) xbmcaddon.Addon.settings['skin_low_power'] = 'true' with patch.object(xbmcgui, 'ListItem', wraps=xbmcgui.ListItem) as made: light = self.build(limit='20') self.assertEqual(made.call_count, 10) self.assertEqual(len(light), 10) self.assertIn('/w342/', light[0][1].art['poster']) self.assertEqual(light[0][0], normal[0][0]) xbmcaddon.Addon.settings['skin_low_power'] = 'false' self.assertIn('/w500/', self.build(limit='10')[0][1].art['poster']) def test_success_empty_is_cached_but_missing_metadata_requests_library_work(self): self.assertEqual(self.build(), []) with patch.object(rows, '_build', side_effect=AssertionError('empty cache miss')): self.assertEqual(self.build(), []) store.set_in_list('movie', 1, True, {'id': 1, 'title': 'Snapshot'}) self.assertEqual(self.build()[0][1].label, 'Snapshot') self.requests.assert_called_with(['library:movie:1']) with patch.object(rows, '_build', wraps=rows._build) as rebuilt: self.build() self.assertEqual(rebuilt.call_count, 1) meta.put(meta.key_movie(1), {'title': 'Now cached', 'genres': []}, 86400) self.assertEqual(self.build()[0][1].label, 'Now cached') def test_missing_season_requests_show_work_and_never_caches_transient_empty(self): self.show(episode_count=2) meta.delete(meta.key_season(1000, 1)) self.assertEqual(self.build('my_series_next'), []) self.requests.assert_called_with(['library:tv:1000']) self.assertEqual(self.snapshot_files(), []) def test_partial_metadata_does_not_cache_unknown_genres_as_success_empty(self): self.movies(1) meta.put(meta.key_movie(1), {'title': 'Partial'}, 86400) self.assertEqual(self.build('my_movies_genres'), []) self.requests.assert_called_with(['library:movie:1']) self.assertEqual(self.snapshot_files(), []) def test_corruption_read_write_and_revision_failures_fall_back(self): self.movies(1) expected = self.signature(self.build()) self.snapshot_files()[0].write_text('{broken') self.assertEqual(self.signature(self.build()), expected) payload = json.loads(self.snapshot_files()[0].read_text()) payload['items'] = ['broken descriptor'] self.snapshot_files()[0].write_text(json.dumps(payload)) self.assertEqual(self.signature(self.build()), expected) with patch.object(row_cache, 'open', side_effect=OSError('read failure'), create=True): self.assertEqual(self.signature(self.build()), expected) with patch.object(row_cache, '_revision', side_effect=OSError('revision failure')): self.assertEqual(self.signature(self.build()), expected) with patch.object(row_cache.os, 'replace', side_effect=OSError('write failure')): self.assertEqual(self.signature(self.build(sort='rating')), expected) self.assertFalse(list(Path(self.tmp.name, 'row_snapshots').glob('*.tmp'))) def test_racing_mutation_cannot_publish_old_snapshot(self): scope = {'db': store.db_path(), 'profile': 'profile', 'date': TODAY.isoformat(), 'language': 'cs'} key = row_cache.key('my_movies_all', {}, scope) with patch.object(row_cache, '_revision', return_value=key[1]['revision'] + 1): self.assertFalse(row_cache.write(key, [])) self.assertEqual(self.snapshot_files(), []) self.assertFalse(list(Path(self.tmp.name, 'row_snapshots').glob('*.tmp'))) def test_unknown_revision_disables_cache_and_search_variants_are_bounded(self): scope = {'db': store.db_path(), 'profile': 'profile', 'date': TODAY.isoformat(), 'language': 'cs'} with patch.object(refresh, 'revisions', return_value={}): self.assertIsNone(row_cache.key('my_movies_all', {}, scope)) with patch.object(row_cache, 'MAX_FILES', 2): for query in ('a', 'b', 'c'): self.assertTrue(row_cache.write(row_cache.key('search_my_movies', {'q': query}, scope), [])) self.assertEqual(len(self.snapshot_files()), 2) def test_listing_order_and_search_fallback_labels_stay_stable(self): self.movies(3) with store.tx() as conn: conn.execute('UPDATE titles SET list_pos=0, list_added_at=100 WHERE tmdb_id=1') conn.execute('UPDATE titles SET list_pos=0, list_added_at=50 WHERE tmdb_id=2') conn.execute('UPDATE titles SET list_pos=NULL WHERE tmdb_id=3') self.assertEqual([e['id'] for e in rows.Data().entries('movie')], [e['id'] for e in store.get_list('movie')]) store.set_in_list('movie', 4, True, {'id': 4}) self.assertIn('Film 4', [i.label for _p, i, _f in self.build('search_my_movies', q='film')]) def test_legacy_dispatch_preserves_all_params(self): from wcs.skin import widgets params = {'row': 'widget_csfd_charts', 'limit': '3', 'sort': 'title', 'style': 'categories', 'click': 'detail', 'hw': '1', 'up': '0', 'ctx': 'slot-1', 'custom': 'retained'} with patch.object(widgets, 'run') as delegated: rows.run(1, params) delegated.assert_called_once_with(1, dict(params, action='widget_csfd_charts')) def test_movie_listing_reuses_metadata_menu_data_and_bulk_logos(self): self.movies(100) for genre, count in ((None, 100), ('1', 1)): xbmcplugin.ITEMS.clear() with self.subTest(genre=genre), patch.object(store, 'connect', wraps=store.connect) as reads, \ patch.object(meta, 'get_many', wraps=meta.get_many) as metadata, \ patch.object(xbmcgui, 'ListItem', wraps=xbmcgui.ListItem) as made, \ patch.object(menu, 'Menu', wraps=menu.Menu) as models: listings.movies(1, genre=genre) titles = [i for _p, i, _f in xbmcplugin.ITEMS[1] if i.getProperty('milionar.type') == 'movie'] self.assertEqual(len(titles), count) self.assertEqual(metadata.call_count, 1) self.assertEqual(models.call_count, 1) self.assertTrue(all(call.kwargs['offscreen'] for call in made.call_args_list)) self.assertLessEqual(reads.call_count, 4) self.assertTrue(all(item.context for item in titles)) def test_series_listing_reuses_episode_order_for_item_and_menu(self): for i in range(1000, 1005): self.show(i, episode_count=10) with patch.object(next_up, 'episode_order', wraps=next_up.episode_order) as orders, \ patch.object(meta, 'get_many', wraps=meta.get_many) as metadata, \ patch.object(store, 'connect', wraps=store.connect) as reads: listings.series(1, genre='1000') self.assertEqual(orders.call_count, 1) self.assertEqual(metadata.call_count, 2) self.assertLessEqual(reads.call_count, 5) self.assertEqual(xbmcplugin.ITEMS[1][0][1].props['TotalEpisodes'], '10') if __name__ == '__main__': unittest.main()