import importlib.util import json import os from pathlib import Path import sqlite3 import tempfile import unittest SERVER = Path(__file__).resolve().parents[1] def load_service(): spec = importlib.util.spec_from_file_location('catalog_api', SERVER / 'catalog_api.py') module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def fixture(directory): path = Path(directory) / 'catalog.sqlite' con = sqlite3.connect(path) con.executescript((SERVER / 'tests' / 'schema.sql').read_text()) metadata = {'built_at_utc': '2026-11-02T07:00:18+00:00', 'provenance': 'Realistic test snapshot', 'catalog_pagination_complete': True, 'series_episode_fetch_complete': True} con.executemany('INSERT INTO metadata VALUES (?,?)', [(k, json.dumps(v)) for k, v in metadata.items()]) con.executemany('INSERT INTO genres VALUES (?,?)', [(1, 'Horror'), (2, 'Action'), (32, 'Top 250'), (48, 'Live')]) con.execute('INSERT INTO countries VALUES (1,?,?)', ('America', 'http://art.test/a%2Fb.jpg?token=Ab+Cd')) for i in range(1, 66): kind = 'serie' if i == 2 else 'movie' title = 'Alpha' if i == 1 else 'Warrior' if i == 2 else 'Film ' + str(i) if i == 63: title = 'نسخه جدید ملودی باکس' if i == 64: kind = 'channel' if i == 65: title = 'Telegram @channelname' item = {'id': i, 'type': kind, 'title': title, 'description': 'Actor Example بازیگران', 'year': 2020 + i % 7, 'imdb': i / 10, 'rating': 4.1, 'image': 'http://art.test/a%2Fb.jpg?token=Ab+Cd', 'cover': 'https://art.test/a b.jpg', 'comment': True, 'genres': [{'id': 2, 'title': 'Action'}], 'country': [{'id': 1, 'title': 'America'}]} if kind == 'movie': source = {'id': 100 + i, 'type': 'mkv', 'quality': '720', 'url': 'http://media.test/Movie%20Name%2F.mkv?token=Ab+Cd&x=%2f#fragment'} item['sources'] = [source] con.execute('INSERT INTO titles(id,type,title,description,year,imdb,rating,raw_json) VALUES (?,?,?,?,?,?,?,?)', (i, kind, title, item['description'], item['year'], item['imdb'], item['rating'], json.dumps(item))) if kind == 'movie': con.execute('INSERT INTO movie_sources VALUES (?,?,?,?,?,?,?)', (i, 0, source['id'], '720', 'mkv', source['url'], json.dumps(source))) con.execute('INSERT INTO title_genres VALUES (?,2)', (i,)) con.execute('INSERT INTO title_countries VALUES (?,1)', (i,)) con.execute('INSERT INTO title_genres VALUES (1,32)') episode = {'id': 501, 'title': 'Episode 1', 'sources': [{'id': 201, 'type': 'mkv', 'quality': None, 'url': 'http://media.test/a%20b.mkv?signature=%2f+Ab'}]} season = {'id': 1001, 'title': 'Season 1', 'episodes': [episode]} con.execute('INSERT INTO seasons VALUES (2,0,1001,?,?)', ('Season 1', json.dumps(season))) con.execute('INSERT INTO episodes(series_id,season_index,episode_index,id,title,raw_json) VALUES (2,0,0,501,?,?)', ('Episode 1', json.dumps(episode))) con.execute('INSERT INTO episode_sources VALUES (2,0,0,0,201,NULL,?,?,?)', ('mkv', episode['sources'][0]['url'], json.dumps(episode['sources'][0]))) con.execute('INSERT INTO series_fetch_status VALUES (2,?,1,?)', ('ok', 'https://old.test/api/season/2')) con.commit() con.close() return path class CatalogTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) self.path = fixture(self.temp.name) def tearDown(self): if hasattr(self, 'catalog'): self.catalog.close() self.temp.cleanup() def open(self): self.api = load_service() self.catalog = self.api.Catalog(self.path) return self.catalog def test_health_is_real_snapshot_counts_and_freshness(self): catalog = self.open() health = catalog.dispatch('/api/health') self.assertEqual(health['status'], 'ok') self.assertEqual(health['counts']['titles'], 65) self.assertEqual(health['counts']['episodes'], 1) self.assertEqual(health['counts']['visible_titles'], 62) self.assertEqual(health['snapshot_built_at_utc'], '2026-11-02T07:00:18+00:00') self.assertEqual(health['provenance'], 'Realistic test snapshot') self.assertFalse(health['media_proxy']) self.assertTrue(health['offline']) self.assertGreaterEqual(health['snapshot_age_seconds'], 0) with self.assertRaises(sqlite3.OperationalError): catalog.db.execute('DELETE FROM titles') def test_catalog_filters_pages_and_original_media_links(self): catalog = self.open() first = catalog.dispatch('/api/movie/by/filtres/0/created/0/public/') second = catalog.dispatch('/api/movie/by/filtres/0/created/1/public/') self.assertEqual(len(first), 30) self.assertEqual(len(second), 30) self.assertFalse({r['id'] for r in first} & {r['id'] for r in second}) self.assertTrue(all(r['id'] <= 62 and r['type'] == 'movie' for r in first + second)) self.assertEqual([r['id'] for r in catalog.dispatch('/api/movie/by/filtres/0/created/2/public/')], [1]) self.assertEqual(catalog.dispatch('/api/movie/by/filtres/0/created/3/public/'), []) self.assertEqual([r['id'] for r in catalog.dispatch('/api/poster/by/filtres/32/1/imdb/0/public/')], [1]) self.assertEqual([r['id'] for r in catalog.dispatch('/api/serie/by/filtres/0/year/0/public/')], [2]) film = catalog.dispatch('/api/movie/by/1/public/') original = json.loads(catalog.db.execute('SELECT raw_json FROM titles WHERE id=1').fetchone()[0]) self.assertEqual(film['sources'], original['sources']) self.assertEqual(film['image'], original['image']) self.assertNotIn('comment', film) seasons = catalog.dispatch('/api/season/by/serie/2/public/') self.assertEqual(seasons[0]['episodes'][0]['sources'][0]['url'], 'http://media.test/a%20b.mkv?signature=%2f+Ab') self.assertEqual(catalog.dispatch('/api/genre/all/public/')[0]['id'], 1) self.assertNotIn(48, [r['id'] for r in catalog.dispatch('/api/genre/all/public/')]) self.assertEqual(catalog.dispatch('/api/country/all/public/')[0]['title'], 'America') for title_id in (63, 64, 65, 999): with self.assertRaises(self.api.ApiError) as error: catalog.dispatch(f'/api/movie/by/{title_id}/public/') self.assertEqual(error.exception.status, 404) def test_home_search_recommendations_curated_and_unexported_metadata(self): catalog = self.open() home = catalog.dispatch('/api/first/public/') self.assertTrue(home['genres'][0]['posters']) self.assertEqual(home['actors'], []) self.assertEqual(home['countries'][0]['id'], 1) search = catalog.dispatch('/api-user/newapi/like.php?action=search-movie-en&q=alpha&pageno=1') self.assertEqual([r['id'] for r in search], [1]) self.assertEqual(catalog.dispatch('/api-user/newapi/like.php?action=search-movie&q=no-match&pageno=1'), []) self.assertEqual(catalog.dispatch('/api-user/newapi/like.php?action=search-movie&q=%27%20OR%201%3D1--'), []) self.assertEqual(catalog.dispatch('/api-user/newapi/like.php?action=best-rand-top&pageno=1')[0]['id'], 62) self.assertTrue(catalog.dispatch('/api-user/newapi/like.php?action=best-rand-year&pageno=1')) curated = catalog.dispatch('/api-user/newapi/pishnahad.php') self.assertTrue(curated) self.assertEqual(len(curated[0]['title']), len(curated[0]['link'])) self.assertTrue(catalog.dispatch(curated[0]['link'][0] + '1')) self.assertEqual([r['id'] for r in catalog.dispatch('/api-user/newapi/filter.php?action=pishnahad&garid=32&pageno=1')], [1]) for target in ['/api-user/newapi/like.php?action=all-actor&pageno=1', '/api-user/newapi/like.php?action=colection&pageno=1', '/api-user/filmahang/star.php?key=public&action=starsearch&q=Example&pageno=1', '/api-user/filmahang/star.php?key=public&action=mazmon&pageno=1', '/api-user/filmahang/star.php?key=public&action=share&pageno=1', '/api/role/by/poster/1/public/', '/api/movie/by/actor/12/public/', '/api/subtitles/by/movie/1/public/', '/api/subtitles/by/episode/501/public/']: self.assertEqual(catalog.dispatch(target), [], target) def test_rejects_proxy_comments_traversal_and_unbounded_inputs(self): catalog = self.open() for target in ['/api/proxy?url=http://169.254.169.254/', '/api/download/movie.mkv', '/api/comments/by/poster/1/public/', '/api/../catalog.sqlite', '/api/%2e%2e/catalog.sqlite', '//evil.test/api/health', '/api/movie/by/1/public/?url=https://media.test/movie.mkv', '/api-user/newapi/like.php?action=comments', '/api-user/newapi/like.php?action=search-movie&q=a&q=b', '/api-user/newapi/like.php?action=search-movie&q=' + 'a' * 201, '/api-user/newapi/like.php?action=search-movie&pageno=-1', '/api-user/newapi/like.php?action=search-movie&pageno=100002', '/api/movie/by/filtres/0/views/0/public/']: with self.assertRaises(self.api.ApiError, msg=target): catalog.dispatch(target) def test_raw_controls_and_malformed_urls_are_rejected_before_parsing(self): catalog = self.open() for target in ['/api/\thealth', '\n/api/health', '/api/health\r', 'http://[', '/api-user/newapi/like.php?action=search-movie&q=a\tb']: with self.subTest(target=repr(target)): with self.assertRaises(self.api.ApiError) as error: catalog.dispatch(target) self.assertEqual(error.exception.status, 400) def test_real_http_is_read_only_bounded_and_returns_direct_source_urls(self): import threading from urllib.request import Request, urlopen from urllib.error import HTTPError catalog = self.open() server = self.api.make_server(catalog, port=0) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() origin = 'http://127.0.0.1:' + str(server.server_port) try: with urlopen(origin + '/api/movie/by/1/public/', timeout=3) as response: detail = json.load(response) self.assertEqual(response.headers['Content-Type'], 'application/json; charset=utf-8') self.assertNotIn('Server', response.headers) self.assertEqual(detail['sources'][0]['url'], 'http://media.test/Movie%20Name%2F.mkv?token=Ab+Cd&x=%2f#fragment') for method in ['POST', 'PUT', 'DELETE', 'PATCH', 'CONNECT']: with self.assertRaises(HTTPError) as error: urlopen(Request(origin + '/api/health', method=method), timeout=3) self.assertEqual(error.exception.code, 405) error.exception.close() with self.assertRaises(HTTPError) as error: urlopen(origin + '/api/proxy?url=https://media.test/a.mkv', timeout=3) self.assertIn(error.exception.code, (400, 404)) error.exception.close() with urlopen(Request(origin + '/api/health', method='HEAD'), timeout=3) as response: self.assertEqual(response.status, 200) self.assertEqual(response.read(), b'') finally: server.shutdown() server.server_close() thread.join(3) def test_worker_saturation_rejects_extra_connections_and_releases_slots(self): import socket import threading import time catalog = self.open() server = self.api.make_server(catalog, port=0) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() sockets = [] def wait_slots(value): deadline = time.monotonic() + 4 while server.slots._value != value and time.monotonic() < deadline: time.sleep(.01) self.assertEqual(server.slots._value, value) try: for _ in range(16): sock = socket.create_connection(server.server_address, timeout=2) sock.sendall(b'GET /api/health HTTP/1.0\r\n') sockets.append(sock) wait_slots(0) with socket.create_connection(server.server_address, timeout=2) as extra: try: self.assertEqual(extra.recv(1), b'') except ConnectionResetError: pass for sock in sockets: sock.close() sockets.clear() wait_slots(16) with socket.create_connection(server.server_address, timeout=2) as recovered: recovered.sendall(b'GET /api/health HTTP/1.0\r\n\r\n') self.assertIn(b'200', recovered.recv(100)) finally: for sock in sockets: sock.close() server.shutdown() server.server_close() thread.join(3) if __name__ == '__main__': unittest.main()