464 lines
27 KiB
Python
464 lines
27 KiB
Python
"""Synchronizer tests use the exported schema, never the upstream service."""
|
|
import copy
|
|
from contextlib import closing
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import ssl
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.error import HTTPError, URLError
|
|
|
|
from test_catalog_api import SERVER, fixture, load_service
|
|
|
|
|
|
def load_sync():
|
|
spec = importlib.util.spec_from_file_location('catalog_sync', SERVER / 'catalog_sync.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
with patch.object(sys, 'path', [str(SERVER)] + sys.path):
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
status = 200
|
|
headers = {}
|
|
|
|
def __init__(self, payload=b'[]', url=None, status=200):
|
|
super().__init__(payload)
|
|
self.url, self.status = url, status
|
|
|
|
def geturl(self):
|
|
return self.url
|
|
|
|
|
|
class QueueOpener:
|
|
def __init__(self, *responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def open(self, request, timeout):
|
|
self.calls.append((request.full_url, request.get_method(), timeout))
|
|
response = self.responses.pop(0)
|
|
if isinstance(response, Exception):
|
|
raise response
|
|
if response.url is None:
|
|
response.url = request.full_url
|
|
return response
|
|
|
|
|
|
class UpstreamTests(unittest.TestCase):
|
|
def test_exact_fixed_origin_metadata_urls_and_quoted_key(self):
|
|
sync = load_sync()
|
|
opener = QueueOpener(*(Response(b'{}' if p.startswith('movie/by/3') else b'[]') for p in
|
|
['genre/all', 'country/all', 'movie/by/35721',
|
|
'movie/by/filtres/0/created/0', 'serie/by/filtres/0/created/12',
|
|
'season/by/serie/35716']))
|
|
client = sync.Upstream('test secret/+#?', interval=0, opener=opener)
|
|
paths = ['genre/all', 'country/all', 'movie/by/35721', 'movie/by/filtres/0/created/0',
|
|
'serie/by/filtres/0/created/12', 'season/by/serie/35716']
|
|
for path in paths:
|
|
client.get(path)
|
|
self.assertEqual(opener.calls, [('https://ninwinios.info/api/' + p +
|
|
'/test%20secret%2F%2B%23%3F/', 'GET', 30) for p in paths])
|
|
self.assertEqual(client.requests_count, 6)
|
|
for bad in ['https://evil.test/api/genre/all', '//evil.test/a', '/genre/all',
|
|
'genre/all/', 'genre/../all', 'genre/%2e%2e/all', 'genre/all?url=x',
|
|
'genre/all#x', 'movie/by/1/surprise', 'download/movie.mkv',
|
|
'movie/by/1\n', 'movie/by/0', 'movie/by/-1', 'movie/by/filtres/0/views/0']:
|
|
with self.subTest(path=repr(bad)), self.assertRaises(ValueError):
|
|
client.get(bad)
|
|
self.assertEqual(client.requests_count, 6)
|
|
|
|
def test_real_upstream_legacy_html_label_still_requires_valid_json(self):
|
|
sync = load_sync()
|
|
response = Response(b'[{"id":1,"title":"Action"}]')
|
|
response.headers = {'Content-Type': 'text/html; charset=UTF-8'}
|
|
client = sync.Upstream('k', interval=0, opener=QueueOpener(response))
|
|
self.assertEqual(client.get('genre/all'), [{'id': 1, 'title': 'Action'}])
|
|
html = Response(b'<html>Service unavailable</html>')
|
|
html.headers = {'Content-Type': 'text/html; charset=UTF-8'}
|
|
opener = QueueOpener(html)
|
|
with self.assertRaises(sync.SyncError):
|
|
sync.Upstream('k', interval=0, opener=opener).get('genre/all')
|
|
self.assertEqual(len(opener.calls), 1)
|
|
|
|
def test_wrong_content_type_and_truncated_length_are_not_retried(self):
|
|
sync = load_sync()
|
|
for headers in ({'Content-Type': 'video/mp4'}, {'Content-Type': 'image/jpeg'},
|
|
{'Content-Length': '999'}, {'Content-Length': 'bad'},
|
|
{'Content-Length': str(32 * 1024 * 1024 + 1)}):
|
|
response = Response(b'[]')
|
|
response.headers = headers
|
|
opener = QueueOpener(response)
|
|
with self.subTest(headers=headers), self.assertRaises(sync.SyncError):
|
|
sync.Upstream('k', interval=0, opener=opener).get('genre/all')
|
|
self.assertEqual(len(opener.calls), 1)
|
|
|
|
def test_bounded_retry_and_rate_limit(self):
|
|
sync = load_sync()
|
|
for status in (429, 503):
|
|
error = HTTPError('https://secret.test/key', status, 'private key', {}, None)
|
|
opener = QueueOpener(error, Response(b'[]'))
|
|
client = sync.Upstream('private-key', interval=.25, opener=opener)
|
|
with patch.object(sync.time, 'sleep') as sleep:
|
|
self.assertEqual(client.get('genre/all'), [])
|
|
self.assertGreaterEqual(sleep.call_count, 1)
|
|
self.assertEqual(client.requests_count, 2)
|
|
opener = QueueOpener(URLError('private URL'), Response(b'[]'))
|
|
with patch.object(sync.time, 'sleep'):
|
|
self.assertEqual(sync.Upstream('k', interval=0, opener=opener).get('genre/all'), [])
|
|
opener = QueueOpener(Response(), Response())
|
|
client = sync.Upstream('k', interval=.25, opener=opener)
|
|
with patch.object(sync.time, 'monotonic', return_value=100), patch.object(sync.time, 'sleep') as sleep:
|
|
client.get('genre/all')
|
|
client.get('country/all')
|
|
sleep.assert_called_once_with(.25)
|
|
|
|
def test_auth_cert_redirect_json_and_size_fail_closed_without_secrets(self):
|
|
sync = load_sync()
|
|
for failure in [HTTPError('https://secret.test/private-key', code, 'private-key', {}, None)
|
|
for code in (301, 302, 307, 308, 401, 403, 404)] + [
|
|
URLError(ssl.SSLCertVerificationError('private-key')),
|
|
ssl.SSLCertVerificationError('private-key')]:
|
|
opener = QueueOpener(failure)
|
|
client = sync.Upstream('private-key', interval=0, opener=opener)
|
|
with self.assertRaises(sync.SyncError) as error:
|
|
client.get('genre/all')
|
|
self.assertNotIn('private-key', str(error.exception))
|
|
self.assertNotIn('https:', str(error.exception))
|
|
self.assertEqual(len(opener.calls), 1)
|
|
for response in [Response(b'not JSON'), Response(b'NaN'), Response(b'{}'),
|
|
Response(b'null'), Response(b'[]', url='https://evil.test/video.mkv'),
|
|
Response(b'[]', status=302), Response(b'[' * (32 * 1024 * 1024 + 1))]:
|
|
opener = QueueOpener(response)
|
|
with self.assertRaises(sync.SyncError):
|
|
sync.Upstream('private-key', interval=0, opener=opener).get('genre/all')
|
|
self.assertEqual(len(opener.calls), 1)
|
|
client = sync.Upstream('k', interval=0)
|
|
from urllib.request import HTTPRedirectHandler
|
|
handlers = [h for h in client._opener.handlers if isinstance(h, HTTPRedirectHandler)]
|
|
self.assertEqual(len(handlers), 1)
|
|
with self.assertRaises(sync.SyncError):
|
|
handlers[0].redirect_request(None, None, 302, 'moved', {}, 'https://evil.test/video')
|
|
|
|
def test_retry_exhaustion_and_configuration_are_bounded(self):
|
|
sync = load_sync()
|
|
opener = QueueOpener(*(HTTPError('https://secret/key', 503, 'key', {}, None) for _ in range(3)))
|
|
with patch.object(sync.time, 'sleep'), self.assertRaises(sync.SyncError):
|
|
sync.Upstream('k', interval=0, opener=opener).get('genre/all')
|
|
self.assertEqual(len(opener.calls), 3)
|
|
for kwargs in ({'attempts': 0}, {'attempts': 11}, {'interval': -1}, {'timeout': 0},
|
|
{'interval': float('nan')}, {'timeout': float('inf')}):
|
|
with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
|
|
sync.Upstream('k', **kwargs)
|
|
|
|
|
|
class FakeUpstream:
|
|
def __init__(self, routes=None):
|
|
self.routes = {'genre/all': [], 'country/all': [],
|
|
'movie/by/filtres/0/created/0': [],
|
|
'movie/by/filtres/0/created/1': [],
|
|
'serie/by/filtres/0/created/0': [],
|
|
'serie/by/filtres/0/created/1': [], 'season/by/serie/2': []}
|
|
self.routes.update(routes or {})
|
|
self.calls = []
|
|
self.requests_count = 0
|
|
|
|
def get(self, path):
|
|
self.calls.append(path)
|
|
self.requests_count += 1
|
|
if path not in self.routes:
|
|
raise AssertionError('Unexpected metadata path: ' + path)
|
|
value = self.routes[path]
|
|
if isinstance(value, Exception):
|
|
raise value
|
|
return copy.deepcopy(value)
|
|
|
|
|
|
class SynchronizeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR'))
|
|
self.source = fixture(self.temp.name)
|
|
self.stage = Path(self.temp.name) / 'stage.sqlite'
|
|
self.sync = load_sync()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def rows(self, path, sql, args=()):
|
|
with closing(sqlite3.connect(path)) as db:
|
|
return db.execute(sql, args).fetchall()
|
|
|
|
def metadata(self, path):
|
|
return {key: json.loads(value) for key, value in self.rows(path, 'SELECT * FROM metadata')}
|
|
|
|
def test_all_known_visible_series_refresh_and_series_samples_are_bounded(self):
|
|
with closing(sqlite3.connect(self.source)) as db:
|
|
for title_id, title, description in [(90, 'An older series', 'Normal description'),
|
|
(91, 'Telegram @channelname', ''),
|
|
(92, 'Disguised advertisement', 'ملودی باکس')]:
|
|
item = {'id': title_id, 'type': 'serie', 'title': title, 'description': description}
|
|
db.execute('INSERT INTO titles(id,type,title,description,raw_json) VALUES (?,?,?,?,?)',
|
|
(title_id, 'serie', title, description, json.dumps(item)))
|
|
db.commit()
|
|
upstream = FakeUpstream({'season/by/serie/90': [{'id': 900, 'title': 'Season 1', 'episodes': [
|
|
{'id': 901, 'title': 'New episode on an old series'}]}]})
|
|
report = self.sync.synchronize(self.source, self.stage, upstream)
|
|
self.assertEqual(report['added']['episodes'], 1)
|
|
self.assertIn('season/by/serie/2', upstream.calls)
|
|
self.assertIn('season/by/serie/90', upstream.calls)
|
|
self.assertNotIn('season/by/serie/91', upstream.calls)
|
|
self.assertNotIn('season/by/serie/92', upstream.calls)
|
|
self.assertEqual(report['counts']['series'], 4)
|
|
self.assertEqual(self.rows(self.stage, 'SELECT COUNT(*) FROM titles WHERE id IN (91,92)'), [(2,)])
|
|
self.stage.unlink()
|
|
records = [{'id': i, 'type': 'serie', 'title': 'New Series ' + str(i)} for i in range(100, 201)]
|
|
routes = {'serie/by/filtres/0/created/0': records, 'season/by/serie/90': []}
|
|
routes.update({f'season/by/serie/{item["id"]}': [] for item in records})
|
|
report = self.sync.synchronize(self.source, self.stage, FakeUpstream(routes))
|
|
self.assertEqual(report['updated_series_count'], 101)
|
|
self.assertEqual(len(report['updated_series_ids']), 100)
|
|
self.assertEqual(report['updated_series_ids'], list(range(100, 200)))
|
|
|
|
def test_progress_failure_discards_candidate_and_missing_source_is_not_created(self):
|
|
original = self.source.read_bytes()
|
|
def reject(_event):
|
|
raise RuntimeError('private-key')
|
|
with self.assertRaises(self.sync.SyncError) as error:
|
|
self.sync.synchronize(self.source, self.stage, FakeUpstream(), progress=reject)
|
|
self.assertNotIn('private-key', str(error.exception))
|
|
self.assertEqual(self.source.read_bytes(), original)
|
|
self.assertFalse(self.stage.exists())
|
|
missing = Path(self.temp.name) / 'missing.sqlite'
|
|
with self.assertRaises(self.sync.SyncError):
|
|
self.sync.synchronize(missing, self.stage, FakeUpstream())
|
|
self.assertFalse(missing.exists())
|
|
self.assertFalse(self.stage.exists())
|
|
|
|
def test_malformed_metadata_discards_candidate_not_source(self):
|
|
original = self.source.read_bytes()
|
|
cases = [
|
|
{'genre/all': {'error': 'not a list'}},
|
|
{'country/all': [{'error': 'bad metadata'}]},
|
|
{'movie/by/filtres/0/created/0': {'error': 'failed'}},
|
|
{'serie/by/filtres/0/created/0': [None]},
|
|
{'serie/by/filtres/0/created/0': [{'id': 70, 'type': 'serie'}]},
|
|
{'serie/by/filtres/0/created/0': [{'id': 70, 'type': 'serie', 'title': {'bad': 'title'}}]},
|
|
{'movie/by/filtres/0/created/0': [{'id': True, 'type': 'movie', 'title': 'Bad'}]},
|
|
{'movie/by/filtres/0/created/0': [{'id': 1, 'type': 'movie', 'title': 'Alpha', 'sources': None}]},
|
|
{'season/by/serie/2': {'error': 'not seasons'}},
|
|
{'season/by/serie/2': [{'error': 'not a season'}]},
|
|
{'season/by/serie/2': [{'id': 1001, 'episodes': {'error': 'not episodes'}}]},
|
|
{'season/by/serie/2': [{'id': 1001, 'episodes': [{'error': 'not an episode'}]}]},
|
|
{'season/by/serie/2': [{'id': 1001, 'episodes': [{'id': 501, 'sources': [{}]}]}]},
|
|
{'season/by/serie/2': [{'id': 1001, 'episodes': [{'id': 501, 'sources': [{'id': 201, 'url': []}]}]}]},
|
|
{'season/by/serie/2': [{'id': 1001, 'episodes': [{'id': 501, 'sources': 'not sources'}]}]},
|
|
{'season/by/serie/2': [{'id': 1001, 'episodes': [{'id': 501}, {'id': 501}]}]},
|
|
{'season/by/serie/2': ValueError('https://private.test/private-key')},
|
|
]
|
|
for routes in cases:
|
|
with self.subTest(routes=routes):
|
|
with self.assertRaises(self.sync.SyncError) as error:
|
|
self.sync.synchronize(self.source, self.stage, FakeUpstream(routes))
|
|
self.assertNotIn('private-key', str(error.exception))
|
|
self.assertFalse(self.stage.exists())
|
|
self.assertEqual(self.source.read_bytes(), original)
|
|
|
|
def test_429_503_exhaustion_rolls_back_staged_writes(self):
|
|
original = self.source.read_bytes()
|
|
for status in (429, 503):
|
|
opener = QueueOpener(Response(b'[{"id":99,"title":"A new genre"}]'), Response(),
|
|
*(HTTPError('https://private.test/private-key', status, 'private-key', {}, None)
|
|
for _ in range(3)))
|
|
upstream = self.sync.Upstream('private-key', interval=0, opener=opener)
|
|
with patch.object(self.sync.time, 'sleep'), self.assertRaises(self.sync.SyncError):
|
|
self.sync.synchronize(self.source, self.stage, upstream)
|
|
self.assertEqual(len(opener.calls), 5)
|
|
self.assertFalse(self.stage.exists())
|
|
self.assertEqual(self.source.read_bytes(), original)
|
|
|
|
def test_page_limit_and_existing_candidates_are_never_published_or_overwritten(self):
|
|
original = self.source.read_bytes()
|
|
routes = {'movie/by/filtres/0/created/0': [{'id': 1, 'type': 'movie', 'title': 'Updated'}]}
|
|
with self.assertRaises(self.sync.SyncError):
|
|
self.sync.synchronize(self.source, self.stage, FakeUpstream(routes), max_pages=1)
|
|
self.assertFalse(self.stage.exists())
|
|
self.assertEqual(self.source.read_bytes(), original)
|
|
self.stage.write_bytes(b'existing candidate')
|
|
with self.assertRaises(FileExistsError):
|
|
self.sync.synchronize(self.source, self.stage, FakeUpstream())
|
|
self.assertEqual(self.stage.read_bytes(), b'existing candidate')
|
|
with self.assertRaises(ValueError):
|
|
self.sync.synchronize(self.source, self.source, FakeUpstream())
|
|
self.assertEqual(self.source.read_bytes(), original)
|
|
|
|
def test_existing_id_does_not_end_pagination_and_missing_ids_use_indices(self):
|
|
routes = {'movie/by/filtres/0/created/0': [{'id': 1, 'type': 'movie', 'title': 'Alpha'}],
|
|
'movie/by/filtres/0/created/1': [{'id': 70, 'type': 'movie', 'title': 'Next page'}],
|
|
'movie/by/filtres/0/created/2': [],
|
|
'movie/by/70': {'id': 70, 'type': 'movie', 'title': 'Next page', 'sources': []},
|
|
'season/by/serie/2': [{'title': 'Season 1 refreshed', 'episodes': [
|
|
{'title': 'Episode 1 refreshed', 'sources': []}, {'title': 'Episode without ID', 'sources': []}]}]}
|
|
report = self.sync.synchronize(self.source, self.stage, FakeUpstream(routes))
|
|
self.assertEqual(report['added']['movies'], 1)
|
|
self.assertEqual(report['added']['episodes'], 1)
|
|
self.assertEqual(self.rows(self.stage, 'SELECT id,title FROM seasons'), [(1001, 'Season 1 refreshed')])
|
|
self.assertEqual(self.rows(self.stage, 'SELECT id,title FROM episodes ORDER BY episode_index'),
|
|
[(501, 'Episode 1 refreshed'), (None, 'Episode without ID')])
|
|
self.assertEqual(self.rows(self.stage, 'SELECT source_id FROM episode_sources'), [(201,)])
|
|
|
|
def test_advertisements_channels_and_detail_ads_are_not_imported(self):
|
|
items = [
|
|
{'id': 70, 'type': 'movie', 'title': 'Telegram @channelname'},
|
|
{'id': 71, 'type': 'movie', 'title': 'Innocent', 'description': 'نسخه جدید ملودی باکس'},
|
|
{'id': 72, 'type': 'channel', 'title': 'Live channel'},
|
|
{'id': 73, 'type': 'movie', 'title': 'Looks like a movie'}]
|
|
upstream = FakeUpstream({'movie/by/filtres/0/created/0': items,
|
|
'movie/by/filtres/0/created/1': [],
|
|
'movie/by/73': {'id': 73, 'type': 'movie', 'title': 'Telegram @channelname'}})
|
|
report = self.sync.synchronize(self.source, self.stage, upstream)
|
|
self.assertEqual(report['added']['titles'], 0)
|
|
self.assertNotIn('movie/by/70', upstream.calls)
|
|
self.assertNotIn('movie/by/71', upstream.calls)
|
|
self.assertNotIn('movie/by/72', upstream.calls)
|
|
self.assertEqual(self.rows(self.stage, 'SELECT COUNT(*) FROM titles'), [(65,)])
|
|
|
|
def test_reordered_stable_ids_and_truncated_nested_lists_retain_every_old_entry(self):
|
|
first = {'season/by/serie/2': [
|
|
{'id': 1001, 'episodes': [{'id': 502, 'title': 'Episode 2', 'sources': [
|
|
{'id': 202, 'url': 'https://video.test/old2'}]}]},
|
|
{'id': 1002, 'title': 'Season 2', 'episodes': [{'id': 601, 'title': 'S2 E1'}]}]}
|
|
self.sync.synchronize(self.source, self.stage, FakeUpstream(first))
|
|
second = Path(self.temp.name) / 'second.sqlite'
|
|
routes = {'movie/by/filtres/0/created/0': [{'id': 1, 'type': 'movie', 'title': 'Updated Alpha',
|
|
'sources': [], 'genres': []}],
|
|
'movie/by/filtres/0/created/1': [],
|
|
'season/by/serie/2': [
|
|
{'id': 1002, 'title': 'Season Two', 'episodes': []},
|
|
{'id': 1001, 'episodes': [{'id': 502, 'title': 'Updated Episode 2', 'sources': [
|
|
{'id': 202, 'url': 'https://video.test/refreshed'}]},
|
|
{'id': 501, 'sources': []}]}]}
|
|
report = self.sync.synchronize(self.stage, second, FakeUpstream(routes))
|
|
self.assertEqual(report['added'], dict.fromkeys(report['added'], 0))
|
|
self.assertEqual(self.rows(second, 'SELECT season_index,id FROM seasons ORDER BY season_index'),
|
|
[(0, 1001), (1, 1002)])
|
|
self.assertEqual(self.rows(second, 'SELECT season_index,episode_index,id FROM episodes ORDER BY 1,2'),
|
|
[(0, 0, 501), (0, 1, 502), (1, 0, 601)])
|
|
self.assertEqual(self.rows(second, 'SELECT url FROM episode_sources WHERE source_id=202'),
|
|
[('https://video.test/refreshed',)])
|
|
for table in ('seasons', 'episodes', 'movie_sources', 'episode_sources'):
|
|
self.assertGreaterEqual(self.rows(second, 'SELECT COUNT(*) FROM ' + table)[0][0],
|
|
self.rows(self.stage, 'SELECT COUNT(*) FROM ' + table)[0][0])
|
|
third = Path(self.temp.name) / 'third.sqlite'
|
|
self.sync.synchronize(second, third, FakeUpstream())
|
|
for table in ('titles', 'seasons', 'episodes', 'movie_sources', 'episode_sources'):
|
|
self.assertEqual(self.rows(second, 'SELECT * FROM ' + table), self.rows(third, 'SELECT * FROM ' + table))
|
|
season = json.loads(self.rows(second, 'SELECT raw_json FROM seasons WHERE id=1001')[0][0])
|
|
self.assertEqual([e['id'] for e in season['episodes']], [501, 502])
|
|
self.assertEqual(season['episodes'][1]['sources'][0]['url'], 'https://video.test/refreshed')
|
|
movie = json.loads(self.rows(second, 'SELECT raw_json FROM titles WHERE id=1')[0][0])
|
|
self.assertTrue(movie['sources'])
|
|
self.assertTrue(movie['genres'])
|
|
|
|
def test_old_series_absent_from_catalogue_and_new_series_receive_episodes_idempotently(self):
|
|
new_series = {'id': 70, 'type': 'serie', 'title': 'New Series'}
|
|
old_episode = {'id': 501, 'title': 'Episode 1', 'sources': []}
|
|
new_episode = {'id': 502, 'title': 'Episode 2', 'sources': [
|
|
{'id': 202, 'quality': '1080', 'type': 'mkv', 'url': 'https://video.test/e2.mkv'}]}
|
|
routes = {'serie/by/filtres/0/created/0': [new_series],
|
|
'serie/by/filtres/0/created/1': [],
|
|
'season/by/serie/2': [{'id': 1001, 'title': 'Season 1', 'episodes': [old_episode, new_episode]}],
|
|
'season/by/serie/70': [{'id': 1070, 'title': 'New Season', 'episodes': [
|
|
{'id': 701, 'title': 'Pilot', 'sources': [
|
|
{'id': 702, 'url': 'https://video.test/pilot.mp4', 'type': 'mp4'}]}]}]}
|
|
report = self.sync.synchronize(self.source, self.stage, FakeUpstream(routes))
|
|
self.assertEqual(report['added'], {'titles': 1, 'movies': 0, 'series': 1, 'episodes': 2, 'sources': 2})
|
|
self.assertEqual(report['updated_series_ids'], [2, 70])
|
|
self.assertEqual(report['updated_series_count'], 2)
|
|
self.assertEqual(self.rows(self.stage, 'SELECT id FROM episodes ORDER BY id'), [(501,), (502,), (701,)])
|
|
raw = json.loads(self.rows(self.stage, 'SELECT raw_json FROM seasons WHERE series_id=2')[0][0])
|
|
self.assertEqual(len(raw['episodes']), 2)
|
|
self.assertEqual(raw['episodes'][0]['sources'][0]['id'], 201)
|
|
self.assertEqual(raw['episodes'][1], new_episode)
|
|
catalog = load_service().Catalog(self.stage)
|
|
try:
|
|
served = catalog.dispatch('/api/season/by/serie/2/public/')
|
|
self.assertEqual(served[0]['episodes'][1], new_episode)
|
|
finally:
|
|
catalog.close()
|
|
second_stage = Path(self.temp.name) / 'second.sqlite'
|
|
again = self.sync.synchronize(self.stage, second_stage, FakeUpstream(routes))
|
|
self.assertEqual(again['added'], dict.fromkeys(report['added'], 0))
|
|
self.assertEqual(again['updated_series_ids'], [])
|
|
for table in ('titles', 'seasons', 'episodes', 'movie_sources', 'episode_sources'):
|
|
self.assertEqual(self.rows(self.stage, 'SELECT * FROM ' + table),
|
|
self.rows(second_stage, 'SELECT * FROM ' + table))
|
|
self.assertEqual(self.metadata(second_stage)['original_snapshot_built_at_utc'],
|
|
self.metadata(self.source)['built_at_utc'])
|
|
for (url,) in self.rows(second_stage, 'SELECT source_url FROM series_fetch_status'):
|
|
self.assertTrue(url.startswith('https://ninwinios.info/api/season/by/serie/'))
|
|
self.assertNotIn('key', url)
|
|
|
|
def test_new_movie_details_staged_source_untouched_and_real_counts(self):
|
|
original = self.source.read_bytes()
|
|
movie = {'id': 70, 'type': 'movie', 'title': 'New Film', 'year': 2026,
|
|
'genres': [{'id': 5, 'title': 'Drama'}], 'country': [{'id': 3, 'title': 'Iran'}],
|
|
'image': 'https://images.test/never-fetch.jpg'}
|
|
detail = dict(movie, description='New description', sources=[
|
|
{'id': 700, 'type': 'mkv', 'url': 'https://video.test/never-fetch.mkv?key=opaque'}])
|
|
upstream = FakeUpstream({'genre/all': [{'id': 5, 'title': 'Drama'}],
|
|
'country/all': [{'id': 3, 'title': 'Iran'}],
|
|
'movie/by/filtres/0/created/0': [movie, movie],
|
|
'movie/by/filtres/0/created/1': [], 'movie/by/70': detail})
|
|
progress = []
|
|
report = self.sync.synchronize(self.source, self.stage, upstream, progress=progress.append)
|
|
self.assertEqual(self.source.read_bytes(), original)
|
|
self.assertEqual(report['added']['titles'], 1)
|
|
self.assertEqual(report['added']['movies'], 1)
|
|
self.assertEqual(report['added']['series'], 0)
|
|
self.assertEqual(report['added']['sources'], 1)
|
|
self.assertEqual(report['counts']['titles'], 66)
|
|
self.assertEqual(report['counts']['episodes'], 1)
|
|
self.assertEqual(report['requests_count'], upstream.requests_count)
|
|
self.assertFalse(report['media_files_downloaded'])
|
|
self.assertEqual(upstream.calls.count('movie/by/70'), 1)
|
|
self.assertIn('season/by/serie/2', upstream.calls)
|
|
self.assertTrue(all('http' not in path and '.mkv' not in path and '.jpg' not in path
|
|
for path in upstream.calls))
|
|
raw = json.loads(self.rows(self.stage, 'SELECT raw_json FROM titles WHERE id=70')[0][0])
|
|
self.assertEqual(raw, detail)
|
|
self.assertEqual(self.rows(self.stage, 'SELECT url FROM movie_sources WHERE title_id=70')[0][0],
|
|
detail['sources'][0]['url'])
|
|
catalog = load_service().Catalog(self.stage)
|
|
try:
|
|
served = catalog.dispatch('/api/movie/by/70/public/')
|
|
self.assertEqual(served, detail)
|
|
finally:
|
|
catalog.close()
|
|
self.assertEqual(self.rows(self.stage, 'PRAGMA integrity_check'), [('ok',)])
|
|
self.assertEqual(self.rows(self.stage, 'PRAGMA foreign_key_check'), [])
|
|
metadata = self.metadata(self.stage)
|
|
self.assertEqual(metadata['original_snapshot_built_at_utc'], self.metadata(self.source)['built_at_utc'])
|
|
self.assertNotEqual(metadata['built_at_utc'], self.metadata(self.source)['built_at_utc'])
|
|
self.assertEqual(metadata['built_at_utc'], metadata['synchronized_at_utc'])
|
|
self.assertEqual(metadata['sync_interval_seconds'], 259200)
|
|
self.assertEqual(metadata['counts']['titles'], 66)
|
|
self.assertFalse(metadata['media_files_downloaded'])
|
|
self.assertEqual(metadata['last_sync']['upstream_base'], 'https://ninwinios.info/api/')
|
|
self.assertEqual(json.loads(json.dumps(report)), report)
|
|
self.assertTrue(progress)
|
|
for item in progress:
|
|
self.assertLessEqual(set(item), {'phase', 'count', 'total'})
|
|
self.assertIsInstance(item['phase'], str)
|
|
self.assertIsInstance(item['count'], int)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|