mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-18 04:41:56 +00:00
6b219f5af6
Deletes tools/web_providers/{brave_free,ddgs,searxng}.py — the three
providers that moved to plugins/web/ in prior commits. tools/web_tools.py
no longer imports them (registry dispatch as of d8735963f), so removing
them is purely a cleanup pass.
Also migrates the existing tests to the new import paths:
tests/tools/test_web_providers_brave_free.py
tests/tools/test_web_providers_ddgs.py
tests/tools/test_web_providers_searxng.py
Mechanical rewrites:
- `from tools.web_providers.X import YSearchProvider`
-> `from plugins.web.X.provider import YWebSearchProvider`
- `.is_configured()` -> `.is_available()` (legacy method -> new method)
- `.provider_name()` -> `.name` (legacy method -> new property)
- `from tools.web_providers.base import WebSearchProvider`
-> `from agent.web_search_provider import WebSearchProvider`
(the subclass-check asserts membership in the new plugin-facing ABC)
- `sys.modules.delitem("tools.web_providers.ddgs")` updated to point at
`plugins.web.ddgs.provider` (cache-busting for lazy ddgs imports)
The TestXBackendWiring / TestXSearchOnlyErrors classes (covering
_is_backend_available, _get_backend, check_web_api_key, and the
"search-only" error paths in web_extract/web_crawl) are untouched —
those still test web_tools.py's backend-selection logic, which continues
to recognize the names "brave-free" / "ddgs" / "searxng" even after the
modules behind them moved to plugins.
tools/web_providers/base.py is intentionally NOT deleted by this commit
— it's the parent ABC of the legacy modules and shares its name with
agent/web_search_provider.py::WebSearchProvider. Removing it surfaces the
naming collision (see PR description Finding 0); the real migration PR
deletes it in the same commit that drops the _WEB_PLUGIN_SKIPLIST
guards in hermes_cli/tools_config.py.
Test results:
bash scripts/run_tests.sh tests/tools/test_web_providers_*.py
-> 65 passed in 3.41s (all rewritten unit tests + unchanged integration tests)
bash scripts/run_tests.sh tests/tools/test_web_*.py
-> 141 passed in 4.70s (full web test set, post-deletion)
337 lines
15 KiB
Python
337 lines
15 KiB
Python
"""Tests for the SearXNG web search provider.
|
|
|
|
Covers:
|
|
- SearXNGWebSearchProvider.is_available() env var gating
|
|
- SearXNGWebSearchProvider.search() — happy path, HTTP error, request error, bad JSON
|
|
- Result normalization (title, url, description, position)
|
|
- Score-based sorting and limit truncation
|
|
- _is_backend_available("searxng") integration
|
|
- _get_backend() recognizes "searxng" as a valid configured backend
|
|
- check_web_api_key() includes searxng in availability check
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SearXNGWebSearchProvider unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSearXNGSearchProviderIsConfigured:
|
|
def test_configured_when_url_set(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
assert SearXNGWebSearchProvider().is_available() is True
|
|
|
|
def test_not_configured_when_url_missing(self, monkeypatch):
|
|
monkeypatch.delenv("SEARXNG_URL", raising=False)
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
assert SearXNGWebSearchProvider().is_available() is False
|
|
|
|
def test_not_configured_when_url_empty_string(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", " ")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
assert SearXNGWebSearchProvider().is_available() is False
|
|
|
|
def test_provider_name(self):
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
assert SearXNGWebSearchProvider().name == "searxng"
|
|
|
|
def test_implements_web_search_provider(self):
|
|
from agent.web_search_provider import WebSearchProvider
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
assert issubclass(SearXNGWebSearchProvider, WebSearchProvider)
|
|
|
|
|
|
class TestSearXNGSearchProviderSearch:
|
|
"""Happy path and error handling for SearXNGWebSearchProvider.search()."""
|
|
|
|
_SAMPLE_RESPONSE = {
|
|
"results": [
|
|
{"title": "Result A", "url": "https://a.example.com", "content": "Desc A", "score": 0.9},
|
|
{"title": "Result B", "url": "https://b.example.com", "content": "Desc B", "score": 0.7},
|
|
{"title": "Result C", "url": "https://c.example.com", "content": "Desc C", "score": 0.5},
|
|
]
|
|
}
|
|
|
|
def _make_mock_response(self, json_data, status_code=200):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = status_code
|
|
mock_resp.json.return_value = json_data
|
|
mock_resp.raise_for_status = MagicMock()
|
|
return mock_resp
|
|
|
|
def test_happy_path_returns_normalized_results(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
mock_resp = self._make_mock_response(self._SAMPLE_RESPONSE)
|
|
|
|
with patch("httpx.get", return_value=mock_resp):
|
|
result = SearXNGWebSearchProvider().search("test query", limit=5)
|
|
|
|
assert result["success"] is True
|
|
web = result["data"]["web"]
|
|
assert len(web) == 3
|
|
assert web[0]["title"] == "Result A"
|
|
assert web[0]["url"] == "https://a.example.com"
|
|
assert web[0]["description"] == "Desc A"
|
|
assert web[0]["position"] == 1
|
|
|
|
def test_results_sorted_by_score_descending(self, monkeypatch):
|
|
"""Results should be sorted by score before limit is applied."""
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
unordered = {
|
|
"results": [
|
|
{"title": "Low", "url": "https://low.example.com", "content": "", "score": 0.1},
|
|
{"title": "High", "url": "https://high.example.com", "content": "", "score": 0.99},
|
|
{"title": "Mid", "url": "https://mid.example.com", "content": "", "score": 0.5},
|
|
]
|
|
}
|
|
mock_resp = self._make_mock_response(unordered)
|
|
|
|
with patch("httpx.get", return_value=mock_resp):
|
|
result = SearXNGWebSearchProvider().search("query", limit=5)
|
|
|
|
assert result["success"] is True
|
|
assert result["data"]["web"][0]["title"] == "High"
|
|
assert result["data"]["web"][1]["title"] == "Mid"
|
|
assert result["data"]["web"][2]["title"] == "Low"
|
|
|
|
def test_limit_is_respected(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
mock_resp = self._make_mock_response(self._SAMPLE_RESPONSE)
|
|
|
|
with patch("httpx.get", return_value=mock_resp):
|
|
result = SearXNGWebSearchProvider().search("query", limit=2)
|
|
|
|
assert result["success"] is True
|
|
assert len(result["data"]["web"]) == 2
|
|
|
|
def test_position_is_one_indexed(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
mock_resp = self._make_mock_response(self._SAMPLE_RESPONSE)
|
|
|
|
with patch("httpx.get", return_value=mock_resp):
|
|
result = SearXNGWebSearchProvider().search("query", limit=5)
|
|
|
|
positions = [r["position"] for r in result["data"]["web"]]
|
|
assert positions == [1, 2, 3]
|
|
|
|
def test_empty_results(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
mock_resp = self._make_mock_response({"results": []})
|
|
|
|
with patch("httpx.get", return_value=mock_resp):
|
|
result = SearXNGWebSearchProvider().search("nothing", limit=5)
|
|
|
|
assert result["success"] is True
|
|
assert result["data"]["web"] == []
|
|
|
|
def test_missing_score_falls_back_to_zero(self, monkeypatch):
|
|
"""Results without a score field should sort to the bottom."""
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
data = {
|
|
"results": [
|
|
{"title": "No score", "url": "https://noscore.example.com", "content": ""},
|
|
{"title": "Has score", "url": "https://scored.example.com", "content": "", "score": 0.8},
|
|
]
|
|
}
|
|
mock_resp = self._make_mock_response(data)
|
|
|
|
with patch("httpx.get", return_value=mock_resp):
|
|
result = SearXNGWebSearchProvider().search("query", limit=5)
|
|
|
|
assert result["success"] is True
|
|
# Has score should sort first (0.8 > 0)
|
|
assert result["data"]["web"][0]["title"] == "Has score"
|
|
|
|
def test_http_error_returns_failure(self, monkeypatch):
|
|
import httpx
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 500
|
|
http_err = httpx.HTTPStatusError("500", request=MagicMock(), response=mock_resp)
|
|
|
|
with patch("httpx.get", side_effect=http_err):
|
|
result = SearXNGWebSearchProvider().search("query", limit=5)
|
|
|
|
assert result["success"] is False
|
|
assert "500" in result["error"]
|
|
|
|
def test_request_error_returns_failure(self, monkeypatch):
|
|
import httpx
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
|
|
with patch("httpx.get", side_effect=httpx.RequestError("connection refused")):
|
|
result = SearXNGWebSearchProvider().search("query", limit=5)
|
|
|
|
assert result["success"] is False
|
|
assert "localhost:8080" in result["error"] or "connection" in result["error"].lower()
|
|
|
|
def test_missing_url_returns_failure(self, monkeypatch):
|
|
monkeypatch.delenv("SEARXNG_URL", raising=False)
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
|
|
result = SearXNGWebSearchProvider().search("query", limit=5)
|
|
assert result["success"] is False
|
|
assert "SEARXNG_URL" in result["error"]
|
|
|
|
def test_trailing_slash_stripped_from_url(self, monkeypatch):
|
|
"""Base URL trailing slash should not produce double-slash in endpoint."""
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080/")
|
|
from plugins.web.searxng.provider import SearXNGWebSearchProvider
|
|
mock_resp = self._make_mock_response({"results": []})
|
|
|
|
calls = []
|
|
def capture_get(url, **kwargs):
|
|
calls.append(url)
|
|
return mock_resp
|
|
|
|
with patch("httpx.get", side_effect=capture_get):
|
|
SearXNGWebSearchProvider().search("query", limit=5)
|
|
|
|
assert calls[0] == "http://localhost:8080/search", f"Got: {calls[0]}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration: _is_backend_available recognizes "searxng"
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestIsBackendAvailable:
|
|
def test_searxng_available_when_url_set(self, monkeypatch):
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
from tools.web_tools import _is_backend_available
|
|
assert _is_backend_available("searxng") is True
|
|
|
|
def test_searxng_unavailable_when_url_missing(self, monkeypatch):
|
|
monkeypatch.delenv("SEARXNG_URL", raising=False)
|
|
from tools.web_tools import _is_backend_available
|
|
assert _is_backend_available("searxng") is False
|
|
|
|
def test_unknown_backend_still_false(self):
|
|
from tools.web_tools import _is_backend_available
|
|
assert _is_backend_available("unknownbackend") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration: _get_backend() accepts "searxng" as configured value
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetBackendSearXNG:
|
|
def test_configured_searxng_returns_searxng(self, monkeypatch):
|
|
from tools import web_tools
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "searxng"})
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
assert web_tools._get_backend() == "searxng"
|
|
|
|
def test_auto_detect_picks_searxng_when_only_url_set(self, monkeypatch):
|
|
"""When no backend is configured but SEARXNG_URL is set, auto-detect returns it."""
|
|
from tools import web_tools
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
|
|
monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
|
|
monkeypatch.delenv("FIRECRAWL_API_URL", raising=False)
|
|
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
|
|
monkeypatch.delenv("TAVILY_API_KEY", raising=False)
|
|
monkeypatch.delenv("EXA_API_KEY", raising=False)
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
# Suppress tool gateway
|
|
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
|
|
assert web_tools._get_backend() == "searxng"
|
|
|
|
def test_searxng_does_not_override_higher_priority_provider(self, monkeypatch):
|
|
"""Tavily (higher priority than searxng) should win in auto-detect."""
|
|
from tools import web_tools
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
|
|
monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
|
|
monkeypatch.delenv("FIRECRAWL_API_URL", raising=False)
|
|
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
|
|
monkeypatch.setenv("TAVILY_API_KEY", "tvly-key")
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
|
|
assert web_tools._get_backend() == "tavily"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration: check_web_api_key includes searxng
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCheckWebApiKey:
|
|
def test_searxng_satisfies_check_web_api_key(self, monkeypatch):
|
|
from tools import web_tools
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "searxng"})
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
assert web_tools.check_web_api_key() is True
|
|
|
|
def test_no_credentials_fails(self, monkeypatch):
|
|
from tools import web_tools
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
|
|
monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
|
|
monkeypatch.delenv("FIRECRAWL_API_URL", raising=False)
|
|
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
|
|
monkeypatch.delenv("TAVILY_API_KEY", raising=False)
|
|
monkeypatch.delenv("EXA_API_KEY", raising=False)
|
|
monkeypatch.delenv("SEARXNG_URL", raising=False)
|
|
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
|
|
monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
|
|
assert web_tools.check_web_api_key() is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# searxng-only: web_extract and web_crawl return clear errors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSearXNGOnlyExtractCrawlErrors:
|
|
"""When searxng is the active backend, extract/crawl must return clear errors."""
|
|
|
|
def test_web_crawl_searxng_returns_clear_error(self, monkeypatch):
|
|
import asyncio
|
|
from tools import web_tools
|
|
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "searxng"})
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
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)
|
|
|
|
import json
|
|
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() or "SearXNG" in result["error"]
|
|
|
|
def test_web_extract_searxng_returns_clear_error(self, monkeypatch):
|
|
import asyncio
|
|
from tools import web_tools
|
|
|
|
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "searxng"})
|
|
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
|
|
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
|
|
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False)
|
|
|
|
import json
|
|
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() or "SearXNG" in result["error"]
|