1134 lines
45 KiB
Python
1134 lines
45 KiB
Python
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.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 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 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")
|
|
|
|
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, "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 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)
|