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

opentui(phase3): launcher integration — HERMES_TUI_ENGINE dual-engine

Browse Source 
hermes --tui launches the native OpenTUI engine (Bun) when
HERMES_TUI_ENGINE=opentui (env) or display.tui_engine=opentui (config);
Ink stays the default and the shipping path is untouched.

- _resolve_tui_engine() (env > config > ink); refuses opentui on
  Windows/Termux (no Bun) -> falls back to ink with a notice.
- _make_opentui_argv() -> [bun, src/entry.real.tsx] (no build step).
- _bun_bin() with HERMES_BUN override.
- Branch at top of _make_tui_argv BEFORE _ensure_tui_node (Bun-only host
  must not bootstrap Node).
- Gate _launch_tui NODE_OPTIONS/--max-old-space-size on engine==ink (Bun
  is JSC; the V8 flag errors/ignores).

Verified end-to-end via tmux: real hermes --tui -> Bun -> OpenTUI ->
real Python gateway streamed a real reply. No-flag default still ink.
This commit is contained in:
alt-glitch
2026-06-08 11:11:54 +00:00
parent 24f74eb888
commit 2bd9c9b881
741 changed files with 17733 additions and 79889 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_approval.py
-38
View File
@@ -9,7 +9,6 @@ from types import SimpleNamespace
from unittest.mock import patch as mock_patch
import tools.approval as approval_module
from hermes_constants import get_hermes_home
from tools.approval import (
_get_approval_mode,
_smart_approve,
@@ -425,22 +424,6 @@ class TestHermesConfigWriteProtection:
dangerous, key, desc = detect_dangerous_command("sed --in-place 's/manual/off/' ~/.hermes/config.yaml")
assert dangerous is True
def test_sed_in_place_absolute_hermes_home_config(self):
config_path = get_hermes_home() / "config.yaml"
dangerous, key, desc = detect_dangerous_command(
f"sed -i 's/manual/off/' {config_path}"
)
assert dangerous is True
assert "hermes config" in desc.lower() or "in-place" in desc.lower()
def test_sed_in_place_absolute_hermes_home_env(self):
env_path = get_hermes_home() / ".env"
dangerous, key, desc = detect_dangerous_command(
f"sed -i 's/API_KEY=.*/API_KEY=x/' {env_path}"
)
assert dangerous is True
assert "hermes config" in desc.lower() or "in-place" in desc.lower()
def test_custom_hermes_home(self):
dangerous, key, desc = detect_dangerous_command("echo x | tee $HERMES_HOME/config.yaml")
assert dangerous is True
@@ -454,33 +437,12 @@ class TestHermesConfigWriteProtection:
assert dangerous is True
assert "in-place" in desc.lower() or "perl" in desc.lower()
def test_perl_in_place_absolute_hermes_home_config(self):
config_path = get_hermes_home() / "config.yaml"
dangerous, key, desc = detect_dangerous_command(
f"perl -i -pe 's/approvals.mode: on/approvals.mode: off/' {config_path}"
)
assert dangerous is True
assert "in-place" in desc.lower() or "perl" in desc.lower()
def test_ruby_in_place_config(self):
dangerous, key, desc = detect_dangerous_command(
"ruby -i -pe 'gsub(/manual/, \"off\")' ~/.hermes/config.yaml"
)
assert dangerous is True
def test_ruby_in_place_absolute_hermes_home_env(self):
env_path = get_hermes_home() / ".env"
dangerous, key, desc = detect_dangerous_command(
f"ruby -i -pe 'gsub(/API_KEY=.*/, \"API_KEY=x\")' {env_path}"
)
assert dangerous is True
def test_regular_absolute_config_path_still_uses_project_rule(self):
dangerous, key, desc = detect_dangerous_command(
"sed -i 's/a/b/' /srv/app/config.yaml"
)
assert dangerous is False
def test_perl_in_place_env(self):
dangerous, key, desc = detect_dangerous_command(
"perl -i -pe 's/SECRET=old/SECRET=new/' ~/.hermes/.env"
tests/tools/test_blueprints.py
-188
View File
@@ -1,188 +0,0 @@
"""Tests for the blueprints layer (skill frontmatter <-> cron automation bridge).
A blueprint is a skill with a metadata.hermes.blueprint block. These verify parsing,
the create-job bridge, and the export round-trip without touching the real
cron store.
"""
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
from tools.blueprints import (
BlueprintError,
BlueprintSpec,
create_blueprint_job,
export_blueprint,
parse_blueprint,
blueprint_spec_for_installed,
)
BLUEPRINT_SKILL = """---
name: morning-brief
description: Summarize unread email and calendar every morning.
version: 1.0.0
metadata:
hermes:
tags: [blueprint, email]
blueprint:
schedule: "0 8 * * *"
deliver: telegram
prompt: "Summarize my unread email and today's calendar."
---
# Morning Brief
Every morning, gather unread email and the day's calendar and send a digest.
"""
PLAIN_SKILL = """---
name: not-a-blueprint
description: Just a regular skill.
metadata:
hermes:
tags: [misc]
---
# Not a blueprint
"""
MALFORMED_BLUEPRINT = """---
name: broken
description: Blueprint with no schedule.
metadata:
hermes:
blueprint:
deliver: origin
---
# Broken
"""
class TestParseBlueprint:
def test_parses_full_blueprint(self):
spec = parse_blueprint(BLUEPRINT_SKILL)
assert spec is not None
assert spec.skill_name == "morning-brief"
assert spec.schedule == "0 8 * * *"
assert spec.deliver == "telegram"
assert spec.prompt is not None and spec.prompt.startswith("Summarize")
def test_plain_skill_is_not_a_blueprint(self):
assert parse_blueprint(PLAIN_SKILL) is None
def test_no_frontmatter_is_not_a_blueprint(self):
assert parse_blueprint("just some text, no frontmatter") is None
def test_missing_schedule_raises(self):
with pytest.raises(BlueprintError):
parse_blueprint(MALFORMED_BLUEPRINT)
def test_blueprint_not_mapping_raises(self):
bad = "---\nname: x\nmetadata:\n hermes:\n blueprint: not-a-dict\n---\n\nbody"
with pytest.raises(BlueprintError):
parse_blueprint(bad)
def test_deliver_defaults_to_origin(self):
skill = (
"---\nname: r\ndescription: d\nmetadata:\n hermes:\n"
' blueprint:\n schedule: "every 1h"\n---\n\nbody'
)
spec = parse_blueprint(skill)
assert spec is not None
assert spec.deliver == "origin"
class TestBlueprintSpecForInstalled:
def test_finds_and_parses_installed_blueprint(self, tmp_path):
skills_dir = tmp_path / "skills"
rec_dir = skills_dir / "productivity" / "morning-brief"
rec_dir.mkdir(parents=True)
(rec_dir / "SKILL.md").write_text(BLUEPRINT_SKILL, encoding="utf-8")
with patch("tools.skills_hub.SKILLS_DIR", skills_dir):
spec = blueprint_spec_for_installed("morning-brief")
assert spec is not None
assert spec.schedule == "0 8 * * *"
def test_missing_skill_returns_none(self, tmp_path):
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
with patch("tools.skills_hub.SKILLS_DIR", skills_dir):
assert blueprint_spec_for_installed("nope") is None
def test_plain_skill_returns_none(self, tmp_path):
skills_dir = tmp_path / "skills"
d = skills_dir / "misc" / "not-a-blueprint"
d.mkdir(parents=True)
(d / "SKILL.md").write_text(PLAIN_SKILL, encoding="utf-8")
with patch("tools.skills_hub.SKILLS_DIR", skills_dir):
assert blueprint_spec_for_installed("not-a-blueprint") is None
class TestCreateBlueprintJob:
def test_bridges_to_create_job(self):
spec = parse_blueprint(BLUEPRINT_SKILL)
assert spec is not None
captured = {}
def fake_create_job(**kwargs):
captured.update(kwargs)
return {"id": "abc123", **kwargs}
with patch("cron.jobs.create_job", fake_create_job):
job = create_blueprint_job(spec, origin={"platform": "telegram"})
assert captured["schedule"] == "0 8 * * *"
assert captured["skills"] == ["morning-brief"]
assert captured["deliver"] == "telegram"
assert captured["prompt"].startswith("Summarize")
assert job["id"] == "abc123"
class TestExportBlueprint:
def test_round_trips_job_to_skill_md(self):
job = {
"name": "My Morning Brief",
"schedule_display": "0 8 * * *",
"skills": ["morning-brief"],
"deliver": "telegram",
"prompt": "Summarize my unread email.",
}
md = export_blueprint(job, "# Morning Brief\n\nDoes the morning digest.")
# The exported SKILL.md must itself parse back as a blueprint.
spec = parse_blueprint(md)
assert spec is not None
assert spec.schedule == "0 8 * * *"
assert spec.deliver == "telegram"
# Name is sanitized to a valid skill identifier.
assert spec.skill_name == "my-morning-brief"
def test_export_has_blueprint_tag(self):
job = {"name": "x", "schedule_display": "every 2h", "skills": ["x"]}
md = export_blueprint(job, "body")
assert "blueprint" in md
assert "automation" in md
def test_export_interval_job_without_display(self):
# Regression: parse_schedule stores interval periods as "minutes" —
# exporting a job with only the parsed schedule dict must round-trip
# the real interval, not fall back to the daily default.
job = {
"name": "poller",
"schedule": {"kind": "interval", "minutes": 30},
"skills": ["poller"],
}
md = export_blueprint(job, "body")
spec = parse_blueprint(md)
assert spec is not None
assert spec.schedule == "every 30m"
job["schedule"] = {"kind": "interval", "minutes": 120}
spec = parse_blueprint(export_blueprint(job, "body"))
assert spec is not None
assert spec.schedule == "every 2h"
tests/tools/test_command_guards.py
-62
View File
@@ -285,65 +285,3 @@ class TestProgrammingErrorsPropagateFromWrapper:
os.environ["HERMES_INTERACTIVE"] = "1"
with pytest.raises(AttributeError, match="bug in wrapper"):
check_all_command_guards("echo hello", "local")
# ---------------------------------------------------------------------------
# Gateway (TUI / desktop) approval notify payload carries allow_permanent
# ---------------------------------------------------------------------------
class TestGatewayApprovalAllowPermanent:
"""The gateway emits the approval prompt to the renderer via the notify
payload (TUI/desktop both consume it). It must carry ``allow_permanent``
so the UI doesn't offer a permanent allow the backend would silently
downgrade to session scope for tirith content-security findings.
"""
def _capture_gateway_payload(self, command, session_key):
"""Run the gateway approval path, denying inline, and return the
single notify payload the renderer would have received."""
from tools.approval import (
register_gateway_notify,
resolve_gateway_approval,
unregister_gateway_notify,
)
captured = []
def notify(data):
captured.append(dict(data))
# The notify fires synchronously before _await_gateway_decision
# blocks, so resolving here releases the wait without a thread.
resolve_gateway_approval(session_key, "deny")
register_gateway_notify(session_key, notify)
token = set_current_session_key(session_key)
os.environ["HERMES_GATEWAY_SESSION"] = "1"
os.environ["HERMES_EXEC_ASK"] = "1"
os.environ["HERMES_SESSION_KEY"] = session_key
try:
check_all_command_guards(command, "local")
finally:
os.environ.pop("HERMES_GATEWAY_SESSION", None)
os.environ.pop("HERMES_EXEC_ASK", None)
os.environ.pop("HERMES_SESSION_KEY", None)
reset_current_session_key(token)
unregister_gateway_notify(session_key)
assert len(captured) == 1
return captured[0]
def test_dangerous_only_allows_permanent(self):
"""No tirith warning → permanent allow is offered."""
payload = self._capture_gateway_payload("rm -rf /important", "gw-allow-perm")
assert payload["command"] == "rm -rf /important"
assert payload["allow_permanent"] is True
@patch(_TIRITH_PATCH,
return_value=_tirith_result("warn",
[{"rule_id": "shortened_url"}],
"shortened URL detected"))
def test_tirith_warning_disallows_permanent(self, mock_tirith):
"""tirith content-security warning → permanent allow is withheld so the
renderer hides "Always allow"."""
payload = self._capture_gateway_payload("curl https://bit.ly/abc", "gw-no-perm")
assert payload["allow_permanent"] is False
tests/tools/test_computer_use.py
+2 -32
View File
@@ -338,7 +338,7 @@ class TestCaptureResponse:
from tools.computer_use.backend import CaptureResult
from tools.computer_use import tool as cu_tool
fake_png = "iVBORw0KGgoAAAANSUhEUgAAAAgAAAAICAYAAADED76LAAAADUlEQVR4nGNgGAUgAAABCAABgukLHQAAAABJRU5ErkJggg=="
fake_png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
class FakeBackend:
def start(self): pass
@@ -372,41 +372,11 @@ class TestCaptureResponse:
assert any(p.get("type") == "image_url" for p in out["content"])
assert any(p.get("type") == "text" for p in out["content"])
def test_capture_tiny_image_returns_text_json(self):
"""Providers can reject <8px images, so placeholders must be omitted."""
from tools.computer_use.backend import CaptureResult, UIElement
from tools.computer_use import tool as cu_tool
tiny_png = "iVBORw0KGgoAAAANSUhEUgAAAAIAAAACCAYAAABytg0kAAAAC0lEQVR4nGNgQAcAABIAAXfx+gAAAAAASUVORK5CYII="
cap = CaptureResult(
mode="som",
width=0,
height=0,
png_b64=tiny_png,
elements=[
UIElement(index=1, role="AXButton", label="Continue", bounds=(10, 20, 30, 30)),
],
app="Safari",
window_title="Example",
png_bytes_len=68,
)
with patch.object(cu_tool, "_should_route_through_aux_vision",
return_value=False):
out = cu_tool._capture_response(cap)
parsed = json.loads(out)
assert parsed["width"] == 2
assert parsed["height"] == 2
assert "screenshot omitted" in parsed["summary"]
assert parsed["elements"][0]["label"] == "Continue"
def test_capture_som_with_elements_formats_index(self):
from tools.computer_use.backend import CaptureResult, UIElement
from tools.computer_use import tool as cu_tool
fake_png = "iVBORw0KGgoAAAANSUhEUgAAAAgAAAAICAYAAADED76LAAAADUlEQVR4nGNgGAUgAAABCAABgukLHQAAAABJRU5ErkJggg=="
fake_png = "iVBORw0KGgo="
class FakeBackend:
def start(self): pass
tests/tools/test_computer_use_capture_routing.py
+3 -3
View File
@@ -33,10 +33,10 @@ import pytest
# Fixtures / helpers
# ---------------------------------------------------------------------------
# 8×8 PNG (transparent) — minimal provider-acceptable bytes that decode cleanly.
# 1×1 PNG (transparent) — minimal bytes that decode cleanly.
_PNG_B64 = (
"iVBORw0KGgoAAAANSUhEUgAAAAgAAAAICAYAAADED76LAAAADUlEQVR4nG"
"NgGAUgAAABCAABgukLHQAAAABJRU5ErkJggg=="
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42m"
"NkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
)
# 1×1 JPEG — used to verify mime detection works for either stream type.
tests/tools/test_cronjob_tools.py
-51
View File
@@ -452,54 +452,3 @@ class TestUnifiedCronjobTool:
assert updated["success"] is True
stored = get_job(created["job_id"])
assert stored["deliver"] == "telegram"
# =========================================================================
# Per-job model/provider override resolution
# =========================================================================
from tools.cronjob_tools import _resolve_model_override # noqa: E402
class TestResolveModelOverride:
"""`_resolve_model_override` must not silently hijack a job that meant to
use a configured custom endpoint (e.g. ``providers.custom`` → cliproxy).
Regression for cron jobs with ``provider: "custom"`` falling back to codex.
"""
def test_keeps_bare_custom_when_a_named_entry_exists(self, monkeypatch):
import hermes_cli.runtime_provider as rp_mod
monkeypatch.setattr(rp_mod, "has_named_custom_provider", lambda name: True)
provider, model = _resolve_model_override(
{"provider": "custom", "model": "gpt-5.4"}
)
assert provider == "custom"
assert model == "gpt-5.4"
def test_pins_main_provider_when_bare_custom_unresolvable(self, monkeypatch):
import hermes_cli.config as cfg_mod
import hermes_cli.runtime_provider as rp_mod
monkeypatch.setattr(rp_mod, "has_named_custom_provider", lambda name: False)
monkeypatch.setattr(
cfg_mod, "load_config", lambda: {"model": {"provider": "openai-codex"}}
)
provider, model = _resolve_model_override(
{"provider": "custom", "model": "gpt-5.4"}
)
# No matching custom entry → fall back to pinning the main provider.
assert provider == "openai-codex"
assert model == "gpt-5.4"
def test_keeps_explicit_custom_name_unchanged(self, monkeypatch):
import hermes_cli.runtime_provider as rp_mod
# Even if the resolver claims no entry, the canonical "custom:<name>"
# form is never stripped or pinned.
monkeypatch.setattr(rp_mod, "has_named_custom_provider", lambda name: False)
provider, model = _resolve_model_override(
{"provider": "custom:cliproxy", "model": "gpt-5.4"}
)
assert provider == "custom:cliproxy"
assert model == "gpt-5.4"
tests/tools/test_docker_environment.py
-1
View File
@@ -109,7 +109,6 @@ def test_ensure_docker_available_uses_resolved_executable(monkeypatch):
"capture_output": True,
"text": True,
"timeout": 5,
"stdin": subprocess.DEVNULL,
})
]
tests/tools/test_dockerfile_pid1_reaping.py
-25
View File
@@ -172,31 +172,6 @@ def test_dockerfile_preinstalls_gateway_messaging_dependencies(dockerfile_text):
)
def test_dockerfile_preinstalls_matrix_dependencies(dockerfile_text):
sync_steps = [
step for step in _run_steps(dockerfile_text)
if "uv sync" in step and "--no-install-project" in step
]
assert sync_steps, "Dockerfile must install Python dependencies with uv sync"
assert any("--extra matrix" in step for step in sync_steps), (
"Published Docker images must preload the [matrix] extra so the "
"Matrix gateway has mautrix[encryption]/python-olm available at "
"runtime instead of relying on first-boot lazy installation into "
"the container venv (#30399)."
)
def test_dockerfile_installs_matrix_native_build_dependencies(dockerfile_text):
instructions = _instruction_text(dockerfile_text)
for package in ("libolm-dev", "cmake", "g++", "make"):
assert package in instructions, (
"Docker image must include native build dependencies needed by "
f"python-olm when preinstalling the [matrix] extra (#30399): {package}"
)
def test_dockerfile_preinstalls_hindsight_memory_dependency(dockerfile_text):
sync_steps = [
step for step in _run_steps(dockerfile_text)
tests/tools/test_local_env_blocklist.py
+11 -77
View File
@@ -388,86 +388,20 @@ class TestSanePathIncludesHomebrew:
assert "/opt/homebrew/sbin" in _SANE_PATH
def test_make_run_env_appends_homebrew_on_minimal_path(self):
"""When PATH is minimal, _make_run_env appends missing sane entries."""
from tools.environments.local import _SANE_PATH, _make_run_env
"""When PATH is minimal (no /usr/bin), _make_run_env should append
_SANE_PATH which now includes Homebrew dirs."""
from tools.environments.local import _make_run_env
minimal_env = {"PATH": "/some/custom/bin"}
with patch.dict(os.environ, minimal_env, clear=True):
result = _make_run_env({})
path_entries = result["PATH"].split(":")
assert path_entries[0] == "/some/custom/bin"
for entry in _SANE_PATH.split(":"):
assert entry in path_entries
assert "/opt/homebrew/bin" in result["PATH"]
assert "/opt/homebrew/sbin" in result["PATH"]
def test_make_run_env_fills_missing_homebrew_when_usr_bin_present(self):
"""macOS launchd PATH can include /usr/bin while missing Homebrew."""
def test_make_run_env_does_not_duplicate_on_full_path(self):
"""When PATH already has /usr/bin, _make_run_env should not append."""
from tools.environments.local import _make_run_env
launchd_env = {"PATH": "/usr/local/bin:/usr/bin:/bin"}
with patch.dict(os.environ, launchd_env, clear=True):
full_env = {"PATH": "/usr/bin:/bin"}
with patch.dict(os.environ, full_env, clear=True):
result = _make_run_env({})
path_entries = result["PATH"].split(":")
assert "/opt/homebrew/bin" in path_entries
assert "/opt/homebrew/sbin" in path_entries
def test_make_run_env_does_not_duplicate_existing_sane_entries(self):
from tools.environments.local import _make_run_env
existing_env = {"PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin"}
with patch.dict(os.environ, existing_env, clear=True):
result = _make_run_env({})
path_entries = result["PATH"].split(":")
assert path_entries.count("/opt/homebrew/bin") == 1
assert path_entries.count("/usr/local/bin") == 1
assert path_entries.count("/usr/bin") == 1
def test_make_run_env_real_launchd_path_gains_homebrew(self):
"""The literal macOS launchd PATH is the production trigger for #35613."""
from tools.environments.local import _make_run_env
launchd_env = {"PATH": "/usr/bin:/bin:/usr/sbin:/sbin"}
with patch.dict(os.environ, launchd_env, clear=True):
result = _make_run_env({})
path_entries = result["PATH"].split(":")
assert "/opt/homebrew/bin" in path_entries
assert "/opt/homebrew/sbin" in path_entries
# Original entries keep their leading precedence.
assert path_entries[:4] == ["/usr/bin", "/bin", "/usr/sbin", "/sbin"]
def test_make_run_env_collapses_duplicate_caller_entries(self):
"""Duplicates already present in the caller PATH are de-duplicated."""
from tools.environments.local import _make_run_env
dup_env = {"PATH": "/usr/bin:/usr/bin:/custom/bin:/custom/bin:/bin"}
with patch.dict(os.environ, dup_env, clear=True):
result = _make_run_env({})
path_entries = result["PATH"].split(":")
assert path_entries.count("/usr/bin") == 1
assert path_entries.count("/custom/bin") == 1
# First-occurrence order is preserved for the caller entries.
assert path_entries[:3] == ["/usr/bin", "/custom/bin", "/bin"]
def test_make_run_env_strips_empty_path_entries(self):
"""Leading/trailing/double colons (== CWD on POSIX) are dropped."""
from tools.environments.local import _make_run_env
empty_env = {"PATH": "/usr/bin::/bin:"}
with patch.dict(os.environ, empty_env, clear=True):
result = _make_run_env({})
path_entries = result["PATH"].split(":")
assert "" not in path_entries
assert "/usr/bin" in path_entries
assert "/opt/homebrew/bin" in path_entries
def test_make_run_env_leaves_windows_path_unchanged(self, monkeypatch):
from tools.environments import local as local_mod
from tools.environments.local import _make_run_env
windows_env = {"PATH": r"C:\Windows\System32;C:\Program Files\Git\bin"}
monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
with patch.dict(os.environ, windows_env, clear=True):
result = _make_run_env({})
assert result["PATH"] == windows_env["PATH"]
def test_make_run_env_preserves_windows_mixed_case_path_key(self, monkeypatch):
from tools.environments import local as local_mod
from tools.environments.local import _make_run_env
windows_env = {"Path": r"C:\Windows\System32;C:\Program Files\Git\bin"}
monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
with patch.object(local_mod.os, "environ", windows_env):
result = _make_run_env({})
assert result["Path"] == windows_env["Path"]
assert "PATH" not in result
# Should keep existing PATH unchanged
assert result["PATH"] == "/usr/bin:/bin"
tests/tools/test_local_shell_init.py
-43
View File
@@ -190,49 +190,6 @@ class TestSnapshotEndToEnd:
"""Spin up a real LocalEnvironment and confirm the snapshot sources
extra init files."""
def test_exported_env_changes_persist_between_commands(self, tmp_path):
env = LocalEnvironment(cwd=str(tmp_path), timeout=15)
try:
first = env.execute(
'export HERMES_SESSION_ENV_PROBE="sticky"; '
'export PATH="/tmp/hermes-session-bin:$PATH"; '
'echo "first=$HERMES_SESSION_ENV_PROBE"'
)
second = env.execute(
'echo "second=$HERMES_SESSION_ENV_PROBE"; echo "PATH=$PATH"'
)
finally:
env.cleanup()
assert first["returncode"] == 0
assert second["returncode"] == 0
assert "first=sticky" in first.get("output", "")
output = second.get("output", "")
assert "second=sticky" in output
assert "/tmp/hermes-session-bin" in output
def test_venv_style_activation_persists_between_commands(self, tmp_path):
venv_bin = tmp_path / ".venv" / "bin"
venv_bin.mkdir(parents=True)
activate = venv_bin / "activate"
activate.write_text(
f'export VIRTUAL_ENV="{tmp_path / ".venv"}"\n'
f'export PATH="{venv_bin}:$PATH"\n'
)
env = LocalEnvironment(cwd=str(tmp_path), timeout=15)
try:
first = env.execute('source .venv/bin/activate; echo "venv=$VIRTUAL_ENV"')
second = env.execute('echo "venv=$VIRTUAL_ENV"; echo "PATH=$PATH"')
finally:
env.cleanup()
assert first["returncode"] == 0
assert second["returncode"] == 0
output = second.get("output", "")
assert f"venv={tmp_path / '.venv'}" in output
assert str(venv_bin) in output
def test_snapshot_picks_up_init_file_exports(self, tmp_path, monkeypatch):
init_file = tmp_path / "custom-init.sh"
init_file.write_text(
tests/tools/test_mcp_capability_gating.py
-158
View File
@@ -1,158 +0,0 @@
"""Tests for capability-gated MCP tool discovery and keepalive.
Prompt-only / resource-only MCP servers do not implement the ``tools/*``
request family. Per the MCP spec, ``InitializeResult.capabilities.tools``
is non-None iff the server supports it. Before this fix, Hermes always
called ``tools/list`` during discovery and as the keepalive probe — both
raised ``McpError(-32601 Method not found)`` against such servers, so a
prompt-only server could never stay connected.
Ported from anomalyco/opencode#31271.
"""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from tools.mcp_tool import MCPServerTask
def _caps(tools=None, prompts=None, resources=None):
"""Build a fake InitializeResult with the given capability sub-objects."""
return SimpleNamespace(
capabilities=SimpleNamespace(tools=tools, prompts=prompts, resources=resources)
)
class TestAdvertisesTools:
def test_true_when_tools_capability_present(self):
task = MCPServerTask("test")
task.initialize_result = _caps(tools=SimpleNamespace(listChanged=True))
assert task._advertises_tools() is True
def test_false_for_prompt_only_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(prompts=SimpleNamespace(listChanged=None))
assert task._advertises_tools() is False
def test_false_for_resource_only_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(resources=SimpleNamespace())
assert task._advertises_tools() is False
def test_legacy_fallback_no_initialize_result(self):
"""No captured capabilities → preserve old always-list_tools behavior."""
task = MCPServerTask("test")
assert task.initialize_result is None
assert task._advertises_tools() is True
def test_legacy_fallback_no_capabilities_attr(self):
task = MCPServerTask("test")
task.initialize_result = SimpleNamespace() # no .capabilities
assert task._advertises_tools() is True
@pytest.mark.asyncio
class TestDiscoverToolsGating:
async def test_skips_list_tools_for_prompt_only_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(prompts=SimpleNamespace())
task.session = SimpleNamespace(list_tools=AsyncMock())
task._tools = ["stale"]
await task._discover_tools()
task.session.list_tools.assert_not_called()
assert task._tools == []
async def test_calls_list_tools_for_tool_capable_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(tools=SimpleNamespace())
fake_tool = SimpleNamespace(name="echo")
task.session = SimpleNamespace(
list_tools=AsyncMock(return_value=SimpleNamespace(tools=[fake_tool]))
)
await task._discover_tools()
task.session.list_tools.assert_awaited_once()
assert task._tools == [fake_tool]
async def test_legacy_fallback_still_calls_list_tools(self):
task = MCPServerTask("test")
task.session = SimpleNamespace(
list_tools=AsyncMock(return_value=SimpleNamespace(tools=[]))
)
await task._discover_tools()
task.session.list_tools.assert_awaited_once()
@pytest.mark.asyncio
class TestRefreshToolsGating:
async def test_refresh_noop_for_prompt_only_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(prompts=SimpleNamespace())
task.session = SimpleNamespace(list_tools=AsyncMock())
await task._refresh_tools()
task.session.list_tools.assert_not_called()
@pytest.mark.asyncio
class TestKeepaliveProbe:
async def _run_one_keepalive_cycle(self, task):
"""Drive _wait_for_lifecycle_event through exactly one keepalive
timeout, then fire shutdown so it returns."""
real_wait = asyncio.wait
cycles = {"n": 0}
async def fake_wait(tasks, timeout=None, return_when=None):
cycles["n"] += 1
if cycles["n"] == 1:
# Simulate keepalive timeout: nothing completed.
return set(), set(tasks)
# Second cycle: let shutdown win.
task._shutdown_event.set()
return await real_wait(
tasks, timeout=0.5, return_when=return_when or asyncio.FIRST_COMPLETED
)
import tools.mcp_tool as mcp_mod
orig = mcp_mod.asyncio.wait
mcp_mod.asyncio.wait = fake_wait
try:
return await task._wait_for_lifecycle_event()
finally:
mcp_mod.asyncio.wait = orig
async def test_keepalive_uses_ping_for_prompt_only_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(prompts=SimpleNamespace())
task.session = SimpleNamespace(
list_tools=AsyncMock(),
send_ping=AsyncMock(),
)
reason = await self._run_one_keepalive_cycle(task)
assert reason == "shutdown"
task.session.send_ping.assert_awaited_once()
task.session.list_tools.assert_not_called()
async def test_keepalive_uses_list_tools_for_tool_capable_server(self):
task = MCPServerTask("test")
task.initialize_result = _caps(tools=SimpleNamespace())
task.session = SimpleNamespace(
list_tools=AsyncMock(return_value=SimpleNamespace(tools=[])),
send_ping=AsyncMock(),
)
reason = await self._run_one_keepalive_cycle(task)
assert reason == "shutdown"
task.session.list_tools.assert_awaited_once()
task.session.send_ping.assert_not_called()
tests/tools/test_mcp_loop_profile_override.py
-139
View File
@@ -1,139 +0,0 @@
"""Regression tests for HERMES_HOME override propagation onto the MCP loop.
Tasks scheduled via run_coroutine_threadsafe are created inside the MCP
event-loop thread, so they copy THAT thread's context — not the scheduling
thread's. A per-request profile scope (dashboard ?profile= endpoints, e.g.
the MCP "Test server" probe) would silently vanish for anything resolving
get_hermes_home() inside the coroutine, most visibly OAuth token-store
paths. _run_on_mcp_loop now wraps scheduled coroutines with the caller's
override (mcp_tool._wrap_with_home_override).
"""
import os
import pytest
@pytest.fixture
def mcp_loop():
import tools.mcp_tool as mcp_tool
mcp_tool._ensure_mcp_loop()
yield mcp_tool
mcp_tool._stop_mcp_loop()
def test_override_propagates_to_mcp_loop(tmp_path, monkeypatch, mcp_loop):
from hermes_constants import (
get_hermes_home,
reset_hermes_home_override,
set_hermes_home_override,
)
process_home = tmp_path / "proc-home"
profile_home = tmp_path / "profile-home"
process_home.mkdir()
profile_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(process_home))
async def read_home():
return str(get_hermes_home())
# Unscoped: the loop task sees the process home.
assert mcp_loop._run_on_mcp_loop(read_home(), timeout=10) == str(process_home)
# Scoped: the caller's override must reach the loop task.
token = set_hermes_home_override(str(profile_home))
try:
assert mcp_loop._run_on_mcp_loop(read_home(), timeout=10) == str(profile_home)
# Factory form must be wrapped too.
assert mcp_loop._run_on_mcp_loop(lambda: read_home(), timeout=10) == str(
profile_home
)
finally:
reset_hermes_home_override(token)
# The loop thread's default context is untouched afterwards.
assert mcp_loop._run_on_mcp_loop(read_home(), timeout=10) == str(process_home)
def test_oauth_token_paths_follow_override(tmp_path, monkeypatch, mcp_loop):
"""The actual symptom path: HermesTokenStorage resolving inside the
probe's MCP-loop coroutine must land in the selected profile's
mcp-tokens dir, not the process home's."""
from hermes_constants import (
reset_hermes_home_override,
set_hermes_home_override,
)
process_home = tmp_path / "proc-home"
profile_home = tmp_path / "profile-home"
process_home.mkdir()
profile_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(process_home))
async def token_path():
from tools.mcp_oauth import HermesTokenStorage
return str(HermesTokenStorage("probe-srv")._tokens_path())
token = set_hermes_home_override(str(profile_home))
try:
path = mcp_loop._run_on_mcp_loop(token_path(), timeout=10)
finally:
reset_hermes_home_override(token)
assert path.startswith(str(profile_home))
assert os.path.join("mcp-tokens", "probe-srv.json") in path
def test_concurrent_scopes_do_not_interfere(tmp_path, monkeypatch, mcp_loop):
"""Two threads carrying DIFFERENT overrides scheduling onto the same
loop must each see their own home — the wrapper is task-local."""
import threading
from hermes_constants import (
get_hermes_home,
reset_hermes_home_override,
set_hermes_home_override,
)
process_home = tmp_path / "proc-home"
home_a = tmp_path / "profile-a"
home_b = tmp_path / "profile-b"
for h in (process_home, home_a, home_b):
h.mkdir()
monkeypatch.setenv("HERMES_HOME", str(process_home))
async def read_home():
return str(get_hermes_home())
results: dict = {}
def scoped_call(key, home):
token = set_hermes_home_override(str(home))
try:
results[key] = mcp_loop._run_on_mcp_loop(read_home(), timeout=10)
finally:
reset_hermes_home_override(token)
threads = [
threading.Thread(target=scoped_call, args=("a", home_a)),
threading.Thread(target=scoped_call, args=("b", home_b)),
]
for t in threads:
t.start()
for t in threads:
t.join(timeout=15)
assert results == {"a": str(home_a), "b": str(home_b)}
def test_wrap_is_noop_without_override(mcp_loop):
"""No active override → the coroutine passes through unwrapped."""
async def trivial():
return 42
coro = trivial()
wrapped = mcp_loop._wrap_with_home_override(coro)
assert wrapped is coro
coro.close()
tests/tools/test_mcp_stability.py
+2 -9
View File
@@ -321,19 +321,12 @@ class TestStdioPgroupReaping:
psutil = pytest.importorskip("psutil")
# Grandchild: sleep forever, write its pid then wait. The pid file
# is written to a temp path and os.replace()d into place so the
# polling reader below can never observe a created-but-empty file
# (CI flake: int('') ValueError when the reader won the race between
# open('w') creating the file and write() filling it).
# Grandchild: sleep forever, write its pid then wait.
grandchild_pid_file = tmp_path / "grandchild.pid"
grandchild_script = tmp_path / "grandchild.py"
grandchild_script.write_text(
"import os, sys, time\n"
f"tmp = {str(grandchild_pid_file)!r} + '.tmp'\n"
"with open(tmp, 'w') as f:\n"
" f.write(str(os.getpid()))\n"
f"os.replace(tmp, {str(grandchild_pid_file)!r})\n"
f"open({str(grandchild_pid_file)!r}, 'w').write(str(os.getpid()))\n"
"while True:\n"
" time.sleep(0.5)\n"
)
tests/tools/test_mcp_tool.py
-77
View File
@@ -82,56 +82,6 @@ class TestLoadMCPConfig:
assert result == {}
class TestMCPStatus:
def test_status_distinguishes_configured_connecting_failed_and_disabled(
self, monkeypatch
):
import tools.mcp_tool as mcp_tool
monkeypatch.setattr(
mcp_tool,
"_load_mcp_config",
lambda: {
"configured": {"command": "docker", "args": ["mcp", "gateway", "run"]},
"connecting": {"command": "slow-mcp"},
"failed": {"command": "bad-mcp"},
"disabled": {"command": "off-mcp", "enabled": False},
},
)
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_connecting = set(mcp_tool._server_connecting)
saved_errors = dict(mcp_tool._server_connect_errors)
mcp_tool._servers.clear()
mcp_tool._server_connecting.clear()
mcp_tool._server_connect_errors.clear()
mcp_tool._server_connecting.add("connecting")
mcp_tool._server_connect_errors["failed"] = "Connection closed"
try:
statuses = {
entry["name"]: entry
for entry in mcp_tool.get_mcp_status()
}
finally:
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_connecting.clear()
mcp_tool._server_connecting.update(saved_connecting)
mcp_tool._server_connect_errors.clear()
mcp_tool._server_connect_errors.update(saved_errors)
assert statuses["configured"]["status"] == "configured"
assert statuses["configured"]["connected"] is False
assert statuses["configured"]["disabled"] is False
assert statuses["connecting"]["status"] == "connecting"
assert statuses["failed"]["status"] == "failed"
assert statuses["failed"]["error"] == "Connection closed"
assert statuses["disabled"]["status"] == "disabled"
assert statuses["disabled"]["disabled"] is True
# ---------------------------------------------------------------------------
# Schema conversion
# ---------------------------------------------------------------------------
@@ -1428,33 +1378,6 @@ class TestBuildSafeEnv:
assert "DATABASE_URL" not in result
assert "API_SECRET" not in result
def test_windows_location_vars_passed_without_secrets(self):
"""Windows launcher tools need location vars, but secrets stay filtered."""
from tools.mcp_tool import _build_safe_env
fake_env = {
"PATH": r"C:\Windows\System32",
"ProgramFiles": r"C:\Program Files",
"ProgramData": r"C:\ProgramData",
"ProgramW6432": r"C:\Program Files",
"LOCALAPPDATA": r"C:\Users\alice\AppData\Local",
"APPDATA": r"C:\Users\alice\AppData\Roaming",
"USERPROFILE": r"C:\Users\alice",
"GITHUB_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"OPENAI_API_KEY": "sk-proj-abc123",
}
with patch.dict("os.environ", fake_env, clear=True):
result = _build_safe_env(None)
assert result["ProgramFiles"] == r"C:\Program Files"
assert result["ProgramData"] == r"C:\ProgramData"
assert result["ProgramW6432"] == r"C:\Program Files"
assert result["LOCALAPPDATA"].endswith("Local")
assert result["APPDATA"].endswith("Roaming")
assert result["USERPROFILE"] == r"C:\Users\alice"
assert "GITHUB_TOKEN" not in result
assert "OPENAI_API_KEY" not in result
# ---------------------------------------------------------------------------
# _sanitize_error
tests/tools/test_process_registry.py
-38
View File
@@ -63,44 +63,6 @@ def _wait_until(predicate, timeout: float = 5.0, interval: float = 0.05) -> bool
return False
def test_write_stdin_uses_str_for_windows_pty(monkeypatch, registry):
"""pywinpty expects str input; bytes raises a PyString conversion error."""
written = []
class _FakePty:
def write(self, value):
written.append(value)
session = _make_session(sid="pty-win")
session._pty = _FakePty()
registry._running[session.id] = session
monkeypatch.setattr("tools.process_registry._IS_WINDOWS", True)
result = registry.write_stdin(session.id, "hello\n")
assert result == {"status": "ok", "bytes_written": 6}
assert written == ["hello\n"]
assert isinstance(written[0], str)
def test_write_stdin_uses_bytes_for_posix_pty(monkeypatch, registry):
written = []
class _FakePty:
def write(self, value):
written.append(value)
session = _make_session(sid="pty-posix")
session._pty = _FakePty()
registry._running[session.id] = session
monkeypatch.setattr("tools.process_registry._IS_WINDOWS", False)
result = registry.write_stdin(session.id, "hello\n")
assert result == {"status": "ok", "bytes_written": 6}
assert written == [b"hello\n"]
# =========================================================================
# Get / Poll
# =========================================================================
tests/tools/test_send_message_target_parse.py
-20
View File
@@ -1,20 +0,0 @@
"""Parser-only tests for send_message targets.
These stay separate from ``test_send_message_tool.py`` because that module
skips wholesale when optional Telegram dependencies are not installed.
"""
from tools.send_message_tool import _parse_target_ref
def test_photon_e164_target_is_explicit() -> None:
chat_id, thread_id, is_explicit = _parse_target_ref("photon", "+15551234567")
assert chat_id == "+15551234567"
assert thread_id is None
assert is_explicit is True
def test_e164_target_still_requires_phone_platform() -> None:
assert _parse_target_ref("matrix", "+15551234567")[2] is False
tests/tools/test_send_message_tool.py
-5
View File
@@ -1199,11 +1199,6 @@ class TestParseTargetRefE164:
assert chat_id == "+15551234567"
assert is_explicit is True
def test_photon_e164_is_explicit(self):
chat_id, _, is_explicit = _parse_target_ref("photon", "+15551234567")
assert chat_id == "+15551234567"
assert is_explicit is True
def test_signal_bare_digits_still_work(self):
"""Bare digit strings continue to match the generic numeric branch."""
chat_id, _, is_explicit = _parse_target_ref("signal", "15551234567")
tests/tools/test_skills_hub.py
+1 -82
View File
@@ -1,8 +1,6 @@
"""Tests for tools/skills_hub.py — source adapters, lock file, taps, dedup logic."""
import json
import time
from typing import List, Optional
from unittest.mock import patch, MagicMock
import httpx
@@ -16,15 +14,13 @@ from tools.skills_hub import (
UrlSource,
WellKnownSkillSource,
OptionalSkillSource,
SkillSource,
SkillBundle,
SkillMeta,
SkillBundle,
HubLockFile,
TapsManager,
bundle_content_hash,
check_for_skill_updates,
create_source_router,
parallel_search_sources,
unified_search,
append_audit_log,
_skill_meta_to_dict,
@@ -2205,80 +2201,3 @@ class TestInstallPathSafety:
assert not (skills_dir / "bad-skill" / "leak.txt").exists()
assert secret.read_text() == "data exfiltration payload\n"
# ---------------------------------------------------------------------------
# parallel_search_sources — overall_timeout must be honoured even when a
# source blocks for far longer than the budget (regression: the executor used
# `with ... as pool`, whose __exit__ calls shutdown(wait=True) and blocked the
# caller on the slow worker, making overall_timeout a no-op).
# ---------------------------------------------------------------------------
class _FakeSource(SkillSource):
def __init__(self, sid: str, sleep: float = 0.0, results=None):
self._sid = sid
self._sleep = sleep
self._results = results or []
def source_id(self) -> str:
return self._sid
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
if self._sleep:
time.sleep(self._sleep)
return list(self._results)
def fetch(self, identifier: str) -> Optional[SkillBundle]:
return None
def inspect(self, identifier: str) -> Optional[SkillMeta]:
return None
class TestParallelSearchSourcesTimeout:
def _meta(self, sid: str) -> SkillMeta:
return SkillMeta(
name=f"{sid}-skill",
description="x",
source=sid,
identifier=f"{sid}/x",
trust_level="community",
)
def test_slow_source_does_not_block_caller(self):
"""A source sleeping well past overall_timeout must not stall the
return. Before the fix the executor's `with` block waited on the slow
worker (~5s); now the call returns promptly and reports the source as
timed out."""
fast = _FakeSource("fast", sleep=0.0, results=[self._meta("fast")])
slow = _FakeSource("slow", sleep=5.0, results=[self._meta("slow")])
start = time.monotonic()
all_results, source_counts, timed_out_ids = parallel_search_sources(
[fast, slow], query="q", overall_timeout=0.3,
)
elapsed = time.monotonic() - start
# Must return long before the slow source's 5s sleep finishes.
assert elapsed < 2.0, f"call blocked for {elapsed:.2f}s (timeout not honoured)"
assert "slow" in timed_out_ids
# Fast source still delivered its result and is not flagged timed out.
assert source_counts.get("fast") == 1
assert "fast" not in timed_out_ids
assert any(r.source == "fast" for r in all_results)
def test_all_fast_sources_complete_without_timeout(self):
"""Happy path: when every source finishes within budget, none are
flagged and all results are collected."""
a = _FakeSource("a", results=[self._meta("a")])
b = _FakeSource("b", results=[self._meta("b")])
all_results, source_counts, timed_out_ids = parallel_search_sources(
[a, b], query="q", overall_timeout=5.0,
)
assert timed_out_ids == []
assert source_counts.get("a") == 1
assert source_counts.get("b") == 1
assert len(all_results) == 2
tests/tools/test_skills_hub_clawhub.py
-196
View File
@@ -350,202 +350,6 @@ class TestClawHubSource(unittest.TestCase):
self.assertIn("b-skill-199", identifiers)
self.assertIn("c-skill-49", identifiers)
@patch("tools.skills_hub._write_index_cache")
@patch("tools.skills_hub._read_index_cache", return_value=None)
@patch("tools.skills_hub.httpx.get")
def test_catalog_walk_aborts_on_budget_and_does_not_poison_cache(
self, mock_get, _mock_read_cache, mock_write_cache
):
"""A walk truncated by the wall-clock budget must stop early and must
NOT write the (partial) result to the cache. Before the budget guard
the walk ran up to 750 pages and cached unconditionally — a truncated
walk poisoned the cache with incomplete catalog data."""
page_calls = {"n": 0}
def side_effect(url, *args, **kwargs):
if url.endswith("/skills"):
idx = page_calls["n"]
page_calls["n"] += 1
# Always advertise another page so the walk would never stop
# on its own — only the budget can break it.
return _MockResponse(
status_code=200,
json_data={
"items": [
{"slug": f"skill-{idx}", "displayName": f"Skill {idx}"}
],
"nextCursor": f"cursor-{idx + 1}",
},
)
return _MockResponse(status_code=404, json_data={})
mock_get.side_effect = side_effect
# Force the deadline to be in the past immediately. Budget only applies
# to bounded browse walks (max_items > 0), not the index builder path.
with patch.object(ClawHubSource, "CATALOG_WALK_BUDGET_SECONDS", -1):
results = self.src._load_catalog_index(max_items=10)
# Walk broke well before the 750-page cap.
self.assertLess(page_calls["n"], 750)
# Truncated walk must not poison the cache.
mock_write_cache.assert_not_called()
# Whatever was gathered is still returned to the caller.
self.assertIsInstance(results, list)
@patch("tools.skills_hub._write_index_cache")
@patch("tools.skills_hub._read_index_cache", return_value=None)
@patch("tools.skills_hub.httpx.get")
def test_catalog_walk_caches_when_terminating_naturally_within_budget(
self, mock_get, _mock_read_cache, mock_write_cache
):
"""Happy path: a walk that exhausts the cursor within the budget DOES
write the cache."""
def side_effect(url, *args, **kwargs):
if url.endswith("/skills"):
return _MockResponse(
status_code=200,
json_data={
"items": [
{"slug": "only-skill", "displayName": "Only Skill"}
],
# No nextCursor -> natural termination.
},
)
return _MockResponse(status_code=404, json_data={})
mock_get.side_effect = side_effect
results = self.src._load_catalog_index()
self.assertEqual(len(results), 1)
self.assertEqual(results[0].identifier, "only-skill")
mock_write_cache.assert_called_once()
class TestClawHubCatalogWalkBounded(unittest.TestCase):
"""max_items bounds the walk so browse's cold-start fallback renders one
page without walking the entire 50k+ catalog. The offline index builder
keeps max_items=0 (unbounded) and walks to exhaustion."""
def setUp(self):
self.src = ClawHubSource()
self._safe_patcher = patch("tools.skills_hub.is_safe_url", return_value=True)
self._policy_patcher = patch("tools.skills_hub.check_website_access", return_value=None)
self._safe_patcher.start()
self._policy_patcher.start()
def tearDown(self):
self._policy_patcher.stop()
self._safe_patcher.stop()
def _infinite_pages(self, page_calls):
"""A side_effect that always advertises another cursor — the walk would
never stop on its own, so only max_items / budget can break it."""
def side_effect(url, *args, **kwargs):
if url.endswith("/skills"):
idx = page_calls["n"]
page_calls["n"] += 1
return _MockResponse(
status_code=200,
json_data={
"items": [
{"slug": f"skill-{idx}", "displayName": f"Skill {idx}"}
],
"nextCursor": f"cursor-{idx + 1}",
},
)
return _MockResponse(status_code=404, json_data={})
return side_effect
@patch("tools.skills_hub._write_index_cache")
@patch("tools.skills_hub._read_index_cache", return_value=None)
@patch("tools.skills_hub.httpx.get")
def test_max_items_stops_walk_early_and_does_not_cache(
self, mock_get, _mock_read_cache, mock_write_cache
):
"""A bounded walk stops as soon as it has >= max_items skills and must
NOT poison the shared full-catalog cache with the partial slice."""
page_calls = {"n": 0}
mock_get.side_effect = self._infinite_pages(page_calls)
results = self.src._load_catalog_index(max_items=5)
# Each mocked page yields exactly 1 item, so ~5 pages cover the bound.
self.assertGreaterEqual(len(results), 5)
self.assertLess(page_calls["n"], 750, "bounded walk should stop well before the cap")
self.assertLess(page_calls["n"], 20, "should stop within a few pages of the bound")
# Partial (bounded) walk must not be cached.
mock_write_cache.assert_not_called()
@patch("tools.skills_hub._write_index_cache")
@patch("tools.skills_hub._read_index_cache", return_value=None)
@patch("tools.skills_hub.httpx.get")
def test_max_items_zero_ignores_wall_clock_budget(
self, mock_get, _mock_read_cache, _mock_write_cache
):
"""Index builder path (max_items=0) must not truncate on the browse budget."""
page_calls = {"n": 0}
mock_get.side_effect = self._infinite_pages(page_calls)
with patch.object(ClawHubSource, "CATALOG_WALK_BUDGET_SECONDS", -1):
results = self.src._load_catalog_index(max_items=0)
# No budget -> walks until the 750-page safety cap, not ~14 pages in 12s.
self.assertEqual(page_calls["n"], 750)
self.assertEqual(len(results), 750)
@patch("tools.skills_hub._write_index_cache")
@patch("tools.skills_hub._read_index_cache", return_value=None)
@patch("tools.skills_hub.httpx.get")
def test_max_items_zero_is_unbounded_and_caches(
self, mock_get, _mock_read_cache, mock_write_cache
):
"""max_items=0 (the index builder's path) walks to natural termination
and DOES cache the complete catalog."""
def side_effect(url, *args, **kwargs):
if url.endswith("/skills"):
return _MockResponse(
status_code=200,
json_data={
"items": [
{"slug": "a", "displayName": "A"},
{"slug": "b", "displayName": "B"},
{"slug": "c", "displayName": "C"},
],
# No nextCursor -> natural termination.
},
)
return _MockResponse(status_code=404, json_data={})
mock_get.side_effect = side_effect
results = self.src._load_catalog_index(max_items=0)
self.assertEqual(len(results), 3)
mock_write_cache.assert_called_once()
@patch("tools.skills_hub._write_index_cache")
@patch("tools.skills_hub._read_index_cache", return_value=None)
@patch("tools.skills_hub.httpx.get")
def test_empty_query_browse_bounds_walk_to_limit(
self, mock_get, _mock_read_cache, _mock_write_cache
):
"""search("", limit=N) is the browse cold-start path — it must bound the
catalog walk to N rather than walking the whole 50k+ catalog."""
page_calls = {"n": 0}
mock_get.side_effect = self._infinite_pages(page_calls)
results = self.src.search("", limit=10)
self.assertEqual(len(results), 10, "browse page should be exactly `limit` items")
# Walk stopped near the bound, not at the 750-page cap.
self.assertLess(page_calls["n"], 30)
if __name__ == "__main__":
unittest.main()
tests/tools/test_skills_tool.py
-20
View File
@@ -373,26 +373,6 @@ class TestSkillView:
assert result["name"] == "my-skill"
assert "Step 1" in result["content"]
def test_view_skill_by_frontmatter_name_when_dir_differs(self, tmp_path):
# The on-disk directory ("alias-dir") differs from the skill's
# frontmatter name ("real-skill-name"). skills_list() exposes the
# frontmatter name, so skill_view(name) must resolve it too.
skill_dir = tmp_path / "alias-dir"
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
"---\n"
"name: real-skill-name\n"
"description: A skill whose directory name differs from its name.\n"
"---\n\n"
"# real-skill-name\n\n"
"Step 1: Do the thing.\n"
)
with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
raw = skill_view("real-skill-name")
result = json.loads(raw)
assert result["success"] is True
assert "Step 1" in result["content"]
def test_skill_view_applies_template_vars(self, tmp_path):
with (
patch("tools.skills_tool.SKILLS_DIR", tmp_path),
tests/tools/test_subprocess_stdin_guard.py
-70
View File
@@ -1,70 +0,0 @@
"""Verify that TUI-context subprocess calls specify stdin=.
This is the pytest wrapper for scripts/check_subprocess_stdin.py.
It runs as part of the test suite so CI catches regressions when new
subprocess calls are added without stdin=subprocess.DEVNULL.
"""
import importlib.util
import subprocess
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "scripts" / "check_subprocess_stdin.py"
def _load_guard():
spec = importlib.util.spec_from_file_location("_stdin_guard", SCRIPT)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def test_all_tui_subprocess_calls_have_stdin():
"""Every subprocess.run/Popen in TUI-context code must set stdin=."""
result = subprocess.run(
[sys.executable, str(SCRIPT)],
capture_output=True,
text=True,
timeout=30,
)
assert result.returncode == 0, (
f"subprocess stdin= check failed:\n{result.stdout}\n{result.stderr}"
)
def test_oauth_setup_token_keeps_inherited_stdin():
"""The interactive 'claude setup-token' login must NOT be muzzled.
Forcing stdin=subprocess.DEVNULL here would feed the OAuth prompt EOF and
break interactive token setup. A blanket DEVNULL sweep over TUI-context
subprocess calls must leave this one inheriting stdin. Regression guard for
the over-application caught while salvaging the stdin-EOF fix.
"""
src = (REPO_ROOT / "agent" / "anthropic_adapter.py").read_text()
assert 'subprocess.run([claude_path, "setup-token"])' in src, (
"interactive setup-token call changed shape; re-verify it still "
"inherits stdin (no stdin=subprocess.DEVNULL)"
)
assert 'subprocess.run([claude_path, "setup-token"], stdin' not in src, (
"setup-token must inherit stdin so the user can complete the OAuth "
"login prompt; do not add stdin=subprocess.DEVNULL"
)
def test_inline_noqa_marker_exempts_a_call():
"""The guard honors an inline 'noqa: subprocess-stdin' exemption marker."""
guard = _load_guard()
flagged = guard.find_subprocess_calls(
"import subprocess\nsubprocess.run(['ls'])\n", "x.py"
)
assert len(flagged) == 1, "unmarked missing-stdin call should be flagged"
exempt = guard.find_subprocess_calls(
"import subprocess\nsubprocess.run(['ls']) # noqa: subprocess-stdin\n",
"x.py",
)
assert exempt == [], "inline marker should exempt the call"
tests/tools/test_terminal_config_env_sync.py
+12 -18
View File
@@ -7,9 +7,9 @@ at startup, by THREE separate code paths:
1. cli.py -> ``env_mappings`` dict (CLI / TUI startup)
2. gateway/run.py -> ``_terminal_env_map`` dict (gateway / messaging
platforms)
3. hermes_cli/config.py:set_config_value
-> bridges via the canonical ``TERMINAL_CONFIG_ENV_MAP``
(one-shot when the user runs ``hermes config set …``)
3. hermes_cli/config.py:save_config_value
-> ``_config_to_env_sync`` dict (one-shot when the
user runs ``hermes config set …``)
If any one of these is missing a key, the corresponding config.yaml setting
silently does nothing for that entry-point. This bug already shipped once
@@ -87,20 +87,14 @@ def _gateway_env_map_keys() -> set[str]:
def _save_config_env_sync_keys() -> set[str]:
"""terminal config keys bridged by ``hermes config set foo bar``.
``set_config_value`` no longer carries its own ``_config_to_env_sync``
dict — it bridges through the canonical ``TERMINAL_CONFIG_ENV_MAP`` via
``terminal_config_env_var_for_key()`` (config.py), excluding ``cwd``
(handled separately). Read the live map so this test tracks the actual
source of truth that the config-set path uses, rather than a string
literal that the consolidation removed.
"""
"""terminal config keys bridged by ``hermes config set foo bar``."""
from hermes_cli import config as hc_config
# set_config_value bridges every TERMINAL_CONFIG_ENV_MAP key except
# terminal.cwd (see the ``key != "terminal.cwd"`` guard in
# set_config_value); mirror that exclusion here.
return {k for k in hc_config.TERMINAL_CONFIG_ENV_MAP if k != "cwd"}
source = inspect.getsource(hc_config.set_config_value)
keys = _extract_dict_keys(source, "_config_to_env_sync")
# set_config_value uses fully-qualified ``terminal.foo`` keys; strip the
# prefix so we can compare against the other two maps which use bare
# leaf keys.
return {k.split(".", 1)[1] for k in keys if k.startswith("terminal.")}
# Keys present in cli.py env_mappings but intentionally absent from
@@ -186,8 +180,8 @@ def test_save_config_set_supports_critical_bridged_keys():
missing = required - save_keys
assert not missing, (
f"`hermes config set terminal.X` doesn't sync these load-bearing "
f"keys to .env: {sorted(missing)}. Add them to TERMINAL_CONFIG_ENV_MAP "
f"in hermes_cli/config.py (set_config_value bridges through it)."
f"keys to .env: {sorted(missing)}. Add them to _config_to_env_sync "
f"in hermes_cli/config.py:set_config_value."
)
tests/tools/test_terminal_tool.py
-75
View File
@@ -22,14 +22,6 @@ def test_searching_for_sudo_does_not_trigger_rewrite(monkeypatch):
assert sudo_stdin is None
def test_terminal_schema_advertises_persistent_env_state():
description = terminal_tool.TERMINAL_TOOL_DESCRIPTION
assert "exported environment variables persist between calls" in description
assert "activate a virtualenv" in description
assert "do not re-source the same environment before every command" in description
def test_printf_literal_sudo_does_not_trigger_rewrite(monkeypatch):
monkeypatch.delenv("SUDO_PASSWORD", raising=False)
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
@@ -98,30 +90,6 @@ def test_cached_sudo_password_is_used_when_env_is_unset(monkeypatch):
assert sudo_stdin == "cached-pass\n"
def test_registered_sudo_callback_is_used_without_interactive_env(monkeypatch):
monkeypatch.delenv("SUDO_PASSWORD", raising=False)
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
monkeypatch.setattr(terminal_tool, "_sudo_nopasswd_works", lambda: False)
calls = []
def sudo_callback():
calls.append("called")
return "callback-pass"
terminal_tool.set_sudo_password_callback(sudo_callback)
try:
transformed, sudo_stdin = terminal_tool._transform_sudo_command(
"echo ok | sudo tee /tmp/hermes-test"
)
finally:
terminal_tool.set_sudo_password_callback(None)
assert calls == ["called"]
assert transformed == "echo ok | sudo -S -p '' tee /tmp/hermes-test"
assert sudo_stdin == "callback-pass\n"
def test_cached_sudo_password_isolated_by_session_key(monkeypatch):
monkeypatch.delenv("SUDO_PASSWORD", raising=False)
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
@@ -200,46 +168,3 @@ def test_validate_workdir_blocks_shell_metacharacters_in_windows_paths():
assert terminal_tool._validate_workdir(r"C:\Users\Alice\project; rm -rf /")
assert terminal_tool._validate_workdir(r"C:\Users\Alice\project$(whoami)")
assert terminal_tool._validate_workdir("C:\\Users\\Alice\\project\nwhoami")
def test_get_env_config_ignores_bad_docker_json_for_local_backend(monkeypatch):
"""Docker-only JSON env vars must not break the default local backend."""
monkeypatch.setenv("TERMINAL_ENV", "local")
monkeypatch.setenv("TERMINAL_DOCKER_VOLUMES", "None")
monkeypatch.setenv("TERMINAL_DOCKER_ENV", "not-json")
monkeypatch.setenv("TERMINAL_DOCKER_FORWARD_ENV", "not-json")
monkeypatch.setenv("TERMINAL_DOCKER_EXTRA_ARGS", "not-json")
config = terminal_tool._get_env_config()
assert config["env_type"] == "local"
assert config["docker_volumes"] == []
assert config["docker_env"] == {}
assert config["docker_forward_env"] == []
assert config["docker_extra_args"] == []
def test_get_env_config_ignores_bad_docker_json_for_ssh_backend(monkeypatch):
"""Non-container remote backends should also ignore Docker-only JSON."""
monkeypatch.setenv("TERMINAL_ENV", "ssh")
monkeypatch.setenv("TERMINAL_DOCKER_VOLUMES", "None")
monkeypatch.setenv("TERMINAL_DOCKER_ENV", "not-json")
config = terminal_tool._get_env_config()
assert config["env_type"] == "ssh"
assert config["docker_volumes"] == []
assert config["docker_env"] == {}
def test_get_env_config_still_rejects_bad_docker_json_for_docker_backend(monkeypatch):
"""Selecting Docker should keep the existing actionable config error."""
monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setenv("TERMINAL_DOCKER_VOLUMES", "None")
try:
terminal_tool._get_env_config()
except ValueError as exc:
assert "TERMINAL_DOCKER_VOLUMES" in str(exc)
else:
raise AssertionError("Docker backend must validate TERMINAL_DOCKER_VOLUMES")
tests/tools/test_tts_gemini.py
-164
View File
@@ -2,7 +2,6 @@
import base64
import struct
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
@@ -256,169 +255,6 @@ class TestGenerateGeminiTts:
assert mock_post.call_args[0][0].startswith("https://custom-gemini.example.com/v1beta/")
def test_persona_prompt_file_appends_labeled_transcript(
self, tmp_path, monkeypatch, mock_gemini_response
):
from tools.tts_tool import _generate_gemini_tts
persona_file = tmp_path / "voice-persona.md"
persona_file.write_text(
"# AUDIO PROFILE: Dry Butler\n\n### DIRECTOR'S NOTES\nStyle: Understated.",
encoding="utf-8",
)
config = {"gemini": {"persona_prompt_file": str(persona_file)}}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Hi", str(tmp_path / "test.wav"), config)
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert "Synthesize speech from the TRANSCRIPT only" in prompt_text
assert "# AUDIO PROFILE: Dry Butler" in prompt_text
assert "### DIRECTOR'S NOTES\nStyle: Understated." in prompt_text
assert "#### TRANSCRIPT\nHi" in prompt_text
def test_persona_prompt_file_supports_transcript_placeholder(
self, tmp_path, monkeypatch, mock_gemini_response
):
from tools.tts_tool import _generate_gemini_tts
persona_file = tmp_path / "voice-persona.md"
persona_file.write_text(
"### DIRECTOR'S NOTES\nPacing: Slow.\n\n#### TRANSCRIPT\n{{ transcript }}",
encoding="utf-8",
)
config = {"gemini": {"persona_prompt_file": str(persona_file)}}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Read this.", str(tmp_path / "test.wav"), config)
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert "{{ transcript }}" not in prompt_text
assert "#### TRANSCRIPT\nRead this." in prompt_text
def test_missing_persona_prompt_file_warns_and_continues(
self, tmp_path, monkeypatch, caplog, mock_gemini_response
):
from tools.tts_tool import _generate_gemini_tts
config = {"gemini": {"persona_prompt_file": str(tmp_path / "missing.md")}}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Hi", str(tmp_path / "test.wav"), config)
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert prompt_text == "Hi"
assert "persona prompt file unavailable" in caplog.text
def test_audio_tags_disabled_does_not_call_rewriter(
self, tmp_path, monkeypatch, mock_gemini_response
):
from tools.tts_tool import _generate_gemini_tts
config = {
"gemini": {
"model": "gemini-3.1-flash-tts-preview",
"audio_tags": False,
}
}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("agent.auxiliary_client.call_llm") as mock_call_llm, \
patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Hi there.", str(tmp_path / "test.wav"), config)
mock_call_llm.assert_not_called()
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert prompt_text == "Hi there."
def test_audio_tags_enabled_rewrites_hidden_tts_script(
self, tmp_path, monkeypatch, mock_gemini_response
):
from tools.tts_tool import _generate_gemini_tts
persona_file = tmp_path / "voice-persona.md"
persona_file.write_text(
"### DIRECTOR'S NOTES\nStyle: Warm and amused.",
encoding="utf-8",
)
response = SimpleNamespace(
choices=[
SimpleNamespace(
message=SimpleNamespace(content="[warmly] Hi there. [soft laugh]")
)
]
)
config = {
"gemini": {
"model": "gemini-3.1-flash-tts-preview",
"audio_tags": True,
"persona_prompt_file": str(persona_file),
}
}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("agent.auxiliary_client.call_llm", return_value=response) as mock_call_llm, \
patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Hi there.", str(tmp_path / "test.wav"), config)
mock_call_llm.assert_called_once()
call_kwargs = mock_call_llm.call_args.kwargs
assert call_kwargs["task"] == "tts_audio_tags"
assert "Audio tags are inline square-bracket modifiers" in call_kwargs["messages"][0]["content"]
assert "Style: Warm and amused." in call_kwargs["messages"][1]["content"]
assert "Hi there." in call_kwargs["messages"][1]["content"]
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert "Synthesize speech from the TRANSCRIPT only" in prompt_text
assert "### DIRECTOR'S NOTES\nStyle: Warm and amused." in prompt_text
assert "#### TRANSCRIPT\n[warmly] Hi there. [soft laugh]" in prompt_text
def test_audio_tags_enabled_skips_non_tag_capable_model(
self, tmp_path, monkeypatch, mock_gemini_response, caplog
):
from tools.tts_tool import _generate_gemini_tts
config = {
"gemini": {
"model": "gemini-2.5-flash-preview-tts",
"audio_tags": True,
}
}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("agent.auxiliary_client.call_llm") as mock_call_llm, \
patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Hi there.", str(tmp_path / "test.wav"), config)
mock_call_llm.assert_not_called()
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert prompt_text == "Hi there."
assert "not known to support Gemini audio tags" in caplog.text
def test_audio_tag_rewrite_failure_falls_back_to_original_text(
self, tmp_path, monkeypatch, mock_gemini_response, caplog
):
from tools.tts_tool import _generate_gemini_tts
config = {
"gemini": {
"model": "gemini-3.1-flash-tts-preview",
"audio_tags": True,
}
}
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
with patch("agent.auxiliary_client.call_llm", side_effect=RuntimeError("boom")), \
patch("requests.post", return_value=mock_gemini_response) as mock_post:
_generate_gemini_tts("Hi there.", str(tmp_path / "test.wav"), config)
prompt_text = mock_post.call_args[1]["json"]["contents"][0]["parts"][0]["text"]
assert prompt_text == "Hi there."
assert "audio tag rewrite failed" in caplog.text
class TestGeminiInCheckRequirements:
def test_gemini_api_key_satisfies_requirements(self, monkeypatch):
tests/tools/test_web_keyless_default_fallback.py
-100
View File
@@ -1,100 +0,0 @@
"""Regression: the keyless Parallel web default must survive a failed sweep.
``web_search`` / ``web_extract`` are documented to work out of the box with
zero setup via the bundled keyless Parallel free-MCP backend. That guarantee
only holds if the bundled ``plugins/web/*`` providers are registered in
``agent.web_search_registry``. The dispatch triggers the general plugin sweep
(:func:`hermes_cli.plugins._ensure_plugins_discovered`) to do that — but the
sweep can finish without registering them (its exception swallowed as a
warning, a packaged layout where it ran before the bundled tree was
importable, or a stale empty-discovery cache). When that happened, *both*
tools dead-ended on "No web {search,extract} provider configured" even though
no setup should be needed.
These tests pin the invariant that :func:`tools.web_tools._ensure_web_plugins_loaded`
guarantees the keyless default is registered regardless of the sweep's outcome,
and that the direct-registration fallback honors an explicit ``plugins.disabled``
entry. Real imports from the bundled plugin modules — no provider mocking.
"""
from __future__ import annotations
import pytest
import agent.web_search_registry as reg
import hermes_cli.plugins as plugins
from tools import web_tools
@pytest.fixture(autouse=True)
def _clean_registry():
reg._reset_for_tests()
yield
reg._reset_for_tests()
def _boom(*_a, **_k):
raise RuntimeError("discovery boom")
def test_keyless_default_registered_when_discovery_raises(monkeypatch):
"""A swallowed discovery failure must not strand the keyless default."""
monkeypatch.setattr(plugins, "_ensure_plugins_discovered", _boom)
assert reg.get_provider("parallel") is None
web_tools._ensure_web_plugins_loaded()
parallel = reg.get_provider("parallel")
assert parallel is not None, "keyless Parallel default not restored"
# It is the universal keyless default precisely because it does both.
assert parallel.supports_search()
assert parallel.supports_extract()
def test_fallback_registers_full_bundled_set(monkeypatch):
"""The fix covers the whole bundled provider class, not just parallel."""
monkeypatch.setattr(plugins, "_ensure_plugins_discovered", _boom)
web_tools._ensure_web_plugins_loaded()
names = {p.name for p in reg.list_providers()}
# Every bundled backend a user might have configured should be reachable
# again, so an explicit ``web.extract_backend: firecrawl`` etc. resolves.
for expected in ("parallel", "firecrawl", "tavily", "exa"):
assert expected in names, f"{expected} missing after fallback"
def test_fallback_honors_explicit_disable(monkeypatch):
"""A backend the user turned off via plugins.disabled stays off."""
monkeypatch.setattr(plugins, "_get_disabled_plugins", lambda: {"web-parallel"})
web_tools._register_bundled_web_providers_directly()
names = {p.name for p in reg.list_providers()}
assert "parallel" not in names, "explicit disable was ignored"
# Other bundled backends are unaffected by the parallel disable.
assert "tavily" in names
def test_fallback_is_noop_when_discovery_already_registered(monkeypatch):
"""Healthy path: don't pay for the direct sweep when parallel is present."""
# Pretend the general sweep already registered the keyless default.
import importlib
class _Ctx:
def register_web_search_provider(self, provider):
reg.register_provider(provider)
importlib.import_module("plugins.web.parallel").register(_Ctx())
monkeypatch.setattr(plugins, "_ensure_plugins_discovered", lambda *a, **k: None)
calls = {"n": 0}
real = web_tools._register_bundled_web_providers_directly
def _spy():
calls["n"] += 1
real()
monkeypatch.setattr(web_tools, "_register_bundled_web_providers_directly", _spy)
web_tools._ensure_web_plugins_loaded()
assert calls["n"] == 0, "direct-registration ran on the healthy path"
tests/tools/test_web_providers.py
+13 -60
View File
@@ -167,21 +167,6 @@ class TestPerCapabilityBackendSelection:
monkeypatch.setenv("TAVILY_API_KEY", "test-key")
assert web_tools._get_search_backend() == "tavily"
def test_explicit_extract_backend_honored_when_unavailable(self, monkeypatch):
"""An explicit per-capability backend is honored even with no creds, so
its setup error surfaces instead of silently rerouting to the keyless
Parallel default (which would send user URLs to a different provider)."""
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {
"extract_backend": "firecrawl",
})
for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "FIRECRAWL_GATEWAY_URL"):
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False, raising=False)
# Resolves to firecrawl (not parallel) despite firecrawl being unavailable.
assert web_tools._get_extract_backend() == "firecrawl"
def test_falls_back_to_generic_backend_when_extract_backend_empty(self, monkeypatch):
from tools import web_tools
@@ -192,7 +177,7 @@ class TestPerCapabilityBackendSelection:
monkeypatch.setenv("PARALLEL_API_KEY", "test-key")
assert web_tools._get_extract_backend() == "parallel"
def test_explicit_search_backend_honored_when_unavailable(self, monkeypatch):
def test_search_backend_ignored_when_not_available(self, monkeypatch):
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {
@@ -201,10 +186,8 @@ class TestPerCapabilityBackendSelection:
})
monkeypatch.delenv("EXA_API_KEY", raising=False)
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-key")
# The explicit per-capability choice (exa) is honored even though it's
# unavailable, so its setup error surfaces — we don't silently reroute
# to the shared backend (or the keyless Parallel default).
assert web_tools._get_search_backend() == "exa"
# Should fall back to firecrawl since exa isn't configured
assert web_tools._get_search_backend() == "firecrawl"
def test_fully_backward_compatible_with_web_backend_only(self, monkeypatch):
from tools import web_tools
@@ -308,55 +291,25 @@ class TestUnconfiguredErrorEnvelopeParity:
):
monkeypatch.delenv(k, raising=False)
def test_extract_empty_urls_does_not_raise(self, monkeypatch):
"""Regression: empty (or fully SSRF-blocked) URL sets skip the dispatch
branch; the free-Parallel flag must still be initialized so the tool
returns an error envelope instead of UnboundLocalError."""
import asyncio
from tools import web_tools
self._clear_web_creds(monkeypatch)
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
out = asyncio.run(web_tools.web_extract_tool([], "markdown"))
# The key assertion is that it returns a normal error envelope (a
# string) rather than raising UnboundLocalError.
assert isinstance(out, str)
result = json.loads(out)
assert "error" in result
def test_unconfigured_search_falls_back_to_free_parallel(self, monkeypatch):
"""``web_search_tool`` with no creds routes to Parallel's free Search
MCP rather than erroring. The MCP transport is mocked so the test
stays offline; we assert dispatch landed on parallel and returned the
standard search envelope.
def test_unconfigured_search_emits_top_level_error(self, monkeypatch):
"""``web_search_tool`` with no creds returns ``{"error": "Error searching web: ..."}``
— matching main's ``tool_error()`` envelope, not a per-result shape.
"""
from tools import web_tools
import plugins.web.parallel.provider as parallel_provider
self._clear_web_creds(monkeypatch)
# Reset firecrawl client cache so the unconfigured state is re-evaluated
monkeypatch.setattr(web_tools, "_firecrawl_client", None, raising=False)
monkeypatch.setattr(web_tools, "_firecrawl_client_config", None, raising=False)
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
captured = {}
def _fake_mcp(query, limit, api_key):
captured["query"] = query
captured["api_key"] = api_key
return {
"success": True,
"data": {"web": [
{"url": "https://example.com", "title": "Example",
"description": "hit", "position": 1},
]},
}
monkeypatch.setattr(parallel_provider, "_mcp_web_search", _fake_mcp)
result = json.loads(web_tools.web_search_tool("hello world", limit=3))
assert result.get("success") is True, f"expected success, got {result}"
assert result["data"]["web"][0]["url"] == "https://example.com"
# Keyless path: dispatched to parallel with no Bearer token.
assert captured == {"query": "hello world", "api_key": None}
assert "error" in result, f"expected top-level 'error' key, got {result}"
# ``Error searching web:`` prefix comes from web_tools' top-level except handler
assert "Error searching web:" in result["error"]
assert "FIRECRAWL_API_KEY" in result["error"]
# No per-result burying
assert "results" not in result
class TestDispatchersTriggerPluginDiscovery:
tests/tools/test_web_providers_ddgs.py
+2 -6
View File
@@ -190,11 +190,7 @@ class TestDDGSBackendWiring:
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True)
assert web_tools._get_backend() == "exa"
def test_auto_detect_prefers_keyless_parallel_over_ddgs(self, monkeypatch):
# With no credentials, keyless Parallel is the auto-detect default even
# when the ddgs package is installed — ddgs is search-only (can't
# extract), so Parallel is preferred so both search and extract work.
# ddgs remains reachable via an explicit web.backend=ddgs.
def test_auto_detect_picks_ddgs_as_last_resort(self, monkeypatch):
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "PARALLEL_API_KEY",
@@ -202,7 +198,7 @@ class TestDDGSBackendWiring:
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True)
assert web_tools._get_backend() == "parallel"
assert web_tools._get_backend() == "ddgs"
def test_check_web_api_key_true_when_ddgs_configured(self, monkeypatch):
from tools import web_tools
tests/tools/test_web_providers_searxng.py
+2 -5
View File
@@ -313,9 +313,7 @@ class TestCheckWebApiKey:
)
assert web_tools.check_web_api_key() is True
def test_no_credentials_usable_via_free_parallel(self, monkeypatch):
"""No credentials → check_web_api_key True: the keyless Parallel free MCP
services calls, so web is usable out of the box."""
def test_no_credentials_fails(self, monkeypatch):
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
@@ -326,8 +324,7 @@ class TestCheckWebApiKey:
monkeypatch.delenv("SEARXNG_URL", raising=False)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False)
assert web_tools.check_web_api_key() is True
assert web_tools.check_web_api_key() is False
# ---------------------------------------------------------------------------
tests/tools/test_web_tools_config.py
+25 -121
View File
@@ -340,13 +340,12 @@ class TestBackendSelection:
patch.dict(os.environ, {"EXA_API_KEY": "exa-test"}):
assert _get_backend() == "exa"
def test_fallback_exa_takes_priority_over_parallel(self):
"""Direct-credential backends are tried in the order tavily > exa > parallel
so an explicit Exa key wins when both Exa and Parallel are configured."""
def test_fallback_parallel_takes_priority_over_exa(self):
"""Exa should only win the fallback path when it is the only configured backend."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch.dict(os.environ, {"EXA_API_KEY": "exa-test", "PARALLEL_API_KEY": "par-test"}):
assert _get_backend() == "exa"
assert _get_backend() == "parallel"
def test_fallback_tavily_only_key(self):
"""Only TAVILY_API_KEY set → 'tavily'."""
@@ -355,27 +354,27 @@ class TestBackendSelection:
patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}):
assert _get_backend() == "tavily"
def test_fallback_tavily_beats_firecrawl_direct(self):
"""Tavily ranks above firecrawl in the explicit-credential block."""
def test_fallback_tavily_with_firecrawl_prefers_firecrawl(self):
"""Tavily + Firecrawl keys, no config → 'firecrawl' (backward compat)."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test", "FIRECRAWL_API_KEY": "fc-test"}):
assert _get_backend() == "tavily"
assert _get_backend() == "firecrawl"
def test_fallback_tavily_beats_parallel(self):
"""Tavily is first in the explicit-credential block so it wins over parallel."""
def test_fallback_tavily_with_parallel_prefers_parallel(self):
"""Tavily + Parallel keys, no config → 'parallel' (Parallel takes priority over Tavily)."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test", "PARALLEL_API_KEY": "par-test"}):
assert _get_backend() == "tavily"
# Parallel + no Firecrawl → parallel
assert _get_backend() == "parallel"
def test_fallback_parallel_beats_firecrawl_direct(self):
"""Parallel + Firecrawl-direct → parallel (parallel is the higher-priority
explicit-credential backend; firecrawl-direct ranks below it)."""
def test_fallback_both_keys_defaults_to_firecrawl(self):
"""Both keys set, no config → 'firecrawl' (backward compat)."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch.dict(os.environ, {"PARALLEL_API_KEY": "test-key", "FIRECRAWL_API_KEY": "fc-test"}):
assert _get_backend() == "parallel"
assert _get_backend() == "firecrawl"
def test_fallback_firecrawl_only_key(self):
"""Only FIRECRAWL_API_KEY set → 'firecrawl'."""
@@ -384,14 +383,11 @@ class TestBackendSelection:
patch.dict(os.environ, {"FIRECRAWL_API_KEY": "fc-test"}):
assert _get_backend() == "firecrawl"
def test_fallback_no_keys_defaults_to_parallel(self):
"""No credentials, no config → 'parallel' (free Search MCP works
keyless). Selection is purely credential-based."""
def test_fallback_no_keys_defaults_to_firecrawl(self):
"""No keys, no config → 'firecrawl' (will fail at client init)."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
patch("tools.web_tools._ddgs_package_importable", return_value=False):
assert _get_backend() == "parallel"
with patch("tools.web_tools._load_web_config", return_value={}):
assert _get_backend() == "firecrawl"
def test_invalid_config_falls_through_to_fallback(self):
"""web.backend=invalid → ignored, uses key-based fallback."""
@@ -400,27 +396,6 @@ class TestBackendSelection:
patch.dict(os.environ, {"PARALLEL_API_KEY": "test-key"}):
assert _get_backend() == "parallel"
def test_managed_gateway_does_not_preempt_explicit_tavily(self):
"""Regression: a Nous OAuth token (managed gateway "ready") must NOT
beat an explicitly configured TAVILY_API_KEY in the fallback path.
Free Nous tiers don't include web search, so the user's deliberate
Tavily setup would fail at runtime with "no subscription" if the
gateway pre-empted it."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=True), \
patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}):
assert _get_backend() == "tavily"
def test_managed_gateway_only_falls_through_to_firecrawl(self):
"""When no explicit-credential backend is configured, a Nous-managed
gateway token still selects firecrawl — the convenience path is
preserved, just no longer pre-empts."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=True):
assert _get_backend() == "firecrawl"
class TestParallelClientConfig:
"""Test suite for Parallel client initialization."""
@@ -626,74 +601,9 @@ class TestCheckWebApiKey:
from tools.web_tools import check_web_api_key
assert check_web_api_key() is True
def test_no_keys_usable_via_free_parallel(self):
"""No credentials → check_web_api_key True: selection resolves to the
keyless Parallel free MCP, which genuinely services calls (web works out
of the box). check_web_api_key is a usability probe, not a key check."""
def test_no_keys_returns_false(self):
from tools.web_tools import check_web_api_key
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
patch("tools.web_tools._ddgs_package_importable", return_value=False), \
patch.dict(os.environ, {}, clear=False):
for k in ("PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL",
"TAVILY_API_KEY", "EXA_API_KEY", "SEARXNG_URL", "BRAVE_SEARCH_API_KEY"):
os.environ.pop(k, None)
assert check_web_api_key() is True
def test_typo_extract_backend_not_masked_by_parallel(self):
"""A typo'd per-capability backend is honored (so dispatch errors)
rather than silently falling through to keyless Parallel."""
from tools.web_tools import _get_extract_backend, check_web_api_key
with patch("tools.web_tools._load_web_config",
return_value={"extract_backend": "parrallel"}):
assert _get_extract_backend() == "parrallel" # not "parallel"
assert check_web_api_key() is False # unknown → unusable
def test_keyless_parallel_unusable_when_provider_disabled(self):
"""If the bundled web-parallel provider is disabled/unregistered, the
keyless free-MCP path must NOT report web as usable — otherwise setup is
skipped but web tools fail at runtime with no provider."""
from tools.web_tools import check_web_api_key
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._parallel_provider_registered", return_value=False), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
patch("tools.web_tools.check_firecrawl_api_key", return_value=False), \
patch("tools.web_tools._ddgs_package_importable", return_value=False), \
patch.dict(os.environ, {}, clear=False):
for var in (
"PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL",
"TAVILY_API_KEY", "EXA_API_KEY", "BRAVE_SEARCH_API_KEY", "SEARXNG_URL",
):
os.environ.pop(var, None)
assert check_web_api_key() is False
def test_extract_autodetect_skips_search_only_for_keyless_parallel(self):
"""A search-only env credential (SEARXNG_URL) must not shadow the keyless
Parallel free-MCP extract fallback: extract auto-detect skips search-only
backends, so _get_extract_backend resolves to parallel (which can fetch),
while search auto-detect still prefers the configured searxng."""
from tools.web_tools import _get_extract_backend, _get_search_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch.dict(os.environ, {}, clear=False):
for var in (
"PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL",
"TAVILY_API_KEY", "EXA_API_KEY", "BRAVE_SEARCH_API_KEY",
):
os.environ.pop(var, None)
os.environ["SEARXNG_URL"] = "http://localhost:8080"
with patch("tools.web_tools._is_tool_gateway_ready", return_value=False):
assert _get_search_backend() == "searxng"
assert _get_extract_backend() == "parallel"
def test_configured_but_unavailable_backend_reports_unusable(self):
"""An explicitly configured backend with no creds (exa, no key) →
check_web_api_key False so diagnostics flag the misconfiguration —
even though the tools stay registered."""
from tools.web_tools import check_web_api_key
with patch("tools.web_tools._load_web_config", return_value={"backend": "exa"}), \
patch.dict(os.environ, {}, clear=False):
os.environ.pop("EXA_API_KEY", None)
assert check_web_api_key() is False
assert check_web_api_key() is False
def test_both_keys_returns_true(self):
with patch.dict(os.environ, {
@@ -756,18 +666,12 @@ class TestCheckWebApiKey:
assert refresh_calls == []
def test_web_tools_registered_even_when_configured_backend_unavailable(self):
# Registration is unconditional (web_tools_registered) so an explicitly
# configured but unavailable backend (exa without EXA_API_KEY) keeps the
# tools registered to surface exa's setup error at call time — while the
# readiness probe (check_web_api_key) honestly reports not-configured.
from tools.web_tools import web_tools_registered, check_web_api_key
assert web_tools_registered() is True
with patch("tools.web_tools._load_web_config", return_value={"backend": "exa"}), \
patch.dict(os.environ, {}, clear=False):
os.environ.pop("EXA_API_KEY", None)
assert web_tools_registered() is True
assert check_web_api_key() is False
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"}):
tests/tools/test_windows_native_support.py
+6 -8
View File
@@ -841,15 +841,13 @@ class TestLocalEnvironmentWindowsTempDir:
class TestLocalEnvironmentPathInjectionGated:
"""Sane PATH completion must stay POSIX-only."""
"""The /usr/bin PATH injection in _make_run_env must be POSIX-only."""
def test_windows_path_is_left_unchanged(self, monkeypatch):
from tools.environments import local as local_mod
from tools.environments.local import _append_missing_sane_path_entries
monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
path = r"C:\Windows\System32;C:\Program Files\Git\bin"
assert _append_missing_sane_path_entries(path) == path
def test_source_gates_path_injection(self):
root = Path(__file__).resolve().parents[2]
source = (root / "tools" / "environments" / "local.py").read_text(encoding="utf-8")
# The fix wraps the injection in `if not _IS_WINDOWS`.
assert 'not _IS_WINDOWS and "/usr/bin" not in existing_path.split(":")' in source
# ---------------------------------------------------------------------------
tests/tools/test_write_approval.py
-384
View File
@@ -1,384 +0,0 @@
"""Tests for the memory/skill write-approval gate (tools/write_approval.py)
and the shared slash-command handlers (hermes_cli/write_approval_commands.py).
Covers the boolean write_approval gate (off by default = write freely; on =
require approval) for both subsystems, the foreground-vs-background staging
split, pending store CRUD, and the list/approve/reject/diff/approval
subcommand dispatch.
"""
import json
import os
import tempfile
import shutil
import pytest
@pytest.fixture
def hermes_home(monkeypatch):
d = tempfile.mkdtemp(prefix="hermes_wa_test_")
home = os.path.join(d, ".hermes")
os.makedirs(home)
monkeypatch.setenv("HERMES_HOME", home)
yield home
shutil.rmtree(d, ignore_errors=True)
def _set_approval(subsystem, enabled):
import hermes_cli.config as cfg
c = cfg.load_config()
c.setdefault(subsystem, {})["write_approval"] = enabled
cfg.save_config(c)
# ---------------------------------------------------------------------------
# Config resolution
# ---------------------------------------------------------------------------
def test_default_gate_is_off(hermes_home):
from tools import write_approval as wa
# Default: gate off → writes flow freely.
assert wa.write_approval_enabled("memory") is False
assert wa.write_approval_enabled("skills") is False
def test_invalid_subsystem_is_off(hermes_home):
from tools import write_approval as wa
assert wa.write_approval_enabled("bogus") is False
def test_normalize_enabled_coerces_values():
from tools import write_approval as wa
# Real bools pass through.
assert wa._normalize_enabled(True) is True
assert wa._normalize_enabled(False) is False
# Truthy strings → True (incl. legacy 'approve').
assert wa._normalize_enabled("on") is True
assert wa._normalize_enabled("approve") is True
assert wa._normalize_enabled("true") is True
# Everything else → False (gate off is the safe default).
assert wa._normalize_enabled("off") is False
assert wa._normalize_enabled("garbage") is False
assert wa._normalize_enabled(None) is False
# ---------------------------------------------------------------------------
# Memory gate
# ---------------------------------------------------------------------------
def test_memory_gate_off_allows_write(hermes_home):
# Default (gate off) → write straight through, no staging.
from tools.memory_tool import memory_tool, MemoryStore
from tools import write_approval as wa
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "user", "save me", store=store))
assert r["success"] is True
assert r["entry_count"] == 1
assert wa.pending_count("memory") == 0
def test_memory_gate_on_no_interactive_stages(hermes_home):
# Gate on, no approval callback / not a gateway context → stage.
from tools.memory_tool import memory_tool, MemoryStore
from tools import write_approval as wa
_set_approval("memory", True)
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "memory", "stage me", store=store))
assert r.get("staged") is True
assert r.get("pending_id")
# Not written to the live store yet.
assert store.memory_entries == []
pend = wa.list_pending("memory")
assert len(pend) == 1
assert pend[0]["id"] == r["pending_id"]
def test_memory_gate_on_then_apply(hermes_home):
from tools.memory_tool import memory_tool, MemoryStore, apply_memory_pending
from tools import write_approval as wa
_set_approval("memory", True)
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "user", "approved entry", store=store))
pid = r["pending_id"]
rec = wa.get_pending("memory", pid)
result = apply_memory_pending(rec["payload"], store)
assert result["success"] is True
assert "approved entry" in store.user_entries[0]
# ---------------------------------------------------------------------------
# Skill gate
# ---------------------------------------------------------------------------
_SKILL = (
"---\nname: test-skill\ndescription: A test skill\nversion: 1.0.0\n---\n"
"# Test\nbody\n"
)
def test_skill_gate_off_allows_create(hermes_home):
# Default (gate off) → skill is created normally, not staged.
import importlib
import tools.skill_manager_tool as smt
importlib.reload(smt)
from tools import write_approval as wa
r = json.loads(smt.skill_manage("create", "free-skill", content=_SKILL))
assert r.get("success") is True
assert wa.pending_count("skills") == 0
def test_skill_gate_on_always_stages(hermes_home):
# Skills stage even in the foreground (too big to review inline).
from tools.skill_manager_tool import skill_manage
from tools import write_approval as wa
_set_approval("skills", True)
r = json.loads(skill_manage("create", "staged-skill", content=_SKILL))
assert r.get("staged") is True
assert "staged-skill" in r.get("gist", "")
assert wa.pending_count("skills") == 1
def test_skill_gate_on_then_apply_writes_file(hermes_home):
# SKILLS_DIR is resolved at import time, so reload the skill module under
# this test's HERMES_HOME to exercise the real on-disk write path.
import importlib
import tools.skill_manager_tool as smt
importlib.reload(smt)
from tools import write_approval as wa
_set_approval("skills", True)
r = json.loads(smt.skill_manage("create", "applied-skill", content=_SKILL))
rec = wa.get_pending("skills", r["pending_id"])
res = json.loads(smt.apply_skill_pending(rec["payload"]))
assert res["success"] is True
assert smt._find_skill("applied-skill") is not None
def test_skill_create_diff_is_full_content(hermes_home):
from tools.skill_manager_tool import skill_manage
from tools import write_approval as wa
_set_approval("skills", True)
r = json.loads(skill_manage("create", "diff-skill", content=_SKILL))
rec = wa.get_pending("skills", r["pending_id"])
diff = wa.skill_pending_diff(rec)
assert "name: test-skill" in diff
# ---------------------------------------------------------------------------
# Pending store CRUD
# ---------------------------------------------------------------------------
def test_pending_store_roundtrip(hermes_home):
from tools import write_approval as wa
rec = wa.stage_write("memory", {"action": "add", "target": "user", "content": "x"},
summary="add x", origin="foreground")
assert wa.pending_count("memory") == 1
got = wa.get_pending("memory", rec["id"])
assert got["payload"]["content"] == "x"
assert wa.discard_pending("memory", rec["id"]) is True
assert wa.pending_count("memory") == 0
assert wa.get_pending("memory", rec["id"]) is None
# ---------------------------------------------------------------------------
# Shared command handler
# ---------------------------------------------------------------------------
def test_handle_pending_list_empty(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
out = handle_pending_subcommand(wa.MEMORY, ["pending"])
assert "No pending memory" in out
def test_handle_approve_all(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools.memory_tool import MemoryStore
from tools import write_approval as wa
store = MemoryStore(); store.load_from_disk()
wa.stage_write("memory", {"action": "add", "target": "user", "content": "a"},
summary="a", origin="foreground")
wa.stage_write("memory", {"action": "add", "target": "user", "content": "b"},
summary="b", origin="foreground")
out = handle_pending_subcommand(wa.MEMORY, ["approve", "all"], memory_store=store)
assert "Approved 2" in out
assert wa.pending_count("memory") == 0
assert len(store.user_entries) == 2
def test_handle_reject(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
rec = wa.stage_write("skills", {"action": "create", "name": "s"},
summary="create s", origin="background_review")
out = handle_pending_subcommand(wa.SKILLS, ["reject", rec["id"]])
assert "Rejected" in out
assert wa.pending_count("skills") == 0
def test_handle_approval_on(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
captured = {}
out = handle_pending_subcommand(
wa.MEMORY, ["approval", "on"],
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
)
assert captured["enabled"] is True
assert "on" in out
def test_handle_approval_off(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
captured = {}
out = handle_pending_subcommand(
wa.SKILLS, ["approval", "off"],
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
)
assert captured["enabled"] is False
assert "off" in out
def test_handle_mode_alias_still_works(hermes_home):
# 'mode' is kept as a back-compat alias for 'approval'.
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
captured = {}
out = handle_pending_subcommand(
wa.MEMORY, ["mode", "on"],
set_mode_fn=lambda enabled: captured.update(enabled=enabled),
)
assert captured["enabled"] is True
assert "on" in out
def test_handle_approval_invalid(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
out = handle_pending_subcommand(wa.MEMORY, ["approval", "bogus"],
set_mode_fn=lambda enabled: None)
assert "Invalid value" in out
def test_handle_unknown_subcommand_returns_none(hermes_home):
from hermes_cli.write_approval_commands import handle_pending_subcommand
from tools import write_approval as wa
# An unrecognized /skills subcommand (e.g. 'search') must return None so
# the CLI falls through to the skills hub.
out = handle_pending_subcommand(wa.SKILLS, ["search", "foo"])
assert out is None
# ---------------------------------------------------------------------------
# Inline (interactive CLI) approval path — regression for the bug where the
# per-thread approval callback was never passed to prompt_dangerous_approval,
# so every gated foreground memory write was silently denied.
# ---------------------------------------------------------------------------
@pytest.fixture
def approval_callback_cleanup():
yield
from tools.terminal_tool import set_approval_callback
set_approval_callback(None)
def test_memory_inline_approve_writes(hermes_home, approval_callback_cleanup):
from tools.memory_tool import memory_tool, MemoryStore
from tools.terminal_tool import set_approval_callback
from tools import write_approval as wa
_set_approval("memory", True)
calls = []
def approve_cb(command, description, **kw):
calls.append((command, description))
return "once"
set_approval_callback(approve_cb)
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "memory", "approved fact", store=store))
assert r["success"] is True
assert r.get("staged") is None # real write, not staged
assert store.memory_entries == ["approved fact"]
assert wa.pending_count("memory") == 0
# The registered callback must actually be invoked (not the input() path).
assert len(calls) == 1
assert "approved fact" in calls[0][0]
def test_memory_inline_deny_blocks(hermes_home, approval_callback_cleanup):
from tools.memory_tool import memory_tool, MemoryStore
from tools.terminal_tool import set_approval_callback
from tools import write_approval as wa
_set_approval("memory", True)
set_approval_callback(lambda command, description, **kw: "deny")
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "memory", "denied fact", store=store))
assert r["success"] is False
assert "denied" in r["error"].lower()
assert store.memory_entries == []
assert wa.pending_count("memory") == 0 # denied, not staged
def test_memory_inline_callback_error_stages(hermes_home, approval_callback_cleanup):
# If the prompt machinery fails, fall back to staging — never drop silently.
from tools.memory_tool import memory_tool, MemoryStore
from tools.terminal_tool import set_approval_callback
from tools import write_approval as wa
_set_approval("memory", True)
def broken_cb(command, description, **kw):
raise RuntimeError("boom")
set_approval_callback(broken_cb)
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "memory", "fallback fact", store=store))
assert r.get("staged") is True
assert wa.pending_count("memory") == 1
def test_gateway_context_stages_not_prompts(hermes_home, monkeypatch):
# A gateway session has no per-thread CLI callback; the dangerous-command
# /approve round-trip lives in the pending-queue machinery which the gate
# does not use. The gate must stage, never attempt an inline prompt
# (which would hit the input() fallback and silently deny).
from tools.memory_tool import memory_tool, MemoryStore
from tools import write_approval as wa
_set_approval("memory", True)
monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "memory", "gateway fact", store=store))
assert r.get("staged") is True
assert store.memory_entries == []
assert wa.pending_count("memory") == 1
def test_skills_never_prompt_inline_even_with_callback(hermes_home, approval_callback_cleanup):
# Skills always stage — even when an interactive callback is registered.
from tools.skill_manager_tool import skill_manage
from tools.terminal_tool import set_approval_callback
from tools import write_approval as wa
_set_approval("skills", True)
calls = []
set_approval_callback(lambda c, d, **kw: calls.append(1) or "once")
r = json.loads(skill_manage(
action="create", name="test-inline-skill",
content="---\nname: test-inline-skill\ndescription: x\n---\nbody\n"))
assert r.get("staged") is True
assert calls == [] # never prompted
assert wa.pending_count("skills") == 1
def test_memory_invalid_params_rejected_before_staging(hermes_home):
# Param validation must run BEFORE the gate so a broken write is rejected
# immediately instead of staged and failing at approve time.
from tools.memory_tool import memory_tool, MemoryStore
from tools import write_approval as wa
_set_approval("memory", True)
store = MemoryStore(); store.load_from_disk()
r = json.loads(memory_tool("add", "memory", None, store=store))
assert r["success"] is False
assert wa.pending_count("memory") == 0