272 lines
14 KiB
Python
272 lines
14 KiB
Python
"""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)
|