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.config import settings from backend.app.network_security import request_trusts_forwarded_headers, validate_notification_target_url from backend.app.models import 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.snapshot import _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): 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 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 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") 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 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 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)