# -*- coding: utf-8 -*- """L0 unit tests: store, meta, urls, legacy list wrappers (docs/plan/milionar-library/120 §2). Run: python3 -m unittest discover -s scripts/library_harness -p 'test_*.py' -v """ import json import os import subprocess import sys import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import stubs # noqa: E402 PROFILE = stubs.install() from wcs.library2 import meta, store, urls # noqa: E402 def fresh_db(): path = os.path.join(tempfile.mkdtemp(prefix='milionar-db-'), 'library.db') store.set_path(path) return path class UrlTests(unittest.TestCase): def test_canonical_strings_are_stable(self): self.assertEqual(urls.movie_url(603), 'plugin://plugin.video.milionar/?action=mplay&type=movie&tmdb=603') self.assertEqual(urls.episode_url(1396, 2, 5), 'plugin://plugin.video.milionar/?action=mplay&type=episode&tmdb=1396&s=2&e=5') def test_round_trip(self): for info in ({'type': 'movie', 'tmdb': 7}, {'type': 'episode', 'tmdb': 9, 's': 3, 'e': 12}): self.assertEqual(urls.parse(urls.canonical(info)), info) def test_foreign_urls_are_not_canonical(self): for url in ('plugin://plugin.video.milionar/?action=play&ident=x', 'plugin://other/?action=mplay&type=movie&tmdb=1', 'plugin://plugin.video.milionar/?action=mplay&type=movie', 'http://x/y', ''): self.assertIsNone(urls.parse(url), url) class StoreTests(unittest.TestCase): def setUp(self): fresh_db() def test_schema_created_and_migration_is_idempotent(self): conn = store.connect() self.assertEqual(conn.execute('PRAGMA user_version').fetchone()[0], store.SCHEMA_VERSION) tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} self.assertTrue({'titles', 'movie_state', 'episode_state', 'play_log', 'meta', 'flags'} <= tables) conn.close() store.connect().close() # second open must not fail or recreate def _write_legacy(self, movies=3, series=12): folder = stubs.PROFILE with open(os.path.join(folder, 'watchlist_movies.json'), 'w', encoding='utf-8') as handle: json.dump([{'id': 100 + i, 'title': 'Film %d' % i, 'year': '2000'} for i in range(movies)], handle) with open(os.path.join(folder, 'watchlist_series.json'), 'w', encoding='utf-8') as handle: json.dump([{'id': 200 + i, 'name': 'Serial %d' % i} for i in range(series)], handle) return folder def test_watchlist_import_3_and_12(self): folder = self._write_legacy() self.assertEqual(store.import_watchlists(folder), {'movie': 3, 'tv': 12}) self.assertEqual([m['id'] for m in store.get_list('movie')], [100, 101, 102]) self.assertEqual(len(store.get_list('tv')), 12) self.assertEqual(store.get_list('tv')[0]['name'], 'Serial 0') self.assertTrue(os.path.exists(os.path.join(folder, 'watchlist_movies.json.migrated'))) self.assertEqual(store.import_watchlists(folder), {}) # second call does nothing def test_replace_list_keeps_order_and_removes_missing(self): store.replace_list('movie', [{'id': 3, 'title': 'c'}, {'id': 1, 'title': 'a'}, {'id': 2, 'title': 'b'}]) self.assertEqual([m['id'] for m in store.get_list('movie')], [3, 1, 2]) store.replace_list('movie', [{'id': 2, 'title': 'b'}, {'id': 3, 'title': 'c'}]) self.assertEqual([m['id'] for m in store.get_list('movie')], [2, 3]) self.assertFalse(store.is_in_list('movie', 1)) store.replace_list('movie', []) self.assertEqual(store.get_list('movie'), []) def test_set_in_list_appends(self): store.set_in_list('tv', 5, True, {'id': 5, 'name': 'five'}) store.set_in_list('tv', 6, True, {'id': 6, 'name': 'six'}) self.assertEqual([s['id'] for s in store.get_list('tv')], [5, 6]) store.set_in_list('tv', 5, False) self.assertEqual([s['id'] for s in store.get_list('tv')], [6]) def test_lists_do_not_touch_watch_state(self): store.save_resume('movie', 1, 300, 6000) store.replace_list('movie', []) self.assertEqual(store.get_state('movie', 1)['resume_pos'], 300) def test_mark_rules_movie(self): store.save_resume('movie', 1, 300, 6000, source={'ident': 'abc', 'provider': 'webshare', 'file_name': 'f.mkv'}) state = store.get_state('movie', 1) self.assertEqual((state['resume_pos'], state['playcount'], state['last_ident']), (300, 0, 'abc')) store.mark_watched('movie', 1) state = store.get_state('movie', 1) self.assertEqual((state['playcount'], state['resume_pos'], state['resume_total']), (1, 0, 0)) self.assertEqual(state['last_ident'], 'abc') # source of the last play survives store.save_resume('movie', 1, 50, 6000) store.mark_unwatched('movie', 1) state = store.get_state('movie', 1) self.assertEqual((state['playcount'], state['resume_pos']), (0, 0)) def test_mark_rules_episode(self): store.save_resume('tv', 9, 120, 2500, season=1, episode=3) store.mark_watched('tv', 9, 1, 3) store.mark_watched('tv', 9, 1, 4) store.mark_unwatched('tv', 9, 1, 3) states = {(s['season'], s['episode']): s for s in store.get_episode_states(9)} self.assertEqual(states[(1, 3)]['playcount'], 0) self.assertEqual(states[(1, 4)]['playcount'], 1) def test_activity_and_hidden_from_continue(self): store.mark_watched('movie', 7, when=1000) self.assertEqual(store.get_title('movie', 7)['last_activity_at'], 1000) store.set_hidden_from_continue('movie', 7, True, when=2000) self.assertEqual(store.get_title('movie', 7)['hidden_from_continue_at'], 2000) store.set_hidden_from_continue('movie', 7, False) self.assertIsNone(store.get_title('movie', 7)['hidden_from_continue_at']) def test_transaction_rolls_back_on_error(self): with self.assertRaises(RuntimeError): with store.tx() as conn: conn.execute("INSERT INTO flags(key, value) VALUES('x', '1')") raise RuntimeError('boom') self.assertIsNone(store.get_flag('x')) def test_two_processes_write_concurrently(self): path = store.db_path() store.connect().close() code = ('import sys; sys.path.insert(0, %r); import stubs; stubs.install(); ' 'from wcs.library2 import store; store.set_path(%r); ' '[store.mark_watched("movie", int(sys.argv[1]) * 1000 + i) for i in range(60)]') % (HERE, path) procs = [subprocess.Popen([sys.executable, '-c', code, str(n)]) for n in (1, 2, 3)] self.assertEqual([p.wait(timeout=60) for p in procs], [0, 0, 0]) conn = store.connect() self.assertEqual(conn.execute('SELECT COUNT(*) FROM movie_state').fetchone()[0], 180) conn.close() class MetaTests(unittest.TestCase): def setUp(self): fresh_db() def test_put_get_and_ttl(self): meta.put(meta.key_movie(1), {'title': 'x'}, 100, now=1000) self.assertEqual(meta.get(meta.key_movie(1), now=1050), ({'title': 'x'}, False)) self.assertEqual(meta.get(meta.key_movie(1), now=1200), ({'title': 'x'}, True)) self.assertEqual(meta.get('movie:404'), (None, True)) self.assertIn('movie:1', meta.stale_keys(now=1200)) def test_ttl_by_show_status(self): self.assertEqual(meta.ttl_for_status('Returning Series'), 12 * 3600) self.assertEqual(meta.ttl_for_status('In Production'), 12 * 3600) self.assertEqual(meta.ttl_for_status('Ended'), 30 * 86400) def test_keys(self): self.assertEqual((meta.key_tv(5), meta.key_season(5, 2), meta.key_collection(8)), ('tv:5', 'season:5:2', 'collection:8')) def test_incomplete_entry_is_refetched(self): from wcs.library2 import tmdb meta.put(meta.key_tv(42), {'id': 42, 'name': 'partial list entry'}, 10 ** 9) # no 'seasons' calls = [] def fetch(path, params): calls.append(path) return {'id': 42, 'status': 'Ended', 'seasons': [{'season_number': 1}]} show = tmdb.show(42, fetch) self.assertEqual((calls, show['status']), (['tv/42'], 'Ended')) self.assertEqual(tmdb.show(42, fetch)['status'], 'Ended') # complete now: cached self.assertEqual(calls, ['tv/42']) def test_strict_listing_raises_instead_of_hiding_a_failed_download(self): from wcs.library2 import tmdb def broken(path, params): raise ValueError('TMDB API klíč není nastaven') self.assertIsNone(tmdb.show(77, broken)) # playback code: None with self.assertRaises(ValueError): tmdb.show(77, broken, strict=True) # listing: error dialog with self.assertRaises(RuntimeError): tmdb.season(77, 1, lambda path, params: {}, strict=True) # swallowed HTTP error def test_listings_and_autoplay_share_one_download(self): from wcs.library2 import tmdb calls = [] def fetch(path, params): calls.append(path) if '/season/' in path: return {'episodes': [{'episode_number': 1}]} return {'id': 5, 'status': 'Ended', 'seasons': [{'season_number': 1}]} tmdb.show(5, fetch, strict=True) # old listing opens the series tmdb.season(5, 1, fetch, strict=True) # ... and a season tmdb.show(5, fetch) # autoplay later tmdb.season(5, 1, fetch) self.assertEqual(calls, ['tv/5', 'tv/5/season/1']) def test_fixture_seed_relative_to_today(self): from datetime import date, timedelta from wcs.library2 import fixture_provider fixture_provider.seed_meta(relative_to_today=True) season = meta.get(meta.key_season(900001, 2))[0] last = season['episodes'][-1]['air_date'] self.assertEqual(last, (date.today() + timedelta(days=14)).isoformat()) # fixture: frozen day + 14 def test_fixture_seed(self): from wcs.library2 import fixture_provider self.assertEqual(fixture_provider.seed_meta(), 2 + 4 + 3 + 1) show, stale = meta.get(meta.key_tv(900001)) self.assertFalse(stale) self.assertEqual(len(meta.get(meta.key_season(900001, 2))[0]['episodes']), 8) class GenericCacheTests(unittest.TestCase): """wcs/metadata/TMDbCache.py is a facade over `meta` (no tmdb_cache.json any more).""" def setUp(self): fresh_db() from wcs.metadata import TMDbCache self.cache = TMDbCache def test_get_or_fetch_downloads_once(self): calls = [] def fetch(): calls.append(1) return {'url': 'x'} self.assertEqual(self.cache.get_or_fetch('/movie/1/images', {'kind': 'logo'}, fetch, 3600), {'url': 'x'}) self.assertEqual(self.cache.get_or_fetch('/movie/1/images', {'kind': 'logo'}, fetch, 3600), {'url': 'x'}) self.assertEqual(len(calls), 1) self.assertEqual(self.cache.get_or_fetch('/movie/1/images', {'kind': 'other'}, fetch, 3600), {'url': 'x'}) self.assertEqual(len(calls), 2) # params are part of the key def test_expiry_is_chosen_by_the_reader(self): self.cache.set_to_cache('/e', {}, {'a': 1}) self.assertEqual(self.cache.get_from_cache('/e', {}, expire_seconds=60), {'a': 1}) key = self.cache._make_key('/e', {}) meta.put(key, {'a': 1}, self.cache.KEEP_SECONDS, now=1000) # fetched long ago self.assertIsNone(self.cache.get_from_cache('/e', {}, expire_seconds=60)) self.assertEqual(self.cache.get_from_cache('/e', {}, expire_seconds=10 ** 10), {'a': 1}) def test_none_is_not_cached(self): self.assertIsNone(self.cache.get_or_fetch('/n', {}, lambda: None)) self.assertIsNone(self.cache.get_from_cache('/n', {})) def test_delete_and_delete_where_touch_only_the_generic_space(self): meta.put(meta.key_tv(1), {'id': 1, 'name': 'show', 'overview': '', 'vote_average': 1, 'poster_path': ''}, 10 ** 9) self.cache.set_to_cache('/csfd/top_movies/page_1', {}, [1]) self.cache.set_to_cache('/csfd/top_movies/page_2', {}, [2]) self.cache.set_to_cache('/tv/9', {'language': 'cs-CZ'}, {'id': 9, 'overview': 'o', 'vote_average': 7, 'poster_path': 'p', 'name': 'n'}) self.assertEqual(self.cache.delete('/csfd/top_movies/page_1', {}), 1) self.assertEqual(self.cache.delete('/csfd/top_movies/page_1', {}), 0) removed = self.cache.delete_where(lambda d: isinstance(d, dict) and 'overview' in d) self.assertEqual(removed, 1) # tv:1 is not generic self.assertIsNotNone(meta.get(meta.key_tv(1))[0]) self.assertEqual(self.cache.get_from_cache('/csfd/top_movies/page_2', {}), [2]) def test_first_write_purges_expired_generic_entries(self): old = self.cache._make_key('/old', {}) meta.put(old, {'x': 1}, 10, now=1000) meta.put('movie:5', {'title': 't'}, 10, now=1000) # stale but not generic self.cache._purged = False self.cache.set_to_cache('/new', {}, {'y': 1}) self.assertIsNone(meta.get_aged(old)) self.assertIsNotNone(meta.get_aged('movie:5')) class LegacyWrapperTests(unittest.TestCase): def setUp(self): fresh_db() for name in ('watchlist_movies.json.migrated', 'watchlist_series.json.migrated'): path = os.path.join(stubs.PROFILE, name) if os.path.exists(path): os.remove(path) with open(os.path.join(stubs.PROFILE, 'watchlist_movies.json'), 'w', encoding='utf-8') as handle: json.dump([{'id': 1, 'title': 'A'}, {'id': 2, 'title': 'B'}, {'id': 3, 'title': 'C'}], handle) with open(os.path.join(stubs.PROFILE, 'watchlist_series.json'), 'w', encoding='utf-8') as handle: json.dump([{'id': 10 + i, 'name': 'S%d' % i} for i in range(12)], handle) store.import_watchlists(stubs.PROFILE) # Explicit migration; wrappers never import automatically. from wcs import user_data self.user_data = user_data user_data._profile = stubs.PROFILE def test_old_functions_show_3_and_12(self): ud = self.user_data self.assertEqual(len(ud.load_tmdb_movies(None)), 3) self.assertEqual(len(ud.load_tmdb_series(None)), 12) def test_store_and_remove_by_index(self): ud = self.user_data movies = ud.load_tmdb_movies(None) movies.append({'id': 4, 'title': 'D'}) ud.store_tmdb_movies(None, movies) self.assertEqual([m['id'] for m in ud.load_tmdb_movies(None)], [1, 2, 3, 4]) self.assertTrue(ud.remove_tmdb_movie_by_index(None, 0)) self.assertFalse(ud.remove_tmdb_movie_by_index(None, 9)) self.assertEqual([m['id'] for m in ud.load_tmdb_movies(None)], [2, 3, 4]) ud.store_tmdb_series(None, []) self.assertEqual(ud.load_tmdb_series(None), []) if __name__ == '__main__': unittest.main()