"""Atomic publication and rollback for the catalogue updater.""" import grp import hashlib from contextlib import closing, redirect_stdout import importlib.util import io import json import os from pathlib import Path import sqlite3 import stat import tempfile import unittest import test_catalog_api as fixtures SERVER = Path(__file__).resolve().parents[1] def runner_module(): spec = importlib.util.spec_from_file_location('catalog_sync_runner', SERVER / 'catalog_sync_runner.py') module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module class SyncRunnerTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.database = fixtures.fixture(self.root) self.state = self.root / 'state' self.stamp = '2026-11-03T19:00:00+00:00' self.api = runner_module() def tearDown(self): self.temp.cleanup() def candidate(self, source, stage, upstream, **kwargs): with closing(sqlite3.connect(source)) as src, closing(sqlite3.connect(stage)) as dest: src.backup(dest) dest.execute('UPDATE metadata SET value_json=? WHERE key=?', (json.dumps(self.stamp), 'built_at_utc')) dest.commit() return {'added_movies': 0, 'added_episodes': 0, 'built_at_utc': self.stamp} def digest(self): return hashlib.sha256(self.database.read_bytes()).hexdigest() def test_progress_normalizes_core_count_without_logging_other_fields(self): for event, completed in (({'count': 0}, 0), ({'count': 3}, 3), ({'completed': 4, 'count': 5}, 4)): with self.subTest(event=event): output = io.StringIO() with redirect_stdout(output): self.api.emit_progress({**event, 'phase': 'movies', 'total': 7, 'api_key': 'never-print-this'}) self.assertEqual(json.loads(output.getvalue()), {'progress': {'phase': 'movies', 'completed': completed, 'total': 7}}) def test_publish_is_atomic_and_records_success_after_health(self): before = self.digest() events = [] def restart(): events.append('restart') with closing(sqlite3.connect(self.database)) as db: stamp = json.loads(db.execute("SELECT value_json FROM metadata WHERE key='built_at_utc'").fetchone()[0]) self.assertEqual(stamp, self.stamp) def health(stamp): self.assertEqual(stamp, self.stamp) self.assertFalse((self.state / 'last-success.json').exists()) events.append('healthy') result = self.api.run_update(self.database, self.state, object(), synchronize_fn=self.candidate, restart_fn=restart, health_fn=health) self.assertEqual(events, ['restart', 'healthy']) self.assertNotEqual(self.digest(), before) backup = self.database.with_name('catalog.previous.sqlite') self.assertEqual(hashlib.sha256(backup.read_bytes()).hexdigest(), before) self.assertEqual(result['status'], 'success') self.assertEqual(json.loads((self.state / 'last-success.json').read_text())['status'], 'success') @unittest.skipUnless(os.geteuid() == 0, 'requires root to set group ownership') def test_publish_preserves_group_readable_database_ownership_and_mode(self): group = grp.getgrnam('www-data').gr_gid os.chown(self.database, 0, group) os.chmod(self.database, 0o640) observed = [] def restart(): published = self.database.stat() observed.append((published.st_uid, published.st_gid, stat.S_IMODE(published.st_mode))) self.api.run_update(self.database, self.state, object(), synchronize_fn=self.candidate, restart_fn=restart, health_fn=lambda _: None) self.assertEqual(observed, [(0, group, 0o640)]) published = self.database.stat() self.assertEqual((published.st_uid, published.st_gid), (0, group)) self.assertEqual(stat.S_IMODE(published.st_mode), 0o640) def test_fetch_failure_never_changes_live_database_or_restarts(self): before = self.digest() def fail(*_args, **_kwargs): raise RuntimeError('network unavailable') with self.assertRaises(self.api.UpdateError): self.api.run_update(self.database, self.state, object(), synchronize_fn=fail, restart_fn=lambda: self.fail('must not restart'), health_fn=lambda _: None) self.assertEqual(self.digest(), before) self.assertFalse((self.state / 'last-success.json').exists()) self.assertEqual(json.loads((self.state / 'last-result.json').read_text())['status'], 'failed') self.assertEqual(list(self.root.glob('.catalog-sync-*.sqlite*')), []) def test_failed_health_rolls_back_before_restarting_old_catalogue(self): before = self.digest() with closing(sqlite3.connect(self.database)) as db: old_stamp = json.loads(db.execute( "SELECT value_json FROM metadata WHERE key='built_at_utc'").fetchone()[0]) restarts = [] health_checks = [] def restart(): restarts.append(self.digest()) def fail_health(stamp): health_checks.append(stamp) if stamp == self.stamp: raise RuntimeError('new API not healthy') self.assertEqual(self.digest(), before) with self.assertRaises(self.api.UpdateError): self.api.run_update(self.database, self.state, object(), synchronize_fn=self.candidate, restart_fn=restart, health_fn=fail_health) self.assertEqual(len(restarts), 2) self.assertNotEqual(restarts[0], before) self.assertEqual(restarts[1], before) self.assertEqual(health_checks, [self.stamp, old_stamp]) self.assertEqual(self.digest(), before) self.assertFalse((self.state / 'last-success.json').exists()) self.assertFalse(json.loads((self.state / 'last-result.json').read_text())['rollback_failed']) def test_failed_rollback_health_requires_operator_attention(self): before = self.digest() with closing(sqlite3.connect(self.database)) as db: old_stamp = json.loads(db.execute( "SELECT value_json FROM metadata WHERE key='built_at_utc'").fetchone()[0]) health_checks = [] restarts = [] def fail_health(stamp): health_checks.append(stamp) raise RuntimeError('API not healthy') with self.assertRaisesRegex(self.api.UpdateError, 'operator attention'): self.api.run_update(self.database, self.state, object(), synchronize_fn=self.candidate, restart_fn=lambda: restarts.append(self.digest()), health_fn=fail_health) self.assertEqual(health_checks, [self.stamp, old_stamp]) self.assertEqual(len(restarts), 2) self.assertEqual(restarts[1], before) self.assertEqual(self.digest(), before) self.assertFalse((self.state / 'last-success.json').exists()) self.assertTrue(json.loads((self.state / 'last-result.json').read_text())['rollback_failed']) def test_concurrent_run_is_rejected_before_fetching(self): import fcntl self.state.mkdir() with (self.state / 'sync.lock').open('a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) with self.assertRaises(self.api.UpdateError): self.api.run_update(self.database, self.state, object(), synchronize_fn=lambda *_a, **_k: self.fail('overlapping fetch'), restart_fn=lambda: None, health_fn=lambda _: None) def assert_candidate_rejected(self, mutate): before = self.digest() restarts = [] def candidate(source, stage, upstream, **kwargs): report = self.candidate(source, stage, upstream, **kwargs) with closing(sqlite3.connect(stage)) as db: mutate(db) db.commit() return report with self.assertRaises(self.api.UpdateError): self.api.run_update(self.database, self.state, object(), synchronize_fn=candidate, restart_fn=lambda: restarts.append(True), health_fn=lambda _: None) self.assertEqual(restarts, []) self.assertEqual(self.digest(), before) self.assertFalse((self.state / 'last-success.json').exists()) def test_required_reference_and_source_tables_cannot_be_removed_or_emptied(self): for table in ('genres', 'countries', 'title_genres', 'title_countries', 'series_fetch_status', 'movie_sources', 'episode_sources'): for statement in ('DELETE FROM ', 'DROP TABLE '): with self.subTest(table=table, statement=statement): self.assert_candidate_rejected(lambda db: db.execute(statement + table)) def test_new_movie_sources_cannot_mask_existing_title_source_loss(self): def replace(db): db.execute("INSERT INTO titles(id,type,raw_json) VALUES (66,'movie','{}')") db.execute('UPDATE movie_sources SET title_id=66 WHERE title_id=1') self.assert_candidate_rejected(replace) def test_other_title_sources_cannot_mask_existing_title_source_loss(self): def replace(db): db.execute('UPDATE movie_sources SET title_id=3,source_index=1 WHERE title_id=1') self.assert_candidate_rejected(replace) def test_new_episode_sources_cannot_mask_existing_episode_source_loss(self): def replace(db): db.execute("INSERT INTO episodes(series_id,season_index,episode_index,id,raw_json) " "VALUES (2,0,1,502,'{}')") db.execute('UPDATE episode_sources SET episode_index=1') self.assert_candidate_rejected(replace) def test_null_id_episode_sources_are_preserved_by_index(self): with closing(sqlite3.connect(self.database)) as db: db.execute('UPDATE episodes SET id=NULL') db.commit() def replace(db): db.execute("INSERT INTO episodes(series_id,season_index,episode_index,raw_json) " "VALUES (2,0,1,'{}')") db.execute('UPDATE episode_sources SET episode_index=1') self.assert_candidate_rejected(replace) def test_source_url_replacement_with_stable_ids_can_be_published(self): replacement = 'https://media.test/replacement.mkv?token=new' def candidate(source, stage, upstream, **kwargs): report = self.candidate(source, stage, upstream, **kwargs) with closing(sqlite3.connect(stage)) as db: for table in ('movie_sources', 'episode_sources'): raw = json.loads(db.execute( 'SELECT raw_json FROM ' + table + ' WHERE source_id=?', (101 if table == 'movie_sources' else 201,)).fetchone()[0]) raw['url'] = replacement db.execute('UPDATE ' + table + ' SET url=?,raw_json=? WHERE source_id=?', (replacement, json.dumps(raw), raw['id'])) for table, title_id in (('titles', 1), ('episodes', 501)): raw = json.loads(db.execute('SELECT raw_json FROM ' + table + ' WHERE id=?', (title_id,)).fetchone()[0]) raw['sources'][0]['url'] = replacement db.execute('UPDATE ' + table + ' SET raw_json=? WHERE id=?', (json.dumps(raw), title_id)) season = json.loads(db.execute('SELECT raw_json FROM seasons').fetchone()[0]) season['episodes'][0]['sources'][0]['url'] = replacement db.execute('UPDATE seasons SET raw_json=?', (json.dumps(season),)) db.commit() return report result = self.api.run_update(self.database, self.state, object(), synchronize_fn=candidate, restart_fn=lambda: None, health_fn=lambda _: None) self.assertEqual(result['status'], 'success') with closing(sqlite3.connect(self.database)) as db: self.assertEqual(db.execute('SELECT source_id,url FROM movie_sources WHERE title_id=1').fetchone(), (101, replacement)) self.assertEqual(db.execute('SELECT source_id,url FROM episode_sources').fetchone(), (201, replacement)) def test_episode_sources_follow_stable_id_when_episode_index_changes(self): def candidate(source, stage, upstream, **kwargs): report = self.candidate(source, stage, upstream, **kwargs) with closing(sqlite3.connect(stage)) as db: db.execute('UPDATE episodes SET episode_index=1 WHERE id=501') db.execute('UPDATE episode_sources SET episode_index=1') db.commit() return report result = self.api.run_update(self.database, self.state, object(), synchronize_fn=candidate, restart_fn=lambda: None, health_fn=lambda _: None) self.assertEqual(result['status'], 'success') def test_incomplete_candidate_cannot_replace_the_catalogue(self): before = self.digest() def truncate(source, stage, upstream, **kwargs): result = self.candidate(source, stage, upstream, **kwargs) with closing(sqlite3.connect(stage)) as db: db.execute('DELETE FROM titles WHERE id=65') db.commit() return result with self.assertRaises(self.api.UpdateError): self.api.run_update(self.database, self.state, object(), synchronize_fn=truncate, restart_fn=lambda: self.fail('invalid candidate'), health_fn=lambda _: None) self.assertEqual(self.digest(), before)