247 lines
13 KiB
Python
247 lines
13 KiB
Python
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()
|