feat(providers): add availability testing for search providers (DAP-51)

Implements DAP-35 provider tests: probe each configured provider for
availability and surface the result to operators.

- BaseProvider.check_availability() runs a throwaway probe search, measures
  latency, and never raises — failures are reported as available=false with
  the captured error.
- check_all_providers() probes the registry concurrently.
- New GET /api/providers endpoint + ProviderStatus schema.
- backend/tests/test_providers.py: unit + endpoint coverage for reachable
  and failing providers (5/5 tests pass in container).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

README.md

@@ -29,6 +29,7 @@ worker (Python polling)
 | Method | Path | Description |
 |--------|------|-------------|
 | GET | /health | Health check |
+| GET | /api/providers | Test each configured provider for availability |
 | GET | /api/search?q= | Search across configured providers |
 | POST | /api/downloads | Queue a download |
 | GET | /api/downloads | List all downloads |
@@ -43,6 +44,23 @@ and registering them in the `_registry`.
 
 No hardcoded piracy sources. All download URLs come from operator-configured providers only.
 
+### Availability testing
+
+`GET /api/providers` probes every registered provider with a throwaway query and
+reports whether it is reachable, its response latency, and the number of results
+returned. Each `BaseProvider` gets this via `check_availability()`; providers may
+override `probe_query`. Probes run concurrently and never raise — a failing
+provider is reported as `available: false` with the captured error.
+
+```json
+[
+  {"name": "mock", "available": true, "latency_ms": 0.12, "result_count": 3, "error": null}
+]
+```
+
+Covered by `backend/tests/test_providers.py` (unit + endpoint tests for both the
+reachable and failing paths).
+
 ## Configuration
 
 All secrets via environment variables or `.env` (never committed):

backend/src/main.py

@@ -3,8 +3,8 @@ from fastapi import FastAPI, Depends, HTTPException
 from sqlalchemy.orm import Session
 from .database import engine, get_db
 from .models import Base, Download, DownloadStatus
-from .schemas import SearchResult, DownloadCreate, DownloadRead
-from .providers import get_providers
+from .schemas import SearchResult, DownloadCreate, DownloadRead, ProviderStatus
+from .providers import get_providers, check_all_providers
 
 
 @asynccontextmanager
@@ -21,6 +21,12 @@ def health():
     return {"status": "ok"}
 
 
+@app.get("/api/providers", response_model=list[ProviderStatus])
+async def providers_status():
+    """Test each configured search provider for availability."""
+    return await check_all_providers()
+
+
 @app.get("/api/search", response_model=list[SearchResult])
 async def search(q: str):
     results: list[SearchResult] = []

backend/src/providers.py

@@ -5,17 +5,45 @@ Providers supply search results from operator-configured, legitimate sources.
 No hardcoded piracy sources — each provider is loaded from runtime config.
 """
 
+import asyncio
+import time
 from abc import ABC, abstractmethod
-from .schemas import SearchResult
+from .schemas import SearchResult, ProviderStatus
 
 
 class BaseProvider(ABC):
     name: str
+    # Lightweight query used to probe availability without leaking real searches.
+    probe_query: str = "test"
 
     @abstractmethod
     async def search(self, query: str) -> list[SearchResult]:
         ...
 
+    async def check_availability(self) -> ProviderStatus:
+        """Probe the provider with a throwaway search and report its status.
+
+        A provider is considered available if a probe search returns without
+        raising. Latency is always measured so slow-but-reachable providers are
+        distinguishable from failing ones.
+        """
+        start = time.perf_counter()
+        try:
+            results = await self.search(self.probe_query)
+        except Exception as exc:
+            return ProviderStatus(
+                name=self.name,
+                available=False,
+                latency_ms=round((time.perf_counter() - start) * 1000, 2),
+                error=f"{type(exc).__name__}: {exc}",
+            )
+        return ProviderStatus(
+            name=self.name,
+            available=True,
+            latency_ms=round((time.perf_counter() - start) * 1000, 2),
+            result_count=len(results),
+        )
+
 
 class MockProvider(BaseProvider):
     """Stub provider for architecture validation. Returns fake results."""
@@ -46,3 +74,10 @@ _registry: dict[str, BaseProvider] = {
 
 def get_providers() -> list[BaseProvider]:
     return list(_registry.values())
+
+
+async def check_all_providers() -> list[ProviderStatus]:
+    """Probe every registered provider concurrently, preserving registry order."""
+    providers = get_providers()
+    statuses = await asyncio.gather(*(p.check_availability() for p in providers))
+    return list(statuses)

backend/src/schemas.py

@@ -15,6 +15,16 @@ class SearchResult(BaseModel):
     media_type: str = "movie"
 
 
+class ProviderStatus(BaseModel):
+    """Availability report for a single search provider."""
+
+    name: str
+    available: bool
+    latency_ms: float | None = None
+    result_count: int | None = None
+    error: str | None = None
+
+
 class DownloadCreate(BaseModel):
     title: str
     url: str

backend/tests/test_providers.py

@@ -0,0 +1,67 @@
+"""Availability tests for the search-provider layer (DAP-35 / DAP-51).
+
+These cover both the unit-level `check_availability` contract and the
+`/api/providers` endpoint that surfaces provider health to operators.
+"""
+
+import asyncio
+
+from fastapi.testclient import TestClient
+
+from src.main import app
+from src.providers import BaseProvider, MockProvider, check_all_providers
+from src.schemas import SearchResult
+
+
+client = TestClient(app)
+
+
+class FailingProvider(BaseProvider):
+    """Provider whose search always fails — simulates an unreachable source."""
+
+    name = "failing"
+
+    async def search(self, query: str) -> list[SearchResult]:
+        raise ConnectionError("upstream unreachable")
+
+
+def test_mock_provider_reports_available():
+    status = asyncio.run(MockProvider().check_availability())
+
+    assert status.name == "mock"
+    assert status.available is True
+    assert status.result_count == 3
+    assert status.error is None
+    assert status.latency_ms is not None and status.latency_ms >= 0
+
+
+def test_failing_provider_reports_unavailable_with_error():
+    status = asyncio.run(FailingProvider().check_availability())
+
+    assert status.name == "failing"
+    assert status.available is False
+    assert status.result_count is None
+    assert "ConnectionError" in status.error
+    assert "upstream unreachable" in status.error
+    assert status.latency_ms is not None
+
+
+def test_check_all_providers_covers_registry():
+    statuses = asyncio.run(check_all_providers())
+
+    assert len(statuses) >= 1
+    names = {s.name for s in statuses}
+    assert "mock" in names
+
+
+def test_providers_endpoint_returns_status_list():
+    response = client.get("/api/providers")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert isinstance(body, list) and body
+
+    mock = next(entry for entry in body if entry["name"] == "mock")
+    assert mock["available"] is True
+    assert mock["result_count"] == 3
+    assert mock["error"] is None