mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-08 03:01:47 +00:00
feat(web): add Brave Search (free tier) and DDGS search providers
Both implement WebSearchProvider via tools/web_providers/ — matching the
existing SearXNG pattern (PR #5c906d702). Search-only; pair with any
extract provider via web.extract_backend.
- tools/web_providers/brave_free.py — Brave Search API (free tier, 2k
queries/mo). Uses BRAVE_SEARCH_API_KEY as X-Subscription-Token.
- tools/web_providers/ddgs.py — DuckDuckGo via the ddgs Python package.
No API key; gated on package importability.
- tools/web_tools.py: both backends added to _get_backend() config list
and auto-detect chain (trails paid providers), _is_backend_available,
web_search_tool dispatch, web_extract_tool + web_crawl_tool search-only
refusals, check_web_api_key, and the __main__ diagnostic. Introduces
_ddgs_package_importable() helper so tests can monkeypatch a single
symbol for the ddgs availability check.
- hermes_cli/tools_config.py: picker entries for both providers; ddgs
gets a post_setup handler that runs `pip install ddgs`.
- hermes_cli/config.py: BRAVE_SEARCH_API_KEY in OPTIONAL_ENV_VARS.
- scripts/release.py: AUTHOR_MAP entry for @Abd0r.
- tests: 14 new tests (brave-free) + 15 new tests (ddgs) covering
provider unit behavior, backend wiring, and search-only refusals.
Salvages the brave-free + ddgs portion of PR #19796. Not included: the
in-line helpers in web_tools.py (replaced with provider modules to match
the shipped architecture), the lynx-based extract path (these backends
should refuse extract with a clear error — users pair with a real
extract provider), and scripts/start-llama-server.sh (unrelated).
Co-authored-by: Abd0r <223003280+Abd0r@users.noreply.github.com>
This commit is contained in:
parent
cdc0a47dd5
commit
04193cf71c
8 changed files with 862 additions and 9 deletions
275
tests/tools/test_web_providers_brave_free.py
Normal file
275
tests/tools/test_web_providers_brave_free.py
Normal file
|
|
@ -0,0 +1,275 @@
|
|||
"""Tests for the Brave Search (free tier) web search provider.
|
||||
|
||||
Covers:
|
||||
- BraveFreeSearchProvider.is_configured() env var gating
|
||||
- BraveFreeSearchProvider.search() — happy path, HTTP error, request error, bad JSON
|
||||
- Result normalization (title, url, description, position)
|
||||
- Limit truncation + Brave's count cap (20)
|
||||
- _is_backend_available("brave-free") integration
|
||||
- _get_backend() recognizes "brave-free" as a valid configured backend
|
||||
- check_web_api_key() includes brave-free in availability check
|
||||
- web_extract / web_crawl return search-only errors when brave-free is active
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# BraveFreeSearchProvider unit tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBraveFreeProviderIsConfigured:
|
||||
def test_configured_when_key_set(self, monkeypatch):
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
assert BraveFreeSearchProvider().is_configured() is True
|
||||
|
||||
def test_not_configured_when_key_missing(self, monkeypatch):
|
||||
monkeypatch.delenv("BRAVE_SEARCH_API_KEY", raising=False)
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
assert BraveFreeSearchProvider().is_configured() is False
|
||||
|
||||
def test_not_configured_when_key_whitespace(self, monkeypatch):
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", " ")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
assert BraveFreeSearchProvider().is_configured() is False
|
||||
|
||||
def test_provider_name(self):
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
assert BraveFreeSearchProvider().provider_name() == "brave-free"
|
||||
|
||||
def test_implements_web_search_provider(self):
|
||||
from tools.web_providers.base import WebSearchProvider
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
assert issubclass(BraveFreeSearchProvider, WebSearchProvider)
|
||||
|
||||
|
||||
class TestBraveFreeProviderSearch:
|
||||
_SAMPLE_RESPONSE = {
|
||||
"web": {
|
||||
"results": [
|
||||
{"title": "A", "url": "https://a.example.com", "description": "desc A"},
|
||||
{"title": "B", "url": "https://b.example.com", "description": "desc B"},
|
||||
{"title": "C", "url": "https://c.example.com", "description": "desc C"},
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _mock_resp(json_data, status_code=200):
|
||||
m = MagicMock()
|
||||
m.status_code = status_code
|
||||
m.json.return_value = json_data
|
||||
m.raise_for_status = MagicMock()
|
||||
return m
|
||||
|
||||
def test_happy_path_normalizes_results(self, monkeypatch):
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
with patch("httpx.get", return_value=self._mock_resp(self._SAMPLE_RESPONSE)):
|
||||
result = BraveFreeSearchProvider().search("test query", 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_sends_subscription_token_header_and_count(self, monkeypatch):
|
||||
"""Brave uses X-Subscription-Token; count maps from limit."""
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_get(url, **kwargs):
|
||||
captured["url"] = url
|
||||
captured["headers"] = kwargs.get("headers", {})
|
||||
captured["params"] = kwargs.get("params", {})
|
||||
return self._mock_resp({"web": {"results": []}})
|
||||
|
||||
with patch("httpx.get", side_effect=fake_get):
|
||||
BraveFreeSearchProvider().search("q", limit=5)
|
||||
|
||||
assert captured["url"] == "https://api.search.brave.com/res/v1/web/search"
|
||||
assert captured["headers"].get("X-Subscription-Token") == "BSAkey123"
|
||||
assert captured["params"].get("q") == "q"
|
||||
assert captured["params"].get("count") == 5
|
||||
|
||||
def test_count_is_capped_at_20(self, monkeypatch):
|
||||
"""Brave caps count at 20 — limit above that clamps."""
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_get(url, **kwargs):
|
||||
captured["params"] = kwargs.get("params", {})
|
||||
return self._mock_resp({"web": {"results": []}})
|
||||
|
||||
with patch("httpx.get", side_effect=fake_get):
|
||||
BraveFreeSearchProvider().search("q", limit=100)
|
||||
|
||||
assert captured["params"].get("count") == 20
|
||||
|
||||
def test_limit_is_respected_client_side(self, monkeypatch):
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
with patch("httpx.get", return_value=self._mock_resp(self._SAMPLE_RESPONSE)):
|
||||
result = BraveFreeSearchProvider().search("q", limit=2)
|
||||
|
||||
assert result["success"] is True
|
||||
assert len(result["data"]["web"]) == 2
|
||||
|
||||
def test_empty_results(self, monkeypatch):
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
with patch("httpx.get", return_value=self._mock_resp({"web": {"results": []}})):
|
||||
result = BraveFreeSearchProvider().search("nothing", limit=5)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["data"]["web"] == []
|
||||
|
||||
def test_missing_web_key_returns_empty(self, monkeypatch):
|
||||
"""Responses without a ``web`` block should produce an empty result set, not crash."""
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
with patch("httpx.get", return_value=self._mock_resp({})):
|
||||
result = BraveFreeSearchProvider().search("q", limit=5)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["data"]["web"] == []
|
||||
|
||||
def test_http_error_returns_failure(self, monkeypatch):
|
||||
import httpx
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
bad = MagicMock()
|
||||
bad.status_code = 429
|
||||
err = httpx.HTTPStatusError("429", request=MagicMock(), response=bad)
|
||||
|
||||
with patch("httpx.get", side_effect=err):
|
||||
result = BraveFreeSearchProvider().search("q", limit=5)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "429" in result["error"]
|
||||
|
||||
def test_request_error_returns_failure(self, monkeypatch):
|
||||
import httpx
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
with patch("httpx.get", side_effect=httpx.RequestError("boom")):
|
||||
result = BraveFreeSearchProvider().search("q", limit=5)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "boom" in result["error"] or "Brave" in result["error"]
|
||||
|
||||
def test_missing_key_returns_failure(self, monkeypatch):
|
||||
monkeypatch.delenv("BRAVE_SEARCH_API_KEY", raising=False)
|
||||
from tools.web_providers.brave_free import BraveFreeSearchProvider
|
||||
|
||||
result = BraveFreeSearchProvider().search("q", limit=5)
|
||||
assert result["success"] is False
|
||||
assert "BRAVE_SEARCH_API_KEY" in result["error"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration: _is_backend_available / _get_backend / check_web_api_key
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBraveFreeBackendWiring:
|
||||
def test_is_backend_available_true_when_key_set(self, monkeypatch):
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
from tools.web_tools import _is_backend_available
|
||||
assert _is_backend_available("brave-free") is True
|
||||
|
||||
def test_is_backend_available_false_when_key_missing(self, monkeypatch):
|
||||
monkeypatch.delenv("BRAVE_SEARCH_API_KEY", raising=False)
|
||||
from tools.web_tools import _is_backend_available
|
||||
assert _is_backend_available("brave-free") is False
|
||||
|
||||
def test_configured_backend_accepted(self, monkeypatch):
|
||||
from tools import web_tools
|
||||
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "brave-free"})
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
assert web_tools._get_backend() == "brave-free"
|
||||
|
||||
def test_auto_detect_picks_brave_free_when_only_key_set(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"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
|
||||
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False)
|
||||
assert web_tools._get_backend() == "brave-free"
|
||||
|
||||
def test_brave_free_does_not_override_paid_provider(self, monkeypatch):
|
||||
"""Tavily (higher priority) should win 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", "EXA_API_KEY", "SEARXNG_URL"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setenv("TAVILY_API_KEY", "tvly")
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
|
||||
assert web_tools._get_backend() == "tavily"
|
||||
|
||||
def test_check_web_api_key_true_when_brave_free_configured(self, monkeypatch):
|
||||
from tools import web_tools
|
||||
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "brave-free"})
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
assert web_tools.check_web_api_key() is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# brave-free is search-only: web_extract / web_crawl return clear errors
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBraveFreeSearchOnlyErrors:
|
||||
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": "brave-free"})
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
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 "brave" 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": "brave-free"})
|
||||
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
|
||||
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 "brave" in result["error"].lower()
|
||||
246
tests/tools/test_web_providers_ddgs.py
Normal file
246
tests/tools/test_web_providers_ddgs.py
Normal file
|
|
@ -0,0 +1,246 @@
|
|||
"""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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue