2320 lines
94 KiB
Python
2320 lines
94 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.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 admin as admin_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 import issue_resolution
|
|
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,
|
|
_build_repair_activity,
|
|
_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_bazarr_subtitle_search_is_explained_in_plain_english(self) -> None:
|
|
message = _operation_result_message(
|
|
"Bazarr",
|
|
"PATCH",
|
|
"/api/episodes/subtitles",
|
|
{"status": True},
|
|
params={"language": "en", "episodeid": 42},
|
|
)
|
|
|
|
self.assertEqual(
|
|
message,
|
|
"Bazarr accepted a fresh EN subtitle search for the selected episode.",
|
|
)
|
|
|
|
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(TempDatabaseMixin, 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_repair_activity_shows_collector_search_before_download(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="144",
|
|
title="Toy Story 2",
|
|
request_type=RequestType.movie,
|
|
state=NormalizedState.searching,
|
|
)
|
|
|
|
activity = _build_repair_activity(
|
|
snapshot,
|
|
action={
|
|
"action_id": "replace_media",
|
|
"status": "ok",
|
|
"message": "Radarr removed the file and started a replacement search.",
|
|
"created_at": "2026-09-01T09:06:55+00:00",
|
|
},
|
|
arr_state="searching",
|
|
arr_details={"availability": {"available": 0, "missing": 1, "total": 1}},
|
|
download={"visible": False, "state": "not_started", "torrents": []},
|
|
jellyfin_found=False,
|
|
)
|
|
|
|
self.assertIsNotNone(activity)
|
|
self.assertEqual(activity["state"], "searching")
|
|
self.assertEqual(activity["headline"], "Replacement search in progress")
|
|
self.assertEqual(activity["steps"][1]["state"], "complete")
|
|
self.assertEqual(activity["steps"][2]["state"], "waiting")
|
|
|
|
def test_repair_activity_tracks_replacement_download(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="144",
|
|
title="Toy Story 2",
|
|
request_type=RequestType.movie,
|
|
state=NormalizedState.downloading,
|
|
)
|
|
|
|
activity = _build_repair_activity(
|
|
snapshot,
|
|
action={
|
|
"action_id": "replace_media",
|
|
"status": "ok",
|
|
"message": "Radarr started a replacement search.",
|
|
"created_at": "2026-09-01T09:06:55+00:00",
|
|
},
|
|
arr_state="searching",
|
|
arr_details={"availability": {"available": 0, "missing": 1, "total": 1}},
|
|
download={
|
|
"visible": True,
|
|
"state": "downloading",
|
|
"summary": "Downloading (1 active).",
|
|
"torrents": [{"progress": 0.25}],
|
|
},
|
|
jellyfin_found=False,
|
|
)
|
|
|
|
self.assertIsNotNone(activity)
|
|
self.assertEqual(activity["state"], "downloading")
|
|
self.assertEqual(activity["headline"], "Replacement download in progress")
|
|
self.assertEqual(activity["steps"][2]["state"], "active")
|
|
|
|
def test_repair_activity_reports_collected_file_and_media_index(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="144",
|
|
title="Toy Story 2",
|
|
request_type=RequestType.movie,
|
|
state=NormalizedState.importing,
|
|
)
|
|
|
|
activity = _build_repair_activity(
|
|
snapshot,
|
|
action={
|
|
"action_id": "replace_media",
|
|
"status": "ok",
|
|
"message": "Radarr started a replacement search.",
|
|
"created_at": "2026-09-01T09:06:55+00:00",
|
|
},
|
|
arr_state="available",
|
|
arr_details={"availability": {"available": 1, "missing": 0, "total": 1}},
|
|
download={"visible": False, "state": "not_started", "torrents": []},
|
|
jellyfin_found=False,
|
|
)
|
|
|
|
self.assertIsNotNone(activity)
|
|
self.assertEqual(activity["state"], "indexing")
|
|
self.assertEqual(activity["steps"][2]["state"], "complete")
|
|
self.assertEqual(
|
|
activity["steps"][2]["detail"],
|
|
"Radarr reports the replacement file as collected and imported.",
|
|
)
|
|
self.assertEqual(activity["steps"][3]["state"], "active")
|
|
|
|
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")
|
|
available_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "available")
|
|
self.assertEqual(search_stage["state"], "complete")
|
|
self.assertEqual(search_stage["actionIds"], [])
|
|
self.assertEqual(available_stage["state"], "complete")
|
|
self.assertEqual(available_stage["stateLabel"], "Ready")
|
|
self.assertEqual(available_stage["label"], "Available to watch")
|
|
self.assertEqual(available_stage["summary"], "This title is ready to watch in Grizzlyflix.")
|
|
self.assertEqual(available_stage["link"], "https://media.test/title/3909")
|
|
|
|
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")
|
|
self.assertEqual(available_stage["stateLabel"], "Indexing")
|
|
self.assertEqual(available_stage["label"], "Adding to Grizzlyflix")
|
|
self.assertEqual(
|
|
available_stage["summary"],
|
|
"The download is complete. Grizzlyflix is indexing this title now.",
|
|
)
|
|
|
|
|
|
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(TempDatabaseMixin, 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,
|
|
"approved": True,
|
|
"rejected": False,
|
|
"downloadAllowed": True,
|
|
"quality": {"quality": {"name": "WEBDL-1080p"}},
|
|
},
|
|
{
|
|
"title": "Example.Show.S01.2160p",
|
|
"guid": "outside-profile",
|
|
"indexerId": 7,
|
|
"approved": False,
|
|
"rejected": True,
|
|
"rejections": ["Quality is not wanted in profile"],
|
|
}
|
|
],
|
|
[],
|
|
]
|
|
),
|
|
)
|
|
|
|
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.assertEqual(len(result["releases"]), 1)
|
|
self.assertTrue(result["releases"][0]["fullSeason"])
|
|
self.assertEqual(result["releases"][0]["seasonNumber"], 1)
|
|
self.assertEqual(result["releases"][0]["quality"], "WEBDL-1080p")
|
|
self.assertTrue(result["releases"][0]["bestPick"])
|
|
self.assertTrue(result["qualityFiltered"])
|
|
|
|
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",
|
|
"approved": True,
|
|
"rejected": False,
|
|
"downloadAllowed": True,
|
|
}
|
|
]
|
|
),
|
|
)
|
|
|
|
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")
|
|
self.assertTrue(result["releases"][0]["bestPick"])
|
|
|
|
def test_manual_release_filter_requires_explicit_arr_approval(self) -> None:
|
|
releases = requests_router._filter_arr_release_results(
|
|
[
|
|
{"title": "Missing decision", "guid": "missing", "indexerId": 1},
|
|
{
|
|
"title": "Temporarily rejected",
|
|
"guid": "temporary",
|
|
"indexerId": 1,
|
|
"approved": True,
|
|
"temporarilyRejected": True,
|
|
},
|
|
{
|
|
"title": "Approved release",
|
|
"guid": "approved",
|
|
"indexerId": 1,
|
|
"approved": True,
|
|
"rejected": False,
|
|
"downloadAllowed": True,
|
|
},
|
|
]
|
|
)
|
|
|
|
self.assertEqual([release["guid"] for release in releases], ["approved"])
|
|
self.assertTrue(releases[0]["bestPick"])
|
|
|
|
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 AdminUserEmailTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
async def test_admin_can_add_and_remove_user_email(self) -> None:
|
|
db.create_user_if_missing("Viewer", "password123", auth_provider="local")
|
|
|
|
saved = await admin_router.update_user_email("viewer", {"email": "viewer@example.com"})
|
|
self.assertEqual(saved["user"]["email"], "viewer@example.com")
|
|
|
|
cleared = await admin_router.update_user_email("VIEWER", {"email": None})
|
|
self.assertIsNone(cleared["user"]["email"])
|
|
|
|
async def test_admin_cannot_assign_duplicate_user_email(self) -> None:
|
|
db.create_user_if_missing(
|
|
"FirstViewer", "password123", email="shared@example.com", auth_provider="local"
|
|
)
|
|
db.create_user_if_missing("SecondViewer", "password123", auth_provider="local")
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
await admin_router.update_user_email(
|
|
"SecondViewer", {"email": "SHARED@example.com"}
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 409)
|
|
self.assertIn("another user", str(context.exception.detail))
|
|
|
|
async def test_admin_user_email_requires_valid_address(self) -> None:
|
|
db.create_user_if_missing("Viewer", "password123", auth_provider="local")
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
await admin_router.update_user_email("Viewer", {"email": "not-an-email"})
|
|
|
|
self.assertEqual(context.exception.status_code, 400)
|
|
|
|
|
|
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_user_can_manage_own_profile_email(self) -> None:
|
|
db.create_user_if_missing("ProfileViewer", "password123", auth_provider="local")
|
|
current_user = {"username": "ProfileViewer", "role": "user"}
|
|
|
|
saved = await auth_router.update_profile_email(
|
|
{"email": "viewer@example.com"}, current_user
|
|
)
|
|
self.assertEqual(saved["email"], "viewer@example.com")
|
|
self.assertEqual(
|
|
db.get_user_by_username("profileviewer").get("email"),
|
|
"viewer@example.com",
|
|
)
|
|
|
|
cleared = await auth_router.update_profile_email({"email": None}, current_user)
|
|
self.assertIsNone(cleared["email"])
|
|
self.assertIsNone(db.get_user_by_username("ProfileViewer").get("email"))
|
|
|
|
async def test_user_cannot_claim_another_accounts_email(self) -> None:
|
|
db.create_user_if_missing(
|
|
"FirstViewer", "password123", email="shared@example.com", auth_provider="local"
|
|
)
|
|
db.create_user_if_missing("SecondViewer", "password123", auth_provider="local")
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
await auth_router.update_profile_email(
|
|
{"email": "SHARED@example.com"},
|
|
{"username": "SecondViewer", "role": "user"},
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 409)
|
|
|
|
async def test_profile_email_requires_valid_address(self) -> None:
|
|
db.create_user_if_missing("ProfileViewer", "password123", auth_provider="local")
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
await auth_router.update_profile_email(
|
|
{"email": "not-an-email"},
|
|
{"username": "ProfileViewer", "role": "user"},
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 400)
|
|
|
|
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_manual_invite_does_not_require_recipient_email(self) -> None:
|
|
current_user = {
|
|
"username": "invite-owner",
|
|
"role": "user",
|
|
"invite_management_enabled": True,
|
|
"profile_id": None,
|
|
}
|
|
result = await auth_router.create_profile_invite(
|
|
{"label": "Family", "recipient_email": None, "send_email": False},
|
|
current_user,
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
self.assertIsNone(result["invite"]["recipient_email"])
|
|
self.assertIsNone(result["email"])
|
|
|
|
async def test_profile_email_delivery_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", "send_email": True},
|
|
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(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
def test_failed_repair_marks_linked_issue_as_blocked(self) -> None:
|
|
issue = {"id": 12, "status": "in_progress"}
|
|
with (
|
|
patch.object(requests_router, "update_portal_item") as update_issue,
|
|
patch.object(requests_router, "add_portal_item_activity"),
|
|
):
|
|
requests_router._record_replacement_activity(
|
|
issue,
|
|
user={"username": "viewer", "role": "user"},
|
|
event_type="replacement_failed",
|
|
message="Radarr could not start the replacement.",
|
|
)
|
|
|
|
update_issue.assert_called_once_with(12, status="blocked", issue_resolved_at=None)
|
|
|
|
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,
|
|
monitor_movie=AsyncMock(return_value={"id": 44, "monitored": 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"),
|
|
patch.object(
|
|
requests_router,
|
|
"get_portal_item",
|
|
return_value={
|
|
"id": 12,
|
|
"kind": "issue",
|
|
"external_ref": "/requests/3914",
|
|
"created_by_username": "admin",
|
|
},
|
|
),
|
|
patch.object(requests_router, "update_portal_item") as update_issue,
|
|
patch.object(requests_router, "add_portal_item_activity") as add_activity,
|
|
):
|
|
result = await requests_router.action_replace_media(
|
|
"3914",
|
|
{"file_id": 77, "confirmed": True, "issue_id": 12},
|
|
{"username": "admin", "role": "admin", "auto_search_enabled": True},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
radarr.monitor_movie.assert_awaited_once_with(44, True)
|
|
radarr.delete_movie_file.assert_awaited_once_with(77)
|
|
radarr.search.assert_awaited_once_with(44)
|
|
add_activity.assert_called_once()
|
|
self.assertEqual(add_activity.call_args.kwargs["event_type"], "replacement_started")
|
|
self.assertIn('"repairTracking"', add_activity.call_args.kwargs["metadata_json"])
|
|
self.assertIn('"originalFileIds":[77]', add_activity.call_args.kwargs["metadata_json"])
|
|
update_issue.assert_called_once_with(12, status="in_progress", issue_resolved_at=None)
|
|
|
|
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))
|
|
|
|
async def test_issue_options_mark_released_missing_episodes_as_best_fit(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="3909",
|
|
title="Target Series",
|
|
request_type=RequestType.tv,
|
|
state=NormalizedState.downloading,
|
|
raw={"arr": {"item": {"id": 22}}},
|
|
)
|
|
sonarr = SimpleNamespace(
|
|
configured=lambda: True,
|
|
get_episodes=AsyncMock(return_value=[
|
|
{
|
|
"id": 101,
|
|
"seasonNumber": 1,
|
|
"episodeNumber": 1,
|
|
"title": "Collected",
|
|
"monitored": True,
|
|
"hasFile": True,
|
|
"episodeFileId": 88,
|
|
"airDateUtc": "2020-01-01T00:00:00Z",
|
|
},
|
|
{
|
|
"id": 102,
|
|
"seasonNumber": 1,
|
|
"episodeNumber": 2,
|
|
"title": "Missing",
|
|
"monitored": True,
|
|
"hasFile": False,
|
|
"episodeFileId": 0,
|
|
"airDateUtc": "2020-01-08T00:00:00Z",
|
|
},
|
|
{
|
|
"id": 103,
|
|
"seasonNumber": 1,
|
|
"episodeNumber": 3,
|
|
"title": "Missing and unmonitored",
|
|
"monitored": False,
|
|
"hasFile": False,
|
|
"episodeFileId": 0,
|
|
"airDateUtc": "2020-01-15T00:00:00Z",
|
|
},
|
|
]),
|
|
)
|
|
runtime = SimpleNamespace(
|
|
jellyseerr_base_url=None,
|
|
jellyseerr_api_key=None,
|
|
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.issue_target_options(
|
|
"3909",
|
|
{"username": "viewer", "role": "user", "auto_search_enabled": True},
|
|
)
|
|
|
|
self.assertEqual(result["seasons"][0]["missing_count"], 2)
|
|
self.assertTrue(result["seasons"][0]["best_fit"])
|
|
missing = next(item for item in result["episodes"] if item["id"] == 102)
|
|
self.assertTrue(missing["missing"])
|
|
self.assertTrue(missing["best_fit"])
|
|
unmonitored = next(item for item in result["episodes"] if item["id"] == 103)
|
|
self.assertFalse(unmonitored["monitored"])
|
|
self.assertTrue(unmonitored["missing"])
|
|
self.assertTrue(unmonitored["best_fit"])
|
|
collected = next(item for item in result["episodes"] if item["id"] == 101)
|
|
self.assertEqual(collected["file_id"], 88)
|
|
self.assertNotIn("file_name", collected)
|
|
self.assertNotIn("quality", collected)
|
|
|
|
async def test_tv_replacement_accepts_multiple_selected_episode_files(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, "relativePath": "S01E01.mkv"},
|
|
{"id": 89, "relativePath": "S01E02.mkv"},
|
|
]),
|
|
get_episodes=AsyncMock(return_value=[
|
|
{"id": 101, "episodeFileId": 88, "seasonNumber": 1, "episodeNumber": 1},
|
|
{"id": 102, "episodeFileId": 89, "seasonNumber": 1, "episodeNumber": 2},
|
|
]),
|
|
monitor_episodes=AsyncMock(return_value={"monitored": True}),
|
|
delete_episode_file=AsyncMock(return_value=None),
|
|
search_episodes=AsyncMock(return_value={"id": 1}),
|
|
)
|
|
runtime = SimpleNamespace(
|
|
jellyseerr_base_url=None,
|
|
jellyseerr_api_key=None,
|
|
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),
|
|
patch.object(requests_router, "save_action"),
|
|
patch.object(requests_router, "get_portal_item", return_value={
|
|
"id": 12,
|
|
"kind": "issue",
|
|
"external_ref": "/requests/3909",
|
|
"created_by_username": "viewer",
|
|
}),
|
|
patch.object(requests_router, "update_portal_item"),
|
|
patch.object(requests_router, "add_portal_item_activity"),
|
|
):
|
|
result = await requests_router.action_replace_media(
|
|
"3909",
|
|
{"file_ids": [88, 89], "confirmed": True, "issue_id": 12},
|
|
{"username": "viewer", "role": "user", "auto_search_enabled": True},
|
|
)
|
|
|
|
self.assertEqual(result["file_ids"], [88, 89])
|
|
sonarr.monitor_episodes.assert_awaited_once_with([101, 102], True)
|
|
self.assertEqual(sonarr.delete_episode_file.await_count, 2)
|
|
sonarr.search_episodes.assert_awaited_once_with([101, 102])
|
|
|
|
async def test_missing_episode_search_monitors_explicit_unmonitored_episode(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="113",
|
|
title="Family Guy",
|
|
request_type=RequestType.tv,
|
|
state=NormalizedState.available,
|
|
raw={"arr": {"item": {"id": 540}}},
|
|
)
|
|
sonarr = SimpleNamespace(
|
|
configured=lambda: True,
|
|
get_episodes=AsyncMock(return_value=[{
|
|
"id": 36899,
|
|
"seasonNumber": 5,
|
|
"episodeNumber": 9,
|
|
"monitored": False,
|
|
"hasFile": False,
|
|
"episodeFileId": 0,
|
|
"airDateUtc": "2006-12-17T00:00:00Z",
|
|
}]),
|
|
monitor_episodes=AsyncMock(return_value={"monitored": True}),
|
|
search_episodes=AsyncMock(return_value={"id": 9001}),
|
|
search=AsyncMock(return_value={"id": 9002}),
|
|
)
|
|
runtime = SimpleNamespace(
|
|
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),
|
|
patch.object(requests_router, "save_action"),
|
|
):
|
|
result = await requests_router.action_search_missing_media(
|
|
"113",
|
|
{"episode_ids": [36899], "season_numbers": [5]},
|
|
{"username": "admin", "role": "admin", "auto_search_enabled": True},
|
|
)
|
|
|
|
self.assertEqual(result["episode_ids"], [36899])
|
|
sonarr.monitor_episodes.assert_awaited_once_with([36899], True)
|
|
sonarr.search_episodes.assert_awaited_once_with([36899])
|
|
sonarr.search.assert_not_awaited()
|
|
|
|
async def test_missing_movie_search_monitors_movie_before_search(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="3914",
|
|
title="Missing Movie",
|
|
request_type=RequestType.movie,
|
|
state=NormalizedState.available,
|
|
raw={"arr": {"item": {"id": 44}}},
|
|
)
|
|
radarr = SimpleNamespace(
|
|
configured=lambda: True,
|
|
monitor_movie=AsyncMock(return_value={"id": 44, "monitored": True}),
|
|
search=AsyncMock(return_value={"id": 9003}),
|
|
)
|
|
runtime = SimpleNamespace(
|
|
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_search_missing_media(
|
|
"3914",
|
|
{"episode_ids": [], "season_numbers": []},
|
|
{"username": "admin", "role": "admin", "auto_search_enabled": True},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
radarr.monitor_movie.assert_awaited_once_with(44, True)
|
|
radarr.search.assert_awaited_once_with(44)
|
|
|
|
async def test_movie_subtitle_issue_starts_bazarr_search_without_replacement(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="3914",
|
|
title="Subtitle Movie",
|
|
request_type=RequestType.movie,
|
|
state=NormalizedState.available,
|
|
raw={"arr": {"item": {"id": 44, "movieFile": {"id": 77}}}},
|
|
)
|
|
bazarr = SimpleNamespace(
|
|
configured=lambda: True,
|
|
search_movie_subtitles=AsyncMock(return_value={"status": True}),
|
|
)
|
|
runtime = SimpleNamespace(
|
|
bazarr_base_url="http://bazarr",
|
|
bazarr_api_key="secret",
|
|
bazarr_default_language="en",
|
|
)
|
|
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, "BazarrClient", return_value=bazarr),
|
|
patch.object(requests_router, "save_action"),
|
|
patch.object(requests_router, "get_portal_item", return_value={
|
|
"id": 12,
|
|
"kind": "issue",
|
|
"external_ref": "/requests/3914",
|
|
"created_by_username": "viewer",
|
|
}),
|
|
patch.object(requests_router, "update_portal_item") as update_issue,
|
|
patch.object(requests_router, "add_portal_item_activity") as add_activity,
|
|
):
|
|
result = await requests_router.action_repair_subtitles(
|
|
"3914",
|
|
{"issue_id": 12, "episode_ids": [], "forced": True},
|
|
{"username": "viewer", "role": "user", "auto_search_enabled": True},
|
|
)
|
|
|
|
self.assertEqual(result["status"], "ok")
|
|
bazarr.search_movie_subtitles.assert_awaited_once_with(44, language="en", forced=True)
|
|
self.assertEqual(add_activity.call_args.kwargs["event_type"], "subtitle_repair_started")
|
|
update_issue.assert_called_once_with(12, status="in_progress", issue_resolved_at=None)
|
|
|
|
|
|
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 InviteOperationalStateTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
async def test_manual_invite_can_be_created_without_recipient_email(self) -> None:
|
|
payload = await admin_router.create_invite(
|
|
{
|
|
"label": "The neighbour",
|
|
"recipient_email": None,
|
|
"send_email": False,
|
|
"max_uses": 1,
|
|
},
|
|
{"username": "admin", "role": "admin"},
|
|
)
|
|
|
|
self.assertEqual(payload["status"], "ok")
|
|
self.assertEqual(payload["invite"]["label"], "The neighbour")
|
|
self.assertIsNone(payload["invite"]["recipient_email"])
|
|
self.assertTrue(payload["invite"]["enabled"])
|
|
|
|
async def test_email_delivery_still_requires_valid_recipient(self) -> None:
|
|
with self.assertRaises(HTTPException) as context:
|
|
await admin_router.create_invite(
|
|
{"label": "Family", "send_email": True},
|
|
{"username": "admin", "role": "admin"},
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 400)
|
|
self.assertIn("required for email delivery", str(context.exception.detail))
|
|
|
|
async def test_invite_list_reports_automatic_operational_states(self) -> None:
|
|
ready = db.create_signup_invite(code="READY", recipient_email="ready@example.com")
|
|
db.create_signup_invite(code="DISABLED", enabled=False, recipient_email="off@example.com")
|
|
used = db.create_signup_invite(code="USED", max_uses=1, recipient_email="used@example.com")
|
|
db.increment_signup_invite_use(int(used["id"]))
|
|
db.create_signup_invite(
|
|
code="EXPIRED",
|
|
expires_at="2000-01-01T00:00:00+00:00",
|
|
recipient_email="expired@example.com",
|
|
)
|
|
db.create_signup_invite(
|
|
code="NO-PROFILE",
|
|
profile_id=999,
|
|
recipient_email="profile@example.com",
|
|
)
|
|
|
|
payload = await admin_router.get_invites()
|
|
states = {invite["code"]: invite["operational_state"] for invite in payload["invites"]}
|
|
|
|
self.assertEqual(states[ready["code"]], "ready")
|
|
self.assertEqual(states["DISABLED"], "disabled")
|
|
self.assertEqual(states["USED"], "exhausted")
|
|
self.assertEqual(states["EXPIRED"], "expired")
|
|
self.assertEqual(states["NO-PROFILE"], "profile_unavailable")
|
|
self.assertEqual(payload["summary"]["total"], 5)
|
|
self.assertEqual(payload["summary"]["ready"], 1)
|
|
self.assertEqual(payload["summary"]["attention"], 4)
|
|
self.assertEqual(payload["summary"]["used_signups"], 1)
|
|
|
|
|
|
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_issue_status_maps_to_public_workflow_progress(self) -> None:
|
|
item = {
|
|
"kind": "issue",
|
|
"status": "blocked",
|
|
"issue_type": "playback",
|
|
"created_by_username": "tester",
|
|
}
|
|
serialized = portal_router._serialize_item(item, {"username": "tester", "role": "user"})
|
|
workflow = (serialized.get("issue") or {}).get("workflow") or {}
|
|
|
|
self.assertEqual(workflow.get("current_step"), 4)
|
|
self.assertEqual(workflow.get("stage"), "repair")
|
|
self.assertEqual(workflow.get("state"), "attention")
|
|
self.assertEqual(len(workflow.get("steps") or []), 6)
|
|
self.assertEqual((workflow.get("steps") or [])[3].get("state"), "attention")
|
|
|
|
def test_resolved_issue_completes_the_public_workflow(self) -> None:
|
|
workflow = portal_router._issue_workflow_payload("closed")
|
|
|
|
self.assertEqual(workflow.get("current_step"), 6)
|
|
self.assertEqual(workflow.get("state"), "complete")
|
|
self.assertTrue(all(step.get("state") == "complete" for step in workflow.get("steps") or []))
|
|
|
|
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)
|
|
|
|
|
|
class PortalIssueDeletionTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
def _create_issue(self) -> dict:
|
|
issue = db.create_portal_item(
|
|
kind="issue",
|
|
title="Delete this issue",
|
|
description="No longer required",
|
|
created_by_username="reporter",
|
|
created_by_id=None,
|
|
issue_type="playback",
|
|
)
|
|
db.add_portal_comment(
|
|
int(issue["id"]),
|
|
author_username="reporter",
|
|
author_role="user",
|
|
message="Issue detail",
|
|
)
|
|
db.add_portal_item_activity(
|
|
int(issue["id"]),
|
|
event_type="item_created",
|
|
actor_username="reporter",
|
|
actor_role="user",
|
|
message="Issue created",
|
|
)
|
|
return issue
|
|
|
|
async def test_only_admin_can_delete_an_issue(self) -> None:
|
|
issue = self._create_issue()
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
await portal_router.portal_delete_item(
|
|
int(issue["id"]),
|
|
current_user={"username": "reporter", "role": "user"},
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 403)
|
|
self.assertIsNotNone(db.get_portal_item(int(issue["id"])))
|
|
|
|
async def test_delete_issue_removes_its_comments_and_activity(self) -> None:
|
|
issue = self._create_issue()
|
|
issue_id = int(issue["id"])
|
|
|
|
result = await portal_router.portal_delete_item(
|
|
issue_id,
|
|
current_user={"username": "admin", "role": "admin"},
|
|
)
|
|
|
|
self.assertEqual(result, {"status": "deleted", "item_id": issue_id})
|
|
self.assertIsNone(db.get_portal_item(issue_id))
|
|
self.assertEqual(db.list_portal_comments(issue_id), [])
|
|
self.assertEqual(db.list_portal_item_activity(issue_id), [])
|
|
|
|
async def test_delete_endpoint_will_not_delete_a_request(self) -> None:
|
|
request_item = db.create_portal_item(
|
|
kind="request",
|
|
title="Keep this request",
|
|
description="The media workflow must remain intact",
|
|
created_by_username="reporter",
|
|
created_by_id=None,
|
|
)
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
await portal_router.portal_delete_item(
|
|
int(request_item["id"]),
|
|
current_user={"username": "admin", "role": "admin"},
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 400)
|
|
self.assertIsNotNone(db.get_portal_item(int(request_item["id"])))
|
|
|
|
|
|
class IssueResolutionWorkflowTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
def _create_issue(self, *, status: str = "in_progress") -> dict:
|
|
return db.create_portal_item(
|
|
kind="issue",
|
|
title="Missing content: Test movie",
|
|
description="The title is missing.",
|
|
created_by_username="reporter",
|
|
created_by_id=None,
|
|
status=status,
|
|
issue_type="missing_content",
|
|
)
|
|
|
|
async def test_marking_issue_fixed_records_contact_and_confirmation(self) -> None:
|
|
issue = self._create_issue()
|
|
reporter = {"username": "reporter", "email": "reporter@example.com"}
|
|
with (
|
|
patch.object(issue_resolution, "_workflow_settings", return_value=(2, 3, "days")),
|
|
patch.object(issue_resolution, "get_user_by_username", return_value=reporter),
|
|
patch.object(issue_resolution, "resolve_user_delivery_email", return_value="reporter@example.com"),
|
|
patch.object(issue_resolution, "send_generic_email", new=AsyncMock(return_value=None)) as send_email,
|
|
):
|
|
waiting = await issue_resolution.begin_issue_confirmation(
|
|
int(issue["id"]),
|
|
actor_username="admin",
|
|
actor_role="admin",
|
|
)
|
|
|
|
self.assertEqual(waiting["status"], "awaiting_confirmation")
|
|
state = issue_resolution.issue_resolution_state(waiting)
|
|
self.assertEqual(state["attemptsSent"], 1)
|
|
self.assertEqual(state["maximumAttempts"], 2)
|
|
send_email.assert_awaited_once()
|
|
|
|
activity = db.list_portal_item_activity(int(issue["id"]))
|
|
self.assertEqual(
|
|
[event["event_type"] for event in activity],
|
|
["resolution_proposed", "confirmation_email_sent"],
|
|
)
|
|
|
|
closed = issue_resolution.respond_to_issue_confirmation(
|
|
int(issue["id"]),
|
|
resolved=True,
|
|
actor_username="reporter",
|
|
actor_role="user",
|
|
)
|
|
self.assertEqual(closed["status"], "closed")
|
|
self.assertIsNotNone(closed["issue_resolved_at"])
|
|
self.assertEqual(db.list_portal_item_activity(int(issue["id"]))[-1]["event_type"], "resolution_confirmed")
|
|
|
|
async def test_replacement_waits_for_jellyfin_to_refresh_existing_movie(self) -> None:
|
|
tracking = {
|
|
"requestId": "144",
|
|
"actionId": "replace_media",
|
|
"mediaType": "movie",
|
|
"collectorId": 12,
|
|
"originalFileIds": [40],
|
|
"episodes": [],
|
|
"jellyfinFoundAtStart": True,
|
|
"jellyfinBaseline": [{"Id": "jf-1", "Etag": "old", "Path": "/movies/test.mkv"}],
|
|
}
|
|
unchanged = Snapshot(
|
|
request_id="144",
|
|
title="Test movie",
|
|
raw={
|
|
"arr": {"item": {"id": 12, "hasFile": True, "movieFile": {"id": 41}}},
|
|
"jellyfin": {
|
|
"found": True,
|
|
"item": {"Id": "jf-1", "Etag": "old", "Path": "/movies/test.mkv"},
|
|
},
|
|
},
|
|
)
|
|
refreshed = unchanged.model_copy(deep=True)
|
|
refreshed.raw["jellyfin"]["item"]["Etag"] = "new"
|
|
|
|
with patch.object(issue_resolution, "build_snapshot", new=AsyncMock(return_value=unchanged)):
|
|
waiting = await issue_resolution._media_repair_evidence(tracking)
|
|
with patch.object(issue_resolution, "build_snapshot", new=AsyncMock(return_value=refreshed)):
|
|
complete = await issue_resolution._media_repair_evidence(tracking)
|
|
|
|
self.assertFalse(waiting["complete"])
|
|
self.assertEqual(waiting["phase"], "indexing")
|
|
self.assertTrue(complete["complete"])
|
|
|
|
async def test_verified_media_repair_starts_confirmation_without_admin(self) -> None:
|
|
issue = self._create_issue()
|
|
db.add_portal_item_activity(
|
|
int(issue["id"]),
|
|
event_type="replacement_started",
|
|
actor_username="reporter",
|
|
actor_role="user",
|
|
message="Radarr started the replacement.",
|
|
metadata_json=(
|
|
'{"repairTracking":{"requestId":"144","actionId":"replace_media",'
|
|
'"mediaType":"movie","collectorId":12,"originalFileIds":[40],'
|
|
'"episodes":[],"jellyfinFoundAtStart":true,'
|
|
'"jellyfinBaseline":[{"Id":"jf-1","Etag":"old"}]}}'
|
|
),
|
|
)
|
|
|
|
with (
|
|
patch.object(
|
|
issue_resolution,
|
|
"_media_repair_evidence",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"complete": True,
|
|
"phase": "complete",
|
|
"message": "Radarr imported the repaired movie and Jellyfin indexed it.",
|
|
}
|
|
),
|
|
),
|
|
patch.object(
|
|
issue_resolution,
|
|
"begin_issue_confirmation",
|
|
new=AsyncMock(return_value={"status": "awaiting_confirmation"}),
|
|
) as begin_confirmation,
|
|
):
|
|
result = await issue_resolution.process_active_media_repairs()
|
|
|
|
self.assertEqual(result["completed"], 1)
|
|
begin_confirmation.assert_awaited_once_with(
|
|
int(issue["id"]),
|
|
actor_username="Magent",
|
|
actor_role="system",
|
|
)
|
|
self.assertEqual(
|
|
db.list_portal_item_activity(int(issue["id"]))[-1]["event_type"],
|
|
"repair_verified",
|
|
)
|
|
|
|
async def test_zero_confirmation_emails_closes_issue_immediately(self) -> None:
|
|
issue = self._create_issue()
|
|
with patch.object(issue_resolution, "_workflow_settings", return_value=(0, 1, "days")):
|
|
closed = await issue_resolution.begin_issue_confirmation(
|
|
int(issue["id"]),
|
|
actor_username="admin",
|
|
actor_role="admin",
|
|
)
|
|
|
|
self.assertEqual(closed["status"], "closed")
|
|
self.assertEqual(
|
|
[event["event_type"] for event in db.list_portal_item_activity(int(issue["id"]))],
|
|
["resolution_proposed", "issue_auto_closed"],
|
|
)
|
|
|
|
async def test_saving_waiting_issue_does_not_restart_confirmation_cycle(self) -> None:
|
|
issue = self._create_issue(status="awaiting_confirmation")
|
|
user = {"username": "admin", "role": "admin"}
|
|
with patch.object(portal_router, "begin_issue_confirmation", new=AsyncMock()) as begin_confirmation:
|
|
result = await portal_router.portal_update_item(
|
|
int(issue["id"]),
|
|
{"title": "Updated title", "status": "awaiting_confirmation"},
|
|
user,
|
|
)
|
|
|
|
self.assertEqual(result["item"]["title"], "Updated title")
|
|
begin_confirmation.assert_not_awaited()
|
|
|
|
def test_public_activity_hides_internal_notes_and_admin_identity(self) -> None:
|
|
issue = self._create_issue()
|
|
db.add_portal_item_activity(
|
|
int(issue["id"]),
|
|
event_type="internal_note_added",
|
|
actor_username="private-admin-name",
|
|
actor_role="admin",
|
|
message="Internal diagnostic detail",
|
|
)
|
|
db.add_portal_item_activity(
|
|
int(issue["id"]),
|
|
event_type="status_changed",
|
|
actor_username="private-admin-name",
|
|
actor_role="admin",
|
|
message="Status changed to in progress.",
|
|
metadata_json='{"private":true}',
|
|
)
|
|
|
|
public_activity = portal_router._activity_payload(issue)
|
|
|
|
self.assertNotIn("Internal diagnostic detail", str(public_activity))
|
|
self.assertNotIn("private-admin-name", str(public_activity))
|
|
self.assertNotIn("metadata_json", public_activity[-1])
|
|
self.assertEqual(public_activity[-1]["actor_username"], "Support team")
|