# -*- coding: utf-8 -*- """Click on a MyTV / MyCinema channel and on the title widgets of the old sections (150): the paths of every option and the headless broadcast of a channel (the queue of the MyTV dialog without the dialog).""" import os import sys import types import unittest from urllib.parse import parse_qsl, urlsplit HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import stubs # noqa: E402 stubs.install() import xbmc # noqa: E402 import xbmcaddon # noqa: E402 import xbmcgui # noqa: E402 from wcs.ai import channel_play # noqa: E402 from wcs.library2 import clicks, meta, store # noqa: E402 from wcs.skin import widgets # noqa: E402 def action_of(path): params = dict(parse_qsl(urlsplit(path).query)) if params.get('action') == 'skin_run': params = dict(parse_qsl(params['run'])) return params.get('action'), params class FakeCache: def __init__(self): self.data, self.flushed = {}, 0 def get(self, key): return self.data.get(key) def set(self, key, value, ttl_seconds=0): self.data[key] = value def delete(self, key): self.data.pop(key, None) def flush(self): self.flushed += 1 class WidgetClickTests(unittest.TestCase): TV = [{'id': 'c1', 'name': 'Komedie', 'icon': 'i.png', 'media_list': [{'id': 2691, 'name': 'Dva a pul chlapa', 'poster_path': '/x.jpg', 'year': '2003'}]}] CACHED = {'widget_tmdb_popular_movies': [{'tmdb_id': 5, 'media_type': 'movie', 'title': 'Film', 'poster': 'p', 'plot': 'x', 'run': 'action=show_wcs_details&media_type=movie&tmdb_id=5'}], 'widget_tmdb_popular_series': [{'tmdb_id': 6, 'media_type': 'series', 'title': 'Serial', 'poster': 'p', 'plot': 'x'}], 'widget_mycinema_recommended': [{'tmdb_id': 7, 'media_type': 'movie', 'title': 'Kino', 'poster': 'p', 'plot': 'x'}]} def setUp(self): import tempfile store.set_path(os.path.join(tempfile.mkdtemp(prefix='milionar-db-'), 'library.db')) meta.put(meta.key_tv(2691), {'id': 2691, 'name': 'Dva a pul chlapa', 'status': 'Ended', 'first_air_date': '2003-01-01', 'seasons': [{'season_number': 1}]}, 10 ** 9) meta.put(meta.key_season(2691, 1), {'episodes': [{'episode_number': n, 'name': 'E{}'.format(n), 'air_date': '2003-01-0{}'.format(n)} for n in range(1, 6)]}, 10 ** 9) self._read, self._cached = widgets._read_json, widgets._cached files = {'mytv_channels.json': self.TV, 'mycinema_channels.json': [dict(self.TV[0], id='k1')], 'ai_recommendations.json': [{'tmdb_id': 8, 'media_type': 'tv', 'title': 'AI serial', 'poster': 'p'}, {'tmdb_id': 9, 'media_type': 'movie', 'title': 'AI film', 'poster': 'p'}], 'global_watch_history.json': [{'tmdb_id': 3, 'media_type': 'movie', 'title': 'Hist', 'poster': 'p', 'year': '1999'}]} widgets._read_json = lambda name, default: files.get(name, default) widgets._cached = lambda name: self.CACHED.get(name, []) from wcs.ai import ChannelHistory self._progress = ChannelHistory.load_channel_history class Progress(dict): def get(self, key, default=None): # every series is at season 1, episode 3 return {'season': 1, 'episode': 3} ChannelHistory.load_channel_history = lambda channel_id: Progress() def tearDown(self): widgets._read_json, widgets._cached = self._read, self._cached from wcs.ai import ChannelHistory ChannelHistory.load_channel_history = self._progress def build(self, name, click): return widgets.WIDGETS[name]({'click': click}) def test_a_channel_click_for_every_option(self): for option in clicks.OPTIONS['tv_channel']: path = self.build('widget_mytv_channels', option)[0].getPath() action, params = action_of(path) if option == 'open': self.assertEqual((action, params['channel_id']), ('my_tv_dialog', 'c1')) else: self.assertEqual((action, params['kind'], params['channel_id'], params['mode']), ('channel_play', 'tv', 'c1', option)) for option in clicks.OPTIONS['cinema_channel']: action, params = action_of(self.build('widget_mycinema_channels', option)[0].getPath()) self.assertEqual(action, 'my_cinema_dialog' if option == 'open' else 'channel_play') self.assertEqual(params['channel_id'], 'k1') self.assertEqual(action_of(self.build('widget_mytv_channels', '')[0].getPath())[1]['mode'], 'live') # unset: broadcast self.assertEqual(action_of(self.build('widget_mytv_channels', 'bogus')[0].getPath())[1]['mode'], 'live') def test_mytv_next_keeps_the_channel_progress_for_play(self): action, params = action_of(self.build('widget_mytv_next', 'play')[0].getPath()) self.assertEqual((action, params['mytv_channel_id'], params['season_number'], params['episode_number']), ('play_episode_from_addon', 'c1', '1', '4')) for option in clicks.OPTIONS['mytv_next']: item = self.build('widget_mytv_next', option)[0] action, params = action_of(item.getPath()) self.assertTrue(action, option) if option == 'channel_open': self.assertEqual((action, params['channel_id']), ('my_tv_dialog', 'c1')) if option == 'season': self.assertEqual((action, params['s']), ('mlist', '1')) # the season of the next episode if option in ('folder', 'season'): self.assertEqual(item.getProperty('IsPlayable'), '') else: self.assertEqual(item.getProperty('IsPlayable'), 'true') def test_every_old_widget_item_has_its_exec(self): for name in ('widget_mytv_channels', 'widget_mycinema_channels', 'widget_mytv_next', 'widget_tmdb_popular_movies', 'widget_tmdb_popular_series', 'widget_mycinema_recommended', 'widget_ai_recent', 'widget_game_topics'): for option in clicks.options(name): for item in self.build(name, option): command = item.getProperty('milionar.exec') self.assertTrue(command, (name, option)) folder = item.getProperty('milionar.folder') == '1' self.assertEqual(command.startswith('ActivateWindow(Videos,'), folder, (name, option)) live = self.build('widget_mytv_channels', 'live')[0].getProperty('milionar.exec') self.assertEqual(live, 'RunPlugin(plugin://plugin.video.milionar?action=channel_play&kind=tv&channel_id=c1&mode=live)') def test_title_widgets_follow_their_click(self): for name, kind in (('widget_tmdb_popular_movies', 'movie'), ('widget_tmdb_popular_series', 'series'), ('widget_mycinema_recommended', 'movie'), ('widget_ai_recent', 'mixed'), ('widget_game_topics', 'game')): for option in clicks.options(name): items = self.build(name, option) self.assertTrue(items, (name, option)) for item in items: if item.label == 'AI chat': self.assertEqual(action_of(item.getPath())[0], 'ai_chat_recommendation') # the button stays continue self.assertTrue(action_of(item.getPath())[0], (name, option)) folder = option in ('folder', 'season') self.assertEqual(item.getProperty('IsPlayable') == 'true', not folder, (name, option)) self.assertEqual(action_of(self.build('widget_tmdb_popular_movies', '')[0].getPath())[0], 'show_wcs_details') # default self.assertEqual(action_of(self.build('widget_game_topics', '')[0].getPath())[0], 'milionar_game') self.assertEqual(action_of(self.build('widget_game_topics', 'detail')[0].getPath())[0], 'show_wcs_details') series = self.build('widget_tmdb_popular_series', 'next')[0].getPath() self.assertEqual(action_of(series)[0], 'ctx_next') ai = {item.label: item for item in self.build('widget_ai_recent', 'stream')} self.assertEqual(action_of(ai['AI serial'].getPath())[1]['src'], 'dialog') class ChannelPlayTests(unittest.TestCase): def setUp(self): xbmcaddon.Addon.settings.clear() self.cache, self.notes = FakeCache(), [] self.addon = xbmcaddon.Addon('plugin.video.milionar') xbmcaddon.Addon.settings.update({'mytv_block_size': '2', 'mytv_upcoming_count': '6'}) xbmcaddon.Addon.strings[32399] = 'Empty' self._note = xbmcgui.Dialog.notification xbmcgui.Dialog.notification = lambda dialog, heading, message, *a, **k: self.notes.append(message) self.progress = {} from wcs.ai import ProgressManager self._get = ProgressManager.get_progress ProgressManager.get_progress = lambda channel_id, series_id: self.progress.get(str(series_id), (1, 0)) self._next = channel_play._next_meta channel_play._next_meta = lambda addon, cache, series_id, season, episode: { 'season': season, 'episode': episode + 1, 'episode_title': 'T{}'.format(episode + 1), 'plot': 'plot', 'poster': 'p', 'fanart': 'f', 'year': '2003', 'genre': 'Komedie', 'rating': '7', 'runtime': '22', 'episode_thumb': 't'} self.channel = types.SimpleNamespace(id='c1', series_list=[{'id': 1, 'name': 'A'}, {'id': 2, 'name': 'B'}]) def tearDown(self): from wcs.ai import ProgressManager ProgressManager.get_progress = self._get channel_play._next_meta = self._next xbmcgui.Dialog.notification = self._note def codes(self, items): return [(i['properties']['series_name'], i['properties']['ep_code']) for i in items] def test_a_queue_is_filled_in_blocks_like_the_dialog(self): items = channel_play.fill(self.addon, self.cache, self.channel, [], 6, shuffle_series=False) self.assertEqual(self.codes(items), [('A', 'S01E01'), ('A', 'S01E02'), ('B', 'S01E01'), ('B', 'S01E02'), ('A', 'S01E03'), ('A', 'S01E04')]) first = items[0] self.assertEqual((first['label'], first['properties']['tmdb_id'], first['art']['thumb']), ('A', '1', 't')) more = channel_play.fill(self.addon, self.cache, self.channel, items[:2], 4, shuffle_series=False) self.assertEqual(len(more), 4) self.assertEqual(len({(i['properties']['tmdb_id'], i['properties']['ep_code']) for i in more}), 4) # never twice def test_watched_episodes_leave_the_queue_and_the_saved_order_wins(self): items = channel_play.fill(self.addon, self.cache, self.channel, [], 4, shuffle_series=False) self.progress = {'1': (1, 1)} # A watched up to E01 self.assertEqual(self.codes(channel_play.drop_watched('c1', items)), [('A', 'S01E02'), ('B', 'S01E01'), ('B', 'S01E02')]) self.cache.set('program_reorder_c1', ['2_S01E02', '1_S01E01'], 0) ordered = channel_play.apply_reorder(self.cache, 'c1', items) self.assertEqual(self.codes(ordered)[:2], [('B', 'S01E02'), ('A', 'S01E01')]) self.assertEqual(len(ordered), 4) # the rest follows self.assertEqual(channel_play.apply_reorder(FakeCache(), 'c1', items), items) def test_the_live_queue_is_stored_back_for_the_dialog_and_fresh_replaces_it(self): queue = channel_play.tv_queue(self.addon, self.cache, self.channel) self.assertEqual(len(queue), 6) # mytv_upcoming_count self.assertEqual(self.cache.get('program_list_c1'), queue) self.assertGreaterEqual(self.cache.flushed, 1) self.progress = {'1': (1, 2)} # A: E01, E02 watched since queue = channel_play.tv_queue(self.addon, self.cache, self.channel) self.assertEqual(len(queue), 6) # topped up again self.assertNotIn(('A', 'S01E01'), self.codes(queue)) self.cache.set('program_reorder_c1', ['x_y'], 0) fresh = channel_play.tv_queue(self.addon, self.cache, self.channel, fresh=True) self.assertEqual(len(fresh), 6) self.assertIsNone(self.cache.get('program_reorder_c1')) # a new shuffle drops the saved order def test_play_tv_modes(self): played = [] manager = types.ModuleType('wcs.ai.MediaChannelManager') manager.MediaChannelManager = lambda addon, kind: types.SimpleNamespace(get_channel_by_id=lambda cid: self.channel if cid == 'c1' else None) caching = types.ModuleType('wcs.caching') caching.get_cache_manager = lambda: self.cache original = dict(sys.modules) sys.modules['wcs.ai.MediaChannelManager'], sys.modules['wcs.caching'] = manager, caching old_play = channel_play.play_episodes channel_play.play_episodes = lambda addon, channel_id, items: played.append((channel_id, len(items))) try: for mode, count in (('live', 6), ('live_shuffle', 6), ('first', 1), ('random', 1)): played.clear() channel_play.play_tv(self.addon, 'c1', mode) self.assertEqual(played, [('c1', count)], mode) channel_play.play_tv(self.addon, 'unknown', 'live') self.assertEqual(played, [('c1', 1)]) self.assertEqual(self.notes, ['Empty']) finally: channel_play.play_episodes = old_play sys.modules.clear() sys.modules.update(original) def test_the_playlist_has_the_dialog_urls_and_a_countdown_monitor(self): monitor = types.SimpleNamespace(calls=[]) monitor.activate = lambda meta=None: monitor.calls.append(meta) fake = types.ModuleType('wcs.playback.PlaylistCountdownMonitor') fake.get_playlist_monitor = lambda: monitor original, old_keep = dict(sys.modules), channel_play._keep_alive sys.modules['wcs.playback.PlaylistCountdownMonitor'] = fake channel_play._keep_alive = lambda *a, **k: None xbmc.Player.played.clear() try: items = channel_play.fill(self.addon, self.cache, self.channel, [], 3, shuffle_series=False) channel_play.play_episodes(self.addon, 'c1', items) playlist = xbmc.Player.played[-1] self.assertEqual(playlist.size(), 3) action, params = action_of(playlist.items[0][0]) self.assertEqual((action, params['mytv_channel_id'], params['series_name'], params['season_number'], params['episode_number']), ('play_episode_from_addon_playlist', 'c1', 'A', '1', '1')) self.assertEqual(monitor.calls, [None]) # several items: playlist mode channel_play.play_episodes(self.addon, 'c1', items[:1]) self.assertEqual(monitor.calls[-1]['tmdb_id'], '1') # one item: the monitor gets its meta finally: channel_play._keep_alive = old_keep sys.modules.clear() sys.modules.update(original) def test_a_channel_without_series_says_so(self): manager = types.ModuleType('wcs.ai.MediaChannelManager') manager.MediaChannelManager = lambda addon, kind: types.SimpleNamespace( get_channel_by_id=lambda cid: types.SimpleNamespace(id=cid, series_list=[])) original = dict(sys.modules) sys.modules['wcs.ai.MediaChannelManager'] = manager try: channel_play.run({'kind': 'tv', 'channel_id': 'c1', 'mode': 'live'}) finally: sys.modules.clear() sys.modules.update(original) self.assertEqual(self.notes, ['Empty']) if __name__ == '__main__': unittest.main()