#!/usr/bin/env python3 """Offline schema-2 upgrade and profile preservation; no Kodi UI or providers.""" import json import os import sqlite3 import sys import tempfile import unittest from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import stubs stubs.install() import xbmcvfs from wcs.library2 import jobs, migration, store class UpgradeTests(unittest.TestCase): TABLES = ('titles', 'movie_state', 'episode_state', 'play_log', 'meta', 'flags') def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='milionar-upgrade-') self.addCleanup(self.temp.cleanup) self.addCleanup(store.set_path, None) self.profile = os.path.join(self.temp.name, 'userdata', 'addon_data', 'plugin.video.milionar') os.makedirs(self.profile) self.path = os.path.join(self.profile, 'library.db') store.set_path(self.path) self.files = {'settings.xml': b'true', 'watchlist_movies.json': json.dumps([{'id': 99, 'title': 'Legacy movie'}]).encode(), 'watchlist_series.json': json.dumps([{'id': 199, 'name': 'Legacy show'}]).encode(), 'search_history.json': b'{"universal":["keep"]}'} for name, data in self.files.items(): with open(os.path.join(self.profile, name), 'wb') as handle: handle.write(data) def seed_v2(self): conn = sqlite3.connect(self.path) for schema in (store.SCHEMA_V1, store.SCHEMA_V2): for statement in schema.split(';'): if statement.strip(): conn.execute(statement) conn.execute('PRAGMA user_version=2') conn.execute('INSERT INTO titles(media_type,tmdb_id,in_list,list_added_at,list_pos,legacy_json,last_activity_at) ' 'VALUES(?,?,?,?,?,?,?)', ('movie', 5, 1, 100, 0, '{"id":5,"title":"Saved movie"}', 200)) conn.execute('INSERT INTO titles(media_type,tmdb_id,in_list,list_added_at,list_pos,legacy_json,last_activity_at) ' 'VALUES(?,?,?,?,?,?,?)', ('tv', 10, 1, 110, 0, '{"id":10,"name":"Saved show"}', 210)) conn.execute('INSERT INTO movie_state(tmdb_id,playcount,last_played_at,resume_pos,resume_total,last_ident,' 'last_provider,last_file_name,mirrored_at,mirrored_playcount,mirrored_resume_pos) ' 'VALUES(5,2,200,300,6000,\'original\',\'webshare\',\'original.mkv\',220,2,300)') conn.execute('INSERT INTO episode_state(show_id,season,episode,playcount,last_played_at,resume_pos,resume_total,' 'mirrored_at,mirrored_playcount,mirrored_resume_pos) VALUES(10,1,2,0,210,120,2400,230,0,120)') conn.execute("INSERT INTO play_log(media_type,tmdb_id,started_at,ended_at,outcome) VALUES('movie',5,150,200,'resume')") conn.execute('INSERT INTO meta(key,json,fetched_at,ttl) VALUES(?,?,?,?)', ('movie:5', '{"id":5,"title":"Cached movie","genres":[]}', 100, 86400)) conn.execute('INSERT INTO flags(key,value) VALUES(?,?)', ('widget.rev.cont_movies', '7')) conn.execute('INSERT INTO flags(key,value) VALUES(?,?)', ('widget.rev.cont_series', '11')) conn.commit() conn.close() return self.snapshot() def snapshot(self): conn = sqlite3.connect(self.path) conn.row_factory = sqlite3.Row try: return {table: [dict(row) for row in conn.execute('SELECT * FROM {} ORDER BY 1,2'.format(table))] for table in self.TABLES} finally: conn.close() def assert_profile_preserved(self): for name, original in self.files.items(): with open(os.path.join(self.profile, name), 'rb') as handle: self.assertEqual(handle.read(), original, name) def test_v2_upgrade_preserves_membership_watch_metadata_and_revisions(self): original = self.seed_v2() conn = store.connect() self.assertEqual(conn.execute('PRAGMA user_version').fetchone()[0], 3) self.assertEqual(conn.execute('SELECT COUNT(*) FROM mirror_pending').fetchone()[0], 0) self.assertEqual(conn.execute('SELECT COUNT(*) FROM pending_marks').fetchone()[0], 0) conn.close() migrated = self.snapshot() for table in ('movie_state', 'episode_state'): self.assertEqual(migrated[table][0].pop('state_version'), 0) self.assertEqual(migrated, original) self.assert_profile_preserved() first = self.snapshot() store.connect().close() self.assertEqual(self.snapshot(), first) def test_failed_migration_rolls_back_columns_tables_and_version(self): original = self.seed_v2() migration_conn = sqlite3.connect(self.path, isolation_level=None) try: migration_conn.execute('PRAGMA journal_mode=WAL') with patch.object(store, 'SCHEMA_V3', store.SCHEMA_V3 + 'INVALID SQL;'): with self.assertRaises(sqlite3.OperationalError): store._migrate(migration_conn) finally: migration_conn.close() conn = sqlite3.connect(self.path) self.assertEqual(conn.execute('PRAGMA user_version').fetchone()[0], 2) self.assertNotIn('state_version', {row[1] for row in conn.execute('PRAGMA table_info(movie_state)')}) self.assertIsNone(conn.execute("SELECT name FROM sqlite_master WHERE name='mirror_pending'").fetchone()) conn.close() self.assertEqual(self.snapshot(), original) self.assert_profile_preserved() def test_new_mirror_and_manual_jobs_survive_reopening_upgraded_database(self): self.seed_v2() store.save_resume('movie', 5, 500, 6000, when=1000) pending = store.pending_mirrors(5, float('inf'))[0] store.retry_mirror(pending) jobs.enqueue('tv', 10, 'episode', 1, 3, True) mirrors = store.pending_mirrors(5, float('inf')) marks = jobs.pending_marks(now=float('inf')) state = store.get_state('movie', 5) store.set_path(self.path) store.connect().close() self.assertEqual(store.pending_mirrors(5, float('inf')), mirrors) self.assertEqual(jobs.pending_marks(now=float('inf')), marks) self.assertEqual(state['state_version'], 1) self.assertEqual(mirrors[0]['version'], state['state_version']) self.assertEqual(mirrors[0]['attempts'], 1) self.assertEqual(store.get_state('movie', 5)['last_ident'], 'original') self.assertEqual([row['id'] for row in store.get_list('movie')], [5]) self.assert_profile_preserved() def test_populated_new_lists_gate_legacy_files_before_any_read_or_import(self): original = self.seed_v2() with patch.object(store, 'inspect_watchlist_import', side_effect=AssertionError('must not read legacy')): self.assertIsNone(migration.offer(self.profile)) self.assertFalse(migration.pending(self.profile)) current = self.snapshot() for table in ('movie_state', 'episode_state'): current[table][0].pop('state_version') self.assertEqual(current, original) self.assert_profile_preserved() def test_empty_new_database_discovery_does_not_import_or_rename_legacy(self): self.assertTrue(migration.pending(self.profile)) self.assertEqual(store.get_list('movie'), []) self.assertEqual(store.get_list('tv'), []) self.assertIsNone(store.get_flag('watchlists.import.v2')) self.assert_profile_preserved() self.assertEqual(sorted(os.listdir(self.profile)), sorted(list(self.files) + ['library.db'])) def test_profile_path_is_stable_and_separate_from_addon_code(self): store.set_path(None) expected = os.path.join(xbmcvfs.translatePath('special://profile/addon_data/plugin.video.milionar/'), 'library.db') self.assertEqual(store.db_path(), expected) self.assertFalse(store.db_path().startswith(stubs.ADDON_DIR + os.sep)) if __name__ == '__main__': unittest.main()