seb/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:NousResearch/hermes-agent into bb/gui

Browse Source
This commit is contained in:
Brooklyn Nicholson
2026-05-21 13:53:53 -05:00
parent fa48c2501f ba9964ff0d
commit 085c33ed70
105 changed files with 5022 additions and 1714 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/conftest.py
+50
View File
@@ -0,0 +1,50 @@
"""Shared fixtures for tests/tools/ web-provider tests.
Per-file subprocess isolation means each test file gets a fresh interpreter,
so module-level state (like the web-search-provider registry) is empty when
a file starts. The ``web_registry_populated`` fixture registers all bundled
providers before each test and resets the registry afterwards — tests that
depend on the registry being populated should use it explicitly or via
``@pytest.mark.usefixtures("web_registry_populated")``.
"""
import pytest
def register_all_web_providers():
"""Register all bundled web-search providers into the global registry.
This is the single source of truth for the provider list used by
test classes that need the registry populated for dispatch checks.
"""
from agent.web_search_registry import register_provider, _reset_for_tests
from plugins.web.brave_free.provider import BraveFreeWebSearchProvider
from plugins.web.ddgs.provider import DDGSWebSearchProvider
from plugins.web.exa.provider import ExaWebSearchProvider
from plugins.web.firecrawl.provider import FirecrawlWebSearchProvider
from plugins.web.parallel.provider import ParallelWebSearchProvider
from plugins.web.searxng.provider import SearXNGWebSearchProvider
from plugins.web.tavily.provider import TavilyWebSearchProvider
from plugins.web.xai.provider import XAIWebSearchProvider
_reset_for_tests()
for cls in (
BraveFreeWebSearchProvider,
DDGSWebSearchProvider,
ExaWebSearchProvider,
FirecrawlWebSearchProvider,
ParallelWebSearchProvider,
SearXNGWebSearchProvider,
TavilyWebSearchProvider,
XAIWebSearchProvider,
):
register_provider(cls())
@pytest.fixture
def web_registry_populated():
"""Populate the web-search-provider registry for one test, then reset."""
register_all_web_providers()
yield
from agent.web_search_registry import _reset_for_tests
_reset_for_tests()
tests/tools/test_approval_plugin_hooks.py
+13 -3
View File
@@ -22,18 +22,28 @@ from tools.approval import (
@pytest.fixture
def isolated_session(monkeypatch):
"""Give each test a fresh session_key and clean approval-state."""
def isolated_session(monkeypatch, tmp_path):
"""Give each test a fresh session_key, clean approval-state, and isolated
HERMES_HOME so the real user's command_allowlist doesn't leak in."""
import tools.approval as _am
session_key = "test:session:approval_hooks"
token = set_current_session_key(session_key)
monkeypatch.setenv("HERMES_SESSION_KEY", session_key)
# Make sure we don't skip guards via yolo / approvals.mode=off
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
# Isolate from the real user's permanent allowlist + session state
_saved_permanent = _am._permanent_approved.copy()
_saved_session = {k: v.copy() for k, v in _am._session_approved.items()}
_am._permanent_approved.clear()
_am._session_approved.clear()
try:
yield session_key
finally:
_am._permanent_approved.update(_saved_permanent)
_am._session_approved.update(_saved_session)
try:
approval_module._approval_session_key.reset(token)
_am._approval_session_key.reset(token)
except Exception:
pass
clear_session(session_key)
tests/tools/test_browser_supervisor.py
+4 -1
View File
@@ -41,7 +41,7 @@ def _find_chrome() -> str:
@pytest.fixture
def chrome_cdp(worker_id):
def chrome_cdp(request):
"""Start a headless Chrome with --remote-debugging-port, yield its WS URL.
Uses a unique port per xdist worker to avoid cross-worker collisions.
@@ -51,6 +51,9 @@ def chrome_cdp(worker_id):
import socket
# xdist worker_id is "master" in single-process mode or "gw0".."gwN" otherwise.
# Under subprocess-per-file isolation there's no xdist, so we fall back
# to "master" via the session-scoped fixture below.
worker_id = request.getfixturevalue("worker_id") if "worker_id" in request.fixturenames else "master"
if worker_id == "master":
port_offset = 0
else:
tests/tools/test_discord_tool.py
+8
View File
@@ -1089,9 +1089,17 @@ class Test403Enrichment:
class TestModelToolsIntegration:
def setup_method(self):
_reset_capability_cache()
from model_tools import _clear_tool_defs_cache
from tools.registry import invalidate_check_fn_cache
_clear_tool_defs_cache()
invalidate_check_fn_cache()
def teardown_method(self):
_reset_capability_cache()
from model_tools import _clear_tool_defs_cache
from tools.registry import invalidate_check_fn_cache
_clear_tool_defs_cache()
invalidate_check_fn_cache()
@patch("tools.discord_tool._discord_request")
def test_discord_admin_schema_rebuilt_by_get_tool_definitions(
tests/tools/test_homeassistant_tool.py
+4 -2
View File
@@ -501,16 +501,18 @@ class TestRegistration:
def test_check_fn_gates_availability(self, monkeypatch):
"""Registry should exclude HA tools when HASS_TOKEN is not set."""
from tools.registry import registry
from tools.registry import invalidate_check_fn_cache, registry
monkeypatch.delenv("HASS_TOKEN", raising=False)
invalidate_check_fn_cache()
defs = registry.get_definitions({"ha_list_entities", "ha_get_state", "ha_call_service"})
assert len(defs) == 0
def test_check_fn_includes_when_token_set(self, monkeypatch):
"""Registry should include HA tools when HASS_TOKEN is set."""
from tools.registry import registry
from tools.registry import invalidate_check_fn_cache, registry
monkeypatch.setenv("HASS_TOKEN", "test-token")
invalidate_check_fn_cache()
defs = registry.get_definitions({"ha_list_entities", "ha_get_state", "ha_call_service"})
assert len(defs) == 3
tests/tools/test_kanban_tools.py
+10
View File
@@ -1093,6 +1093,11 @@ def test_kanban_guidance_not_in_normal_prompt(monkeypatch, tmp_path):
from pathlib import Path as _P
monkeypatch.setattr(_P, "home", lambda: tmp_path)
from tools.registry import invalidate_check_fn_cache
from model_tools import _clear_tool_defs_cache
invalidate_check_fn_cache()
_clear_tool_defs_cache()
from run_agent import AIAgent
a = AIAgent(
api_key="test",
@@ -1116,6 +1121,11 @@ def test_kanban_guidance_in_worker_prompt(monkeypatch, tmp_path):
from pathlib import Path as _P
monkeypatch.setattr(_P, "home", lambda: tmp_path)
from tools.registry import invalidate_check_fn_cache
from model_tools import _clear_tool_defs_cache
invalidate_check_fn_cache()
_clear_tool_defs_cache()
from run_agent import AIAgent
a = AIAgent(
api_key="test",
tests/tools/test_send_message_tool.py
+6
View File
@@ -10,6 +10,12 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# python-telegram-bot is an optional dep — skip the entire module when
# it isn't installed (e.g. CI bare env). Tests that patch telegram.Bot
# or call _send_telegram need it; tests for other platforms don't but
# keeping the whole file consistent is simpler.
_HAS_TELEGRAM = pytest.importorskip("telegram", reason="python-telegram-bot not installed") is not None
@pytest.fixture(autouse=True)
def _reset_signal_scheduler():
tests/tools/test_skills_hub.py
+27 -8
View File
@@ -1279,10 +1279,11 @@ class TestUnifiedSearchDedup:
return src
def test_dedup_keeps_first_seen(self):
# Same identifier from two sources — only the first (community) is kept when equal trust.
s1 = SkillMeta(name="skill", description="from A", source="a",
identifier="a/skill", trust_level="community")
identifier="shared/skill", trust_level="community")
s2 = SkillMeta(name="skill", description="from B", source="b",
identifier="b/skill", trust_level="community")
identifier="shared/skill", trust_level="community")
src_a = self._make_source("a", [s1])
src_b = self._make_source("b", [s2])
results = unified_search("skill", [src_a, src_b])
@@ -1290,10 +1291,11 @@ class TestUnifiedSearchDedup:
assert results[0].description == "from A"
def test_dedup_prefers_trusted_over_community(self):
# Same identifier — trusted wins over community.
community = SkillMeta(name="skill", description="community", source="a",
identifier="a/skill", trust_level="community")
identifier="shared/skill", trust_level="community")
trusted = SkillMeta(name="skill", description="trusted", source="b",
identifier="b/skill", trust_level="trusted")
identifier="shared/skill", trust_level="trusted")
src_a = self._make_source("a", [community])
src_b = self._make_source("b", [trusted])
results = unified_search("skill", [src_a, src_b])
@@ -1303,9 +1305,9 @@ class TestUnifiedSearchDedup:
def test_dedup_prefers_builtin_over_trusted(self):
"""Regression: builtin must not be overwritten by trusted."""
builtin = SkillMeta(name="skill", description="builtin", source="a",
identifier="a/skill", trust_level="builtin")
identifier="shared/skill", trust_level="builtin")
trusted = SkillMeta(name="skill", description="trusted", source="b",
identifier="b/skill", trust_level="trusted")
identifier="shared/skill", trust_level="trusted")
src_a = self._make_source("a", [builtin])
src_b = self._make_source("b", [trusted])
results = unified_search("skill", [src_a, src_b])
@@ -1314,14 +1316,31 @@ class TestUnifiedSearchDedup:
def test_dedup_trusted_not_overwritten_by_community(self):
trusted = SkillMeta(name="skill", description="trusted", source="a",
identifier="a/skill", trust_level="trusted")
identifier="shared/skill", trust_level="trusted")
community = SkillMeta(name="skill", description="community", source="b",
identifier="b/skill", trust_level="community")
identifier="shared/skill", trust_level="community")
src_a = self._make_source("a", [trusted])
src_b = self._make_source("b", [community])
results = unified_search("skill", [src_a, src_b])
assert results[0].trust_level == "trusted"
def test_browse_sh_same_name_different_site_not_deduped(self):
# Browse.sh skills from different hostnames share task names (e.g. "search-listings")
# but have unique identifiers. They must NOT be collapsed into one result.
airbnb = SkillMeta(
name="search-listings", description="Airbnb search", source="browse-sh",
identifier="browse-sh/airbnb.com/search-listings-ddgioa", trust_level="community",
)
booking = SkillMeta(
name="search-listings", description="Booking.com search", source="browse-sh",
identifier="browse-sh/booking.com/search-listings-xyzab", trust_level="community",
)
src = self._make_source("browse-sh", [airbnb, booking])
results = unified_search("search-listings", [src])
assert len(results) == 2, (
"browse-sh skills with the same name but different sites must not be deduplicated"
)
def test_source_filter(self):
s1 = SkillMeta(name="s1", description="d", source="a",
identifier="x", trust_level="community")
tests/tools/test_terminal_tool_requirements.py
+15
View File
@@ -2,11 +2,26 @@
import importlib
import pytest
from model_tools import get_tool_definitions
terminal_tool_module = importlib.import_module("tools.terminal_tool")
@pytest.fixture(autouse=True)
def _clear_caches():
"""Invalidate check_fn and tool-definitions caches before each test
so that monkeypatched env vars / config take effect."""
from tools.registry import invalidate_check_fn_cache
from model_tools import _clear_tool_defs_cache
invalidate_check_fn_cache()
_clear_tool_defs_cache()
yield
invalidate_check_fn_cache()
_clear_tool_defs_cache()
class TestTerminalRequirements:
def test_local_backend_requirements(self, monkeypatch):
monkeypatch.setattr(
tests/tools/test_video_generation_tool_surface_matrix.py
+3 -1
View File
@@ -95,7 +95,9 @@ def _invoke_tool(home, cfg: dict, args: dict) -> dict:
if hasattr(cfg_mod, "_invalidate_load_config_cache"):
cfg_mod._invalidate_load_config_cache()
from tools.registry import registry
from tools.registry import discover_builtin_tools, registry
if "video_generate" not in registry._tools:
discover_builtin_tools()
handler = registry._tools["video_generate"].handler
return json.loads(handler(args))
tests/tools/test_web_providers.py
+11
View File
@@ -13,6 +13,8 @@ from typing import Any, Dict, List
import pytest
from tests.tools.conftest import register_all_web_providers
# ---------------------------------------------------------------------------
# ABC enforcement
@@ -276,6 +278,15 @@ class TestUnconfiguredErrorEnvelopeParity:
``result.get("error")`` detect the failure cleanly.
"""
_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 _clear_web_creds(self, monkeypatch):
for k in (
"BRAVE_SEARCH_API_KEY",
tests/tools/test_web_providers_brave_free.py
+16
View File
@@ -15,6 +15,10 @@ from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
import pytest
from tests.tools.conftest import register_all_web_providers
# ---------------------------------------------------------------------------
# BraveFreeWebSearchProvider unit tests
@@ -239,6 +243,15 @@ class TestBraveFreeBackendWiring:
class TestBraveFreeSearchOnlyErrors:
_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_extract_returns_search_only_error(self, monkeypatch):
import asyncio
from tools import web_tools
@@ -246,6 +259,7 @@ class TestBraveFreeSearchOnlyErrors:
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, "is_safe_url", lambda url: True)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False)
result_str = asyncio.get_event_loop().run_until_complete(
@@ -264,6 +278,8 @@ class TestBraveFreeSearchOnlyErrors:
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(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)
result_str = asyncio.get_event_loop().run_until_complete(
tests/tools/test_web_providers_ddgs.py
+16
View File
@@ -14,6 +14,10 @@ import sys
import types
from unittest.mock import MagicMock
import pytest
from tests.tools.conftest import register_all_web_providers
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.
@@ -210,6 +214,15 @@ class TestDDGSBackendWiring:
class TestDDGSSearchOnlyErrors:
_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_extract_returns_search_only_error(self, monkeypatch):
import asyncio
from tools import web_tools
@@ -217,6 +230,7 @@ class TestDDGSSearchOnlyErrors:
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, "is_safe_url", lambda url: True)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False)
result_str = asyncio.get_event_loop().run_until_complete(
@@ -235,6 +249,8 @@ class TestDDGSSearchOnlyErrors:
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(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)
result_str = asyncio.get_event_loop().run_until_complete(
tests/tools/test_web_providers_searxng.py
+14
View File
@@ -17,6 +17,8 @@ from unittest.mock import MagicMock, patch
import pytest
from tests.tools.conftest import register_all_web_providers
# ---------------------------------------------------------------------------
# SearXNGWebSearchProvider unit tests
@@ -301,6 +303,15 @@ class TestCheckWebApiKey:
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
@@ -309,6 +320,8 @@ class TestSearXNGOnlyExtractCrawlErrors:
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
@@ -326,6 +339,7 @@ class TestSearXNGOnlyExtractCrawlErrors:
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
tests/tools/test_web_tools_tavily.py
+29
View File
@@ -13,6 +13,8 @@ import asyncio
import pytest
from unittest.mock import patch, MagicMock
from tests.tools.conftest import register_all_web_providers
# ─── _tavily_request ─────────────────────────────────────────────────────────
@@ -163,6 +165,15 @@ class TestNormalizeTavilyDocuments:
class TestWebSearchTavily:
"""Test web_search_tool dispatch to Tavily."""
_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_search_dispatches_to_tavily(self):
mock_response = MagicMock()
mock_response.json.return_value = {
@@ -186,6 +197,15 @@ class TestWebSearchTavily:
class TestWebExtractTavily:
"""Test web_extract_tool dispatch to Tavily."""
_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_extract_dispatches_to_tavily(self):
mock_response = MagicMock()
mock_response.json.return_value = {
@@ -211,6 +231,15 @@ class TestWebExtractTavily:
class TestWebCrawlTavily:
"""Test web_crawl_tool dispatch to Tavily."""
_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_crawl_dispatches_to_tavily(self):
mock_response = MagicMock()
mock_response.json.return_value = {
tests/tools/test_website_policy.py
+184 -167
View File
@@ -4,6 +4,8 @@ from pathlib import Path
import pytest
import yaml
from tests.tools.conftest import register_all_web_providers
from tools.website_policy import WebsitePolicyError, check_website_access, load_website_blocklist
@@ -347,40 +349,191 @@ def test_browser_navigate_allows_when_shared_file_missing(monkeypatch, tmp_path)
assert result is None
@pytest.mark.asyncio
async def test_web_extract_short_circuits_blocked_url(monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
class TestWebToolPolicy:
"""Tests that exercise web_extract_tool / web_crawl_tool with website-policy gates.
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
# The per-URL website-policy gate moved into the firecrawl plugin's
# extract() during the web-provider migration. Patch it at the new
# location; the dispatcher-level gate (used by web_crawl_tool's
# pre-flight) still lives on tools.web_tools.
monkeypatch.setattr(
firecrawl_provider,
"check_website_access",
lambda url: {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
},
)
monkeypatch.setattr(
firecrawl_provider,
"_get_firecrawl_client",
lambda: pytest.fail("firecrawl should not run for blocked URL"),
)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
# Force the firecrawl plugin to be the active extract provider.
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
These tests need the bundled web providers to be registered in the
agent.web_search_registry so the tool dispatchers can find an active
provider. Without registration, the tools return an error dict that
lacks a ``results`` key, causing ``KeyError``.
"""
result = json.loads(await web_tools.web_extract_tool(["https://blocked.test"], use_llm_processing=False))
_register_providers = staticmethod(register_all_web_providers)
assert result["results"][0]["url"] == "https://blocked.test"
assert "Blocked by website policy" in result["results"][0]["error"]
@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()
@pytest.mark.asyncio
async def test_web_extract_short_circuits_blocked_url(self, monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
# The per-URL website-policy gate moved into the firecrawl plugin's
# extract() during the web-provider migration. Patch it at the new
# location; the dispatcher-level gate (used by web_crawl_tool's
# pre-flight) still lives on tools.web_tools.
monkeypatch.setattr(
firecrawl_provider,
"check_website_access",
lambda url: {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
},
)
monkeypatch.setattr(
firecrawl_provider,
"_get_firecrawl_client",
lambda: pytest.fail("firecrawl should not run for blocked URL"),
)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
# Force the firecrawl plugin to be the active extract provider.
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
result = json.loads(await web_tools.web_extract_tool(["https://blocked.test"], use_llm_processing=False))
assert result["results"][0]["url"] == "https://blocked.test"
assert "Blocked by website policy" in result["results"][0]["error"]
@pytest.mark.asyncio
async def test_web_extract_blocks_redirected_final_url(self, monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
def fake_check(url):
if url == "https://allowed.test":
return None
if url == "https://blocked.test/final":
return {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
}
pytest.fail(f"unexpected URL checked: {url}")
class FakeFirecrawlClient:
def scrape(self, url, formats):
return {
"markdown": "secret content",
"metadata": {
"title": "Redirected",
"sourceURL": "https://blocked.test/final",
},
}
# After the web-provider migration, the per-URL gate + firecrawl client
# live in the plugin. Patch both at the plugin location.
monkeypatch.setattr(firecrawl_provider, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "_get_firecrawl_client", lambda: FakeFirecrawlClient())
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
result = json.loads(await web_tools.web_extract_tool(["https://allowed.test"], use_llm_processing=False))
assert result["results"][0]["url"] == "https://blocked.test/final"
assert result["results"][0]["content"] == ""
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_crawl_short_circuits_blocked_url(self, monkeypatch):
from tools import web_tools
# web_crawl_tool checks for Firecrawl env before website policy
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
# The dispatcher-level (seed-URL) policy gate still lives on web_tools.
# No per-page gate runs in this test because the dispatcher returns
# immediately when the seed is blocked, before delegating to the plugin.
monkeypatch.setattr(
web_tools,
"check_website_access",
lambda url: {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
},
)
# If the dispatcher ever reaches the firecrawl plugin's crawl(), the test
# fails — pin the plugin module's client lookup so we'd notice.
from plugins.web.firecrawl import provider as firecrawl_provider
monkeypatch.setattr(
firecrawl_provider,
"_get_firecrawl_client",
lambda: pytest.fail("firecrawl plugin should not run for blocked crawl URL"),
)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
result = json.loads(await web_tools.web_crawl_tool("https://blocked.test", use_llm_processing=False))
assert result["results"][0]["url"] == "https://blocked.test"
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_crawl_blocks_redirected_final_url(self, monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
# Force the firecrawl plugin to be the active crawl provider.
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
def fake_check(url):
# Dispatcher seed-URL gate (web_tools.check_website_access call)
# and plugin per-page gate (firecrawl_provider.check_website_access
# call) both flow through this single fake_check.
if url == "https://allowed.test":
return None
if url == "https://blocked.test/final":
return {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
}
pytest.fail(f"unexpected URL checked: {url}")
class FakeCrawlClient:
def crawl(self, url, **kwargs):
return {
"data": [
{
"markdown": "secret crawl content",
"metadata": {
"title": "Redirected crawl page",
"sourceURL": "https://blocked.test/final",
},
}
]
}
# After PR #25182 follow-up: per-page policy gate lives in
# plugins.web.firecrawl.provider.crawl(). Patch the gate + client at
# the plugin location. The dispatcher-level (seed) gate also reads
# web_tools.check_website_access — patch both.
monkeypatch.setattr(web_tools, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "_get_firecrawl_client", lambda: FakeCrawlClient())
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
result = json.loads(await web_tools.web_crawl_tool("https://allowed.test", use_llm_processing=False))
assert result["results"][0]["content"] == ""
assert result["results"][0]["error"] == "Blocked by website policy"
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
def test_check_website_access_fails_open_on_malformed_config(tmp_path, monkeypatch):
@@ -400,139 +553,3 @@ def test_check_website_access_fails_open_on_malformed_config(tmp_path, monkeypat
# With default path, errors are caught and fail open
result = check_website_access("https://example.com")
assert result is None # allowed, not crashed
@pytest.mark.asyncio
async def test_web_extract_blocks_redirected_final_url(monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
def fake_check(url):
if url == "https://allowed.test":
return None
if url == "https://blocked.test/final":
return {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
}
pytest.fail(f"unexpected URL checked: {url}")
class FakeFirecrawlClient:
def scrape(self, url, formats):
return {
"markdown": "secret content",
"metadata": {
"title": "Redirected",
"sourceURL": "https://blocked.test/final",
},
}
# After the web-provider migration, the per-URL gate + firecrawl client
# live in the plugin. Patch both at the plugin location.
monkeypatch.setattr(firecrawl_provider, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "_get_firecrawl_client", lambda: FakeFirecrawlClient())
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
result = json.loads(await web_tools.web_extract_tool(["https://allowed.test"], use_llm_processing=False))
assert result["results"][0]["url"] == "https://blocked.test/final"
assert result["results"][0]["content"] == ""
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_crawl_short_circuits_blocked_url(monkeypatch):
from tools import web_tools
# web_crawl_tool checks for Firecrawl env before website policy
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
# The dispatcher-level (seed-URL) policy gate still lives on web_tools.
# No per-page gate runs in this test because the dispatcher returns
# immediately when the seed is blocked, before delegating to the plugin.
monkeypatch.setattr(
web_tools,
"check_website_access",
lambda url: {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
},
)
# If the dispatcher ever reaches the firecrawl plugin's crawl(), the test
# fails — pin the plugin module's client lookup so we'd notice.
from plugins.web.firecrawl import provider as firecrawl_provider
monkeypatch.setattr(
firecrawl_provider,
"_get_firecrawl_client",
lambda: pytest.fail("firecrawl plugin should not run for blocked crawl URL"),
)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
result = json.loads(await web_tools.web_crawl_tool("https://blocked.test", use_llm_processing=False))
assert result["results"][0]["url"] == "https://blocked.test"
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_crawl_blocks_redirected_final_url(monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
# Force the firecrawl plugin to be the active crawl provider.
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
def fake_check(url):
# Dispatcher seed-URL gate (web_tools.check_website_access call)
# and plugin per-page gate (firecrawl_provider.check_website_access
# call) both flow through this single fake_check.
if url == "https://allowed.test":
return None
if url == "https://blocked.test/final":
return {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
}
pytest.fail(f"unexpected URL checked: {url}")
class FakeCrawlClient:
def crawl(self, url, **kwargs):
return {
"data": [
{
"markdown": "secret crawl content",
"metadata": {
"title": "Redirected crawl page",
"sourceURL": "https://blocked.test/final",
},
}
]
}
# After PR #25182 follow-up: per-page policy gate lives in
# plugins.web.firecrawl.provider.crawl(). Patch the gate + client at
# the plugin location. The dispatcher-level (seed) gate also reads
# web_tools.check_website_access — patch both.
monkeypatch.setattr(web_tools, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "_get_firecrawl_client", lambda: FakeCrawlClient())
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
result = json.loads(await web_tools.web_crawl_tool("https://allowed.test", use_llm_processing=False))
assert result["results"][0]["content"] == ""
assert result["results"][0]["error"] == "Blocked by website policy"
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
tests/tools/test_write_deny.py
+43 -2
View File
@@ -1,8 +1,10 @@
"""Tests for _is_write_denied() — verifies deny list blocks sensitive paths on all platforms."""
import os
import pytest
from pathlib import Path
from unittest.mock import patch
from tools.file_operations import _is_write_denied
@@ -41,6 +43,31 @@ class TestWriteDenyExactPaths:
path = str(get_hermes_home() / ".env")
assert _is_write_denied(path) is True
def test_hermes_root_env_when_running_under_profile(self, tmp_path, monkeypatch):
"""Top-level ``<root>/.env`` stays write-denied even when running under
a profile (#15981).
Before the fix, ``build_write_denied_paths`` only added
``<active_profile>/.env`` to the deny list, so the global
``~/.hermes/.env`` (whose credentials are inherited by every profile)
could be silently overwritten by ``write_file`` while a profile was
active.
"""
root = tmp_path / "hermes_root"
profile_home = root / "profiles" / "coder"
profile_home.mkdir(parents=True)
global_env = root / ".env"
global_env.write_text("OPENAI_API_KEY=sk-real\n")
monkeypatch.setenv("HERMES_HOME", str(profile_home))
# Sanity check: HERMES_HOME does point to the profile dir, not the root.
from hermes_constants import get_hermes_home, get_default_hermes_root
assert get_hermes_home() == profile_home
assert get_default_hermes_root() == root
assert _is_write_denied(str(global_env)) is True
def test_shell_profiles(self):
home = str(Path.home())
for name in [".bashrc", ".zshrc", ".profile", ".bash_profile", ".zprofile"]:
@@ -72,8 +99,22 @@ class TestWriteDenyPrefixes:
def test_sudoers_d_prefix(self):
assert _is_write_denied("/etc/sudoers.d/custom") is True
def test_systemd_prefix(self):
assert _is_write_denied("/etc/systemd/system/evil.service") is True
def test_systemd_prefix(self, tmp_path):
# On NixOS, /etc/systemd is a symlink into /nix/store, so
# realpath() resolves it to a store path that doesn't match
# the /etc/systemd/ prefix. Build a real directory tree so
# realpath is a no-op and prefix matching works.
fake_etc = tmp_path / "etc" / "systemd" / "system"
fake_etc.mkdir(parents=True)
target = str(fake_etc / "evil.service")
# Patch the prefix builder to include our tmp_path prefix
import agent.file_safety as _fs
_orig = _fs.build_write_denied_prefixes
_extra_prefix = str(tmp_path / "etc" / "systemd") + os.sep
def _patched(home):
return _orig(home) + [_extra_prefix]
with patch.object(_fs, "build_write_denied_prefixes", _patched):
assert _is_write_denied(target) is True
class TestWriteAllowed:
tests/tools/test_x_search_tool.py
+287
View File
@@ -436,3 +436,290 @@ def test_x_search_registered_in_registry_with_check_fn():
assert entry.check_fn.__name__ == "check_x_search_requirements"
assert "XAI_API_KEY" in entry.requires_env
assert entry.emoji == "🐦"
# ---------------------------------------------------------------------------
# Date validation — fail fast before burning an API call on a window that
# cannot possibly return X posts. xAI itself happily 200s with a fluff
# answer when the range is malformed or pure-future, which is hard for
# callers to distinguish from a real result.
# ---------------------------------------------------------------------------
def _no_post_allowed(monkeypatch):
"""Guard: any test that should fail before HTTP can hit this fence."""
def _fail(*_, **__):
raise AssertionError("requests.post must not be called — validation should reject first")
monkeypatch.setattr("requests.post", _fail)
def test_x_search_rejects_malformed_from_date(monkeypatch):
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
_no_post_allowed(monkeypatch)
result = json.loads(x_search_tool(query="anything", from_date="not-a-date"))
assert "from_date must be YYYY-MM-DD" in result["error"]
def test_x_search_rejects_malformed_to_date(monkeypatch):
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
_no_post_allowed(monkeypatch)
result = json.loads(x_search_tool(query="anything", to_date="2026/05/01"))
assert "to_date must be YYYY-MM-DD" in result["error"]
def test_x_search_rejects_inverted_date_range(monkeypatch):
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
_no_post_allowed(monkeypatch)
result = json.loads(
x_search_tool(
query="anything",
from_date="2026-05-10",
to_date="2026-05-01",
)
)
assert "from_date (2026-05-10) must be on or before to_date (2026-05-01)" in result["error"]
def test_x_search_rejects_future_from_date(monkeypatch):
"""``from_date`` in the future can never match any post → reject."""
import datetime as _dt
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
_no_post_allowed(monkeypatch)
class _FrozenDateTime(_dt.datetime):
@classmethod
def now(cls, tz=None):
return _dt.datetime(2026, 5, 21, 12, 0, 0, tzinfo=tz or _dt.timezone.utc)
monkeypatch.setattr("tools.x_search_tool.datetime", _FrozenDateTime)
result = json.loads(x_search_tool(query="anything", from_date="2030-01-01"))
assert "from_date (2030-01-01) is in the future" in result["error"]
def test_x_search_allows_future_to_date(monkeypatch):
"""``to_date`` in the future is fine — caller may want posts as they arrive."""
import datetime as _dt
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
class _FrozenDateTime(_dt.datetime):
@classmethod
def now(cls, tz=None):
return _dt.datetime(2026, 5, 21, 12, 0, 0, tzinfo=tz or _dt.timezone.utc)
monkeypatch.setattr("tools.x_search_tool.datetime", _FrozenDateTime)
def _fake_post(url, headers=None, json=None, timeout=None):
return _FakeResponse(
{"output_text": "future to_date is allowed", "citations": []}
)
monkeypatch.setattr("requests.post", _fake_post)
result = json.loads(
x_search_tool(
query="anything",
from_date="2026-05-20",
to_date="2030-01-01",
)
)
assert result["success"] is True
assert result["answer"] == "future to_date is allowed"
def test_x_search_accepts_today_as_from_date(monkeypatch):
"""``from_date == today UTC`` is a valid edge case (today is past + present)."""
import datetime as _dt
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
class _FrozenDateTime(_dt.datetime):
@classmethod
def now(cls, tz=None):
return _dt.datetime(2026, 5, 21, 12, 0, 0, tzinfo=tz or _dt.timezone.utc)
monkeypatch.setattr("tools.x_search_tool.datetime", _FrozenDateTime)
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse({"output_text": "ok", "citations": []}),
)
result = json.loads(x_search_tool(query="anything", from_date="2026-05-21"))
assert result["success"] is True
# ---------------------------------------------------------------------------
# Degraded-result flag — distinguish citation-backed answers from
# unsourced fluff when narrowing filters returned nothing.
# ---------------------------------------------------------------------------
def test_x_search_marks_degraded_when_handle_filter_returns_no_citations(monkeypatch):
"""allowed_x_handles set + zero citations → degraded=True."""
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse(
{"output_text": "Generic encyclopedic answer with no citations.", "citations": []}
),
)
result = json.loads(
x_search_tool(query="what has @ghostuser posted", allowed_x_handles=["ghostuser"])
)
assert result["success"] is True
assert result["degraded"] is True
assert "allowed_x_handles" in result["degraded_reason"]
def test_x_search_marks_degraded_when_excluded_handles_and_no_citations(monkeypatch):
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse({"output_text": "fluff", "citations": []}),
)
result = json.loads(
x_search_tool(query="anything", excluded_x_handles=["someuser"])
)
assert result["degraded"] is True
assert "excluded_x_handles" in result["degraded_reason"]
def test_x_search_marks_degraded_when_date_range_and_no_citations(monkeypatch):
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse({"output_text": "fluff", "citations": []}),
)
result = json.loads(
x_search_tool(
query="anything",
from_date="2026-04-01",
to_date="2026-04-02",
)
)
assert result["degraded"] is True
assert "from_date" in result["degraded_reason"]
assert "to_date" in result["degraded_reason"]
def test_x_search_not_degraded_when_filter_returns_inline_citations(monkeypatch):
"""A real citation from the inline annotations clears the degraded flag."""
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse(
{
"output": [
{
"type": "message",
"content": [
{
"type": "output_text",
"text": "Real post from xai.",
"annotations": [
{
"type": "url_citation",
"url": "https://x.com/xai/status/1",
"title": "xAI post",
"start_index": 0,
"end_index": 4,
}
],
}
],
}
]
}
),
)
result = json.loads(
x_search_tool(query="latest xAI post", allowed_x_handles=["xai"])
)
assert result["success"] is True
assert result["degraded"] is False
assert result["degraded_reason"] is None
assert len(result["inline_citations"]) == 1
def test_x_search_not_degraded_when_filter_returns_top_level_citations(monkeypatch):
"""A real citation from xAI's top-level ``citations`` array also clears the flag."""
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse(
{
"output_text": "Found discussion.",
"citations": [{"url": "https://x.com/example/status/1", "title": "Example"}],
}
),
)
result = json.loads(
x_search_tool(query="anything", allowed_x_handles=["xai"])
)
assert result["degraded"] is False
assert result["degraded_reason"] is None
def test_x_search_not_degraded_when_no_filters_active(monkeypatch):
"""A broad query that returns no citations isn't necessarily degraded.
Without any narrowing filter, an empty-citations response is a generic
unsourced answer, not a "filter miss". The caller can already tell from
``inline_citations == []`` if they care.
"""
from tools.x_search_tool import x_search_tool
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
monkeypatch.setattr(
"requests.post",
lambda *a, **k: _FakeResponse({"output_text": "broad answer", "citations": []}),
)
result = json.loads(x_search_tool(query="anything"))
assert result["success"] is True
assert result["degraded"] is False
assert result["degraded_reason"] is None