#!/usr/bin/env python3 """Offline checks of the prepared Home rows (wcs/skin/journal.py): what the service writes down is what the plugin call hands over. Run: python3 scripts/skin_harness/test_journal.py """ from pathlib import Path import sys import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / 'scripts/library_harness')) import stubs stubs.install() import xbmc import xbmcgui import xbmcplugin from wcs.library2 import rows from wcs.skin import context, journal PARAMS = {'action': 'widget_row', 'row': 'my_movies_added', 'sec': 'my_movies', 'slot': '2', 'limit': '50', 'sort': 'added', 'style': 'poster', 'click': '', 'hw': '', 'up': '', 'reload': '7', 'format': '2', 'profile': '', 'hb': 'ab12', 'mode': ''} def fill(item, number): item.setArt({'poster': 'p{}.jpg'.format(number)}) item.setProperty('milionar.hero.title', 'Title {}'.format(number)) item.addContextMenuItems([('Menu', 'RunPlugin(x)')], True) tag = item.getVideoInfoTag() tag.setTitle('Title {}'.format(number)) tag.setYear(2000 + number) return item def fake_run(count=3): """A row listing the way the real ones work: items made with xbmcgui.ListItem, sent with addDirectoryItems.""" def run(handle, params): items = [('plugin://x/?n={}'.format(n), fill(xbmcgui.ListItem('Label {}'.format(n), offscreen=True), n), n == 2) for n in range(count)] xbmcplugin.setContent(handle, 'movies') xbmcplugin.addDirectoryItems(handle, items, len(items)) xbmcplugin.endOfDirectory(handle, succeeded=True, cacheToDisc=False) return run def state(item): tag = item.getVideoInfoTag() return (item.getLabel(), item.getPath(), dict(item.art), dict(item.props), list(item.context), {key: value for key, value in vars(tag).items() if not key.startswith('_')}) class JournalTests(unittest.TestCase): def setUp(self): xbmcgui.Window.props.clear() xbmcplugin.ITEMS.clear() journal._stamp[:] = [1] for path in Path(journal._path('x')).parent.glob('*.json'): path.unlink() self.context = patch.object(context, 'install_listing_context', lambda params: None) self.context.start() self.addCleanup(self.context.stop) def serve(self, params): xbmcplugin.ITEMS.clear() served = journal.serve(5, journal.normalize(params)) return served, xbmcplugin.ITEMS.get(5, []) def test_the_plugin_call_hands_over_the_items_the_service_built(self): real_functions = (xbmcplugin.addDirectoryItems, xbmcplugin.endOfDirectory, xbmcplugin.setContent) with patch.object(rows, 'run', fake_run()): self.assertTrue(journal.prepare(PARAMS)) self.assertEqual((xbmcplugin.addDirectoryItems, xbmcplugin.endOfDirectory, xbmcplugin.setContent), real_functions) self.assertFalse(xbmcplugin.ITEMS) # nothing was sent while building served, items = self.serve(PARAMS) self.assertTrue(served) expected = [('plugin://x/?n={}'.format(n), state(fill(journal._real_item('Label {}'.format(n), offscreen=True), n)), n == 2) for n in range(3)] self.assertEqual([(path, state(item), folder) for path, item, folder in items], expected) self.assertIsInstance(items[0][1].context[0], tuple) # tuples survive the journal self.assertEqual(xbmcplugin.CONTENT.get(5), 'movies') def test_it_is_read_from_the_file_after_a_restart_and_only_for_the_same_request_and_code(self): with patch.object(rows, 'run', fake_run()): journal.prepare(PARAMS) xbmcgui.Window.props.clear() # Kodi restarted: no Home properties self.assertTrue(self.serve(PARAMS)[0]) self.assertFalse(self.serve(dict(PARAMS, reload='8'))[0]) # another revision of the row self.assertFalse(self.serve(dict(PARAMS, sort='title'))[0]) self.assertFalse(self.serve({'action': 'widget_row', 'row': 'my_movies_added'})[0]) # not a Home row journal._stamp[:] = [2] # the addon was updated self.assertFalse(self.serve(PARAMS)[0]) def test_a_row_is_never_older_than_its_data(self): from wcs.skin import refresh with patch.object(rows, 'run', fake_run()): journal.prepare(PARAMS) self.assertTrue(self.serve(PARAMS)[0]) refresh.changed(['my_movies_added']) # the library changed, nothing was published yet self.assertFalse(self.serve(PARAMS)[0]) # the plugin call builds the row itself with patch.object(rows, 'run', fake_run()): journal.prepare(PARAMS) self.assertTrue(self.serve(PARAMS)[0]) def test_a_row_has_at_most_twenty_cards_on_both_sides(self): self.assertEqual(journal.normalize(PARAMS)['limit'], '20') self.assertEqual(journal.normalize(dict(PARAMS, limit='10'))['limit'], '10') self.assertNotIn('click', journal.normalize(PARAMS)) # Kodi's URL parsing drops empty values with patch.object(rows, 'run', fake_run(30)): journal.prepare(PARAMS) self.assertEqual(len(self.serve(PARAMS)[1]), 20) def test_what_cannot_be_written_down_is_left_to_the_plugin_call(self): def foreign(handle, params): xbmcplugin.addDirectoryItems(handle, [('plugin://x', journal._real_item('made elsewhere'), False)], 1) def odd(handle, params): item = xbmcgui.ListItem('odd') item.setProperty('milionar.x', object()) xbmcplugin.addDirectoryItems(handle, [('plugin://x', item, False)], 1) for run in (foreign, odd): with patch.object(rows, 'run', run): self.assertIsNone(journal.prepare(PARAMS)) self.assertFalse(self.serve(PARAMS)[0]) def test_a_failed_build_keeps_the_journal_it_had(self): with patch.object(rows, 'run', fake_run()): journal.prepare(PARAMS) with patch.object(rows, 'run', side_effect=RuntimeError('broken row')): self.assertIsNone(journal.prepare(PARAMS)) self.assertEqual(len(self.serve(PARAMS)[1]), 3) def test_only_the_building_thread_gets_journal_items(self): journal.install() self.assertNotIsInstance(xbmcgui.ListItem('plain'), journal.Item) def test_old_journals_of_a_slot_are_dropped(self): with patch.object(rows, 'run', fake_run(1)): for revision in range(journal.KEEP + 3): journal.prepare(dict(PARAMS, reload=str(revision))) files = list(Path(journal._path('x')).parent.glob('my_movies.2.*.json')) self.assertEqual(len(files), journal.KEEP) self.assertEqual(len([key for key in xbmcgui.Window.props if key.startswith(journal.PROPERTY)]), journal.KEEP) def test_request_parameters_are_what_the_skin_puts_into_the_url(self): sys.path.insert(0, str(ROOT / 'scripts/skin_harness')) import gen_slots strings = {'mil.foryou.1.type': 'fy_continue', 'mil.foryou.1.limit': '30', 'mil.foryou.1.sort': 's', 'mil.foryou.1.style': 'episode', 'mil.foryou.1.click': 'play', 'mil.foryou.hidewatched': 'true', 'mil.foryou.upcoming': '', 'mil.foryou.1.rev': '12', 'mil.foryou.herorev': 'ff', 'mil.foryou.mode': 'tv'} with patch.object(xbmc, 'getInfoLabel', side_effect=lambda label: strings.get(label[len('Skin.String('):-1], '')): params = journal.request_params('foryou', 1) self.assertIsNone(journal.request_params('foryou', 2)) url = gen_slots.CONTENT.format(sec='foryou', n=1).replace('&', '&') from urllib.parse import parse_qsl import re filled = re.sub(r'\$INFO\[Skin\.String\(([^)]*)\)\]', lambda found: strings.get(found.group(1), ''), url) filled = re.sub(r'\$INFO\[Window\(Home\)\.Property\([^)]*\)\]', '', filled) self.assertEqual(params, journal.normalize(dict(parse_qsl(filled.split('?', 1)[1])))) if __name__ == '__main__': unittest.main()