"""Tests for the DuckDuckGo (ddgs) web search provider. Covers: - DDGSSearchProvider.is_configured() — reflects package importability - DDGSSearchProvider.search() — happy path, missing package, runtime error - Result normalization (title, url, description, position) - _is_backend_available("ddgs") / _get_backend() integration - web_extract / web_crawl return search-only errors when ddgs is active """ from __future__ import annotations import json import sys import types from unittest.mock import MagicMock def _install_fake_ddgs(monkeypatch, *, text_results=None, text_raises=None): """Install a stub ``ddgs`` module in sys.modules for the duration of a test. ``text_results``: iterable of dicts to yield from DDGS().text(...). ``text_raises``: if set, DDGS().text raises this exception instead. """ fake = types.ModuleType("ddgs") class _FakeDDGS: def __enter__(self): return self def __exit__(self, *_a): return False def text(self, query, max_results=5): if text_raises is not None: raise text_raises for hit in (text_results or []): yield hit fake.DDGS = _FakeDDGS monkeypatch.setitem(sys.modules, "ddgs", fake) return fake # --------------------------------------------------------------------------- # DDGSSearchProvider unit tests # --------------------------------------------------------------------------- class TestDDGSProviderIsConfigured: def test_configured_when_package_importable(self, monkeypatch): _install_fake_ddgs(monkeypatch) # Drop any cached ``tools.web_providers.ddgs`` so is_configured re-imports ddgs fresh monkeypatch.delitem(sys.modules, "tools.web_providers.ddgs", raising=False) from tools.web_providers.ddgs import DDGSSearchProvider assert DDGSSearchProvider().is_configured() is True def test_not_configured_when_package_missing(self, monkeypatch): monkeypatch.delitem(sys.modules, "ddgs", raising=False) monkeypatch.delitem(sys.modules, "tools.web_providers.ddgs", raising=False) # Block the import so ``import ddgs`` raises ImportError even if the package is actually installed import builtins orig_import = builtins.__import__ def blocked_import(name, *args, **kwargs): if name == "ddgs": raise ImportError("blocked for test") return orig_import(name, *args, **kwargs) monkeypatch.setattr(builtins, "__import__", blocked_import) from tools.web_providers.ddgs import DDGSSearchProvider assert DDGSSearchProvider().is_configured() is False def test_provider_name(self): from tools.web_providers.ddgs import DDGSSearchProvider assert DDGSSearchProvider().provider_name() == "ddgs" def test_implements_web_search_provider(self): from tools.web_providers.base import WebSearchProvider from tools.web_providers.ddgs import DDGSSearchProvider assert issubclass(DDGSSearchProvider, WebSearchProvider) class TestDDGSProviderSearch: def test_happy_path_normalizes_results(self, monkeypatch): _install_fake_ddgs(monkeypatch, text_results=[ {"title": "A", "href": "https://a.example.com", "body": "desc A"}, {"title": "B", "href": "https://b.example.com", "body": "desc B"}, {"title": "C", "href": "https://c.example.com", "body": "desc C"}, ]) from tools.web_providers.ddgs import DDGSSearchProvider result = DDGSSearchProvider().search("q", limit=5) assert result["success"] is True web = result["data"]["web"] assert len(web) == 3 assert web[0] == {"title": "A", "url": "https://a.example.com", "description": "desc A", "position": 1} assert web[2]["position"] == 3 def test_accepts_url_key_as_fallback_for_href(self, monkeypatch): _install_fake_ddgs(monkeypatch, text_results=[ {"title": "A", "url": "https://a.example.com", "body": "desc A"}, ]) from tools.web_providers.ddgs import DDGSSearchProvider result = DDGSSearchProvider().search("q", limit=5) assert result["success"] is True assert result["data"]["web"][0]["url"] == "https://a.example.com" def test_limit_is_respected(self, monkeypatch): _install_fake_ddgs(monkeypatch, text_results=[ {"title": f"R{i}", "href": f"https://r{i}.example.com", "body": ""} for i in range(10) ]) from tools.web_providers.ddgs import DDGSSearchProvider result = DDGSSearchProvider().search("q", limit=3) assert result["success"] is True assert len(result["data"]["web"]) == 3 def test_missing_package_returns_failure(self, monkeypatch): monkeypatch.delitem(sys.modules, "ddgs", raising=False) monkeypatch.delitem(sys.modules, "tools.web_providers.ddgs", raising=False) import builtins orig_import = builtins.__import__ def blocked_import(name, *args, **kwargs): if name == "ddgs": raise ImportError("blocked for test") return orig_import(name, *args, **kwargs) monkeypatch.setattr(builtins, "__import__", blocked_import) from tools.web_providers.ddgs import DDGSSearchProvider result = DDGSSearchProvider().search("q", limit=5) assert result["success"] is False assert "ddgs" in result["error"].lower() def test_runtime_error_returns_failure(self, monkeypatch): _install_fake_ddgs(monkeypatch, text_raises=RuntimeError("rate limited 202")) from tools.web_providers.ddgs import DDGSSearchProvider result = DDGSSearchProvider().search("q", limit=5) assert result["success"] is False assert "rate limited" in result["error"] or "failed" in result["error"].lower() def test_empty_results(self, monkeypatch): _install_fake_ddgs(monkeypatch, text_results=[]) from tools.web_providers.ddgs import DDGSSearchProvider result = DDGSSearchProvider().search("nothing", limit=5) assert result["success"] is True assert result["data"]["web"] == [] # --------------------------------------------------------------------------- # Integration: _is_backend_available / _get_backend / check_web_api_key # --------------------------------------------------------------------------- class TestDDGSBackendWiring: def test_is_backend_available_true_when_package_importable(self, monkeypatch): from tools import web_tools monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) assert web_tools._is_backend_available("ddgs") is True def test_is_backend_available_false_when_package_missing(self, monkeypatch): from tools import web_tools monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False) assert web_tools._is_backend_available("ddgs") is False def test_configured_backend_accepted(self, monkeypatch): from tools import web_tools monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "ddgs"}) monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) assert web_tools._get_backend() == "ddgs" def test_ddgs_trails_paid_providers_in_auto_detect(self, monkeypatch): """Exa (priority) should win over ddgs in auto-detect.""" from tools import web_tools monkeypatch.setattr(web_tools, "_load_web_config", lambda: {}) for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "PARALLEL_API_KEY", "TAVILY_API_KEY", "SEARXNG_URL", "BRAVE_SEARCH_API_KEY"): monkeypatch.delenv(key, raising=False) monkeypatch.setenv("EXA_API_KEY", "exa-key") monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False) monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) assert web_tools._get_backend() == "exa" def test_auto_detect_picks_ddgs_as_last_resort(self, monkeypatch): from tools import web_tools monkeypatch.setattr(web_tools, "_load_web_config", lambda: {}) for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "PARALLEL_API_KEY", "TAVILY_API_KEY", "EXA_API_KEY", "SEARXNG_URL", "BRAVE_SEARCH_API_KEY"): monkeypatch.delenv(key, raising=False) monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False) monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) assert web_tools._get_backend() == "ddgs" def test_check_web_api_key_true_when_ddgs_configured(self, monkeypatch): from tools import web_tools monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "ddgs"}) monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) assert web_tools.check_web_api_key() is True # --------------------------------------------------------------------------- # ddgs is search-only: web_extract / web_crawl return clear errors # --------------------------------------------------------------------------- class TestDDGSSearchOnlyErrors: def test_web_extract_returns_search_only_error(self, monkeypatch): import asyncio from tools import web_tools monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "ddgs"}) monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False) monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False) result_str = asyncio.get_event_loop().run_until_complete( web_tools.web_extract_tool(["https://example.com"]) ) result = json.loads(result_str) assert result["success"] is False assert "search-only" in result["error"].lower() assert "duckduckgo" in result["error"].lower() or "ddgs" in result["error"].lower() def test_web_crawl_returns_search_only_error(self, monkeypatch): import asyncio from tools import web_tools monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "ddgs"}) monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True) monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False) monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False) monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False) result_str = asyncio.get_event_loop().run_until_complete( web_tools.web_crawl_tool("https://example.com") ) result = json.loads(result_str) assert result["success"] is False assert "search-only" in result["error"].lower() assert "duckduckgo" in result["error"].lower() or "ddgs" in result["error"].lower()