alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1da67f38c7b1ea22d1b74f40f0fc57aa7fef62c7
manage/backend/tests/test_api.py
T
Developer 08a3b616f6 refactor(monitoring): decommission legacy SSH-scraping poller (slice 1) 
The 2026-06-16/17 observability update externalised metrics to
Prometheus + node_exporter + Grafana, but the legacy Manage-side
SSH-scraping monitor was never removed. It duplicated the new stack,
ran SSH df on every machine every 300s, and fed nothing (its UI was
deleted in e2ad731). This slice decommissions the duplication.

Removed (backend):
- services/monitoring_poller.py (MonitoringPoller) — entire file
- services/monitoring_actions.py (disk_space, run_machine_operation,
  poll_machine_snapshot, build_machine_client) — entire file;
  run_machine_operation had only 2 callers (the poller + /disk), both gone
- tests/test_monitoring_actions.py
- endpoints: POST /api/monitoring/poller, GET /machines/{id}/actions,
  GET /disk (and the now-dead _resolve_machine helper)
- lifespan wiring (main.py), dependency wrapper (dependencies.py),
  poller.start()/kick() from machine save (routers/settings.py)
- SettingsStore: monitoring_machine_actions table CREATE + 2 indexes +
  record/list/prune_machine_actions methods; DROP TABLE IF EXISTS on
  startup cleans existing DBs (user-approved)
- config knobs: monitoring_poll_interval_seconds,
  monitoring_poll_initial_delay_seconds, monitoring_action_retention_days
- test_api.py: TestMonitoring._ensure_machine + test_disk

Kept (fits the new model): /machines, /prometheus-targets, /alerts,
/alertmanager-status, /alertmanager-webhook; the disk_usage JOB template
(manual on-demand, not monitoring); node_exporter_* machine fields
(they point Prometheus at the right host).

Gate: backend pytest 173 passed; ruff clean.
2026-06-17 20:48:56 +00:00

754 lines
30 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_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_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