diff --git a/backend/src/media_library_viewer_api/integrations/alertmanager.py b/backend/src/media_library_viewer_api/integrations/alertmanager.py index 64d8016..ccbef80 100644 --- a/backend/src/media_library_viewer_api/integrations/alertmanager.py +++ b/backend/src/media_library_viewer_api/integrations/alertmanager.py @@ -2,17 +2,24 @@ from __future__ import annotations -from typing import Any +from typing import TYPE_CHECKING, Any + +import requests from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, + TestResult, WidgetConfigBase, + translate_connection_error, widget_kind, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + class AlertmanagerConfig(ServiceConfigBase): """Non-secret Alertmanager connection config.""" @@ -68,6 +75,28 @@ def summarize_alerts( } +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """GET /api/v2/status with optional bearer auth.""" + try: + base_url = str(config.get("base_url") or "").rstrip("/") + timeout = int(config.get("timeout_seconds") or 5) + headers: dict[str, str] = {} + api_key = str(secrets.get("api_key") or "") + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + resp = requests.get(f"{base_url}/api/v2/status", headers=headers, timeout=timeout) + resp.raise_for_status() + payload = resp.json() + version = str(payload.get("versionInfo", {}).get("version", "") or "connected") + return TestResult(ok=True, detail="Connected to Alertmanager.", evidence=version) + except Exception as exc: + return translate_connection_error(exc, context="Alertmanager") + + DEFINITION = ServiceDefinition( service_type="alertmanager", name="Alertmanager", @@ -86,4 +115,5 @@ DEFINITION = ServiceDefinition( refresh_interval_ms=30_000, ), ], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/integrations/authentik.py b/backend/src/media_library_viewer_api/integrations/authentik.py index bfb53a6..6fa97bd 100644 --- a/backend/src/media_library_viewer_api/integrations/authentik.py +++ b/backend/src/media_library_viewer_api/integrations/authentik.py @@ -8,13 +8,39 @@ is unchanged -- this service type is for the directory, not SSO. from __future__ import annotations +from typing import TYPE_CHECKING, Any + +from media_library_viewer_api.clients.authentik import AuthentikClient from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, + TestResult, + translate_connection_error, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + + +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """Probe AuthentikClient.users(page=1, page_size=1) — lightest directory call.""" + try: + base_url = str(config.get("base_url") or "").rstrip("/") + api_token = str(secrets.get("api_token") or "") + timeout = float(config.get("timeout_seconds") or 10) + client = AuthentikClient(base_url=base_url, api_token=api_token, timeout=timeout) + result = client.users(page=1, page_size=1) + total = result.get("total", 0) if isinstance(result, dict) else 0 + return TestResult(ok=True, detail="Connected to Authentik.", evidence=f"{total} users") + except Exception as exc: + return translate_connection_error(exc, context="Authentik") + class AuthentikConfig(ServiceConfigBase): """Non-secret Authentik connection config.""" @@ -32,4 +58,5 @@ DEFINITION = ServiceDefinition( SecretField(key="api_token", label="API token", required=True), ], widget_kinds=[], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/integrations/base.py b/backend/src/media_library_viewer_api/integrations/base.py index 4204684..a2dbe4e 100644 --- a/backend/src/media_library_viewer_api/integrations/base.py +++ b/backend/src/media_library_viewer_api/integrations/base.py @@ -15,11 +15,16 @@ map. There is no runtime plugin loading. from __future__ import annotations +import asyncio from dataclasses import dataclass, field -from typing import Annotated, Any +from typing import TYPE_CHECKING, Annotated, Any, Callable +import requests from pydantic import BaseModel, BeforeValidator, Field +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + def _validate_service_base_url(value: Any) -> str: """Require an absolute http(s) URL for service ``base_url`` fields. @@ -93,6 +98,20 @@ class WidgetKind: config_model: type[WidgetConfigBase] | None = None +@dataclass(frozen=True) +class TestResult: + """Outcome of a credential/connectivity test for a service instance.""" + + ok: bool + detail: str + evidence: str | None = None + + +#: A test routine receives (config, secrets, store). The store is needed for +#: ssh_tasks (SSH-key resolution). Other types ignore it. +TestCallable = Callable[[dict[str, Any], dict[str, str], "SettingsStore"], TestResult] + + @dataclass(frozen=True) class ServiceDefinition: """Closed description of an external service type.""" @@ -103,6 +122,7 @@ class ServiceDefinition: config_model: type[ServiceConfigBase] secret_fields: list[SecretField] widget_kinds: list[WidgetKind] + test_callable: TestCallable | None = None @property def config_schema(self) -> dict[str, Any]: @@ -148,3 +168,58 @@ def validate_config(model_cls: type[BaseModel], config: dict[str, Any] | None) - """Validate a config dict against a Pydantic model and return the cleaned dict.""" instance = model_cls.model_validate(config or {}) return instance.model_dump(exclude_none=True) + + +def translate_connection_error(exc: Exception, *, context: str = "") -> TestResult: + """Map a common connection/auth exception to a human-friendly TestResult. + + Handles patterns extracted from ``test_machine_ssh`` (settings.py) plus + HTTP-client patterns from the widget sources. Each per-type test routine + calls this for unexpected exceptions, but handles its **type-specific** + auth failures directly (e.g., qBit ``"Fails."``). + """ + message = str(exc) + lowered = message.lower() + + # Auth failures (HTTP 401/403) + if isinstance(exc, requests.HTTPError): + status_code = exc.response.status_code if exc.response is not None else 0 + if status_code in (401, 403): + return TestResult( + ok=False, + detail=f"Authentication failed — the service rejected the credentials ({status_code}).", + ) + if "authentication failed" in lowered or "no authentication methods available" in lowered: + return TestResult(ok=False, detail="Authentication failed — check the credentials, API key, or SSH key.") + + # Timeout (before OSError check, since requests.Timeout is a subclass of OSError) + if isinstance(exc, (requests.Timeout, TimeoutError, asyncio.TimeoutError)): + return TestResult(ok=False, detail="Connection timed out — the service did not respond in time.") + + # Connection refused / DNS / unreachable + if isinstance(exc, (requests.ConnectionError, ConnectionRefusedError, OSError)): + if ( + "name or service not known" in lowered + or "nodename nor servname" in lowered + or "getaddrinfo failed" in lowered + ): + return TestResult(ok=False, detail="Host not found — check the URL/hostname for typos.") + return TestResult( + ok=False, + detail="Connection refused — the service is not reachable at the configured address.", + ) + + # SSL / certificate errors + if "ssl" in lowered or "certificate" in lowered: + return TestResult(ok=False, detail="SSL/TLS error — the service's certificate is invalid or untrusted.") + + # SSH banner (from test_machine_ssh pattern) + if "protocol banner" in lowered: + return TestResult( + ok=False, + detail="SSH banner not received — confirm the SSH service is running and the port is correct.", + ) + + # Fallback + prefix = f"{context}: " if context else "" + return TestResult(ok=False, detail=f"{prefix}{message[:200]}") diff --git a/backend/src/media_library_viewer_api/integrations/jellyfin.py b/backend/src/media_library_viewer_api/integrations/jellyfin.py index 833c63c..7b00058 100644 --- a/backend/src/media_library_viewer_api/integrations/jellyfin.py +++ b/backend/src/media_library_viewer_api/integrations/jellyfin.py @@ -2,15 +2,40 @@ from __future__ import annotations +from typing import TYPE_CHECKING, Any + +from media_library_viewer_api.clients.jellyfin import JellyfinClient from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, + TestResult, WidgetConfigBase, + translate_connection_error, widget_kind, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + + +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """Call JellyfinClient.users() — the lightest authenticated probe.""" + try: + base_url = str(config.get("base_url") or "") + api_key = str(secrets.get("api_key") or "") + timeout = int(config.get("timeout_seconds") or 10) + client = JellyfinClient(base_url, api_key, timeout=timeout) + users = client.users() + return TestResult(ok=True, detail="Connected to Jellyfin.", evidence=f"{len(users)} users") + except Exception as exc: + return translate_connection_error(exc, context="Jellyfin") + class JellyfinConfig(ServiceConfigBase): """Non-secret Jellyfin connection config. @@ -68,4 +93,5 @@ DEFINITION = ServiceDefinition( refresh_interval_ms=30_000, ), ], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/integrations/nextcloud.py b/backend/src/media_library_viewer_api/integrations/nextcloud.py index fe62d97..87731bd 100644 --- a/backend/src/media_library_viewer_api/integrations/nextcloud.py +++ b/backend/src/media_library_viewer_api/integrations/nextcloud.py @@ -6,13 +6,39 @@ dashboard widgets yet; its service page holds connection config only. from __future__ import annotations +from typing import TYPE_CHECKING, Any + +import requests + from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, + TestResult, + translate_connection_error, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + + +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """GET {base_url}/status.php (unauthenticated server probe).""" + try: + base_url = str(config.get("base_url") or "").rstrip("/") + resp = requests.get(f"{base_url}/status.php", timeout=10) + resp.raise_for_status() + payload = resp.json() + version = str(payload.get("version", "") or "connected") + return TestResult(ok=True, detail="Connected to Nextcloud.", evidence=version) + except Exception as exc: + return translate_connection_error(exc, context="Nextcloud") + class NextcloudConfig(ServiceConfigBase): """Non-secret Nextcloud connection config.""" @@ -30,4 +56,5 @@ DEFINITION = ServiceDefinition( SecretField(key="app_password", label="App password", required=True), ], widget_kinds=[], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/integrations/prometheus.py b/backend/src/media_library_viewer_api/integrations/prometheus.py index eb088c8..dffdf53 100644 --- a/backend/src/media_library_viewer_api/integrations/prometheus.py +++ b/backend/src/media_library_viewer_api/integrations/prometheus.py @@ -2,15 +2,71 @@ from __future__ import annotations +from typing import TYPE_CHECKING, Any + +import requests + from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, + TestResult, WidgetConfigBase, + translate_connection_error, widget_kind, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + + +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """POST {grafana_url}/api/ds/query with expr 'up' via the Grafana gateway.""" + try: + grafana_url = str(config.get("grafana_url") or "").rstrip("/") + api_key = str(secrets.get("grafana_api_key") or "") + datasource_uid = str(config.get("datasource_uid") or "prometheus") + timeout = int(config.get("timeout_seconds") or 10) + if not grafana_url: + return TestResult(ok=False, detail="Grafana gateway URL is required.") + if not api_key: + return TestResult(ok=False, detail="Grafana API key is required.") + body = { + "queries": [ + { + "datasource": {"uid": datasource_uid, "type": "prometheus"}, + "expr": "up", + "format": "time_series", + "intervalMs": 15000, + "maxDataPoints": 1, + "refId": "A", + } + ], + "from": "now-1m", + "to": "now", + } + resp = requests.post( + f"{grafana_url}/api/ds/query", + json=body, + timeout=timeout, + headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, + ) + resp.raise_for_status() + return TestResult( + ok=True, + detail="Grafana gateway reachable.", + evidence="Gateway reachable; datasource responded.", + ) + except requests.HTTPError as exc: + return translate_connection_error(exc, context="Prometheus via Grafana") + except Exception as exc: + return translate_connection_error(exc, context="Prometheus via Grafana") + class PrometheusConfig(ServiceConfigBase): """Non-secret Prometheus-via-Grafana gateway config.""" @@ -99,4 +155,5 @@ DEFINITION = ServiceDefinition( refresh_interval_ms=60_000, ), ], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/integrations/qbittorrent.py b/backend/src/media_library_viewer_api/integrations/qbittorrent.py index e505534..82cb64d 100644 --- a/backend/src/media_library_viewer_api/integrations/qbittorrent.py +++ b/backend/src/media_library_viewer_api/integrations/qbittorrent.py @@ -7,15 +7,47 @@ password), and three widget kinds (totals, active, speed). Models on from __future__ import annotations +from typing import TYPE_CHECKING, Any + +from media_library_viewer_api.clients.qbittorrent import QbittorrentClient from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, + TestResult, WidgetConfigBase, + translate_connection_error, widget_kind, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + + +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """Login + probe maindata; surface auth failures specifically.""" + try: + base_url = str(config.get("base_url") or "") + username = str(secrets.get("username") or "") + password = str(secrets.get("password") or "") + timeout = int(config.get("timeout_seconds") or 10) + client = QbittorrentClient(base_url, username, password, timeout=timeout) + data = client.maindata() + version = str(data.get("server_state", {}).get("qbittorrent_version", "") or "connected") + return TestResult(ok=True, detail="Connected to qBittorrent.", evidence=version) + except RuntimeError as exc: + lowered = str(exc).lower() + if "login failed" in lowered: + return TestResult(ok=False, detail="Authentication failed — qBittorrent rejected the credentials.") + return translate_connection_error(exc, context="qBittorrent") + except Exception as exc: + return translate_connection_error(exc, context="qBittorrent") + class QbittorrentConfig(ServiceConfigBase): """Non-secret qBittorrent connection config.""" @@ -65,4 +97,5 @@ DEFINITION = ServiceDefinition( refresh_interval_ms=5_000, ), ], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/integrations/ssh_tasks.py b/backend/src/media_library_viewer_api/integrations/ssh_tasks.py index a2eeb89..57c3b03 100644 --- a/backend/src/media_library_viewer_api/integrations/ssh_tasks.py +++ b/backend/src/media_library_viewer_api/integrations/ssh_tasks.py @@ -8,14 +8,74 @@ and shown as history on the instance's service page. from __future__ import annotations +from typing import TYPE_CHECKING, Any + from media_library_viewer_api.integrations.base import ( SecretField, ServiceConfigBase, ServiceDefinition, + TestResult, WidgetConfigBase, + translate_connection_error, widget_kind, ) +if TYPE_CHECKING: + from media_library_viewer_api.services.settings_store import SettingsStore + + +def test_connection( + config: dict[str, Any], + secrets: dict[str, str], + store: SettingsStore, +) -> TestResult: + """Build an SSH client via build_ssh_client and attempt .connect(). + + Reuses the same error-translation patterns as test_machine_ssh (banner, + auth failed). Known-host recording is preserved. + """ + from media_library_viewer_api.services.task_runner import build_ssh_client + from media_library_viewer_api.widgets.sources import ServiceRecord + + host = str(config.get("host") or "").strip() + port = int(config.get("port") or 22) + try: + service = ServiceRecord( + id="", + service_type="ssh_tasks", + name="test", + config=config, + secrets=secrets, + enabled=True, + ) + client = build_ssh_client(store, service) + try: + client.connect() + except Exception as exc: + lowered = str(exc).lower() + if "protocol banner" in lowered: + return TestResult( + ok=False, + detail=f"SSH banner not received from {host}:{port}; confirm the SSH service is running.", + ) + if "no authentication methods available" in lowered or "authentication failed" in lowered: + return TestResult( + ok=False, + detail=f"SSH authentication failed for {host}:{port}; check the SSH key, passphrase, or username.", + ) + return translate_connection_error(exc, context=f"SSH {host}:{port}") + finally: + client.close() + return TestResult( + ok=True, + detail=f"SSH connection succeeded for {host}:{port}.", + evidence=f"Connected to {host}:{port}", + ) + except ValueError as exc: + return TestResult(ok=False, detail=str(exc)) + except Exception as exc: + return translate_connection_error(exc, context=f"SSH {host}:{port}") + class SshTasksConfig(ServiceConfigBase): """Non-secret SSH task runner config. @@ -57,4 +117,5 @@ DEFINITION = ServiceDefinition( refresh_interval_ms=0, ), ], + test_callable=test_connection, ) diff --git a/backend/src/media_library_viewer_api/routers/services.py b/backend/src/media_library_viewer_api/routers/services.py index df00631..2ee1f2f 100644 --- a/backend/src/media_library_viewer_api/routers/services.py +++ b/backend/src/media_library_viewer_api/routers/services.py @@ -12,7 +12,7 @@ from typing import Any from fastapi import APIRouter, Depends, HTTPException, status from media_library_viewer_api.dependencies import get_settings_store -from media_library_viewer_api.integrations.base import validate_config +from media_library_viewer_api.integrations.base import TestResult, validate_config from media_library_viewer_api.integrations.registry import ( SERVICE_DEFINITIONS, get_service_definition, @@ -180,3 +180,30 @@ def delete_instance( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Service not found") store.delete_service(service_id) return {"status": "deleted"} + + +@router.post("/test") +def test_instance( + body: ServiceInstanceInput, + store: SettingsStore = Depends(get_settings_store), +) -> dict[str, Any]: + """Test connectivity + credentials for unsaved service input. + + Validates first (422 on malformed config), dispatches to the per-type + test_callable, and returns ``{ok, detail, evidence}``. Does NOT persist. + """ + _validate_input(body) # raises HTTPException(422) on bad config/type/secrets + definition = require_service_definition(body.service_type) + + if definition.test_callable is None: + logger.info("test requested type=%s ok=true (no test_callable)", body.service_type) + return {"ok": True, "detail": "No connection test for this service type", "evidence": None} + + try: + result: TestResult = definition.test_callable(body.config, body.secrets, store) + except Exception as exc: + logger.exception("test_callable raised for type=%s", body.service_type) + result = TestResult(ok=False, detail=f"Test failed unexpectedly: {exc}") + + logger.info("test requested type=%s ok=%s", body.service_type, result.ok) + return {"ok": result.ok, "detail": result.detail, "evidence": result.evidence} diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index ddc85e2..4044d0d 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -845,3 +845,74 @@ class TestPrometheusStartupValidation: # Best-effort validator logs a migration hint referencing grafana_url. assert "grafana_url" in caplog.text assert any(record.levelno == logging.WARNING for record in caplog.records) + + +# --- Service credential tester endpoint (CT-101..CT-113) --- + + +class TestServiceTestEndpoint: + """Tests for POST /api/services/test — dispatch, validation-first, no-persistence, no-secret-logs.""" + + def test_backups_returns_no_test_needed(self, test_client: TestClient) -> None: + """backups has test_callable=None → returns ok=true with 'No test' detail.""" + response = test_client.post( + "/api/services/test", + json={ + "service_type": "backups", + "name": "test", + "config": {"ingestion_label": "default"}, + "secrets": {}, + "enabled": True, + }, + ) + assert response.status_code == 200 + body = response.json() + assert body["ok"] is True + assert "No" in body["detail"] + + def test_validation_first_rejects_malformed_config(self, test_client: TestClient) -> None: + """Malformed config (schema-less base_url) → 422, no test_callable called.""" + response = test_client.post( + "/api/services/test", + json={ + "service_type": "qbittorrent", + "name": "test", + "config": {"base_url": "localhost:8080"}, # missing http:// + "secrets": {"username": "u", "password": "p"}, + "enabled": True, + }, + ) + assert response.status_code == 422 + + def test_no_persistence_after_test(self, test_client: TestClient, tmp_path) -> None: + """Calling /test does not create a service row.""" + store = app.dependency_overrides[get_settings_store]() + before = len(store.list_services()) + test_client.post( + "/api/services/test", + json={ + "service_type": "backups", + "name": "test", + "config": {"ingestion_label": "default"}, + "secrets": {}, + "enabled": True, + }, + ) + after = len(store.list_services()) + assert before == after + + def test_secrets_not_logged(self, test_client: TestClient, caplog) -> None: + """No log line contains the secret value.""" + secret_value = "super-secret-hunter2" + with caplog.at_level(logging.INFO): + test_client.post( + "/api/services/test", + json={ + "service_type": "backups", + "name": "test", + "config": {"ingestion_label": "default"}, + "secrets": {}, + "enabled": True, + }, + ) + assert secret_value not in caplog.text diff --git a/backend/tests/test_credential_tester.py b/backend/tests/test_credential_tester.py new file mode 100644 index 0000000..c4272e3 --- /dev/null +++ b/backend/tests/test_credential_tester.py @@ -0,0 +1,260 @@ +"""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