diff --git a/src/nsct/api/main.py b/src/nsct/api/main.py
index d44aae6..dee38e3 100644
--- a/src/nsct/api/main.py
+++ b/src/nsct/api/main.py
@@ -75,6 +75,10 @@ def create_app() -> FastAPI:
from nsct.api.health import router as health_router
app.include_router(health_router, tags=["system"])
+ # Mount search router
+ from nsct.api.search import router as search_router
+ app.include_router(search_router, tags=["search"])
+
return app
diff --git a/src/nsct/api/search.py b/src/nsct/api/search.py
new file mode 100644
index 0000000..9d640ef
--- /dev/null
+++ b/src/nsct/api/search.py
@@ -0,0 +1,120 @@
+"""Search endpoint — POST /search returns normalized search results."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from nsct.providers.abstract import MultiProviderSearch, NormalizedResult, SearchProvider
+from nsct.providers.duckduckgo import DuckDuckGoProvider
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+# ---------------------------------------------------------------------------
+# Provider factory — creates provider instances from env
+# ---------------------------------------------------------------------------
+
+
+def _build_search_engine() -> MultiProviderSearch:
+ """Baue den Search-Engine mit allen konfigurierten Providern auf.
+
+ DuckDuckGo ist der Default-Provider. Weitere Provider können in
+ Zukunft hinzugefügt werden, ohne den API-Code zu ändern.
+ """
+ engine = MultiProviderSearch()
+
+ # DuckDuckGo — immer aktiviert als Fallback-Provider
+ ddgs = DuckDuckGoProvider()
+ engine.add_provider(ddgs)
+
+ logger.info("Search engine initialized with providers: %s", [p.__class__.__name__ for p in engine._providers])
+ return engine
+
+
+# Lazy initialization — created on first request
+_search_engine: MultiProviderSearch | None = None
+
+
+def get_search_engine() -> MultiProviderSearch:
+ """Lazy-init des Search-Engines (einmal beim ersten Aufruf)."""
+ global _search_engine
+ if _search_engine is None:
+ _search_engine = _build_search_engine()
+ return _search_engine
+
+
+# ---------------------------------------------------------------------------
+# Request / Response Schemas
+# ---------------------------------------------------------------------------
+
+
+class SearchRequest(BaseModel):
+ """Eingabe für den Search-Endpoint."""
+
+ query: str = Field(..., min_length=1, max_length=500, description="Suchanfrage.")
+ language: str = Field(default="de", description="Sprachcode (z.B. 'de', 'en').")
+ max_results: int = Field(default=10, ge=1, le=50, description="Maximale Ergebnisanzahl.")
+ providers: list[str] | None = Field(
+ default=None,
+ description="Optionale Liste von Provider-Namen. Leer = alle aktiven.",
+ )
+
+
+class SearchResponse(BaseModel):
+ """Ausgabe des Search-Endpoints."""
+
+ results: list[NormalizedResult] = Field(
+ description="Normalisierte Suchergebnisse. rank ist KEIN truth_score."
+ )
+ debug: dict[str, Any] | None = Field(
+ default=None,
+ description="Debug-Informationen wenn NSCT_DEBUG=true.",
+ )
+
+
+# ---------------------------------------------------------------------------
+# Endpoint
+# ---------------------------------------------------------------------------
+
+
+@router.post("/search", response_model=SearchResponse, summary="Search via normalized search providers")
+async def search(request: SearchRequest) -> SearchResponse:
+ """POST /search — normale Suchanfrage mit normalisierten Ergebnissen.
+
+ Kein LLM — nur rohe Suchergebnisse normalisieren.
+ Bei NSCT_DEBUG=true werden Debug-Informationen mitgeliefert.
+ """
+ engine = get_search_engine()
+
+ debug_enabled = os.environ.get("NSCT_DEBUG", "false").lower() == "true"
+
+ try:
+ if debug_enabled:
+ debug_result = await engine.search_with_debug(
+ query=request.query,
+ language=request.language,
+ max_results=request.max_results,
+ provider_names=request.providers,
+ )
+ return SearchResponse(
+ results=debug_result["results"],
+ debug=debug_result["debug"],
+ )
+ else:
+ results = await engine.search(
+ query=request.query,
+ language=request.language,
+ max_results=request.max_results,
+ provider_names=request.providers,
+ )
+ return SearchResponse(results=results)
+ except Exception as exc:
+ logger.error("Search failed: %s", exc)
+ raise HTTPException(status_code=503, detail="Search service unavailable") from exc
\ No newline at end of file
diff --git a/src/nsct/providers/abstract.py b/src/nsct/providers/abstract.py
new file mode 100644
index 0000000..89a0cf1
--- /dev/null
+++ b/src/nsct/providers/abstract.py
@@ -0,0 +1,345 @@
+"""Search provider interface — neutral ranking, no trust implication."""
+
+from __future__ import annotations
+
+import abc
+import os
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+from pydantic import BaseModel, Field, field_validator
+
+
+# ---------------------------------------------------------------------------
+# NormalizedResult
+# ---------------------------------------------------------------------------
+
+
+class NormalizedResult(BaseModel):
+ """Normalisiertes Suchergebnis — provider-unabhängig.
+
+ WICHTIG: rank ist KEIN truth_score.
+ Suchergebnis auf Position 1 ist nicht automatisch glaubwürdiger als Position 8.
+ """
+
+ title: str = Field(..., min_length=1, description="Titel des Suchergebnisses.")
+ url: str = Field(..., description="Original-URL der Quelle.")
+ snippet: str = Field(default="", description="Auszug/Snippet aus dem Suchergebnis.")
+ provider: str = Field(..., description="Name des Providers, der dieses Ergebnis geliefert hat.")
+ rank: int = Field(
+ ...,
+ ge=1,
+ description="Original-Ranking des Providers. KEIN Vertrauens- oder Wahrheitsindikator.",
+ )
+ retrieved_at: datetime = Field(
+ default_factory=lambda: datetime.now(timezone.utc),
+ description="Zeitpunkt der Abrufung (UTC).",
+ )
+ extra: dict[str, Any] = Field(
+ default_factory=dict,
+ description="Zusätzliche provider-spezifische Metadaten.",
+ )
+
+ @field_validator("url")
+ @classmethod
+ def _validate_url(cls, v: str) -> str:
+ """Stelle sicher, dass URL einen validen Scheme hat."""
+ if not v.startswith(("http://", "https://")):
+ raise ValueError("URL muss mit http:// oder https:// beginnen")
+ return v
+
+ @field_validator("title")
+ @classmethod
+ def _validate_title(cls, v: str) -> str:
+ if not v.strip():
+ raise ValueError("title darf nicht leer sein")
+ return v
+
+ @classmethod
+ def from_raw(
+ cls,
+ provider_name: str,
+ title: str,
+ url: str,
+ snippet: str = "",
+ rank: int = 1,
+ extra: dict[str, Any] | None = None,
+ retrieved_at: datetime | None = None,
+ ) -> "NormalizedResult":
+ """Factory: Erzeuge ein NormalizedResult aus rohen Provider-Daten."""
+ if retrieved_at is None:
+ retrieved_at = datetime.now(timezone.utc)
+ return cls(
+ title=title.strip() if title else "",
+ url=url.strip(),
+ snippet=snippet.strip() if snippet else "",
+ provider=provider_name,
+ rank=rank,
+ retrieved_at=retrieved_at,
+ extra=extra or {},
+ )
+
+
+# ---------------------------------------------------------------------------
+# ProviderConfig
+# ---------------------------------------------------------------------------
+
+
+class ProviderConfig(BaseModel):
+ """Konfiguration für einen einzelnen SearchProvider."""
+
+ name: str = Field(..., description="Eindeutiger Provider-Name.")
+ enabled: bool = Field(default=True, description="Ob der Provider aktiviert ist.")
+ timeout_seconds: float = Field(
+ default=10.0,
+ description="Timeout pro Provider-Anfrage in Sekunden.",
+ gt=0,
+ )
+ max_results: int = Field(
+ default=10,
+ ge=1,
+ description="Max Ergebnisse pro Provider.",
+ )
+
+
+# ---------------------------------------------------------------------------
+# SearchProvider Interface
+# ---------------------------------------------------------------------------
+
+
+class SearchProvider(abc.ABC):
+ """Schnittstelle für Suchmaschinen.
+
+ Kein Provider darf als Evidenz-Ranking verwendet werden.
+ Das Ergebnis-Ranking einer Suchmaschine ist KEIN Vertrauensindikator.
+ """
+
+ _provider_name: str = ""
+ _provider_config: ProviderConfig | None = None
+
+ @abc.abstractmethod
+ async def search(
+ self,
+ query: str,
+ language: str = "de",
+ max_results: int = 10,
+ ) -> list[NormalizedResult]:
+ """Suche durchführen.
+
+ Parameters
+ ----------
+ query: Suchanfrage-Text.
+ language: Sprachcode (z.B. 'de', 'en').
+ max_results: Maximale Anzahl zurückzugebender Ergebnisse.
+
+ Returns
+ -------
+ list[NormalizedResult] — ohne Ranking-Implikation.
+ """
+ ...
+
+ @abc.abstractmethod
+ async def get_metadata(self) -> dict[str, Any]:
+ """Provider-Metadata (Name, Version, capabilities)."""
+ ...
+
+ async def health_check(self) -> bool:
+ """Prüfe, ob der Provider erreichbar ist."""
+ try:
+ meta = await self.get_metadata()
+ return bool(meta.get("name"))
+ except Exception:
+ return False
+
+
+# ---------------------------------------------------------------------------
+# Multi-Provider-Orchestrator
+# ---------------------------------------------------------------------------
+
+
+class MultiProviderSearch:
+ """Orchestrator für mehrere SearchProvider parallel.
+
+ - Sammelt Ergebnisse aller aktiven Provider.
+ - Dedupliziert nach URL.
+ - Behält provider-Information für jedes Ergebnis.
+ - Fällt stillschweigend auf aus, wenn ein Provider ausfällt.
+ """
+
+ def __init__(self, providers: list[SearchProvider] | None = None) -> None:
+ """Initialisiere mit einer Liste von SearchProvider-Instanzen."""
+ self._providers: list[SearchProvider] = providers or []
+ self._name_index: dict[str, SearchProvider] = {}
+ for p in self._providers:
+ name = getattr(p, "_provider_name", p.__class__.__name__)
+ self._name_index[name] = p
+
+ def add_provider(self, provider: SearchProvider, config: ProviderConfig | None = None) -> None:
+ """Füge einen Provider hinzu."""
+ self._providers.append(provider)
+ name = getattr(provider, "_provider_name", provider.__class__.__name__)
+ self._name_index[name] = provider
+
+ def get_provider(self, name: str) -> SearchProvider | None:
+ """Gib einen Provider nach Namen zurück."""
+ return self._name_index.get(name)
+
+ def enabled_providers(self) -> list[tuple[str, SearchProvider, ProviderConfig]]:
+ """Gib alle aktivierten (Name, Provider, Config)-Tupel zurück."""
+ results: list[tuple[str, SearchProvider, ProviderConfig]] = []
+ for p in self._providers:
+ name = getattr(p, "_provider_name", p.__class__.__name__)
+ cfg = getattr(p, "_provider_config", None)
+ if cfg is None:
+ cfg = ProviderConfig(name=name, enabled=True)
+ if not isinstance(cfg, ProviderConfig):
+ cfg = ProviderConfig(name=name, enabled=True)
+ if cfg.enabled:
+ results.append((name, p, cfg))
+ return results
+
+ async def search(
+ self,
+ query: str,
+ language: str = "de",
+ max_results: int = 10,
+ provider_names: list[str] | None = None,
+ timeout_seconds: float | None = None,
+ ) -> list[NormalizedResult]:
+ """Starte parallele Suche über alle (oder ausgewählte) Provider.
+
+ Returns
+ -------
+ Deduplizierte, normalisierte Ergebnisse.
+ """
+ import asyncio
+
+ candidates: list[tuple[str, SearchProvider, ProviderConfig]] = []
+ if provider_names:
+ for pname in provider_names:
+ p = self.get_provider(pname)
+ if p is None:
+ continue
+ cfg = getattr(p, "_provider_config", None)
+ if cfg is None:
+ cfg = ProviderConfig(name=pname, enabled=True)
+ elif not isinstance(cfg, ProviderConfig):
+ cfg = ProviderConfig(name=pname, enabled=True)
+ candidates.append((pname, p, cfg))
+ else:
+ candidates = self.enabled_providers()
+
+ if not candidates:
+ return []
+
+ async def _run(p: SearchProvider, cfg: ProviderConfig) -> list[NormalizedResult]:
+ to = timeout_seconds if timeout_seconds else cfg.timeout_seconds
+ try:
+ return await asyncio.wait_for(
+ p.search(query, language=language, max_results=cfg.max_results),
+ timeout=to,
+ )
+ except asyncio.TimeoutError:
+ return []
+ except Exception:
+ return []
+
+ tasks = [_run(p, cfg) for _, p, cfg in candidates]
+ raw_results: list[list[NormalizedResult]] = await asyncio.gather(*tasks)
+
+ # Zusammenführen und deduplizieren nach URL
+ seen_urls: set[str] = set()
+ deduped: list[NormalizedResult] = []
+ for batch in raw_results:
+ for r in batch:
+ url_key = r.url.lower().rstrip("/")
+ if url_key not in seen_urls:
+ seen_urls.add(url_key)
+ deduped.append(r)
+ if len(deduped) >= max_results:
+ break
+ if len(deduped) >= max_results:
+ break
+
+ return deduped[:max_results]
+
+ async def search_with_debug(
+ self,
+ query: str,
+ language: str = "de",
+ max_results: int = 10,
+ provider_names: list[str] | None = None,
+ timeout_seconds: float | None = None,
+ ) -> dict[str, Any]:
+ """Parallele Suche mit Debug-Informationen (Timing, Provider-Status)."""
+ import asyncio
+
+ start = time.monotonic()
+
+ candidates: list[tuple[str, SearchProvider, ProviderConfig]] = []
+ if provider_names:
+ for pname in provider_names:
+ p = self.get_provider(pname)
+ if p is None:
+ continue
+ cfg = getattr(p, "_provider_config", None)
+ if cfg is None:
+ cfg = ProviderConfig(name=pname, enabled=True)
+ elif not isinstance(cfg, ProviderConfig):
+ cfg = ProviderConfig(name=pname, enabled=True)
+ candidates.append((pname, p, cfg))
+ else:
+ candidates = self.enabled_providers()
+
+ timings: dict[str, float] = {}
+ results_by_provider: dict[str, list[NormalizedResult]] = {}
+ errors: dict[str, str] = {}
+
+ async def _run_debug(p: SearchProvider, cfg: ProviderConfig) -> None:
+ to = timeout_seconds if timeout_seconds else cfg.timeout_seconds
+ t0 = time.monotonic()
+ try:
+ res = await asyncio.wait_for(
+ p.search(query, language=language, max_results=cfg.max_results),
+ timeout=to,
+ )
+ timings[p.__class__.__name__] = time.monotonic() - t0
+ results_by_provider[p.__class__.__name__] = res
+ except asyncio.TimeoutError:
+ timings[p.__class__.__name__] = time.monotonic() - t0
+ errors[p.__class__.__name__] = "timeout"
+ except Exception as exc:
+ timings[p.__class__.__name__] = time.monotonic() - t0
+ errors[p.__class__.__name__] = str(exc)
+
+ tasks = [_run_debug(p, cfg) for _, p, cfg in candidates]
+ await asyncio.gather(*tasks)
+
+ # Dedupliziere
+ seen: set[str] = set()
+ deduped: list[NormalizedResult] = []
+ for batch in results_by_provider.values():
+ for r in batch:
+ u = r.url.lower().rstrip("/")
+ if u not in seen:
+ seen.add(u)
+ deduped.append(r)
+ deduped = deduped[:max_results]
+
+ elapsed = time.monotonic() - start
+ return {
+ "results": deduped,
+ "debug": {
+ "total_time_seconds": round(elapsed, 3),
+ "providers": {
+ name: {
+ "timing_seconds": round(timings.get(name, 0), 3),
+ "result_count": len(results_by_provider.get(name, [])),
+ "error": errors.get(name),
+ }
+ for name, _, _ in candidates
+ },
+ "total_results": len(deduped),
+ },
+ }
\ No newline at end of file
diff --git a/src/nsct/providers/duckduckgo.py b/src/nsct/providers/duckduckgo.py
new file mode 100644
index 0000000..5f87bb3
--- /dev/null
+++ b/src/nsct/providers/duckduckgo.py
@@ -0,0 +1,192 @@
+"""DuckDuckGo search provider — lightweight HTTP-based search.
+
+Uses DuckDuckGo's HTML search page (no API key required).
+This is a fallback provider — if DuckDuckGo is unavailable, the
+system continues functioning normally.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime, timezone
+from html.parser import HTMLParser
+from typing import Any
+
+import httpx
+
+from nsct.providers.abstract import NormalizedResult, SearchProvider
+
+# Environment-Variable für das DuckDuckGo Gateway
+# DDGS_BASE_URL — Base URL (default: DuckDuckGo HTML)
+# DDGS_USER_AGENT — Custom User-Agent string
+_DDGS_BASE_URL: str = os.environ.get("DDGS_BASE_URL", "https://html.duckduckgo.com")
+_DDGS_USER_AGENT: str = os.environ.get(
+ "DDGS_USER_AGENT",
+ "NSCT/1.0 Neutral Search Crawler Tool (research@localhost)",
+)
+
+
+class DuckDuckGoProvider(SearchProvider):
+ """DuckDuckGo Search Provider.
+
+ Nutzt die HTML-Version von DuckDuckGo über HTTP – keine API-Keys,
+ keine externen Services. Fallback-fähig: wenn DuckDuckGo nicht
+ erreichbar ist, gibt der Provider leere Ergebnisse zurück und das
+ Gesamtsystem funktioniert weiter.
+ """
+
+ _provider_name: str = "duckduckgo"
+
+ def __init__(self, base_url: str | None = None, user_agent: str | None = None) -> None:
+ """Initialisiere den DuckDuckGo-Provider.
+
+ Parameters
+ ----------
+ base_url: Optionale URL-Override (env DDGS_BASE_URL takes priority).
+ user_agent: Optionale User-Agent-Override.
+ """
+ self._base_url: str = base_url or _DDGS_BASE_URL
+ self._user_agent: str = user_agent or _DDGS_USER_AGENT
+ self._client: httpx.AsyncClient | None = None
+
+ @property
+ def _http_client(self) -> httpx.AsyncClient:
+ if self._client is None:
+ self._client = httpx.AsyncClient(
+ timeout=httpx.Timeout(15.0, connect=5.0),
+ headers={
+ "User-Agent": self._user_agent,
+ "Accept": "text/html,application/xhtml+xml",
+ },
+ follow_redirects=True,
+ )
+ return self._client
+
+ async def _search_html(self, query: str, language: str = "de") -> str:
+ """Führe die DuckDuckGo-Suche durch und liefere den HTML-Body."""
+ lang_map: dict[str, str] = {
+ "de": "de_de",
+ "en": "en_us",
+ "fr": "fr_fr",
+ "es": "es_es",
+ "it": "it_it",
+ "pt": "pt_br",
+ "nl": "nl_nl",
+ "ru": "ru_ru",
+ "ja": "ja_jp",
+ "zh": "zh_cn",
+ }
+ lang_param = lang_map.get(language, "en_us")
+
+ params: dict[str, str] = {
+ "q": query,
+ "lang": language,
+ }
+
+ query_str = "&".join(f"{k}={v}" for k, v in params.items())
+ url = f"{self._base_url}/html/?{query_str}"
+
+ resp = await self._http_client.get(url)
+ resp.raise_for_status()
+ return resp.text
+
+ def _parse_html(self, html: str, query: str) -> list[dict[str, str]]:
+ """Parst die DuckDuckGo-HTML-Seite und extrahiert Web-Ergebnisse."""
+ results: list[dict[str, str]] = []
+
+ # DuckDuckGo's HTML structure uses for result links
+ # and for snippets
+ import re
+
+ # Find result links
+ result_links = re.finditer(
+ r']*class="[^"]*result__a[^"]*"[^>]*>(.*?)',
+ html,
+ re.DOTALL,
+ )
+ # Find snippets
+ snippets = re.finditer(
+ r']*class="[^"]*result__snippet[^"]*"[^>]*>(.*?)',
+ html,
+ re.DOTALL,
+ )
+
+ link_urls = list(re.finditer(r']*class="[^"]*result__a[^"]*"[^>]*href="([^"]*)"', html))
+ snippet_texts = list(re.finditer(
+ r']*class="[^"]*result__snippet[^"]*"[^>]*>(.*?)',
+ html,
+ re.DOTALL,
+ ))
+
+ link_titles = list(re.finditer(
+ r']*class="[^"]*result__a[^"]*"[^>]*>(.*?)',
+ html,
+ re.DOTALL,
+ ))
+
+ for idx, link in enumerate(link_urls):
+ url = link.group(1)
+ title = link_titles[idx].group(1).strip() if idx < len(link_titles) else ""
+ snippet = snippet_texts[idx].group(1).strip() if idx < len(snippet_texts) else ""
+
+ # Strip HTML tags from title and snippet
+ title = re.sub(r"<[^>]*>", "", title).strip()
+ snippet = re.sub(r"<[^>]*>", "", snippet).strip()
+
+ if url and url.startswith(("http://", "https://")):
+ results.append({
+ "title": title or query,
+ "url": url,
+ "snippet": snippet[:500],
+ })
+
+ return results
+
+ async def search(
+ self,
+ query: str,
+ language: str = "de",
+ max_results: int = 10,
+ ) -> list[NormalizedResult]:
+ """Suche über DuckDuckGo HTML und normalisiere Ergebnisse."""
+ results: list[NormalizedResult] = []
+ try:
+ html = await self._search_html(query, language)
+ parsed = self._parse_html(html, query)
+ for idx, item in enumerate(parsed[:max_results], start=1):
+ results.append(
+ NormalizedResult.from_raw(
+ provider_name="duckduckgo",
+ title=item.get("title", query),
+ url=item.get("url", ""),
+ snippet=item.get("snippet", ""),
+ rank=idx,
+ )
+ )
+ except Exception:
+ # Fallback: kein Fehler werfen – leere Liste zurückgeben.
+ # Das System funktioniert trotzdem weiter.
+ pass
+ return results
+
+ async def get_metadata(self) -> dict[str, Any]:
+ """Gib Provider-Metadata zurück."""
+ return {
+ "name": "duckduckgo",
+ "version": "0.1.0",
+ "provider": self.__class__.__name__,
+ "capabilities": ["web_search", "html_scraping"],
+ "requires_api_key": False,
+ "base_url": self._base_url,
+ "language_support": ["de", "en", "fr", "es", "it", "pt", "nl", "ru", "ja", "zh"],
+ "note": "Search ranking is NOT a trust indicator.",
+ }
+
+ async def health_check(self) -> bool:
+ """Prüfe die Erreichbarkeit von DuckDuckGo."""
+ try:
+ async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=2.0)) as client:
+ resp = await client.get(f"{self._base_url}/", follow_redirects=True)
+ return resp.status_code == 200
+ except Exception:
+ return False
\ No newline at end of file
diff --git a/src/nsct/providers/multi.py b/src/nsct/providers/multi.py
new file mode 100644
index 0000000..5298225
--- /dev/null
+++ b/src/nsct/providers/multi.py
@@ -0,0 +1,12 @@
+"""Multi-provider orchestrator — re-exported from abstract for convenience."""
+
+from __future__ import annotations
+
+from nsct.providers.abstract import MultiProviderSearch, NormalizedResult, ProviderConfig, SearchProvider
+
+__all__ = [
+ "MultiProviderSearch",
+ "NormalizedResult",
+ "ProviderConfig",
+ "SearchProvider",
+]
\ No newline at end of file
diff --git a/tests/test_search.py b/tests/test_search.py
new file mode 100644
index 0000000..e7e78af
--- /dev/null
+++ b/tests/test_search.py
@@ -0,0 +1,369 @@
+"""Tests for the search provider layer."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+import pytest
+
+from nsct.providers.abstract import MultiProviderSearch, NormalizedResult, ProviderConfig, SearchProvider
+from nsct.providers.duckduckgo import DuckDuckGoProvider
+
+
+# ---------------------------------------------------------------------------
+# Fake provider for testing
+# ---------------------------------------------------------------------------
+
+
+class FakeSearchProvider(SearchProvider):
+ """Simpler Test-Provider, der vordefinierte Ergebnisse zurückgibt."""
+
+ _provider_name = "fake"
+
+ def __init__(self, results: list[dict], fail: bool = False) -> None:
+ """
+ Parameters
+ ----------
+ results: Liste von Dicts mit keys: title, url, snippet, rank.
+ fail: Wenn True, wirft search() eine Exception.
+ """
+ self._results = results
+ self._fail = fail
+ self.call_count = 0
+ self._provider_config = ProviderConfig(name="fake", enabled=True)
+
+ async def search(
+ self,
+ query: str,
+ language: str = "de",
+ max_results: int = 10,
+ ) -> list[NormalizedResult]:
+ self.call_count += 1
+ if self._fail:
+ raise RuntimeError("Simulated provider failure")
+ normalized: list[NormalizedResult] = []
+ for idx, item in enumerate(self._results[:max_results], start=1):
+ normalized.append(
+ NormalizedResult.from_raw(
+ provider_name="fake",
+ title=item.get("title", ""),
+ url=item.get("url", ""),
+ snippet=item.get("snippet", ""),
+ rank=idx,
+ )
+ )
+ return normalized
+
+ async def get_metadata(self) -> dict:
+ return {"name": "fake", "version": "0.0.1", "provider": "FakeSearchProvider"}
+
+ async def health_check(self) -> bool:
+ return not self._fail
+
+
+# ---------------------------------------------------------------------------
+# Tests: NormalizedResult
+# ---------------------------------------------------------------------------
+
+
+class TestNormalizedResult:
+ """Tests für NormalizedResult-Schema."""
+
+ def test_required_fields(self):
+ """Alle required fields müssen vorhanden sein."""
+ r = NormalizedResult.from_raw(
+ provider_name="test",
+ title="Test Title",
+ url="https://example.com",
+ snippet="Test snippet",
+ rank=1,
+ )
+ assert r.title == "Test Title"
+ assert r.url == "https://example.com"
+ assert r.snippet == "Test snippet"
+ assert r.provider == "test"
+ assert r.rank == 1
+ assert isinstance(r.retrieved_at, datetime)
+
+ def test_rank_is_not_truth(self):
+ """rank ist ein Ranking-Indikator, kein truth_score."""
+ r = NormalizedResult.from_raw(
+ provider_name="test",
+ title="Rank 5",
+ url="https://example.com/5",
+ rank=5,
+ )
+ assert r.rank == 5
+ assert r.rank != r.rank + 1
+ assert r.title != "" # Titel ist die Quelle der Wahrheit, nicht rank
+
+ def test_url_validation(self):
+ """URLs müssen mit http:// oder https:// beginnen."""
+ with pytest.raises(ValueError):
+ NormalizedResult.from_raw(
+ provider_name="test",
+ title="Bad URL",
+ url="not-a-url",
+ rank=1,
+ )
+
+ def test_empty_title_raises(self):
+ """Leere titles sind ungültig."""
+ with pytest.raises(ValueError):
+ NormalizedResult.from_raw(
+ provider_name="test",
+ title="",
+ url="https://example.com",
+ rank=1,
+ )
+
+ def test_extra_fields(self):
+ """Zusätzliche Metadaten können gespeichert werden."""
+ r = NormalizedResult.from_raw(
+ provider_name="test",
+ title="With extra",
+ url="https://example.com",
+ rank=1,
+ extra={"domain": "example.com", "category": "news"},
+ )
+ assert r.extra["domain"] == "example.com"
+ assert r.extra["category"] == "news"
+
+ def test_timestamp_default_utc(self):
+ """Standard-Zeitstempel muss UTC sein."""
+ r = NormalizedResult.from_raw(
+ provider_name="test",
+ title="Timestamp test",
+ url="https://example.com",
+ rank=1,
+ )
+ assert r.retrieved_at.tzinfo is not None
+
+ def test_from_raw_factory(self):
+ """Factory-Methode erzeugt korrekte NormalizedResult-Instanzen."""
+ r = NormalizedResult.from_raw(
+ provider_name="duckduckgo",
+ title="Test",
+ url="https://example.com",
+ snippet="Snippet text",
+ rank=1,
+ extra={"source": "web"},
+ )
+ assert r.provider == "duckduckgo"
+ assert r.extra == {"source": "web"}
+
+ def test_custom_retrieved_at(self):
+ """Benutzerdefinierter retrieved_at wird übernommen."""
+ custom = datetime(2025, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+ r = NormalizedResult.from_raw(
+ provider_name="test",
+ title="Custom time",
+ url="https://example.com",
+ rank=1,
+ retrieved_at=custom,
+ )
+ assert r.retrieved_at == custom
+
+
+# ---------------------------------------------------------------------------
+# Tests: MultiProviderSearch
+# ---------------------------------------------------------------------------
+
+
+class TestMultiProviderSearch:
+ """Tests für MultiProviderSearch-Orchestrator."""
+
+ def test_add_and_run_provider(self):
+ """Provider hinzufügen und Suche ausführen."""
+ fake = FakeSearchProvider([
+ {"title": "Result 1", "url": "https://example.com/1", "snippet": "First", "rank": 1},
+ {"title": "Result 2", "url": "https://example.com/2", "snippet": "Second", "rank": 2},
+ ])
+ engine = MultiProviderSearch()
+ engine.add_provider(fake)
+
+ results = asyncio.run(engine.search("test query"))
+ assert len(results) == 2
+ assert results[0].title == "Result 1"
+ assert results[1].url == "https://example.com/2"
+
+ def test_parallel_multiple_providers(self):
+ """Multiple Provider parallel abfragen."""
+ fake_a = FakeSearchProvider([
+ {"title": "A-1", "url": "https://a.example/1", "snippet": "A first", "rank": 1},
+ {"title": "A-2", "url": "https://a.example/2", "snippet": "A second", "rank": 2},
+ ])
+ fake_b = FakeSearchProvider([
+ {"title": "B-1", "url": "https://b.example/1", "snippet": "B first", "rank": 1},
+ ])
+
+ engine = MultiProviderSearch()
+ engine.add_provider(fake_a)
+ engine.add_provider(fake_b)
+
+ results = asyncio.run(engine.search("test query"))
+ # Alle 3 Ergebnisse sollten vorhanden sein
+ assert len(results) == 3
+ urls = {r.url for r in results}
+ assert "https://a.example/1" in urls
+ assert "https://a.example/2" in urls
+ assert "https://b.example/1" in urls
+
+ def test_url_deduplication(self):
+ """Doppelte URLs nach URL werden dedupliziert."""
+ fake_a = FakeSearchProvider([
+ {"title": "Same URL A", "url": "https://example.com/same", "snippet": "From A", "rank": 1},
+ ])
+ fake_b = FakeSearchProvider([
+ {"title": "Same URL B", "url": "https://example.com/same", "snippet": "From B", "rank": 1},
+ ])
+
+ engine = MultiProviderSearch()
+ engine.add_provider(fake_a)
+ engine.add_provider(fake_b)
+
+ results = asyncio.run(engine.search("test query"))
+ # Nur eine Instanz der URL sollte vorkommen
+ assert len(results) == 1
+ assert results[0].url == "https://example.com/same"
+
+ def test_fallback_when_provider_fails(self):
+ """Wenn ein Provider ausfällt, funktionieren andere weiter."""
+ failing = FakeSearchProvider([], fail=True)
+ working = FakeSearchProvider([
+ {"title": "Still works", "url": "https://works.com", "snippet": "I work", "rank": 1},
+ ])
+
+ engine = MultiProviderSearch()
+ engine.add_provider(failing)
+ engine.add_provider(working)
+
+ results = asyncio.run(engine.search("test query"))
+ # Nur das Ergebnis des working-Providers sollte da sein
+ assert len(results) == 1
+ assert results[0].url == "https://works.com"
+
+ def test_empty_provider_list(self):
+ """Keine Provider → leere Ergebnisse."""
+ engine = MultiProviderSearch()
+ results = asyncio.run(engine.search("test query"))
+ assert len(results) == 0
+
+ def test_max_results_limit(self):
+ """max_results limitiert die Rückgabeanzahl."""
+ big_provider = FakeSearchProvider([
+ {"title": f"Item {i}", "url": f"https://example.com/{i}", "snippet": f"Snip {i}", "rank": i}
+ for i in range(1, 11) # 10 results
+ ])
+
+ engine = MultiProviderSearch()
+ engine.add_provider(big_provider)
+
+ results = asyncio.run(engine.search("test query", max_results=3))
+ assert len(results) <= 3
+
+ def test_get_provider_by_name(self):
+ """Provider nach Namen lookup."""
+ fake = FakeSearchProvider([])
+ engine = MultiProviderSearch()
+ engine.add_provider(fake)
+
+ retrieved = engine.get_provider("fake")
+ assert retrieved is fake
+
+ def test_enabled_providers(self):
+ """Nur enabled Provider werden zurückgegeben."""
+ fake = FakeSearchProvider([])
+ engine = MultiProviderSearch()
+ engine.add_provider(fake)
+
+ enabled = engine.enabled_providers()
+ assert len(enabled) == 1
+ assert enabled[0][0] == "fake"
+
+ def test_disabled_provider_not_searched(self):
+ """Deaktivierte Provider werden nicht abgefragt."""
+ fake = FakeSearchProvider(
+ [{"title": "Disabled", "url": "https://disabled.com", "snippet": "nope", "rank": 1}]
+ )
+ fake._provider_config = ProviderConfig(name="fake", enabled=False)
+
+ engine = MultiProviderSearch()
+ engine.add_provider(fake)
+
+ results = asyncio.run(engine.search("test query"))
+ assert len(results) == 0
+
+ def test_provider_names_filter(self):
+ """Nur explizit angegebene Provider werden abgefragt."""
+ fake_a = FakeSearchProvider([{"title": "A", "url": "https://a.com", "snippet": "a", "rank": 1}])
+ fake_b = FakeSearchProvider([{"title": "B", "url": "https://b.com", "snippet": "b", "rank": 1}])
+
+ engine = MultiProviderSearch()
+ engine.add_provider(fake_a)
+ engine.add_provider(fake_b)
+
+ # query mit provider_names
+ results = asyncio.run(engine.search("test query", provider_names=["fake"]))
+ assert len(results) == 1 # Beide haben den Namen "fake", nur eine Instanz
+
+ def test_search_with_debug_empty(self):
+ """Debug-Mode mit keiner Provider gibt leere Results."""
+ engine = MultiProviderSearch()
+ result = asyncio.run(engine.search_with_debug("test"))
+ assert result["debug"]["total_results"] == 0
+ assert len(result["results"]) == 0
+
+
+# ---------------------------------------------------------------------------
+# Tests: DuckDuckGo Provider
+# ---------------------------------------------------------------------------
+
+
+class TestDuckDuckGoProvider:
+ """Tests für DuckDuckGoProvider."""
+
+ def test_metadata(self):
+ """Provider gibt korrekte Metadata zurück."""
+ provider = DuckDuckGoProvider()
+ meta = asyncio.run(provider.get_metadata())
+ assert meta["name"] == "duckduckgo"
+ assert meta["requires_api_key"] is False
+ assert meta["capabilities"] == ["web_search", "html_scraping"]
+
+ def test_metadata_structure(self):
+ """Metadata enthält alle erwarteten Felder."""
+ provider = DuckDuckGoProvider()
+ meta = asyncio.run(provider.get_metadata())
+ assert "name" in meta
+ assert "version" in meta
+ assert "provider" in meta
+ assert "capabilities" in meta
+ assert "note" in meta
+ assert "Search ranking is NOT a trust indicator." in meta["note"]
+
+ def test_is_search_provider(self):
+ """DuckDuckGoProvider implementiert SearchProvider."""
+ provider = DuckDuckGoProvider()
+ assert isinstance(provider, SearchProvider)
+
+ def test_no_api_key_required(self):
+ """DuckDuckGo benötigt keinen API-Key."""
+ provider = DuckDuckGoProvider()
+ meta = asyncio.run(provider.get_metadata())
+ assert meta["requires_api_key"] is False
+
+ def test_url_validation_for_duckduckgo(self):
+ """Ergebnisse von DuckDuckGo durchlaufen URL-Validierung."""
+ provider = DuckDuckGoProvider()
+ # Der Provider selbst ist ein SearchProvider, das ist der Test.
+ assert provider._provider_name == "duckduckgo"
+
+ def test_health_check_interface(self):
+ """health_check gibt bool zurück."""
+ provider = DuckDuckGoProvider()
+ # Wir können nicht die tatsächliche Erreichbarkeit testen ohne Netzwerk,
+ # aber wir prüfen, dass die Methode existiert und bool zurückgibt.
+ result = asyncio.run(provider.health_check())
+ assert isinstance(result, bool)
\ No newline at end of file