# -*- coding: utf-8 -*- """Focused offline migration audit; mocked dialogs and providers, no real GUI. Run: python3 -m unittest discover -s scripts/library_harness -p test_migration.py -v """ import json import os import sys import tempfile import threading import unittest from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import stubs stubs.install() import xbmc import xbmcaddon import xbmcgui from wcs.library2 import meta, migration, store from wcs.skin import refresh class MigrationTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='migration-test-') self.addCleanup(self.temp.cleanup) self.profile = self.temp.name old_path = store._override_path store.set_path(os.path.join(self.profile, 'library.db')) self.addCleanup(store.set_path, old_path) self.write('movie', [{'id': 1, 'title': 'One', 'genres': []}]) self.write('tv', []) self.dialogs, self.progresses, self.events = [], [], [] owner = self class Dialog: answer = True def __init__(self): owner.dialogs.append(self) self.prompts, self.messages = [], [] def yesno(self, heading, message, **kwargs): self.prompts.append((heading, message, kwargs)) return self.answer def ok(self, heading, message): self.messages.append(message) owner.events.append('message') class Progress: cancelled = False def __init__(self): self.closed, self.updates = False, [] owner.progresses.append(self) def create(self, *args): owner.events.append('open') def update(self, percent, message): self.updates.append((percent, message)) def iscanceled(self): return self.cancelled def close(self): self.closed = True owner.events.append('close') self.Dialog, self.Progress = Dialog, Progress self.addCleanup(patch.stopall) patch.object(migration, '_profile', return_value=self.profile).start() patch.object(xbmcgui, 'Dialog', Dialog).start() patch.object(xbmcgui, 'DialogProgress', Progress, create=True).start() # A provider call omitted from a test fails instead of reaching the network. self.provider = patch.object(migration.tmdb, '_fetch', side_effect=AssertionError('Live provider forbidden')).start() patch.dict(xbmcaddon.Addon.strings, {32660: 'Import', 32661: '{movie}/{tv}; existing={existing}; duplicates={duplicates}; invalid={invalid}; files={files}', 32662: 'Importing', 32663: 'Metadata {done}/{total}', 32664: 'Home', 32665: 'Complete', 32666: 'Pending metadata {count}', 32667: 'Failed; retry', 32668: 'Home timeout; retry', 32669: 'No sources'}).start() xbmcgui.Window.props.clear() def path(self, kind): return os.path.join(self.profile, dict(store.LEGACY_FILES)[kind]) def write(self, kind, data): with open(self.path(kind), 'w', encoding='utf-8') as handle: json.dump(data, handle) def test_summary_merge_order_dedup_and_existing_watch_state(self): store.replace_list('movie', [{'id': 7, 'title': 'New'}, {'id': 1, 'title': 'Edited'}]) store.save_resume('movie', 1, 240, 1800) store.mark_watched('tv', 20, 1, 1) before = store.get_state('movie', 1), store.get_state('tv', 20, 1, 1) self.write('movie', [{'id': '1', 'title': 'Old'}, {'id': 2, 'title': 'Two'}, {'id': '2'}, {}, False, {'id': True}, {'id': 1.5}, {'id': -1}, {'id': 2**63}]) self.write('tv', [{'id': 20, 'name': 'Show'}]) snapshot = migration.offer(self.profile, manual=True) self.assertEqual(snapshot['counts'], {'movie': 9, 'tv': 1}) self.assertEqual((snapshot['existing'], snapshot['duplicates'], snapshot['invalid']), (1, 1, 6)) self.assertEqual(store.import_watchlists(self.profile, snapshot), {'movie': 2, 'tv': 1}) self.assertEqual([e['id'] for e in store.get_list('movie')], [7, 1, 2]) self.assertEqual(store.get_list('movie')[1]['title'], 'Edited') self.assertEqual(before, (store.get_state('movie', 1), store.get_state('tv', 20, 1, 1))) self.assertEqual(store.import_watchlists(self.profile), {}) def test_originals_backups_and_changed_source_are_retained(self): with open(self.path('movie'), 'rb') as handle: original = handle.read() store.import_watchlists(self.profile) with open(self.path('movie') + '.migrated', 'rb') as handle: self.assertEqual(handle.read(), original) self.assertTrue(os.path.exists(self.path('movie'))) self.write('movie', [{'id': 2}]) self.assertEqual(store.import_watchlists(self.profile), {'movie': 1, 'tv': 0}) self.assertEqual([e['id'] for e in store.get_list('movie')], [1, 2]) with open(self.path('movie') + '.migrated', 'rb') as handle: self.assertEqual(handle.read(), original) self.assertEqual(len([n for n in os.listdir(self.profile) if n.endswith('.bak')]), 1) def test_malformed_second_source_never_commits_first_or_marks_imported(self): for raw in ('{', '{}', 'null', '[NaN]', '\udfff'): with open(self.path('tv'), 'wb') as handle: handle.write(raw.encode('utf-8', errors='surrogatepass')) with self.assertRaises((ValueError, UnicodeError)): store.import_watchlists(self.profile) self.assertEqual(store.get_list('movie'), []) self.assertIsNone(store.get_flag('watchlists_imported')) self.assertTrue(migration.pending(self.profile)) self.write('tv', []) self.assertEqual(store.import_watchlists(self.profile)['movie'], 1) def test_failed_read_and_backup_do_not_commit(self): from builtins import open as real_open def unreadable(path, *args, **kwargs): if path == self.path('tv'): raise PermissionError('Denied') return real_open(path, *args, **kwargs) with patch('builtins.open', side_effect=unreadable), self.assertRaises(PermissionError): store.import_watchlists(self.profile) with patch.object(store, '_import_backup', side_effect=OSError('Full disk')), self.assertRaises(OSError): store.import_watchlists(self.profile) self.assertEqual(store.get_list('movie'), []) self.assertIsNone(store.get_flag('watchlists.import.v2')) def test_transaction_failure_rolls_back_both_lists_and_flags_then_retry(self): self.write('tv', [{'id': 20}]) with store.tx() as conn: conn.execute("CREATE TRIGGER fail_tv BEFORE UPDATE ON titles WHEN NEW.media_type='tv' " "BEGIN SELECT RAISE(ABORT,'simulated crash'); END") with self.assertRaises(Exception): store.import_watchlists(self.profile) self.assertEqual(store.get_list('movie'), []) self.assertIsNone(store.get_flag('watchlists_imported')) with store.tx() as conn: conn.execute('DROP TRIGGER fail_tv') self.assertEqual(store.import_watchlists(self.profile), {'movie': 1, 'tv': 1}) self.assertEqual(store.import_watchlists(self.profile), {}) def test_changed_reviewed_snapshot_is_rejected(self): snapshot = migration.offer(self.profile) self.write('movie', [{'id': 9}]) with self.assertRaises(ValueError): store.import_watchlists(self.profile, snapshot) self.assertEqual(store.get_list('movie'), []) def test_legacy_renamed_files_are_discovered(self): os.rename(self.path('movie'), self.path('movie') + '.migrated') self.assertEqual(migration.offer(self.profile)['counts']['movie'], 1) self.assertEqual(store.import_watchlists(self.profile)['movie'], 1) self.assertTrue(os.path.exists(self.path('movie') + '.migrated')) def test_decline_preserves_bytes_and_automatic_suppression_manual_reopens(self): self.Dialog.answer = False snapshot = migration.offer(self.profile) self.assertEqual(migration.run({'auto': '1'})['status'], 'declined') self.assertEqual(store.get_list('movie'), []) self.assertFalse(migration.pending(self.profile)) self.assertIsNotNone(migration.offer(self.profile, manual=True)) self.assertEqual(migration.run({'auto': '1'})['status'], 'skipped') self.assertEqual(migration.run()['status'], 'declined') with open(self.path('movie'), 'rb') as handle: self.assertEqual(handle.read(), snapshot['files'][0]['raw']) self.assertFalse(self.progresses) self.write('movie', [{'id': 2}]) self.assertTrue(migration.pending(self.profile)) def test_import_retry_waits_before_completed_and_closes_afterwards(self): def wait(targets, slots, progress): self.events.append('wait') self.assertFalse(progress.closed) self.assertFalse(store.get_flag(migration.COMPLETED)) self.assertEqual(refresh.revisions(targets), targets) return True with patch.object(migration, '_wait_home', side_effect=wait): result = migration.run() self.assertEqual(result['status'], 'complete') self.assertEqual(self.events, ['open', 'wait', 'close', 'message']) self.assertFalse(migration.pending(self.profile)) self.provider.assert_not_called() with patch.object(migration, '_wait_home', return_value=True): self.assertEqual(migration.run()['status'], 'complete') self.assertEqual(len(store.get_list('movie')), 1) def test_provider_failure_is_pending_and_retry_uses_cached_episode_coverage(self): self.write('tv', [{'id': 20, 'name': 'Show', 'genres': [], 'seasons': [{'season_number': 1, 'episode_count': 2}]}]) with patch.object(migration, '_wait_home', return_value=True): result = migration.run() self.assertEqual(result['status'], 'pending') self.assertEqual(result['pending'], ['season:20:1']) self.assertFalse(store.get_flag(migration.COMPLETED)) self.assertEqual(self.dialogs[-1].messages, ['Pending metadata 1']) meta.put('season:20:1', {'episodes': [{'episode_number': 1}, {'episode_number': 2}]}, 60) with patch.object(migration, '_wait_home', return_value=True): self.assertEqual(migration.run()['status'], 'complete') self.assertEqual(len(store.get_list('tv')), 1) def test_home_timeout_preserves_import_and_retries_without_duplicates(self): with patch.object(migration, '_wait_home', return_value=False): result = migration.run() self.assertEqual(result['status'], 'pending') self.assertFalse(store.get_flag(migration.COMPLETED)) self.assertEqual(self.dialogs[-1].messages, ['Home timeout; retry']) self.assertTrue(self.progresses[-1].closed) with patch.object(migration, '_wait_home', return_value=True): self.assertEqual(migration.run()['status'], 'complete') self.assertEqual(len(store.get_list('movie')), 1) def test_failure_and_cancel_close_modal_without_success(self): with patch.object(store, 'import_watchlists', side_effect=OSError('disk')): self.assertEqual(migration.run()['status'], 'failed') self.assertTrue(self.progresses[-1].closed) self.assertEqual(self.dialogs[-1].messages, ['Failed; retry']) self.Progress.cancelled = True self.assertEqual(migration.run()['status'], 'cancelled') self.assertTrue(self.progresses[-1].closed) self.assertFalse(store.get_flag(migration.COMPLETED)) def test_no_modal_on_background_thread(self): failures = [] def run(): try: migration.run() except RuntimeError as exc: failures.append(exc) worker = threading.Thread(target=run) worker.start() worker.join() self.assertEqual(len(failures), 1) self.assertFalse(self.dialogs) def test_wrappers_never_implicitly_import(self): from wcs import user_data with patch.object(store, 'import_watchlists', side_effect=AssertionError('Implicit import')): self.assertEqual(user_data._library_list('movie'), []) self.assertTrue(user_data._library_store('movie', [{'id': 4}])) self.assertEqual([e['id'] for e in store.get_list('movie')], [4]) self.assertTrue(os.path.exists(self.path('movie'))) def test_visible_widgets_wait_for_publication_and_update_cycle(self): row = 'my_movies_all' refresh.changed({row}) targets = refresh.revisions({row}) slot = {'section': 'my_movies', 'slot': 1, 'row': row, 'container': 29100} home = xbmcgui.Window(10000) ticks = [0] def tick(ms): ticks[0] += 1 home.setProperty('milionar.rows.' + row, str(targets[row])) home.setProperty('milionar.widgets.my_movies.1', str(targets[row])) def visible(cond): return ticks[0] <= 2 if 'IsUpdating' in cond else True with patch.object(xbmc, 'sleep', side_effect=tick), patch.object(xbmc, 'getCondVisibility', side_effect=visible), \ patch.object(migration.time, 'monotonic', side_effect=lambda: ticks[0] * 0.1): self.assertTrue(migration._wait_home(targets, [slot], self.Progress(), timeout=2)) self.assertGreaterEqual(ticks[0], 8) def test_idle_visible_container_without_item_evidence_times_out(self): row = 'my_movies_all' targets = refresh.revisions({row}) refresh.publish({row}) home = xbmcgui.Window(10000) home.setProperty('milionar.widgets.my_movies.1', str(targets[row])) slot = {'section': 'my_movies', 'slot': 1, 'row': row, 'container': 29100} ticks = [0] with patch.object(xbmc, 'sleep', side_effect=lambda ms: ticks.__setitem__(0, ticks[0] + 1)), \ patch.object(xbmc, 'getCondVisibility', side_effect=lambda c: 'IsUpdating' not in c), \ patch.object(migration.time, 'monotonic', side_effect=lambda: ticks[0] * 0.1): self.assertFalse(migration._wait_home(targets, [slot], self.Progress(), timeout=1)) def test_fast_update_uses_matching_items_and_fences_newer_revisions(self): row = 'my_movies_all' targets = refresh.revisions({row}) refresh.publish({row}) xbmcgui.Window(10000).setProperty('milionar.widgets.my_movies.1', str(targets[row])) slot = {'section': 'my_movies', 'slot': 1, 'row': row, 'container': 29100, 'expected': [('One', 'plugin://one')]} def label(query): return '1' if query.endswith('.NumItems') else 'One' if query.endswith('.Label') else 'plugin://one' ticks = [0] with patch.object(xbmc, 'getInfoLabel', side_effect=label), \ patch.object(xbmc, 'getCondVisibility', side_effect=lambda c: 'IsUpdating' not in c), \ patch.object(xbmc, 'sleep', side_effect=lambda ms: ticks.__setitem__(0, ticks[0] + 1)), \ patch.object(migration.time, 'monotonic', side_effect=lambda: ticks[0] * 0.1): self.assertTrue(migration._wait_home(targets, [slot], self.Progress(), timeout=1)) refresh.changed({row}) ticks[0] = 0 new_slot = dict(slot, expected=[('Changed', 'plugin://changed')]) with patch.object(migration, '_prepare_widgets', return_value=[new_slot]) as prepare: self.assertFalse(migration._wait_home(targets, [slot], self.Progress(), timeout=1)) prepare.assert_called_once() self.assertEqual(targets, refresh.revisions({row})) def test_truncated_backup_crash_is_retryable_without_overwrite(self): import hashlib with open(self.path('movie'), 'rb') as handle: raw = handle.read() digest_backup = self.path('movie') + '.migration-' + hashlib.sha256(raw).hexdigest() + '.bak' for path in (self.path('movie') + '.migrated', digest_backup): with open(path, 'wb') as handle: handle.write(b'truncated') with patch.object(os, 'link', side_effect=OSError('Unsupported')): self.assertEqual(store.import_watchlists(self.profile)['movie'], 1) with open(digest_backup + '.1', 'rb') as handle: self.assertEqual(handle.read(), raw) for path in (self.path('movie') + '.migrated', digest_backup): with open(path, 'rb') as handle: self.assertEqual(handle.read(), b'truncated') def test_accepted_manual_retry_clears_decline_even_when_partial(self): snapshot = migration.offer(self.profile) store.set_flag(migration.DECLINED, snapshot['fingerprint']) with patch.object(migration, '_wait_home', return_value=False): self.assertEqual(migration.run()['status'], 'pending') self.assertFalse(migration.pending(self.profile)) self.assertIsNotNone(migration.offer(self.profile, manual=True)) self.assertEqual(json.loads(store.get_flag(migration.STATE))['phase'], 'pending_home') def test_automatic_offer_requires_both_new_lists_empty(self): self.assertTrue(migration.pending(self.profile)) self.assertIsNotNone(migration.offer(self.profile)) for kind in ('movie', 'tv'): store.set_in_list(kind, 900, True, {'id': 900, 'title': 'New'}) with patch.object(store, 'inspect_watchlist_import', side_effect=AssertionError('No legacy reads')): self.assertFalse(migration.pending(self.profile)) self.assertIsNone(migration.offer(self.profile)) self.assertEqual(migration.run({'auto': '1'})['status'], 'skipped') self.assertFalse(self.dialogs[-1].prompts) self.assertFalse(self.progresses) self.assertIsNotNone(migration.offer(self.profile, manual=True)) store.set_in_list(kind, 900, False) def test_valid_empty_sources_skip_auto_but_malformed_sources_report_failure(self): self.write('movie', []) self.write('tv', []) self.assertFalse(migration.pending(self.profile)) self.assertIsNone(migration.offer(self.profile)) self.assertEqual(migration.run({'auto': '1'})['status'], 'skipped') self.assertFalse(self.dialogs[-1].prompts) self.assertIsNotNone(migration.offer(self.profile, manual=True)) with patch.object(migration, '_wait_home', return_value=False): self.assertEqual(migration.run()['status'], 'pending') self.assertIsNotNone(migration.offer(self.profile, manual=True)) self.assertEqual(json.loads(store.get_flag(migration.STATE))['phase'], 'pending_home') with open(self.path('tv'), 'w', encoding='utf-8') as handle: handle.write('{') self.assertTrue(migration.pending(self.profile)) self.assertEqual(migration.run({'auto': '1'})['status'], 'failed') self.assertEqual(self.dialogs[-1].messages, ['Failed; retry']) def test_new_version_updates_of_populated_lists_never_offer_again(self): store.set_in_list('movie', 900, True, {'id': 900, 'title': 'New', 'genres': []}) for version in ('2.0', '2.1', '3.0'): self.write('movie', [{'id': 1, 'title': 'Legacy ' + version, 'genres': []}]) self.assertFalse(migration.pending(self.profile)) self.assertEqual(migration.run({'auto': '1'})['status'], 'skipped') self.assertFalse(any(d.prompts for d in self.dialogs)) self.assertEqual([e['id'] for e in store.get_list('movie')], [900]) with patch.object(migration, '_wait_home', return_value=True): self.assertEqual(migration.run()['status'], 'complete') self.assertEqual([e['id'] for e in store.get_list('movie')], [900, 1]) def test_interrupted_import_with_pending_metadata_only_retries_manually(self): self.write('tv', [{'id': 20, 'name': 'Show', 'genres': [], 'seasons': [{'season_number': 1, 'episode_count': 1}]}]) with patch.object(migration, '_wait_home', return_value=True): self.assertEqual(migration.run({'auto': '1'})['status'], 'pending') self.assertFalse(migration.pending(self.profile)) self.assertEqual(migration.run({'auto': '1'})['status'], 'skipped') self.assertEqual(json.loads(store.get_flag(migration.STATE))['phase'], 'pending_metadata') self.assertIsNotNone(migration.offer(self.profile, manual=True)) meta.put('season:20:1', {'episodes': [{'episode_number': 1}]}, 60) with patch.object(migration, '_wait_home', return_value=True): self.assertEqual(migration.run()['status'], 'complete') self.assertEqual(len(store.get_list('tv')), 1) def test_build_warms_catalogue_and_legacy_with_slot_context(self): item = xbmcgui.ListItem(label='Prepared', path='plugin://prepared') slots = [{'row': 'my_movies_all', 'params': {'sec': 'my_movies', 'slot': '1'}}, {'row': 'widget_game_topics', 'params': {'sec': 'game', 'slot': '2'}}] from wcs.library2 import rows from wcs.skin import widgets with patch.object(migration, '_slots', return_value=slots), \ patch.object(rows, 'build', return_value=[('plugin://prepared', item, False)]) as library_build, \ patch.object(widgets, 'build', return_value=[item], create=True) as legacy_build: result = migration._prepare_widgets({'my_movies_all', 'widget_game_topics'}, self.Progress()) library_build.assert_called_once_with('my_movies_all', {'sec': 'my_movies', 'slot': '1'}) legacy_build.assert_called_once_with('widget_game_topics', {'sec': 'game', 'slot': '2'}) self.assertEqual([s['expected'] for s in result], [[('Prepared', 'plugin://prepared')]] * 2) def test_real_cached_builders_align_movie_series_game_and_channel_slots(self): from wcs.library2 import rows from wcs.skin import widgets self.write('tv', [{'id': 20, 'name': 'Show', 'genres': [], 'seasons': [ {'season_number': 1, 'episode_count': 1, 'episodes': [ {'episode_number': 1, 'name': 'Pilot', 'air_date': '2020-01-01'}]}]}]) configured = {'my_movies': 'my_movies_all', 'my_series': 'my_series_all', 'game': 'widget_game_topics', 'cont': 'widget_mytv_next'} labels = {f'Skin.String(mil.{sec}.1.type)': row for sec, row in configured.items()} labels.update({f'Skin.String(mil.{sec}.1.limit)': '10' for sec in configured}) labels['System.Language'] = 'cs' def wait(targets, slots, progress): self.assertIn('widget_mytv_next', targets) self.assertEqual(len(slots), 4) for slot in slots: items = rows.build(slot['row'], slot['params']) if slot['row'] in rows.CATALOGUE \ else [(i.getPath(), i, False) for i in widgets.build(slot['row'], slot['params'])] self.assertEqual(slot['expected'], [(i.getLabel(), i.getPath()) for _p, i, _f in items]) self.assertTrue(os.path.isdir(os.path.join(self.profile, 'row_snapshots'))) return True with patch.object(widgets, 'PROFILE', self.profile), \ patch.object(xbmc, 'getInfoLabel', side_effect=lambda q: labels.get(q, '')), \ patch.object(migration, '_wait_home', side_effect=wait): result = migration.run() self.assertEqual(result['status'], 'complete') self.provider.assert_not_called() if __name__ == '__main__': unittest.main()