# -*- coding: utf-8 -*- """L1 unit tests: thresholds, reconcile, tracker engine, schema v2 (docs/plan/milionar-library/040, 120 ยง2).""" import os import sqlite3 import sys import tempfile import unittest 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 jobs, mirror, policy, store, tracker # noqa: E402 def fresh_db(): store.set_path(os.path.join(tempfile.mkdtemp(prefix='milionar-db-'), 'library.db')) class ThresholdTests(unittest.TestCase): def test_short_below_threshold(self): self.assertEqual(policy.outcome(179, 6000, False), 'short') self.assertEqual(policy.outcome(0, 6000, False), 'short') def test_resume_from_threshold(self): self.assertEqual(policy.outcome(181, 6000, False), 'resume') self.assertEqual(policy.outcome(180, 6000, False), 'resume') def test_percent_boundary(self): self.assertEqual(policy.outcome(5399, 6000, False), 'resume') # 89.98 % self.assertEqual(policy.outcome(5400, 6000, False), 'watched') # 90 % self.assertEqual(policy.outcome(5800, 6000, False), 'watched') def test_ended_naturally_is_watched_past_threshold(self): self.assertEqual(policy.outcome(2000, 6000, True), 'watched') self.assertEqual(policy.outcome(100, 100, True), 'short') # a clip shorter than the threshold def test_custom_thresholds(self): self.assertEqual(policy.outcome(10, 100, False, min_seconds=10, watched_percent=95), 'resume') self.assertEqual(policy.outcome(95, 100, False, min_seconds=10, watched_percent=95), 'watched') def test_zero_total_uses_runtime(self): self.assertEqual(policy.effective_total(0, 7200), 7200.0) self.assertEqual(policy.effective_total(5000, 7200), 5000.0) total = policy.effective_total(0, 6000) self.assertEqual(policy.outcome(5500, total, False), 'watched') # 91.7 % of the runtime self.assertEqual(policy.outcome(5500, 0, False), 'resume') # unknown total: never watched by percent class ReconcileTests(unittest.TestCase): MIRRORED = {'playcount': 0, 'resume_pos': 600.0, 'at': 1000} def test_unchanged(self): self.assertIsNone(policy.reconcile({'last_played_at': 900}, {'playcount': 0, 'resume_pos': 600.5}, self.MIRRORED)) def test_kodi_mark_watched_is_adopted(self): self.assertEqual(policy.reconcile({'last_played_at': 900}, {'playcount': 1, 'resume_pos': 0}, self.MIRRORED), {'playcount': 1, 'resume_pos': 0.0}) def test_kodi_remove_resume_point_is_adopted(self): self.assertEqual(policy.reconcile({'last_played_at': 900}, {'playcount': 0, 'resume_pos': 0}, self.MIRRORED), {'playcount': 0, 'resume_pos': 0.0}) def test_kodi_stop_bookmark_older_than_ours_is_ignored(self): # a play ended at 1200 (after the mirror at 1000): Kodi's own bookmark must not win self.assertIsNone(policy.reconcile({'last_played_at': 1200}, {'playcount': 0, 'resume_pos': 1500}, self.MIRRORED)) def test_never_mirrored(self): self.assertIsNone(policy.reconcile({'last_played_at': 900}, {'playcount': 1, 'resume_pos': 0}, None)) self.assertIsNone(policy.reconcile({'last_played_at': 900}, {'playcount': 1, 'resume_pos': 0}, {'playcount': 0, 'resume_pos': 0, 'at': None})) def test_np_matching(self): np = {'stream_url': 'https://cdn/x.mkv', 'ts': 1000} self.assertTrue(policy.np_matches(np, 'https://cdn/x.mkv', 5000)) # exact match, any age self.assertTrue(policy.np_matches(np, 'https://cdn/other', 1010)) # fresh, not a plugin url self.assertFalse(policy.np_matches(np, 'https://cdn/other', 1100)) # stale self.assertFalse(policy.np_matches(np, 'plugin://plugin.video.milionar/?x', 1010)) self.assertFalse(policy.np_matches(None, 'https://cdn/x.mkv', 1000)) class SchemaV2Tests(unittest.TestCase): def test_migration_from_v1_adds_mirror_columns(self): path = os.path.join(tempfile.mkdtemp(prefix='milionar-db-'), 'library.db') conn = sqlite3.connect(path) for statement in store.SCHEMA_V1.strip().split(';'): if statement.strip(): conn.execute(statement) conn.execute('PRAGMA user_version=1') conn.execute('INSERT INTO movie_state(tmdb_id, playcount) VALUES(5, 2)') conn.commit() conn.close() store.set_path(path) conn = store.connect() self.assertEqual(conn.execute('PRAGMA user_version').fetchone()[0], store.SCHEMA_VERSION) columns = {r[1] for r in conn.execute('PRAGMA table_info(movie_state)')} self.assertIn('mirrored_playcount', columns) self.assertEqual(conn.execute('SELECT playcount FROM movie_state WHERE tmdb_id=5').fetchone()[0], 2) conn.close() class FakePlayer: def __init__(self): self.file, self.time, self.total, self.seeks = '', 0.0, 0.0, [] def getPlayingFile(self): return self.file def getTime(self): return self.time def getTotalTime(self): return self.total def seekTime(self, seconds): self.seeks.append(seconds) self.time = seconds class QuietEngine(tracker.Engine): def _spawn(self, target, *args): # no plan prefetch / dialogs here: autoplay has its own tests pass class EngineTests(unittest.TestCase): STREAM = 'https://cdn.example/film.mkv' def setUp(self): fresh_db() xbmc.RPC_CALLS.clear() xbmcgui.Window.props.clear() xbmcaddon.Addon.settings.clear() self.clock = [1000.0] self.player = FakePlayer() self.engine = QuietEngine(self.player, xbmc.Monitor(), now=lambda: self.clock[0]) def np(self, **extra): import json contract = {'token': 't', 'type': 'movie', 'tmdb': 77, 's': None, 'e': None, 'stream_url': self.STREAM, 'ident': 'abc', 'provider': 'webshare', 'file_name': 'Film.mkv', 'start_at': 0, 'ts': self.clock[0], 'runtime_s': 0} contract.update(extra) xbmcgui.Window.props['milionar.np'] = json.dumps(contract) def start(self, total=6000.0, **np_extra): self.np(**np_extra) self.player.file, self.player.total, self.player.time = self.STREAM, total, 0.0 self.engine.handle('started', self.clock[0]) def advance(self, seconds, position): self.clock[0] += seconds self.player.time = position self.engine.tick() def test_short_play_saves_nothing(self): self.start() self.advance(60, 120) self.engine.handle('stopped', self.clock[0]) self.assertIsNone(store.get_state('movie', 77)) self.assertEqual(store.recent_logs(1)[0]['outcome'], 'short') def test_stop_in_progress_saves_resume_and_source(self): self.start() self.advance(200, 200) self.engine.handle('stopped', self.clock[0]) state = store.get_state('movie', 77) self.assertEqual((state['resume_pos'], state['playcount'], state['last_ident']), (200.0, 0, 'abc')) self.assertEqual(store.recent_logs(1)[0]['outcome'], 'resume') def test_stop_at_91_percent_is_watched(self): self.start() self.advance(5000, 5460) self.engine.handle('stopped', self.clock[0]) state = store.get_state('movie', 77) self.assertEqual((state['playcount'], state['resume_pos']), (1, 0)) def test_natural_end_is_watched(self): self.start() self.advance(5990, 5990) # the player polls every 0.5 s: at the end it is at the end self.engine.handle('ended', self.clock[0]) self.assertEqual(store.get_state('movie', 77)['playcount'], 1) def test_a_broken_stream_is_not_watched(self): self.start() self.advance(3, 3) # the file fails 3 s in and Kodi reports 'ended' self.engine.handle('ended', self.clock[0]) self.assertIsNone(store.get_state('movie', 77)) self.assertEqual(store.recent_logs(1)[0]['outcome'], 'short') def test_autosave_every_10_seconds_and_mirror_after_3(self): self.start() self.advance(11, 300) self.assertEqual(store.get_state('movie', 77)['resume_pos'], 300.0) # saved by the 10 s autosave self.advance(1, 301) self.assertEqual(store.get_state('movie', 77)['resume_pos'], 300.0) # not saved again yet self.assertEqual([c for c in xbmc.RPC_CALLS if c['method'] == 'Files.SetFileDetails'], []) self.advance(3, 304) jobs.process_mirrors(now=10 ** 10) # the mirror RPC runs in the jobs thread, never in the tracker tick calls = [c for c in xbmc.RPC_CALLS if c['method'] == 'Files.SetFileDetails'] self.assertEqual(len(calls), 1) self.assertEqual(calls[0]['params']['file'], 'plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=77') self.assertEqual(calls[0]['params']['resume']['position'], 300.0) self.assertIsNotNone(store.get_state('movie', 77)['mirrored_at']) def test_playback_without_contract_is_ignored(self): # MyTV / MyCinema / library playback never set milionar.np (scenario 7) self.player.file, self.player.total = 'plugin://plugin.video.milionar/?action=play&ident=x', 3000.0 self.engine.handle('started', self.clock[0]) self.advance(500, 500) self.engine.handle('stopped', self.clock[0]) self.assertIsNone(store.get_state('movie', 77)) self.assertEqual(store.recent_logs(5), []) def test_zero_total_uses_contract_runtime(self): self.start(total=0.0, runtime_s=6000) self.advance(5500, 5500) self.engine.handle('stopped', self.clock[0]) self.assertEqual(store.get_state('movie', 77)['playcount'], 1) def test_episode_state_and_safety_seek(self): self.start(type='episode', tmdb=900001, s=1, e=3, start_at=1234.0) self.advance(4, 0.0) # Kodi did not honour the resume point self.assertEqual(self.player.seeks, [1234.0]) self.advance(1, 1300.0) self.advance(20, 1320.0) self.engine.handle('stopped', self.clock[0]) state = store.get_state('tv', 900001, 1, 3) self.assertEqual(state['resume_pos'], 1320.0) def test_safety_seek_not_needed_when_kodi_resumed(self): self.start(start_at=500.0) self.advance(4, 502.0) self.assertEqual(self.player.seeks, []) def test_crash_recovery_closes_open_log(self): self.start() self.advance(25, 250) # autosaved at 250, then "Kodi dies": no stop event self.assertEqual(store.close_open_logs(), 1) row = store.recent_logs(1)[0] self.assertEqual((row['outcome'], row['position']), ('crash', 250.0)) self.assertEqual(store.get_state('movie', 77)['resume_pos'], 250.0) class MirrorTests(unittest.TestCase): def setUp(self): fresh_db() xbmc.RPC_CALLS.clear() xbmc.RPC_REPLIES.clear() def kodi_says(self, playcount, position, total=6000.0): xbmc.RPC_REPLIES['Files.GetFileDetails'] = {'filedetails': { 'playcount': playcount, 'resume': {'position': position, 'total': total}}} def test_push_remembers_mirrored_values(self): store.save_resume('movie', 5, 600, 6000) self.assertTrue(mirror.push('movie', 5)) state = store.get_state('movie', 5) self.assertEqual((state['mirrored_playcount'], state['mirrored_resume_pos']), (0, 600.0)) def test_kodi_mark_watched_adopted_within_check(self): store.save_resume('movie', 5, 600, 6000, when=1000) mirror.push('movie', 5) store.set_mirrored('movie', 5, 0, 600.0, when=1500, version=store.get_state('movie', 5)['state_version']) self.kodi_says(1, 0) self.assertEqual(mirror.check('movie', 5), 'adopted') state = store.get_state('movie', 5) self.assertEqual((state['playcount'], state['resume_pos']), (1, 0.0)) def test_stop_time_bookmark_does_not_beat_newer_data(self): store.save_resume('movie', 5, 600, 6000, when=1000) store.set_mirrored('movie', 5, 0, 600.0, when=900, version=store.get_state('movie', 5)['state_version']) # mirror older than our last play self.kodi_says(0, 1500) # Kodi's own bookmark self.assertEqual(mirror.check('movie', 5), 'remirrored') self.assertEqual(store.get_state('movie', 5)['resume_pos'], 600.0) def test_unchanged_returns_none(self): store.save_resume('movie', 5, 600, 6000, when=1000) store.set_mirrored('movie', 5, 0, 600.0, when=1500, version=store.get_state('movie', 5)['state_version']) self.kodi_says(0, 600) self.assertIsNone(mirror.check('movie', 5)) def test_check_paths_counts_adopted_only_canonical(self): store.save_resume('movie', 5, 600, 6000, when=1000) store.set_mirrored('movie', 5, 0, 600.0, when=1500, version=store.get_state('movie', 5)['state_version']) self.kodi_says(1, 0) paths = ['plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=5', 'plugin://plugin.video.milionar/?action=play&ident=x', 'smb://nas/film.mkv'] self.assertEqual(mirror.check_paths(paths), 1) class PlayRequestTests(unittest.TestCase): CANON = 'plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=5' def setUp(self): xbmcgui.Window.props.clear() self.builtins = [] xbmc.executebuiltin = lambda cmd, wait=False: self.builtins.append(cmd) def test_request_sets_property_and_plays_canonical(self): from wcs.library2 import play play.request(self.CANON, src='auto', start='resume', pick={'ident': 'x', 'name': 'f.mkv', 'provider': 'kraska', 'junk': 1}) self.assertIn('PlayMedia({},resume)'.format(self.CANON), self.builtins) result = play.consume(self.CANON) self.assertEqual((result['src'], result['from']), ('auto', 'resume')) self.assertEqual(result['pick'], {'ident': 'x', 'name': 'f.mkv', 'provider': 'kraska'}) self.assertEqual(xbmcgui.Window().getProperty('milionar.play_req'), '') # consumed once def test_foreign_or_stale_request_is_ignored(self): import json import time from wcs.library2 import play xbmcgui.Window.props['milionar.play_req'] = json.dumps( {'canonical': 'plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=6', 'src': 'auto', 'from': 'resume', 'ts': time.time()}) self.assertEqual(play.consume(self.CANON), {'src': 'dialog', 'from': 'start', 'pick': None}) xbmcgui.Window.props['milionar.play_req'] = json.dumps( {'canonical': self.CANON, 'src': 'auto', 'from': 'resume', 'ts': time.time() - 60}) self.assertEqual(play.consume(self.CANON), {'src': 'dialog', 'from': 'start', 'pick': None}) class ResetTests(unittest.TestCase): def setUp(self): fresh_db() xbmc.RPC_CALLS.clear() xbmc.RPC_REPLIES.clear() folder = os.path.join(stubs.PROFILE, 'addon_data', 'plugin.video.milionar') for sub in ('', 'mytv_history', 'mycinema_history'): os.makedirs(os.path.join(folder, sub), exist_ok=True) self.folder = folder for name in ('global_watch_history.json', 'search_history.json', os.path.join('mytv_history', 'c1.json')): with open(os.path.join(folder, name), 'w') as handle: handle.write('[]') def test_store_reset_keeps_lists(self): store.replace_list('movie', [{'id': 1, 'title': 'a'}]) store.save_resume('movie', 1, 300, 6000) store.mark_watched('tv', 9, 1, 1) store.log_play('movie', 1, None, None, 1, 2, 3, 4, 'resume') self.assertEqual(store.reset_watch_state(), {'movies': 1, 'episodes': 1, 'plays': 1}) self.assertEqual(store.all_state_keys(), []) self.assertEqual([m['id'] for m in store.get_list('movie')], [1]) self.assertIsNone(store.get_title('movie', 1)['last_activity_at']) def test_reset_all(self): from wcs.library2 import reset store.save_resume('movie', 5, 300, 6000) store.mark_watched('tv', 9, 2, 3) xbmc.RPC_REPLIES['VideoLibrary.GetMovies'] = {'movies': [{'movieid': 11, 'playcount': 1}]} xbmc.RPC_REPLIES['VideoLibrary.GetEpisodes'] = {'episodes': [{'episodeid': 21, 'playcount': 2}, {'episodeid': 22, 'playcount': 0}]} counts = reset.reset_all() self.assertEqual((counts['movies'], counts['episodes'], counts['kodi_library'], counts['kodi_mirror']), (1, 1, 3, 2)) methods = [c['method'] for c in xbmc.RPC_CALLS] self.assertEqual(methods.count('Files.SetFileDetails'), 2) self.assertEqual(methods.count('VideoLibrary.SetMovieDetails'), 1) self.assertEqual(methods.count('VideoLibrary.SetEpisodeDetails'), 2) self.assertFalse(os.path.exists(os.path.join(self.folder, 'global_watch_history.json'))) self.assertFalse(os.path.exists(os.path.join(self.folder, 'mytv_history', 'c1.json'))) self.assertTrue(os.path.exists(os.path.join(self.folder, 'search_history.json'))) # not watch history self.assertEqual(store.all_state_keys(), []) if __name__ == '__main__': unittest.main()