common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

test: use subprocesses for each test file (#29016)

Browse Source 
* 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.
This commit is contained in:
ethernet
2026-05-21 07:10:04 -04:00
committed by GitHub
parent 87d9239009
commit 48be2e0e4d
35 changed files with 1694 additions and 582 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
tests/agent/test_auxiliary_client.py
View File

@@ -40,6 +40,16 @@ def _clean_env(monkeypatch):
"ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN",
):
monkeypatch.delenv(key, raising=False)
# Module-level unhealthy cache (10-min TTL) leaks between tests;
# earlier tests that call _mark_provider_unhealthy() poison the
# cache for later ones, causing _resolve_auto to skip providers
# that the test patched to return valid clients.
import agent.auxiliary_client as _aux_mod
_aux_mod._aux_unhealthy_until.clear()
_aux_mod._aux_unhealthy_logged_at.clear()
yield
_aux_mod._aux_unhealthy_until.clear()
_aux_mod._aux_unhealthy_logged_at.clear()
@pytest.fixture
@@ -461,6 +471,17 @@ class TestExpiredCodexFallback:
import base64
import time as _time
# Belt-and-suspenders: _try_openrouter marks openrouter unhealthy
# when OPENROUTER_API_KEY is absent (which the preceding test in
# this class exercises). The file-level _clean_env autouse fixture
# clears the cache, but fixture ordering with the conftest
# _hermetic_environment autouse can leave a narrow window where
# the mark reappears. Explicitly clear here so this test is
# independent of run order.
import agent.auxiliary_client as _aux_mod
_aux_mod._aux_unhealthy_until.clear()
_aux_mod._aux_unhealthy_logged_at.clear()
header = base64.urlsafe_b64encode(b'{"alg":"RS256","typ":"JWT"}').rstrip(b"=").decode()
payload_data = json.dumps({"exp": int(_time.time()) - 3600}).encode()
payload = base64.urlsafe_b64encode(payload_data).rstrip(b"=").decode()
@@ -1047,6 +1068,20 @@ class TestGetProviderChain:
class TestTryPaymentFallback:
"""_try_payment_fallback skips the failed provider and tries alternatives."""
@pytest.fixture(autouse=True)
def _clear_unhealthy_cache(self):
"""Earlier tests in this file call _mark_provider_unhealthy() which
pollutes the module-level ``_aux_unhealthy_until`` dict (10-min TTL).
Without this cleanup the fallback chain skips providers we've patched
to return valid clients — the patched function is never called.
"""
from agent.auxiliary_client import _aux_unhealthy_until, _aux_unhealthy_logged_at
_aux_unhealthy_until.clear()
_aux_unhealthy_logged_at.clear()
yield
_aux_unhealthy_until.clear()
_aux_unhealthy_logged_at.clear()
def test_skips_failed_provider(self):
mock_client = MagicMock()
with patch("agent.auxiliary_client._try_openrouter", return_value=(None, None)), \

11
tests/agent/test_skill_commands.py
View File

@@ -556,10 +556,11 @@ Generate some audio.
raising=False,
)
with patch.dict(
os.environ, {"HERMES_SESSION_PLATFORM": "telegram"}, clear=False
):
with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
from gateway.session_context import clear_session_vars, set_session_vars
tokens = set_session_vars(platform="telegram")
try:
_make_skill(
tmp_path,
"test-skill",
@@ -571,6 +572,8 @@ Generate some audio.
)
scan_skill_commands()
msg = build_skill_invocation_message("/test-skill", "do stuff")
finally:
clear_session_vars(tokens)
assert msg is not None
assert "local cli" in msg.lower()

218
tests/conftest.py
View File

@@ -20,12 +20,9 @@ test runner at ``scripts/run_tests.sh``.
"""
import asyncio
import logging
import os
import re
import signal
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch
@@ -37,6 +34,22 @@ if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
# ── Per-file process isolation ──────────────────────────────────────────────
# Tests run via ``scripts/run_tests_parallel.py``, which spawns a fresh
# ``python -m pytest <file>`` subprocess per test file. Cross-file state
# leakage (module-level dicts, ContextVars, caches) is impossible: each
# file gets a clean Python interpreter. Intra-file ordering is the test
# author's responsibility — if test A in foo.py mutates state that test B
# in foo.py reads, that's a real bug to fix in the file (it would also
# bite anyone running ``pytest tests/foo.py`` directly).
#
# This replaces the historic _reset_module_state autouse fixture (manual
# state clearing) and the brief experiment with subprocess-per-test
# isolation (too slow at ~17k tests).
#
# See ``scripts/run_tests_parallel.py`` for the runner.
# ── Credential env-var filter ──────────────────────────────────────────────
#
# Any env var in the current process matching ONE of these patterns is
@@ -279,7 +292,7 @@ _HERMES_BEHAVIORAL_VARS = frozenset({
"WECOM_HOME_CHANNEL_NAME",
# Platform gating — set by load_gateway_config() as a side effect when
# a config.yaml is present, so individual test bodies that call the
# loader leak these values into later tests on the same xdist worker.
# loader leak these values into later tests in the same process.
# Force-clear on every test setup so the leak can't happen.
"SLACK_REQUIRE_MENTION",
"SLACK_STRICT_MENTION",
@@ -368,144 +381,21 @@ def _isolate_hermes_home(_hermetic_environment):
return None
# ── Module-level state reset ───────────────────────────────────────────────
# ── Module-level state reset — replaced by per-file process isolation ──────
#
# Python modules are singletons per process, and pytest-xdist workers are
# long-lived. Module-level dicts/sets (tool registries, approval state,
# interrupt flags) and ContextVars persist across tests in the same worker,
# causing tests that pass alone to fail when run with siblings.
# Each test FILE runs in a freshly-spawned ``python -m pytest <file>``
# subprocess via ``scripts/run_tests_parallel.py``, so module-level dicts /
# sets / ContextVars from tests in one file cannot leak into tests in
# another file. No manual per-module clearing needed.
#
# Each entry in this fixture clears state that belongs to a specific module.
# New state buckets go here too — this is the single gate that prevents
# "works alone, flakes in CI" bugs from state leakage.
# Within a single file, ordering is the author's responsibility. If your
# tests in the same file share mutable state, either reset it explicitly
# in a fixture or split them across files.
#
# The skill `test-suite-cascade-diagnosis` documents the concrete patterns
# this closes; the running example was `test_command_guards` failing 12/15
# CI runs because ``tools.approval._session_approved`` carried approvals
# from one test's session into another's.
@pytest.fixture(autouse=True)
def _reset_module_state():
"""Clear module-level mutable state and ContextVars between tests.
Keeps state from leaking across tests on the same xdist worker. Modules
that don't exist yet (test collection before production import) are
skipped silently — production import later creates fresh empty state.
"""
# --- logging — quiet/one-shot paths mutate process-global logger state ---
logging.disable(logging.NOTSET)
for _logger_name in ("tools", "run_agent", "trajectory_compressor", "cron", "hermes_cli"):
_logger = logging.getLogger(_logger_name)
_logger.disabled = False
_logger.setLevel(logging.NOTSET)
_logger.propagate = True
# --- tools.approval — the single biggest source of cross-test pollution ---
try:
from tools import approval as _approval_mod
_approval_mod._session_approved.clear()
_approval_mod._session_yolo.clear()
_approval_mod._permanent_approved.clear()
_approval_mod._pending.clear()
_approval_mod._gateway_queues.clear()
_approval_mod._gateway_notify_cbs.clear()
# ContextVar: reset to empty string so get_current_session_key()
# falls through to the env var / default path, matching a fresh
# process.
_approval_mod._approval_session_key.set("")
except Exception:
pass
# --- tools.interrupt — per-thread interrupt flag set ---
try:
from tools import interrupt as _interrupt_mod
with _interrupt_mod._lock:
_interrupt_mod._interrupted_threads.clear()
except Exception:
pass
# --- gateway.session_context — 9 ContextVars that represent
# the active gateway session. If set in one test and not reset,
# the next test's get_session_env() reads stale values.
try:
from gateway import session_context as _sc_mod
for _cv in (
_sc_mod._SESSION_PLATFORM,
_sc_mod._SESSION_CHAT_ID,
_sc_mod._SESSION_CHAT_NAME,
_sc_mod._SESSION_THREAD_ID,
_sc_mod._SESSION_USER_ID,
_sc_mod._SESSION_USER_NAME,
_sc_mod._SESSION_KEY,
_sc_mod._CRON_AUTO_DELIVER_PLATFORM,
_sc_mod._CRON_AUTO_DELIVER_CHAT_ID,
_sc_mod._CRON_AUTO_DELIVER_THREAD_ID,
):
_cv.set(_sc_mod._UNSET)
except Exception:
pass
# --- tools.env_passthrough — ContextVar<set[str]> with no default ---
# LookupError is normal if the test never set it. Setting it to an
# empty set unconditionally normalizes the starting state.
try:
from tools import env_passthrough as _envp_mod
_envp_mod._allowed_env_vars_var.set(set())
except Exception:
pass
# --- tools.terminal_tool — active environment/cwd cache ---
# File tools prefer a live terminal cwd when one is cached for the task.
# Clear terminal environments between tests so a prior terminal call can't
# override TERMINAL_CWD in path-resolution tests.
try:
from tools import terminal_tool as _term_mod
_envs_to_cleanup = []
with _term_mod._env_lock:
_envs_to_cleanup = list(_term_mod._active_environments.values())
_term_mod._active_environments.clear()
_term_mod._last_activity.clear()
_term_mod._creation_locks.clear()
for _env in _envs_to_cleanup:
try:
_env.cleanup()
except Exception:
pass
except Exception:
pass
# --- tools.credential_files — ContextVar<dict> ---
try:
from tools import credential_files as _credf_mod
_credf_mod._registered_files_var.set({})
except Exception:
pass
# --- agent.auxiliary_client — runtime main provider/model override and
# payment-error health cache. Both are process-global in production;
# reset them per test so one worker's fallback/402 test does not make
# later auxiliary-client tests skip otherwise-available providers.
try:
from agent import auxiliary_client as _aux_mod
_aux_mod.clear_runtime_main()
_aux_mod._reset_aux_unhealthy_cache()
except Exception:
pass
# --- tools.file_tools — per-task read history + file-ops cache ---
# _read_tracker accumulates per-task_id read history for loop detection,
# capped by _READ_HISTORY_CAP. If entries from a prior test persist, the
# cap is hit faster than expected and capacity-related tests flake.
try:
from tools import file_tools as _ft_mod
with _ft_mod._read_tracker_lock:
_ft_mod._read_tracker.clear()
with _ft_mod._file_ops_lock:
_ft_mod._file_ops_cache.clear()
except Exception:
pass
yield
# The skill ``test-suite-cascade-diagnosis`` documents the cascade patterns
# this replaces; the running example was ``test_command_guards`` failing
# 12/15 CI runs because ``tools.approval._session_approved`` carried
# approvals from one test's session into another's.
@pytest.fixture()
@@ -532,13 +422,12 @@ def mock_config():
}
# ── Global test timeout ─────────────────────────────────────────────────────
# Kill any individual test that takes longer than 30 seconds.
# Prevents hanging tests (subprocess spawns, blocking I/O) from stalling the
# entire test suite.
# ── Per-test timeout — handled by the isolation plugin ─────────────────────
#
# The subprocess-per-test plugin enforces the configured ``isolate_timeout``
# ini key by terminating the child if it overruns. The old SIGALRM-based
# fixture (POSIX-only, didn't work on Windows) is gone.
def _timeout_handler(signum, frame):
raise TimeoutError("Test exceeded 30 second timeout")
@pytest.fixture(autouse=True)
def _ensure_current_event_loop(request):
@@ -584,45 +473,6 @@ def _ensure_current_event_loop(request):
asyncio.set_event_loop(None)
@pytest.fixture(autouse=True)
def _enforce_test_timeout():
"""Kill any individual test that takes longer than 30 seconds.
SIGALRM is Unix-only; skip on Windows."""
if sys.platform == "win32":
yield
return
old = signal.signal(signal.SIGALRM, _timeout_handler)
signal.alarm(30)
yield
signal.alarm(0)
signal.signal(signal.SIGALRM, old)
@pytest.fixture(autouse=True)
def _reset_tool_registry_caches():
"""Clear tool-registry-level caches between tests.
The production registry caches ``check_fn()`` results for 30 s
(see tools/registry.py) and :func:`get_tool_definitions` memoizes
its result (see model_tools.py). Both are keyed on state that tests
routinely mutate (env vars, registry._generation, config.yaml mtime)
— but a stale result from test A can still be served to test B
because 30 s covers the entire suite, and xdist worker reuse means
one test's cache lands in another's process. Clearing before every
test keeps hermetic behavior.
"""
try:
from tools.registry import invalidate_check_fn_cache
invalidate_check_fn_cache()
except ImportError:
pass
try:
from model_tools import _clear_tool_defs_cache
_clear_tool_defs_cache()
except ImportError:
pass
# ── Live-system guard ──────────────────────────────────────────────────────
#
# Several test files exercise the gateway-restart / kill code paths

133
tests/gateway/conftest.py
View File

@@ -313,19 +313,30 @@ def _scan_for_plugin_adapter_antipattern(source: str) -> list[str]:
return offenses
def pytest_configure(config):
"""Reject plugin-adapter tests that use the sys.path anti-pattern.
def _fingerprint_gateway_tests() -> str:
"""Return a short fingerprint that changes when any gateway test file changes.
Runs once per pytest session on the controller, BEFORE any xdist
worker is spawned. If any file under ``tests/gateway/`` matches the
anti-pattern, we fail the whole session with a clear message —
before a polluted ``sys.path`` can cascade across workers.
Uses (mtime, size) pairs instead of content hashing — fast to compute
(stat-only, no reads) and sufficient for cache invalidation across
per-file subprocess runs.
"""
# Only run on the xdist controller (or in non-xdist runs). Skip on
# worker subprocesses so we don't scan the filesystem N times.
if hasattr(config, "workerinput"):
return
import hashlib
h = hashlib.sha256()
for path in sorted(_GATEWAY_DIR.rglob("test_*.py")):
try:
st = path.stat()
h.update(f"{path.name}:{st.st_mtime_ns}:{st.st_size}".encode())
except OSError:
h.update(f"{path.name}:missing".encode())
return h.hexdigest()[:16]
def _run_adapter_antipattern_scan() -> list[str]:
"""Scan gateway test files for the plugin-adapter anti-pattern.
Returns a list of violation strings (empty if clean).
"""
violations: list[str] = []
for path in _GATEWAY_DIR.rglob("test_*.py"):
if path.name in {"_plugin_adapter_loader.py", "conftest.py"}:
@@ -334,20 +345,108 @@ def pytest_configure(config):
source = path.read_text(encoding="utf-8")
except OSError:
continue
# Fast string pre-filter: skip files that can't possibly violate.
# A violating file MUST contain both (a) an adapter/plugins/platforms
# reference AND (b) either sys.path manipulation or a bare adapter import.
if "adapter" not in source and "plugins/platforms" not in source:
continue
if not (
"sys.path" in source
or "import adapter" in source
or "from adapter import" in source
):
continue
offenses = _scan_for_plugin_adapter_antipattern(source)
if offenses:
violations.append(
f" {path.relative_to(_GATEWAY_DIR.parent.parent)}:\n "
+ "\n ".join(offenses)
)
return violations
if violations:
raise pytest.UsageError(
"Plugin-adapter-import anti-pattern detected in gateway tests:\n"
+ "\n".join(violations)
+ "\n\n"
+ _GUARD_HINT
)
def pytest_configure(config):
"""Reject plugin-adapter tests that use the sys.path anti-pattern.
Runs once per pytest session on the controller, BEFORE any xdist
worker is spawned. If any file under ``tests/gateway/`` matches the
anti-pattern, we fail the whole session with a clear message —
before a polluted ``sys.path`` can cascade across workers.
**Performance**: in the per-file subprocess isolation model (no xdist),
every subprocess is a "controller" — so the naive scan would run 257
times, each costing ~1s of AST walking. We avoid this with two
strategies:
1. **Tight string pre-filter**: a file can only violate if it contains
*both* an adapter/plugins/platforms reference *and* a sys.path
manipulation or bare ``import adapter``. This drops ~95% of files
from needing AST parsing.
2. **File-locked cache**: the scan result is cached in
``.pytest-cache/gw-adapter-guard-<fingerprint>`` keyed on a
fingerprint of the gateway test file mtimes/sizes. Concurrent
subprocesses acquire a lock; only the first performs the scan;
the rest wait and read the cached result.
"""
# Only run on the xdist controller (or in non-xdist runs). Skip on
# worker subprocesses so we don't scan the filesystem N times.
if hasattr(config, "workerinput"):
return
fp = _fingerprint_gateway_tests()
cache_dir = Path.cwd() / ".pytest-cache"
cache_file = cache_dir / f"gw-adapter-guard-{fp}"
lock_file = cache_dir / f".gw-adapter-guard-{fp}.lock"
cache_dir.mkdir(parents=True, exist_ok=True)
# Evict stale cache entries from previous fingerprints (best-effort).
try:
for old in cache_dir.glob("gw-adapter-guard-*"):
if old.name != f"gw-adapter-guard-{fp}":
old.unlink(missing_ok=True)
for old in cache_dir.glob(".gw-adapter-guard-*.lock"):
if old.name != f".gw-adapter-guard-{fp}.lock":
old.unlink(missing_ok=True)
except OSError:
pass # Non-critical; old files are harmless.
# Use filelock to ensure only one process scans at a time.
# Concurrent subprocesses all hit pytest_configure simultaneously;
# without a lock they'd all find no cache and all run the scan.
try:
from filelock import FileLock
lock = FileLock(str(lock_file), timeout=120)
except ImportError:
# Fallback: no locking (still correct, just slower under contention).
import contextlib
class _NoLock:
def __enter__(self):
return self
def __exit__(self, *a):
pass
lock = _NoLock()
with lock:
if cache_file.exists():
cached = cache_file.read_text(encoding="utf-8")
if cached == "clean":
return
raise pytest.UsageError(cached)
# Slow path: this process is the first to acquire the lock.
violations = _run_adapter_antipattern_scan()
if violations:
msg = (
"Plugin-adapter-import anti-pattern detected in gateway tests:\n"
+ "\n".join(violations)
+ "\n\n"
+ _GUARD_HINT
)
cache_file.write_text(msg, encoding="utf-8")
raise pytest.UsageError(msg)
else:
cache_file.write_text("clean", encoding="utf-8")

27
tests/gateway/test_dm_topics.py
View File

@@ -22,19 +22,26 @@ from gateway.config import PlatformConfig
def _ensure_telegram_mock():
if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
return
telegram_mod = MagicMock()
telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
telegram_mod.constants.ChatType.GROUP = "group"
telegram_mod.constants.ChatType.SUPERGROUP = "supergroup"
telegram_mod.constants.ChatType.CHANNEL = "channel"
telegram_mod.constants.ChatType.PRIVATE = "private"
for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
sys.modules.setdefault(name, telegram_mod)
# Register telegram.constants as a separate module mock so that
# ``from telegram.constants import ChatType`` resolves to our mock
# with string-valued members (not auto-generated MagicMocks).
constants_mod = MagicMock()
constants_mod.ParseMode.MARKDOWN_V2 = "MarkdownV2"
constants_mod.ChatType.GROUP = "group"
constants_mod.ChatType.SUPERGROUP = "supergroup"
constants_mod.ChatType.CHANNEL = "channel"
constants_mod.ChatType.PRIVATE = "private"
sys.modules["telegram"] = telegram_mod
sys.modules["telegram.ext"] = telegram_mod.ext
sys.modules["telegram.constants"] = constants_mod
sys.modules["telegram.request"] = telegram_mod.request
# Force reimport so the adapter picks up the mock ChatType.
sys.modules.pop("gateway.platforms.telegram", None)
_ensure_telegram_mock()

17
tests/gateway/test_google_chat.py
View File

@@ -22,6 +22,11 @@ import pytest
from gateway.config import Platform, PlatformConfig, load_gateway_config
# Platform uses _missing_() for dynamic members, so "google_chat" is
# resolvable via Platform("google_chat") even without a static
# GOOGLE_CHAT attribute on the enum class.
_GC = Platform("google_chat")
# ---------------------------------------------------------------------------
# Mock the google-* packages if they are not installed
@@ -229,7 +234,7 @@ def _make_chat_envelope(text="hello", sender_email="u@example.com", sender_type=
class TestPlatformRegistration:
def test_enum_value(self):
assert Platform.GOOGLE_CHAT.value == "google_chat"
assert _GC.value == "google_chat"
def test_requirements_check_returns_true_when_available(self):
# The shim flag is True in this test module.
@@ -266,14 +271,14 @@ class TestEnvConfigLoading:
monkeypatch.setenv("GOOGLE_CHAT_PROJECT_ID", "p")
# No subscription.
cfg = load_gateway_config()
assert Platform.GOOGLE_CHAT not in cfg.platforms
assert _GC not in cfg.platforms
def test_missing_project_does_not_enable(self, monkeypatch):
self._clean_env(monkeypatch)
monkeypatch.setenv("GOOGLE_CHAT_SUBSCRIPTION_NAME",
"projects/p/subscriptions/s")
cfg = load_gateway_config()
assert Platform.GOOGLE_CHAT not in cfg.platforms
assert _GC not in cfg.platforms
@@ -2583,7 +2588,7 @@ class TestAuthorizationEmailMatch:
runner.pairing_store.is_approved = MagicMock(return_value=False)
source = SessionSource(
platform=Platform.GOOGLE_CHAT,
platform=_GC,
chat_id="spaces/S",
chat_type="dm",
user_id="alice@example.com", # post-swap: email is canonical
@@ -2604,7 +2609,7 @@ class TestAuthorizationEmailMatch:
runner.pairing_store.is_approved = MagicMock(return_value=False)
source = SessionSource(
platform=Platform.GOOGLE_CHAT,
platform=_GC,
chat_id="spaces/S",
chat_type="dm",
user_id="bob@example.com",
@@ -2630,7 +2635,7 @@ class TestAuthorizationEmailMatch:
runner.pairing_store.is_approved = MagicMock(return_value=False)
source = SessionSource(
platform=Platform.GOOGLE_CHAT,
platform=_GC,
chat_id="spaces/S",
chat_type="dm",
user_id="users/77777", # no email available — resource name wins

3
tests/hermes_cli/test_pty_bridge.py
View File

@@ -7,6 +7,7 @@ printf) to verify it behaves like a PTY you can read/write/resize/close.
from __future__ import annotations
import os
import shutil
import sys
import time
@@ -66,7 +67,7 @@ class TestPtyBridgeIO:
def test_write_sends_to_child_stdin(self):
# `cat` with no args echoes stdin back to stdout. We write a line,
# read it back, then signal EOF to let cat exit cleanly.
bridge = PtyBridge.spawn(["/bin/cat"])
bridge = PtyBridge.spawn([shutil.which("cat") or "cat"])
try:
bridge.write(b"hello-pty\n")
output = _read_until(bridge, b"hello-pty")

9
tests/plugins/test_achievements_plugin.py
View File

@@ -62,8 +62,9 @@ def plugin_api(tmp_path, monkeypatch):
class _FakeSessionDB:
"""Stand-in for hermes_state.SessionDB that records scan calls."""
def __init__(self, session_count: int):
def __init__(self, session_count: int, scan_delay: float = 0):
self.session_count = session_count
self.scan_delay = scan_delay
self.last_limit: Optional[int] = None
self.last_include_children: Optional[bool] = None
self.list_calls = 0
@@ -78,6 +79,8 @@ class _FakeSessionDB:
include_children: bool = False,
project_compression_tips: bool = True,
) -> List[Dict[str, Any]]:
if self.scan_delay:
time.sleep(self.scan_delay)
self.last_limit = limit
self.last_include_children = include_children
self.list_calls += 1
@@ -225,10 +228,8 @@ def test_evaluate_all_stale_cache_serves_stale_and_refreshes_in_background(plugi
the stale data immediately and kicks a background refresh. Users don't
stare at a loading spinner every time TTL expires.
"""
fake_db = _FakeSessionDB(session_count=10)
fake_db = _FakeSessionDB(session_count=10, scan_delay=2.0)
_install_fake_session_db(plugin_api, fake_db)
# Seed a stale snapshot on disk.
stale_generated_at = int(time.time()) - plugin_api.SNAPSHOT_TTL_SECONDS - 60
stale_payload = {
"achievements": [],

39
tests/plugins/web/test_web_search_provider_plugins.py
View File

@@ -2,8 +2,8 @@
Covers:
- All seven bundled plugins (brave-free, ddgs, searxng, exa, parallel,
tavily, firecrawl) instantiate and self-report the expected
- All eight bundled plugins (brave-free, ddgs, searxng, exa, parallel,
tavily, firecrawl, xai) instantiate and self-report the expected
capabilities + ABC-derived defaults.
- Each plugin's ``is_available()`` correctly reflects env-var presence.
- The web_search_registry resolves an active provider in the documented
@@ -47,6 +47,7 @@ def _clear_web_env(monkeypatch: pytest.MonkeyPatch) -> None:
"FIRECRAWL_GATEWAY_URL",
"TOOL_GATEWAY_DOMAIN",
"TOOL_GATEWAY_USER_TOKEN",
"XAI_API_KEY",
):
monkeypatch.delenv(k, raising=False)
@@ -70,7 +71,7 @@ def _isolate_env(monkeypatch: pytest.MonkeyPatch) -> None:
class TestBundledPluginsRegister:
"""All seven bundled web plugins discover and register correctly."""
"""All eight bundled web plugins discover and register correctly."""
def test_all_seven_plugins_present_in_registry(self) -> None:
_ensure_plugins_loaded()
@@ -85,6 +86,7 @@ class TestBundledPluginsRegister:
"parallel",
"searxng",
"tavily",
"xai",
]
@pytest.mark.parametrize(
@@ -100,6 +102,8 @@ class TestBundledPluginsRegister:
# disabled in the migration (fell through to a legacy inline
# path); the follow-up commit enabled it natively.
("firecrawl", True, True, True),
# xai: search-only via Grok's agentic web_search tool.
("xai", True, False, False),
],
)
def test_capability_flags_match_spec(
@@ -120,7 +124,7 @@ class TestBundledPluginsRegister:
@pytest.mark.parametrize(
"plugin_name",
["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl"],
["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl", "xai"],
)
def test_each_plugin_has_name_and_display_name(self, plugin_name: str) -> None:
_ensure_plugins_loaded()
@@ -133,7 +137,7 @@ class TestBundledPluginsRegister:
@pytest.mark.parametrize(
"plugin_name",
["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl"],
["brave-free", "ddgs", "searxng", "exa", "parallel", "tavily", "firecrawl", "xai"],
)
def test_each_plugin_has_setup_schema(self, plugin_name: str) -> None:
"""``get_setup_schema()`` returns a dict the picker can consume."""
@@ -239,6 +243,17 @@ class TestIsAvailable:
# Truthy or falsy, just must not raise.
_ = bool(p.is_available())
def test_xai_requires_api_key_or_oauth(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""xAI needs XAI_API_KEY or OAuth tokens in auth.json."""
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
p = get_provider("xai")
assert p is not None
assert p.is_available() is False # no XAI_API_KEY, no auth.json
monkeypatch.setenv("XAI_API_KEY", "real")
assert p.is_available() is True
# ---------------------------------------------------------------------------
# Registry resolution semantics (Option B — conservative smart fallback)
@@ -455,7 +470,7 @@ class TestErrorResponseShapes:
if result["results"]:
assert "error" in result["results"][0]
def test_firecrawl_crawl_returns_error_dict_when_unconfigured(self) -> None:
def test_firecrawl_crawl_returns_error_dict_when_unconfigured(self):
"""firecrawl crawl is async (wraps SDK in to_thread); error must be
surfaced via the per-page result shape, not raised."""
_ensure_plugins_loaded()
@@ -473,3 +488,15 @@ class TestErrorResponseShapes:
assert len(result["results"]) >= 1
assert "error" in result["results"][0]
assert result["results"][0]["url"] == "https://example.com"
def test_xai_search_returns_error_dict_when_unconfigured(self) -> None:
"""xAI returns a typed error dict (no XAI_API_KEY)."""
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
p = get_provider("xai")
assert p is not None
result = p.search("test", limit=5)
assert isinstance(result, dict)
assert result.get("success") is False
assert "error" in result

187
tests/test_run_tests_parallel.py Normal file
View File

@@ -0,0 +1,187 @@
"""Verify scripts/run_tests_parallel.py kills test-spawned grandchildren.
Setup
-----
A test in this file spawns a long-lived Python grandchild that writes
its PID + a nonce to a tempfile, then exits without cleaning up.
With the old ``subprocess.run`` runner, that grandchild would orphan
and outlive the test (and the whole runner). With the current Popen +
``start_new_session`` + ``_kill_tree`` runner, the grandchild gets
SIGKILL'd via process-group kill when its file's pytest exits.
The leaker test always passes — its only job is to spawn a grandchild
and walk away. The verifier runs the runner over the leaker file in a
subprocess, then waits for the grandchild PID to disappear from the
kernel's process table.
POSIX-only: Windows has its own grandchild lifecycle (no shared session,
``taskkill /F /T`` semantics). Marked accordingly.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import textwrap
import time
from pathlib import Path
import pytest
# Both tests share the same handoff file: the leaker writes here, the
# verifier reads here. We park it in $TMPDIR with a unique-per-run name
# so concurrent invocations of the suite don't clobber each other.
_HANDOFF_DIR = Path(os.environ.get("TMPDIR", "/tmp")) / "hermes-isolation-probe"
_HANDOFF_DIR.mkdir(exist_ok=True)
def _handoff_path_for(nonce: str) -> Path:
return _HANDOFF_DIR / f"grandchild-{nonce}.json"
def _pid_alive(pid: int) -> bool:
"""POSIX: send signal 0 to probe whether ``pid`` is still alive.
``os.kill(pid, 0)`` raises ``ProcessLookupError`` if the process is
gone, ``PermissionError`` if it exists but we can't signal it
(someone else's pid). We treat PermissionError as "alive" because
the process exists and that's all we need to know.
"""
if sys.platform == "win32": # pragma: no cover — POSIX-only test
# On Windows we'd use OpenProcess + GetExitCodeProcess; this
# test is skipped on Windows so the path is unreachable.
raise RuntimeError("_pid_alive POSIX-only")
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only probe")
@pytest.mark.live_system_guard_bypass
def test_grandchild_leak_is_killed_by_runner(tmp_path: Path) -> None:
"""Run the parallel runner over a probe file and verify cleanup.
1. Materialize a probe file that spawns a long-lived grandchild and
writes its PID to disk before exiting.
2. Invoke ``scripts/run_tests_parallel.py`` against the probe file.
3. Wait for the grandchild PID to vanish (poll for ~5s).
4. Assert the runner exited cleanly AND the grandchild is dead.
"""
repo_root = Path(__file__).resolve().parent.parent
runner = repo_root / "scripts" / "run_tests_parallel.py"
assert runner.exists(), f"runner missing at {runner}"
# Probe lives in a temp dir, NOT under tests/, so the regular suite
# never picks it up — only our explicit invocation does.
probe_dir = tmp_path / "probe"
probe_dir.mkdir()
probe = probe_dir / "test_probe_leaker.py"
nonce = f"{os.getpid()}-{int(time.time() * 1000)}"
handoff = _handoff_path_for(nonce)
if handoff.exists():
handoff.unlink()
probe_src = textwrap.dedent(f"""
import json, os, subprocess, sys, time
from pathlib import Path
HANDOFF = Path({str(handoff)!r})
def test_spawns_grandchild_and_walks_away():
# Long-lived grandchild: detached, ignores SIGTERM (we want
# SIGKILL or process-group kill to be the only thing that
# works, simulating a misbehaving server).
child = subprocess.Popen(
[
sys.executable, "-c",
"import os, signal, sys, time; "
"signal.signal(signal.SIGTERM, signal.SIG_IGN); "
"sys.stdout.write(f'gc-pgid={{os.getpgid(0)}} gc-pid={{os.getpid()}}\\\\n'); "
"sys.stdout.flush(); "
"time.sleep(600)",
],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
# IMPORTANT: do NOT pass start_new_session here. We want
# the grandchild to inherit the pytest subprocess's
# process group, so when the runner kills the group the
# grandchild dies too.
)
# Read the first line so we can record gc's pgid in the
# handoff, then walk away — don't close the pipe (would
# signal EOF and let the child see SIGPIPE on next write).
first_line = child.stdout.readline().decode().strip()
HANDOFF.write_text(json.dumps({{
"pid": child.pid,
"diag": first_line,
"test_pid": os.getpid(),
"test_pgid": os.getpgid(0),
}}))
assert child.pid > 0
""").strip()
probe.write_text(probe_src + "\n")
# Run the parallel runner against just the probe file. The runner
# discovers under ``tests/`` by default, so we override via --paths.
proc = subprocess.run(
[
sys.executable,
str(runner),
"--paths",
str(probe_dir),
"-j",
"1",
# Tight per-file timeout: the probe finishes in <1s, no
# need for 10min.
"--file-timeout",
"30",
],
cwd=repo_root,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
timeout=60,
)
assert handoff.exists(), (
f"probe never wrote handoff file; runner output:\n{proc.stdout}"
)
handoff_data = json.loads(handoff.read_text())
grandchild_pid = handoff_data["pid"]
diag = handoff_data.get("diag", "(no diag)")
test_pid = handoff_data.get("test_pid")
test_pgid = handoff_data.get("test_pgid")
handoff.unlink()
# The runner must have exited cleanly (probe test passes).
assert proc.returncode == 0, (
f"runner exited {proc.returncode}; output:\n{proc.stdout}"
)
# The grandchild must be gone. Poll for a bit because process-group
# SIGKILL + reaping isn't synchronous; on a loaded box it can take
# a beat.
deadline = time.monotonic() + 5.0
while time.monotonic() < deadline:
if not _pid_alive(grandchild_pid):
break
time.sleep(0.05)
else:
# Test cleanup: kill the leaked grandchild ourselves so a
# FAILED assertion doesn't leave a sleep(600) running.
try:
os.kill(grandchild_pid, 9)
except ProcessLookupError:
pass
pytest.fail(
f"grandchild PID {grandchild_pid} survived runner exit; "
f"diag={diag!r} test_pid={test_pid} test_pgid={test_pgid}; "
f"runner output:\n{proc.stdout}"
)

50
tests/tools/conftest.py Normal file
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()

16
tests/tools/test_approval_plugin_hooks.py
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)

5
tests/tools/test_browser_supervisor.py
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:

8
tests/tools/test_discord_tool.py
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(

6
tests/tools/test_homeassistant_tool.py
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

10
tests/tools/test_kanban_tools.py
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",

6
tests/tools/test_send_message_tool.py
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():

15
tests/tools/test_terminal_tool_requirements.py
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(

4
tests/tools/test_video_generation_tool_surface_matrix.py
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))

11
tests/tools/test_web_providers.py
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",

16
tests/tools/test_web_providers_brave_free.py
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(

16
tests/tools/test_web_providers_ddgs.py
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(

14
tests/tools/test_web_providers_searxng.py
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

29
tests/tools/test_web_tools_tavily.py
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 = {

351
tests/tools/test_website_policy.py
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"

20
tests/tools/test_write_deny.py
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
@@ -97,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: