Developer
3391fbc85d
TestResult dataclass + translate_connection_error shared helper in base.py. test_callable field on ServiceDefinition (default None). 7 per-type test_connection routines (qbittorrent, prometheus via Grafana gateway, alertmanager, jellyfin, authentik, ssh_tasks via build_ssh_client, nextcloud). POST /api/services/test endpoint: validation-first (422 on malformed config), dispatch, no-persistence, no-secret-logs. backups has test_callable=None. qBit 'Fails.' → specific auth message (resolves #3 at API layer). Backend: 362 pytest pass (+31 new), ruff clean. Frontend: build green.
261 lines
12 KiB
Python
261 lines
12 KiB
Python
"""Tests for the service credential tester (CT-101..CT-113, CT-119)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import requests
|
|
|
|
from media_library_viewer_api.integrations.alertmanager import test_connection as am_test
|
|
from media_library_viewer_api.integrations.authentik import test_connection as ak_test
|
|
from media_library_viewer_api.integrations.base import translate_connection_error
|
|
from media_library_viewer_api.integrations.jellyfin import test_connection as jf_test
|
|
from media_library_viewer_api.integrations.nextcloud import test_connection as nc_test
|
|
from media_library_viewer_api.integrations.prometheus import test_connection as prom_test
|
|
from media_library_viewer_api.integrations.qbittorrent import test_connection as qbit_test
|
|
from media_library_viewer_api.integrations.ssh_tasks import test_connection as ssh_test
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# translate_connection_error (CT-119)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTranslateConnectionError:
|
|
def test_http_401_maps_to_auth_message(self) -> None:
|
|
resp = SimpleNamespace(status_code=401)
|
|
exc = requests.HTTPError(response=resp)
|
|
result = translate_connection_error(exc)
|
|
assert result.ok is False
|
|
assert "Authentication failed" in result.detail
|
|
|
|
def test_http_403_maps_to_auth_message(self) -> None:
|
|
resp = SimpleNamespace(status_code=403)
|
|
exc = requests.HTTPError(response=resp)
|
|
result = translate_connection_error(exc)
|
|
assert result.ok is False
|
|
assert "Authentication failed" in result.detail
|
|
|
|
def test_connection_error_dns_maps_to_host_not_found(self) -> None:
|
|
exc = requests.ConnectionError("getaddrinfo failed")
|
|
result = translate_connection_error(exc)
|
|
assert result.ok is False
|
|
assert "Host not found" in result.detail
|
|
|
|
def test_timeout_maps_to_timed_out(self) -> None:
|
|
exc = requests.Timeout("timed out")
|
|
result = translate_connection_error(exc)
|
|
assert result.ok is False
|
|
assert "timed out" in result.detail.lower()
|
|
|
|
def test_generic_fallback_includes_context(self) -> None:
|
|
exc = ValueError("something weird happened")
|
|
result = translate_connection_error(exc, context="qBittorrent")
|
|
assert result.ok is False
|
|
assert "qBittorrent" in result.detail
|
|
assert "something weird happened" in result.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# qbittorrent (CT-104)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestQbittorrentTestConnection:
|
|
def test_success_returns_version(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.maindata.return_value = {"server_state": {"qbittorrent_version": "v4.6.0"}}
|
|
with patch("media_library_viewer_api.integrations.qbittorrent.QbittorrentClient", return_value=mock_client):
|
|
result = qbit_test(
|
|
{"base_url": "http://qb:8080", "timeout_seconds": 5},
|
|
{"username": "u", "password": "p"},
|
|
MagicMock(),
|
|
)
|
|
assert result.ok is True
|
|
assert result.evidence == "v4.6.0"
|
|
|
|
def test_login_failed_translates_to_auth_message(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.maindata.side_effect = RuntimeError("qBittorrent login failed: Fails.")
|
|
with patch("media_library_viewer_api.integrations.qbittorrent.QbittorrentClient", return_value=mock_client):
|
|
result = qbit_test({"base_url": "http://qb:8080"}, {"username": "u", "password": "p"}, MagicMock())
|
|
assert result.ok is False
|
|
assert "Authentication failed" in result.detail
|
|
|
|
def test_connection_error_translates(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.maindata.side_effect = requests.ConnectionError("Connection refused")
|
|
with patch("media_library_viewer_api.integrations.qbittorrent.QbittorrentClient", return_value=mock_client):
|
|
result = qbit_test({"base_url": "http://qb:8080"}, {"username": "u", "password": "p"}, MagicMock())
|
|
assert result.ok is False
|
|
assert "Connection refused" in result.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# prometheus (CT-105)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPrometheusTestConnection:
|
|
def test_success_returns_gateway_evidence(self) -> None:
|
|
payload = SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"results": {}})
|
|
with patch("media_library_viewer_api.integrations.prometheus.requests.post", return_value=payload):
|
|
result = prom_test(
|
|
{"grafana_url": "http://grafana:3000", "datasource_uid": "prometheus"},
|
|
{"grafana_api_key": "tok"},
|
|
MagicMock(),
|
|
)
|
|
assert result.ok is True
|
|
assert "Gateway" in (result.evidence or "")
|
|
|
|
def test_missing_url_returns_error_without_network(self) -> None:
|
|
result = prom_test({}, {"grafana_api_key": "tok"}, MagicMock())
|
|
assert result.ok is False
|
|
assert "URL" in result.detail
|
|
|
|
def test_missing_api_key_returns_error_without_network(self) -> None:
|
|
result = prom_test({"grafana_url": "http://grafana:3000"}, {}, MagicMock())
|
|
assert result.ok is False
|
|
assert "API key" in result.detail
|
|
|
|
def test_http_401_translates_to_auth(self) -> None:
|
|
exc = requests.HTTPError(response=SimpleNamespace(status_code=401))
|
|
payload = SimpleNamespace(raise_for_status=MagicMock(side_effect=exc))
|
|
with patch("media_library_viewer_api.integrations.prometheus.requests.post", return_value=payload):
|
|
result = prom_test(
|
|
{"grafana_url": "http://grafana:3000"},
|
|
{"grafana_api_key": "wrong"},
|
|
MagicMock(),
|
|
)
|
|
assert result.ok is False
|
|
assert "Authentication failed" in result.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# alertmanager (CT-106)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAlertmanagerTestConnection:
|
|
def test_success_returns_version(self) -> None:
|
|
payload = SimpleNamespace(
|
|
raise_for_status=lambda: None,
|
|
json=lambda: {"versionInfo": {"version": "0.27.0"}},
|
|
)
|
|
with patch("media_library_viewer_api.integrations.alertmanager.requests.get", return_value=payload):
|
|
result = am_test({"base_url": "http://am:9093"}, {}, MagicMock())
|
|
assert result.ok is True
|
|
assert result.evidence == "0.27.0"
|
|
|
|
def test_connection_refused_translates(self) -> None:
|
|
with patch(
|
|
"media_library_viewer_api.integrations.alertmanager.requests.get",
|
|
side_effect=requests.ConnectionError("refused"),
|
|
):
|
|
result = am_test({"base_url": "http://am:9093"}, {}, MagicMock())
|
|
assert result.ok is False
|
|
assert "Connection refused" in result.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# jellyfin (CT-107)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestJellyfinTestConnection:
|
|
def test_success_returns_user_count(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.users.return_value = [{"Name": "a"}, {"Name": "b"}]
|
|
with patch("media_library_viewer_api.integrations.jellyfin.JellyfinClient", return_value=mock_client):
|
|
result = jf_test({"base_url": "http://jf:8096"}, {"api_key": "k"}, MagicMock())
|
|
assert result.ok is True
|
|
assert "2 users" == result.evidence
|
|
|
|
def test_http_401_translates_to_auth(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.users.side_effect = requests.HTTPError(response=SimpleNamespace(status_code=401))
|
|
with patch("media_library_viewer_api.integrations.jellyfin.JellyfinClient", return_value=mock_client):
|
|
result = jf_test({"base_url": "http://jf:8096"}, {"api_key": "wrong"}, MagicMock())
|
|
assert result.ok is False
|
|
assert "Authentication failed" in result.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# authentik (CT-108)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAuthentikTestConnection:
|
|
def test_success_returns_user_count(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.users.return_value = {"total": 5, "items": []}
|
|
with patch("media_library_viewer_api.integrations.authentik.AuthentikClient", return_value=mock_client):
|
|
result = ak_test({"base_url": "http://ak:9000"}, {"api_token": "tok"}, MagicMock())
|
|
assert result.ok is True
|
|
assert "5 users" == result.evidence
|
|
|
|
def test_connection_error_translates(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.users.side_effect = requests.ConnectionError("refused")
|
|
with patch("media_library_viewer_api.integrations.authentik.AuthentikClient", return_value=mock_client):
|
|
result = ak_test({"base_url": "http://ak:9000"}, {"api_token": "tok"}, MagicMock())
|
|
assert result.ok is False
|
|
assert "Connection refused" in result.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ssh_tasks (CT-109)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSshTasksTestConnection:
|
|
def test_success_returns_connected_evidence(self) -> None:
|
|
mock_client = MagicMock()
|
|
with patch("media_library_viewer_api.services.task_runner.build_ssh_client", return_value=mock_client):
|
|
result = ssh_test({"host": "srv", "port": 22, "username": "u"}, {"passphrase": ""}, MagicMock())
|
|
assert result.ok is True
|
|
assert "Connected to srv:22" == result.evidence
|
|
|
|
def test_auth_failed_translates_to_ssh_auth_message(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.connect.side_effect = Exception("SSH authentication failed")
|
|
with patch("media_library_viewer_api.services.task_runner.build_ssh_client", return_value=mock_client):
|
|
result = ssh_test({"host": "srv", "port": 22, "username": "u"}, {}, MagicMock())
|
|
assert result.ok is False
|
|
assert "SSH authentication failed" in result.detail
|
|
|
|
def test_protocol_banner_translates(self) -> None:
|
|
mock_client = MagicMock()
|
|
mock_client.connect.side_effect = Exception("protocol banner error")
|
|
with patch("media_library_viewer_api.services.task_runner.build_ssh_client", return_value=mock_client):
|
|
result = ssh_test({"host": "srv", "port": 22, "username": "u"}, {}, MagicMock())
|
|
assert result.ok is False
|
|
assert "SSH banner" in result.detail
|
|
|
|
def test_missing_host_returns_value_error(self) -> None:
|
|
result = ssh_test({"host": "", "username": "u"}, {}, MagicMock())
|
|
assert result.ok is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# nextcloud (CT-110)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNextcloudTestConnection:
|
|
def test_success_returns_version(self) -> None:
|
|
payload = SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"version": "29.0.0"})
|
|
with patch("media_library_viewer_api.integrations.nextcloud.requests.get", return_value=payload):
|
|
result = nc_test({"base_url": "http://nc:80"}, {}, MagicMock())
|
|
assert result.ok is True
|
|
assert result.evidence == "29.0.0"
|
|
|
|
def test_connection_error_translates(self) -> None:
|
|
with patch(
|
|
"media_library_viewer_api.integrations.nextcloud.requests.get",
|
|
side_effect=requests.ConnectionError("refused"),
|
|
):
|
|
result = nc_test({"base_url": "http://nc:80"}, {}, MagicMock())
|
|
assert result.ok is False
|
|
assert "Connection refused" in result.detail
|