alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7d49df3e7d6446bb92dc5a6c10eaae191c6eba27
manage/backend/tests/test_api.py
T
Developer d4f95b64d4 chore(observability): externalize stack from root compose files 
Manage now connects to existing Grafana/Prometheus/Alertmanager instances
and never deploys its own stack.

- docker-compose.yml / docker-compose.dev.yml: removed prometheus, loki,
  alloy, grafana, alertmanager, node-exporter services, the monitoring
  network, and observability named volumes; they now ship only backend +
  frontend. Dev frontend now joins the web network so the Vite dev proxy
  can reach the backend.
- backend: alertmanager_url default is now empty; /api/monitoring/alerts
  and /alertmanager-status return graceful "not configured" responses
  when ALERTMANAGER_URL is unset. Added not-configured tests.
- docker-compose.observability.yml: kept as the optional standalone
  example; header clarifies Manage does not deploy it.
- Removed orphaned combined monitoring/prometheus/prometheus.yml
  (standalone stack uses prometheus.standalone.yml).
- Docs (README, REQUIREMENTS decision log, monitoring-logging-design,
  observability-runbooks, context.md, MIGRATION_PLAN, frontend/README,
  CHANGELOG) updated to the connect-to-existing model.

VITE_GRAFANA_URL / VITE_PROMETHEUS_URL remain as optional frontend
deep-link overrides. .env.example still needs a manual update (safety
policy blocks assistant edits): set ALERTMANAGER_URL empty/optional and
move standalone-only vars out of the root file.
2026-06-23 21:20:07 +00:00

775 lines
31 KiB
Python
Raw Blame History

"""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
# --- 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
# --- 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 TestSettingsMachines:
def test_post_machine_rewrites_prometheus_targets(self, test_client):
with patch("media_library_viewer_api.routers.settings.write_prometheus_targets") as write_targets:
with patch("media_library_viewer_api.routers.settings._validate_saved_machine_ssh"):
response = test_client.post(
"/api/settings/machines",
json={
"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",
},
)
assert response.status_code == 201
write_targets.assert_called_once()
def test_delete_machine_rewrites_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",
}
)
with patch("media_library_viewer_api.routers.settings.write_prometheus_targets") as write_targets:
response = test_client.delete(f"/api/settings/machines/{machine['id']}")
assert response.status_code == 200
write_targets.assert_called_once()
class TestAlertmanager:
def test_alerts_endpoint_when_alertmanager_unreachable(self, test_client):
with patch("media_library_viewer_api.routers.monitoring._alertmanager_client") as mock_client:
session = MagicMock()
session.get.side_effect = Exception("connection refused")
mock_client.return_value = (session, "http://alertmanager:9093")
response = test_client.get("/api/monitoring/alerts")
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["error"] == "alertmanager_unreachable"
def test_alerts_endpoint_when_not_configured(self, test_client):
with patch("media_library_viewer_api.routers.monitoring._alertmanager_client") as mock_client:
session = MagicMock()
mock_client.return_value = (session, "")
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"
session.get.assert_not_called()
def test_alerts_endpoint_returns_summary(self, test_client):
with patch("media_library_viewer_api.routers.monitoring._alertmanager_client") as mock_client:
session = MagicMock()
session.get.return_value.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",
}
],
}
session.get.return_value.raise_for_status = MagicMock()
mock_client.return_value = (session, "http://alertmanager:9093")
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"
def test_alertmanager_status_endpoint_when_not_configured(self, test_client):
with patch("media_library_viewer_api.routers.monitoring._alertmanager_client") as mock_client:
session = MagicMock()
mock_client.return_value = (session, "")
response = test_client.get("/api/monitoring/alertmanager-status")
assert response.status_code == 200
data = response.json()
assert data["up"] is False
session.get.assert_not_called()
def test_alertmanager_status_endpoint_when_unreachable(self, test_client):
with patch("media_library_viewer_api.routers.monitoring._alertmanager_client") as mock_client:
session = MagicMock()
session.get.side_effect = Exception("connection refused")
mock_client.return_value = (session, "http://alertmanager:9093")
response = test_client.get("/api/monitoring/alertmanager-status")
assert response.status_code == 200
data = response.json()
assert data["up"] is False
class TestAlertmanagerWebhook:
"""Tests for POST /api/monitoring/alertmanager-webhook."""
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
def test_webhook_forwards_when_configured(self, test_client, caplog, monkeypatch):
payload = {"status": "resolved", "alerts": []}
forwarded = {"captured": False}
class FakeResponse:
def raise_for_status(self):
pass
def fake_post(url, json, timeout):
forwarded["captured"] = True
forwarded["url"] = url
forwarded["payload"] = json
return FakeResponse()
monkeypatch.setattr("requests.Session.post", lambda _self, url, json, timeout: fake_post(url, json, timeout))
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 forwarded["captured"] is False
Reference in New Issue View Git Blame Copy Permalink