#!/usr/bin/env python3 """Offline watched-state transactions, durable retries, version races and bounded adoption.""" import json import os import queue import sys import tempfile import unittest from datetime import date from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import stubs stubs.install() import xbmc import xbmcaddon import xbmcgui from wcs.library2 import ctx, jobs, meta, mirror, store, tracker, urls class MarkPerformanceTests(unittest.TestCase): def setUp(self): self.folder = tempfile.TemporaryDirectory(prefix='milionar-mark-') self.addCleanup(self.folder.cleanup) store.set_path(os.path.join(self.folder.name, 'library.db')) xbmc.RPC_CALLS.clear() xbmc.RPC_REPLIES.clear() xbmcgui.NOTIFICATIONS.clear() xbmcgui.Window.props.clear() xbmcaddon.Addon.settings.clear() while not jobs._adoptions.empty(): jobs._adoptions.get_nowait() def seed(self, show_id=10, seasons=2, episodes=3): show = {'seasons': [{'season_number': s, 'episode_count': episodes} for s in range(1, seasons + 1)]} meta.put(meta.key_tv(show_id), show, 86400) for s in range(1, seasons + 1): meta.put(meta.key_season(show_id, s), {'season_number': s, 'episodes': [ {'episode_number': e, 'air_date': '2020-01-01'} for e in range(1, episodes + 1)]}, 86400) return show def pending(self): conn = store.connect() try: return [dict(r) for r in conn.execute('SELECT * FROM pending_marks ORDER BY id')] finally: conn.close() def test_large_cached_series_one_transaction_no_rpc_or_network(self): self.seed(seasons=20, episodes=20) with patch.object(store, 'tx', wraps=store.tx) as tx, \ patch.object(store, 'mark_many', wraps=store.mark_many) as mark, \ patch.object(jobs.tmdb, '_fetch', side_effect=AssertionError('network')): self.assertEqual(ctx.mark('tv', 10), 400) self.assertEqual((tx.call_count, mark.call_count), (1, 1)) self.assertEqual(len(store.get_episode_states(10)), 400) self.assertEqual(xbmc.RPC_CALLS, []) self.assertEqual(len(store.pending_mirrors(500, float('inf'))), 400) self.assertEqual(ctx.mark('tv', 10), 0) self.assertEqual({r['playcount'] for r in store.get_episode_states(10)}, {1}) def test_movie_and_episode_one_transaction_without_metadata(self): for kind, media, season, episode in [('movie', 'movie', None, None), ('episode', 'tv', 1, 2)]: with self.subTest(kind=kind), patch.object(store, 'tx', wraps=store.tx) as tx, \ patch.object(jobs.tmdb, '_fetch', side_effect=AssertionError('network')): self.assertEqual(ctx.mark(kind, 77, season, episode), 1) self.assertEqual(tx.call_count, 1) self.assertEqual(store.get_state(media, 77, season, episode)['playcount'], 1) self.assertEqual(xbmc.RPC_CALLS, []) def test_identical_resume_preserves_source_activity_versions_and_mirror_queue(self): from wcs.skin import refresh source = {'ident': 'original', 'provider': 'webshare', 'file_name': 'original.mkv'} for media, season, episode in [('movie', None, None), ('tv', 1, 2)]: with self.subTest(media=media): store.save_resume(media, 77, 300, 6000, season, episode, source=source, when=1000) state = store.get_state(media, 77, season, episode) title = store.get_title(media, 77) versions = refresh.revisions(refresh.media_rows(media)) pending = store.pending_mirrors(5, float('inf')) with patch.object(xbmc, 'executebuiltin') as builtin: for when, repeated_source in [(1100, None), (1200, source), (1300, {'provider': 'webshare', 'ident': None})]: store.save_resume(media, 77, 300, 6000, season, episode, source=repeated_source, when=when) builtin.assert_not_called() self.assertEqual(store.get_state(media, 77, season, episode), state) self.assertEqual(store.get_title(media, 77), title) self.assertEqual(refresh.revisions(refresh.media_rows(media)), versions) self.assertEqual(store.pending_mirrors(5, float('inf')), pending) store.set_mirrored(media, 77, 0, 300, season, episode, version=state['state_version']) mirrored = store.get_state(media, 77, season, episode) store.save_resume(media, 77, 300, 6000, season, episode, when=1400) self.assertEqual(store.get_state(media, 77, season, episode), mirrored) self.assertFalse(store.mirror_pending(media, 77, season, episode)) def test_resume_source_change_bumps_versions_and_preserves_other_source_fields(self): from wcs.skin import refresh source = {'ident': 'original', 'provider': 'webshare', 'file_name': 'original.mkv'} for media, season, episode in [('movie', None, None), ('tv', 1, 2)]: for tmdb_id, field in enumerate(source, 80): with self.subTest(media=media, field=field): store.save_resume(media, tmdb_id, 300, 6000, season, episode, source=source, when=1000) before = store.get_state(media, tmdb_id, season, episode) versions = refresh.revisions(refresh.media_rows(media)) store.set_mirrored(media, tmdb_id, 0, 300, season, episode, version=before['state_version']) store.save_resume(media, tmdb_id, 300, 6000, season, episode, source={field: 'changed'}, when=2000) state = store.get_state(media, tmdb_id, season, episode) self.assertEqual(state['state_version'], before['state_version'] + 1) self.assertEqual((state['resume_pos'], state['resume_total']), (300, 6000)) self.assertEqual(state['last_played_at'], 2000) self.assertEqual(store.get_title(media, tmdb_id)['last_activity_at'], 2000) self.assertEqual(refresh.revisions(refresh.media_rows(media)), {row: version + 1 for row, version in versions.items()}) self.assertEqual({key: state['last_' + key] for key in source}, dict(source, **{field: 'changed'})) pending = [r for r in store.pending_mirrors(20, float('inf')) if r['media_type'] == media and r['tmdb_id'] == tmdb_id] self.assertEqual(len(pending), 1) self.assertEqual(pending[0]['version'], state['state_version']) def test_incomplete_cache_is_queued_without_partial_state(self): self.seed() meta.delete(meta.key_season(10, 2)) with patch.object(jobs.tmdb, '_fetch', side_effect=AssertionError('network')), \ patch.object(xbmcaddon.Addon, 'getLocalizedString', side_effect=lambda i: 'translated-{}'.format(i)): self.assertEqual(ctx.mark('tv', 10), 0) self.assertEqual(store.get_episode_states(10), []) self.assertEqual(len(self.pending()), 1) self.assertEqual(xbmcgui.NOTIFICATIONS[-1][1], 'translated-32653') self.assertEqual(xbmc.RPC_CALLS, []) def test_short_season_and_empty_results_are_never_success(self): self.seed() meta.put(meta.key_season(10, 2), {'episodes': [ {'episode_number': e, 'air_date': '2020-01-01'} for e in (1, 3, 4)]}, 86400) self.assertIsNone(jobs.cached_targets('tv', 10)) meta.put(meta.key_season(10, 2), {'episodes': [{'episode_number': 1, 'air_date': '2020-01-01'}]}, 86400) self.assertIsNone(jobs.cached_targets('tv', 10)) meta.put(meta.key_season(10, 2), {'episodes': []}, 86400) self.assertIsNone(jobs.cached_targets('tv', 10)) ctx.mark('tv', 10) with patch.object(jobs.tmdb, '_fetch', return_value={'episodes': []}), \ patch.object(jobs.tmdb, '_cached', wraps=jobs.tmdb._cached) as cached: self.assertEqual(jobs.process_marks(now=100), 0) self.assertTrue(all(call.kwargs.get('strict') for call in cached.call_args_list)) self.assertEqual(store.get_episode_states(10), []) self.assertEqual(self.pending()[0]['attempts'], 1) def test_cached_marks_keep_aired_and_specials_semantics(self): meta.put(meta.key_tv(10), {'seasons': [{'season_number': 0, 'episode_count': 1}, {'season_number': 1, 'episode_count': 4}]}, 86400) meta.put(meta.key_season(10, 0), {'episodes': [{'episode_number': 1, 'air_date': '2020-01-01'}]}, 86400) meta.put(meta.key_season(10, 1), {'episodes': [ {'episode_number': e, 'air_date': air} for e, air in [(1, '2020-01-01'), (2, '2026-06-16'), (3, None), (4, '2026-06-15')]]}, 86400) frozen = type('FrozenDate', (), {'today': staticmethod(lambda: date(2026, 6, 15))}) with patch.object(jobs, 'date', frozen): self.assertEqual(jobs.cached_targets('tv', 10), [(1, 1), (1, 4)]) self.assertEqual(jobs.cached_targets('season', 10, 0), [(0, 1)]) self.assertEqual(ctx.mark('tv', 10), 2) self.assertIsNone(store.get_state('tv', 10, 0, 1)) self.assertIsNone(store.get_state('tv', 10, 1, 2)) self.assertIsNone(store.get_state('tv', 10, 1, 3)) def test_worker_fetch_is_strict_and_preserves_later_episode_intent(self): ctx.mark('tv', 10) ctx.mark('episode', 10, 1, 2, watched=False) self.assertEqual([r['kind'] for r in self.pending()], ['tv', 'episode']) show = {'seasons': [{'season_number': 1, 'episode_count': 2}]} data = {'episodes': [{'episode_number': e, 'air_date': '2020-01-01'} for e in (1, 2)]} with patch.object(jobs.tmdb, '_fetch', side_effect=lambda path, params: show if path == 'tv/10' else data) as fetch, \ patch.object(jobs.tmdb, '_cached', wraps=jobs.tmdb._cached) as cached: self.assertEqual(jobs.process_marks(now=100), 1) self.assertEqual([call.args[0] for call in fetch.call_args_list], ['tv/10', 'tv/10/season/1']) self.assertTrue(all(call.kwargs.get('strict') for call in cached.call_args_list)) self.assertEqual(store.get_state('tv', 10, 1, 2)['playcount'], 1) self.assertEqual(jobs.process_marks(now=100), 1) self.assertEqual(store.get_state('tv', 10, 1, 2)['playcount'], 0) self.assertEqual(self.pending(), []) def test_backoff_is_durable_bounded_and_blocks_only_its_show(self): ctx.mark('tv', 10) ctx.mark('episode', 10, 1, 1, watched=False) ctx.mark('tv', 20) with patch.object(jobs.tmdb, '_fetch', side_effect=RuntimeError('offline')): self.assertEqual(jobs.process_marks(now=100), 0) pending = self.pending() self.assertEqual((pending[0]['attempts'], pending[0]['next_attempt']), (1, 102)) self.assertEqual([r['tmdb_id'] for r in jobs.pending_marks(5, 101)], [20]) with store.tx() as conn: conn.execute('UPDATE pending_marks SET attempts=99,next_attempt=0 WHERE id=?', (pending[0]['id'],)) with patch.object(jobs.tmdb, '_fetch', return_value=None): jobs.process_marks(now=200) self.assertEqual(self.pending()[0]['next_attempt'], 800) store.set_path(store.db_path()) # reopen the same database as after a service restart self.assertEqual(self.pending()[0]['attempts'], 100) def test_failed_metadata_load_keeps_old_snapshot(self): self.seed() old = {'episodes': [{'episode_number': 1, 'air_date': '2020-01-01'}]} meta.put(meta.key_season(10, 2), old, 86400) ctx.mark('tv', 10) with patch.object(jobs.tmdb, '_fetch', side_effect=RuntimeError('offline')): self.assertEqual(jobs.process_marks(now=100), 0) self.assertEqual(meta.get(meta.key_season(10, 2))[0], old) self.assertEqual(store.get_episode_states(10), []) def test_queue_delete_failure_rolls_back_watch_state(self): jobs.enqueue('tv', 10, 'episode', 1, 1, True) with store.tx() as conn: conn.execute("CREATE TRIGGER fail_completion BEFORE DELETE ON pending_marks BEGIN " "SELECT RAISE(ABORT, 'crash at completion'); END") self.assertEqual(jobs.process_marks(now=100), 0) self.assertIsNone(store.get_state('tv', 10, 1, 1)) self.assertFalse(store.mirror_pending('tv', 10, 1, 1)) self.assertEqual(self.pending()[0]['attempts'], 1) with store.tx() as conn: conn.execute('DROP TRIGGER fail_completion') self.assertEqual(jobs.process_marks(now=102), 1) self.assertEqual(store.get_state('tv', 10, 1, 1)['playcount'], 1) self.assertEqual(self.pending(), []) def test_queued_noop_deletes_job_atomically_without_second_view(self): store.mark_watched('tv', 10, 1, 1) store.mark_watched('tv', 10, 1, 1) jobs.enqueue('tv', 10, 'episode', 1, 1, True) with patch.object(store, 'tx', wraps=store.tx) as tx: self.assertEqual(jobs.process_marks(now=100), 1) self.assertEqual(tx.call_count, 1) self.assertEqual(self.pending(), []) self.assertEqual(store.get_state('tv', 10, 1, 1)['playcount'], 2) def test_reset_during_loaded_job_never_resurrects_watch_state(self): jobs.enqueue('tv', 10, 'tv', None, None, True) def loaded(*args): store.reset_watch_state() return [(1, 1)] with patch.object(jobs, '_load_targets', side_effect=loaded): jobs.process_marks(now=100) self.assertIsNone(store.get_state('tv', 10, 1, 1)) self.assertEqual(self.pending(), []) self.assertEqual(store.pending_mirrors(5, float('inf')), []) def test_reset_and_new_job_cannot_reuse_inflight_job_id(self): jobs.enqueue('tv', 10, 'tv', None, None, True) old_id = self.pending()[0]['id'] def loaded(*args): store.reset_watch_state() jobs.enqueue('tv', 10, 'episode', 1, 2, False) return [(1, 1)] with patch.object(jobs, '_load_targets', side_effect=loaded): jobs.process_marks(now=100) self.assertGreater(self.pending()[0]['id'], old_id) self.assertEqual(self.pending()[0]['kind'], 'episode') self.assertIsNone(store.get_state('tv', 10, 1, 1)) def test_toggle_uses_pending_intent_before_cache_changes(self): with patch.object(jobs.tmdb, '_fetch', side_effect=AssertionError('network')): ctx.toggle_watched('tv', 10) ctx.toggle_watched('tv', 10) self.assertEqual([r['watched'] for r in self.pending()], [1, 0]) def test_mirror_worker_batch_and_durable_retry(self): for movie in range(7): ctx.mark('movie', movie + 1) with patch.object(mirror, 'push', return_value=False) as push: self.assertEqual(jobs.process_mirrors(float('inf')), 0) self.assertEqual(push.call_count, 5) conn = store.connect() try: attempts = [r[0] for r in conn.execute('SELECT attempts FROM mirror_pending')] finally: conn.close() self.assertEqual(sorted(attempts), [0, 0, 1, 1, 1, 1, 1]) self.assertEqual(jobs.process_mirrors(float('inf')), 5) self.assertEqual(len(store.pending_mirrors(10, float('inf'))), 2) def test_stale_mirror_completion_keeps_latest_version_queued(self): ctx.mark('movie', 1) original = xbmc.executeJSONRPC def rpc(raw): if json.loads(raw)['method'] == 'Files.SetFileDetails': ctx.mark('movie', 1, watched=False) return original(raw) with patch.object(xbmc, 'executeJSONRPC', side_effect=rpc): self.assertFalse(mirror.push('movie', 1)) state = store.get_state('movie', 1) self.assertEqual(state['playcount'], 0) self.assertIsNone(state['mirrored_at']) self.assertEqual(store.pending_mirrors(5, float('inf'))[0]['version'], state['state_version']) def test_adoption_skips_pending_mirrors_and_metadata_marks(self): ctx.mark('episode', 10, 1, 1) self.assertIsNone(mirror.check('episode', 10, 1, 1)) self.assertEqual(xbmc.RPC_CALLS, []) mirror.push('episode', 10, 1, 1) xbmc.RPC_CALLS.clear() jobs.enqueue('tv', 10, 'tv', None, None, False) self.assertIsNone(mirror.check('episode', 10, 1, 1)) self.assertEqual(xbmc.RPC_CALLS, []) def test_adoption_queues_mirror_and_rechecks_version_after_rpc(self): ctx.mark('movie', 1) mirror.push('movie', 1) xbmc.RPC_REPLIES['Files.GetFileDetails'] = {'filedetails': {'playcount': 0, 'resume': {}}} xbmc.RPC_CALLS.clear() self.assertEqual(mirror.check('movie', 1), 'adopted') self.assertEqual([r['method'] for r in xbmc.RPC_CALLS], ['Files.GetFileDetails']) self.assertTrue(store.mirror_pending('movie', 1)) mirror.push('movie', 1) xbmc.RPC_REPLIES['Files.GetFileDetails'] = {'filedetails': {'playcount': 2, 'resume': {}}} original = xbmc.executeJSONRPC def rpc(raw): ctx.mark('movie', 1) return original(raw) with patch.object(xbmc, 'executeJSONRPC', side_effect=rpc): self.assertIsNone(mirror.check('movie', 1)) self.assertEqual(store.get_state('movie', 1)['playcount'], 1) def test_newer_local_bookmark_is_remirrored_only_by_worker(self): store.save_resume('movie', 1, 600, 6000, when=2000) mirror.push('movie', 1) state = store.get_state('movie', 1) store.set_mirrored('movie', 1, 0, 600, when=1000, version=state['state_version']) xbmc.RPC_REPLIES['Files.GetFileDetails'] = {'filedetails': { 'playcount': 0, 'resume': {'position': 1500, 'total': 6000}}} xbmc.RPC_CALLS.clear() self.assertEqual(mirror.check('movie', 1), 'remirrored') self.assertEqual(store.get_state('movie', 1)['resume_pos'], 600) self.assertEqual([r['method'] for r in xbmc.RPC_CALLS], ['Files.GetFileDetails']) self.assertTrue(store.mirror_pending('movie', 1)) def test_context_refresh_only_actual_addon_videos(self): for active, path, refresh in [(False, urls.BASE, False), (True, 'smb://other', False), (True, urls.BASE + '?action=mlibrary', True)]: with self.subTest(active=active, path=path), \ patch.object(xbmc, 'getCondVisibility', return_value=active), \ patch.object(xbmc, 'getInfoLabel', return_value=path), \ patch.object(xbmc, 'executebuiltin') as builtin: ctx._done() self.assertEqual(builtin.call_count, int(refresh)) def test_tracker_snapshots_twelve_then_idles_until_event_or_fallback(self): clock = [100.0] engine = tracker.Engine(xbmc.Player(), xbmc.Monitor(), now=lambda: clock[0]) labels = {'Container.FolderPath': urls.BASE + '?action=mlibrary', 'Container.NumItems': '40', 'Container.ListItem.FileNameAndPath': urls.movie_url(99)} reads = [] def label(key): if 'ListItemAbsolute' in key: reads.append(key) return urls.movie_url(int(key.split('(')[1].split(')')[0]) + 1) return labels.get(key, '') with patch.object(xbmc, 'getCondVisibility', side_effect=lambda c: c == 'Window.IsActive(videos)'), \ patch.object(xbmc, 'getInfoLabel', side_effect=label), \ patch.object(jobs, 'request_adoption') as request: for _ in range(4): before = len(reads) engine.adopt_focused() self.assertLessEqual(len(reads) - before, 12) self.assertEqual(len(reads), 40) self.assertTrue(all(len(call.args[0]) <= 12 for call in request.call_args_list)) request.reset_mock() engine.adopt_focused() request.assert_not_called() clock[0] = 131 engine.adopt_focused() self.assertEqual(request.call_count, 1) engine.handle('adopt_changed', 131) engine.adopt_focused() self.assertEqual(request.call_count, 2) self.assertEqual(xbmc.RPC_CALLS, []) def test_tracker_tick_only_queues_durable_mirrors_and_row_events(self): player = xbmc.Player() clock = [100.0] engine = tracker.Engine(player, xbmc.Monitor(), now=lambda: clock[0]) engine.session = tracker.Session({'type': 'movie', 'tmdb': 5, 's': None, 'e': None, 'runtime_s': 6000, 'token': 't', 'start_at': 0}, store.log_start('movie', 5, None, None, 100), 100) with patch.object(player, 'getTime', return_value=300), \ patch.object(player, 'getTotalTime', return_value=6000, create=True), \ patch.object(engine, '_check_autoplay'), patch.object(xbmc, 'executebuiltin') as builtin: engine.tick() clock[0] += 4 engine.tick() self.assertEqual(store.get_state('movie', 5)['resume_pos'], 300) self.assertTrue(store.mirror_pending('movie', 5)) self.assertEqual(xbmc.RPC_CALLS, []) commands = [call.args[0] for call in builtin.call_args_list] self.assertTrue(any('skin_rows_changed' in command for command in commands)) self.assertTrue(all('skin_widget_refresh' not in command and 'Container.Refresh' not in command for command in commands)) def test_notification_callback_only_queues(self): events = queue.Queue() monitor = tracker.AdoptionMonitor(events) monitor.onNotification('kodi', 'VideoLibrary.OnUpdate', '{}') self.assertEqual(events.get_nowait()[0], 'adopt_changed') self.assertEqual(xbmc.RPC_CALLS, []) if __name__ == '__main__': unittest.main()