import os from types import SimpleNamespace import tempfile import unittest from unittest.mock import AsyncMock, patch import httpx from fastapi import HTTPException from starlette.requests import Request from backend.app import db from backend.app.clients.base import _operation_error_message, _operation_result_message from backend.app.clients.jellyfin import _availability_message from backend.app.clients.qbittorrent import _torrent_result_message from backend.app.auth import require_admin from backend.app.config import settings from backend.app.network_security import request_trusts_forwarded_headers, validate_notification_target_url from backend.app.models import ActionOption, NormalizedState, RequestType, Snapshot, TimelineHop from backend.app.routers import auth as auth_router from backend.app.routers import portal as portal_router from backend.app.routers import requests as requests_router from backend.app.routers import site as site_router from backend.app.routers import status as status_router from backend.app.security import PASSWORD_POLICY_MESSAGE, validate_password_policy from backend.app.services import password_reset from backend.app.services.operation_progress import ( begin_operation, finish_operation, finish_remote_call, get_operation, reset_operation, start_remote_call, ) from backend.app.services.snapshot import ( _apply_arr_identity, _build_presentation, _episode_availability, _torrent_progress, ) def _build_request(ip: str = "127.0.0.1", user_agent: str = "backend-test") -> Request: scope = { "type": "http", "http_version": "1.1", "method": "POST", "scheme": "http", "path": "/auth/password/forgot", "raw_path": b"/auth/password/forgot", "query_string": b"", "headers": [(b"user-agent", user_agent.encode("utf-8"))], "client": (ip, 12345), "server": ("testserver", 8000), } async def receive() -> dict: return {"type": "http.request", "body": b"", "more_body": False} return Request(scope, receive) class TempDatabaseMixin: def setUp(self) -> None: super_method = getattr(super(), "setUp", None) if callable(super_method): super_method() self._tempdir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self._original_sqlite_path = settings.sqlite_path self._original_journal_mode = getattr(settings, "sqlite_journal_mode", "DELETE") settings.sqlite_path = os.path.join(self._tempdir.name, "test.db") settings.sqlite_journal_mode = "DELETE" auth_router._LOGIN_ATTEMPTS_BY_IP.clear() auth_router._LOGIN_ATTEMPTS_BY_USER.clear() auth_router._RESET_ATTEMPTS_BY_IP.clear() auth_router._RESET_ATTEMPTS_BY_IDENTIFIER.clear() db.init_db() def tearDown(self) -> None: settings.sqlite_path = self._original_sqlite_path settings.sqlite_journal_mode = self._original_journal_mode auth_router._LOGIN_ATTEMPTS_BY_IP.clear() auth_router._LOGIN_ATTEMPTS_BY_USER.clear() auth_router._RESET_ATTEMPTS_BY_IP.clear() auth_router._RESET_ATTEMPTS_BY_IDENTIFIER.clear() self._tempdir.cleanup() super_method = getattr(super(), "tearDown", None) if callable(super_method): super_method() class PasswordPolicyTests(unittest.TestCase): def test_validate_password_policy_rejects_short_passwords(self) -> None: with self.assertRaisesRegex(ValueError, PASSWORD_POLICY_MESSAGE): validate_password_policy("short") def test_validate_password_policy_trims_whitespace(self) -> None: self.assertEqual(validate_password_policy(" password123 "), "password123") class NetworkSecurityTests(unittest.TestCase): def test_notification_targets_reject_loopback(self) -> None: with self.assertRaisesRegex(ValueError, "Private or local notification targets are not allowed."): validate_notification_target_url("http://127.0.0.1:8080/webhook") def test_forwarded_headers_require_trusted_proxy(self) -> None: original_enabled = settings.magent_proxy_enabled original_trust = settings.magent_proxy_trust_forwarded_headers original_proxies = settings.magent_proxy_trusted_proxies settings.magent_proxy_enabled = True settings.magent_proxy_trust_forwarded_headers = True settings.magent_proxy_trusted_proxies = "127.0.0.1,::1" try: self.assertTrue(request_trusts_forwarded_headers("127.0.0.1")) self.assertFalse(request_trusts_forwarded_headers("203.0.113.10")) finally: settings.magent_proxy_enabled = original_enabled settings.magent_proxy_trust_forwarded_headers = original_trust settings.magent_proxy_trusted_proxies = original_proxies class ServiceStatusTests(unittest.IsolatedAsyncioTestCase): def test_status_router_requires_admin(self) -> None: dependencies = [getattr(dependency, "dependency", None) for dependency in status_router.router.dependencies] self.assertIn(require_admin, dependencies) async def test_qbittorrent_login_accepts_modern_empty_response_with_session_cookie(self) -> None: class FakeClient: def __init__(self) -> None: self.cookies = httpx.Cookies() async def post(self, *_args, **_kwargs) -> httpx.Response: self.cookies.set("QBT_SID_8080", "session") return httpx.Response(204, request=httpx.Request("POST", "http://10.0.0.2:8080/api/v2/auth/login")) client = status_router.QBittorrentClient("http://10.0.0.2:8080", "admin", "secret") await client._login(FakeClient()) async def test_qbittorrent_incomplete_credentials_report_degraded_when_reachable(self) -> None: client = status_router.QBittorrentClient("http://10.0.0.2:8080", "admin", None) with patch.object(client, "is_webui_reachable", new=AsyncMock(return_value=True)): result = await status_router._check_qbittorrent(client) self.assertEqual(result["status"], "degraded") self.assertIn("credentials", result["message"].lower()) async def test_qbittorrent_rejected_credentials_report_degraded_when_reachable(self) -> None: client = status_router.QBittorrentClient("http://10.0.0.2:8080", "admin", "secret") with patch.object( client, "get_app_version", new=AsyncMock(side_effect=RuntimeError("qBittorrent login failed")), ), patch.object(client, "is_webui_reachable", new=AsyncMock(return_value=True)): result = await status_router._check_qbittorrent(client) self.assertEqual(result["status"], "degraded") self.assertIn("credentials", result["message"].lower()) class OperationProgressTests(unittest.TestCase): def test_remote_interaction_is_visible_until_operation_completes(self) -> None: operation_id = "operation-progress-test" token = begin_operation( operation_id, label="Recheck request status", path="/requests/3914/actions/recheck", ) try: event_id = start_remote_call("Radarr") active = get_operation(operation_id) self.assertEqual(active["status"], "running") self.assertEqual(active["events"][-1]["service"], "Radarr") self.assertEqual(active["events"][-1]["state"], "active") finish_remote_call( event_id, success=True, status_code=200, message="Radarr responded in 0.2s.", ) finish_operation(operation_id, success=True, status_code=200) finally: reset_operation(token) completed = get_operation(operation_id) self.assertEqual(completed["status"], "complete") self.assertEqual(completed["events"][-2]["state"], "complete") self.assertEqual(completed["events"][-2]["status_code"], 200) self.assertEqual(completed["events"][-1]["service"], "Magent") class OperationMessageTests(unittest.TestCase): def test_radarr_lookup_explains_whether_movie_was_found(self) -> None: found = _operation_result_message( "Radarr", "GET", "/api/v3/movie", [{"title": "Arrival"}], ) missing = _operation_result_message( "Radarr", "GET", "/api/v3/movie", [], ) self.assertEqual(found, 'Radarr found "Arrival" in its library list.') self.assertEqual(missing, "This movie is not currently in Radarr.") def test_radarr_add_explains_that_download_search_started(self) -> None: message = _operation_result_message( "Radarr", "POST", "/api/v3/movie", {"title": "Arrival", "id": 42}, payload={ "title": "Arrival", "addOptions": {"searchForMovie": True}, }, ) self.assertEqual(message, 'Radarr added "Arrival" and started looking for a download.') def test_queue_and_indexer_health_results_are_summarized(self) -> None: queue_message = _operation_result_message( "Sonarr", "GET", "/api/v3/queue", {"totalRecords": 0, "records": []}, ) health_message = _operation_result_message( "Prowlarr", "GET", "/api/v1/health", [], ) self.assertEqual(queue_message, "Sonarr has no matching downloads in its queue.") self.assertEqual(health_message, "Prowlarr reports that all configured indexers are healthy.") def test_download_and_jellyfin_results_include_actual_state(self) -> None: torrent_message = _torrent_result_message( [{"name": "Arrival.2016", "state": "downloading", "progress": 0.42}] ) self.assertEqual( torrent_message, 'qBittorrent found "Arrival.2016"; it is downloading and 42% complete.', ) self.assertEqual( _availability_message({"TotalRecordCount": 0, "Items": []}), "The title is not currently available in Jellyfin.", ) def test_http_failures_are_translated_without_exposing_stack_traces(self) -> None: self.assertEqual( _operation_error_message("Radarr", 500), "Radarr encountered an internal error while processing the request.", ) self.assertEqual( _operation_error_message("Sonarr", 401), "Sonarr rejected Magent's login details.", ) class SiteInfoTests(unittest.TestCase): def test_site_public_exposes_requests_navigation_toggle(self) -> None: runtime = SimpleNamespace( site_build_number="test-build", site_banner_enabled=False, site_banner_message="", site_banner_tone="info", site_login_show_jellyfin_login=True, site_login_show_local_login=True, site_login_show_forgot_password=True, site_login_show_signup_link=True, site_nav_show_requests=False, ) with patch.object(site_router, "get_runtime_settings", return_value=runtime): info = site_router._build_site_info(False) self.assertEqual(info["navigation"], {"showRequests": False}) class RequestCacheTests(unittest.TestCase): def tearDown(self) -> None: requests_router._detail_cache.clear() requests_router._failed_detail_cache.clear() def test_successful_detail_cache_write_clears_prior_failure(self) -> None: key = "request:123" requests_router._failure_cache_set(key) self.assertTrue(requests_router._failure_cache_has(key)) requests_router._cache_set(key, {"id": 123}) self.assertFalse(requests_router._failure_cache_has(key)) self.assertEqual(requests_router._cache_get(key), {"id": 123}) class RequestVisibilityTests(unittest.TestCase): def test_non_admin_snapshot_excludes_advanced_identifying_data(self) -> None: snapshot = Snapshot( request_id="3925", title="Example", timeline=[ TimelineHop( service="Seerr", status="approved", details={"requestedBy": "viewer@example.com"}, ) ], raw={ "jellyseerr": {"requestedBy": {"email": "viewer@example.com"}}, "qbittorrent": {"downloadIds": ["secret-hash"]}, }, actions=[ ActionOption( id="search_releases", label="Search and choose a download", risk="safe", ) ], presentation={"status": {"label": "Needs attention"}}, ) filtered = requests_router._filter_snapshot_for_user( snapshot, {"username": "helper", "role": "user"} ) self.assertEqual(filtered.timeline, []) self.assertEqual(filtered.raw, {}) self.assertEqual([action.id for action in filtered.actions], ["search_releases"]) self.assertEqual(filtered.presentation["status"]["label"], "Needs attention") def test_admin_snapshot_retains_advanced_diagnostics(self) -> None: snapshot = Snapshot( request_id="3925", title="Example", timeline=[TimelineHop(service="Seerr", status="approved")], raw={"jellyseerr": {"id": 3925}}, ) filtered = requests_router._filter_snapshot_for_user( snapshot, {"username": "admin", "role": "admin"} ) self.assertEqual(len(filtered.timeline), 1) self.assertEqual(filtered.raw["jellyseerr"]["id"], 3925) def test_non_admin_cannot_request_advanced_history(self) -> None: with self.assertRaises(HTTPException) as context: requests_router._require_advanced_request_access( {"username": "helper", "role": "user"} ) self.assertEqual(context.exception.status_code, 403) def test_my_requests_cache_only_returns_signed_in_users_rows(self) -> None: previous = dict(requests_router._recent_cache) requests_router._recent_cache["items"] = [ {"request_id": 100, "requested_by_id": 7, "requested_by_norm": "zak"}, {"request_id": 101, "requested_by_id": 8, "requested_by_norm": "someone-else"}, ] try: rows = requests_router._get_recent_from_cache( requested_by_norm="zak", requested_by_id=7, limit=10, offset=0, since_iso=None, ) finally: requests_router._recent_cache.clear() requests_router._recent_cache.update(previous) self.assertEqual([row["request_id"] for row in rows], [100]) class RequestPresentationTests(unittest.TestCase): def test_torrent_progress_keeps_tenths_for_live_updates(self) -> None: self.assertEqual(_torrent_progress({"progress": 0.1344}), 13.4) def test_episode_availability_counts_only_aired_monitored_episodes(self) -> None: episodes = [ {"seasonNumber": 1, "episodeNumber": 1, "monitored": True, "hasFile": True}, {"seasonNumber": 1, "episodeNumber": 2, "monitored": True, "hasFile": False}, {"seasonNumber": 1, "episodeNumber": 3, "monitored": False, "hasFile": False}, { "seasonNumber": 1, "episodeNumber": 4, "monitored": True, "hasFile": False, "airDateUtc": "2999-01-01T00:00:00Z", }, ] availability = _episode_availability(episodes) self.assertEqual(availability["available"], 1) self.assertEqual(availability["missing"], 1) self.assertEqual(availability["total"], 2) def test_presentation_hides_download_without_download_evidence(self) -> None: snapshot = Snapshot( request_id="3909", title="Example", request_type=RequestType.tv, state=NormalizedState.added_to_arr, actions=[], ) presentation = _build_presentation( snapshot, approved=True, arr_state="added", arr_details={ "availability": {"available": 0, "missing": 6, "total": 6, "seasons": []} }, prowlarr_state="ok", download={ "visible": False, "state": "not_started", "summary": "No download attempt has been observed.", "torrents": [], }, jellyfin_found=False, jellyfin_link=None, ) self.assertFalse(presentation["download"]["visible"]) self.assertIn("waiting for 6 episodes", presentation["status"]["label"]) download_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "download") self.assertEqual(download_stage["summary"], "No download attempt yet") def test_available_content_replaces_stale_download_warning_with_completion(self) -> None: snapshot = Snapshot( request_id="3909", title="Example", request_type=RequestType.tv, state=NormalizedState.completed, actions=[], ) presentation = _build_presentation( snapshot, approved=True, arr_state="available", arr_details={ "availability": {"available": 6, "missing": 0, "total": 6, "seasons": []} }, prowlarr_state="ok", download={ "visible": True, "state": "missing", "summary": "A previous download was observed, but it is not currently visible in qBittorrent.", "torrents": [], }, jellyfin_found=True, jellyfin_link="https://media.test/title/3909", ) self.assertFalse(presentation["download"]["visible"]) self.assertEqual(presentation["download"]["state"], "completed") self.assertEqual(presentation["nextStep"]["title"], "Ready to watch") self.assertEqual(presentation["nextStep"]["actionIds"], []) download_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "download") self.assertEqual(download_stage["label"], "Download complete") self.assertEqual(download_stage["state"], "complete") self.assertFalse(download_stage["visible"]) self.assertEqual( download_stage["summary"], "The requested content has been collected and is available to watch. No further action is needed.", ) search_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "search") self.assertEqual(search_stage["state"], "complete") self.assertEqual(search_stage["actionIds"], []) def test_partially_available_content_keeps_missing_download_attention(self) -> None: snapshot = Snapshot( request_id="3909", title="Example", request_type=RequestType.tv, state=NormalizedState.importing, actions=[], ) presentation = _build_presentation( snapshot, approved=True, arr_state="added", arr_details={ "availability": {"available": 3, "missing": 3, "total": 6, "seasons": []} }, prowlarr_state="ok", download={ "visible": True, "state": "missing", "summary": "A previous download is no longer visible.", "torrents": [], }, jellyfin_found=True, jellyfin_link="https://media.test/title/3909", ) self.assertTrue(presentation["download"]["visible"]) download_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "download") self.assertEqual(download_stage["label"], "Download") self.assertEqual(download_stage["state"], "attention") self.assertTrue(download_stage["visible"]) def test_collector_file_waits_for_media_server_before_marking_available(self) -> None: snapshot = Snapshot( request_id="3914", title="I See You", request_type=RequestType.movie, state=NormalizedState.importing, actions=[], ) presentation = _build_presentation( snapshot, approved=True, arr_state="available", arr_details={ "availability": {"available": 1, "missing": 0, "total": 1, "seasons": []} }, prowlarr_state="ok", download={ "visible": False, "state": "not_started", "summary": "No download attempt has been observed.", "torrents": [], }, jellyfin_found=False, jellyfin_link=None, ) self.assertEqual(presentation["status"]["label"], "Collected — waiting for the media server") self.assertEqual(presentation["nextStep"]["title"], "Wait for the media server to index this title") search_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "search") download_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "download") available_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "available") self.assertEqual(search_stage["state"], "complete") self.assertEqual(download_stage["state"], "complete") self.assertEqual(available_stage["state"], "active") class RequestCreationFlowTests(unittest.IsolatedAsyncioTestCase): def test_sparse_seerr_request_is_enriched_with_media_lookup(self) -> None: sparse = { "id": 3925, "type": "movie", "status": 2, "media": {"id": 2444, "mediaType": "movie", "tmdbId": 209112}, } details = { "title": "Batman v Superman: Dawn of Justice", "releaseDate": "2016-03-23", "posterPath": "/poster.jpg", } enriched = requests_router._merge_request_media_details(sparse, details) parsed = requests_router._parse_request_payload(enriched) self.assertEqual(parsed["title"], "Batman v Superman: Dawn of Justice") self.assertEqual(parsed["year"], 2016) self.assertEqual(enriched["media"]["posterPath"], "/poster.jpg") self.assertNotIn("title", sparse["media"]) async def test_seerr_search_percent_encodes_multi_word_titles(self) -> None: client = requests_router.JellyseerrClient("http://seerr.test", "key") client.get = AsyncMock(return_value={"results": []}) await client.search("Ricky Gervais Alley Cats", page=2) client.get.assert_awaited_once_with( "/api/v1/search?query=Ricky%20Gervais%20Alley%20Cats&page=2" ) async def test_seerr_request_includes_validated_destination_and_profile(self) -> None: client = requests_router.JellyseerrClient("http://seerr.test", "key") client.post = AsyncMock(return_value={"id": 42}) await client.create_request( media_type="tv", media_id=123, seasons=[1, 2], server_id=0, profile_id=7, root_folder="/TV98", ) client.post.assert_awaited_once_with( "/api/v1/request", payload={ "mediaType": "tv", "mediaId": 123, "seasons": [1, 2], "serverId": 0, "profileId": 7, "rootFolder": "/TV98", }, ) async def test_seerr_write_completes_csrf_cookie_handshake(self) -> None: observed: list[httpx.Request] = [] def handler(request: httpx.Request) -> httpx.Response: observed.append(request) if request.method == "GET": return httpx.Response( 200, headers=[ ("set-cookie", "_csrf=secret-value; Path=/; Secure; HttpOnly; SameSite=Strict"), ("set-cookie", "XSRF-TOKEN=csrf%2Etoken; Path=/; Secure; SameSite=Strict"), ], json={"id": 1}, ) return httpx.Response(201, json={"id": 42}) transport = httpx.MockTransport(handler) seerr = requests_router.JellyseerrClient("https://seerr.test", "api-key") async with httpx.AsyncClient(transport=transport) as client: response = await seerr._send_request( client, "POST", "https://seerr.test/api/v1/request", headers=seerr.headers(), params=None, payload={"mediaType": "movie", "mediaId": 209112}, ) self.assertEqual(response.status_code, 201) self.assertEqual([request.method for request in observed], ["GET", "POST"]) write_request = observed[1] self.assertEqual(write_request.headers.get("XSRF-TOKEN"), "csrf.token") self.assertEqual(write_request.headers.get("Origin"), "https://seerr.test") self.assertIn("_csrf=secret-value", write_request.headers.get("Cookie", "")) self.assertIn("XSRF-TOKEN=csrf%2Etoken", write_request.headers.get("Cookie", "")) async def test_base_request_passes_payload_to_transport_hook(self) -> None: captured: dict = {} async def send_request( _client: httpx.AsyncClient, method: str, url: str, *, headers: dict, params: dict | None, payload: dict | None, ) -> httpx.Response: captured.update( method=method, url=url, headers=headers, params=params, payload=payload, ) return httpx.Response(200, request=httpx.Request(method, url), json={"ok": True}) client = requests_router.JellyseerrClient("https://seerr.test", "api-key") with patch.object(client, "_send_request", new=send_request): result = await client._request( "POST", "/api/v1/request", payload={"mediaType": "movie", "mediaId": 209112}, ) self.assertEqual(result, {"ok": True}) self.assertEqual(captured["payload"], {"mediaType": "movie", "mediaId": 209112}) async def test_request_destination_only_offers_live_sonarr_profiles(self) -> None: runtime = SimpleNamespace( sonarr_base_url="http://sonarr.test", sonarr_api_key="key", sonarr_quality_profile_id=7, sonarr_root_folder="/tv", ) seerr = SimpleNamespace( get_service_settings=AsyncMock( return_value=[ { "id": 4, "name": "Main Sonarr", "isDefault": True, "is4k": False, "activeProfileId": 7, "activeDirectory": "/tv", } ] ) ) sonarr = SimpleNamespace( configured=lambda: True, get_quality_profiles=AsyncMock( return_value=[{"id": 7, "name": "WEB-1080p"}, {"id": 10, "name": "Optimal"}] ), get_root_folders=AsyncMock(return_value=[{"id": 1, "path": "/tv"}]), ) with patch.object(requests_router, "SonarrClient", return_value=sonarr): destination = await requests_router._resolve_request_destination( runtime, seerr, "tv", requested_profile_id=10 ) self.assertEqual(destination["profile_id"], 10) self.assertEqual(destination["default_profile_id"], 7) self.assertEqual(destination["root_folder"], "/tv") self.assertEqual(destination["profiles"], [ {"id": 7, "name": "WEB-1080p"}, {"id": 10, "name": "Optimal"}, ]) async def test_request_destination_rejects_stale_profile_id(self) -> None: runtime = SimpleNamespace( radarr_base_url="http://radarr.test", radarr_api_key="key", radarr_quality_profile_id=6, radarr_root_folder="/movies", ) seerr = SimpleNamespace( get_service_settings=AsyncMock( return_value=[ { "id": 2, "name": "Main Radarr", "isDefault": True, "is4k": False, "activeProfileId": 6, "activeDirectory": "/movies", } ] ) ) radarr = SimpleNamespace( configured=lambda: True, get_quality_profiles=AsyncMock(return_value=[{"id": 6, "name": "HD"}]), get_root_folders=AsyncMock(return_value=[{"id": 1, "path": "/movies"}]), ) with patch.object(requests_router, "RadarrClient", return_value=radarr): with self.assertRaises(HTTPException) as context: await requests_router._resolve_request_destination( runtime, seerr, "movie", requested_profile_id=999 ) self.assertEqual(context.exception.status_code, 400) self.assertIn("not available in Radarr", context.exception.detail) class RequestRecheckTests(unittest.IsolatedAsyncioTestCase): async def test_recheck_refreshes_seerr_cache_and_returns_rebuilt_snapshot(self) -> None: runtime = SimpleNamespace( jellyseerr_base_url="http://seerr.test", jellyseerr_api_key="seerr-key", ) fresh_request = { "id": 3914, "type": "movie", "status": 2, "createdAt": "2026-08-30T00:00:00Z", "updatedAt": "2026-08-30T01:00:00Z", "requestedBy": {"username": "viewer"}, "media": { "id": 9001, "mediaType": "movie", "tmdbId": 524251, "title": "I See You", "year": 2019, }, } seerr = SimpleNamespace( configured=lambda: True, get_request=AsyncMock(return_value=fresh_request), ) snapshot = Snapshot( request_id="3914", title="I See You", request_type=RequestType.movie, state=NormalizedState.importing, presentation={ "status": {"label": "Collected — waiting for the media server"}, }, ) with patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object( requests_router, "JellyseerrClient", return_value=seerr ), patch.object(requests_router, "upsert_request_cache") as upsert, patch.object( requests_router, "_cache_set" ) as cache_set, patch.object(requests_router, "_refresh_recent_cache_from_db"), patch.object( requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot) ), patch.object(requests_router, "save_action") as save_action: result = await requests_router.action_recheck( "3914", user={"username": "viewer", "role": "user"} ) seerr.get_request.assert_awaited_once_with("3914") upsert.assert_called_once() cache_set.assert_called_once_with("request:3914", fresh_request) save_action.assert_called_once() self.assertEqual(result["status"], "ok") self.assertIs(result["snapshot"], snapshot) async def test_recheck_hydrates_sparse_seerr_request_before_caching(self) -> None: runtime = SimpleNamespace(jellyseerr_base_url="http://seerr.test", jellyseerr_api_key="key") sparse_request = { "id": 3925, "type": "movie", "status": 2, "requestedBy": {"username": "viewer"}, "media": {"id": 2444, "mediaType": "movie", "tmdbId": 209112}, } seerr = SimpleNamespace( configured=lambda: True, get_request=AsyncMock(return_value=sparse_request), get_movie=AsyncMock( return_value={ "title": "Batman v Superman: Dawn of Justice", "releaseDate": "2016-03-23", } ), ) snapshot = Snapshot( request_id="3925", title="Batman v Superman: Dawn of Justice", request_type=RequestType.movie, state=NormalizedState.downloading, presentation={"status": {"label": "Download in progress"}}, ) with patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object( requests_router, "JellyseerrClient", return_value=seerr ), patch.object( requests_router, "_get_media_details", new=AsyncMock( return_value={ "title": "Batman v Superman: Dawn of Justice", "releaseDate": "2016-03-23", } ), ), patch.object(requests_router, "upsert_request_cache") as upsert, patch.object( requests_router, "_cache_set" ) as cache_set, patch.object(requests_router, "_refresh_recent_cache_from_db"), patch.object( requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot) ), patch.object(requests_router, "save_action"): await requests_router.action_recheck( "3925", user={"username": "viewer", "role": "user"} ) cached_record = upsert.call_args.kwargs self.assertEqual(cached_record["title"], "Batman v Superman: Dawn of Justice") cached_payload = cache_set.call_args.args[1] self.assertEqual(cached_payload["media"]["title"], "Batman v Superman: Dawn of Justice") class SnapshotIdentityTests(unittest.TestCase): def test_radarr_identity_replaces_unknown_cached_title(self) -> None: snapshot = Snapshot(request_id="3925", title="Unknown", request_type=RequestType.movie) _apply_arr_identity( snapshot, {"title": "Batman v Superman: Dawn of Justice", "year": 2016}, ) self.assertEqual(snapshot.title, "Batman v Superman: Dawn of Justice") self.assertEqual(snapshot.year, 2016) class LiveDownloadProgressTests(unittest.IsolatedAsyncioTestCase): async def test_live_download_progress_uses_saved_hash_and_current_qbittorrent_value(self) -> None: runtime = SimpleNamespace( jellyseerr_base_url=None, jellyseerr_api_key=None, qbittorrent_base_url="http://qbittorrent.test", qbittorrent_username="magent", qbittorrent_password="secret", ) evidence = { "observed": True, "torrents": [{"hash": "abc123", "progress": 0.12}], } current = [{"hash": "abc123", "name": "Example", "progress": 0.1344, "state": "downloading"}] with patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object( requests_router, "get_request_download_evidence", return_value=evidence, ), patch.object( requests_router.QBittorrentClient, "get_torrents_by_hashes", new=AsyncMock(return_value=current), ) as get_torrents: result = await requests_router.get_download_progress( "3909", user={"username": "viewer", "role": "user"} ) get_torrents.assert_awaited_once_with("abc123") self.assertEqual(result["state"], "downloading") self.assertEqual(result["torrents"][0]["progressPercent"], 13.4) class ArrAddPayloadTests(unittest.IsolatedAsyncioTestCase): async def test_radarr_add_resolves_title_before_posting_movie(self) -> None: client = requests_router.RadarrClient("http://radarr.test", "radarr-key") with patch.object( client, "get", new=AsyncMock(return_value={"title": "A Grand Day Out", "tmdbId": 530}), ) as lookup, patch.object( client, "post", new=AsyncMock(return_value={"id": 12, "title": "A Grand Day Out"}), ) as create: result = await client.add_movie(530, 1, "/movies") lookup.assert_awaited_once_with("/api/v3/movie/lookup/tmdb", params={"tmdbId": 530}) payload = create.await_args.kwargs["payload"] self.assertEqual(payload["title"], "A Grand Day Out") self.assertEqual(payload["tmdbId"], 530) self.assertEqual(result["id"], 12) async def test_sonarr_add_resolves_matching_series_title_before_posting(self) -> None: client = requests_router.SonarrClient("http://sonarr.test", "sonarr-key") lookup_response = [ {"title": "Wrong Show", "tvdbId": 111}, {"title": "Example Show", "tvdbId": 222}, ] with patch.object( client, "get", new=AsyncMock(return_value=lookup_response), ) as lookup, patch.object( client, "post", new=AsyncMock(return_value={"id": 42, "title": "Example Show"}), ) as create: result = await client.add_series(222, 2, "/television") lookup.assert_awaited_once_with("/api/v3/series/lookup", params={"term": "tvdb:222"}) payload = create.await_args.kwargs["payload"] self.assertEqual(payload["title"], "Example Show") self.assertEqual(payload["tvdbId"], 222) self.assertEqual(result["id"], 42) def test_arr_error_message_does_not_expose_upstream_stack_trace(self) -> None: response = httpx.Response( 500, request=httpx.Request("POST", "http://radarr.test/api/v3/movie"), json={ "message": "Object reference not set to an instance of an object.", "description": "System.NullReferenceException\n at Radarr.Internal.SecretMethod()", }, ) error = httpx.HTTPStatusError("Radarr failed", request=response.request, response=response) message = requests_router._format_upstream_error("Radarr", error) self.assertIn("Object reference", message) self.assertNotIn("NullReferenceException", message) self.assertNotIn("SecretMethod", message) class CollectorManualDownloadTests(unittest.IsolatedAsyncioTestCase): @staticmethod def _runtime() -> SimpleNamespace: return SimpleNamespace( jellyseerr_base_url=None, jellyseerr_api_key=None, sonarr_base_url="http://sonarr.test", sonarr_api_key="sonarr-key", radarr_base_url="http://radarr.test", radarr_api_key="radarr-key", ) async def test_tv_manual_search_uses_sonarr_and_keeps_season_packs(self) -> None: snapshot = Snapshot( request_id="3909", title="Example Show", request_type=RequestType.tv, raw={"arr": {"item": {"id": 42}}}, ) sonarr = SimpleNamespace( configured=lambda: True, get_episodes=AsyncMock( return_value=[ {"id": 101, "seasonNumber": 1, "monitored": True, "hasFile": False}, {"id": 201, "seasonNumber": 2, "monitored": True, "hasFile": False}, {"id": 202, "seasonNumber": 2, "monitored": True, "hasFile": True}, ] ), search_releases=AsyncMock( side_effect=[ [ { "title": "Example.Show.S01.1080p", "guid": "season-one", "indexerId": 7, "indexer": "Prowlarr", "protocol": "torrent", "fullSeason": True, "seasonNumber": 1, } ], [], ] ), ) with patch.object(requests_router, "get_runtime_settings", return_value=self._runtime()), patch.object( requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot) ), patch.object(requests_router, "SonarrClient", return_value=sonarr), patch.object( requests_router, "save_action" ): result = await requests_router.action_search( "3909", user={"username": "viewer", "role": "user"} ) sonarr.search_releases.assert_any_await(42, 1) sonarr.search_releases.assert_any_await(42, 2) self.assertEqual(result["collector"], "Sonarr") self.assertTrue(result["releases"][0]["fullSeason"]) self.assertEqual(result["releases"][0]["seasonNumber"], 1) async def test_movie_manual_search_uses_radarr(self) -> None: snapshot = Snapshot( request_id="4000", title="Example Movie", request_type=RequestType.movie, raw={"arr": {"item": {"id": 84}}}, ) radarr = SimpleNamespace( configured=lambda: True, search_releases=AsyncMock( return_value=[ { "title": "Example.Movie.2026.1080p", "guid": "movie-release", "indexerId": 9, "indexer": "Prowlarr", "protocol": "torrent", } ] ), ) with patch.object(requests_router, "get_runtime_settings", return_value=self._runtime()), patch.object( requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot) ), patch.object(requests_router, "RadarrClient", return_value=radarr), patch.object( requests_router, "save_action" ): result = await requests_router.action_search( "4000", user={"username": "viewer", "role": "user"} ) radarr.search_releases.assert_awaited_once_with(84) self.assertEqual(result["collector"], "Radarr") self.assertEqual(result["releases"][0]["guid"], "movie-release") async def test_tv_manual_grab_is_sent_to_sonarr_not_qbittorrent(self) -> None: snapshot = Snapshot( request_id="3909", title="Example Show", request_type=RequestType.tv, ) sonarr = SimpleNamespace( configured=lambda: True, grab_release=AsyncMock(return_value={"guid": "season-one", "indexerId": 7}), push_release=AsyncMock(), ) payload = { "title": "Example.Show.S01.1080p", "guid": "season-one", "indexerId": 7, "protocol": "torrent", } with patch.object(requests_router, "get_runtime_settings", return_value=self._runtime()), patch.object( requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot) ), patch.object(requests_router, "SonarrClient", return_value=sonarr), patch.object( requests_router, "save_action" ): result = await requests_router.action_grab( "3909", payload, user={"username": "viewer", "role": "user"} ) sonarr.grab_release.assert_awaited_once_with("season-one", 7) sonarr.push_release.assert_not_awaited() self.assertEqual(result["response"], {"collector": "Sonarr", "queued": True}) async def test_stale_movie_release_still_routes_through_radarr_push(self) -> None: snapshot = Snapshot( request_id="4000", title="Example Movie", request_type=RequestType.movie, ) response = httpx.Response( 404, request=httpx.Request("POST", "http://radarr.test/api/v3/release"), json={"message": "release cache expired"}, ) cache_miss = httpx.HTTPStatusError( "release cache expired", request=response.request, response=response, ) radarr = SimpleNamespace( configured=lambda: True, grab_release=AsyncMock(side_effect=cache_miss), push_release=AsyncMock(return_value=[{"approved": True, "downloadAllowed": True}]), ) payload = { "title": "Example.Movie.2026.1080p", "guid": "stale-release", "indexerId": 9, "indexer": "Prowlarr", "protocol": "torrent", "publishDate": "2026-08-29T00:00:00Z", "downloadUrl": "http://prowlarr.test/download/1", } with patch.object(requests_router, "get_runtime_settings", return_value=self._runtime()), patch.object( requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot) ), patch.object(requests_router, "RadarrClient", return_value=radarr), patch.object( requests_router, "save_action" ): result = await requests_router.action_grab( "4000", payload, user={"username": "viewer", "role": "user"} ) radarr.push_release.assert_awaited_once() pushed = radarr.push_release.await_args.args[0] self.assertEqual(pushed["downloadUrl"], "http://prowlarr.test/download/1") self.assertEqual(pushed["protocol"], "torrent") self.assertEqual(result["response"], {"collector": "Radarr", "queued": True}) class DatabaseEmailTests(TempDatabaseMixin, unittest.TestCase): def test_set_user_email_is_case_insensitive(self) -> None: created = db.create_user_if_missing( "MixedCaseUser", "password123", email=None, auth_provider="local", ) self.assertTrue(created) updated = db.set_user_email("mixedcaseuser", "mixed@example.com") self.assertTrue(updated) stored = db.get_user_by_username("MIXEDCASEUSER") self.assertIsNotNone(stored) self.assertEqual(stored.get("email"), "mixed@example.com") class SnapshotHistoryTests(TempDatabaseMixin, unittest.TestCase): def test_duplicate_snapshots_are_not_saved_and_download_evidence_is_retained(self) -> None: snapshot = Snapshot( request_id="3909", title="Example", request_type=RequestType.tv, state=NormalizedState.downloading, state_reason="Downloading one episode.", timeline=[ TimelineHop( service="qBittorrent", status="downloading", details={ "summary": "Downloading one item.", "torrents": [{"hash": "abc", "progress": 0.5}], }, ) ], ) db.save_snapshot(snapshot) db.save_snapshot(snapshot) history = db.get_recent_snapshots("3909", 10) evidence = db.get_request_download_evidence("3909") self.assertEqual(len(history), 1) self.assertTrue(evidence["observed"]) self.assertEqual(evidence["state"], "downloading") class AuthFlowTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase): async def test_forgot_password_is_rate_limited(self) -> None: request = _build_request(ip="10.1.2.3") payload = {"identifier": "resetuser@example.com"} with patch.object(auth_router, "smtp_email_config_ready", return_value=(True, "")), patch.object( auth_router, "request_password_reset", new=AsyncMock(return_value={"status": "ok", "issued": False}), ): for _ in range(3): result = await auth_router.forgot_password(payload, request) self.assertEqual(result["status"], "ok") with self.assertRaises(HTTPException) as context: await auth_router.forgot_password(payload, request) self.assertEqual(context.exception.status_code, 429) self.assertEqual( context.exception.detail, "Too many password reset attempts. Try again shortly.", ) async def test_request_password_reset_prefers_local_user_email(self) -> None: db.create_user_if_missing( "ResetUser", "password123", email="local@example.com", auth_provider="local", ) with patch.object( password_reset, "send_password_reset_email", new=AsyncMock(return_value={"status": "ok"}), ) as send_email: result = await password_reset.request_password_reset("ResetUser") self.assertTrue(result["issued"]) self.assertEqual(result["recipient_email"], "local@example.com") send_email.assert_awaited_once() self.assertEqual(send_email.await_args.kwargs["recipient_email"], "local@example.com") async def test_profile_invite_requires_recipient_email(self) -> None: current_user = { "username": "invite-owner", "role": "user", "invite_management_enabled": True, "profile_id": None, } with self.assertRaises(HTTPException) as context: await auth_router.create_profile_invite({"label": "Missing email"}, current_user) self.assertEqual(context.exception.status_code, 400) self.assertEqual( context.exception.detail, "recipient_email is required and must be a valid email address.", ) class MediaReplacementTests(unittest.IsolatedAsyncioTestCase): async def test_movie_replacement_validates_file_then_deletes_and_searches(self) -> None: snapshot = Snapshot( request_id="3914", title="Replacement Movie", request_type=RequestType.movie, state=NormalizedState.available, raw={ "arr": { "item": { "id": 44, "movieFile": { "id": 77, "relativePath": "Replacement.Movie.1080p.mkv", }, } } }, ) radarr = SimpleNamespace( configured=lambda: True, delete_movie_file=AsyncMock(return_value=None), search=AsyncMock(return_value={"id": 1}), ) runtime = SimpleNamespace( jellyseerr_base_url="http://seerr", jellyseerr_api_key="secret", radarr_base_url="http://radarr", radarr_api_key="secret", ) with ( patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object(requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot)), patch.object(requests_router, "RadarrClient", return_value=radarr), patch.object(requests_router, "save_action"), ): result = await requests_router.action_replace_media( "3914", {"file_id": 77, "confirmed": True}, {"username": "admin", "role": "admin", "auto_search_enabled": True}, ) self.assertEqual(result["status"], "ok") radarr.delete_movie_file.assert_awaited_once_with(77) radarr.search.assert_awaited_once_with(44) async def test_tv_replacement_options_return_only_safe_file_details(self) -> None: snapshot = Snapshot( request_id="3909", title="Replacement Series", request_type=RequestType.tv, state=NormalizedState.available, raw={"arr": {"item": {"id": 22}}}, ) sonarr = SimpleNamespace( configured=lambda: True, get_episode_files=AsyncMock(return_value=[{ "id": 88, "seasonNumber": 2, "path": "/private/library/Replacement.Series.S02E03.mkv", "size": 1024, "quality": {"quality": {"name": "WEBDL-1080p"}}, }]), get_episodes=AsyncMock(return_value=[{ "id": 101, "episodeFileId": 88, "seasonNumber": 2, "episodeNumber": 3, }]), ) runtime = SimpleNamespace( jellyseerr_base_url="http://seerr", jellyseerr_api_key="secret", sonarr_base_url="http://sonarr", sonarr_api_key="secret", ) with ( patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object(requests_router, "build_snapshot", new=AsyncMock(return_value=snapshot)), patch.object(requests_router, "SonarrClient", return_value=sonarr), ): result = await requests_router.replacement_options( "3909", {"username": "viewer", "role": "user", "auto_search_enabled": True}, ) self.assertEqual(result["files"][0]["name"], "Replacement.Series.S02E03.mkv") self.assertEqual(result["files"][0]["episodes"], ["S02E03"]) self.assertNotIn("/private/library", str(result)) class PortalMediaStatusTests(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: portal_router._MEDIA_STATUS_CACHE.update(expires_at=0.0, payload=None) async def test_media_status_is_live_and_removes_session_identity(self) -> None: client = SimpleNamespace( configured=lambda: True, get_system_info=AsyncMock( return_value={ "Version": "10.10.7", "HasPendingRestart": False, "WanAddress": "https://private.example", } ), get_sessions=AsyncMock( return_value=[ { "UserName": "private-user", "DeviceName": "Living room television", "NowPlayingItem": {"Name": "Private title"}, "TranscodingInfo": {"VideoCodec": "h264"}, } ] ), ) with ( patch.object(portal_router, "get_runtime_settings", return_value=SimpleNamespace( jellyfin_base_url="http://jellyfin", jellyfin_api_key="secret", )), patch.object(portal_router, "JellyfinClient", return_value=client), ): result = await portal_router.portal_media_status() self.assertEqual(result["status"], "up") self.assertEqual(result["activity"]["active_streams"], 1) self.assertEqual(result["activity"]["transcoding_streams"], 1) serialized = str(result) self.assertNotIn("private-user", serialized) self.assertNotIn("Living room television", serialized) self.assertNotIn("Private title", serialized) self.assertNotIn("private.example", serialized) async def test_media_status_reports_unavailable_without_exposing_exception(self) -> None: client = SimpleNamespace( configured=lambda: True, get_system_info=AsyncMock(side_effect=RuntimeError("secret upstream failure")), ) with ( patch.object(portal_router, "get_runtime_settings", return_value=SimpleNamespace( jellyfin_base_url="http://jellyfin", jellyfin_api_key="secret", )), patch.object(portal_router, "JellyfinClient", return_value=client), ): result = await portal_router.portal_media_status() self.assertEqual(result["status"], "down") self.assertNotIn("secret upstream failure", str(result)) class PortalWorkflowTests(TempDatabaseMixin, unittest.TestCase): def test_legacy_request_status_maps_to_workflow(self) -> None: item = {"kind": "request", "status": "in_progress"} serialized = portal_router._serialize_item(item, {"username": "tester", "role": "user"}) workflow = serialized.get("workflow") or {} self.assertEqual(workflow.get("request_status"), "approved") self.assertEqual(workflow.get("media_status"), "processing") def test_invalid_pipeline_transition_is_rejected(self) -> None: with self.assertRaises(HTTPException) as context: portal_router._validate_pipeline_transition( "approved", "processing", "pending", "pending", ) self.assertEqual(context.exception.status_code, 400) def test_portal_workflow_filters(self) -> None: db.create_portal_item( kind="request", title="Request A", description="A", created_by_username="alpha", created_by_id=None, status="processing", workflow_request_status="approved", workflow_media_status="processing", ) db.create_portal_item( kind="request", title="Request B", description="B", created_by_username="bravo", created_by_id=None, status="pending", workflow_request_status="pending", workflow_media_status="pending", ) processing = db.list_portal_items( kind="request", workflow_request_status="approved", workflow_media_status="processing", limit=10, offset=0, ) pending_count = db.count_portal_items( kind="request", workflow_request_status="pending", workflow_media_status="pending", ) self.assertEqual(len(processing), 1) self.assertEqual(pending_count, 1)