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