# -*- coding: utf-8 -*- """L2 unit tests: autoplay engine and collection offer (docs/plan/milionar-library/050 §4-5, 120 §2).""" import json import os import queue import sys import tempfile import unittest from datetime import date 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.library2 import autoplay, fixture_provider, meta, store, tracker # noqa: E402 TODAY = date(2026, 6, 15) class FakePlayer: def __init__(self): self.file, self.time, self.total, self.events = '', 0.0, 0.0, queue.Queue() def getPlayingFile(self): return self.file def getTime(self): return self.time def getTotalTime(self): return self.total def seekTime(self, seconds): self.time = seconds class TestEngine(tracker.Engine): """Engine with Kodi side effects recorded and worker threads run inline.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.dialog_action = 'timeout' self.played, self.dropped, self.notes, self.dialogs = [], [], [], [] def _spawn(self, target, *args): target(*args) def _show_dialog(self, plan, duration, token): self.dialogs.append((plan['kind'], duration)) return self.dialog_action def play_next(self, plan): self.played.append(plan['canonical']) def drop_queue(self, s): if s.queued_url: self.dropped.append(s.queued_url) s.queued_url = None def _text(self, string_id): return {32280: 'Next {} airs {}', 32281: 'Series finished', 32283: 'Next in collection'}.get(string_id, str(string_id)) def _notify(self, message): self.notes.append(message) class AutoplayTests(unittest.TestCase): STREAM = 'https://cdn.example/ep.mkv' def setUp(self): store.set_path(os.path.join(tempfile.mkdtemp(prefix='milionar-db-'), 'library.db')) xbmcgui.Window.props.clear() xbmcaddon.Addon.settings.clear() xbmcaddon.Addon.settings.update({'autoplay_enabled_global': 'true', 'autoplay_countdown_seconds': '45', 'autoplay_trigger_seconds': '15', 'lib_resume_min_seconds': '10'}) fixture_provider.seed_meta() self.clock = [1000.0] self.player = FakePlayer() self.engine = TestEngine(self.player, xbmc.Monitor(), now=lambda: self.clock[0]) autoplay_today = lambda: TODAY # noqa: E731 self._real_date = autoplay.date autoplay.date = type('D', (), {'today': staticmethod(autoplay_today)}) def tearDown(self): autoplay.date = self._real_date def start(self, kind='episode', tmdb=900001, s=1, e=10, total=2400.0, **extra): contract = {'token': 't{}'.format(self.clock[0]), 'type': kind, 'tmdb': tmdb, 's': s, 'e': e, 'stream_url': self.STREAM, 'ident': 'abc', 'provider': 'webshare', 'file_name': 'f.mkv', 'start_at': 0, 'ts': self.clock[0], 'runtime_s': 0} contract.update(extra) xbmcgui.Window.props['milionar.np'] = json.dumps(contract) self.player.file, self.player.total, self.player.time = self.STREAM, total, 0.0 self.engine.handle('started', self.clock[0]) self.pump() # the prefetched plan arrives like in the service def play_to(self, position): self.clock[0] += 20 self.player.time = position self.engine.tick() self.pump() def pump(self): while not self.player.events.empty(): event, when, data = self.player.events.get_nowait() self.engine.handle(event, when, data) def request(self): raw = xbmcgui.Window().getProperty('milionar.play_req') return json.loads(raw) if raw else None NEXT = 'plugin://plugin.video.milionar/?action=mplay&type=episode&tmdb=900001&s=2&e=1' def test_before_countdown_nothing_happens(self): self.start() self.play_to(2000) # 400 s left, countdown at 45 s self.assertEqual((self.engine.played, self.engine.dialogs), ([], [])) def test_rollover_timeout_marks_watched_and_plays_next_once(self): self.start() self.play_to(2360) # 40 s left: countdown point passed self.assertEqual(self.engine.dialogs, [('series_next', 25)]) # min(40 - 15, 45 - 15) self.assertEqual(self.engine.played, [self.NEXT]) req = self.request() self.assertEqual((req['canonical'], req['src'], req['from']), (self.NEXT, 'auto', 'start')) self.assertGreater(req['max_age'], 100) # lives until the lazy resolve self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) # a second action (the user also pressed play now) must not start it twice self.engine.handle('autoplay_action', 0, {'token': self.engine.session.np['token'], 'action': 'play_now'}) self.assertEqual(self.engine.played, [self.NEXT]) # Kodi switches to the next item: the old session ends without a second count and without a replay self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) self.assertEqual(store.recent_logs(1)[0]['outcome'], 'watched') def test_play_now_same_as_timeout(self): self.engine.dialog_action = 'play_now' self.start() self.play_to(2360) self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) def test_cancel_drops_the_queue_and_plays_to_the_end(self): self.engine.dialog_action = 'cancel' self.start() self.play_to(2360) self.assertEqual((self.engine.dropped, self.engine.played), ([self.NEXT], [])) self.play_to(2395) self.assertEqual(self.engine.dialogs, [('series_next', 25)]) # not offered again self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, []) # and nothing follows self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) # watched by the natural end def test_dismiss_lets_the_episode_end_then_next_follows(self): self.engine.dialog_action = 'dismiss' self.start() self.play_to(2360) self.assertEqual((self.engine.played, self.engine.dropped), ([], [])) self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) def test_user_stop_drops_the_queue(self): self.engine.dialog_action = 'dismiss' self.start() self.play_to(2360) self.engine.handle('stopped', self.clock[0]) self.assertEqual((self.engine.dropped, self.engine.played), ([self.NEXT], [])) def test_next_not_aired_notifies_at_the_end(self): self.start(s=2, e=6, total=2400.0) self.play_to(2360) self.assertEqual((self.engine.played, self.engine.dialogs), ([], [])) self.assertEqual(self.engine.notes, []) # not yet, only at the end self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.notes, ['Next S02E07 airs 22. 6.']) self.assertEqual(self.engine.played, []) def test_last_episode_of_ended_series(self): self.start(tmdb=900002, s=1, e=3) self.play_to(2360) self.assertEqual(self.engine.dialogs, []) self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.notes, ['Series finished']) def test_special_has_no_followup(self): self.start(s=0, e=1) self.play_to(2360) self.assertEqual((self.engine.played, self.engine.dialogs, self.engine.notes), ([], [], [])) def test_autoplay_disabled(self): xbmcaddon.Addon.settings['autoplay_enabled_global'] = 'false' self.start() self.play_to(2390) self.engine.handle('ended', self.clock[0]) self.assertEqual((self.engine.played, self.engine.dialogs), ([], [])) def test_collection_offer_never_starts_by_itself(self): self.engine.dialog_action = 'timeout' self.start(kind='movie', tmdb=900101, s=None, e=None, total=6000.0) self.play_to(5960) self.assertEqual(self.engine.dialogs, [('collection', 25)]) self.assertEqual(self.engine.played, []) self.assertEqual(self.engine.session.plan['canonical'], 'plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=900102') self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, []) # still nothing def test_collection_play_marks_current_and_starts_next(self): self.engine.dialog_action = 'play_now' self.start(kind='movie', tmdb=900101, s=None, e=None, total=6000.0) self.play_to(5960) self.assertEqual(self.engine.played, ['plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=900102']) self.assertEqual(store.get_state('movie', 900101)['playcount'], 1) self.assertEqual(self.request()['src'], 'auto') def test_movie_without_collection(self): meta.put(meta.key_movie(777), {'id': 777, 'title': 'Solo', 'release_date': '2020-01-01'}, 10 ** 9) self.start(kind='movie', tmdb=777, s=None, e=None, total=6000.0) self.play_to(5960) self.assertEqual((self.engine.dialogs, self.engine.played), ([], [])) # prefetch / follow-up at the natural end (140 §2) def test_plan_is_prefetched_at_start_and_shows_nothing_yet(self): calls = [] real = autoplay.plan autoplay.plan = lambda np: calls.append(np['e']) or real(np) try: self.start() finally: autoplay.plan = real self.assertEqual(calls, [10]) self.assertEqual(self.engine.session.plan['kind'], 'series_next') self.assertEqual((self.engine.dialogs, self.engine.played, self.engine.notes), ([], [], [])) self.assertIsNone(self.request()) # no request until the countdown point def test_countdown_uses_the_prefetched_plan_without_a_new_lookup(self): self.start() real = autoplay.plan autoplay.plan = lambda np: self.fail('plan computed again') try: self.play_to(2360) finally: autoplay.plan = real self.assertEqual(self.engine.played, [self.NEXT]) def test_seek_to_the_very_end_still_plays_next_once_and_counts_once(self): self.engine.dialog_action = None # the dialog never gets an answer self.start() self.play_to(2399) # 1 s left: dialog too late to matter self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) self.assertEqual(self.request()['src'], 'auto') self.engine.handle('ended', self.clock[0]) # a stray second end event: no session, nothing self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) def test_seek_straight_to_the_end_without_any_countdown_tick(self): self.start() self.player.time = 2399.0 self.engine.handle('seek', self.clock[0]) self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) def test_natural_end_counts_even_when_the_last_poll_was_stale(self): self.start() self.play_to(5) # last poll long before the seek to the end self.clock[0] += 60 self.engine.handle('ended', self.clock[0]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) self.assertEqual(self.engine.played, [self.NEXT]) def test_a_stream_that_breaks_off_is_not_watched_and_plays_nothing_next(self): self.start() self.play_to(3) # Kodi read 3 s, then the file "ended" self.engine.handle('ended', self.clock[0]) self.assertEqual((store.get_state('tv', 900001, 1, 10) or {}).get('playcount', 0), 0) self.assertEqual(self.engine.played, []) def test_a_file_cut_short_keeps_its_resume_point(self): self.start() self.play_to(900) # 15 min in, far from the end, then EOF self.engine.handle('ended', self.clock[0]) state = store.get_state('tv', 900001, 1, 10) self.assertEqual(state['playcount'], 0) self.assertAlmostEqual(state['resume_pos'], 900.0) def test_missing_plan_is_computed_at_the_end(self): self.engine.session = None self.start() self.engine.session.plan = None # prefetch has not arrived (slow network) self.player.time = 2399.0 self.engine.handle('seek', self.clock[0]) self.engine.handle('ended', self.clock[0]) self.assertEqual(self.engine.played, [self.NEXT]) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) def test_autoplay_off_does_nothing_at_the_end(self): xbmcaddon.Addon.settings['autoplay_enabled_global'] = 'false' self.start() self.player.time = 2399.0 self.engine.handle('seek', self.clock[0]) self.engine.handle('ended', self.clock[0]) self.assertEqual((self.engine.played, self.engine.dialogs, self.engine.notes), ([], [], [])) self.assertEqual(store.get_state('tv', 900001, 1, 10)['playcount'], 1) # still watched by the end def test_stale_prefetch_of_another_playback_is_ignored(self): self.start() self.engine.session.plan = None self.engine.handle('autoplay_prefetch', 0, {'token': 'other', 'plan': {'kind': 'finished'}}) self.assertIsNone(self.engine.session.plan) def test_no_resume_point_is_autosaved_past_the_watched_mark(self): self.start() self.play_to(2000) # 83 %: saved self.assertEqual(int(store.get_state('tv', 900001, 1, 10)['resume_pos']), 2000) self.play_to(2300) # 96 %: a crash now must not leave this behind self.assertEqual(int(store.get_state('tv', 900001, 1, 10)['resume_pos']), 2000) def test_dialog_duration_rules(self): self.assertEqual(autoplay.dialog_duration(40, 45, 15), 25) self.assertEqual(autoplay.dialog_duration(45, 45, 15), 30) self.assertEqual(autoplay.dialog_duration(10, 45, 15), 3) xbmcaddon.Addon.settings.update({'autoplay_countdown_seconds': '5', 'autoplay_trigger_seconds': '50'}) self.assertEqual(autoplay.settings(), (True, 10, 5)) # clamped like the MyTV monitor if __name__ == '__main__': unittest.main()