422 lines
17 KiB
Python
422 lines
17 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.config import settings
|
|
from backend.app.network_security import request_trusts_forwarded_headers, validate_notification_target_url
|
|
from backend.app.models import NormalizedState, RequestType, Snapshot, TimelineHop
|
|
from backend.app.routers import auth as auth_router
|
|
from backend.app.routers import portal as portal_router
|
|
from backend.app.routers import requests as requests_router
|
|
from backend.app.routers import site as site_router
|
|
from backend.app.routers import status as status_router
|
|
from backend.app.security import PASSWORD_POLICY_MESSAGE, validate_password_policy
|
|
from backend.app.services import password_reset
|
|
from backend.app.services.snapshot import _build_presentation, _episode_availability, _torrent_progress
|
|
|
|
|
|
def _build_request(ip: str = "127.0.0.1", user_agent: str = "backend-test") -> Request:
|
|
scope = {
|
|
"type": "http",
|
|
"http_version": "1.1",
|
|
"method": "POST",
|
|
"scheme": "http",
|
|
"path": "/auth/password/forgot",
|
|
"raw_path": b"/auth/password/forgot",
|
|
"query_string": b"",
|
|
"headers": [(b"user-agent", user_agent.encode("utf-8"))],
|
|
"client": (ip, 12345),
|
|
"server": ("testserver", 8000),
|
|
}
|
|
|
|
async def receive() -> dict:
|
|
return {"type": "http.request", "body": b"", "more_body": False}
|
|
|
|
return Request(scope, receive)
|
|
|
|
|
|
class TempDatabaseMixin:
|
|
def setUp(self) -> None:
|
|
super_method = getattr(super(), "setUp", None)
|
|
if callable(super_method):
|
|
super_method()
|
|
self._tempdir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|
self._original_sqlite_path = settings.sqlite_path
|
|
self._original_journal_mode = getattr(settings, "sqlite_journal_mode", "DELETE")
|
|
settings.sqlite_path = os.path.join(self._tempdir.name, "test.db")
|
|
settings.sqlite_journal_mode = "DELETE"
|
|
auth_router._LOGIN_ATTEMPTS_BY_IP.clear()
|
|
auth_router._LOGIN_ATTEMPTS_BY_USER.clear()
|
|
auth_router._RESET_ATTEMPTS_BY_IP.clear()
|
|
auth_router._RESET_ATTEMPTS_BY_IDENTIFIER.clear()
|
|
db.init_db()
|
|
|
|
def tearDown(self) -> None:
|
|
settings.sqlite_path = self._original_sqlite_path
|
|
settings.sqlite_journal_mode = self._original_journal_mode
|
|
auth_router._LOGIN_ATTEMPTS_BY_IP.clear()
|
|
auth_router._LOGIN_ATTEMPTS_BY_USER.clear()
|
|
auth_router._RESET_ATTEMPTS_BY_IP.clear()
|
|
auth_router._RESET_ATTEMPTS_BY_IDENTIFIER.clear()
|
|
self._tempdir.cleanup()
|
|
super_method = getattr(super(), "tearDown", None)
|
|
if callable(super_method):
|
|
super_method()
|
|
|
|
|
|
class PasswordPolicyTests(unittest.TestCase):
|
|
def test_validate_password_policy_rejects_short_passwords(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, PASSWORD_POLICY_MESSAGE):
|
|
validate_password_policy("short")
|
|
|
|
def test_validate_password_policy_trims_whitespace(self) -> None:
|
|
self.assertEqual(validate_password_policy(" password123 "), "password123")
|
|
|
|
|
|
class NetworkSecurityTests(unittest.TestCase):
|
|
def test_notification_targets_reject_loopback(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "Private or local notification targets are not allowed."):
|
|
validate_notification_target_url("http://127.0.0.1:8080/webhook")
|
|
|
|
def test_forwarded_headers_require_trusted_proxy(self) -> None:
|
|
original_enabled = settings.magent_proxy_enabled
|
|
original_trust = settings.magent_proxy_trust_forwarded_headers
|
|
original_proxies = settings.magent_proxy_trusted_proxies
|
|
settings.magent_proxy_enabled = True
|
|
settings.magent_proxy_trust_forwarded_headers = True
|
|
settings.magent_proxy_trusted_proxies = "127.0.0.1,::1"
|
|
try:
|
|
self.assertTrue(request_trusts_forwarded_headers("127.0.0.1"))
|
|
self.assertFalse(request_trusts_forwarded_headers("203.0.113.10"))
|
|
finally:
|
|
settings.magent_proxy_enabled = original_enabled
|
|
settings.magent_proxy_trust_forwarded_headers = original_trust
|
|
settings.magent_proxy_trusted_proxies = original_proxies
|
|
|
|
|
|
class ServiceStatusTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_qbittorrent_login_accepts_modern_empty_response_with_session_cookie(self) -> None:
|
|
class FakeClient:
|
|
def __init__(self) -> None:
|
|
self.cookies = httpx.Cookies()
|
|
|
|
async def post(self, *_args, **_kwargs) -> httpx.Response:
|
|
self.cookies.set("QBT_SID_8080", "session")
|
|
return httpx.Response(204, request=httpx.Request("POST", "http://10.0.0.2:8080/api/v2/auth/login"))
|
|
|
|
client = status_router.QBittorrentClient("http://10.0.0.2:8080", "admin", "secret")
|
|
|
|
await client._login(FakeClient())
|
|
|
|
async def test_qbittorrent_incomplete_credentials_report_degraded_when_reachable(self) -> None:
|
|
client = status_router.QBittorrentClient("http://10.0.0.2:8080", "admin", None)
|
|
with patch.object(client, "is_webui_reachable", new=AsyncMock(return_value=True)):
|
|
result = await status_router._check_qbittorrent(client)
|
|
|
|
self.assertEqual(result["status"], "degraded")
|
|
self.assertIn("credentials", result["message"].lower())
|
|
|
|
async def test_qbittorrent_rejected_credentials_report_degraded_when_reachable(self) -> None:
|
|
client = status_router.QBittorrentClient("http://10.0.0.2:8080", "admin", "secret")
|
|
with patch.object(
|
|
client,
|
|
"get_app_version",
|
|
new=AsyncMock(side_effect=RuntimeError("qBittorrent login failed")),
|
|
), patch.object(client, "is_webui_reachable", new=AsyncMock(return_value=True)):
|
|
result = await status_router._check_qbittorrent(client)
|
|
|
|
self.assertEqual(result["status"], "degraded")
|
|
self.assertIn("credentials", result["message"].lower())
|
|
|
|
|
|
class SiteInfoTests(unittest.TestCase):
|
|
def test_site_public_exposes_requests_navigation_toggle(self) -> None:
|
|
runtime = SimpleNamespace(
|
|
site_build_number="test-build",
|
|
site_banner_enabled=False,
|
|
site_banner_message="",
|
|
site_banner_tone="info",
|
|
site_login_show_jellyfin_login=True,
|
|
site_login_show_local_login=True,
|
|
site_login_show_forgot_password=True,
|
|
site_login_show_signup_link=True,
|
|
site_nav_show_requests=False,
|
|
)
|
|
|
|
with patch.object(site_router, "get_runtime_settings", return_value=runtime):
|
|
info = site_router._build_site_info(False)
|
|
|
|
self.assertEqual(info["navigation"], {"showRequests": False})
|
|
|
|
|
|
class RequestCacheTests(unittest.TestCase):
|
|
def tearDown(self) -> None:
|
|
requests_router._detail_cache.clear()
|
|
requests_router._failed_detail_cache.clear()
|
|
|
|
def test_successful_detail_cache_write_clears_prior_failure(self) -> None:
|
|
key = "request:123"
|
|
requests_router._failure_cache_set(key)
|
|
self.assertTrue(requests_router._failure_cache_has(key))
|
|
|
|
requests_router._cache_set(key, {"id": 123})
|
|
|
|
self.assertFalse(requests_router._failure_cache_has(key))
|
|
self.assertEqual(requests_router._cache_get(key), {"id": 123})
|
|
|
|
|
|
class RequestPresentationTests(unittest.TestCase):
|
|
def test_torrent_progress_keeps_tenths_for_live_updates(self) -> None:
|
|
self.assertEqual(_torrent_progress({"progress": 0.1344}), 13.4)
|
|
|
|
def test_episode_availability_counts_only_aired_monitored_episodes(self) -> None:
|
|
episodes = [
|
|
{"seasonNumber": 1, "episodeNumber": 1, "monitored": True, "hasFile": True},
|
|
{"seasonNumber": 1, "episodeNumber": 2, "monitored": True, "hasFile": False},
|
|
{"seasonNumber": 1, "episodeNumber": 3, "monitored": False, "hasFile": False},
|
|
{
|
|
"seasonNumber": 1,
|
|
"episodeNumber": 4,
|
|
"monitored": True,
|
|
"hasFile": False,
|
|
"airDateUtc": "2999-01-01T00:00:00Z",
|
|
},
|
|
]
|
|
|
|
availability = _episode_availability(episodes)
|
|
|
|
self.assertEqual(availability["available"], 1)
|
|
self.assertEqual(availability["missing"], 1)
|
|
self.assertEqual(availability["total"], 2)
|
|
|
|
def test_presentation_hides_download_without_download_evidence(self) -> None:
|
|
snapshot = Snapshot(
|
|
request_id="3909",
|
|
title="Example",
|
|
request_type=RequestType.tv,
|
|
state=NormalizedState.added_to_arr,
|
|
actions=[],
|
|
)
|
|
|
|
presentation = _build_presentation(
|
|
snapshot,
|
|
approved=True,
|
|
arr_state="added",
|
|
arr_details={
|
|
"availability": {"available": 0, "missing": 6, "total": 6, "seasons": []}
|
|
},
|
|
prowlarr_state="ok",
|
|
download={
|
|
"visible": False,
|
|
"state": "not_started",
|
|
"summary": "No download attempt has been observed.",
|
|
"torrents": [],
|
|
},
|
|
jellyfin_found=False,
|
|
jellyfin_link=None,
|
|
)
|
|
|
|
self.assertFalse(presentation["download"]["visible"])
|
|
self.assertIn("waiting for 6 episodes", presentation["status"]["label"])
|
|
download_stage = next(stage for stage in presentation["pipeline"] if stage["id"] == "download")
|
|
self.assertEqual(download_stage["summary"], "No download attempt yet")
|
|
|
|
|
|
class LiveDownloadProgressTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_live_download_progress_uses_saved_hash_and_current_qbittorrent_value(self) -> None:
|
|
runtime = SimpleNamespace(
|
|
jellyseerr_base_url=None,
|
|
jellyseerr_api_key=None,
|
|
qbittorrent_base_url="http://qbittorrent.test",
|
|
qbittorrent_username="magent",
|
|
qbittorrent_password="secret",
|
|
)
|
|
evidence = {
|
|
"observed": True,
|
|
"torrents": [{"hash": "abc123", "progress": 0.12}],
|
|
}
|
|
current = [{"hash": "abc123", "name": "Example", "progress": 0.1344, "state": "downloading"}]
|
|
|
|
with patch.object(requests_router, "get_runtime_settings", return_value=runtime), patch.object(
|
|
requests_router,
|
|
"get_request_download_evidence",
|
|
return_value=evidence,
|
|
), patch.object(
|
|
requests_router.QBittorrentClient,
|
|
"get_torrents_by_hashes",
|
|
new=AsyncMock(return_value=current),
|
|
) as get_torrents:
|
|
result = await requests_router.get_download_progress(
|
|
"3909", user={"username": "viewer", "role": "user"}
|
|
)
|
|
|
|
get_torrents.assert_awaited_once_with("abc123")
|
|
self.assertEqual(result["state"], "downloading")
|
|
self.assertEqual(result["torrents"][0]["progressPercent"], 13.4)
|
|
|
|
|
|
class 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)
|