Developer
6d46de26c4
The credential tester always reported "Authentication failed — qBittorrent rejected the credentials" for the qBittorrent service, even when credentials were correct. test_connection classified any RuntimeError whose message contained "login failed" as an auth failure — and the gateway-timeout error (502/503/504 from the reverse proxy) and the wrong-URL diagnostic both started with "qBittorrent login failed:", so a proxy timeout was reported as a credentials rejection. That sent users down the wrong path (re-entering correct passwords to fix a 504). - QbittorrentClient._login: gateway and URL/routing errors no longer contain "login failed"; only a genuine "Fails." body carries the "invalid username or password" signal. - integrations/qbittorrent.test_connection: key the auth message off "invalid username or password" specifically; all other login errors flow through translate_connection_error so the real reason (proxy timeout, wrong URL, empty body) is surfaced. After this, a failing test reports the actual cause (e.g. "qBittorrent is unreachable: reverse proxy returned HTTP 504 ...") instead of accusing the credentials. New regression test asserts a gateway error is NOT reported as "Authentication failed". 386/386 backend tests pass; ruff clean.
277 lines
13 KiB
Python
277 lines
13 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()
|
|
# Mirrors the real _login auth-failure message for a "Fails." body.
|
|
mock_client.maindata.side_effect = RuntimeError(
|
|
"qBittorrent login failed (HTTP 200): invalid username or password"
|
|
)
|
|
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_gateway_error_does_not_masquerade_as_auth_failure(self) -> None:
|
|
mock_client = MagicMock()
|
|
# A 504 from the reverse proxy must NOT be reported as "Authentication
|
|
# failed" — that misled users into re-entering correct credentials.
|
|
mock_client.maindata.side_effect = RuntimeError(
|
|
"qBittorrent is unreachable: reverse proxy returned HTTP 504 for http://qb:8080/api/v2/auth/login."
|
|
)
|
|
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" not in result.detail
|
|
assert "504" 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
|