Phase 2: Docker and OIDC auth
This commit is contained in:
+420
-10
@@ -5,14 +5,23 @@ without requiring real remote connections.
|
||||
"""
|
||||
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from media_library_viewer_api.main import app
|
||||
from media_library_viewer_api.dependencies import get_ssh_client, get_jellyfin_client, get_user_id
|
||||
from media_library_viewer_api.dependencies import (
|
||||
get_ssh_client,
|
||||
get_jellyfin_client,
|
||||
get_jellyseerr_client,
|
||||
get_mail_queue,
|
||||
get_user_id,
|
||||
)
|
||||
from media_library_viewer_api.clients.ssh import CommandResult
|
||||
from media_library_viewer_api.routers.media import get_media_index
|
||||
from media_library_viewer_api.services.media_index import MediaIndex
|
||||
|
||||
|
||||
# --- Fixtures ---
|
||||
@@ -26,11 +35,15 @@ def mock_jellyfin():
|
||||
{"Id": "lib1", "Name": "Movies", "CollectionType": "movies"},
|
||||
{"Id": "lib2", "Name": "TV Shows", "CollectionType": "tvshows"},
|
||||
]
|
||||
client.users.return_value = [
|
||||
{"Id": "jf1", "Name": "alex"},
|
||||
{"Id": "jf2", "Name": "sam"},
|
||||
]
|
||||
client.library_item_counts.return_value = [
|
||||
{"library": "Movies", "type": "movies", "movies": 100, "series": 0, "episodes": 0, "total": 100},
|
||||
{"library": "TV Shows", "type": "tvshows", "movies": 0, "series": 20, "episodes": 500, "total": 520},
|
||||
]
|
||||
client.active_sessions.return_value = [
|
||||
client.sessions.return_value = [
|
||||
{
|
||||
"Id": "sess1",
|
||||
"UserName": "alex",
|
||||
@@ -38,11 +51,51 @@ def mock_jellyfin():
|
||||
"NowPlayingItem": {"Name": "Test Movie", "Type": "Movie"},
|
||||
"PlayState": {"IsPaused": False},
|
||||
"TranscodingInfo": {"IsVideoDirect": True, "IsAudioDirect": False},
|
||||
}
|
||||
},
|
||||
{
|
||||
"Id": "sess2",
|
||||
"UserName": "sam",
|
||||
"DeviceName": "Android",
|
||||
"NowPlayingItem": None,
|
||||
"PlayState": {},
|
||||
"TranscodingInfo": None,
|
||||
},
|
||||
]
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_jellyseerr():
|
||||
"""Mock Jellyseerr client."""
|
||||
client = MagicMock()
|
||||
client.jellyfin_users.return_value = [
|
||||
{"id": "jf1", "username": "alex", "thumb": "/avatarproxy/alex", "email": "alex@example.com"},
|
||||
{"id": "jf2", "username": "sam", "thumb": "/avatarproxy/sam", "email": "sam@example.com"},
|
||||
]
|
||||
client.users.return_value = [
|
||||
{
|
||||
"id": 7,
|
||||
"username": "alex",
|
||||
"email": "alex@example.com",
|
||||
"avatar": "/avatarproxy/alex",
|
||||
"userType": 3,
|
||||
"permissions": 10,
|
||||
"requestCount": 3,
|
||||
},
|
||||
{
|
||||
"id": 8,
|
||||
"username": "sam",
|
||||
"email": "sam@example.com",
|
||||
"avatar": "/avatarproxy/sam",
|
||||
"userType": 2,
|
||||
"permissions": 32,
|
||||
"requestCount": 1,
|
||||
},
|
||||
]
|
||||
client.absolute_url.side_effect = lambda path: f"https://requests.example.com{path}"
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ssh():
|
||||
"""Mock SSH client."""
|
||||
@@ -73,9 +126,10 @@ def mock_ssh():
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_client(mock_jellyfin, mock_ssh):
|
||||
def test_client(mock_jellyfin, mock_jellyseerr, mock_ssh):
|
||||
"""FastAPI test client with mocked dependencies."""
|
||||
app.dependency_overrides[get_jellyfin_client] = lambda: mock_jellyfin
|
||||
app.dependency_overrides[get_jellyseerr_client] = lambda: mock_jellyseerr
|
||||
app.dependency_overrides[get_ssh_client] = lambda: mock_ssh
|
||||
app.dependency_overrides[get_user_id] = lambda: "user123"
|
||||
client = TestClient(app)
|
||||
@@ -111,17 +165,244 @@ class TestDashboard:
|
||||
assert data[0]["library"] == "Movies"
|
||||
assert data[1]["library"] == "TV Shows"
|
||||
|
||||
def test_now_playing(self, test_client):
|
||||
def test_activity(self, test_client):
|
||||
response = test_client.get("/api/dashboard/activity")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 2
|
||||
|
||||
playing_row = next(row for row in data if row["user"] == "alex")
|
||||
assert playing_row["title"] == "Test Movie"
|
||||
assert playing_row["state"] == "playing"
|
||||
assert playing_row["transcoding"] == "yes"
|
||||
assert "audio" in playing_row["transcoding_type"]
|
||||
|
||||
idle_row = next(row for row in data if row["user"] == "sam")
|
||||
assert idle_row["state"] == "idle"
|
||||
assert idle_row["title"] == "(idle)"
|
||||
|
||||
def test_now_playing_alias(self, test_client):
|
||||
response = test_client.get("/api/dashboard/now-playing")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 1
|
||||
assert data[0]["user"] == "alex"
|
||||
assert data[0]["title"] == "Test Movie"
|
||||
assert data[0]["transcoding"] == "yes"
|
||||
assert "audio" in data[0]["transcoding_type"]
|
||||
assert len(data) == 2
|
||||
|
||||
|
||||
# --- Users ---
|
||||
|
||||
class TestUsers:
|
||||
def test_users_list_enriched(self, test_client):
|
||||
response = test_client.get("/api/users")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total"] == 2
|
||||
assert data["jellyseerr_configured"] is True
|
||||
assert data["jellyseerr_available"] is True
|
||||
assert data["jellyseerr_error"] == ""
|
||||
|
||||
alex = next(item for item in data["items"] if item["username"] == "alex")
|
||||
assert alex["email"] == "alex@example.com"
|
||||
assert alex["email_source"] == "jellyseerr:user"
|
||||
assert alex["contactable"] is True
|
||||
assert alex["avatar"].startswith("https://requests.example.com/")
|
||||
assert alex["avatar_source"] == "jellyseerr:user"
|
||||
assert alex["permissions"] == 10
|
||||
assert alex["permissions_label"] == "admin, manage_users"
|
||||
assert alex["role"] == "admin"
|
||||
assert alex["user_type_label"] == "jellyfin"
|
||||
assert alex["request_count"] == 3
|
||||
assert "name=jellyfin" in alex["source_summary"]
|
||||
assert "email=jellyseerr:user" in alex["source_summary"]
|
||||
|
||||
sam = next(item for item in data["items"] if item["username"] == "sam")
|
||||
assert sam["role"] == "requester"
|
||||
assert sam["user_type_label"] == "local"
|
||||
assert sam["email"] == "sam@example.com"
|
||||
|
||||
def test_users_message_status(self, test_client):
|
||||
mail_queue = MagicMock()
|
||||
mail_queue.status.return_value = {
|
||||
"state": "idle",
|
||||
"worker_running": True,
|
||||
"stop_requested": False,
|
||||
"pending_count": 0,
|
||||
"active_request_id": None,
|
||||
"last_request_id": None,
|
||||
"last_result": None,
|
||||
"last_error": "",
|
||||
"last_error_at": None,
|
||||
"last_success_at": None,
|
||||
"last_activity_at": None,
|
||||
"sent_count": 0,
|
||||
"failed_count": 0,
|
||||
}
|
||||
app.dependency_overrides[get_mail_queue] = lambda: mail_queue
|
||||
try:
|
||||
response = test_client.get("/api/users/message/status")
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_mail_queue, None)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["state"] == "idle"
|
||||
assert response.json()["pending_count"] == 0
|
||||
|
||||
def test_users_message_test_smtp(self, test_client):
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.fastmail.com",
|
||||
smtp_port=587,
|
||||
smtp_username="main@fastmail.com",
|
||||
smtp_password="app-password",
|
||||
smtp_from_address="alias@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=True,
|
||||
smtp_use_ssl=False,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
app.dependency_overrides[get_mail_queue] = lambda: MagicMock()
|
||||
try:
|
||||
with patch("media_library_viewer_api.routers.users.get_settings", return_value=settings), patch(
|
||||
"media_library_viewer_api.routers.users.test_smtp_connection",
|
||||
return_value={
|
||||
"status": "ok",
|
||||
"message": "SMTP connection successful using Fastmail STARTTLS 587",
|
||||
"from_address": "alias@example.com",
|
||||
"from_name": "Media Library Viewer",
|
||||
"smtp_host": "smtp.fastmail.com",
|
||||
"smtp_port": 587,
|
||||
"use_tls": True,
|
||||
"use_ssl": False,
|
||||
"authenticated": True,
|
||||
"selected_mode": {
|
||||
"label": "Fastmail STARTTLS 587",
|
||||
"smtp_host": "smtp.fastmail.com",
|
||||
"smtp_port": 587,
|
||||
"use_tls": True,
|
||||
"use_ssl": False,
|
||||
},
|
||||
"attempts": [
|
||||
{
|
||||
"label": "Fastmail STARTTLS 587",
|
||||
"smtp_host": "smtp.fastmail.com",
|
||||
"smtp_port": 587,
|
||||
"use_tls": True,
|
||||
"use_ssl": False,
|
||||
"status": "ok",
|
||||
}
|
||||
],
|
||||
},
|
||||
):
|
||||
response = test_client.post("/api/users/message/test-smtp")
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_mail_queue, None)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "ok"
|
||||
assert data["smtp_host"] == "smtp.fastmail.com"
|
||||
assert data["selected_mode"]["label"] == "Fastmail STARTTLS 587"
|
||||
|
||||
def test_users_message_test_smtp_timeout_message(self, test_client):
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.fastmail.com",
|
||||
smtp_port=587,
|
||||
smtp_username="main@fastmail.com",
|
||||
smtp_password="app-password",
|
||||
smtp_from_address="alias@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=True,
|
||||
smtp_use_ssl=False,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
app.dependency_overrides[get_mail_queue] = lambda: MagicMock()
|
||||
try:
|
||||
with patch("media_library_viewer_api.routers.users.get_settings", return_value=settings), patch(
|
||||
"media_library_viewer_api.routers.users.test_smtp_connection",
|
||||
return_value={
|
||||
"status": "error",
|
||||
"message": "SMTP connection timed out while waiting for the server greeting. Check host, port, network access, and SMTP_TIMEOUT.",
|
||||
"from_address": "alias@example.com",
|
||||
"from_name": "Media Library Viewer",
|
||||
"smtp_host": "smtp.fastmail.com",
|
||||
"smtp_port": 587,
|
||||
"use_tls": True,
|
||||
"use_ssl": False,
|
||||
"authenticated": True,
|
||||
"selected_mode": None,
|
||||
"attempts": [
|
||||
{
|
||||
"label": "configured",
|
||||
"smtp_host": "smtp.fastmail.com",
|
||||
"smtp_port": 587,
|
||||
"use_tls": True,
|
||||
"use_ssl": False,
|
||||
"status": "failed",
|
||||
"error": "SMTP connection timed out while waiting for the server greeting. Check host, port, network access, and SMTP_TIMEOUT.",
|
||||
}
|
||||
],
|
||||
},
|
||||
):
|
||||
response = test_client.post("/api/users/message/test-smtp")
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_mail_queue, None)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["status"] == "error"
|
||||
assert "timed out" in response.json()["message"].lower()
|
||||
|
||||
def test_users_message_is_queued(self, test_client):
|
||||
mail_queue = MagicMock()
|
||||
mail_queue.status.return_value = {
|
||||
"state": "idle",
|
||||
"worker_running": True,
|
||||
"stop_requested": False,
|
||||
"pending_count": 0,
|
||||
"active_request_id": None,
|
||||
"last_request_id": None,
|
||||
"last_result": None,
|
||||
"last_error": "",
|
||||
"last_error_at": None,
|
||||
"last_success_at": None,
|
||||
"last_activity_at": None,
|
||||
"sent_count": 0,
|
||||
"failed_count": 0,
|
||||
}
|
||||
mail_queue.enqueue.return_value = "mail-123456"
|
||||
app.dependency_overrides[get_mail_queue] = lambda: mail_queue
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.example.com",
|
||||
smtp_port=587,
|
||||
smtp_username="mailer@example.com",
|
||||
smtp_password="secret",
|
||||
smtp_from_address="mailer@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=True,
|
||||
smtp_use_ssl=False,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
|
||||
try:
|
||||
with patch("media_library_viewer_api.routers.users.get_settings", return_value=settings):
|
||||
response = test_client.post(
|
||||
"/api/users/message",
|
||||
data={
|
||||
"recipient_ids": json.dumps(["jf1", "jf2"]),
|
||||
"subject": "Hello team",
|
||||
"html_body": "<p>Hi there</p>",
|
||||
"text_body": "Hi there",
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_mail_queue, None)
|
||||
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "queued"
|
||||
assert data["request_id"] == "mail-123456"
|
||||
assert data["recipient_count"] == 2
|
||||
assert data["attachment_count"] == 0
|
||||
mail_queue.enqueue.assert_called_once()
|
||||
kwargs = mail_queue.enqueue.call_args.kwargs
|
||||
assert kwargs["recipients"] == ["alex@example.com", "sam@example.com"]
|
||||
assert kwargs["subject"] == "Hello team"
|
||||
assert kwargs["settings"] is settings
|
||||
|
||||
# --- Files ---
|
||||
|
||||
class TestFiles:
|
||||
@@ -163,6 +444,135 @@ class TestFiles:
|
||||
assert "Permission denied" in response.text
|
||||
|
||||
|
||||
# --- Media index build ---
|
||||
|
||||
class TestMediaIndexApi:
|
||||
def test_status_includes_build_progress(self, test_client, tmp_path):
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
index.init_schema()
|
||||
index.set_metadata("build_running", "true")
|
||||
index.set_metadata("build_stage", "building")
|
||||
index.set_metadata("build_message", "3 / 10 items")
|
||||
index.set_metadata("build_progress", "0.3")
|
||||
index.set_metadata("build_items_processed", "3")
|
||||
index.set_metadata("build_items_total", "10")
|
||||
index.set_metadata("build_current_library", "Movies")
|
||||
index.set_metadata("build_library_index", "1")
|
||||
index.set_metadata("build_libraries_total", "2")
|
||||
index.set_metadata("build_library_progress", "0.5")
|
||||
index.set_metadata("build_library_items_processed", "1")
|
||||
index.set_metadata("build_library_items_total", "2")
|
||||
index.set_metadata("build_elapsed_seconds", "12.0")
|
||||
index.set_metadata("build_eta_seconds", "28.0")
|
||||
index.set_metadata("build_library_elapsed_seconds", "4.0")
|
||||
index.set_metadata("build_library_eta_seconds", "4.0")
|
||||
index.set_metadata("build_pid", "4321")
|
||||
app.dependency_overrides[get_media_index] = lambda: index
|
||||
with patch("media_library_viewer_api.routers.media._pid_is_alive", return_value=True):
|
||||
response = test_client.get("/api/media/status")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["build_running"] is True
|
||||
assert data["build_stage"] == "building"
|
||||
assert data["build_message"] == "3 / 10 items"
|
||||
assert data["build_progress"] == 0.3
|
||||
assert data["build_items_processed"] == 3
|
||||
assert data["build_items_total"] == 10
|
||||
assert data["build_current_library"] == "Movies"
|
||||
assert data["build_library_index"] == 1
|
||||
assert data["build_libraries_total"] == 2
|
||||
assert data["build_library_progress"] == 0.5
|
||||
assert data["build_library_items_processed"] == 1
|
||||
assert data["build_library_items_total"] == 2
|
||||
assert data["build_elapsed_seconds"] == 12.0
|
||||
assert data["build_eta_seconds"] == 28.0
|
||||
assert data["build_library_elapsed_seconds"] == 4.0
|
||||
assert data["build_library_eta_seconds"] == 4.0
|
||||
assert data["build_pid"] == 4321
|
||||
|
||||
def test_build_returns_started_when_background_build_is_queued(self, test_client, tmp_path, mock_jellyfin):
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
app.dependency_overrides[get_media_index] = lambda: index
|
||||
|
||||
class FakeProcess:
|
||||
pid = 4321
|
||||
|
||||
with patch("media_library_viewer_api.routers.media._start_worker", return_value=FakeProcess()) as start_worker:
|
||||
try:
|
||||
response = test_client.post("/api/media/build")
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "started"
|
||||
assert data["build_running"] is True
|
||||
assert data["build_stage"] == "queued"
|
||||
assert data["build_libraries_total"] == len(mock_jellyfin.libraries.return_value)
|
||||
assert data["build_pid"] == 4321
|
||||
start_worker.assert_called_once()
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_media_index, None)
|
||||
|
||||
def test_stop_requests_cancel(self, test_client, tmp_path):
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
index.init_schema()
|
||||
index.set_metadata("build_running", "true")
|
||||
index.set_metadata("build_stage", "building")
|
||||
index.set_metadata("build_pid", "4321")
|
||||
app.dependency_overrides[get_media_index] = lambda: index
|
||||
|
||||
with patch("media_library_viewer_api.routers.media._pid_is_alive", return_value=True):
|
||||
try:
|
||||
response = test_client.post("/api/media/stop")
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "stop_requested"
|
||||
assert data["build_cancel_requested"] is True
|
||||
status = test_client.get("/api/media/status").json()
|
||||
assert status["build_cancel_requested"] is True
|
||||
assert status["build_stage"] == "canceling"
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_media_index, None)
|
||||
|
||||
def test_force_stop_terminates_worker(self, test_client, tmp_path):
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
index.init_schema()
|
||||
index.set_metadata("build_running", "true")
|
||||
index.set_metadata("build_stage", "building")
|
||||
index.set_metadata("build_pid", "4321")
|
||||
app.dependency_overrides[get_media_index] = lambda: index
|
||||
|
||||
alive_calls = {"count": 0}
|
||||
|
||||
def fake_pid_is_alive(pid):
|
||||
alive_calls["count"] += 1
|
||||
return alive_calls["count"] <= 2
|
||||
|
||||
with patch("media_library_viewer_api.routers.media._pid_is_alive", side_effect=fake_pid_is_alive), patch(
|
||||
"media_library_viewer_api.routers.media.os.killpg"
|
||||
) as killpg, patch("media_library_viewer_api.routers.media.time.sleep", return_value=None):
|
||||
try:
|
||||
response = test_client.post("/api/media/force-stop")
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
assert data["status"] == "force_stopped"
|
||||
killpg.assert_called()
|
||||
status = test_client.get("/api/media/status").json()
|
||||
assert status["build_running"] is False
|
||||
assert status["build_stage"] == "force-stopped"
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_media_index, None)
|
||||
|
||||
def test_force_stop_returns_conflict_when_idle(self, test_client, tmp_path):
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
index.init_schema()
|
||||
app.dependency_overrides[get_media_index] = lambda: index
|
||||
|
||||
try:
|
||||
response = test_client.post("/api/media/force-stop")
|
||||
assert response.status_code == 409
|
||||
finally:
|
||||
app.dependency_overrides.pop(get_media_index, None)
|
||||
|
||||
|
||||
# --- Jobs ---
|
||||
|
||||
class TestJobs:
|
||||
|
||||
@@ -12,12 +12,17 @@ class TestSettings:
|
||||
assert settings.ssh_host == ""
|
||||
assert settings.ssh_port == 22
|
||||
assert settings.jellyfin_url == ""
|
||||
assert settings.jellyseerr_url == ""
|
||||
assert settings.log_level == "INFO"
|
||||
assert settings.remote_media_root == ""
|
||||
|
||||
def test_from_env(self):
|
||||
env = {
|
||||
"JELLYFIN_URL": "https://test.example.com",
|
||||
"JELLYFIN_API_KEY": "key123",
|
||||
"JELLYSEERR_URL": "https://requests.example.com",
|
||||
"JELLYSEERR_API_KEY": "seerr123",
|
||||
"LOG_LEVEL": "DEBUG",
|
||||
"SSH_HOST": "192.168.1.1",
|
||||
"SSH_USERNAME": "testuser",
|
||||
"SSH_PORT": "2222",
|
||||
@@ -27,6 +32,9 @@ class TestSettings:
|
||||
settings = Settings(_env_file=None)
|
||||
assert settings.jellyfin_url == "https://test.example.com"
|
||||
assert settings.jellyfin_api_key == "key123"
|
||||
assert settings.jellyseerr_url == "https://requests.example.com"
|
||||
assert settings.jellyseerr_api_key == "seerr123"
|
||||
assert settings.log_level == "DEBUG"
|
||||
assert settings.ssh_host == "192.168.1.1"
|
||||
assert settings.ssh_username == "testuser"
|
||||
assert settings.ssh_port == 2222
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Unit tests for the Jellyseerr client."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from media_library_viewer_api.clients.jellyseerr import JellyseerrClient
|
||||
|
||||
|
||||
class JellyseerrClientTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.client = JellyseerrClient("https://requests.example.com/api/v1", "api-key")
|
||||
self.session = MagicMock()
|
||||
self.client.session = self.session
|
||||
|
||||
def test_jellyfin_users_accepts_wrapped_payload(self) -> None:
|
||||
response = MagicMock()
|
||||
response.raise_for_status.return_value = None
|
||||
response.json.return_value = {
|
||||
"users": [
|
||||
{"id": "jf1", "username": "alex", "email": "alex@example.com", "thumb": "/avatar"},
|
||||
{"id": "jf2", "username": "sam", "email": "sam@example.com", "thumb": "/avatar2"},
|
||||
]
|
||||
}
|
||||
self.session.get.return_value = response
|
||||
|
||||
users = self.client.jellyfin_users()
|
||||
|
||||
self.assertEqual(len(users), 2)
|
||||
self.assertEqual(users[0]["email"], "alex@example.com")
|
||||
self.assertEqual(users[1]["username"], "sam")
|
||||
self.session.get.assert_called_once()
|
||||
|
||||
def test_jellyfin_users_accepts_list_payload(self) -> None:
|
||||
response = MagicMock()
|
||||
response.raise_for_status.return_value = None
|
||||
response.json.return_value = [
|
||||
{"id": "jf1", "username": "alex"},
|
||||
{"id": "jf2", "username": "sam"},
|
||||
]
|
||||
self.session.get.return_value = response
|
||||
|
||||
users = self.client.jellyfin_users()
|
||||
|
||||
self.assertEqual([u["username"] for u in users], ["alex", "sam"])
|
||||
|
||||
def test_users_uses_take_and_skip(self) -> None:
|
||||
first = MagicMock()
|
||||
first.raise_for_status.return_value = None
|
||||
first.json.return_value = {
|
||||
"pageInfo": {"results": 3, "pages": 2, "pageSize": 2, "page": 1},
|
||||
"results": [
|
||||
{"id": 1, "username": "alex", "email": "alex@example.com"},
|
||||
{"id": 2, "username": "sam", "email": "sam@example.com"},
|
||||
],
|
||||
}
|
||||
second = MagicMock()
|
||||
second.raise_for_status.return_value = None
|
||||
second.json.return_value = {
|
||||
"pageInfo": {"results": 3, "pages": 2, "pageSize": 2, "page": 2},
|
||||
"results": [
|
||||
{"id": 3, "username": "max", "email": "max@example.com"},
|
||||
],
|
||||
}
|
||||
self.session.get.side_effect = [first, second]
|
||||
|
||||
users = self.client.users(page_size=2)
|
||||
|
||||
self.assertEqual([u["username"] for u in users], ["alex", "sam", "max"])
|
||||
self.assertEqual(self.session.get.call_args_list[0].kwargs["params"], {"take": 2, "skip": 0})
|
||||
self.assertEqual(self.session.get.call_args_list[1].kwargs["params"], {"take": 2, "skip": 2})
|
||||
|
||||
def test_absolute_url_normalizes_relative_paths(self) -> None:
|
||||
self.assertEqual(self.client.absolute_url("/avatar.png"), "https://requests.example.com/avatar.png")
|
||||
self.assertEqual(self.client.absolute_url("avatar.png"), "https://requests.example.com/avatar.png")
|
||||
self.assertEqual(self.client.absolute_url("https://cdn.example.com/x.png"), "https://cdn.example.com/x.png")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,207 @@
|
||||
"""Unit tests for SMTP mail helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import smtplib
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from media_library_viewer_api.services.mailer import (
|
||||
EmailAttachment,
|
||||
describe_smtp_error,
|
||||
html_to_text,
|
||||
send_email_message,
|
||||
test_smtp_connection as smtp_connection_probe,
|
||||
)
|
||||
|
||||
|
||||
class _SMTPContext:
|
||||
def __init__(self, smtp: MagicMock):
|
||||
self.smtp = smtp
|
||||
|
||||
def __enter__(self):
|
||||
return self.smtp
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
|
||||
class MailerTests(unittest.TestCase):
|
||||
def test_html_to_text_strips_tags(self) -> None:
|
||||
text = html_to_text("<p>Hello <strong>world</strong></p><p>Line 2</p>")
|
||||
self.assertIn("Hello", text)
|
||||
self.assertIn("world", text)
|
||||
self.assertIn("Line 2", text)
|
||||
|
||||
def test_send_email_message_uses_smtp_with_attachments(self) -> None:
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.example.com",
|
||||
smtp_port=587,
|
||||
smtp_username="mailer@example.com",
|
||||
smtp_password="secret",
|
||||
smtp_from_address="mailer@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=True,
|
||||
smtp_use_ssl=False,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
smtp = MagicMock()
|
||||
smtp.send_message.return_value = {}
|
||||
smtp_factory = MagicMock(return_value=_SMTPContext(smtp))
|
||||
|
||||
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP", smtp_factory), patch(
|
||||
"media_library_viewer_api.services.mailer.smtplib.SMTP_SSL"
|
||||
) as smtp_ssl:
|
||||
result = send_email_message(
|
||||
settings,
|
||||
recipients=["alex@example.com", "sam@example.com"],
|
||||
subject="Hello",
|
||||
html_body="<p><strong>Hi</strong> there</p>",
|
||||
attachments=[EmailAttachment(filename="note.txt", content_type="text/plain", data=b"note")],
|
||||
)
|
||||
|
||||
smtp_ssl.assert_not_called()
|
||||
smtp.starttls.assert_called_once()
|
||||
smtp.login.assert_called_once_with("mailer@example.com", "secret")
|
||||
smtp.send_message.assert_called_once()
|
||||
message = smtp.send_message.call_args.args[0]
|
||||
self.assertEqual(message["Subject"], "Hello")
|
||||
self.assertEqual(message["From"], "Media Library Viewer <mailer@example.com>")
|
||||
self.assertEqual(result["recipient_count"], 2)
|
||||
self.assertEqual(result["attachment_count"], 1)
|
||||
|
||||
def test_test_smtp_connection_uses_starttls_and_login(self) -> None:
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.example.com",
|
||||
smtp_port=587,
|
||||
smtp_username="mailer@example.com",
|
||||
smtp_password="secret",
|
||||
smtp_from_address="alias@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=True,
|
||||
smtp_use_ssl=False,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
smtp = MagicMock()
|
||||
smtp_factory = MagicMock(return_value=_SMTPContext(smtp))
|
||||
|
||||
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP", smtp_factory), patch(
|
||||
"media_library_viewer_api.services.mailer.smtplib.SMTP_SSL"
|
||||
) as smtp_ssl:
|
||||
result = smtp_connection_probe(settings)
|
||||
|
||||
smtp_ssl.assert_not_called()
|
||||
smtp.ehlo.assert_called()
|
||||
smtp.starttls.assert_called_once()
|
||||
smtp.login.assert_called_once_with("mailer@example.com", "secret")
|
||||
smtp.noop.assert_called_once()
|
||||
self.assertEqual(result["status"], "ok")
|
||||
self.assertEqual(result["from_address"], "alias@example.com")
|
||||
self.assertEqual(result["smtp_host"], "smtp.example.com")
|
||||
self.assertEqual(result["selected_mode"]["label"], "configured")
|
||||
|
||||
def test_test_smtp_connection_falls_back_to_fastmail_mode(self) -> None:
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.fastmail.com",
|
||||
smtp_port=465,
|
||||
smtp_username="mailer@example.com",
|
||||
smtp_password="secret",
|
||||
smtp_from_address="alias@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=False,
|
||||
smtp_use_ssl=True,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
smtp_ssl = MagicMock(side_effect=TimeoutError("timed out"))
|
||||
fallback_smtp = MagicMock()
|
||||
smtp_factory = MagicMock(return_value=_SMTPContext(fallback_smtp))
|
||||
|
||||
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP_SSL", smtp_ssl), patch(
|
||||
"media_library_viewer_api.services.mailer.smtplib.SMTP",
|
||||
smtp_factory,
|
||||
):
|
||||
result = smtp_connection_probe(settings)
|
||||
|
||||
self.assertEqual(result["status"], "ok")
|
||||
self.assertEqual(result["selected_mode"]["label"], "Fastmail STARTTLS 587")
|
||||
self.assertEqual(len(result["attempts"]), 2)
|
||||
self.assertEqual(result["attempts"][0]["status"], "failed")
|
||||
self.assertEqual(result["attempts"][1]["status"], "ok")
|
||||
fallback_smtp.starttls.assert_called_once()
|
||||
fallback_smtp.login.assert_called_once_with("mailer@example.com", "secret")
|
||||
|
||||
def test_send_email_message_falls_back_to_fastmail_mode(self) -> None:
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.fastmail.com",
|
||||
smtp_port=465,
|
||||
smtp_username="mailer@example.com",
|
||||
smtp_password="secret",
|
||||
smtp_from_address="alias@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=False,
|
||||
smtp_use_ssl=True,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
smtp_ssl_factory = MagicMock(side_effect=TimeoutError("timed out"))
|
||||
fallback_smtp = MagicMock()
|
||||
fallback_factory = MagicMock(return_value=_SMTPContext(fallback_smtp))
|
||||
|
||||
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP_SSL", smtp_ssl_factory), patch(
|
||||
"media_library_viewer_api.services.mailer.smtplib.SMTP",
|
||||
fallback_factory,
|
||||
):
|
||||
result = send_email_message(
|
||||
settings,
|
||||
recipients=["alex@example.com"],
|
||||
subject="Hello",
|
||||
html_body="<p>Hello</p>",
|
||||
)
|
||||
|
||||
self.assertEqual(result["selected_mode"]["label"], "Fastmail STARTTLS 587")
|
||||
self.assertEqual(result["authenticated_as"], "mailer@example.com")
|
||||
self.assertEqual(len(result["attempts"]), 2)
|
||||
self.assertEqual(result["attempts"][0]["status"], "failed")
|
||||
self.assertEqual(result["attempts"][1]["status"], "ok")
|
||||
fallback_smtp.send_message.assert_called_once()
|
||||
|
||||
def test_send_email_message_rejects_unauthorized_from_address(self) -> None:
|
||||
settings = SimpleNamespace(
|
||||
smtp_host="smtp.example.com",
|
||||
smtp_port=587,
|
||||
smtp_username="mailer@example.com",
|
||||
smtp_password="secret",
|
||||
smtp_from_address="alias@example.com",
|
||||
smtp_from_name="Media Library Viewer",
|
||||
smtp_use_tls=True,
|
||||
smtp_use_ssl=False,
|
||||
smtp_timeout=15,
|
||||
)
|
||||
smtp = MagicMock()
|
||||
smtp.send_message.side_effect = smtplib.SMTPDataError(
|
||||
551, b"5.7.1 Not authorised to send from this header address"
|
||||
)
|
||||
smtp_factory = MagicMock(return_value=_SMTPContext(smtp))
|
||||
|
||||
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP", smtp_factory), patch(
|
||||
"media_library_viewer_api.services.mailer.smtplib.SMTP_SSL"
|
||||
) as smtp_ssl:
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
send_email_message(
|
||||
settings,
|
||||
recipients=["alex@example.com"],
|
||||
subject="Hello",
|
||||
html_body="<p>Hello</p>",
|
||||
)
|
||||
|
||||
smtp_ssl.assert_not_called()
|
||||
self.assertIn("authorized alias", str(ctx.exception).lower())
|
||||
self.assertEqual(smtp.send_message.call_count, 1)
|
||||
|
||||
def test_describe_smtp_error_handles_timeout(self) -> None:
|
||||
detail = describe_smtp_error(TimeoutError("timed out"))
|
||||
self.assertIn("timed out", detail.lower())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -3,7 +3,12 @@
|
||||
import pytest
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from media_library_viewer_api.services.media_index import MediaIndex
|
||||
from typing import Any
|
||||
from media_library_viewer_api.services.media_index import (
|
||||
MediaIndex,
|
||||
MediaIndexBuildCancelled,
|
||||
build_media_index,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -242,3 +247,126 @@ class TestMediaIndexMetadata:
|
||||
index.set_metadata("build_duration_seconds", "12.5")
|
||||
status = index.status()
|
||||
assert status.build_duration_seconds == 12.5
|
||||
|
||||
|
||||
class TestMediaIndexBuildPaths:
|
||||
def test_build_media_index_patches_remote_media_root(self, tmp_path):
|
||||
class FakeClient:
|
||||
def items(self, **kwargs):
|
||||
return {
|
||||
"Items": [
|
||||
{
|
||||
"Id": "m1",
|
||||
"Name": "Movie One",
|
||||
"Type": "Movie",
|
||||
"Path": "/media/movies/Movie One/file.mkv",
|
||||
}
|
||||
],
|
||||
"TotalRecordCount": 1,
|
||||
}
|
||||
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
count = build_media_index(
|
||||
FakeClient(),
|
||||
"user1",
|
||||
[{"Id": "lib1", "Name": "Movies"}],
|
||||
index,
|
||||
media_root="/srv/media",
|
||||
fallback_prefix="",
|
||||
)
|
||||
assert count == 1
|
||||
rows, total = index.query(library_ids=["lib1"], media_types=["Movie"])
|
||||
assert total == 1
|
||||
assert rows[0]["path"] == "/srv/media/movies/Movie One/file.mkv"
|
||||
|
||||
def test_build_media_index_reports_progress(self, tmp_path):
|
||||
class FakeClient:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def items(self, **kwargs):
|
||||
self.calls.append(kwargs.get("start_index", 0))
|
||||
if kwargs.get("start_index", 0) == 0:
|
||||
return {
|
||||
"Items": [
|
||||
{"Id": "m1", "Name": "Movie One", "Type": "Movie", "Path": "/media/a.mkv"}
|
||||
],
|
||||
"TotalRecordCount": 2,
|
||||
}
|
||||
return {
|
||||
"Items": [
|
||||
{"Id": "m2", "Name": "Movie Two", "Type": "Movie", "Path": "/media/b.mkv"}
|
||||
],
|
||||
"TotalRecordCount": 2,
|
||||
}
|
||||
|
||||
events: list[dict[str, Any]] = []
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
count = build_media_index(
|
||||
FakeClient(),
|
||||
"user1",
|
||||
[{"Id": "lib1", "Name": "Movies"}],
|
||||
index,
|
||||
page_size=1,
|
||||
progress_callback=events.append,
|
||||
)
|
||||
assert count == 2
|
||||
assert events[0]["stage"] == "starting"
|
||||
assert events[0]["progress"] is None
|
||||
assert any(event["stage"] == "building" for event in events)
|
||||
building_event = next(event for event in events if event["stage"] == "building")
|
||||
assert building_event["library"] == "Movies"
|
||||
assert building_event["library_progress"] in (0.5, 1.0)
|
||||
assert "elapsed_seconds" in building_event
|
||||
assert "eta_seconds" in building_event
|
||||
assert events[-1]["stage"] == "completed"
|
||||
assert events[-1]["progress"] == 1.0
|
||||
assert events[-1]["library_progress"] == 1.0
|
||||
|
||||
def test_build_media_index_can_be_cancelled(self, tmp_path):
|
||||
class FakeClient:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def items(self, **kwargs):
|
||||
self.calls.append(kwargs.get("start_index", 0))
|
||||
if kwargs.get("start_index", 0) == 0:
|
||||
return {
|
||||
"Items": [
|
||||
{"Id": "m1", "Name": "Movie One", "Type": "Movie", "Path": "/media/a.mkv"}
|
||||
],
|
||||
"TotalRecordCount": 2,
|
||||
}
|
||||
return {
|
||||
"Items": [
|
||||
{"Id": "m2", "Name": "Movie Two", "Type": "Movie", "Path": "/media/b.mkv"}
|
||||
],
|
||||
"TotalRecordCount": 2,
|
||||
}
|
||||
|
||||
events: list[dict[str, Any]] = []
|
||||
cancel_after_building = [False]
|
||||
|
||||
def progress_callback(state: dict[str, Any]) -> None:
|
||||
events.append(state)
|
||||
if state["stage"] == "building":
|
||||
cancel_after_building[0] = True
|
||||
|
||||
def should_cancel() -> bool:
|
||||
return cancel_after_building[0]
|
||||
|
||||
index = MediaIndex(tmp_path / "index.sqlite")
|
||||
client = FakeClient()
|
||||
with pytest.raises(MediaIndexBuildCancelled):
|
||||
build_media_index(
|
||||
client,
|
||||
"user1",
|
||||
[{"Id": "lib1", "Name": "Movies"}],
|
||||
index,
|
||||
page_size=1,
|
||||
progress_callback=progress_callback,
|
||||
should_cancel=should_cancel,
|
||||
)
|
||||
assert any(event["stage"] == "building" for event in events)
|
||||
assert events[-1]["stage"] == "building"
|
||||
assert client.calls == [0]
|
||||
|
||||
Reference in New Issue
Block a user