103 lines
7.1 KiB
Python
103 lines
7.1 KiB
Python
import asyncio
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
from fastapi import HTTPException
|
|
from backend.app.config import settings
|
|
from backend.app.services import manual_releases as manual
|
|
from backend.app.routers import requests
|
|
from backend.app.models import Snapshot, RequestType
|
|
from backend.app.feature_access import permissions, update_permissions
|
|
from backend.app import db
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
|
|
|
|
class ManualSelectionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
secret = patch.object(settings, 'jwt_secret', 'manual-selection-test-secret-123456789')
|
|
secret.start(); self.addCleanup(secret.stop)
|
|
self.user = {'username': 'viewer', 'role': 'user', 'features': {'ignore_profile_limits': True}}
|
|
self.release = {'guid': 'release', 'indexerId': 7, 'title': 'Example', 'requiresOverride': True,
|
|
'rejections': ['WEBDL-2160p is not wanted in profile']}
|
|
self.payload = {**self.release, 'ignoreProfileLimits': True,
|
|
'selectionToken': manual.issue_selection(self.release, '42', self.user, 'http://sonarr', 55)}
|
|
|
|
def test_profile_only_rejections_are_overridable(self):
|
|
for reason in ['WEBDL-2160p is not wanted in profile', 'Custom format score below minimum', 'File is larger than maximum size', 'Language is not wanted']:
|
|
self.assertTrue(manual.decision({'approved': False, 'rejections': [reason]})[1])
|
|
for reason in ['Unknown series', 'Release is blocklisted', 'No download client available', 'Already in queue']:
|
|
self.assertFalse(manual.decision({'rejections': [self.release['rejections'][0], reason]})[1])
|
|
self.assertFalse(manual.decision({'approved': True, 'downloadAllowed': False})[0])
|
|
|
|
def test_receipt_binds_request_user_source_item_and_release(self):
|
|
self.assertTrue(manual.verify_selection(self.payload, '42', self.user, 'http://sonarr', 55)['override'])
|
|
attempts = [({**self.payload, 'guid': 'other'}, '42', self.user, 'http://sonarr', 55),
|
|
(self.payload, '43', self.user, 'http://sonarr', 55),
|
|
(self.payload, '42', {**self.user, 'username': 'other'}, 'http://sonarr', 55),
|
|
(self.payload, '42', self.user, 'http://other', 55),
|
|
(self.payload, '42', self.user, 'http://sonarr', 56),
|
|
({**self.payload, 'selectionToken': 'forged'}, '42', self.user, 'http://sonarr', 55)]
|
|
for args in attempts:
|
|
with self.assertRaises(HTTPException): manual.verify_selection(*args)
|
|
|
|
def test_permission_revocation_and_literal_confirmation_enforced(self):
|
|
for payload, user, code in [(self.payload, {**self.user, 'features': {}}, 403),
|
|
({**self.payload, 'ignoreProfileLimits': 'true'}, self.user, 400)]:
|
|
with self.assertRaises(HTTPException) as error:
|
|
manual.verify_selection(payload, '42', user, 'http://sonarr', 55)
|
|
self.assertEqual(error.exception.status_code, code)
|
|
|
|
|
|
class ManualPermissionTests(TempDatabaseMixin, unittest.TestCase):
|
|
def test_default_off_individual_and_bulk(self):
|
|
for name in ('one', 'two'): db.create_user(name, 'Password123!', role='user')
|
|
one, two = [db.get_user_by_username(n) for n in ('one', 'two')]
|
|
self.assertFalse(permissions(one)['ignore_profile_limits'])
|
|
update_permissions({'ignore_profile_limits': True}, 'one')
|
|
self.assertTrue(permissions(one)['ignore_profile_limits'])
|
|
self.assertFalse(permissions(two)['ignore_profile_limits'])
|
|
update_permissions({'ignore_profile_limits': False})
|
|
self.assertFalse(permissions(one)['ignore_profile_limits'])
|
|
|
|
|
|
class ManualEpisodeSearchTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
access = patch.object(
|
|
requests,
|
|
'_ensure_request_mutation_access',
|
|
new=AsyncMock(return_value=None),
|
|
)
|
|
access.start()
|
|
self.addCleanup(access.stop)
|
|
|
|
async def test_episode_batch_is_bounded_and_exposes_next_page(self):
|
|
episodes = [{'id': i, 'seasonNumber': 1, 'monitored': True, 'hasFile': False} for i in range(1, 26)]
|
|
episodes += [{'id': 26, 'seasonNumber': 1, 'monitored': True, 'hasFile': True}]
|
|
active = peak = 0
|
|
async def search(identity):
|
|
nonlocal active, peak
|
|
active += 1; peak = max(peak, active)
|
|
await asyncio.sleep(0.001); active -= 1
|
|
return []
|
|
sonarr = SimpleNamespace(configured=lambda: True, get_episodes=AsyncMock(return_value=episodes), search_episode_releases=AsyncMock(side_effect=search))
|
|
runtime = SimpleNamespace(jellyseerr_base_url=None,jellyseerr_api_key=None,sonarr_base_url='http://sonarr',sonarr_api_key='test')
|
|
snapshot = Snapshot(request_id='42',request_type=RequestType.tv,title='Example',raw={'arr':{'item':{'id':55,'qualityProfileId':9}}})
|
|
with patch.object(requests, 'get_runtime_settings', return_value=runtime), patch.object(requests, 'build_snapshot', new=AsyncMock(return_value=snapshot)), patch.object(requests,'SonarrClient',return_value=sonarr), patch.object(requests,'save_action'):
|
|
first = await requests.action_search('42', {'username':'viewer','role':'user'})
|
|
second = await requests.action_search('42', {'username':'viewer','role':'user'}, offset=24)
|
|
self.assertEqual(first['nextOffset'],3); self.assertIsNone(second['nextOffset'])
|
|
self.assertEqual(sonarr.search_episode_releases.await_count,4)
|
|
self.assertLessEqual(peak,3)
|
|
self.assertEqual(first['totalMissingEpisodes'],25)
|
|
|
|
async def test_auto_search_preserves_current_profile(self):
|
|
for kind, service in [(RequestType.tv,'SonarrClient'),(RequestType.movie,'RadarrClient')]:
|
|
client=SimpleNamespace(configured=lambda:True, update_series=AsyncMock(), update_movie=AsyncMock(),
|
|
get_episodes=AsyncMock(return_value=[{'id':1,'seasonNumber':1,'monitored':True,'hasFile':False}]),
|
|
search_episodes=AsyncMock(return_value={'id':1}), search=AsyncMock(return_value={'id':1}))
|
|
runtime=SimpleNamespace(jellyseerr_base_url=None,jellyseerr_api_key=None,sonarr_base_url='http://sonarr',sonarr_api_key='test',radarr_base_url='http://radarr',radarr_api_key='test',sonarr_quality_profile_id=6,radarr_quality_profile_id=6)
|
|
snapshot=Snapshot(request_id='42',request_type=kind,title='Example',raw={'arr':{'item':{'id':55,'qualityProfileId':9}}})
|
|
with patch.object(requests,'get_runtime_settings',return_value=runtime),patch.object(requests,'build_snapshot',new=AsyncMock(return_value=snapshot)),patch.object(requests,service,return_value=client),patch.object(requests,'save_action'),patch.object(requests,'series_search_outcome',new=AsyncMock(return_value={'status':'attention','message':'Nothing queued'})),patch.object(requests,'movie_search_outcome',new=AsyncMock(return_value={'status':'attention','message':'Nothing queued'})):
|
|
await requests.action_search_auto('42',{'username':'admin','role':'admin'})
|
|
client.update_series.assert_not_awaited(); client.update_movie.assert_not_awaited()
|