mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-05-30 06:41:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 15
hermes-agent/tests/tools/test_web_providers_searxng.py
ethernet 48be2e0e4d
test: use subprocesses for each test file (#29016) 
* ci(tests): install ripgrep from prebuilt tarball instead of apt

apt-get update + install of ripgrep takes ~4 min on the GHA Ubuntu
runners (the apt-get update against archive.ubuntu.com is the slow
part; ripgrep itself is small). Switching to the upstream musl
binary tarball cuts the step to a few seconds.

- Pinned to ripgrep 15.1.0 with sha256 verification (same hash as
  published in the releases sha256 sidecar file).
- Drops the `rg` binary into /usr/local/bin so it is on PATH for
  every subsequent step without GITHUB_PATH manipulation.
- Applied to both the test and e2e jobs in tests.yml.

* fix(cli): compile syntax check to tempdir, not source __pycache__

`_validate_critical_files_syntax` runs `py_compile.compile()` on each
critical bootstrap file after a successful `git pull`. The default
`py_compile` writes the resulting `.pyc` next to the source under
`__pycache__/`, which causes two real problems:

1. Parallel test workers walking the same source tree (e.g. running
   the suite under per-file process isolation) can race against each
   other on the `__pycache__` write — manifests as flaky 'directory
   not empty' errors during teardown.
2. In production, the post-pull syntax check leaves a `.pyc` behind
   that the next interpreter run might pick up — fine when the
   interpreter version matches, sketchy if it doesn't.

Fix: write the compiled output to a `tempfile.TemporaryDirectory()`
that's discarded on function exit. We only care about the compile-or-not
signal, not the artifact.

* test(runner): per-file process isolation, drop manual state reset + xdist

Replace fragile manual _reset_module_state test fixtures with robust
per-file subprocess isolation. Each test file runs in a fresh
`python -m pytest <file>` subprocess via ThreadPoolExecutor. No xdist,
no custom pytest plugin, no shared worker state.

Key changes:
  * scripts/run_tests_parallel.py — new runner: discovers test files,
    runs N in parallel via ThreadPoolExecutor, captures stdout per file,
    treats exit code 5 (no tests collected) as pass, kills all children
    on exit. Change from cpu_count to cpu_count*2. The runner is
    I/O-bound (waiting on subprocess.communicate() from pytest children)
    The parent process does almost no CPU work, so 2x oversubscription
    keeps more pipes full. When a file fails, immediately show the last
    30 lines of pytest output (stack traces + FAILED summary) plus a
    ready-to-copy repro command:
      python -m pytest tests/agent/test_auxiliary_client.py
  * scripts/run_tests.sh — delegates to run_tests_parallel.py
  * .github/workflows/tests.yml — test step: python
scripts/run_tests_parallel.py
  * pyproject.toml — drop pytest-xdist, pytest-split; simplify addopts
  * tests/conftest.py — remove ~200 lines of manual state-reset fixtures
  * AGENTS.md — update Testing section for per-file design

* test(runner): speed gateway test antipattern scan up

* fix(test): web search provider plugin test missing xai

* fix(tests): make 14 test files pass under per-file subprocess isolation

Tests that relied on cross-file state pollution from xdist workers
fail when run in isolation (per-file subprocess model). Root causes
and fixes:

Tool registry not populated:
  - test_video_generation_tool_surface_matrix: add discover_builtin_tools()
  - test_web_providers_brave_free/ddgs/searxng/general: autouse fixtures
    registering all 8 bundled web providers, reset after each test
  - test_website_policy: same provider registration pattern
  - test_web_tools_tavily: same pattern across 3 dispatch test classes
  - Also add is_safe_url/check_website_access mocks where SSRF check
    blocks example.com (DNS resolution fails in isolated envs)

Stale check_fn cache:
  - test_kanban_tools: invalidate_check_fn_cache() + _clear_tool_defs_cache()
    in both kanban guidance tests (prior test cached False for kanban_show)
  - test_discord_tool: cache invalidation in setup/teardown
  - test_homeassistant_tool: invalidate_check_fn_cache() before registry queries

Module-level state pollution:
  - test_auxiliary_client: autouse fixture clearing _aux_unhealthy_until cache
  - test_skill_commands: set_session_vars() instead of patch.dict(os.environ)
    (ContextVar takes precedence over os.environ)
  - test_dm_topics: overwrite sys.modules + separate telegram.constants mock
    + force-reimport of gateway.platforms.telegram
  - test_terminal_tool_requirements: removed duplicate class declaration,
    autouse _clear_caches fixture

* change(tests): run_tests.sh explicitly includes env vars

instead of manually dropping some vars, now we just only include some

* fix(tests): 5 more isolation/NixOS fixes

- test_approval_plugin_hooks: isolate HERMES_HOME so real user's
  command_allowlist doesn't short-circuit the approval path
- test_google_chat: skipif when Platform.GOOGLE_CHAT not in enum
  (feature not merged on this branch)
- test_write_deny: test systemd prefix against tmp_path instead of
  /etc/systemd which resolves to /nix/store on NixOS
- test_pty_bridge: use shutil.which('cat') instead of /bin/cat
  (doesn't exist on NixOS)
- profiles.py: rmtree onexc handler chmod's parent dirs too, fixing
  profile deletion when copytree preserved read-only modes from
  nix store

* fix(tests): clear unhealthy cache in autouse fixture for auxiliary_client

* fix(tests): skip send_message when telegram not installed; handle missing worker_id in browser_supervisor

* fix: py3.11 rmtree onexc compat + belt-and-suspenders unhealthy cache clear for expired codex test

* fix: address PR #29016 review feedback

- Remove tracked .pytest-cache/ artifact and add to .gitignore
- Fix stale 'xdist worker' comment in conftest.py
- Deduplicate web provider registration into tests/tools/conftest.py
  shared helper (register_all_web_providers), replacing 8 copy-pasted
  blocks across 6 test files
- Update PR description: remove stale recovered-test-files claim,
  fix worker count to match code (cpu_count*2)

* fix: eliminate race in stale-cache achievements test

The background scan thread could complete and overwrite _SNAPSHOT_CACHE
before evaluate_all() returned the stale data — only 10 fake sessions
made the scan finish instantly. Added scan_delay param to _FakeSessionDB
and set it to 2s in the stale-cache test so the background thread can't
win the race.
2026-05-21 16:40:04 +05:30

351 lines
16 KiB
Python
Raw Blame History

"""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
from tests.tools.conftest import register_all_web_providers
# ---------------------------------------------------------------------------
# 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."""
_register_providers = staticmethod(register_all_web_providers)
@pytest.fixture(autouse=True)
def _populate_web_registry(self):
self._register_providers()
yield
from agent.web_search_registry import _reset_for_tests
_reset_for_tests()
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(web_tools, "is_safe_url", lambda url: True)
monkeypatch.setattr(web_tools, "check_website_access", lambda url: None)
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(web_tools, "is_safe_url", lambda url: True)
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"]
Reference in a new issue View git blame Copy permalink