alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dedup _resolve_service_record into shared service_resolution module

Browse Source 
Extract the duplicated _resolve_service_record helper (identical in
routers/monitoring.py and routers/authentik_users.py) into a shared
services/service_resolution.py module. Both routers now import
resolve_service_record from the shared module.

The authentik router previously hardcoded service_type='authentik' in
its local copy; the shared helper takes service_type as a param (same
as monitoring's did).

Tests updated: test_api.py patches now target the correct module paths
(resolve_service_record on the monitoring module where it's imported,
build_service_record on the service_resolution module).

283 backend tests pass; ruff clean.
This commit is contained in:
Developer
2026-07-06 12:25:17 +00:00
parent 1e636fdbe2
commit 691d78ff06
4 changed files with 71 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_api.py
+23 -22
View File
@@ -26,6 +26,7 @@ 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"
_SVC = "media_library_viewer_api.services.service_resolution"
# --- Fixtures ---
@@ -496,7 +497,7 @@ class TestMonitoring:
class TestResolveServiceRecord:
"""Unit tests for _resolve_service_record (service_id + first-enabled paths)."""
"""Unit tests for resolve_service_record (service_id + first-enabled paths)."""
def _store(self, rows):
store = MagicMock()
@@ -509,31 +510,31 @@ class TestResolveServiceRecord:
return store
def test_service_id_match_returns_record(self):
from media_library_viewer_api.routers.monitoring import _resolve_service_record
from media_library_viewer_api.services.service_resolution 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")
with patch(f"{_SVC}.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
from media_library_viewer_api.services.service_resolution 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
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
from media_library_viewer_api.services.service_resolution 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
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
from media_library_viewer_api.services.service_resolution import resolve_service_record
rows = [
{
@@ -554,16 +555,16 @@ class TestResolveServiceRecord:
},
]
store = self._store(rows)
with patch(f"{_MON}.build_service_record", return_value="RECORD") as mock_build:
result = _resolve_service_record(store, "alertmanager", None)
with patch(f"{_SVC}.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
from media_library_viewer_api.services.service_resolution import resolve_service_record
store = self._store([])
assert _resolve_service_record(store, "alertmanager", None) is None
assert resolve_service_record(store, "alertmanager", None) is None
class TestSettingsMachines:
@@ -624,7 +625,7 @@ class TestAlertmanager:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", side_effect=Exception("connection refused")),
):
response = test_client.get("/api/monitoring/alerts")
@@ -651,7 +652,7 @@ class TestAlertmanager:
}
resp.raise_for_status = MagicMock()
with (
patch(f"{_MON}._resolve_service_record", return_value=service),
patch(f"{_MON}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", return_value=resp),
):
response = test_client.get("/api/monitoring/alerts")
@@ -669,7 +670,7 @@ class TestAlertmanager:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", return_value=resp) as mock_get,
):
test_client.get("/api/monitoring/alerts")
@@ -687,7 +688,7 @@ class TestAlertmanager:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", side_effect=Exception("refused")),
):
response = test_client.get("/api/monitoring/alertmanager-status")
@@ -707,7 +708,7 @@ class TestAlertmanager:
}
resp.raise_for_status = MagicMock()
with (
patch(f"{_MON}._resolve_service_record", return_value=service),
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")
@@ -753,7 +754,7 @@ class TestGrafanaStatus:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", side_effect=Exception("refused")),
):
response = test_client.get("/api/monitoring/grafana-status")
@@ -771,7 +772,7 @@ class TestGrafanaStatus:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", return_value=resp),
):
response = test_client.get("/api/monitoring/grafana-status")
@@ -795,7 +796,7 @@ class TestPrometheusStatus:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", side_effect=Exception("refused")),
):
response = test_client.get("/api/monitoring/prometheus-status")
@@ -814,7 +815,7 @@ class TestPrometheusStatus:
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}.resolve_service_record", return_value=service),
patch(f"{_MON}.requests.get", side_effect=[health, build_info]),
):
response = test_client.get("/api/monitoring/prometheus-status")