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

feat: add managed tool gateway and Nous subscription support

Browse Source 
- add managed modal and gateway-backed tool integrations\n- improve CLI setup, auth, and configuration for subscriber flows\n- expand tests and docs for managed tool support
This commit is contained in:
Robin Fernandes
2026-03-26 15:27:27 -07:00
parent cbf195e806
commit 95dc9aaa75
44 changed files with 4567 additions and 423 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
tests/agent/test_prompt_builder.py
View File

@@ -5,6 +5,8 @@ import importlib
import logging
import sys
import pytest
from agent.prompt_builder import (
_scan_context_content,
_truncate_content,
@@ -15,6 +17,7 @@ from agent.prompt_builder import (
_find_git_root,
_strip_yaml_frontmatter,
build_skills_system_prompt,
build_nous_subscription_prompt,
build_context_files_prompt,
CONTEXT_FILE_MAX_CHARS,
DEFAULT_AGENT_IDENTITY,
@@ -22,6 +25,7 @@ from agent.prompt_builder import (
SESSION_SEARCH_GUIDANCE,
PLATFORM_HINTS,
)
from hermes_cli.nous_subscription import NousFeatureState, NousSubscriptionFeatures
# =========================================================================
@@ -395,6 +399,53 @@ class TestBuildSkillsSystemPrompt:
assert "backend-skill" in result
class TestBuildNousSubscriptionPrompt:
def test_includes_active_subscription_features(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_subscription_features",
lambda config=None: NousSubscriptionFeatures(
subscribed=True,
nous_auth_present=True,
provider_is_nous=True,
features={
"web": NousFeatureState("web", "Web tools", True, True, True, True, False, True, "firecrawl"),
"image_gen": NousFeatureState("image_gen", "Image generation", True, True, True, True, False, True, "Nous Subscription"),
"tts": NousFeatureState("tts", "OpenAI TTS", True, True, True, True, False, True, "OpenAI TTS"),
"browser": NousFeatureState("browser", "Browser automation", True, True, True, True, False, True, "Browserbase"),
"modal": NousFeatureState("modal", "Modal execution", False, True, False, False, False, True, "local"),
},
),
)
prompt = build_nous_subscription_prompt({"web_search", "browser_navigate"})
assert "Browserbase" in prompt
assert "Modal execution is optional" in prompt
assert "do not ask the user for Firecrawl, FAL, OpenAI TTS, or Browserbase API keys" in prompt
def test_non_subscriber_prompt_includes_relevant_upgrade_guidance(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_subscription_features",
lambda config=None: NousSubscriptionFeatures(
subscribed=False,
nous_auth_present=False,
provider_is_nous=False,
features={
"web": NousFeatureState("web", "Web tools", True, False, False, False, False, True, ""),
"image_gen": NousFeatureState("image_gen", "Image generation", True, False, False, False, False, True, ""),
"tts": NousFeatureState("tts", "OpenAI TTS", True, False, False, False, False, True, ""),
"browser": NousFeatureState("browser", "Browser automation", True, False, False, False, False, True, ""),
"modal": NousFeatureState("modal", "Modal execution", False, False, False, False, False, True, ""),
},
),
)
prompt = build_nous_subscription_prompt({"image_generate"})
assert "suggest Nous subscription as one option" in prompt
assert "Do not mention subscription unless" in prompt
# =========================================================================
# Context files prompt builder
# =========================================================================
@@ -562,8 +613,12 @@ class TestBuildContextFilesPrompt:
assert "Lowercase claude rules" in result
def test_claude_md_uppercase_takes_priority(self, tmp_path):
(tmp_path / "CLAUDE.md").write_text("From uppercase.")
(tmp_path / "claude.md").write_text("From lowercase.")
uppercase = tmp_path / "CLAUDE.md"
lowercase = tmp_path / "claude.md"
uppercase.write_text("From uppercase.")
lowercase.write_text("From lowercase.")
if uppercase.samefile(lowercase):
pytest.skip("filesystem is case-insensitive")
result = build_context_files_prompt(cwd=str(tmp_path))
assert "From uppercase" in result
assert "From lowercase" not in result

172
tests/hermes_cli/test_setup.py
View File

@@ -1,4 +1,6 @@
import json
import sys
import types
from hermes_cli.auth import _update_config_for_provider, get_active_provider
from hermes_cli.config import load_config, save_config
@@ -136,6 +138,8 @@ def test_codex_setup_uses_runtime_access_token_for_live_model_list(tmp_path, mon
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 2 # OpenAI Codex
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
return 0
tts_idx = _maybe_keep_current_tts(question, choices)
@@ -176,3 +180,171 @@ def test_codex_setup_uses_runtime_access_token_for_live_model_list(tmp_path, mon
assert reloaded["model"]["provider"] == "openai-codex"
assert reloaded["model"]["default"] == "gpt-5.2-codex"
assert reloaded["model"]["base_url"] == "https://chatgpt.com/backend-api/codex"
def test_nous_setup_sets_managed_openai_tts_when_unconfigured(tmp_path, monkeypatch, capsys):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 1
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
return len(choices) - 1
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
def _fake_login_nous(*args, **kwargs):
auth_path = tmp_path / "auth.json"
auth_path.write_text(json.dumps({"active_provider": "nous", "providers": {"nous": {"access_token": "nous-token"}}}))
_update_config_for_provider("nous", "https://inference.example.com/v1")
monkeypatch.setattr("hermes_cli.auth._login_nous", _fake_login_nous)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["gemini-3-flash"],
)
setup_model_provider(config)
out = capsys.readouterr().out
assert config["tts"]["provider"] == "openai"
assert "Nous subscription enables managed web tools" in out
assert "OpenAI TTS via your Nous subscription" in out
def test_nous_setup_preserves_existing_tts_provider(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
config["tts"] = {"provider": "elevenlabs"}
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 1
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
return len(choices) - 1
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr(
"hermes_cli.auth._login_nous",
lambda *args, **kwargs: (tmp_path / "auth.json").write_text(
json.dumps({"active_provider": "nous", "providers": {"nous": {"access_token": "nous-token"}}})
),
)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["gemini-3-flash"],
)
setup_model_provider(config)
assert config["tts"]["provider"] == "elevenlabs"
def test_modal_setup_can_use_nous_subscription_without_modal_creds(tmp_path, monkeypatch, capsys):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select terminal backend:":
return 2
if question == "Select how Modal execution should be billed:":
return 0
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_prompt(message, *args, **kwargs):
assert "Modal Token" not in message
raise AssertionError(f"Unexpected prompt call: {message}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", fake_prompt)
monkeypatch.setattr("hermes_cli.setup._prompt_container_resources", lambda config: None)
monkeypatch.setattr(
"hermes_cli.setup.get_nous_subscription_features",
lambda config: type("Features", (), {"nous_auth_present": True})(),
)
monkeypatch.setitem(
sys.modules,
"tools.managed_tool_gateway",
types.SimpleNamespace(
is_managed_tool_gateway_ready=lambda vendor: vendor == "modal",
resolve_managed_tool_gateway=lambda vendor: None,
),
)
from hermes_cli.setup import setup_terminal_backend
setup_terminal_backend(config)
out = capsys.readouterr().out
assert config["terminal"]["backend"] == "modal"
assert config["terminal"]["modal_mode"] == "managed"
assert "bill to your subscription" in out
def test_modal_setup_persists_direct_mode_when_user_chooses_their_own_account(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("MODAL_TOKEN_ID", raising=False)
monkeypatch.delenv("MODAL_TOKEN_SECRET", raising=False)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select terminal backend:":
return 2
if question == "Select how Modal execution should be billed:":
return 1
raise AssertionError(f"Unexpected prompt_choice call: {question}")
prompt_values = iter(["token-id", "token-secret", ""])
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: next(prompt_values))
monkeypatch.setattr("hermes_cli.setup._prompt_container_resources", lambda config: None)
monkeypatch.setattr(
"hermes_cli.setup.get_nous_subscription_features",
lambda config: type("Features", (), {"nous_auth_present": True})(),
)
monkeypatch.setitem(
sys.modules,
"tools.managed_tool_gateway",
types.SimpleNamespace(
is_managed_tool_gateway_ready=lambda vendor: vendor == "modal",
resolve_managed_tool_gateway=lambda vendor: None,
),
)
monkeypatch.setitem(sys.modules, "swe_rex", object())
from hermes_cli.setup import setup_terminal_backend
setup_terminal_backend(config)
assert config["terminal"]["backend"] == "modal"
assert config["terminal"]["modal_mode"] == "direct"

47
tests/hermes_cli/test_setup_noninteractive.py
View File

@@ -1,7 +1,7 @@
"""Tests for non-interactive setup and first-run headless behavior."""
from argparse import Namespace
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
@@ -92,3 +92,48 @@ class TestNonInteractiveSetup:
mock_setup.assert_not_called()
out = capsys.readouterr().out
assert "hermes config set model.provider custom" in out
def test_returning_user_terminal_menu_choice_dispatches_terminal_section(self, tmp_path):
"""Returning-user menu should map Terminal Backend to the terminal setup, not TTS."""
from hermes_cli import setup as setup_mod
args = _make_setup_args()
config = {}
model_section = MagicMock()
tts_section = MagicMock()
terminal_section = MagicMock()
gateway_section = MagicMock()
tools_section = MagicMock()
agent_section = MagicMock()
with (
patch.object(setup_mod, "ensure_hermes_home"),
patch.object(setup_mod, "load_config", return_value=config),
patch.object(setup_mod, "get_hermes_home", return_value=tmp_path),
patch.object(setup_mod, "is_interactive_stdin", return_value=True),
patch.object(
setup_mod,
"get_env_value",
side_effect=lambda key: "sk-test" if key == "OPENROUTER_API_KEY" else "",
),
patch("hermes_cli.auth.get_active_provider", return_value=None),
patch.object(setup_mod, "prompt_choice", return_value=4),
patch.object(
setup_mod,
"SETUP_SECTIONS",
[
("model", "Model & Provider", model_section),
("tts", "Text-to-Speech", tts_section),
("terminal", "Terminal Backend", terminal_section),
("gateway", "Messaging Platforms (Gateway)", gateway_section),
("tools", "Tools", tools_section),
("agent", "Agent Settings", agent_section),
],
),
patch.object(setup_mod, "save_config"),
patch.object(setup_mod, "_print_setup_summary"),
):
setup_mod.run_setup_wizard(args)
terminal_section.assert_called_once_with(config)
tts_section.assert_not_called()

41
tests/hermes_cli/test_status_model_provider.py
View File

@@ -2,6 +2,8 @@
from types import SimpleNamespace
from hermes_cli.nous_subscription import NousFeatureState, NousSubscriptionFeatures
def _patch_common_status_deps(monkeypatch, status_mod, tmp_path, *, openai_base_url=""):
import hermes_cli.auth as auth_mod
@@ -59,3 +61,42 @@ def test_show_status_displays_legacy_string_model_and_custom_endpoint(monkeypatc
out = capsys.readouterr().out
assert "Model: qwen3:latest" in out
assert "Provider: Custom endpoint" in out
def test_show_status_reports_managed_nous_features(monkeypatch, capsys, tmp_path):
from hermes_cli import status as status_mod
_patch_common_status_deps(monkeypatch, status_mod, tmp_path)
monkeypatch.setattr(
status_mod,
"load_config",
lambda: {"model": {"default": "claude-opus-4-6", "provider": "nous"}},
raising=False,
)
monkeypatch.setattr(status_mod, "resolve_requested_provider", lambda requested=None: "nous", raising=False)
monkeypatch.setattr(status_mod, "resolve_provider", lambda requested=None, **kwargs: "nous", raising=False)
monkeypatch.setattr(status_mod, "provider_label", lambda provider: "Nous Portal", raising=False)
monkeypatch.setattr(
status_mod,
"get_nous_subscription_features",
lambda config: NousSubscriptionFeatures(
subscribed=True,
nous_auth_present=True,
provider_is_nous=True,
features={
"web": NousFeatureState("web", "Web tools", True, True, True, True, False, True, "firecrawl"),
"image_gen": NousFeatureState("image_gen", "Image generation", True, True, True, True, False, True, "Nous Subscription"),
"tts": NousFeatureState("tts", "OpenAI TTS", True, True, True, True, False, True, "OpenAI TTS"),
"browser": NousFeatureState("browser", "Browser automation", True, True, True, True, False, True, "Browserbase"),
"modal": NousFeatureState("modal", "Modal execution", False, True, False, False, False, True, "local"),
},
),
raising=False,
)
status_mod.show_status(SimpleNamespace(all=False, deep=False))
out = capsys.readouterr().out
assert "Nous Subscription Features" in out
assert "Browser automation" in out
assert "active via Nous subscription" in out

79
tests/hermes_cli/test_tools_config.py
View File

@@ -3,10 +3,14 @@
from unittest.mock import patch
from hermes_cli.tools_config import (
_configure_provider,
_get_platform_tools,
_platform_toolset_summary,
_save_platform_tools,
_toolset_has_keys,
TOOL_CATEGORIES,
_visible_providers,
tools_command,
)
@@ -45,6 +49,10 @@ def test_toolset_has_keys_for_vision_accepts_codex_auth(tmp_path, monkeypatch):
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.delenv("AUXILIARY_VISION_PROVIDER", raising=False)
monkeypatch.delenv("CONTEXT_VISION_PROVIDER", raising=False)
monkeypatch.setattr(
"agent.auxiliary_client.resolve_vision_provider_client",
lambda: ("openai-codex", object(), "gpt-4.1"),
)
assert _toolset_has_keys("vision") is True
@@ -204,3 +212,74 @@ def test_save_platform_tools_still_preserves_mcp_with_platform_default_present()
# Deselected configurable toolset removed
assert "terminal" not in saved
def test_visible_providers_include_nous_subscription_when_logged_in(monkeypatch):
config = {"model": {"provider": "nous"}}
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_auth_status",
lambda: {"logged_in": True},
)
providers = _visible_providers(TOOL_CATEGORIES["browser"], config)
assert providers[0]["name"].startswith("Nous Subscription")
def test_local_browser_provider_is_saved_explicitly(monkeypatch):
config = {}
local_provider = next(
provider
for provider in TOOL_CATEGORIES["browser"]["providers"]
if provider.get("browser_provider") == "local"
)
monkeypatch.setattr("hermes_cli.tools_config._run_post_setup", lambda key: None)
_configure_provider(local_provider, config)
assert config["browser"]["cloud_provider"] == "local"
def test_first_install_nous_auto_configures_managed_defaults(monkeypatch):
config = {
"model": {"provider": "nous"},
"platform_toolsets": {"cli": []},
}
for env_var in (
"VOICE_TOOLS_OPENAI_KEY",
"OPENAI_API_KEY",
"ELEVENLABS_API_KEY",
"FIRECRAWL_API_KEY",
"FIRECRAWL_API_URL",
"TAVILY_API_KEY",
"PARALLEL_API_KEY",
"BROWSERBASE_API_KEY",
"BROWSERBASE_PROJECT_ID",
"BROWSER_USE_API_KEY",
"FAL_KEY",
):
monkeypatch.delenv(env_var, raising=False)
monkeypatch.setattr(
"hermes_cli.tools_config._prompt_toolset_checklist",
lambda *args, **kwargs: {"web", "image_gen", "tts", "browser"},
)
monkeypatch.setattr("hermes_cli.tools_config.save_config", lambda config: None)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_auth_status",
lambda: {"logged_in": True},
)
configured = []
monkeypatch.setattr(
"hermes_cli.tools_config._configure_toolset",
lambda ts_key, config: configured.append(ts_key),
)
tools_command(first_install=True, config=config)
assert config["web"]["backend"] == "firecrawl"
assert config["tts"]["provider"] == "openai"
assert config["browser"]["cloud_provider"] == "browserbase"
assert configured == []

135
tests/test_cli_provider_resolution.py
View File

@@ -78,6 +78,13 @@ def _install_prompt_toolkit_stubs():
def _import_cli():
for name in list(sys.modules):
if name == "cli" or name == "run_agent" or name == "tools" or name.startswith("tools."):
sys.modules.pop(name, None)
if "firecrawl" not in sys.modules:
sys.modules["firecrawl"] = types.SimpleNamespace(Firecrawl=object)
try:
importlib.import_module("prompt_toolkit")
except ModuleNotFoundError:
@@ -269,6 +276,81 @@ def test_codex_provider_replaces_incompatible_default_model(monkeypatch):
assert shell.model == "gpt-5.2-codex"
def test_model_flow_nous_prints_subscription_guidance_without_mutating_explicit_tts(monkeypatch, capsys):
config = {
"model": {"provider": "nous", "default": "claude-opus-4-6"},
"tts": {"provider": "elevenlabs"},
"browser": {"cloud_provider": "browser-use"},
}
monkeypatch.setattr(
"hermes_cli.auth.get_provider_auth_state",
lambda provider: {"access_token": "nous-token"},
)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["claude-opus-4-6"],
)
monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda model_ids, current_model="": "claude-opus-4-6")
monkeypatch.setattr("hermes_cli.auth._save_model_choice", lambda model: None)
monkeypatch.setattr("hermes_cli.auth._update_config_for_provider", lambda provider, url: None)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_subscription_explainer_lines",
lambda: ["Nous subscription enables managed web tools."],
)
hermes_main._model_flow_nous(config, current_model="claude-opus-4-6")
out = capsys.readouterr().out
assert "Nous subscription enables managed web tools." in out
assert config["tts"]["provider"] == "elevenlabs"
assert config["browser"]["cloud_provider"] == "browser-use"
def test_model_flow_nous_applies_managed_tts_default_when_unconfigured(monkeypatch, capsys):
config = {
"model": {"provider": "nous", "default": "claude-opus-4-6"},
"tts": {"provider": "edge"},
}
monkeypatch.setattr(
"hermes_cli.auth.get_provider_auth_state",
lambda provider: {"access_token": "nous-token"},
)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["claude-opus-4-6"],
)
monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda model_ids, current_model="": "claude-opus-4-6")
monkeypatch.setattr("hermes_cli.auth._save_model_choice", lambda model: None)
monkeypatch.setattr("hermes_cli.auth._update_config_for_provider", lambda provider, url: None)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_subscription_explainer_lines",
lambda: ["Nous subscription enables managed web tools."],
)
hermes_main._model_flow_nous(config, current_model="claude-opus-4-6")
out = capsys.readouterr().out
assert "Nous subscription enables managed web tools." in out
assert "OpenAI TTS via your Nous subscription" in out
assert config["tts"]["provider"] == "openai"
def test_codex_provider_uses_config_model(monkeypatch):
"""Model comes from config.yaml, not LLM_MODEL env var.
Config.yaml is the single source of truth to avoid multi-agent conflicts."""
@@ -468,4 +550,55 @@ def test_model_flow_custom_saves_verified_v1_base_url(monkeypatch, capsys):
assert "Saving the working base URL instead" in output
assert saved_env["OPENAI_BASE_URL"] == "http://localhost:8000/v1"
assert saved_env["OPENAI_API_KEY"] == "local-key"
assert saved_env["MODEL"] == "llm"
assert saved_env["MODEL"] == "llm"
def test_cmd_model_forwards_nous_login_tls_options(monkeypatch):
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"model": {"default": "gpt-5", "provider": "nous"}},
)
monkeypatch.setattr("hermes_cli.config.save_config", lambda cfg: None)
monkeypatch.setattr("hermes_cli.config.get_env_value", lambda key: "")
monkeypatch.setattr("hermes_cli.config.save_env_value", lambda key, value: None)
monkeypatch.setattr("hermes_cli.auth.resolve_provider", lambda requested, **kwargs: "nous")
monkeypatch.setattr("hermes_cli.auth.get_provider_auth_state", lambda provider_id: None)
monkeypatch.setattr(hermes_main, "_prompt_provider_choice", lambda choices: 0)
captured = {}
def _fake_login(login_args, provider_config):
captured["portal_url"] = login_args.portal_url
captured["inference_url"] = login_args.inference_url
captured["client_id"] = login_args.client_id
captured["scope"] = login_args.scope
captured["no_browser"] = login_args.no_browser
captured["timeout"] = login_args.timeout
captured["ca_bundle"] = login_args.ca_bundle
captured["insecure"] = login_args.insecure
monkeypatch.setattr("hermes_cli.auth._login_nous", _fake_login)
hermes_main.cmd_model(
SimpleNamespace(
portal_url="https://portal.nousresearch.com",
inference_url="https://inference.nousresearch.com/v1",
client_id="hermes-local",
scope="openid profile",
no_browser=True,
timeout=7.5,
ca_bundle="/tmp/local-ca.pem",
insecure=True,
)
)
assert captured == {
"portal_url": "https://portal.nousresearch.com",
"inference_url": "https://inference.nousresearch.com/v1",
"client_id": "hermes-local",
"scope": "openid profile",
"no_browser": True,
"timeout": 7.5,
"ca_bundle": "/tmp/local-ca.pem",
"insecure": True,
}

5
tests/test_run_agent.py
View File

@@ -584,6 +584,11 @@ class TestBuildSystemPrompt:
# Should contain current date info like "Conversation started:"
assert "Conversation started:" in prompt
def test_includes_nous_subscription_prompt(self, agent, monkeypatch):
monkeypatch.setattr(run_agent, "build_nous_subscription_prompt", lambda tool_names: "NOUS SUBSCRIPTION BLOCK")
prompt = agent._build_system_prompt()
assert "NOUS SUBSCRIPTION BLOCK" in prompt
class TestInvalidateSystemPrompt:
def test_clears_cache(self, agent):

418
tests/tools/test_managed_browserbase_and_modal.py Normal file
View File

@@ -0,0 +1,418 @@
import os
import sys
import tempfile
import threading
import types
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from unittest.mock import patch
import pytest
TOOLS_DIR = Path(__file__).resolve().parents[2] / "tools"
def _load_tool_module(module_name: str, filename: str):
spec = spec_from_file_location(module_name, TOOLS_DIR / filename)
assert spec and spec.loader
module = module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
def _reset_modules(prefixes: tuple[str, ...]):
for name in list(sys.modules):
if name.startswith(prefixes):
sys.modules.pop(name, None)
@pytest.fixture(autouse=True)
def _restore_tool_and_agent_modules():
original_modules = {
name: module
for name, module in sys.modules.items()
if name == "tools"
or name.startswith("tools.")
or name == "agent"
or name.startswith("agent.")
}
try:
yield
finally:
_reset_modules(("tools", "agent"))
sys.modules.update(original_modules)
def _install_fake_tools_package():
_reset_modules(("tools", "agent"))
tools_package = types.ModuleType("tools")
tools_package.__path__ = [str(TOOLS_DIR)] # type: ignore[attr-defined]
sys.modules["tools"] = tools_package
env_package = types.ModuleType("tools.environments")
env_package.__path__ = [str(TOOLS_DIR / "environments")] # type: ignore[attr-defined]
sys.modules["tools.environments"] = env_package
agent_package = types.ModuleType("agent")
agent_package.__path__ = [] # type: ignore[attr-defined]
sys.modules["agent"] = agent_package
sys.modules["agent.auxiliary_client"] = types.SimpleNamespace(
call_llm=lambda *args, **kwargs: "",
)
sys.modules["tools.managed_tool_gateway"] = _load_tool_module(
"tools.managed_tool_gateway",
"managed_tool_gateway.py",
)
interrupt_event = threading.Event()
sys.modules["tools.interrupt"] = types.SimpleNamespace(
set_interrupt=lambda value=True: interrupt_event.set() if value else interrupt_event.clear(),
is_interrupted=lambda: interrupt_event.is_set(),
_interrupt_event=interrupt_event,
)
sys.modules["tools.approval"] = types.SimpleNamespace(
detect_dangerous_command=lambda *args, **kwargs: None,
check_dangerous_command=lambda *args, **kwargs: {"approved": True},
check_all_command_guards=lambda *args, **kwargs: {"approved": True},
load_permanent_allowlist=lambda *args, **kwargs: [],
DANGEROUS_PATTERNS=[],
)
class _Registry:
def register(self, **kwargs):
return None
sys.modules["tools.registry"] = types.SimpleNamespace(registry=_Registry())
class _DummyEnvironment:
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
def cleanup(self):
return None
sys.modules["tools.environments.base"] = types.SimpleNamespace(BaseEnvironment=_DummyEnvironment)
sys.modules["tools.environments.local"] = types.SimpleNamespace(LocalEnvironment=_DummyEnvironment)
sys.modules["tools.environments.singularity"] = types.SimpleNamespace(
_get_scratch_dir=lambda: Path(tempfile.gettempdir()),
SingularityEnvironment=_DummyEnvironment,
)
sys.modules["tools.environments.ssh"] = types.SimpleNamespace(SSHEnvironment=_DummyEnvironment)
sys.modules["tools.environments.docker"] = types.SimpleNamespace(DockerEnvironment=_DummyEnvironment)
sys.modules["tools.environments.modal"] = types.SimpleNamespace(ModalEnvironment=_DummyEnvironment)
sys.modules["tools.environments.managed_modal"] = types.SimpleNamespace(ManagedModalEnvironment=_DummyEnvironment)
def test_browserbase_explicit_local_mode_stays_local_even_when_managed_gateway_is_ready(tmp_path):
_install_fake_tools_package()
(tmp_path / "config.yaml").write_text("browser:\n cloud_provider: local\n", encoding="utf-8")
env = os.environ.copy()
env.pop("BROWSERBASE_API_KEY", None)
env.pop("BROWSERBASE_PROJECT_ID", None)
env.update({
"HERMES_HOME": str(tmp_path),
"TOOL_GATEWAY_USER_TOKEN": "nous-token",
"BROWSERBASE_GATEWAY_URL": "http://127.0.0.1:3009",
})
with patch.dict(os.environ, env, clear=True):
browser_tool = _load_tool_module("tools.browser_tool", "browser_tool.py")
local_mode = browser_tool._is_local_mode()
provider = browser_tool._get_cloud_provider()
assert local_mode is True
assert provider is None
def test_browserbase_managed_gateway_adds_idempotency_key_and_persists_external_call_id():
_install_fake_tools_package()
env = os.environ.copy()
env.pop("BROWSERBASE_API_KEY", None)
env.pop("BROWSERBASE_PROJECT_ID", None)
env.update({
"TOOL_GATEWAY_USER_TOKEN": "nous-token",
"BROWSERBASE_GATEWAY_URL": "http://127.0.0.1:3009",
})
class _Response:
status_code = 200
ok = True
text = ""
headers = {"x-external-call-id": "call-browserbase-1"}
def json(self):
return {
"id": "bb_local_session_1",
"connectUrl": "wss://connect.browserbase.example/session",
}
with patch.dict(os.environ, env, clear=True):
browserbase_module = _load_tool_module(
"tools.browser_providers.browserbase",
"browser_providers/browserbase.py",
)
with patch.object(browserbase_module.requests, "post", return_value=_Response()) as post:
provider = browserbase_module.BrowserbaseProvider()
session = provider.create_session("task-browserbase-managed")
sent_headers = post.call_args.kwargs["headers"]
assert sent_headers["X-BB-API-Key"] == "nous-token"
assert sent_headers["X-Idempotency-Key"].startswith("browserbase-session-create:")
assert session["external_call_id"] == "call-browserbase-1"
def test_browserbase_managed_gateway_reuses_pending_idempotency_key_after_timeout():
_install_fake_tools_package()
env = os.environ.copy()
env.pop("BROWSERBASE_API_KEY", None)
env.pop("BROWSERBASE_PROJECT_ID", None)
env.update({
"TOOL_GATEWAY_USER_TOKEN": "nous-token",
"BROWSERBASE_GATEWAY_URL": "http://127.0.0.1:3009",
})
class _Response:
status_code = 200
ok = True
text = ""
headers = {"x-external-call-id": "call-browserbase-2"}
def json(self):
return {
"id": "bb_local_session_2",
"connectUrl": "wss://connect.browserbase.example/session2",
}
with patch.dict(os.environ, env, clear=True):
browserbase_module = _load_tool_module(
"tools.browser_providers.browserbase",
"browser_providers/browserbase.py",
)
provider = browserbase_module.BrowserbaseProvider()
timeout = browserbase_module.requests.Timeout("timed out")
with patch.object(
browserbase_module.requests,
"post",
side_effect=[timeout, _Response()],
) as post:
try:
provider.create_session("task-browserbase-timeout")
except browserbase_module.requests.Timeout:
pass
else:
raise AssertionError("Expected Browserbase create_session to propagate timeout")
provider.create_session("task-browserbase-timeout")
first_headers = post.call_args_list[0].kwargs["headers"]
second_headers = post.call_args_list[1].kwargs["headers"]
assert first_headers["X-Idempotency-Key"] == second_headers["X-Idempotency-Key"]
def test_browserbase_managed_gateway_preserves_pending_idempotency_key_for_in_progress_conflicts():
_install_fake_tools_package()
env = os.environ.copy()
env.pop("BROWSERBASE_API_KEY", None)
env.pop("BROWSERBASE_PROJECT_ID", None)
env.update({
"TOOL_GATEWAY_USER_TOKEN": "nous-token",
"BROWSERBASE_GATEWAY_URL": "http://127.0.0.1:3009",
})
class _ConflictResponse:
status_code = 409
ok = False
text = '{"error":{"code":"CONFLICT","message":"Managed Browserbase session creation is already in progress for this idempotency key"}}'
headers = {}
def json(self):
return {
"error": {
"code": "CONFLICT",
"message": "Managed Browserbase session creation is already in progress for this idempotency key",
}
}
class _SuccessResponse:
status_code = 200
ok = True
text = ""
headers = {"x-external-call-id": "call-browserbase-4"}
def json(self):
return {
"id": "bb_local_session_4",
"connectUrl": "wss://connect.browserbase.example/session4",
}
with patch.dict(os.environ, env, clear=True):
browserbase_module = _load_tool_module(
"tools.browser_providers.browserbase",
"browser_providers/browserbase.py",
)
provider = browserbase_module.BrowserbaseProvider()
with patch.object(
browserbase_module.requests,
"post",
side_effect=[_ConflictResponse(), _SuccessResponse()],
) as post:
try:
provider.create_session("task-browserbase-conflict")
except RuntimeError:
pass
else:
raise AssertionError("Expected Browserbase create_session to propagate the in-progress conflict")
provider.create_session("task-browserbase-conflict")
first_headers = post.call_args_list[0].kwargs["headers"]
second_headers = post.call_args_list[1].kwargs["headers"]
assert first_headers["X-Idempotency-Key"] == second_headers["X-Idempotency-Key"]
def test_browserbase_managed_gateway_uses_new_idempotency_key_for_a_new_session_after_success():
_install_fake_tools_package()
env = os.environ.copy()
env.pop("BROWSERBASE_API_KEY", None)
env.pop("BROWSERBASE_PROJECT_ID", None)
env.update({
"TOOL_GATEWAY_USER_TOKEN": "nous-token",
"BROWSERBASE_GATEWAY_URL": "http://127.0.0.1:3009",
})
class _Response:
status_code = 200
ok = True
text = ""
headers = {"x-external-call-id": "call-browserbase-3"}
def json(self):
return {
"id": "bb_local_session_3",
"connectUrl": "wss://connect.browserbase.example/session3",
}
with patch.dict(os.environ, env, clear=True):
browserbase_module = _load_tool_module(
"tools.browser_providers.browserbase",
"browser_providers/browserbase.py",
)
provider = browserbase_module.BrowserbaseProvider()
with patch.object(browserbase_module.requests, "post", side_effect=[_Response(), _Response()]) as post:
provider.create_session("task-browserbase-new")
provider.create_session("task-browserbase-new")
first_headers = post.call_args_list[0].kwargs["headers"]
second_headers = post.call_args_list[1].kwargs["headers"]
assert first_headers["X-Idempotency-Key"] != second_headers["X-Idempotency-Key"]
def test_terminal_tool_prefers_managed_modal_when_gateway_ready_and_no_direct_creds():
_install_fake_tools_package()
env = os.environ.copy()
env.pop("MODAL_TOKEN_ID", None)
env.pop("MODAL_TOKEN_SECRET", None)
with patch.dict(os.environ, env, clear=True):
terminal_tool = _load_tool_module("tools.terminal_tool", "terminal_tool.py")
with (
patch.object(terminal_tool, "is_managed_tool_gateway_ready", return_value=True),
patch.object(terminal_tool, "_ManagedModalEnvironment", return_value="managed-modal-env") as managed_ctor,
patch.object(terminal_tool, "_ModalEnvironment", return_value="direct-modal-env") as direct_ctor,
patch.object(Path, "exists", return_value=False),
):
result = terminal_tool._create_environment(
env_type="modal",
image="python:3.11",
cwd="/root",
timeout=60,
container_config={
"container_cpu": 1,
"container_memory": 2048,
"container_disk": 1024,
"container_persistent": True,
"modal_mode": "auto",
},
task_id="task-modal-managed",
)
assert result == "managed-modal-env"
assert managed_ctor.called
assert not direct_ctor.called
def test_terminal_tool_keeps_direct_modal_when_direct_credentials_exist():
_install_fake_tools_package()
env = os.environ.copy()
env.update({
"MODAL_TOKEN_ID": "tok-id",
"MODAL_TOKEN_SECRET": "tok-secret",
})
with patch.dict(os.environ, env, clear=True):
terminal_tool = _load_tool_module("tools.terminal_tool", "terminal_tool.py")
with (
patch.object(terminal_tool, "is_managed_tool_gateway_ready", return_value=True),
patch.object(terminal_tool, "_ManagedModalEnvironment", return_value="managed-modal-env") as managed_ctor,
patch.object(terminal_tool, "_ModalEnvironment", return_value="direct-modal-env") as direct_ctor,
):
result = terminal_tool._create_environment(
env_type="modal",
image="python:3.11",
cwd="/root",
timeout=60,
container_config={
"container_cpu": 1,
"container_memory": 2048,
"container_disk": 1024,
"container_persistent": True,
"modal_mode": "auto",
},
task_id="task-modal-direct",
)
assert result == "direct-modal-env"
assert direct_ctor.called
assert not managed_ctor.called
def test_terminal_tool_respects_direct_modal_mode_without_falling_back_to_managed():
_install_fake_tools_package()
env = os.environ.copy()
env.pop("MODAL_TOKEN_ID", None)
env.pop("MODAL_TOKEN_SECRET", None)
with patch.dict(os.environ, env, clear=True):
terminal_tool = _load_tool_module("tools.terminal_tool", "terminal_tool.py")
with (
patch.object(terminal_tool, "is_managed_tool_gateway_ready", return_value=True),
patch.object(Path, "exists", return_value=False),
):
with pytest.raises(ValueError, match="direct Modal credentials"):
terminal_tool._create_environment(
env_type="modal",
image="python:3.11",
cwd="/root",
timeout=60,
container_config={
"container_cpu": 1,
"container_memory": 2048,
"container_disk": 1024,
"container_persistent": True,
"modal_mode": "direct",
},
task_id="task-modal-direct-only",
)

288
tests/tools/test_managed_media_gateways.py Normal file
View File

@@ -0,0 +1,288 @@
import sys
import types
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
import pytest
TOOLS_DIR = Path(__file__).resolve().parents[2] / "tools"
def _load_tool_module(module_name: str, filename: str):
spec = spec_from_file_location(module_name, TOOLS_DIR / filename)
assert spec and spec.loader
module = module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
@pytest.fixture(autouse=True)
def _restore_tool_and_agent_modules():
original_modules = {
name: module
for name, module in sys.modules.items()
if name == "tools"
or name.startswith("tools.")
or name == "agent"
or name.startswith("agent.")
or name in {"fal_client", "openai"}
}
try:
yield
finally:
for name in list(sys.modules):
if (
name == "tools"
or name.startswith("tools.")
or name == "agent"
or name.startswith("agent.")
or name in {"fal_client", "openai"}
):
sys.modules.pop(name, None)
sys.modules.update(original_modules)
def _install_fake_tools_package():
tools_package = types.ModuleType("tools")
tools_package.__path__ = [str(TOOLS_DIR)] # type: ignore[attr-defined]
sys.modules["tools"] = tools_package
sys.modules["tools.debug_helpers"] = types.SimpleNamespace(
DebugSession=lambda *args, **kwargs: types.SimpleNamespace(
active=False,
session_id="debug-session",
log_call=lambda *a, **k: None,
save=lambda: None,
get_session_info=lambda: {},
)
)
sys.modules["tools.managed_tool_gateway"] = _load_tool_module(
"tools.managed_tool_gateway",
"managed_tool_gateway.py",
)
def _install_fake_fal_client(captured):
def submit(model, arguments=None, headers=None):
raise AssertionError("managed FAL gateway mode should use fal_client.SyncClient")
class FakeResponse:
def json(self):
return {
"request_id": "req-123",
"response_url": "http://127.0.0.1:3009/requests/req-123",
"status_url": "http://127.0.0.1:3009/requests/req-123/status",
"cancel_url": "http://127.0.0.1:3009/requests/req-123/cancel",
}
def _maybe_retry_request(client, method, url, json=None, timeout=None, headers=None):
captured["submit_via"] = "managed_client"
captured["http_client"] = client
captured["method"] = method
captured["submit_url"] = url
captured["arguments"] = json
captured["timeout"] = timeout
captured["headers"] = headers
return FakeResponse()
class SyncRequestHandle:
def __init__(self, request_id, response_url, status_url, cancel_url, client):
captured["request_id"] = request_id
captured["response_url"] = response_url
captured["status_url"] = status_url
captured["cancel_url"] = cancel_url
captured["handle_client"] = client
class SyncClient:
def __init__(self, key=None, default_timeout=120.0):
captured["sync_client_inits"] = captured.get("sync_client_inits", 0) + 1
captured["client_key"] = key
captured["client_timeout"] = default_timeout
self.default_timeout = default_timeout
self._client = object()
fal_client_module = types.SimpleNamespace(
submit=submit,
SyncClient=SyncClient,
client=types.SimpleNamespace(
_maybe_retry_request=_maybe_retry_request,
_raise_for_status=lambda response: None,
SyncRequestHandle=SyncRequestHandle,
),
)
sys.modules["fal_client"] = fal_client_module
return fal_client_module
def _install_fake_openai_module(captured, transcription_response=None):
class FakeSpeechResponse:
def stream_to_file(self, output_path):
captured["stream_to_file"] = output_path
class FakeOpenAI:
def __init__(self, api_key, base_url, **kwargs):
captured["api_key"] = api_key
captured["base_url"] = base_url
captured["client_kwargs"] = kwargs
captured["close_calls"] = captured.get("close_calls", 0)
def create_speech(**kwargs):
captured["speech_kwargs"] = kwargs
return FakeSpeechResponse()
def create_transcription(**kwargs):
captured["transcription_kwargs"] = kwargs
return transcription_response
self.audio = types.SimpleNamespace(
speech=types.SimpleNamespace(
create=create_speech
),
transcriptions=types.SimpleNamespace(
create=create_transcription
),
)
def close(self):
captured["close_calls"] += 1
fake_module = types.SimpleNamespace(
OpenAI=FakeOpenAI,
APIError=Exception,
APIConnectionError=Exception,
APITimeoutError=Exception,
)
sys.modules["openai"] = fake_module
def test_managed_fal_submit_uses_gateway_origin_and_nous_token(monkeypatch):
captured = {}
_install_fake_tools_package()
_install_fake_fal_client(captured)
monkeypatch.delenv("FAL_KEY", raising=False)
monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-token")
image_generation_tool = _load_tool_module(
"tools.image_generation_tool",
"image_generation_tool.py",
)
monkeypatch.setattr(image_generation_tool.uuid, "uuid4", lambda: "fal-submit-123")
image_generation_tool._submit_fal_request(
"fal-ai/flux-2-pro",
{"prompt": "test prompt", "num_images": 1},
)
assert captured["submit_via"] == "managed_client"
assert captured["client_key"] == "nous-token"
assert captured["submit_url"] == "http://127.0.0.1:3009/fal-ai/flux-2-pro"
assert captured["method"] == "POST"
assert captured["arguments"] == {"prompt": "test prompt", "num_images": 1}
assert captured["headers"] == {"x-idempotency-key": "fal-submit-123"}
assert captured["sync_client_inits"] == 1
def test_managed_fal_submit_reuses_cached_sync_client(monkeypatch):
captured = {}
_install_fake_tools_package()
_install_fake_fal_client(captured)
monkeypatch.delenv("FAL_KEY", raising=False)
monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-token")
image_generation_tool = _load_tool_module(
"tools.image_generation_tool",
"image_generation_tool.py",
)
image_generation_tool._submit_fal_request("fal-ai/flux-2-pro", {"prompt": "first"})
first_client = captured["http_client"]
image_generation_tool._submit_fal_request("fal-ai/flux-2-pro", {"prompt": "second"})
assert captured["sync_client_inits"] == 1
assert captured["http_client"] is first_client
def test_openai_tts_uses_managed_audio_gateway_when_direct_key_absent(monkeypatch, tmp_path):
captured = {}
_install_fake_tools_package()
_install_fake_openai_module(captured)
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
monkeypatch.setenv("TOOL_GATEWAY_DOMAIN", "nousresearch.com")
monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-token")
tts_tool = _load_tool_module("tools.tts_tool", "tts_tool.py")
monkeypatch.setattr(tts_tool.uuid, "uuid4", lambda: "tts-call-123")
output_path = tmp_path / "speech.mp3"
tts_tool._generate_openai_tts("hello world", str(output_path), {"openai": {}})
assert captured["api_key"] == "nous-token"
assert captured["base_url"] == "https://openai-audio-gateway.nousresearch.com/v1"
assert captured["speech_kwargs"]["model"] == "gpt-4o-mini-tts"
assert captured["speech_kwargs"]["extra_headers"] == {"x-idempotency-key": "tts-call-123"}
assert captured["stream_to_file"] == str(output_path)
assert captured["close_calls"] == 1
def test_openai_tts_accepts_openai_api_key_as_direct_fallback(monkeypatch, tmp_path):
captured = {}
_install_fake_tools_package()
_install_fake_openai_module(captured)
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
monkeypatch.setenv("OPENAI_API_KEY", "openai-direct-key")
monkeypatch.setenv("TOOL_GATEWAY_DOMAIN", "nousresearch.com")
monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-token")
tts_tool = _load_tool_module("tools.tts_tool", "tts_tool.py")
output_path = tmp_path / "speech.mp3"
tts_tool._generate_openai_tts("hello world", str(output_path), {"openai": {}})
assert captured["api_key"] == "openai-direct-key"
assert captured["base_url"] == "https://api.openai.com/v1"
assert captured["close_calls"] == 1
def test_transcription_uses_model_specific_response_formats(monkeypatch, tmp_path):
whisper_capture = {}
_install_fake_tools_package()
_install_fake_openai_module(whisper_capture, transcription_response="hello from whisper")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text("stt:\n provider: openai\n")
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
monkeypatch.setenv("TOOL_GATEWAY_DOMAIN", "nousresearch.com")
monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-token")
transcription_tools = _load_tool_module(
"tools.transcription_tools",
"transcription_tools.py",
)
transcription_tools._load_stt_config = lambda: {"provider": "openai"}
audio_path = tmp_path / "audio.wav"
audio_path.write_bytes(b"RIFF0000WAVEfmt ")
whisper_result = transcription_tools.transcribe_audio(str(audio_path), model="whisper-1")
assert whisper_result["success"] is True
assert whisper_capture["base_url"] == "https://openai-audio-gateway.nousresearch.com/v1"
assert whisper_capture["transcription_kwargs"]["response_format"] == "text"
assert whisper_capture["close_calls"] == 1
json_capture = {}
_install_fake_openai_module(
json_capture,
transcription_response=types.SimpleNamespace(text="hello from gpt-4o"),
)
transcription_tools = _load_tool_module(
"tools.transcription_tools",
"transcription_tools.py",
)
json_result = transcription_tools.transcribe_audio(
str(audio_path),
model="gpt-4o-mini-transcribe",
)
assert json_result["success"] is True
assert json_result["transcript"] == "hello from gpt-4o"
assert json_capture["transcription_kwargs"]["response_format"] == "json"
assert json_capture["close_calls"] == 1

213
tests/tools/test_managed_modal_environment.py Normal file
View File

@@ -0,0 +1,213 @@
import json
import sys
import tempfile
import threading
import types
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
TOOLS_DIR = Path(__file__).resolve().parents[2] / "tools"
def _load_tool_module(module_name: str, filename: str):
spec = spec_from_file_location(module_name, TOOLS_DIR / filename)
assert spec and spec.loader
module = module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
def _reset_modules(prefixes: tuple[str, ...]):
for name in list(sys.modules):
if name.startswith(prefixes):
sys.modules.pop(name, None)
def _install_fake_tools_package():
_reset_modules(("tools", "agent", "hermes_cli"))
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
sys.modules["hermes_cli"] = hermes_cli
sys.modules["hermes_cli.config"] = types.SimpleNamespace(
get_hermes_home=lambda: Path(tempfile.gettempdir()) / "hermes-home",
)
tools_package = types.ModuleType("tools")
tools_package.__path__ = [str(TOOLS_DIR)] # type: ignore[attr-defined]
sys.modules["tools"] = tools_package
env_package = types.ModuleType("tools.environments")
env_package.__path__ = [str(TOOLS_DIR / "environments")] # type: ignore[attr-defined]
sys.modules["tools.environments"] = env_package
interrupt_event = threading.Event()
sys.modules["tools.interrupt"] = types.SimpleNamespace(
set_interrupt=lambda value=True: interrupt_event.set() if value else interrupt_event.clear(),
is_interrupted=lambda: interrupt_event.is_set(),
_interrupt_event=interrupt_event,
)
class _DummyBaseEnvironment:
def __init__(self, cwd: str, timeout: int, env=None):
self.cwd = cwd
self.timeout = timeout
self.env = env or {}
def _prepare_command(self, command: str):
return command, None
sys.modules["tools.environments.base"] = types.SimpleNamespace(BaseEnvironment=_DummyBaseEnvironment)
sys.modules["tools.managed_tool_gateway"] = types.SimpleNamespace(
resolve_managed_tool_gateway=lambda vendor: types.SimpleNamespace(
vendor=vendor,
gateway_origin="https://modal-gateway.example.com",
nous_user_token="user-token",
managed_mode=True,
)
)
return interrupt_event
class _FakeResponse:
def __init__(self, status_code: int, payload=None, text: str = ""):
self.status_code = status_code
self._payload = payload
self.text = text
def json(self):
if isinstance(self._payload, Exception):
raise self._payload
return self._payload
def test_managed_modal_execute_polls_until_completed(monkeypatch):
_install_fake_tools_package()
managed_modal = _load_tool_module("tools.environments.managed_modal", "environments/managed_modal.py")
calls = []
poll_count = {"value": 0}
def fake_request(method, url, headers=None, json=None, timeout=None):
calls.append((method, url, json, timeout))
if method == "POST" and url.endswith("/v1/sandboxes"):
return _FakeResponse(200, {"id": "sandbox-1"})
if method == "POST" and url.endswith("/execs"):
return _FakeResponse(202, {"execId": json["execId"], "status": "running"})
if method == "GET" and "/execs/" in url:
poll_count["value"] += 1
if poll_count["value"] == 1:
return _FakeResponse(200, {"execId": url.rsplit("/", 1)[-1], "status": "running"})
return _FakeResponse(200, {
"execId": url.rsplit("/", 1)[-1],
"status": "completed",
"output": "hello",
"returncode": 0,
})
if method == "POST" and url.endswith("/terminate"):
return _FakeResponse(200, {"status": "terminated"})
raise AssertionError(f"Unexpected request: {method} {url}")
monkeypatch.setattr(managed_modal.requests, "request", fake_request)
monkeypatch.setattr(managed_modal.time, "sleep", lambda _: None)
env = managed_modal.ManagedModalEnvironment(image="python:3.11")
result = env.execute("echo hello")
env.cleanup()
assert result == {"output": "hello", "returncode": 0}
assert any(call[0] == "POST" and call[1].endswith("/execs") for call in calls)
def test_managed_modal_create_sends_a_stable_idempotency_key(monkeypatch):
_install_fake_tools_package()
managed_modal = _load_tool_module("tools.environments.managed_modal", "environments/managed_modal.py")
create_headers = []
def fake_request(method, url, headers=None, json=None, timeout=None):
if method == "POST" and url.endswith("/v1/sandboxes"):
create_headers.append(headers or {})
return _FakeResponse(200, {"id": "sandbox-1"})
if method == "POST" and url.endswith("/terminate"):
return _FakeResponse(200, {"status": "terminated"})
raise AssertionError(f"Unexpected request: {method} {url}")
monkeypatch.setattr(managed_modal.requests, "request", fake_request)
env = managed_modal.ManagedModalEnvironment(image="python:3.11")
env.cleanup()
assert len(create_headers) == 1
assert isinstance(create_headers[0].get("x-idempotency-key"), str)
assert create_headers[0]["x-idempotency-key"]
def test_managed_modal_execute_cancels_on_interrupt(monkeypatch):
interrupt_event = _install_fake_tools_package()
managed_modal = _load_tool_module("tools.environments.managed_modal", "environments/managed_modal.py")
calls = []
def fake_request(method, url, headers=None, json=None, timeout=None):
calls.append((method, url, json, timeout))
if method == "POST" and url.endswith("/v1/sandboxes"):
return _FakeResponse(200, {"id": "sandbox-1"})
if method == "POST" and url.endswith("/execs"):
return _FakeResponse(202, {"execId": json["execId"], "status": "running"})
if method == "GET" and "/execs/" in url:
return _FakeResponse(200, {"execId": url.rsplit("/", 1)[-1], "status": "running"})
if method == "POST" and url.endswith("/cancel"):
return _FakeResponse(202, {"status": "cancelling"})
if method == "POST" and url.endswith("/terminate"):
return _FakeResponse(200, {"status": "terminated"})
raise AssertionError(f"Unexpected request: {method} {url}")
def fake_sleep(_seconds):
interrupt_event.set()
monkeypatch.setattr(managed_modal.requests, "request", fake_request)
monkeypatch.setattr(managed_modal.time, "sleep", fake_sleep)
env = managed_modal.ManagedModalEnvironment(image="python:3.11")
result = env.execute("sleep 30")
env.cleanup()
assert result == {
"output": "[Command interrupted - Modal sandbox exec cancelled]",
"returncode": 130,
}
assert any(call[0] == "POST" and call[1].endswith("/cancel") for call in calls)
poll_calls = [call for call in calls if call[0] == "GET" and "/execs/" in call[1]]
cancel_calls = [call for call in calls if call[0] == "POST" and call[1].endswith("/cancel")]
assert poll_calls[0][3] == (1.0, 5.0)
assert cancel_calls[0][3] == (1.0, 5.0)
def test_managed_modal_execute_returns_descriptive_error_on_missing_exec(monkeypatch):
_install_fake_tools_package()
managed_modal = _load_tool_module("tools.environments.managed_modal", "environments/managed_modal.py")
def fake_request(method, url, headers=None, json=None, timeout=None):
if method == "POST" and url.endswith("/v1/sandboxes"):
return _FakeResponse(200, {"id": "sandbox-1"})
if method == "POST" and url.endswith("/execs"):
return _FakeResponse(202, {"execId": json["execId"], "status": "running"})
if method == "GET" and "/execs/" in url:
return _FakeResponse(404, {"error": "not found"}, text="not found")
if method == "POST" and url.endswith("/terminate"):
return _FakeResponse(200, {"status": "terminated"})
raise AssertionError(f"Unexpected request: {method} {url}")
monkeypatch.setattr(managed_modal.requests, "request", fake_request)
monkeypatch.setattr(managed_modal.time, "sleep", lambda _: None)
env = managed_modal.ManagedModalEnvironment(image="python:3.11")
result = env.execute("echo hello")
env.cleanup()
assert result["returncode"] == 1
assert "not found" in result["output"].lower()

70
tests/tools/test_managed_tool_gateway.py Normal file
View File

@@ -0,0 +1,70 @@
import os
import json
from datetime import datetime, timedelta, timezone
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
import sys
from unittest.mock import patch
MODULE_PATH = Path(__file__).resolve().parents[2] / "tools" / "managed_tool_gateway.py"
MODULE_SPEC = spec_from_file_location("managed_tool_gateway_test_module", MODULE_PATH)
assert MODULE_SPEC and MODULE_SPEC.loader
managed_tool_gateway = module_from_spec(MODULE_SPEC)
sys.modules[MODULE_SPEC.name] = managed_tool_gateway
MODULE_SPEC.loader.exec_module(managed_tool_gateway)
resolve_managed_tool_gateway = managed_tool_gateway.resolve_managed_tool_gateway
def test_resolve_managed_tool_gateway_derives_vendor_origin_from_shared_domain():
with patch.dict(os.environ, {"TOOL_GATEWAY_DOMAIN": "nousresearch.com"}, clear=False):
result = resolve_managed_tool_gateway(
"firecrawl",
token_reader=lambda: "nous-token",
)
assert result is not None
assert result.gateway_origin == "https://firecrawl-gateway.nousresearch.com"
assert result.nous_user_token == "nous-token"
assert result.managed_mode is True
def test_resolve_managed_tool_gateway_uses_vendor_specific_override():
with patch.dict(os.environ, {"BROWSERBASE_GATEWAY_URL": "http://browserbase-gateway.localhost:3009/"}, clear=False):
result = resolve_managed_tool_gateway(
"browserbase",
token_reader=lambda: "nous-token",
)
assert result is not None
assert result.gateway_origin == "http://browserbase-gateway.localhost:3009"
def test_resolve_managed_tool_gateway_is_inactive_without_nous_token():
with patch.dict(os.environ, {"TOOL_GATEWAY_DOMAIN": "nousresearch.com"}, clear=False):
result = resolve_managed_tool_gateway(
"firecrawl",
token_reader=lambda: None,
)
assert result is None
def test_read_nous_access_token_refreshes_expiring_cached_token(tmp_path, monkeypatch):
monkeypatch.delenv("TOOL_GATEWAY_USER_TOKEN", raising=False)
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
expires_at = (datetime.now(timezone.utc) + timedelta(seconds=30)).isoformat()
(tmp_path / "auth.json").write_text(json.dumps({
"providers": {
"nous": {
"access_token": "stale-token",
"refresh_token": "refresh-token",
"expires_at": expires_at,
}
}
}))
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_access_token",
lambda refresh_skew_seconds=120: "fresh-token",
)
assert managed_tool_gateway.read_nous_access_token() == "fresh-token"

188
tests/tools/test_modal_snapshot_isolation.py Normal file
View File

@@ -0,0 +1,188 @@
import json
import sys
import types
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[2]
TOOLS_DIR = REPO_ROOT / "tools"
def _load_module(module_name: str, path: Path):
spec = spec_from_file_location(module_name, path)
assert spec and spec.loader
module = module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
def _reset_modules(prefixes: tuple[str, ...]):
for name in list(sys.modules):
if name.startswith(prefixes):
sys.modules.pop(name, None)
def _install_modal_test_modules(
tmp_path: Path,
*,
fail_on_snapshot_ids: set[str] | None = None,
snapshot_id: str = "im-fresh",
):
_reset_modules(("tools", "hermes_cli", "swerex", "modal"))
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = [] # type: ignore[attr-defined]
sys.modules["hermes_cli"] = hermes_cli
hermes_home = tmp_path / "hermes-home"
sys.modules["hermes_cli.config"] = types.SimpleNamespace(
get_hermes_home=lambda: hermes_home,
)
tools_package = types.ModuleType("tools")
tools_package.__path__ = [str(TOOLS_DIR)] # type: ignore[attr-defined]
sys.modules["tools"] = tools_package
env_package = types.ModuleType("tools.environments")
env_package.__path__ = [str(TOOLS_DIR / "environments")] # type: ignore[attr-defined]
sys.modules["tools.environments"] = env_package
class _DummyBaseEnvironment:
def __init__(self, cwd: str, timeout: int, env=None):
self.cwd = cwd
self.timeout = timeout
self.env = env or {}
def _prepare_command(self, command: str):
return command, None
sys.modules["tools.environments.base"] = types.SimpleNamespace(BaseEnvironment=_DummyBaseEnvironment)
sys.modules["tools.interrupt"] = types.SimpleNamespace(is_interrupted=lambda: False)
from_id_calls: list[str] = []
registry_calls: list[tuple[str, list[str] | None]] = []
deployment_calls: list[dict] = []
class _FakeImage:
@staticmethod
def from_id(image_id: str):
from_id_calls.append(image_id)
return {"kind": "snapshot", "image_id": image_id}
@staticmethod
def from_registry(image: str, setup_dockerfile_commands=None):
registry_calls.append((image, setup_dockerfile_commands))
return {"kind": "registry", "image": image}
class _FakeRuntime:
async def execute(self, _command):
return types.SimpleNamespace(stdout="ok", exit_code=0)
class _FakeModalDeployment:
def __init__(self, **kwargs):
deployment_calls.append(dict(kwargs))
self.image = kwargs["image"]
self.runtime = _FakeRuntime()
async def _snapshot_aio():
return types.SimpleNamespace(object_id=snapshot_id)
self._sandbox = types.SimpleNamespace(
snapshot_filesystem=types.SimpleNamespace(aio=_snapshot_aio),
)
async def start(self):
image = self.image if isinstance(self.image, dict) else {}
image_id = image.get("image_id")
if fail_on_snapshot_ids and image_id in fail_on_snapshot_ids:
raise RuntimeError(f"cannot restore {image_id}")
async def stop(self):
return None
class _FakeRexCommand:
def __init__(self, **kwargs):
self.kwargs = kwargs
sys.modules["modal"] = types.SimpleNamespace(Image=_FakeImage)
swerex = types.ModuleType("swerex")
swerex.__path__ = [] # type: ignore[attr-defined]
sys.modules["swerex"] = swerex
swerex_deployment = types.ModuleType("swerex.deployment")
swerex_deployment.__path__ = [] # type: ignore[attr-defined]
sys.modules["swerex.deployment"] = swerex_deployment
sys.modules["swerex.deployment.modal"] = types.SimpleNamespace(ModalDeployment=_FakeModalDeployment)
swerex_runtime = types.ModuleType("swerex.runtime")
swerex_runtime.__path__ = [] # type: ignore[attr-defined]
sys.modules["swerex.runtime"] = swerex_runtime
sys.modules["swerex.runtime.abstract"] = types.SimpleNamespace(Command=_FakeRexCommand)
return {
"snapshot_store": hermes_home / "modal_snapshots.json",
"deployment_calls": deployment_calls,
"from_id_calls": from_id_calls,
"registry_calls": registry_calls,
}
def test_modal_environment_migrates_legacy_snapshot_key_and_uses_snapshot_id(tmp_path):
state = _install_modal_test_modules(tmp_path)
snapshot_store = state["snapshot_store"]
snapshot_store.parent.mkdir(parents=True, exist_ok=True)
snapshot_store.write_text(json.dumps({"task-legacy": "im-legacy123"}))
modal_module = _load_module("tools.environments.modal", TOOLS_DIR / "environments" / "modal.py")
env = modal_module.ModalEnvironment(image="python:3.11", task_id="task-legacy")
try:
assert state["from_id_calls"] == ["im-legacy123"]
assert state["deployment_calls"][0]["image"] == {"kind": "snapshot", "image_id": "im-legacy123"}
assert json.loads(snapshot_store.read_text()) == {"direct:task-legacy": "im-legacy123"}
finally:
env.cleanup()
def test_modal_environment_prunes_stale_direct_snapshot_and_retries_base_image(tmp_path):
state = _install_modal_test_modules(tmp_path, fail_on_snapshot_ids={"im-stale123"})
snapshot_store = state["snapshot_store"]
snapshot_store.parent.mkdir(parents=True, exist_ok=True)
snapshot_store.write_text(json.dumps({"direct:task-stale": "im-stale123"}))
modal_module = _load_module("tools.environments.modal", TOOLS_DIR / "environments" / "modal.py")
env = modal_module.ModalEnvironment(image="python:3.11", task_id="task-stale")
try:
assert [call["image"] for call in state["deployment_calls"]] == [
{"kind": "snapshot", "image_id": "im-stale123"},
{"kind": "registry", "image": "python:3.11"},
]
assert json.loads(snapshot_store.read_text()) == {}
finally:
env.cleanup()
def test_modal_environment_cleanup_writes_namespaced_snapshot_key(tmp_path):
state = _install_modal_test_modules(tmp_path, snapshot_id="im-cleanup456")
snapshot_store = state["snapshot_store"]
modal_module = _load_module("tools.environments.modal", TOOLS_DIR / "environments" / "modal.py")
env = modal_module.ModalEnvironment(image="python:3.11", task_id="task-cleanup")
env.cleanup()
assert json.loads(snapshot_store.read_text()) == {"direct:task-cleanup": "im-cleanup456"}
def test_resolve_modal_image_uses_snapshot_ids_and_registry_images(tmp_path):
state = _install_modal_test_modules(tmp_path)
modal_module = _load_module("tools.environments.modal", TOOLS_DIR / "environments" / "modal.py")
snapshot_image = modal_module._resolve_modal_image("im-snapshot123")
registry_image = modal_module._resolve_modal_image("python:3.11")
assert snapshot_image == {"kind": "snapshot", "image_id": "im-snapshot123"}
assert registry_image == {"kind": "registry", "image": "python:3.11"}
assert state["from_id_calls"] == ["im-snapshot123"]
assert state["registry_calls"][0][0] == "python:3.11"
assert "ensurepip" in state["registry_calls"][0][1][0]

45
tests/tools/test_terminal_requirements.py
View File

@@ -8,9 +8,11 @@ def _clear_terminal_env(monkeypatch):
"""Remove terminal env vars that could affect requirements checks."""
keys = [
"TERMINAL_ENV",
"TERMINAL_MODAL_MODE",
"TERMINAL_SSH_HOST",
"TERMINAL_SSH_USER",
"MODAL_TOKEN_ID",
"MODAL_TOKEN_SECRET",
"HOME",
"USERPROFILE",
]
@@ -63,7 +65,7 @@ def test_modal_backend_without_token_or_config_logs_specific_error(monkeypatch,
monkeypatch.setenv("TERMINAL_ENV", "modal")
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
# Pretend swerex is installed
monkeypatch.setattr(terminal_tool_module, "is_managed_tool_gateway_ready", lambda _vendor: False)
monkeypatch.setattr(terminal_tool_module.importlib.util, "find_spec", lambda _name: object())
with caplog.at_level(logging.ERROR):
@@ -71,6 +73,45 @@ def test_modal_backend_without_token_or_config_logs_specific_error(monkeypatch,
assert ok is False
assert any(
"Modal backend selected but no MODAL_TOKEN_ID environment variable" in record.getMessage()
"Modal backend selected but no direct Modal credentials/config or managed tool gateway was found" in record.getMessage()
for record in caplog.records
)
def test_modal_backend_with_managed_gateway_does_not_require_direct_creds_or_minisweagent(monkeypatch, tmp_path):
_clear_terminal_env(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "modal")
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.setenv("TERMINAL_MODAL_MODE", "managed")
monkeypatch.setattr(terminal_tool_module, "is_managed_tool_gateway_ready", lambda _vendor: True)
monkeypatch.setattr(
terminal_tool_module,
"ensure_minisweagent_on_path",
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("should not be called")),
)
monkeypatch.setattr(
terminal_tool_module.importlib.util,
"find_spec",
lambda _name: (_ for _ in ()).throw(AssertionError("should not be called")),
)
assert terminal_tool_module.check_terminal_requirements() is True
def test_modal_backend_direct_mode_does_not_fall_back_to_managed(monkeypatch, caplog, tmp_path):
_clear_terminal_env(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "modal")
monkeypatch.setenv("TERMINAL_MODAL_MODE", "direct")
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.setattr(terminal_tool_module, "is_managed_tool_gateway_ready", lambda _vendor: True)
with caplog.at_level(logging.ERROR):
ok = terminal_tool_module.check_terminal_requirements()
assert ok is False
assert any(
"TERMINAL_MODAL_MODE=direct" in record.getMessage()
for record in caplog.records
)

27
tests/tools/test_terminal_tool_requirements.py
View File

@@ -26,3 +26,30 @@ class TestTerminalRequirements:
names = {tool["function"]["name"] for tool in tools}
assert "terminal" in names
assert {"read_file", "write_file", "patch", "search_files"}.issubset(names)
def test_terminal_and_execute_code_tools_resolve_for_managed_modal(self, monkeypatch, tmp_path):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("MODAL_TOKEN_ID", raising=False)
monkeypatch.delenv("MODAL_TOKEN_SECRET", raising=False)
monkeypatch.setattr(
terminal_tool_module,
"_get_env_config",
lambda: {"env_type": "modal", "modal_mode": "managed"},
)
monkeypatch.setattr(
terminal_tool_module,
"is_managed_tool_gateway_ready",
lambda _vendor: True,
)
monkeypatch.setattr(
terminal_tool_module,
"ensure_minisweagent_on_path",
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("should not be called")),
)
tools = get_tool_definitions(enabled_toolsets=["terminal", "code_execution"], quiet_mode=True)
names = {tool["function"]["name"] for tool in tools}
assert "terminal" in names
assert "execute_code" in names

4
tests/tools/test_transcription_tools.py
View File

@@ -231,6 +231,7 @@ class TestTranscribeGroq:
assert result["success"] is True
assert result["transcript"] == "hello world"
assert result["provider"] == "groq"
mock_client.close.assert_called_once()
def test_whitespace_stripped(self, monkeypatch, sample_wav):
monkeypatch.setenv("GROQ_API_KEY", "gsk-test")
@@ -272,6 +273,7 @@ class TestTranscribeGroq:
assert result["success"] is False
assert "API error" in result["error"]
mock_client.close.assert_called_once()
def test_permission_error(self, monkeypatch, sample_wav):
monkeypatch.setenv("GROQ_API_KEY", "gsk-test")
@@ -327,6 +329,7 @@ class TestTranscribeOpenAIExtended:
result = _transcribe_openai(sample_wav, "whisper-1")
assert result["transcript"] == "hello"
mock_client.close.assert_called_once()
def test_permission_error(self, monkeypatch, sample_wav):
monkeypatch.setenv("VOICE_TOOLS_OPENAI_KEY", "sk-test")
@@ -341,6 +344,7 @@ class TestTranscribeOpenAIExtended:
assert result["success"] is False
assert "Permission denied" in result["error"]
mock_client.close.assert_called_once()
class TestTranscribeLocalCommand:

249
tests/tools/test_web_tools_config.py
View File

@@ -5,12 +5,14 @@ Coverage:
constructor failure recovery, return value verification, edge cases.
_get_backend() — backend selection logic with env var combinations.
_get_parallel_client() — Parallel client configuration, singleton caching.
check_web_api_key() — unified availability check.
check_web_api_key() — unified availability check across all web backends.
"""
import importlib
import json
import os
import pytest
from unittest.mock import patch, MagicMock
from unittest.mock import patch, MagicMock, AsyncMock
class TestFirecrawlClientConfig:
@@ -20,14 +22,30 @@ class TestFirecrawlClientConfig:
"""Reset client and env vars before each test."""
import tools.web_tools
tools.web_tools._firecrawl_client = None
for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL"):
tools.web_tools._firecrawl_client_config = None
for key in (
"FIRECRAWL_API_KEY",
"FIRECRAWL_API_URL",
"FIRECRAWL_GATEWAY_URL",
"TOOL_GATEWAY_DOMAIN",
"TOOL_GATEWAY_SCHEME",
"TOOL_GATEWAY_USER_TOKEN",
):
os.environ.pop(key, None)
def teardown_method(self):
"""Reset client after each test."""
import tools.web_tools
tools.web_tools._firecrawl_client = None
for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL"):
tools.web_tools._firecrawl_client_config = None
for key in (
"FIRECRAWL_API_KEY",
"FIRECRAWL_API_URL",
"FIRECRAWL_GATEWAY_URL",
"TOOL_GATEWAY_DOMAIN",
"TOOL_GATEWAY_SCHEME",
"TOOL_GATEWAY_USER_TOKEN",
):
os.environ.pop(key, None)
# ── Configuration matrix ─────────────────────────────────────────
@@ -67,9 +85,152 @@ class TestFirecrawlClientConfig:
def test_no_config_raises_with_helpful_message(self):
"""Neither key nor URL → ValueError with guidance."""
with patch("tools.web_tools.Firecrawl"):
from tools.web_tools import _get_firecrawl_client
with pytest.raises(ValueError, match="FIRECRAWL_API_KEY"):
with patch("tools.web_tools._read_nous_access_token", return_value=None):
from tools.web_tools import _get_firecrawl_client
with pytest.raises(ValueError, match="FIRECRAWL_API_KEY"):
_get_firecrawl_client()
def test_tool_gateway_domain_builds_firecrawl_gateway_origin(self):
"""Shared gateway domain should derive the Firecrawl vendor hostname."""
with patch.dict(os.environ, {"TOOL_GATEWAY_DOMAIN": "nousresearch.com"}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch("tools.web_tools.Firecrawl") as mock_fc:
from tools.web_tools import _get_firecrawl_client
result = _get_firecrawl_client()
mock_fc.assert_called_once_with(
api_key="nous-token",
api_url="https://firecrawl-gateway.nousresearch.com",
)
assert result is mock_fc.return_value
def test_tool_gateway_scheme_can_switch_derived_gateway_origin_to_http(self):
"""Shared gateway scheme should allow local plain-http vendor hosts."""
with patch.dict(os.environ, {
"TOOL_GATEWAY_DOMAIN": "nousresearch.com",
"TOOL_GATEWAY_SCHEME": "http",
}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch("tools.web_tools.Firecrawl") as mock_fc:
from tools.web_tools import _get_firecrawl_client
result = _get_firecrawl_client()
mock_fc.assert_called_once_with(
api_key="nous-token",
api_url="http://firecrawl-gateway.nousresearch.com",
)
assert result is mock_fc.return_value
def test_invalid_tool_gateway_scheme_raises(self):
"""Unexpected shared gateway schemes should fail fast."""
with patch.dict(os.environ, {
"TOOL_GATEWAY_DOMAIN": "nousresearch.com",
"TOOL_GATEWAY_SCHEME": "ftp",
}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
from tools.web_tools import _get_firecrawl_client
with pytest.raises(ValueError, match="TOOL_GATEWAY_SCHEME"):
_get_firecrawl_client()
def test_explicit_firecrawl_gateway_url_takes_precedence(self):
"""An explicit Firecrawl gateway origin should override the shared domain."""
with patch.dict(os.environ, {
"FIRECRAWL_GATEWAY_URL": "https://firecrawl-gateway.localhost:3009/",
"TOOL_GATEWAY_DOMAIN": "nousresearch.com",
}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch("tools.web_tools.Firecrawl") as mock_fc:
from tools.web_tools import _get_firecrawl_client
_get_firecrawl_client()
mock_fc.assert_called_once_with(
api_key="nous-token",
api_url="https://firecrawl-gateway.localhost:3009",
)
def test_default_gateway_domain_targets_nous_production_origin(self):
"""Default gateway origin should point at the Firecrawl vendor hostname."""
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch("tools.web_tools.Firecrawl") as mock_fc:
from tools.web_tools import _get_firecrawl_client
_get_firecrawl_client()
mock_fc.assert_called_once_with(
api_key="nous-token",
api_url="https://firecrawl-gateway.nousresearch.com",
)
def test_direct_mode_is_preferred_over_tool_gateway(self):
"""Explicit Firecrawl config should win over the gateway fallback."""
with patch.dict(os.environ, {
"FIRECRAWL_API_KEY": "fc-test",
"TOOL_GATEWAY_DOMAIN": "nousresearch.com",
}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch("tools.web_tools.Firecrawl") as mock_fc:
from tools.web_tools import _get_firecrawl_client
_get_firecrawl_client()
mock_fc.assert_called_once_with(api_key="fc-test")
def test_nous_auth_token_respects_hermes_home_override(self, tmp_path):
"""Auth lookup should read from HERMES_HOME/auth.json, not ~/.hermes/auth.json."""
real_home = tmp_path / "real-home"
(real_home / ".hermes").mkdir(parents=True)
hermes_home = tmp_path / "hermes-home"
hermes_home.mkdir()
(hermes_home / "auth.json").write_text(json.dumps({
"providers": {
"nous": {
"access_token": "nous-token",
}
}
}))
with patch.dict(os.environ, {
"HOME": str(real_home),
"HERMES_HOME": str(hermes_home),
}, clear=False):
import tools.web_tools
importlib.reload(tools.web_tools)
assert tools.web_tools._read_nous_access_token() == "nous-token"
def test_check_auxiliary_model_re_resolves_backend_each_call(self):
"""Availability checks should not be pinned to module import state."""
import tools.web_tools
# Simulate the pre-fix import-time cache slot for regression coverage.
tools.web_tools.__dict__["_aux_async_client"] = None
with patch(
"tools.web_tools.get_async_text_auxiliary_client",
side_effect=[(None, None), (MagicMock(base_url="https://api.openrouter.ai/v1"), "test-model")],
):
assert tools.web_tools.check_auxiliary_model() is False
assert tools.web_tools.check_auxiliary_model() is True
@pytest.mark.asyncio
async def test_summarizer_re_resolves_backend_after_initial_unavailable_state(self):
"""Summarization should pick up a backend that becomes available later in-process."""
import tools.web_tools
tools.web_tools.__dict__["_aux_async_client"] = None
response = MagicMock()
response.choices = [MagicMock(message=MagicMock(content="summary text"))]
fake_client = MagicMock(base_url="https://api.openrouter.ai/v1")
fake_client.chat.completions.create = AsyncMock(return_value=response)
with patch(
"tools.web_tools.get_async_text_auxiliary_client",
side_effect=[(None, None), (fake_client, "test-model")],
):
assert tools.web_tools.check_auxiliary_model() is False
result = await tools.web_tools._call_summarizer_llm(
"Some content worth summarizing",
"Source: https://example.com\n\n",
None,
)
assert result == "summary text"
fake_client.chat.completions.create.assert_awaited_once()
# ── Singleton caching ────────────────────────────────────────────
@@ -117,9 +278,10 @@ class TestFirecrawlClientConfig:
"""FIRECRAWL_API_KEY='' with no URL → should raise."""
with patch.dict(os.environ, {"FIRECRAWL_API_KEY": ""}):
with patch("tools.web_tools.Firecrawl"):
from tools.web_tools import _get_firecrawl_client
with pytest.raises(ValueError):
_get_firecrawl_client()
with patch("tools.web_tools._read_nous_access_token", return_value=None):
from tools.web_tools import _get_firecrawl_client
with pytest.raises(ValueError):
_get_firecrawl_client()
class TestBackendSelection:
@@ -130,7 +292,16 @@ class TestBackendSelection:
setups.
"""
_ENV_KEYS = ("PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "TAVILY_API_KEY")
_ENV_KEYS = (
"PARALLEL_API_KEY",
"FIRECRAWL_API_KEY",
"FIRECRAWL_API_URL",
"FIRECRAWL_GATEWAY_URL",
"TOOL_GATEWAY_DOMAIN",
"TOOL_GATEWAY_SCHEME",
"TOOL_GATEWAY_USER_TOKEN",
"TAVILY_API_KEY",
)
def setup_method(self):
for key in self._ENV_KEYS:
@@ -276,10 +447,47 @@ class TestParallelClientConfig:
assert client1 is client2
class TestWebSearchErrorHandling:
"""Test suite for web_search_tool() error responses."""
def test_search_error_response_does_not_expose_diagnostics(self):
import tools.web_tools
firecrawl_client = MagicMock()
firecrawl_client.search.side_effect = RuntimeError("boom")
with patch("tools.web_tools._get_backend", return_value="firecrawl"), \
patch("tools.web_tools._get_firecrawl_client", return_value=firecrawl_client), \
patch("tools.interrupt.is_interrupted", return_value=False), \
patch.object(tools.web_tools._debug, "log_call") as mock_log_call, \
patch.object(tools.web_tools._debug, "save"):
result = json.loads(tools.web_tools.web_search_tool("test query", limit=3))
assert result == {"error": "Error searching web: boom"}
debug_payload = mock_log_call.call_args.args[1]
assert debug_payload["error"] == "Error searching web: boom"
assert "traceback" not in debug_payload["error"]
assert "exception_type" not in debug_payload["error"]
assert "config" not in result
assert "exception_type" not in result
assert "exception_chain" not in result
assert "traceback" not in result
class TestCheckWebApiKey:
"""Test suite for check_web_api_key() unified availability check."""
_ENV_KEYS = ("PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "TAVILY_API_KEY")
_ENV_KEYS = (
"PARALLEL_API_KEY",
"FIRECRAWL_API_KEY",
"FIRECRAWL_API_URL",
"FIRECRAWL_GATEWAY_URL",
"TOOL_GATEWAY_DOMAIN",
"TOOL_GATEWAY_SCHEME",
"TOOL_GATEWAY_USER_TOKEN",
"TAVILY_API_KEY",
)
def setup_method(self):
for key in self._ENV_KEYS:
@@ -329,3 +537,22 @@ class TestCheckWebApiKey:
}):
from tools.web_tools import check_web_api_key
assert check_web_api_key() is True
def test_tool_gateway_returns_true(self):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
from tools.web_tools import check_web_api_key
assert check_web_api_key() is True
def test_configured_backend_must_match_available_provider(self):
with patch("tools.web_tools._load_web_config", return_value={"backend": "parallel"}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch.dict(os.environ, {"FIRECRAWL_GATEWAY_URL": "http://127.0.0.1:3002"}, clear=False):
from tools.web_tools import check_web_api_key
assert check_web_api_key() is False
def test_configured_firecrawl_backend_accepts_managed_gateway(self):
with patch("tools.web_tools._load_web_config", return_value={"backend": "firecrawl"}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch.dict(os.environ, {"FIRECRAWL_GATEWAY_URL": "http://127.0.0.1:3002"}, clear=False):
from tools.web_tools import check_web_api_key
assert check_web_api_key() is True