#!/usr/bin/env python3 """Targeted offline persistence/config hooks. No Kodi, providers or real profile. Run: python3 scripts/skin_harness/test_refresh_hooks.py """ import copy import importlib import importlib.util import json from pathlib import Path import re import sys import tempfile import types import unittest from unittest.mock import Mock, patch REPO = Path(__file__).resolve().parents[2] TV = {'widget_mytv_channels', 'widget_mytv_next'} CINEMA = {'widget_mycinema_channels', 'widget_mycinema_recommended'} class HookTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.profile = Path(temporary.name) self.values, self.properties, self.commands = {}, {}, [] xbmc = types.ModuleType('xbmc') xbmc.LOGINFO, xbmc.LOGWARNING, xbmc.LOGERROR = 1, 2, 3 xbmc.log = Mock() xbmc.getInfoLabel = lambda label: self.values.get(label[12:-1], '') if label.startswith('Skin.String(') else '' xbmc.executebuiltin = self.execute xbmc.getSkinDir = lambda: 'skin.milionar' xbmc.getCondVisibility = lambda condition: False addon = Mock() addon.getAddonInfo.return_value = str(self.profile) addon.getLocalizedString.side_effect = lambda number: str(number) xbmcaddon = types.ModuleType('xbmcaddon') xbmcaddon.Addon = Mock(return_value=addon) vfs = types.ModuleType('xbmcvfs') vfs.translatePath = lambda path: str(self.profile) if path.startswith('special://') else path vfs.mkdirs = lambda path: Path(path).mkdir(parents=True, exist_ok=True) home = Mock() home.setProperty.side_effect = lambda key, value: self.properties.__setitem__(key, value) home.getProperty.side_effect = lambda key: self.properties.get(key, '') home.clearProperty.side_effect = lambda key: self.properties.pop(key, None) gui = types.ModuleType('xbmcgui') gui.Window = Mock(return_value=home) refresh = types.ModuleType('wcs.skin.refresh') refresh.changed, refresh.request = Mock(), Mock() user_data = types.ModuleType('wcs.user_data') self.persisted = {'tv': [], 'cinema': []} for kind, prefix in (('tv', 'mytv'), ('cinema', 'mycinema')): setattr(user_data, 'load_' + prefix + '_channels', Mock(side_effect=lambda addon, k=kind: copy.deepcopy(self.persisted[k]))) setattr(user_data, 'store_' + prefix + '_channels', Mock(side_effect=lambda addon, values, k=kind: self.persisted.__setitem__(k, copy.deepcopy(values)))) library = types.ModuleType('wcs.library') library.LibraryManager = Mock() dialogs = types.ModuleType('wcs.skin.context_dialogs') dialogs.edit = Mock() modules = {'xbmc': xbmc, 'xbmcgui': gui, 'xbmcaddon': xbmcaddon, 'xbmcvfs': vfs, 'wcs.skin.refresh': refresh, 'wcs.user_data': user_data, 'wcs.library': library, 'wcs.skin.context_dialogs': dialogs} for name in ('wcs', 'wcs.ai', 'wcs.skin', 'wcs.library2'): package = types.ModuleType(name) package.__path__ = [str(REPO.joinpath(*name.split('.')))] modules[name] = package context = patch.dict(sys.modules, modules) context.start() self.addCleanup(context.stop) self.refresh, self.user_data, self.dialogs = refresh, user_data, dialogs self.manager = importlib.import_module('wcs.ai.MediaChannelManager') self.history = importlib.import_module('wcs.ai.ChannelHistory') self.recommendations = importlib.import_module('wcs.ai.RecommendationHistory') self.slots = importlib.import_module('wcs.skin.slots') self.context = importlib.import_module('wcs.skin.context') self.addon = addon self.snapshot = self.profile / 'widget_mycinema_recommended.json' self.snapshot.write_text('{"items":[{"title":"Previous"}]}') self.addCleanup(lambda: self.assertEqual(self.snapshot.read_bytes(), b'{"items":[{"title":"Previous"}]}')) def execute(self, command, *args): self.commands.append(command) match = re.fullmatch(r'Skin.SetString\((.*?),"(.*)"\)', command) if match: self.values[match[1]] = match[2] match = re.fullmatch(r'Skin.Reset\((.*?)\)', command) if match: self.values.pop(match[1], None) def channel_manager(self, kind='tv'): return self.manager.MediaChannelManager(self.addon, kind) def reset_events(self): self.refresh.changed.reset_mock() self.refresh.request.reset_mock() self.commands.clear() self.properties.clear() def draft(self, sec='my_movies'): self.slots.apply_defaults(sec) widgets = [self.slots.current(sec, n) for n in range(1, len(self.slots.SECTIONS[sec][2]) + 1)] self.reset_events() return {'sec': sec, 'widgets': widgets, 'hw': False, 'up': sec in self.slots.UPCOMING} def test_mytv_edits_publish_only_tv_and_identical_save_is_noop(self): manager = self.channel_manager() channel = manager.create_channel('TV') self.refresh.changed.assert_called_once_with(TV) self.refresh.request.assert_not_called() self.reset_events() manager.rename_channel(channel.id, 'TV') self.refresh.changed.assert_not_called() self.assertEqual(self.user_data.store_mytv_channels.call_count, 1) manager.rename_channel(channel.id, 'New TV') self.refresh.changed.assert_called_once_with(TV) def test_cinema_rebuild_requested_only_when_movie_sources_change(self): manager = self.channel_manager('cinema') channel = manager.create_channel('Cinema') self.refresh.changed.assert_called_once_with({'widget_mycinema_channels'}) self.refresh.request.assert_not_called() self.reset_events() channel.media_list = [{'id': 42}] manager._save_channels() self.refresh.changed.assert_called_once_with(CINEMA) self.refresh.request.assert_called_once_with({'widget_mycinema_recommended'}, force=True) self.reset_events() manager.rename_channel(channel.id, 'New cinema') self.refresh.changed.assert_called_once_with({'widget_mycinema_channels'}) self.refresh.request.assert_not_called() self.reset_events() channel.media_list = [] manager._save_channels() self.refresh.request.assert_called_once_with({'widget_mycinema_recommended'}, force=True) def test_shared_writer_failure_does_not_publish_and_legacy_noop_is_preserved(self): manager = self.channel_manager() self.user_data.store_mytv_channels.side_effect = lambda *args: None manager.create_channel('Not saved') self.refresh.changed.assert_not_called() self.persisted['tv'] = [{'id': 'legacy', 'name': 'TV', 'series_list': [{'id': 1}]}] manager._load_channels() manager._save_channels() self.refresh.changed.assert_not_called() self.assertEqual(self.user_data.store_mytv_channels.call_count, 1) def test_writer_false_suppresses_hook_but_none_and_true_are_accepted(self): for status in (False, None, True): with self.subTest(status=status): self.persisted['tv'] = [] self.reset_events() def store(addon, values): self.persisted['tv'] = copy.deepcopy(values) return status self.user_data.store_mytv_channels.side_effect = store self.channel_manager().create_channel('TV') if status is False: self.refresh.changed.assert_not_called() else: self.refresh.changed.assert_called_once_with(TV) def test_history_progress_resume_clear_and_delete_publish_at_persistence(self): self.history.update_series_progress('tv', 1, 1, 3) self.refresh.changed.assert_called_once_with(TV) path = Path(self.history._get_history_file('tv')) initial = path.read_bytes() self.reset_events() self.history.update_series_progress('tv', 1, 1, 3) self.refresh.changed.assert_not_called() self.assertEqual(initial, path.read_bytes()) self.history.set_resume_point('tv', 1, 1, 4, 10, 100) self.refresh.changed.assert_called_once_with(TV) self.reset_events() self.history.set_resume_point('tv', 1, 1, 4, 10, 100) self.refresh.changed.assert_not_called() self.history.clear_resume_point('tv', 1) self.refresh.changed.assert_called_once_with(TV) self.reset_events() self.history.clear_resume_point('tv', 1) self.refresh.changed.assert_not_called() self.assertTrue(self.history.delete_channel_history('tv')) self.refresh.changed.assert_called_once_with(TV) self.reset_events() self.assertFalse(self.history.delete_channel_history('tv')) self.refresh.changed.assert_not_called() def test_cinema_history_has_no_network_rebuild_and_noop_has_no_write(self): self.history.mark_movie_watched('cinema', 42) self.refresh.changed.assert_called_once_with({'widget_mycinema_channels'}) self.refresh.request.assert_not_called() self.reset_events() with patch.object(self.history.os, 'replace', side_effect=AssertionError('redundant write')): self.history.mark_movie_watched('cinema', 42) self.refresh.changed.assert_not_called() def test_atomic_history_failure_retains_previous_file_and_has_no_event(self): self.history.update_series_progress('tv', 1, 1, 3) path = Path(self.history._get_history_file('tv')) before = path.read_bytes() self.reset_events() with patch.object(self.history.os, 'replace', side_effect=OSError('disk error')): self.history.update_series_progress('tv', 1, 1, 4) self.assertEqual(path.read_bytes(), before) self.assertEqual(list(path.parent.glob('*.tmp')), []) self.refresh.changed.assert_not_called() def test_recommendations_changed_once_same_content_later_is_noop(self): entry = {'tmdb_id': 42, 'media_type': 'movie', 'title': 'Film'} with patch.object(self.recommendations.time, 'time', return_value=100): self.recommendations.add([entry]) self.refresh.changed.assert_called_once_with({'widget_ai_recent'}) previous = Path(self.recommendations._FILE).read_bytes() self.reset_events() with patch.object(self.recommendations.time, 'time', return_value=200): self.recommendations.add([entry]) self.assertEqual(Path(self.recommendations._FILE).read_bytes(), previous) self.refresh.changed.assert_not_called() self.recommendations.add([dict(entry, title='Changed')]) self.refresh.changed.assert_called_once_with({'widget_ai_recent'}) self.refresh.request.assert_not_called() def test_recommendation_reorder_is_change_and_failed_replace_retains_file(self): entries = [{'tmdb_id': number, 'media_type': 'movie', 'title': str(number)} for number in (1, 2)] self.recommendations.add(entries) self.reset_events() self.recommendations.add(entries[::-1]) self.refresh.changed.assert_called_once_with({'widget_ai_recent'}) previous = Path(self.recommendations._FILE).read_bytes() self.reset_events() with patch.object(self.recommendations.os, 'replace', side_effect=OSError('disk error')): with self.assertRaises(OSError): self.recommendations.add(entries) self.assertEqual(Path(self.recommendations._FILE).read_bytes(), previous) self.assertEqual(list(self.profile.glob('*.tmp')), []) self.refresh.changed.assert_not_called() def test_slot_nonlayout_change_writes_only_changed_values_and_placement(self): draft = self.draft() self.properties['milionar.widgets.cont.1'] = 'untouched duplicate' draft['widgets'][0]['limit'] = '10' self.slots.commit(draft) self.assertEqual([c for c in self.commands if c.startswith('Skin.')], ['Skin.SetString(mil.my_movies.1.limit,"10")']) self.assertEqual(set(self.properties), {'milionar.widgets.my_movies.1', 'milionar.widgets.cont.1'}) self.assertEqual(self.properties['milionar.widgets.cont.1'], 'untouched duplicate') self.assertNotIn('ReloadSkin()', self.commands) self.refresh.changed.assert_not_called() def test_sort_click_context_only_changes_never_reload_other_slots(self): for key, value in (('sort', 'title'), ('click', 'detail'), ('menu', {'main': ['detail'], 'more': [], 'hidden': ['play']})): with self.subTest(key=key): draft = self.draft() draft['widgets'][1][key] = value self.slots.commit(draft) self.assertEqual(set(self.properties), {'milionar.widgets.my_movies.2'}) self.assertNotIn('ReloadSkin()', self.commands) def test_unchanged_commit_and_defaults_make_no_refresh_or_reload(self): draft = self.draft() self.slots.commit(draft) self.assertEqual(self.commands, ['Dialog.Close(1123)']) self.assertEqual(self.properties, {}) self.commands.clear() self.slots.run({'action': 'skin_slots_defaults', 'sec': 'my_movies'}) self.assertEqual(self.commands, []) self.assertEqual(self.properties, {}) def test_style_and_removal_reload_once_but_flags_only_refresh_section_slots(self): draft = self.draft() draft['widgets'][1]['style'] = 'poster' self.slots.commit(draft) self.assertEqual(self.commands.count('ReloadSkin()'), 1) self.assertEqual(set(self.properties), {'milionar.widgets.my_movies.2'}) draft = self.draft() draft['widgets'].pop() self.slots.commit(draft) self.assertEqual(self.commands.count('ReloadSkin()'), 1) draft = self.draft() draft['hw'] = True self.slots.commit(draft) self.assertNotIn('ReloadSkin()', self.commands) self.assertEqual(set(self.properties), {'milionar.widgets.my_movies.1', 'milionar.widgets.my_movies.2', 'milionar.widgets.my_movies.3'}) def test_saved_context_only_touches_its_slot_and_identical_save_is_noop(self): draft = self.draft('mytv') self.dialogs.edit.return_value = {'main': ['open'], 'more': ['live'], 'hidden': []} widget = draft['widgets'][0] with patch.object(self.context, 'find_widget', return_value=('mytv', 1, widget)): self.slots.edit_saved_context(widget['uid']) self.assertEqual(set(self.properties), {'milionar.widgets.mytv.1'}) self.assertNotIn('Container.Refresh', self.commands) self.assertNotIn('ReloadSkin()', self.commands) self.reset_events() widget = self.slots.current('mytv', 1) with patch.object(self.context, 'find_widget', return_value=('mytv', 1, widget)): self.slots.edit_saved_context(widget['uid']) self.assertEqual(self.commands, []) self.assertEqual(self.properties, {}) def test_row_publication_reaches_mytv_duplicates_across_sections_only(self): self.values.update({'mil.mytv.1.type': 'widget_mytv_channels', 'mil.cont.1.type': 'widget_mytv_channels', 'mil.mytv.2.type': 'widget_mytv_next', 'mil.cont.2.type': 'cont_series'}) spec = importlib.util.spec_from_file_location('refresh_publication_under_test', REPO / 'wcs/skin/refresh.py') refresh = importlib.util.module_from_spec(spec) spec.loader.exec_module(refresh) with patch.object(refresh, 'revisions', return_value={row: 1 for row in TV}): refresh.publish(TV) self.assertEqual(set(self.properties), {'milionar.rows.widget_mytv_channels', 'milionar.rows.widget_mytv_next', 'milionar.widgets.mytv.1', 'milionar.widgets.mytv.2', 'milionar.widgets.cont.1', 'milionar.warm.queue'}) if __name__ == '__main__': unittest.main(verbosity=2)