560 lines
22 KiB
Python
560 lines
22 KiB
Python
"""Integration tests for FastAPI endpoints using TestClient.
|
|
|
|
These tests mock the SSH and Jellyfin clients to test the API layer
|
|
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_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 ---
|
|
|
|
@pytest.fixture
|
|
def mock_jellyfin():
|
|
"""Mock Jellyfin client."""
|
|
client = MagicMock()
|
|
client.media_counts.return_value = {"movies": 100, "series": 20, "episodes": 500}
|
|
client.libraries.return_value = [
|
|
{"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.sessions.return_value = [
|
|
{
|
|
"Id": "sess1",
|
|
"UserName": "alex",
|
|
"DeviceName": "Chrome",
|
|
"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."""
|
|
client = MagicMock()
|
|
client.host = "test-host"
|
|
# Default list_dir response
|
|
entries = [
|
|
{"type": "d", "size": 4096, "mtime": 1700000000, "name": "Movies"},
|
|
{"type": "f", "size": 5000000000, "mtime": 1700000100, "name": "movie.mkv"},
|
|
]
|
|
client.list_dir.return_value = CommandResult(
|
|
command="find ...",
|
|
exit_status=0,
|
|
stdout=json.dumps(entries),
|
|
stderr="",
|
|
)
|
|
client.stat_path.return_value = CommandResult(
|
|
command="stat ...",
|
|
exit_status=0,
|
|
stdout="regular file\n5000000000 bytes\n2024-01-01\n/path",
|
|
stderr="",
|
|
)
|
|
client.ffprobe_json.return_value = {
|
|
"format": {"filename": "movie.mkv", "format_name": "matroska"},
|
|
"streams": [{"codec_type": "video", "codec_name": "hevc"}],
|
|
}
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
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"
|
|
auth_settings = SimpleNamespace(auth_enabled=False)
|
|
with patch("media_library_viewer_api.auth.get_settings", return_value=auth_settings):
|
|
client = TestClient(app)
|
|
yield client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# --- Health ---
|
|
|
|
class TestHealth:
|
|
def test_health(self, test_client):
|
|
response = test_client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
|
|
|
|
# --- Dashboard ---
|
|
|
|
class TestDashboard:
|
|
def test_counts(self, test_client):
|
|
response = test_client.get("/api/dashboard/counts")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["movies"] == 100
|
|
assert data["series"] == 20
|
|
assert data["episodes"] == 500
|
|
|
|
def test_libraries(self, test_client):
|
|
response = test_client.get("/api/dashboard/libraries")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert data[0]["library"] == "Movies"
|
|
assert data[1]["library"] == "TV Shows"
|
|
|
|
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) == 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_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="Manage",
|
|
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:
|
|
def test_list_directory(self, test_client):
|
|
response = test_client.get("/api/files/list", params={"path": "/media"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["path"] == "/media"
|
|
assert data["count"] == 2
|
|
assert data["entries"][0]["name"] == "Movies"
|
|
|
|
def test_ffprobe(self, test_client):
|
|
response = test_client.get("/api/files/ffprobe", params={"path": "/media/movie.mkv"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["format"]["format_name"] == "matroska"
|
|
assert data["streams"][0]["codec_name"] == "hevc"
|
|
|
|
def test_stat(self, test_client):
|
|
response = test_client.get("/api/files/stat", params={"path": "/media/movie.mkv"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["path"] == "/media/movie.mkv"
|
|
assert "bytes" in data["output"]
|
|
|
|
def test_resolve_path(self, test_client):
|
|
response = test_client.get("/api/files/resolve-path", params={"path": "/media/shows/X"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["original"] == "/media/shows/X"
|
|
assert "resolved" in data
|
|
|
|
def test_list_failure(self, test_client, mock_ssh):
|
|
mock_ssh.list_dir.return_value = CommandResult(
|
|
command="find ...", exit_status=1, stdout="", stderr="Permission denied"
|
|
)
|
|
response = test_client.get("/api/files/list", params={"path": "/root"})
|
|
assert response.status_code == 400
|
|
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:
|
|
def test_list_templates(self, test_client):
|
|
response = test_client.get("/api/jobs/templates")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) >= 3
|
|
keys = [t["key"] for t in data]
|
|
assert "disk_usage" in keys
|
|
assert "ffprobe" in keys
|
|
|
|
def test_run_job(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="du -sh '/media/test'",
|
|
exit_status=0,
|
|
stdout="5.0G\t/media/test\n",
|
|
stderr="",
|
|
)
|
|
response = test_client.post("/api/jobs/run", json={"job_key": "disk_usage", "path": "/media/test"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["exit_status"] == 0
|
|
assert "5.0G" in data["stdout"]
|
|
|
|
def test_run_unknown_job(self, test_client):
|
|
response = test_client.post("/api/jobs/run", json={"job_key": "nonexistent", "path": "/x"})
|
|
assert response.status_code == 400
|
|
|
|
|
|
# --- Monitoring ---
|
|
|
|
class TestMonitoring:
|
|
def test_status(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="...", exit_status=0, stdout="running pid=1234\n", stderr=""
|
|
)
|
|
response = test_client.get("/api/monitoring/status")
|
|
assert response.status_code == 200
|
|
assert "running" in response.json()["status"]
|
|
|
|
def test_metrics_empty(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="...", exit_status=0, stdout="", stderr=""
|
|
)
|
|
response = test_client.get("/api/monitoring/metrics")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["samples"] == []
|
|
|
|
def test_disk(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="df ...",
|
|
exit_status=0,
|
|
stdout='{"filesystem":"/dev/sda1","size":1000000000,"used":500000000,"available":500000000,"used_pct":"50%","mount":"/"}',
|
|
stderr="",
|
|
)
|
|
response = test_client.get("/api/monitoring/disk")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["used_pct"] == "50%"
|
|
|
|
def test_start(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="...", exit_status=0, stdout="started pid=5678\n", stderr=""
|
|
)
|
|
response = test_client.post("/api/monitoring/start")
|
|
assert response.status_code == 200
|
|
assert "started" in response.json()["message"]
|
|
|
|
def test_stop(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="...", exit_status=0, stdout="stopped pid=5678\n", stderr=""
|
|
)
|
|
response = test_client.post("/api/monitoring/stop")
|
|
assert response.status_code == 200
|
|
|
|
def test_restart(self, test_client, mock_ssh):
|
|
mock_ssh.run.return_value = CommandResult(
|
|
command="...", exit_status=0, stdout="stopped pid=5678\nstarted pid=9999\n", stderr=""
|
|
)
|
|
response = test_client.post("/api/monitoring/restart")
|
|
assert response.status_code == 200
|