manage/openspec/changes/service-credential-tester/design.md

SDD Design: Service Credential Tester

Change: service-credential-tester Phase: design Date: 2026-07-09

0. Source findings (read before anything else)

The proposal and spec were written against a mental model. Reading actual source surfaced deviations the design must account for. Trust source, not assumptions.

Spec/proposal claim	Actual source reality	Design impact
Spec CT-115: "the edit dialog on ServicePage.tsx"	ServicePage.tsx has no edit dialog. It's a read-only tabbed view (Overview + content tabs + Widgets). Service editing lives in Settings.tsx as ServiceConfigEditor (a master-detail panel, lines ~1435–1596). The "Save" button calls saveService.mutateAsync(buildInput()) directly (no try/catch — a latent error-swallow like the pre-#1-fix create dialog).	The Test button is added to two edit surfaces: CreateServiceDialog (ServicesPage.tsx) and ServiceConfigEditor (Settings.tsx). NOT ServicePage.tsx. Spec CT-115 should note this textual drift.
Proposal §5.1.3: "the edit dialog on ServicePage.tsx"	Same as above.	Same fix — target is Settings.tsx::ServiceConfigEditor.
Spec CT-109: "ssh_tasks test reuses the test_machine_ssh connect flow"	test_machine_ssh (settings.py:122–180) is a router endpoint, not a reusable function. Its logic (build client → connect → translate errors) is inlined. However, build_ssh_client in task_runner.py already builds an RemoteSSHClient from a ServiceRecord — the exact pattern needed.	The ssh_tasks test routine constructs an RemoteSSHClient via build_ssh_client(store, service_record) and calls .connect(), then translates errors using the same message patterns as test_machine_ssh (protocol banner / auth failed). It needs a SettingsStore (for SSH key resolution) — see §3.6 for how test_callables get store access.
Proposal: "construct a QbittorrentClient from the config + decrypted secrets"	QbittorrentClient.__init__(base_url, username, password, timeout) takes positional strings, not a config dict. The qbittorrent config has base_url + timeout_seconds; secrets have username + password.	The test routine extracts fields from config/secrets dicts explicitly: QbittorrentClient(config["base_url"], secrets["username"], secrets["password"], config.get("timeout_seconds", 10)).
Proposal: "Alertmanager test probes /api/v2/alerts"	AlertmanagerWidgetSource.fetch (sources.py) already hits /api/v2/alerts with optional bearer auth. The exact URL pattern + auth-header logic is reusable.	The test routine mirrors the widget source's request shape (GET {base_url}/api/v2/alerts, optional Authorization: Bearer {api_key}).
Proposal: "Authentik test probes its directory endpoint"	routers/authentik_users.py::_build_client builds an AuthentikClient(base_url, api_token, timeout). The client has a .users() method that hits the directory endpoint.	The test routine constructs an AuthentikClient and calls .users(page=1, page_size=1) — the lightest possible probe.
Proposal: "Nextcloud test probes /status.php"	NextcloudConfig has base_url + username (not a ServiceBaseUrl). Secret is app_password. /status.php is unauthenticated.	The test routine does a simple GET {base_url}/status.php (no auth headers); extracts version from the JSON response.
Proposal: "test_callable field on ServiceDefinition"	ServiceDefinition is a @dataclass(frozen=True) with fields: service_type, name, description, config_model, secret_fields, widget_kinds. Adding test_callable after widget_kinds works (dataclass field ordering — it has a default).	New field `test_callable: TestCallable

No proposal/spec scope change is required — the intent (per-type credential tester) still holds. The findings above refine implementation details and correct the edit-surface target.

---

1. Architecture overview

A new POST /api/services/test endpoint validates the input (reusing _validate_input), resolves the service definition's test_callable, and runs it. Each integration declares its own test_connection routine alongside its DEFINITION. A shared error-translation helper maps common exception types to human-friendly strings. The frontend gains a Test button + result pill + gating in both the create dialog and the settings edit panel.

 POST /api/services/test
   └► _validate_input(body)              → 422 on malformed config (CT-102)
   └► definition = get_service_definition(body.service_type)
   └► if definition.test_callable is None → {ok: true, detail: "No test..."} (CT-103/111)
   └► result = test_callable(config, secrets, store)
        └► per-type routine:
             qbittorrent → QbittorrentClient.maindata()
             prometheus  → POST {grafana_url}/api/ds/query (expr "up")
             alertmanager → GET {base_url}/api/v2/alerts
             jellyfin    → JellyfinClient.users()
             authentik   → AuthentikClient.users(page=1, page_size=1)
             ssh_tasks   → RemoteSSHClient.connect() via build_ssh_client
             nextcloud   → GET {base_url}/status.php
        └► translate_connection_error(exc) → TestResult(ok=False, detail=...)
   └► return {ok, detail, evidence}      → HTTP 200 (CT-101)

No persistence. No secrets in logs. Auth-gated identically to every other /api/services/* endpoint.

---

2. Core data structures

2.1 TestResult dataclass (CT-101)

File: backend/src/media_library_viewer_api/integrations/base.py

@dataclass(frozen=True)
class TestResult:
    """Outcome of a credential/connectivity test for a service instance."""
    ok: bool
    detail: str
    evidence: str | None = None

2.2 TestCallable type alias + test_callable field on ServiceDefinition

File: backend/src/media_library_viewer_api/integrations/base.py

from typing import Callable

# A test routine receives (config, secrets, store). The store is needed for
# ssh_tasks (SSH-key resolution via store.get_ssh_key). Other types ignore it.
TestCallable = Callable[["dict[str, Any]", "dict[str, str]", "SettingsStore"], TestResult]

ServiceDefinition gains (placed last, after widget_kinds, with a default):

@dataclass(frozen=True)
class ServiceDefinition:
    service_type: str
    name: str
    description: str
    config_model: type[ServiceConfigBase]
    secret_fields: list[SecretField]
    widget_kinds: list[WidgetKind]
    test_callable: TestCallable | None = None  # NEW — placed last

Why SettingsStore in the signature: ssh_tasks needs store.get_ssh_key(ssh_key_id) to resolve the private key material. Every other type ignores it. Making it a uniform parameter avoids a special-case dispatch for one type.

2.3 Shared error-translation helper

File: backend/src/media_library_viewer_api/integrations/base.py (or a new integrations/test_helpers.py if base.py gets too large — design recommends base.py for discoverability)

def translate_connection_error(exc: Exception, *, context: str = "") -> TestResult:
    """Map a common connection/auth exception to a human-friendly TestResult.

    Handles the patterns extracted from ``test_machine_ssh`` (settings.py) plus
    the HTTP-client patterns from the widget sources.
    """
    message = str(exc)
    lowered = message.lower()

    # Auth failures (HTTP 401/403 or auth-specific strings)
    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.")

    # 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.")

    # Timeout
    if isinstance(exc, (requests.Timeout, TimeoutError, asyncio.TimeoutError)):
        return TestResult(ok=False, detail="Connection timed out — the service did not respond in time.")

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

Each per-type routine wraps its probe in try/except and calls translate_connection_error for unexpected exceptions, but handles its type-specific auth failures directly (e.g., qBit "Fails." response).

---

3. Per-type test routines

Each routine lives alongside its integration's DEFINITION. The signature is (config: dict, secrets: dict, store: SettingsStore) -> TestResult. All use requests (synchronous) wrapped in a short timeout — no asyncio needed (the endpoint handler calls the callable synchronously).

3.1 qbittorrent (CT-104)

File: backend/src/media_library_viewer_api/integrations/qbittorrent.py

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:
        # QbittorrentClient._login raises RuntimeError("qBittorrent login failed: Fails.")
        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")

Evidence: qBittorrent version from server_state (fallback "connected").

3.2 prometheus (CT-105)

File: backend/src/media_library_viewer_api/integrations/prometheus.py

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

Evidence: "Gateway reachable; datasource responded." — validates the full path (Grafana up + datasource reachable + Prom responding).

3.3 alertmanager (CT-106)

File: backend/src/media_library_viewer_api/integrations/alertmanager.py

def test_connection(config: dict[str, Any], secrets: dict[str, str], store: SettingsStore) -> TestResult:
    """GET /api/v2/alerts (or /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 = {}
        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")

Evidence: Alertmanager cluster version from /api/v2/status.

3.4 jellyfin (CT-107)

File: backend/src/media_library_viewer_api/integrations/jellyfin.py

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

Evidence: "<N> users".

3.5 authentik (CT-108)

File: backend/src/media_library_viewer_api/integrations/authentik.py

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)
        return TestResult(ok=True, detail="Connected to Authentik.", evidence=f"{result.get('total', 0)} users")
    except Exception as exc:
        return translate_connection_error(exc, context="Authentik")

Evidence: "<N> users" from the directory total.

3.6 ssh_tasks (CT-109)

File: backend/src/media_library_viewer_api/integrations/ssh_tasks.py

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 (first successful connect
    records the host key, same as test_machine_ssh).
    """
    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}")

Evidence: "Connected to <host>:<port>". Known-host recording: preserved — RemoteSSHClient.connect() records the host key on first successful connect (same as test_machine_ssh).

3.7 nextcloud (CT-110)

File: backend/src/media_library_viewer_api/integrations/nextcloud.py

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

Evidence: Nextcloud version from /status.php.

3.8 backups (CT-111)

File: backend/src/media_library_viewer_api/integrations/backups.py

# No test_connection function. The DEFINITION's test_callable stays None (default).
# The endpoint returns {ok: true, detail: "No connection test for this service type"}.

---

4. Backend endpoint design

4.1 POST /api/services/test (CT-101, CT-102, CT-103, CT-112, CT-113)

File: backend/src/media_library_viewer_api/routers/services.py

from media_library_viewer_api.integrations.base import TestResult

@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:
        # A routine should never raise (it catches internally), but defend.
        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}

Key properties:

• Validation-first (CT-102): _validate_input runs before any network call. Malformed config → 422 with the same detail format as create.
• No persistence (CT-112): no store.upsert_service, no store.update_setting. The body is consumed and discarded.
• No secret logging (CT-113): only test requested type=%s ok=%s at INFO. No body/config/secrets in any log line.
• Dispatch (CT-103): keys off definition.test_callable (None → default ok response).

4.2 DEFINITION updates — wiring test_callable

Each integration's DEFINITION gains test_callable=test_connection:

# integrations/qbittorrent.py
DEFINITION = ServiceDefinition(
    ...,
    test_callable=test_connection,
)

# integrations/backups.py
DEFINITION = ServiceDefinition(
    ...,  # test_callable stays None (default)
)

Every integration module that has a test_connection function imports TestResult + translate_connection_error from base.py and requests as needed.

---

5. Frontend design

5.1 TestResult type (CT-114)

File: frontend/src/types/index.ts

export interface ServiceTestResult {
  ok: boolean;
  detail: string;
  evidence: string | null;
}

5.2 API client function (CT-114)

File: frontend/src/api/services.ts

import type { ServiceTestResult } from "../types";

export async function testServiceInstance(
  input: ServiceInstanceInput,
): Promise<ServiceTestResult> {
  return post<ServiceTestResult>("/api/services/test", input);
}

5.3 useTestServiceInstance hook (CT-114)

File: frontend/src/hooks/useServices.ts

import { testServiceInstance } from "../api/services";

export function useTestServiceInstance() {
  return useMutation({
    mutationFn: (input: ServiceInstanceInput) => testServiceInstance(input),
  });
}

No cache invalidation needed — the test is a one-shot mutation with no query to refresh.

5.4 Shared ServiceTestPanel component

File: frontend/src/components/ServiceTestPanel.tsx (NEW — shared by both surfaces)

Both CreateServiceDialog and ServiceConfigEditor need the same UI: a Test button, a result pill, testPassed state, and Save-anyway toggle. Extracting it avoids duplication.

interface Props {
  input: ServiceInstanceInput | null;  // null = no draft yet
  onTestPassed: (passed: boolean) => void;
}

export function ServiceTestPanel({ input, onTestPassed }: Props) {
  const testService = useTestServiceInstance();
  const [result, setResult] = useState<ServiceTestResult | null>(null);
  const [saveAnyway, setSaveAnyway] = useState(false);

  // Clear result when input changes (CT-118)
  useEffect(() => {
    setResult(null);
    onTestPassed(false);
  }, [input]);  // input is a new object on every field edit → re-test required

  async function handleTest() {
    if (!input) return;
    setResult(null);
    try {
      const res = await testService.mutateAsync(input);
      setResult(res);
      onTestPassed(res.ok);
    } catch (err) {
      setResult({ ok: false, detail: err instanceof Error ? err.message : String(err), evidence: null });
      onTestPassed(false);
    }
  }

  const testPassed = result?.ok === true;
  const canSave = testPassed || saveAnyway;

  return (
    <div className="flex flex-col gap-2">
      <Button variant="outline" size="sm" onClick={handleTest} disabled={testService.isPending || !input}>
        {testService.isPending ? "Testing…" : "Test credentials"}
      </Button>
      {result ? (
        <Alert variant={result.ok ? "default" : "destructive"}>
          <AlertDescription>
            {result.ok ? `✓ Connected${result.evidence ? ` — ${result.evidence}` : ""}` : `✗ ${result.detail}`}
          </AlertDescription>
        </Alert>
      ) : null}
      <label className="flex items-center gap-2 text-xs text-muted-foreground">
        <input type="checkbox" checked={saveAnyway} onChange={(e) => { setSaveAnyway(e.target.checked); onTestPassed(e.target.checked || testPassed); }} />
        Save anyway (skip test)
      </label>
      {/* Hidden signal: parent reads canSave via onTestPassed callback */}
    </div>
  );
}

CT-118 (field-edit clears result): the useEffect([input]) triggers on every render where input is a new object reference (React state updates create new objects). Since setDraft({...draft, config}) creates a new object, any field edit resets the result.

CT-117 (Save gating): the parent passes onTestPassed to track the gating state. The parent's confirm button is disabled unless testPassed || saveAnyway.

5.5 Integration into CreateServiceDialog (CT-115, CT-117)

File: frontend/src/pages/ServicesPage.tsx

function CreateServiceDialog({ open, onClose }) {
  // ... existing state ...
  const [testPassed, setTestPassed] = useState(false);

  function reset() {
    setDraft(null);
    setSubmitError(null);
    setTestPassed(false);
  }

  // Build the input object from the draft (for the test panel)
  const testInput: ServiceInstanceInput | null = draft ? {
    service_type: draft.serviceType,
    name: draft.name.trim(),
    config: draft.config,
    secrets: draft.secrets,
    enabled: draft.enabled,
  } : null;

  return (
    <Dialog ...>
      <DialogContent>
        {/* ... existing fields ... */}
        {draft ? (
          <ServiceTestPanel input={testInput} onTestPassed={setTestPassed} />
        ) : null}
        {submitError ? <Alert variant="destructive">...</Alert> : null}
        {draft ? (
          <DialogFooter
            onCancel={reset}
            onConfirm={save}
            confirmLabel="Create service"
            confirmDisabled={!draft.name.trim() || saveService.isPending || !testPassed}
          />
        ) : null}
      </DialogContent>
    </Dialog>
  );
}

5.6 Integration into ServiceConfigEditor (CT-115, CT-117)

File: frontend/src/pages/Settings.tsx

function ServiceConfigEditor({ instance, typeInfo }) {
  // ... existing state ...
  const [testPassed, setTestPassed] = useState(false);

  const testInput: ServiceInstanceInput = {
    id: instance.id,
    service_type: instance.service_type,
    name,
    config: draftConfig,
    secrets: Object.fromEntries(Object.entries(draftSecrets).filter(([, v]) => v !== "")),
    enabled,
  };

  return (
    <>
      <div className="rounded-lg border p-4">
        {/* ... existing fields ... */}
        <ServiceTestPanel input={testInput} onTestPassed={setTestPassed} />
        <div className="flex justify-between">
          <Button onClick={handleSave} disabled={saveService.isPending || !testPassed}>Save</Button>
          {/* ... delete button ... */}
        </div>
      </div>
    </>
  );
}

---

6. Slice plan

Slice 1: Backend (~300–380 lines)

Goal: endpoint + test_callable + shared helper + 7 per-type routines + tests.

Task	File(s)	Lines
TestResult dataclass + TestCallable type + translate_connection_error helper	integrations/base.py	~60
test_callable field on ServiceDefinition	integrations/base.py	~3
7 per-type test_connection routines + wire into DEFINITIONs	integrations/{qbittorrent,prometheus,alertmanager,jellyfin,authentik,ssh_tasks,nextcloud}.py	~180
POST /api/services/test endpoint	routers/services.py	~30
Backend tests: per-type mocked routines, endpoint dispatch, validation-first, no-persistence	tests/test_services.py + tests/test_api.py	~120

Exit gate: pytest + ruff green.

Slice 2: Frontend (~250–320 lines)

Goal: API client + hook + shared Test panel + wire into both surfaces + tests.

Task	File(s)	Lines
ServiceTestResult type	types/index.ts	~5
testServiceInstance API function	api/services.ts	~5
useTestServiceInstance hook	hooks/useServices.ts	~8
ServiceTestPanel component	components/ServiceTestPanel.tsx	~70
Wire into CreateServiceDialog	pages/ServicesPage.tsx	~20
Wire into ServiceConfigEditor	pages/Settings.tsx	~15
Frontend tests: panel renders both states, button fires mutation, gating, field-clear	components/__tests__/ServiceTestPanel.test.tsx + existing page tests	~130

Exit gate: npm run build + npm run lint + npm run test green.

---

7. Test strategy

7.1 Backend tests (CT-119)

• Per-type test_connection (mocked): each routine gets a success case (mocked client returns data → {ok: true, evidence: ...}) and at least one failure case (mocked client raises → {ok: false, detail: ...}). qBit gets a specific "Fails." auth-failure test.
• Endpoint dispatch: POST /api/services/test with each service type dispatches correctly; backups returns the default ok-no-test.
• Validation-first: schema-less URL → 422, no network call mocked.
• No-persistence: call /test, assert store count unchanged.

7.2 Frontend tests (CT-120)

• ServiceTestPanel: renders Test button; click fires mocked mutation; success → green pill with evidence; failure → red pill with detail; Save-anyway checkbox toggles gating.
• CreateServiceDialog: confirm disabled until test passes; Save anyway re-enables; editing a field clears the result.
• ServiceConfigEditor: same gating behavior.

---

8. Security considerations

• No persistence (CT-112): the endpoint never calls any store.* method except read-only get_ssh_key (for SSH key resolution in ssh_tasks). No upsert_service, no update_setting.
• No secret logging (CT-113): only test requested type=%s ok=%s at INFO. No request body in logs. The sanitize_log_extra helper is available if structured logging is added later.
• Auth gating: identical to every other /api/services/* endpoint (JWT/API-key via the app-level middleware).
• Secrets in transit: plaintext in the request body over TLS — identical to the existing POST /api/services/instances create endpoint. No new attack surface.
• Timeout: each routine uses a short timeout (≤10s) from the service config. No long-running probes.

---

9. Key design decisions summary

1. TestResult is a frozen dataclass in integrations/base.py — immutable, serializable, co-located with the ServiceDefinition it augments.
2. test_callable signature includes SettingsStore — needed for ssh_tasks SSH-key resolution; other types ignore it. Uniform signature avoids per-type dispatch special-cases.
3. translate_connection_error is shared — extracts the test_machine_ssh message-translation pattern into a reusable helper covering HTTP errors, connection errors, SSL, timeout, SSH banner. Type-specific auth failures (qBit "Fails.") are handled in-routine.
4. POST /api/services/test validation-first — reuses _validate_input so malformed configs get the same 422 as create. The #1 validation-surfacing fix covers this endpoint too.
5. Shared ServiceTestPanel component — extracted to avoid duplicating the Test button + pill + gating between CreateServiceDialog and ServiceConfigEditor.
6. Field-edit-clears-result via useEffect([input]) — React state updates create new object references on every edit, naturally clearing the result without explicit field-tracking.
7. Prometheus test queries via Grafana gateway — mirrors MetricSource._gateway_query exactly (POST /api/ds/query with expr: "up"). No direct Prom.
8. ssh_tasks test reuses build_ssh_client — the same task_runner.build_ssh_client(store, ServiceRecord) that powers task execution, wrapped with the test_machine_ssh error-translation patterns.