Developer
7d252489de
On a fresh deploy with no Jellyfin service configured yet, get_jellyfin_client (and get_user_id / get_ssh_client) raised a plain RuntimeError, which bubbled up as a 500 traceback on every Jellyfin-dependent route (dashboard counts/libraries/activity, media, users). Convert those RuntimeErrors to HTTPException(503) with a clear detail message so FastAPI returns a clean 503 JSON response instead of a 500, and the frontend can render a not-configured state. - dependencies.py: get_jellyfin_client (no service / missing creds), get_user_id (no users discovered), and get_ssh_client (no SSH machine + no legacy key path) now raise HTTPException(503, detail=...). - tests/test_api.py: added TestDashboard.test_jellyfin_endpoints_return_503_when_not_configured covering /api/dashboard/counts and /activity. ruff clean; 240 backend tests pass.
977 lines
39 KiB
Python
977 lines
39 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.clients.ssh import CommandResult
|
|
from media_library_viewer_api.dependencies import (
|
|
get_jellyfin_client,
|
|
get_jellyseerr_client,
|
|
get_mail_queue,
|
|
get_settings_store,
|
|
get_ssh_client,
|
|
get_user_id,
|
|
)
|
|
from media_library_viewer_api.main import app
|
|
from media_library_viewer_api.routers.media import get_media_index
|
|
from media_library_viewer_api.services.media_index import MediaIndex
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
from media_library_viewer_api.widgets.sources import ServiceRecord
|
|
|
|
# Short alias for the monitoring router module under test.
|
|
_MON = "media_library_viewer_api.routers.monitoring"
|
|
|
|
# --- 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, tmp_path):
|
|
"""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"
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
app.dependency_overrides[get_settings_store] = lambda: store
|
|
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"}
|
|
|
|
def test_metrics_endpoint(self, test_client):
|
|
response = test_client.get("/metrics")
|
|
assert response.status_code == 200
|
|
assert "manage_api_requests_total" in response.text
|
|
assert response.headers["content-type"].startswith("text/plain")
|
|
|
|
def test_request_id_header(self, test_client):
|
|
response = test_client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert "x-request-id" in response.headers
|
|
assert len(response.headers["x-request-id"]) > 0
|
|
|
|
|
|
# --- 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
|
|
|
|
def test_jellyfin_endpoints_return_503_when_not_configured(self, test_client):
|
|
# Remove the mocked Jellyfin dependency so the real one runs; with no
|
|
# Jellyfin service seeded, endpoints must degrade to 503, not 500.
|
|
app.dependency_overrides.pop(get_jellyfin_client, None)
|
|
app.dependency_overrides.pop(get_user_id, None)
|
|
try:
|
|
for path in ("/api/dashboard/counts", "/api/dashboard/activity"):
|
|
response = test_client.get(path)
|
|
assert response.status_code == 503, path
|
|
detail = response.json()["detail"]
|
|
assert "configured" in detail, path
|
|
finally:
|
|
# Restore the mocks for subsequent tests in this fixture session.
|
|
app.dependency_overrides[get_jellyfin_client] = lambda: MagicMock()
|
|
app.dependency_overrides[get_user_id] = lambda: "user123"
|
|
|
|
|
|
# --- Settings reset ---
|
|
|
|
|
|
class TestSettingsReset:
|
|
def test_reset_local_database_requires_full_confirmation(self, test_client, tmp_path):
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
media_db = tmp_path / "media.sqlite"
|
|
media_db.write_text("placeholder", encoding="utf-8")
|
|
media_wal = tmp_path / "media.sqlite-wal"
|
|
media_wal.write_text("wal", encoding="utf-8")
|
|
|
|
app.dependency_overrides[get_settings_store] = lambda: store
|
|
settings_module = __import__("media_library_viewer_api.routers.settings", fromlist=["MediaIndex"])
|
|
original_media_index = settings_module.MediaIndex
|
|
settings_module.MediaIndex = lambda: SimpleNamespace(db_path=media_db)
|
|
try:
|
|
response = test_client.post(
|
|
"/api/settings/reset-local-database",
|
|
json={
|
|
"confirm_phrase": "RESET LOCAL DATABASE",
|
|
"acknowledge_settings_loss": True,
|
|
"acknowledge_media_index_loss": False,
|
|
"acknowledge_irreversible": True,
|
|
},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.pop(get_settings_store, None)
|
|
settings_module.MediaIndex = original_media_index
|
|
assert response.status_code == 400
|
|
|
|
def test_reset_local_database_wipes_state_and_leaves_no_machines(self, test_client, tmp_path):
|
|
store = SettingsStore(tmp_path / "settings.sqlite")
|
|
store.ensure_defaults()
|
|
media_db = tmp_path / "media.sqlite"
|
|
media_db.write_text("placeholder", encoding="utf-8")
|
|
media_wal = tmp_path / "media.sqlite-wal"
|
|
media_wal.write_text("wal", encoding="utf-8")
|
|
|
|
app.dependency_overrides[get_settings_store] = lambda: store
|
|
settings_module = __import__("media_library_viewer_api.routers.settings", fromlist=["MediaIndex"])
|
|
original_media_index = settings_module.MediaIndex
|
|
settings_module.MediaIndex = lambda: SimpleNamespace(db_path=media_db)
|
|
try:
|
|
response = test_client.post(
|
|
"/api/settings/reset-local-database",
|
|
json={
|
|
"confirm_phrase": "RESET LOCAL DATABASE",
|
|
"acknowledge_settings_loss": True,
|
|
"acknowledge_media_index_loss": True,
|
|
"acknowledge_irreversible": True,
|
|
},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.pop(get_settings_store, None)
|
|
settings_module.MediaIndex = original_media_index
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["status"] == "reset"
|
|
assert not media_db.exists()
|
|
assert not media_wal.exists()
|
|
assert store.get_machine("local") is None
|
|
assert len(store.list_machines()) == 0
|
|
|
|
|
|
# --- 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_impl.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_prometheus_targets_empty(self, test_client):
|
|
response = test_client.get("/api/monitoring/prometheus-targets")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
def test_prometheus_targets_returns_enabled_ssh_node_exporter(self, test_client):
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
store.upsert_machine(
|
|
{
|
|
"name": "remote1",
|
|
"mode": "ssh",
|
|
"enabled": True,
|
|
"services": ["monitoring"],
|
|
"host": "10.0.0.5",
|
|
"username": "u",
|
|
"node_exporter_enabled": True,
|
|
"node_exporter_port": 9200,
|
|
"node_exporter_scrape_host": "1.2.3.4",
|
|
}
|
|
)
|
|
response = test_client.get("/api/monitoring/prometheus-targets")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 1
|
|
assert data[0]["targets"] == ["1.2.3.4:9200"]
|
|
assert data[0]["labels"]["job"] == "node-exporter-remote"
|
|
|
|
|
|
class TestResolveServiceRecord:
|
|
"""Unit tests for _resolve_service_record (service_id + first-enabled paths)."""
|
|
|
|
def _store(self, rows):
|
|
store = MagicMock()
|
|
store.get_service = lambda sid: next((r for r in rows if r["id"] == sid), None)
|
|
|
|
def list_filtered(service_type=None):
|
|
return [r for r in rows if r["service_type"] == (service_type or r["service_type"])]
|
|
|
|
store.list_services = list_filtered
|
|
return store
|
|
|
|
def test_service_id_match_returns_record(self):
|
|
from media_library_viewer_api.routers.monitoring import _resolve_service_record
|
|
|
|
row = {"id": "am1", "service_type": "alertmanager", "name": "AM", "enabled": True, "config": {}, "secrets": {}}
|
|
store = self._store([row])
|
|
with patch(f"{_MON}.build_service_record", return_value="RECORD") as mock_build:
|
|
result = _resolve_service_record(store, "alertmanager", "am1")
|
|
assert result == "RECORD"
|
|
mock_build.assert_called_once_with(store, row)
|
|
|
|
def test_service_id_type_mismatch_returns_none(self):
|
|
from media_library_viewer_api.routers.monitoring import _resolve_service_record
|
|
|
|
row = {"id": "x1", "service_type": "grafana", "name": "G", "enabled": True, "config": {}, "secrets": {}}
|
|
store = self._store([row])
|
|
assert _resolve_service_record(store, "alertmanager", "x1") is None
|
|
|
|
def test_service_id_disabled_returns_none(self):
|
|
from media_library_viewer_api.routers.monitoring import _resolve_service_record
|
|
|
|
row = {"id": "am1", "service_type": "alertmanager", "name": "AM", "enabled": False, "config": {}, "secrets": {}}
|
|
store = self._store([row])
|
|
assert _resolve_service_record(store, "alertmanager", "am1") is None
|
|
|
|
def test_no_service_id_returns_first_enabled(self):
|
|
from media_library_viewer_api.routers.monitoring import _resolve_service_record
|
|
|
|
rows = [
|
|
{
|
|
"id": "am1",
|
|
"service_type": "alertmanager",
|
|
"name": "Disabled",
|
|
"enabled": False,
|
|
"config": {},
|
|
"secrets": {},
|
|
},
|
|
{
|
|
"id": "am2",
|
|
"service_type": "alertmanager",
|
|
"name": "Active",
|
|
"enabled": True,
|
|
"config": {},
|
|
"secrets": {},
|
|
},
|
|
]
|
|
store = self._store(rows)
|
|
with patch(f"{_MON}.build_service_record", return_value="RECORD") as mock_build:
|
|
result = _resolve_service_record(store, "alertmanager", None)
|
|
assert result == "RECORD"
|
|
mock_build.assert_called_once_with(store, rows[1])
|
|
|
|
def test_no_service_id_and_none_enabled_returns_none(self):
|
|
from media_library_viewer_api.routers.monitoring import _resolve_service_record
|
|
|
|
store = self._store([])
|
|
assert _resolve_service_record(store, "alertmanager", None) is None
|
|
|
|
|
|
class TestSettingsMachines:
|
|
def test_machine_appears_in_prometheus_targets(self, test_client):
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
store.upsert_machine(
|
|
{
|
|
"name": "remote1",
|
|
"mode": "ssh",
|
|
"enabled": True,
|
|
"services": ["monitoring"],
|
|
"host": "10.0.0.5",
|
|
"username": "u",
|
|
"node_exporter_enabled": True,
|
|
"node_exporter_port": 9200,
|
|
"node_exporter_scrape_host": "1.2.3.4",
|
|
}
|
|
)
|
|
targets = test_client.get("/api/monitoring/prometheus-targets").json()
|
|
assert len(targets) == 1
|
|
assert targets[0]["targets"] == ["1.2.3.4:9200"]
|
|
|
|
def test_delete_machine_removed_from_prometheus_targets(self, test_client):
|
|
store = app.dependency_overrides[get_settings_store]()
|
|
machine = store.upsert_machine(
|
|
{
|
|
"name": "remote1",
|
|
"mode": "ssh",
|
|
"enabled": True,
|
|
"services": ["monitoring"],
|
|
"host": "10.0.0.5",
|
|
"username": "u",
|
|
"node_exporter_enabled": True,
|
|
"node_exporter_port": 9200,
|
|
"node_exporter_scrape_host": "1.2.3.4",
|
|
}
|
|
)
|
|
response = test_client.delete(f"/api/settings/machines/{machine['id']}")
|
|
assert response.status_code == 200
|
|
assert test_client.get("/api/monitoring/prometheus-targets").json() == []
|
|
|
|
|
|
def _am_service(name="Alertmanager", **config):
|
|
cfg = {"base_url": "http://alertmanager:9093", "timeout_seconds": 5}
|
|
cfg.update(config)
|
|
return ServiceRecord(id="am1", service_type="alertmanager", name=name, config=cfg)
|
|
|
|
|
|
class TestAlertmanager:
|
|
def test_alerts_endpoint_when_not_configured(self, test_client):
|
|
# No alertmanager service seeded -> registry resolves None.
|
|
response = test_client.get("/api/monitoring/alerts")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 0
|
|
assert data["error"] == "alertmanager_not_configured"
|
|
|
|
def test_alerts_endpoint_when_unreachable(self, test_client):
|
|
service = _am_service()
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", side_effect=Exception("connection refused")),
|
|
):
|
|
response = test_client.get("/api/monitoring/alerts")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 0
|
|
assert data["error"] == "alertmanager_unreachable"
|
|
assert data["service_id"] == "am1"
|
|
assert data["name"] == "Alertmanager"
|
|
|
|
def test_alerts_endpoint_returns_summary(self, test_client):
|
|
service = _am_service()
|
|
resp = MagicMock()
|
|
resp.json.return_value = {
|
|
"status": "success",
|
|
"data": [
|
|
{
|
|
"labels": {"alertname": "BackupJobFailed", "severity": "critical", "job_name": "test"},
|
|
"annotations": {"summary": "Backup failed", "description": "details"},
|
|
"startsAt": "2026-05-11T02:00:00Z",
|
|
"status": "firing",
|
|
}
|
|
],
|
|
}
|
|
resp.raise_for_status = MagicMock()
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", return_value=resp),
|
|
):
|
|
response = test_client.get("/api/monitoring/alerts")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 1
|
|
assert data["by_severity"]["critical"] == 1
|
|
assert data["alerts"][0]["name"] == "BackupJobFailed"
|
|
assert data["service_id"] == "am1"
|
|
|
|
def test_alerts_endpoint_sends_bearer_token(self, test_client):
|
|
service = _am_service()
|
|
service.secrets = {"api_key": "tok"}
|
|
resp = MagicMock()
|
|
resp.json.return_value = {"status": "success", "data": []}
|
|
resp.raise_for_status = MagicMock()
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", return_value=resp) as mock_get,
|
|
):
|
|
test_client.get("/api/monitoring/alerts")
|
|
_, kwargs = mock_get.call_args
|
|
assert kwargs["headers"]["Authorization"] == "Bearer tok"
|
|
|
|
def test_alertmanager_status_when_not_configured(self, test_client):
|
|
response = test_client.get("/api/monitoring/alertmanager-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is False
|
|
assert data["error"] == "alertmanager_not_configured"
|
|
assert data["peers"] == []
|
|
|
|
def test_alertmanager_status_when_unreachable(self, test_client):
|
|
service = _am_service()
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", side_effect=Exception("refused")),
|
|
):
|
|
response = test_client.get("/api/monitoring/alertmanager-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is False
|
|
assert data["error"] == "alertmanager_unreachable"
|
|
assert data["name"] == "Alertmanager"
|
|
|
|
def test_alertmanager_status_returns_cluster_info(self, test_client):
|
|
service = _am_service()
|
|
resp = MagicMock()
|
|
resp.json.return_value = {
|
|
"versionInfo": {"version": "0.27.0"},
|
|
"clusterStatus": {"createdAt": "2026-06-01T00:00:00Z"},
|
|
"cluster": {"peers": [{"name": "am-1"}, {"name": "am-2"}]},
|
|
}
|
|
resp.raise_for_status = MagicMock()
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", return_value=resp),
|
|
):
|
|
response = test_client.get("/api/monitoring/alertmanager-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is True
|
|
assert data["version"] == "0.27.0"
|
|
assert data["peers"] == ["am-1", "am-2"]
|
|
|
|
|
|
class TestAlertmanagerWebhook:
|
|
"""Tests for POST /api/monitoring/alertmanager-webhook (log-only receiver)."""
|
|
|
|
def test_webhook_receives_payload_and_logs(self, test_client, caplog):
|
|
payload = {
|
|
"status": "firing",
|
|
"alerts": [
|
|
{
|
|
"status": "firing",
|
|
"labels": {"alertname": "BackupJobFailed", "severity": "critical"},
|
|
"annotations": {"summary": "Backup failed"},
|
|
"startsAt": "2026-06-13T10:00:00Z",
|
|
}
|
|
],
|
|
}
|
|
with caplog.at_level("INFO", logger="media_library_viewer_api.routers.monitoring"):
|
|
response = test_client.post("/api/monitoring/alertmanager-webhook", json=payload)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "received"
|
|
assert "Received Alertmanager webhook with 1 alert(s)" in caplog.text
|
|
|
|
|
|
class TestGrafanaStatus:
|
|
def test_grafana_status_when_not_configured(self, test_client):
|
|
response = test_client.get("/api/monitoring/grafana-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is False
|
|
assert data["error"] == "no_service_configured"
|
|
|
|
def test_grafana_status_when_unreachable(self, test_client):
|
|
service = ServiceRecord(
|
|
id="g1", service_type="grafana", name="Grafana", config={"base_url": "http://grafana:3000"}
|
|
)
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", side_effect=Exception("refused")),
|
|
):
|
|
response = test_client.get("/api/monitoring/grafana-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is False
|
|
assert data["error"] == "grafana_unreachable"
|
|
assert data["name"] == "Grafana"
|
|
|
|
def test_grafana_status_returns_version(self, test_client):
|
|
service = ServiceRecord(
|
|
id="g1", service_type="grafana", name="Grafana", config={"base_url": "http://grafana:3000"}
|
|
)
|
|
resp = MagicMock()
|
|
resp.json.return_value = {"version": "11.3.1", "database": "ok"}
|
|
resp.raise_for_status = MagicMock()
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", return_value=resp),
|
|
):
|
|
response = test_client.get("/api/monitoring/grafana-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is True
|
|
assert data["version"] == "11.3.1"
|
|
assert data["service_id"] == "g1"
|
|
|
|
|
|
class TestPrometheusStatus:
|
|
def test_prometheus_status_when_not_configured(self, test_client):
|
|
response = test_client.get("/api/monitoring/prometheus-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is False
|
|
assert data["error"] == "no_service_configured"
|
|
|
|
def test_prometheus_status_when_unreachable(self, test_client):
|
|
service = ServiceRecord(
|
|
id="p1", service_type="prometheus", name="Prometheus", config={"base_url": "http://prometheus:9090"}
|
|
)
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", side_effect=Exception("refused")),
|
|
):
|
|
response = test_client.get("/api/monitoring/prometheus-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is False
|
|
assert data["error"] == "prometheus_unreachable"
|
|
|
|
def test_prometheus_status_returns_version(self, test_client):
|
|
service = ServiceRecord(
|
|
id="p1", service_type="prometheus", name="Prometheus", config={"base_url": "http://prometheus:9090"}
|
|
)
|
|
health = MagicMock()
|
|
health.raise_for_status = MagicMock()
|
|
build_info = MagicMock()
|
|
build_info.raise_for_status = MagicMock()
|
|
build_info.json.return_value = {"status": "success", "data": {"version": "2.55.1"}}
|
|
with (
|
|
patch(f"{_MON}._resolve_service_record", return_value=service),
|
|
patch(f"{_MON}.requests.get", side_effect=[health, build_info]),
|
|
):
|
|
response = test_client.get("/api/monitoring/prometheus-status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["up"] is True
|
|
assert data["version"] == "2.55.1"
|
|
assert data["service_id"] == "p1"
|