# -*- coding: utf-8 -*- """Focused offline refresh contracts. No Kodi process, threads, RPC or providers. Run: python3 scripts/skin_harness/test_refresh_service.py The real scheduler runs against bounded virtual time and manually completed jobs. """ import contextlib import importlib import json import os from pathlib import Path import queue import sys import tempfile import unittest from datetime import date, timedelta from unittest.mock import Mock, patch from urllib.parse import quote, unquote sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'library_harness')) import stubs stubs.install() import xbmc import xbmcaddon import xbmcgui from wcs.library2 import store from wcs.skin import refresh, refresher, service @contextlib.contextmanager def other_threads(): """The service's helper threads that these tests do not start (one context: a `with` takes only so many).""" with patch('wcs.skin.perflog.start'), patch('wcs.skin.commands.start'), patch('wcs.skin.bytecode.start'): yield MOVIES = 'widget_tmdb_popular_movies' SERIES = 'widget_tmdb_popular_series' def payload(names, force=False): return quote(json.dumps({'names': list(names), 'force': force}, ensure_ascii=False), safe='') class Loop: """A bounded scheduler driver; queued jobs stay inflight until complete() is called.""" def __init__(self, steps=8, active=(), due=False): self.steps, self.active, self.due = steps, set(active), due self.step, self.now = 0, 0.0 self.day, self.home_visible = date(2026, 10, 4), False self.conditions, self.condition_reads = {}, [] self.pending_effect = None self.shared_pending = False # the MyTV "shared progress" switch has work for the service self.monitor = service.Monitor() self.monitor.abortRequested = lambda: self.step >= self.steps self.monitor.waitForAbort = self.wait self.jobs = self.results = None self.enqueued, self.commands, self.properties = [], [], [] self.at = {} self.thread = Mock() self.waits = [] self.bootstrap_effect = self.properties_effect = self.publish_effect = self.revision_effect = None self.versions = {name: 0 for name in refresh.ALL_ROWS} def event(self, method, names=(), force=False): self.monitor.onNotification(service.ADDON_ID, 'Other.' + method, payload(names, force)) def complete(self, name, generation=1, ok=True): self.results.put((name, generation, ok)) def start_thread(self, **kwargs): self.jobs, self.results = kwargs['args'][1:] return self.thread def wait(self, interval): if interval not in (0.25, 1): raise AssertionError('Unexpected blocking scheduler wait') self.waits.append(interval) while self.jobs is not None: try: name, generation = self.jobs.get_nowait() self.enqueued.append((self.now, name, generation)) except queue.Empty: break self.properties.append((self.now, dict(xbmcgui.Window.props))) self.step += 1 self.now += interval if self.step in self.at: self.at[self.step](self) return self.step >= self.steps def run(self, expect_worker=True): # All peripheral work is replaced before invoking the real scheduler. from wcs.library2 import migration fake_date = type('D', (), {'today': staticmethod(lambda: self.day)}) def visible(key): self.condition_reads.append(key) return self.conditions.get(key, self.home_visible if key == 'Window.IsActive(home)' else False) with patch.object(service, 'Monitor', return_value=self.monitor), \ patch.object(service.threading, 'Thread', side_effect=self.start_thread), \ patch('wcs.skin.warm.start'), patch('wcs.skin.journal.start'), other_threads(), \ patch.object(service.time, 'monotonic', side_effect=lambda: self.now), \ patch.object(service.time, 'time', side_effect=lambda: 1700000000 + self.now), \ patch.object(service, 'date', fake_date), \ patch.object(service, '_properties', side_effect=self.properties_effect) as props, \ patch.object(service, '_active', side_effect=lambda: self.active), \ patch.object(service, '_due', return_value=self.due), \ patch.object(service.shared, 'pending', side_effect=lambda: self.shared_pending), \ patch.object(service, '_build', side_effect=AssertionError('unexpected live worker')), \ patch.object(refresh, 'bootstrap', side_effect=self.bootstrap_effect) as boot, \ patch.object(refresh, 'revisions', side_effect=self.revision_effect or (lambda names: {name: self.versions.get(name, 0) for name in names})), \ patch.object(refresh, 'changed') as changed, \ patch.object(refresh, 'publish', side_effect=self.publish_effect) as published, \ patch.object(migration, 'pending', side_effect=self.pending_effect or (lambda: False)) as pending, \ patch.object(xbmc, 'getCondVisibility', side_effect=visible), \ patch.object(xbmc, 'executeJSONRPC', side_effect=AssertionError('unexpected RPC')), \ patch.object(xbmc, 'executebuiltin', side_effect=lambda cmd: self.commands.append((self.now, cmd))): service.run() if expect_worker: self.thread.start.assert_called_once() else: self.thread.start.assert_not_called() self.property_calls, self.changed, self.published, self.bootstraps = props.call_count, changed, published, boot.call_count self.pending_calls = pending.call_count def notices(self): return [(when, cmd) for when, cmd in self.commands if cmd.startswith('Notification(')] class RefreshServiceTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix='milionar-refresh-service-') self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) previous = store.db_path() self.addCleanup(store.set_path, previous) store.set_path(str(self.root / 'library.db')) properties = dict(xbmcgui.Window.props) strings = dict(xbmcaddon.Addon.strings) self.addCleanup(self.restore, xbmcgui.Window.props, properties) self.addCleanup(self.restore, xbmcaddon.Addon.strings, strings) xbmcgui.Window.props.clear() xbmcaddon.Addon.strings.update({32536: 'SUCCESS', 32651: 'FAILURE'}) self.builtins = self.enter(patch.object(xbmc, 'executebuiltin')) self.rpc = Mock() # RPCs other than the change notifications def rpc(text): # refresh.notify sends JSONRPC.NotifyAll; the tests read it in the builtin's form from self.builtins request = json.loads(text) if request.get('method') != 'JSONRPC.NotifyAll': self.rpc(text) raise AssertionError('unexpected RPC') params = request['params'] xbmc.executebuiltin('NotifyAll({},{},{})'.format(params['sender'], params['message'], quote( json.dumps(params['data'], separators=(',', ':')), safe=''))) self.enter(patch.object(xbmc, 'executeJSONRPC', side_effect=rpc)) self.enter(patch.object(refresher, 'CACHE_DIR', str(self.root / 'skin_cache'))) 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 test_properties_projects_low_power_and_provider_availability_to_home(self): home = xbmcgui.Window home.props['unrelated'] = 'keep' cases = ( ({'skin_low_power': 'true', 'sc_auth_token': 'synthetic-token', 'groq_api_key': 'synthetic-key', 'wsuser': 'synthetic-user'}, 'true', '1'), ({}, 'false', ''), ) for settings, low_power, available in cases: with self.subTest(low_power=low_power), \ patch.object(xbmcaddon.Addon, 'getSetting', side_effect=lambda key: settings.get(key, '')), \ patch.object(xbmcgui, 'Window', wraps=home) as windows, \ patch.object(service, '_build', side_effect=AssertionError('properties called a provider')): service._properties() self.assertTrue(windows.called) self.assertTrue(all(call.args == (10000,) for call in windows.call_args_list)) expected = { 'unrelated': 'keep', 'milionar.provider.kraska': available, 'milionar.provider.ai': available, 'milionar.provider.webshare': available, } if low_power == 'true': # off = property absent, so the row URLs stay unchanged expected['milionar.low_power'] = low_power self.assertEqual(home.props, expected) self.builtins.assert_not_called() self.rpc.assert_not_called() def test_grouped_transaction_revisions_persist_and_publish_does_not_increment(self): own, other = 'my_movies_all', 'my_series_all' self.assertEqual(refresh.revisions([own, other]), {own: 0, other: 0}) with store.tx() as conn: store.dirty(conn, [own]) store.dirty(conn, [own, other]) conn.execute("INSERT INTO titles(media_type,tmdb_id,in_list) VALUES('movie',1,1)") self.assertEqual(refresh.revisions([own, other]), {own: 0, other: 0}) self.assertEqual(refresh.revisions([own, other]), {own: 1, other: 1}) self.assertEqual(self.builtins.call_count, 1) message = self.builtins.call_args.args[0] decoded = json.loads(unquote(message.split(',', 2)[2][:-1])) self.assertEqual(set(decoded['names']), {own, other}) importlib.reload(refresh) self.assertEqual(refresh.revisions([own, other]), {own: 1, other: 1}) with patch.object(xbmc, 'getInfoLabel', return_value=own): refresh.publish([own]) self.assertEqual(refresh.revisions([own, other]), {own: 1, other: 1}) self.assertEqual(refresh.revisions([MOVIES]), {MOVIES: 0}) def test_rollback_and_repeated_mutations_are_revision_noops(self): with self.assertRaisesRegex(RuntimeError, 'rollback'): with store.tx() as conn: store.dirty(conn, ['my_movies_all']) raise RuntimeError('rollback') self.assertEqual(refresh.revisions(['my_movies_all']), {'my_movies_all': 0}) self.builtins.assert_not_called() store.set_in_list('movie', 1, True, {'id': 1, 'title': 'One'}) before = refresh.revisions(refresh.ALL_ROWS) self.builtins.reset_mock() store.set_in_list('movie', 1, True, {'id': 1, 'title': 'One'}) store.clear_resume('movie', 1) self.assertEqual(refresh.revisions(refresh.ALL_ROWS), before) self.builtins.assert_not_called() targets = [(1, number) for number in range(1, 121)] self.assertEqual(store.mark_many('tv', 2, targets), 120) marked = refresh.revisions(refresh.ALL_ROWS) self.assertEqual(marked['my_series_all'], before['my_series_all'] + 1) self.builtins.reset_mock() self.assertEqual(store.mark_many('tv', 2, targets), 0) self.assertEqual(refresh.revisions(refresh.ALL_ROWS), marked) self.builtins.assert_not_called() self.rpc.assert_not_called() def test_publish_matches_all_60_slots_and_preserves_unrelated_properties(self): own, other = 'my_movies_all', 'my_series_all' refresh.changed([own]) selected = {('cont', 6), ('my_movies', 1), ('mytv', 3)} labels = {} for section in refresh.SECTIONS: for number in range(1, 7): labels['Skin.String(mil.{}.{}.type)'.format(section, number)] = own if (section, number) in selected else other xbmcgui.Window.props['milionar.widgets.{}.{}'.format(section, number)] = 'old' xbmcgui.Window.props.update({'milionar.widgets.token': 'global-unchanged', 'unrelated': 'keep', 'milionar.rows.' + other: 'other-unchanged'}) before = refresh.revisions([own, other]) with patch.object(xbmc, 'getInfoLabel', side_effect=labels.get) as reads: refresh.publish([own]) self.assertEqual(reads.call_count, 63) # 60 slot types + the persisted revision of the 3 matching slots with patch.object(xbmcgui.Window, 'setProperty', wraps=xbmcgui.Window(10000).setProperty) as written: refresh.publish([own]) self.assertFalse(any(call.args[0].startswith('milionar.widgets.') for call in written.call_args_list)) for section in refresh.SECTIONS: for number in range(1, 7): self.assertEqual(xbmcgui.Window.props['milionar.widgets.{}.{}'.format(section, number)], '1' if (section, number) in selected else 'old') self.assertEqual(xbmcgui.Window.props['unrelated'], 'keep') self.assertEqual(xbmcgui.Window.props['milionar.widgets.token'], 'global-unchanged') self.assertEqual(xbmcgui.Window.props['milionar.rows.' + other], 'other-unchanged') self.assertEqual(refresh.revisions([own, other]), before) def test_notifications_are_deduplicated_and_parsed_with_force(self): names = [MOVIES, 'library:movie:7', 'quoted, Česky "title"', MOVIES] with patch.object(xbmc, 'executeJSONRPC') as rpc: refresh.request(names, force=True) sent = json.loads(rpc.call_args.args[0]) self.assertEqual(sent['method'], 'JSONRPC.NotifyAll') sender, method, data = sent['params']['sender'], sent['params']['message'], sent['params']['data'] self.assertEqual(data, {'names': sorted(set(names)), 'force': True}) monitor = service.Monitor() monitor.onNotification(sender, 'Other.' + method, json.dumps(data)) # Kodi hands the data over as JSON self.assertEqual(monitor.events.get_nowait(), ('skin_widget_force_refresh', set(names))) self.assertEqual(service.names_from('{"names":["one",null,3,"one"]}'), {'one'}) self.assertEqual(service.names_from('%invalid'), set()) self.assertEqual(service.names_from('[]'), set()) with patch.object(xbmc, 'executeJSONRPC') as rpc: refresh.request([]) rpc.assert_not_called() def test_monitor_callbacks_only_enqueue_and_never_access_db_rpc_or_providers(self): monitor = service.Monitor() with patch.object(store, 'connect', side_effect=AssertionError('callback DB access')), \ patch.object(service, '_properties', side_effect=AssertionError('callback settings work')), \ patch.object(service, '_build', side_effect=AssertionError('callback provider')), \ patch.object(refresher, 'refresh', side_effect=AssertionError('callback provider')), \ patch.object(refresh, 'publish', side_effect=AssertionError('callback publication')): monitor.onSettingsChanged() monitor.onNotification(service.ADDON_ID, 'Other.skin_rows_changed', payload(['my_movies_all'])) monitor.onNotification(service.ADDON_ID, 'skin_widget_refresh', payload([MOVIES])) monitor.onNotification(service.ADDON_ID, 'skin_widgets_refresh', '{}') monitor.onNotification(service.ADDON_ID, 'vinyl_refresh', payload(['vinyl_songs'])) monitor.onNotification('another.addon', 'skin_widget_refresh', payload([SERIES])) self.assertEqual([monitor.events.get_nowait()[0] for _ in range(5)], ['settings', 'skin_rows_changed', 'skin_widget_refresh', 'skin_widgets_refresh', 'vinyl_refresh']) self.assertTrue(monitor.events.empty()) self.rpc.assert_not_called() self.builtins.assert_not_called() def test_worker_failure_preserves_last_snapshot_and_revision(self): refresher.write_cache(MOVIES, [{'title': 'Last good', 'tmdb_id': 1}]) cache = Path(refresher.cache_path(MOVIES)) previous, versions = cache.read_bytes(), refresh.revisions([MOVIES]) monitor = Mock() monitor.abortRequested.side_effect = [False, True] jobs, results = queue.Queue(), queue.Queue() jobs.put((MOVIES, 7)) with patch.object(refresher, 'build_tmdb_popular', side_effect=RuntimeError('synthetic provider failure')) as build: service._worker(monitor, jobs, results) self.assertEqual(results.get_nowait(), (MOVIES, 7, False)) self.assertEqual(cache.read_bytes(), previous) self.assertEqual(refresh.revisions([MOVIES]), versions) build.assert_called_once() self.rpc.assert_not_called() self.assertFalse(any(call.args[0].startswith('Notification(') for call in self.builtins.call_args_list)) def test_success_empty_replaces_cache_but_identical_output_is_a_revision_noop(self): refresher.write_cache(MOVIES, [{'title': 'Old'}]) before = refresh.revisions([MOVIES])[MOVIES] with patch.object(refresher, 'build_tmdb_popular', return_value=[]): self.assertTrue(refresher.refresh(xbmcaddon.Addon(), MOVIES)) self.assertEqual(json.loads(Path(refresher.cache_path(MOVIES)).read_text())['items'], []) self.assertEqual(refresh.revisions([MOVIES])[MOVIES], before + 1) self.assertTrue(refresher.refresh(xbmcaddon.Addon(), MOVIES)) self.assertEqual(refresh.revisions([MOVIES])[MOVIES], before + 1) def test_manual_success_waits_for_every_result(self): xbmcgui.Window.props['milionar.widgets.refreshing'] = '1' loop = Loop(active=[MOVIES, SERIES]) loop.event('skin_widgets_refresh') loop.at[1] = lambda run: run.complete(MOVIES) loop.at[3] = lambda run: run.complete(SERIES) loop.run() self.assertEqual({(name, generation) for _when, name, generation in loop.enqueued}, {(MOVIES, 1), (SERIES, 1)}) self.assertEqual(loop.notices(), [(0.75, 'Notification(Milionář,SUCCESS)')]) self.assertTrue(all(props.get('milionar.widgets.refreshing') == '1' for when, props in loop.properties if when < 0.75)) self.assertNotIn('milionar.widgets.refreshing', xbmcgui.Window.props) self.assertEqual(xbmcgui.Window.props['milionar.widgets.refreshed'], '1700000000.75') loop.changed.assert_called_once_with(refresh.ALL_ROWS) def test_manual_failure_waits_for_remaining_jobs_and_reports_failure(self): loop = Loop(active=[MOVIES, SERIES]) loop.event('skin_widgets_refresh') loop.at[1] = lambda run: run.complete(MOVIES, ok=False) loop.at[3] = lambda run: run.complete(SERIES) loop.run() self.assertEqual(loop.notices(), [(0.75, 'Notification(Milionář,FAILURE)')]) def test_manual_refresh_during_inflight_waits_for_the_forced_generation(self): loop = Loop(active=[MOVIES]) loop.event('skin_widget_refresh', [MOVIES]) loop.at[1] = lambda run: run.event('skin_widgets_refresh') loop.at[2] = lambda run: run.complete(MOVIES, 1) loop.at[4] = lambda run: run.complete(MOVIES, 2) loop.run() self.assertEqual([(name, generation) for _when, name, generation in loop.enqueued], [(MOVIES, 1), (MOVIES, 2)]) self.assertEqual(loop.notices(), [(1.0, 'Notification(Milionář,SUCCESS)')]) def test_requests_coalesce_both_before_dispatch_and_while_inflight(self): loop = Loop(active=[MOVIES]) for _ in range(3): loop.event('skin_widget_refresh', [MOVIES, MOVIES]) loop.at[1] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.at[2] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.at[3] = lambda run: run.complete(MOVIES) loop.run() self.assertEqual(loop.enqueued, [(0.0, MOVIES, 1)]) self.assertEqual(loop.notices(), []) def test_source_force_only_queues_named_vinyl_source(self): loop = Loop(active=[MOVIES, 'vinyl_albums']) loop.event('vinyl_refresh', ['vinyl_songs']) loop.at[1] = lambda run: run.complete('vinyl_songs') loop.run() self.assertEqual(loop.enqueued, [(0.0, 'vinyl_songs', 1)]) def test_retry_window_blocks_misses_but_explicit_force_bypasses_it(self): loop = Loop(steps=50, active=[MOVIES]) loop.event('skin_widget_refresh', [MOVIES]) loop.at[1] = lambda run: run.complete(MOVIES, ok=False) for step in (2, 3, 20): loop.at[step] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.at[44] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.at[45] = lambda run: run.complete(MOVIES, 2, ok=False) loop.at[47] = lambda run: run.event('skin_widget_refresh', [MOVIES], force=True) loop.run() self.assertEqual(loop.enqueued, [(0.0, MOVIES, 1), (11.0, MOVIES, 2), (11.75, MOVIES, 3)]) def test_due_retry_is_scheduled_after_backoff_with_bounded_virtual_time(self): loop = Loop(steps=62, active=[MOVIES], due=True) loop.event('skin_widget_refresh', [MOVIES]) loop.at[1] = lambda run: run.complete(MOVIES, ok=False) loop.run() self.assertEqual(loop.enqueued, [(0.0, MOVIES, 1), (15.0, MOVIES, 2)]) def test_settings_force_active_sources_in_the_loop_and_local_changes_only_publish(self): loop = Loop(active=[MOVIES, SERIES]) for _ in range(3): loop.monitor.onSettingsChanged() loop.event('skin_rows_changed', ['my_movies_all']) loop.run() self.assertEqual({name for _when, name, _generation in loop.enqueued}, {MOVIES, SERIES}) self.assertEqual(loop.property_calls, 2) # startup + the queued settings event loop.published.assert_called_once_with({'my_movies_all'}) def test_startup_bootstrap_failure_retries_before_starting_worker(self): loop = Loop(steps=5) loop.bootstrap_effect = [OSError('synthetic DB not ready'), None] loop.run() self.assertEqual(loop.bootstraps, 2) self.assertEqual(loop.property_calls, 2) self.assertEqual(loop.waits[0], 1) self.assertEqual(loop.enqueued, []) def test_startup_failure_can_abort_without_starting_worker(self): loop = Loop(steps=2) loop.bootstrap_effect = OSError('synthetic DB not ready') loop.run(expect_worker=False) self.assertEqual(loop.bootstraps, 2) self.assertEqual(loop.waits, [1, 1]) def test_initial_revision_read_failure_also_retries_startup(self): loop = Loop(steps=5) attempts = [] def revisions(names): attempts.append(1) if len(attempts) == 1: raise OSError('synthetic DB became unavailable after bootstrap') return {name: 0 for name in names} loop.revision_effect = revisions loop.run() self.assertGreaterEqual(len(attempts), 2) def test_failed_hidden_library_job_retries_without_a_new_widget_request(self): name = 'library:tv:7' loop = Loop(steps=62, active=[]) loop.event('skin_widget_refresh', [name]) loop.at[1] = lambda run: run.complete(name, ok=False) loop.run() self.assertEqual(loop.enqueued, [(0.0, name, 1), (15.0, name, 2)]) def test_success_resets_backoff_for_the_next_failure(self): loop = Loop(steps=91, active=[MOVIES]) loop.event('skin_widget_refresh', [MOVIES]) loop.at[1] = lambda run: run.complete(MOVIES, 1, ok=False) loop.at[44] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.at[45] = lambda run: run.complete(MOVIES, 2) loop.at[46] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.at[47] = lambda run: run.complete(MOVIES, 3, ok=False) loop.at[88] = lambda run: run.event('skin_widget_refresh', [MOVIES]) loop.run() self.assertEqual(loop.enqueued, [(0.0, MOVIES, 1), (11.0, MOVIES, 2), (11.5, MOVIES, 3), (22.0, MOVIES, 4)]) def test_midnight_changes_only_calendar_dependent_rows_once(self): loop = Loop() loop.at[1] = lambda run: setattr(run, 'day', run.day + timedelta(days=1)) loop.run() loop.changed.assert_called_once_with(refresh.SERIES_ROWS | {'widget_mytv_next'}) self.assertNotIn('my_movies_all', loop.changed.call_args.args[0]) def test_failed_publication_is_recovered_by_the_revision_poll(self): loop = Loop(steps=22) loop.event('skin_rows_changed', ['my_movies_all']) calls = [] def publish(names): calls.append(names) if len(calls) == 1: loop.versions['my_movies_all'] = 1 raise OSError('synthetic publication failure') loop.publish_effect = publish loop.run() self.assertEqual(calls, [{'my_movies_all'}, {'my_movies_all'}]) def test_migration_waits_for_skin_installation_to_finish(self): xbmcgui.Window.props['milionar.skin.installing'] = '1' blocked = Loop(steps=4) blocked.pending_effect = lambda: True blocked.run() self.assertEqual(blocked.pending_calls, 0) self.assertEqual(blocked.commands, []) loop = Loop(steps=6) loop.pending_effect = lambda: True loop.at[2] = lambda run: xbmcgui.Window.props.pop('milionar.skin.installing') loop.run() self.assertEqual(loop.pending_calls, 1) commands = [(when, command) for when, command in loop.commands if 'action=library_migrate' in command] self.assertEqual(commands, [(0.5, 'RunPlugin(plugin://plugin.video.milionar/?action=library_migrate&auto=1)')]) self.assertNotIn('Window.IsActive(home)', loop.condition_reads) def test_empty_library_requests_pending_migration_immediately_without_ui_or_home_wait(self): from wcs.library2 import migration, tmdb (self.root / 'watchlist_movies.json').write_text(json.dumps([{'id': 7, 'title': 'Legacy'}])) loop = Loop() loop.pending_effect = migration.pending # exercise the real SQL-empty/discovery gate before = refresh.revisions(refresh.ALL_ROWS) with patch.object(migration, '_profile', return_value=str(self.root)), \ patch.object(migration, 'run', side_effect=AssertionError('migration UI executed')), \ patch.object(xbmcgui, 'Dialog', side_effect=AssertionError('service opened a dialog')), \ patch.object(tmdb, '_fetch', side_effect=AssertionError('service called a provider')): loop.run() self.assertEqual(loop.pending_calls, 1) self.assertEqual(loop.commands, [(0.0, 'RunPlugin(plugin://plugin.video.milionar/?action=library_migrate&auto=1)')]) self.assertEqual(loop.notices(), []) self.assertEqual(loop.enqueued, []) self.assertNotIn('Window.IsActive(home)', loop.condition_reads) self.assertEqual(refresh.revisions(refresh.ALL_ROWS), before) self.assertEqual(store.get_list('movie'), []) def test_empty_library_without_legacy_sources_is_quiet_and_checked_only_once(self): from wcs.library2 import migration loop = Loop() loop.pending_effect = migration.pending with patch.object(migration, '_profile', return_value=str(self.root)), \ patch.object(xbmcgui, 'Dialog', side_effect=AssertionError('service opened a dialog')): loop.run() self.assertEqual(loop.pending_calls, 1) self.assertEqual(loop.commands, []) self.assertEqual(loop.enqueued, []) def test_populated_library_suppresses_auto_run_before_reading_any_legacy_files(self): from wcs.library2 import migration for media in ('movie', 'tv'): with self.subTest(media=media): store.set_path(str(self.root / (media + '.db'))) store.set_in_list(media, 1, True, {'id': 1}) loop = Loop() loop.pending_effect = migration.pending with patch.object(store, 'inspect_watchlist_import', side_effect=AssertionError('legacy files read')) as inspect, \ patch.object(migration, '_profile', side_effect=AssertionError('legacy profile read')) as profile: loop.run() self.assertEqual(loop.pending_calls, 1) inspect.assert_not_called() profile.assert_not_called() self.assertEqual(loop.commands, []) def test_auto_migration_waits_for_video_or_modal_dialog_then_runs_once(self): for condition in ('Player.HasVideo', 'System.HasActiveModalDialog'): with self.subTest(condition=condition): loop = Loop(steps=6) loop.pending_effect = lambda: True loop.conditions[condition] = True loop.at[2] = lambda run, key=condition: run.conditions.pop(key) loop.run() self.assertEqual(loop.pending_calls, 1) self.assertEqual(loop.commands, [(0.5, 'RunPlugin(plugin://plugin.video.milionar/?action=library_migrate&auto=1)')]) def test_migration_discovery_exception_retries_and_then_requests_once(self): loop = Loop() loop.pending_effect = [OSError('synthetic discovery failure'), True] loop.run() self.assertEqual(loop.pending_calls, 2) self.assertEqual(loop.commands, [(0.25, 'RunPlugin(plugin://plugin.video.milionar/?action=library_migrate&auto=1)')]) self.assertEqual(loop.notices(), []) def test_migration_discovery_exception_then_no_offer_stays_quiet(self): loop = Loop() loop.pending_effect = [OSError('synthetic discovery failure'), False] loop.run() self.assertEqual(loop.pending_calls, 2) self.assertEqual(loop.commands, []) class SharedProgressSyncTests(unittest.TestCase): """The service merges the channels and the store when the MyTV "shared progress" switch has pending work.""" def setUp(self): temporary = tempfile.TemporaryDirectory(prefix='milionar-shared-service-') self.addCleanup(temporary.cleanup) previous = store.db_path() self.addCleanup(store.set_path, previous) store.set_path(os.path.join(temporary.name, 'library.db')) properties = dict(xbmcgui.Window.props) self.addCleanup(lambda: (xbmcgui.Window.props.clear(), xbmcgui.Window.props.update(properties))) xbmcgui.Window.props.clear() def test_pending_switch_is_synced_once_at_start(self): loop = Loop(steps=6) loop.shared_pending = True loop.run() self.assertEqual(loop.enqueued, [(0.0, service.shared.SYNC_JOB, 1)]) def test_nothing_pending_queues_nothing(self): loop = Loop(steps=6) loop.run() self.assertEqual(loop.enqueued, []) def test_a_settings_change_checks_the_switch_again(self): loop = Loop(steps=8) def flip(current): current.shared_pending = True current.monitor.onSettingsChanged() loop.at[3] = flip loop.run() self.assertEqual([name for _when, name, _generation in loop.enqueued], [service.shared.SYNC_JOB]) self.assertGreater(loop.enqueued[0][0], 0.0) def test_the_job_runs_the_sync(self): with patch.object(service.shared, 'sync', return_value=True) as sync: self.assertTrue(service._build(service.shared.SYNC_JOB)) sync.assert_called_once_with() with patch.object(service.shared, 'sync', return_value=False): self.assertFalse(service._build(service.shared.SYNC_JOB)) # not complete: the scheduler retries if __name__ == '__main__': unittest.main()