feat(service-credential-tester): slice 1 — backend test endpoint + per-type routines

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.

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,
 )

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,
 )

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]}")

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,
 )

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,
 )

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,
 )

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,
 )

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,
 )

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}

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

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