# -*- coding: utf-8 -*- """Offline playback scenarios; no Kodi, GUI, network or provider calls.""" import ast import copy import importlib.util import json import os import sys import unittest import zipfile import zipimport from unittest.mock import Mock, patch sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'scripts', 'library_harness')) import stubs stubs.install() import xbmc from wcs import playback release_zip = os.environ.get('MILIONAR_TEST_RELEASE_ZIP') if release_zip: loader = zipimport.zipimporter(os.path.join(os.path.abspath(release_zip), 'plugin.video.milionar', 'wcs', 'playback')) spec = importlib.util.spec_from_loader('wcs.playback.language', loader) language = importlib.util.module_from_spec(spec) spec.loader.exec_module(language) sys.modules[spec.name] = language playback.language = language else: from wcs.playback import language def stream(index, tag, **extra): return dict({'index': index, 'language': tag, 'name': '', 'channels': 2}, **extra) class FakePlayer(language.LanguagePlayer): def __init__(self, clock, properties): super().__init__(now=lambda: clock[0]) self.properties = copy.deepcopy(properties) self.playing = True self.path = 'https://example.invalid/video.mkv' self.actions = [] def isPlayingVideo(self): return self.playing def getPlayingFile(self): return self.path def setAudioStream(self, index): self.actions.append(('audio', index)) self.properties['currentaudiostream'] = next(s for s in self.properties['audiostreams'] if s['index'] == index) def setSubtitleStream(self, index): self.actions.append(('subtitle', index)) self.properties['currentsubtitle'] = next(s for s in self.properties['subtitles'] if s['index'] == index) def showSubtitles(self, visible): self.actions.append(('visible', visible)) self.properties['subtitleenabled'] = visible class PlaybackLanguageTests(unittest.TestCase): def setUp(self): self.clock = [100.0] self.properties = {'audiostreams': [stream(0, 'eng'), stream(1, 'cze'), stream(2, 'slo')], 'currentaudiostream': stream(0, 'eng'), 'subtitles': [stream(0, 'eng'), stream(1, 'ces')]} self.player = FakePlayer(self.clock, self.properties) self.rpc = patch.object(xbmc, 'executeJSONRPC', side_effect=self.reply).start() self.addCleanup(patch.stopall) def reply(self, request): request = json.loads(request) self.assertEqual(request['method'], 'Player.GetProperties') self.assertEqual(request['params']['playerid'], 1) return json.dumps({'result': self.player.properties}) def start(self): self.player.onAVStarted() self.player.tick() self.player.tick() # confirm the actual player state before ending initialization def test_czech_audio_disables_all_subtitles(self): self.start() self.assertEqual(self.player.actions, [('audio', 1), ('visible', False)]) self.assertIsNone(self.player.deadline) def test_czech_already_selected_still_disables_subtitles(self): self.player.properties['currentaudiostream'] = stream(1, 'cze') self.start() self.assertEqual(self.player.actions, [('visible', False)]) def test_language_codes_and_region_tags(self): for tag in ('cze', 'ces', 'cs', 'cs-CZ', 'cs_CZ', 'CZE'): with self.subTest(tag=tag): self.player.properties['audiostreams'][1]['language'] = tag self.player.actions.clear() self.start() self.assertIn(('visible', False), self.player.actions) self.assertEqual(self.player.properties['currentaudiostream']['index'], 1) def test_plain_dubbing_before_commentary_and_audio_description(self): self.player.properties['audiostreams'] = [ stream(0, 'eng'), stream(1, 'cze', channels=2), stream(2, 'ces', channels=8, name='Český komentář'), stream(3, 'ces', channels=8, isimpaired=True), stream(4, 'cze', channels=6)] self.start() self.assertEqual(self.player.actions, [('audio', 4), ('visible', False)]) def test_english_with_czech_subtitles_never_slovak(self): self.player.properties['audiostreams'] = [stream(0, 'slo'), stream(1, 'eng')] self.player.properties['currentaudiostream'] = stream(0, 'slo') self.start() self.assertEqual(self.player.actions, [('audio', 1), ('subtitle', 1), ('visible', True)]) def test_english_already_selected_is_kept(self): self.player.properties['audiostreams'] = [stream(0, 'eng')] self.start() self.assertEqual(self.player.actions, [('subtitle', 1), ('visible', True)]) def test_full_czech_subtitles_before_forced_or_impaired(self): self.player.properties['audiostreams'] = [stream(0, 'en')] self.player.properties['subtitles'] = [stream(0, 'en'), stream(1, 'cs', isforced=True), stream(2, 'cs', isimpaired=True), stream(3, 'cs')] self.start() self.assertEqual(self.player.actions, [('subtitle', 3), ('visible', True)]) def test_forced_subtitles_disabled_with_czech_audio(self): self.player.properties['subtitles'] = [stream(0, 'cs', isforced=True)] self.start() self.assertEqual(self.player.actions[-1], ('visible', False)) def test_english_subtitles_not_enabled_when_czech_missing(self): self.player.properties['audiostreams'] = [stream(0, 'eng')] self.player.properties['subtitles'] = [stream(0, 'eng'), stream(1, 'slo')] self.start() self.assertEqual(self.player.actions, [('visible', False)]) self.assertIsNone(self.player.deadline) def test_delayed_czech_subtitles(self): self.player.properties['audiostreams'] = [stream(0, 'eng')] self.player.properties['subtitles'] = [] self.start() self.clock[0] += 1 self.player.properties['subtitles'] = [stream(0, 'eng'), stream(1, 'cs-CZ')] self.player.properties['currentsubtitle'] = stream(0, 'eng') self.player.properties['subtitleenabled'] = True # Kodi may activate its default as subtitles appear. self.player.tick() self.assertEqual(self.player.actions, [('subtitle', 1), ('visible', True)]) def test_delayed_audio_streams(self): self.player.properties['audiostreams'] = [] self.start() self.player.properties['audiostreams'] = self.properties['audiostreams'] self.player.tick() self.assertEqual(self.player.actions, [('audio', 1), ('visible', False)]) def test_manual_changes_remain_after_initial_selection(self): self.start() self.player.properties['currentaudiostream'] = stream(0, 'eng') self.player.properties['subtitles'] = [stream(0, 'eng')] self.player.actions.clear() reads = self.rpc.call_count for _ in range(10): self.clock[0] += 1 self.player.tick() self.assertEqual(self.player.actions, []) self.assertEqual(self.rpc.call_count, reads) def test_next_episode_and_replay_reapply_preferences(self): self.start() self.player.actions.clear() self.player.properties['currentaudiostream'] = stream(0, 'eng') self.start() # Same URL can be played again; every AV start is a new session. self.assertEqual(self.player.actions, [('audio', 1), ('visible', False)]) def test_every_video_is_initialized_without_item_properties(self): self.start() self.assertEqual(self.player.actions, [('audio', 1), ('visible', False)]) self.assertIsNone(self.player.deadline) def test_music_is_ignored(self): self.player.playing = False self.start() self.rpc.assert_not_called() def test_unapplied_audio_is_retried_until_confirmed(self): original = self.player.setAudioStream with patch.object(self.player, 'setAudioStream') as setter: self.player.onAVStarted() self.player.tick() self.assertIsNotNone(self.player.deadline) self.player.tick() self.assertEqual(setter.call_count, 2) self.assertIsNotNone(self.player.deadline) original(1) self.player.tick() self.assertIsNone(self.player.deadline) def test_stop_cancels_pending_selection(self): self.player.properties['audiostreams'] = [] self.start() self.player.onPlayBackStopped() self.player.tick() self.assertIsNone(self.player.deadline) self.assertEqual(self.player.actions, []) def test_file_change_during_startup_cancels_selection(self): self.player.properties['audiostreams'] = [] self.start() self.player.path = 'https://example.invalid/other-video.mkv' self.player.properties = self.properties self.player.tick() self.assertEqual(self.player.actions, []) self.assertIsNone(self.player.deadline) def test_rpc_error_retries_only_during_startup(self): self.rpc.side_effect = None self.rpc.return_value = json.dumps({'error': {'code': -32100}}) self.start() self.assertIsNotNone(self.player.deadline) self.clock[0] += language.STARTUP_TIMEOUT_S self.player.tick() self.assertIsNone(self.player.deadline) self.assertEqual(self.player.actions, []) def test_name_used_only_when_language_tag_missing(self): self.assertTrue(language._language(stream(0, 'und', name='Czech AC3 5.1'), 'cs')) self.assertTrue(language._language(stream(0, '', name='Český dabing'), 'cs')) self.assertFalse(language._language(stream(0, 'slo', name='Czech'), 'cs')) self.assertFalse(language._language(stream(0, 'und', name='AC3 5.1'), 'cs')) class ReleaseServiceTests(unittest.TestCase): def test_packaged_service_dispatches_callbacks_after_startup(self): archive = os.environ.get('MILIONAR_TEST_RELEASE_ZIP') if not archive: self.skipTest('Set MILIONAR_TEST_RELEASE_ZIP to verify the installable ZIP') self.assertEqual(os.path.abspath(language.__file__), os.path.join( os.path.abspath(archive), 'plugin.video.milionar', 'wcs', 'playback', 'language.py')) with zipfile.ZipFile(archive) as package: source = package.read('plugin.video.milionar/service.py') self.assertNotIn(b'run_dev_harness', source) tree = ast.parse(source) entrypoint = next(node for node in tree.body if isinstance(node, ast.If)) # Kodi delivers Player callbacks on the thread where the Player was created. # Startup alone must not leave that thread without a callback dispatch loop. started = Mock() delivered = Mock() monitor = Mock() monitor.abortRequested.side_effect = [False, True] monitor.waitForAbort.side_effect = lambda *args: delivered() context = {'__name__': '__main__', 'xbmc': xbmc, 'main': lambda: self.assertTrue(started.called), 'start_library_tracker': Mock(), 'start_skin_installer': Mock(), 'start_skin_properties': Mock(), 'start_foryou': Mock()} with patch.object(language, 'start', started), patch('threading.Timer'), \ patch.object(xbmc, 'Monitor', return_value=monitor): exec(compile(ast.Module(body=[entrypoint], type_ignores=[]), 'service.py', 'exec'), context) delivered.assert_called() if __name__ == '__main__': unittest.main()