Merge upstream/main and address Copilot review feedback

Merge resolved conflicts in web/src/{i18n/{en,zh,types}.ts,lib/api.ts}
by keeping both this branch's `profiles` additions and upstream's new
`models` page additions.

Copilot review feedback:
- Implement POST /api/profiles/{name}/open-terminal endpoint (already
  present); align Windows branch to `cmd.exe /c start "" <cmd>` so it
  matches the new test and spawns a fresh window instead of /k reusing
  the parent console.
- Move backslash escaping out of the macOS AppleScript f-string
  expression (Python <3.12 disallows backslashes inside f-string
  expression parts).
- Patch `_get_wrapper_dir` via monkeypatch in
  test_profiles_create_creates_wrapper_alias_when_safe so the test no
  longer writes to the real `~/.local/bin`.
- Extend test_dashboard_browser_safe_imports to scan `.ts` files in
  addition to `.tsx`.
- Switch upstream's new ModelsPage.tsx away from the `@nous-research/ui`
  root barrel onto per-component subpaths to satisfy the stricter scan.
- Fix NouiTypography `leading-1.4` -> `leading-[1.4]` so Tailwind
  actually emits the line-height for the `sm` variant.
- Guard ProfilesPage.openSoulEditor against out-of-order responses by
  tracking the latest requested profile via a ref.
- Replace ProfilesPage's hand-rolled setup command with a fetch to
  `/api/profiles/{name}/setup-command` so the copied command always
  matches what the backend would actually run (handles wrapper-alias
  collisions and reserved names correctly).
- Wire SOUL.md textarea label `htmlFor` -> textarea `id` so screen
  readers and clicking the label work as expected.
This commit is contained in:
VinceZ-Hms-Coder
2026-04-30 06:43:22 -04:00
496 changed files with 47359 additions and 2846 deletions
@@ -1097,3 +1097,63 @@ class TestHuggingFaceModels:
from hermes_cli.models import _PROVIDER_LABELS
assert "huggingface" in _PROVIDER_LABELS
assert _PROVIDER_LABELS["huggingface"] == "Hugging Face"
# =============================================================================
# MiniMax OAuth provider tests (added by feat/minimax-oauth-provider)
# =============================================================================
class TestMinimaxOAuthProvider:
"""Tests for the minimax-oauth OAuth provider."""
def test_minimax_oauth_in_provider_registry(self):
assert "minimax-oauth" in PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY["minimax-oauth"]
assert pconfig.auth_type == "oauth_minimax"
assert pconfig.id == "minimax-oauth"
def test_minimax_oauth_has_correct_endpoints(self):
from hermes_cli.auth import (
MINIMAX_OAUTH_GLOBAL_BASE,
MINIMAX_OAUTH_GLOBAL_INFERENCE,
MINIMAX_OAUTH_CN_BASE,
MINIMAX_OAUTH_CN_INFERENCE,
)
pconfig = PROVIDER_REGISTRY["minimax-oauth"]
assert pconfig.portal_base_url == MINIMAX_OAUTH_GLOBAL_BASE
assert pconfig.inference_base_url == MINIMAX_OAUTH_GLOBAL_INFERENCE
assert pconfig.extra["cn_portal_base_url"] == MINIMAX_OAUTH_CN_BASE
assert pconfig.extra["cn_inference_base_url"] == MINIMAX_OAUTH_CN_INFERENCE
def test_minimax_oauth_alias_resolves_portal(self):
result = resolve_provider("minimax-portal")
assert result == "minimax-oauth"
def test_minimax_oauth_alias_resolves_global(self):
result = resolve_provider("minimax-global")
assert result == "minimax-oauth"
def test_minimax_oauth_alias_resolves_underscore(self):
result = resolve_provider("minimax_oauth")
assert result == "minimax-oauth"
def test_minimax_oauth_listed_in_canonical_providers(self):
from hermes_cli.models import CANONICAL_PROVIDERS
slugs = [p.slug for p in CANONICAL_PROVIDERS]
assert "minimax-oauth" in slugs
def test_minimax_oauth_models_alias_in_models_py(self):
from hermes_cli.models import _PROVIDER_ALIASES
assert _PROVIDER_ALIASES.get("minimax-portal") == "minimax-oauth"
assert _PROVIDER_ALIASES.get("minimax-global") == "minimax-oauth"
assert _PROVIDER_ALIASES.get("minimax_oauth") == "minimax-oauth"
def test_minimax_oauth_has_models(self):
from hermes_cli.models import _PROVIDER_MODELS
models = _PROVIDER_MODELS.get("minimax-oauth", [])
assert len(models) >= 1
def test_minimax_oauth_aux_model_registered(self):
from agent.auxiliary_client import _API_KEY_PROVIDER_AUX_MODELS
assert "minimax-oauth" in _API_KEY_PROVIDER_AUX_MODELS
assert _API_KEY_PROVIDER_AUX_MODELS["minimax-oauth"] # non-empty
+18 -5
View File
@@ -1446,23 +1446,36 @@ def test_seed_custom_pool_respects_config_suppression(tmp_path, monkeypatch):
def test_credential_sources_registry_has_expected_steps():
"""Sanity check — the registry contains the expected RemovalSteps.
Guards against accidentally dropping a step during future refactors.
If you add a new credential source, add it to the expected set below.
Adding a new credential source is routine, so this is a structural
invariant check (every step has a description, every step is unique,
core steps are present) rather than a frozen snapshot. Frozen
snapshots of catalog-like data violate the AGENTS.md "don't write
change-detector tests" rule — they break every time someone adds a
provider.
"""
from agent.credential_sources import _REGISTRY
descriptions = {step.description for step in _REGISTRY}
expected = {
descriptions = [step.description for step in _REGISTRY]
# No empty descriptions, no duplicates.
assert all(d for d in descriptions), "Every removal step must have a description"
assert len(descriptions) == len(set(descriptions)), (
f"Registry has duplicate step descriptions: {descriptions}"
)
# Core steps must be present — these are the ones the rest of the code
# assumes exist. When deliberately dropping one, update this list.
required = {
"gh auth token / COPILOT_GITHUB_TOKEN / GH_TOKEN",
"Any env-seeded credential (XAI_API_KEY, DEEPSEEK_API_KEY, etc.)",
"~/.claude/.credentials.json",
"~/.hermes/.anthropic_oauth.json",
"auth.json providers.nous",
"auth.json providers.openai-codex + ~/.codex/auth.json",
"auth.json providers.minimax-oauth",
"~/.qwen/oauth_creds.json",
"Custom provider config.yaml api_key field",
}
assert descriptions == expected, f"Registry mismatch. Got: {descriptions}"
missing = required - set(descriptions)
assert not missing, f"Registry missing required steps: {missing}"
def test_credential_sources_find_step_returns_none_for_manual():
+5
View File
@@ -526,6 +526,11 @@ class TestCmdMigrate:
class TestCmdCleanup:
"""Test the cleanup command handler."""
@pytest.fixture(autouse=True)
def _mock_openclaw_running(self):
with patch.object(claw_mod, "_detect_openclaw_processes", return_value=[]):
yield
def test_no_dirs_found(self, tmp_path, capsys):
args = Namespace(source=None, dry_run=False, yes=False)
with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[]):
@@ -72,7 +72,10 @@ class TestLoadConfigExpansion:
monkeypatch.setenv("GOOGLE_API_KEY", "gsk-test-key")
monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "1234567:ABC-token")
monkeypatch.setattr("hermes_cli.config.get_config_path", lambda: config_file)
# Patch the imported function's own globals. Other tests may reload
# hermes_cli.config, making string-target monkeypatches hit a different
# module object than this collection-time imported load_config().
monkeypatch.setitem(load_config.__globals__, "get_config_path", lambda: config_file)
config = load_config()
@@ -86,7 +89,7 @@ class TestLoadConfigExpansion:
config_file.write_text(config_yaml)
monkeypatch.delenv("NOT_SET_XYZ_123", raising=False)
monkeypatch.setattr("hermes_cli.config.get_config_path", lambda: config_file)
monkeypatch.setitem(load_config.__globals__, "get_config_path", lambda: config_file)
config = load_config()
+1 -1
View File
@@ -105,7 +105,7 @@ def test_get_container_exec_info_defaults():
)
with patch("hermes_constants.is_container", return_value=False), \
patch("hermes_cli.config.get_hermes_home", return_value=hermes_home), \
patch.dict(get_container_exec_info.__globals__, {"get_hermes_home": lambda: hermes_home}), \
patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_DEV", None)
info = get_container_exec_info()
@@ -7,9 +7,10 @@ WEB_SRC = Path(__file__).resolve().parents[2] / "web" / "src"
def test_dashboard_does_not_import_nous_ui_root_barrel():
offenders = []
for path in WEB_SRC.rglob("*.tsx"):
content = path.read_text(encoding="utf-8")
if 'from "@nous-research/ui"' in content or "from '@nous-research/ui'" in content:
offenders.append(str(path.relative_to(WEB_SRC)))
for ext in ("*.tsx", "*.ts"):
for path in WEB_SRC.rglob(ext):
content = path.read_text(encoding="utf-8")
if 'from "@nous-research/ui"' in content or "from '@nous-research/ui'" in content:
offenders.append(str(path.relative_to(WEB_SRC)))
assert offenders == []
@@ -0,0 +1,181 @@
"""Tests for ``hermes dashboard --stop`` / ``--status`` flags.
These flags share the detection + kill path with the post-``hermes update``
cleanup, so the heavy coverage of SIGTERM / SIGKILL / Windows taskkill lives
in ``test_update_stale_dashboard.py``. This file just verifies the flag
dispatch: argparse wiring, no-op when nothing is running, and correct
exit codes.
"""
from __future__ import annotations
import argparse
import sys
from unittest.mock import patch, MagicMock
import pytest
from hermes_cli.main import cmd_dashboard, _report_dashboard_status
def _ns(**kw):
"""Build an argparse.Namespace with dashboard defaults plus overrides."""
defaults = dict(
port=9119, host="127.0.0.1", no_open=False, insecure=False,
tui=False, stop=False, status=False,
)
defaults.update(kw)
return argparse.Namespace(**defaults)
class TestDashboardStatus:
def test_status_no_processes(self, capsys):
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(status=True))
assert exc.value.code == 0
out = capsys.readouterr().out
assert "No hermes dashboard processes running" in out
def test_status_with_processes(self, capsys):
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[12345, 12346]), \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(status=True))
# Status is informational — always exits 0.
assert exc.value.code == 0
out = capsys.readouterr().out
assert "2 hermes dashboard process(es) running" in out
assert "PID 12345" in out
assert "PID 12346" in out
def test_status_does_not_try_to_import_fastapi(self):
"""`--status` must not require dashboard runtime deps — it's a
process-table scan only. We prove this by making fastapi import
fail and confirming --status still succeeds."""
orig_import = __import__
def fake_import(name, *a, **kw):
if name == "fastapi":
raise ImportError("fastapi missing")
return orig_import(name, *a, **kw)
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
patch("builtins.__import__", side_effect=fake_import), \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(status=True))
assert exc.value.code == 0
class TestDashboardStop:
def test_stop_when_nothing_running(self, capsys):
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(stop=True))
assert exc.value.code == 0
out = capsys.readouterr().out
assert "No hermes dashboard processes running" in out
def test_stop_kills_and_exits_zero_when_all_killed(self, capsys):
"""After the kill, if the second scan returns empty we exit 0."""
# First scan: finds two processes. Second (verification) scan: empty.
scans = iter([[12345, 12346], []])
with patch("hermes_cli.main._find_stale_dashboard_pids",
side_effect=lambda: next(scans)), \
patch("hermes_cli.main._kill_stale_dashboard_processes") as mock_kill, \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(stop=True))
mock_kill.assert_called_once()
# --stop should pass a reason so the output doesn't say "running
# backend no longer matches the updated frontend" (that wording is
# for the post-`hermes update` path).
kwargs = mock_kill.call_args.kwargs
assert "reason" in kwargs
assert "stop" in kwargs["reason"].lower()
assert exc.value.code == 0
def test_stop_exits_nonzero_if_kill_leaves_survivors(self):
"""If the second scan still finds PIDs, we exit 1 so scripts can
detect that the stop didn't succeed (e.g. permission denied)."""
scans = iter([[12345], [12345]]) # both scans find the same PID
with patch("hermes_cli.main._find_stale_dashboard_pids",
side_effect=lambda: next(scans)), \
patch("hermes_cli.main._kill_stale_dashboard_processes"), \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(stop=True))
assert exc.value.code == 1
def test_stop_does_not_try_to_import_fastapi(self):
"""Like --status, --stop must work without dashboard runtime deps."""
orig_import = __import__
def fake_import(name, *a, **kw):
if name == "fastapi":
raise ImportError("fastapi missing")
return orig_import(name, *a, **kw)
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
patch("builtins.__import__", side_effect=fake_import), \
pytest.raises(SystemExit) as exc:
cmd_dashboard(_ns(stop=True))
assert exc.value.code == 0
class TestLifecycleFlagsTakePrecedence:
"""If both --stop and --status are set, --status wins (it's listed
first in cmd_dashboard). Neither is allowed to fall through to the
server-start path, which is the critical safety property — a user
who typed ``hermes dashboard --stop`` must not end up ALSO starting
a new server."""
def test_status_wins_over_stop(self, capsys):
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
patch("hermes_cli.main._kill_stale_dashboard_processes") as mock_kill, \
pytest.raises(SystemExit):
cmd_dashboard(_ns(status=True, stop=True))
# Kill path must NOT run when --status is also set.
mock_kill.assert_not_called()
def test_stop_does_not_fall_through_to_server_start(self):
"""Covers the worst-case regression: if --stop ever stopped exiting
early, the user would start the dashboard they just asked to stop."""
called = {"start": False}
def fake_start_server(**kw):
called["start"] = True
# Provide a fake web_server module so the import doesn't matter.
fake_ws = MagicMock()
fake_ws.start_server = fake_start_server
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
patch.dict(sys.modules, {"hermes_cli.web_server": fake_ws}), \
pytest.raises(SystemExit):
cmd_dashboard(_ns(stop=True))
assert called["start"] is False
class TestArgparseWiring:
"""Confirm the flags are exposed via the real argparse tree so
``hermes dashboard --stop`` / ``--status`` actually parse."""
def test_flags_are_registered(self):
from hermes_cli.main import main as _cli_main # noqa: F401
# Rebuild the argparse tree by re-running the section of main()
# that builds it. Cheapest way: introspect via --help on the
# already-built parser would require refactoring; instead we
# parse the flags directly via a minimal replay.
import importlib
mod = importlib.import_module("hermes_cli.main")
# Find the dashboard_parser instance by running build logic would
# be too invasive. Instead parse args as if via the CLI by
# intercepting parse_args. This is overkill for a smoke test —
# we just want to know the flags don't KeyError.
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[]), \
pytest.raises(SystemExit) as exc:
mod.cmd_dashboard(_ns(status=True))
assert exc.value.code == 0
+32
View File
@@ -161,6 +161,38 @@ def test_check_gateway_service_linger_skips_when_service_not_installed(monkeypat
assert issues == []
def test_doctor_reports_vercel_backend_diagnostics(monkeypatch, tmp_path):
monkeypatch.setenv("TERMINAL_ENV", "vercel_sandbox")
monkeypatch.setenv("TERMINAL_VERCEL_RUNTIME", "python3.13")
monkeypatch.setenv("TERMINAL_CONTAINER_DISK", "2048")
monkeypatch.setenv("VERCEL_TOKEN", "super-secret-value")
monkeypatch.delenv("VERCEL_PROJECT_ID", raising=False)
monkeypatch.setenv("VERCEL_TEAM_ID", "team")
monkeypatch.setattr(doctor_mod.importlib.util, "find_spec", lambda name: object() if name == "vercel" else None)
fake_model_tools = types.SimpleNamespace(
check_tool_availability=lambda *a, **kw: ([], []),
TOOLSET_REQUIREMENTS={},
)
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
doctor_mod.run_doctor(Namespace(fix=False))
out = buf.getvalue()
assert "Vercel runtime" in out
assert "python3.13" in out
assert "Vercel custom disk unsupported" in out
assert "Vercel auth incomplete" in out
assert "VERCEL_PROJECT_ID" in out
assert "Vercel auth mode: incomplete access token" in out
assert "Vercel auth present env: VERCEL_TOKEN, VERCEL_TEAM_ID" in out
assert "Vercel auth missing env: VERCEL_PROJECT_ID" in out
assert "super-secret-value" not in out
assert "snapshot filesystem only" in out
# ── Memory provider section (doctor should only check the *active* provider) ──
+44 -2
View File
@@ -14,6 +14,26 @@ from gateway.restart import (
)
class TestUserSystemdPrivateSocketPreflight:
def test_preflight_accepts_private_socket_without_dbus_bus(self, monkeypatch):
monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: None)
monkeypatch.setattr(gateway_cli, "_user_dbus_socket_path", lambda: Path("/tmp/missing-bus"))
monkeypatch.setattr(gateway_cli, "_user_systemd_private_socket_path", lambda: Path("/tmp/private-socket"))
monkeypatch.setattr(Path, "exists", lambda self: str(self) == "/tmp/private-socket")
gateway_cli._preflight_user_systemd(auto_enable_linger=False)
def test_wait_for_user_dbus_socket_accepts_private_socket(self, monkeypatch):
calls = []
monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("env"))
monkeypatch.setattr(gateway_cli, "_user_dbus_socket_path", lambda: Path("/tmp/missing-bus"))
monkeypatch.setattr(gateway_cli, "_user_systemd_private_socket_path", lambda: Path("/tmp/private-socket"))
monkeypatch.setattr(Path, "exists", lambda self: str(self) == "/tmp/private-socket")
assert gateway_cli._wait_for_user_dbus_socket(timeout=0.1) is True
assert calls == ["env"]
class TestSystemdServiceRefresh:
def test_systemd_install_repairs_outdated_unit_without_force(self, tmp_path, monkeypatch):
unit_path = tmp_path / "hermes-gateway.service"
@@ -235,7 +255,8 @@ class TestLaunchdServiceRecovery:
target = f"{domain}/{label}"
def fake_run(cmd, check=False, **kwargs):
calls.append(cmd)
if cmd and cmd[0] == "launchctl":
calls.append(cmd)
if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
return SimpleNamespace(returncode=0, stdout="", stderr="")
@@ -262,7 +283,8 @@ class TestLaunchdServiceRecovery:
target = f"{domain}/{label}"
def fake_run(cmd, check=False, **kwargs):
calls.append(cmd)
if cmd and cmd[0] == "launchctl":
calls.append(cmd)
if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
raise gateway_cli.subprocess.CalledProcessError(113, cmd, stderr="Could not find service")
return SimpleNamespace(returncode=0, stdout="", stderr="")
@@ -1105,6 +1127,10 @@ class TestPreflightUserSystemd:
gateway_cli, "_user_dbus_socket_path",
lambda: type("P", (), {"exists": lambda self: True})(),
)
monkeypatch.setattr(
gateway_cli, "_user_systemd_private_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
# Should not raise, no subprocess calls needed.
gateway_cli._preflight_user_systemd()
@@ -1114,6 +1140,10 @@ class TestPreflightUserSystemd:
gateway_cli, "_user_dbus_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "_user_systemd_private_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "get_systemd_linger_status", lambda: (False, ""),
)
@@ -1142,6 +1172,10 @@ class TestPreflightUserSystemd:
gateway_cli, "_user_dbus_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "_user_systemd_private_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "get_systemd_linger_status",
lambda: (None, "loginctl not found"),
@@ -1159,6 +1193,10 @@ class TestPreflightUserSystemd:
gateway_cli, "_user_dbus_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "_user_systemd_private_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "get_systemd_linger_status", lambda: (True, ""),
)
@@ -1177,6 +1215,10 @@ class TestPreflightUserSystemd:
gateway_cli, "_user_dbus_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "_user_systemd_private_socket_path",
lambda: type("P", (), {"exists": lambda self: False})(),
)
monkeypatch.setattr(
gateway_cli, "get_systemd_linger_status", lambda: (False, ""),
)
@@ -224,22 +224,21 @@ class TestArgparseFlagsRegistered:
assert args.ignore_rules is True
def test_main_py_registers_both_flags(self):
"""E2E: the real hermes_cli/main.py parser accepts both flags.
"""E2E: the real hermes parser accepts both flags."""
from hermes_cli._parser import build_top_level_parser
We invoke the real argparse tree builder from hermes_cli.main.
"""
import hermes_cli.main as hm
parser, _subparsers, chat_parser = build_top_level_parser()
top_dests = {a.dest for a in parser._actions}
chat_dests = {a.dest for a in chat_parser._actions}
assert "ignore_user_config" in top_dests
assert "ignore_rules" in top_dests
assert "ignore_user_config" in chat_dests
assert "ignore_rules" in chat_dests
# hm has a helper that builds the argparse tree inside main().
# We can extract it by catching the SystemExit on --help.
# Simpler: just grep the source for the flag strings. Both approaches
# are brittle; we use a combined test.
import inspect
src = inspect.getsource(hm)
assert '"--ignore-user-config"' in src, \
"chat subparser must register --ignore-user-config"
assert '"--ignore-rules"' in src, \
"chat subparser must register --ignore-rules"
# And the cmd_chat env-var wiring must be present
import inspect
import hermes_cli.main as hm
src = inspect.getsource(hm)
assert "HERMES_IGNORE_USER_CONFIG" in src
assert "HERMES_IGNORE_RULES" in src
@@ -0,0 +1,91 @@
"""Tests for the approvals.mcp_reload_confirm config gate.
When the user runs /reload-mcp, the MCP tool set is rebuilt which
invalidates the provider prompt cache for the active session. That's
expensive on long-context / high-reasoning models. The config gate
adds a three-option confirmation (Approve Once / Always Approve /
Cancel); "Always Approve" flips this key to false so subsequent reloads
run silently.
"""
from __future__ import annotations
from copy import deepcopy
from hermes_cli.config import DEFAULT_CONFIG
class TestMcpReloadConfirmDefault:
def test_default_config_has_the_key(self):
approvals = DEFAULT_CONFIG.get("approvals")
assert isinstance(approvals, dict)
assert "mcp_reload_confirm" in approvals
def test_default_is_true(self):
# New installs confirm by default — this is the safe behavior.
assert DEFAULT_CONFIG["approvals"]["mcp_reload_confirm"] is True
def test_shape_matches_other_approval_keys(self):
# Same flat dict level as `mode` / `timeout` / `cron_mode`.
approvals = DEFAULT_CONFIG["approvals"]
assert isinstance(approvals.get("mode"), str)
assert isinstance(approvals.get("timeout"), int)
assert isinstance(approvals.get("cron_mode"), str)
assert isinstance(approvals.get("mcp_reload_confirm"), bool)
class TestUserConfigMerge:
"""If a user has a pre-existing config without this key, load_config
should fill it in from DEFAULT_CONFIG (deep merge preserves keys the
user didn't override).
"""
def test_existing_user_config_without_key_gets_default(self, tmp_path, monkeypatch):
import yaml
# Simulate a legacy user config without the new key.
home = tmp_path / ".hermes"
home.mkdir()
cfg_path = home / "config.yaml"
legacy = {
"approvals": {"mode": "manual", "timeout": 60, "cron_mode": "deny"},
}
cfg_path.write_text(yaml.safe_dump(legacy))
monkeypatch.setenv("HERMES_HOME", str(home))
# Force a fresh reimport of config.py so the HERMES_HOME is honored.
import importlib
import hermes_cli.config as cfg_mod
importlib.reload(cfg_mod)
cfg = cfg_mod.load_config()
assert cfg["approvals"]["mcp_reload_confirm"] is True
def test_existing_user_config_with_false_key_survives_merge(
self, tmp_path, monkeypatch,
):
"""A user who has clicked "Always Approve" (key=false) must keep
that setting across reloads — the default_true value must not win.
"""
import yaml
home = tmp_path / ".hermes"
home.mkdir()
cfg_path = home / "config.yaml"
user_cfg = {
"approvals": {
"mode": "manual",
"timeout": 60,
"cron_mode": "deny",
"mcp_reload_confirm": False,
},
}
cfg_path.write_text(yaml.safe_dump(user_cfg))
monkeypatch.setenv("HERMES_HOME", str(home))
import importlib
import hermes_cli.config as cfg_mod
importlib.reload(cfg_mod)
cfg = cfg_mod.load_config()
assert cfg["approvals"]["mcp_reload_confirm"] is False
+17
View File
@@ -188,6 +188,23 @@ class TestCreateProfile:
assert not (profile_dir / "gateway_state.json").exists()
assert not (profile_dir / "processes.json").exists()
def test_clone_all_excludes_sibling_profiles_tree(self, profile_env):
"""--clone-all from default ~/.hermes must not copy profiles/* (nested explosion)."""
tmp_path = profile_env
default_home = tmp_path / ".hermes"
profiles_root = default_home / "profiles"
profiles_root.mkdir(exist_ok=True)
(profiles_root / "other").mkdir(parents=True, exist_ok=True)
(profiles_root / "other" / "marker.txt").write_text("sibling data")
(default_home / "memories").mkdir(exist_ok=True)
(default_home / "memories" / "note.md").write_text("remember this")
profile_dir = create_profile("coder", clone_all=True, no_alias=True)
assert (profile_dir / "memories" / "note.md").read_text() == "remember this"
assert not (profile_dir / "profiles").exists()
def test_clone_config_missing_files_skipped(self, profile_env):
"""Clone config gracefully skips files that don't exist in source."""
profile_dir = create_profile("coder", clone_config=True, no_alias=True)
+10 -3
View File
@@ -96,10 +96,17 @@ class TestPtyBridgeIO:
@skip_on_windows
class TestPtyBridgeResize:
def test_resize_updates_child_winsize(self):
# tput reads COLUMNS/LINES from the TTY ioctl (TIOCGWINSZ).
# Spawn a shell, resize, then ask tput for the dimensions.
# Query the TTY ioctl directly instead of using tput, which requires
# TERM and fails in GitHub Actions' non-interactive environment.
winsize_script = (
"import fcntl, struct, termios, time; "
"time.sleep(0.1); "
"rows, cols, *_ = struct.unpack('HHHH', "
"fcntl.ioctl(0, termios.TIOCGWINSZ, b'\\0' * 8)); "
"print(cols); print(rows)"
)
bridge = PtyBridge.spawn(
["/bin/sh", "-c", "sleep 0.1; tput cols; tput lines"],
[sys.executable, "-c", winsize_script],
cols=80,
rows=24,
)
+155
View File
@@ -0,0 +1,155 @@
"""Tests for hermes_cli.relaunch — unified self-relaunch utility."""
import sys
import pytest
from hermes_cli import relaunch as relaunch_mod
class TestResolveHermesBin:
def test_prefers_absolute_argv0_when_executable(self, monkeypatch):
fake = "/nix/store/abc/bin/hermes"
monkeypatch.setattr(sys, "argv", [fake])
monkeypatch.setattr(relaunch_mod.os.path, "isfile", lambda p: p == fake)
monkeypatch.setattr(relaunch_mod.os, "access", lambda p, mode: p == fake)
assert relaunch_mod.resolve_hermes_bin() == fake
def test_resolves_relative_argv0(self, monkeypatch, tmp_path):
fake = tmp_path / "hermes"
fake.write_text("#!/bin/sh\n")
fake.chmod(0o755)
monkeypatch.setattr(sys, "argv", [str(fake.name)])
monkeypatch.chdir(tmp_path)
# Ensure we don't accidentally match a real 'hermes' on PATH
monkeypatch.setattr(relaunch_mod.shutil, "which", lambda _name: None)
assert relaunch_mod.resolve_hermes_bin() == str(fake)
def test_falls_back_to_path_which(self, monkeypatch):
monkeypatch.setattr(sys, "argv", ["-c"]) # not a real path
monkeypatch.setattr(
relaunch_mod.shutil, "which", lambda name: "/usr/bin/hermes" if name == "hermes" else None
)
assert relaunch_mod.resolve_hermes_bin() == "/usr/bin/hermes"
def test_returns_none_when_unresolvable(self, monkeypatch):
monkeypatch.setattr(sys, "argv", ["-c"])
monkeypatch.setattr(relaunch_mod.shutil, "which", lambda _name: None)
assert relaunch_mod.resolve_hermes_bin() is None
class TestExtractInheritedFlags:
def test_extracts_tui_and_dev(self):
argv = ["--tui", "--dev", "chat"]
assert relaunch_mod._extract_inherited_flags(argv) == ["--tui", "--dev"]
def test_extracts_profile_with_value(self):
argv = ["--profile", "work", "chat"]
assert relaunch_mod._extract_inherited_flags(argv) == ["--profile", "work"]
def test_extracts_short_p_with_value(self):
argv = ["-p", "work"]
assert relaunch_mod._extract_inherited_flags(argv) == ["-p", "work"]
def test_extracts_equals_form(self):
argv = ["--profile=work", "--model=anthropic/claude-sonnet-4"]
assert relaunch_mod._extract_inherited_flags(argv) == [
"--profile=work",
"--model=anthropic/claude-sonnet-4",
]
def test_skips_unknown_flags(self):
argv = ["--foo", "bar", "--tui"]
assert relaunch_mod._extract_inherited_flags(argv) == ["--tui"]
def test_does_not_consume_flag_like_value(self):
argv = ["--tui", "--resume", "abc123"]
assert relaunch_mod._extract_inherited_flags(argv) == ["--tui"]
def test_preserves_multiple_skills(self):
argv = ["-s", "foo", "-s", "bar", "--tui"]
assert relaunch_mod._extract_inherited_flags(argv) == ["-s", "foo", "-s", "bar", "--tui"]
class TestInheritedFlagTable:
"""Sanity-check the argparse-introspected table that drives extraction."""
def test_short_and_long_aliases_are_paired(self):
table = dict(relaunch_mod._INHERITED_FLAGS_TABLE)
# Each pair declared together in the parser shares takes_value.
for short, long_ in [
("-p", "--profile"),
("-m", "--model"),
("-s", "--skills"),
]:
assert table[short] == table[long_], f"{short}/{long_} disagree"
def test_store_true_flags_do_not_take_value(self):
table = dict(relaunch_mod._INHERITED_FLAGS_TABLE)
for flag in ["--tui", "--dev", "--yolo", "--ignore-user-config", "--ignore-rules"]:
assert table[flag] is False, f"{flag} should not take a value"
def test_value_flags_take_value(self):
table = dict(relaunch_mod._INHERITED_FLAGS_TABLE)
for flag in ["--profile", "--model", "--provider", "--skills"]:
assert table[flag] is True, f"{flag} should take a value"
def test_excluded_flags_are_not_inherited(self):
table = dict(relaunch_mod._INHERITED_FLAGS_TABLE)
# --worktree creates a new worktree per process; inheriting would
# orphan the parent's. Chat-only flags (--quiet/-Q, --verbose/-v,
# --source) can't be in argv at the existing relaunch callsites.
for flag in ["-w", "--worktree", "-Q", "--quiet", "-v", "--verbose", "--source"]:
assert flag not in table, f"{flag} should not be inherited"
class TestBuildRelaunchArgv:
def test_uses_bin_when_available(self, monkeypatch):
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: "/usr/bin/hermes")
argv = relaunch_mod.build_relaunch_argv(["--resume", "abc"])
assert argv[0] == "/usr/bin/hermes"
def test_falls_back_to_python_module(self, monkeypatch):
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: None)
argv = relaunch_mod.build_relaunch_argv(["--resume", "abc"])
assert argv == [sys.executable, "-m", "hermes_cli.main", "--resume", "abc"]
def test_preserves_inherited_flags(self, monkeypatch):
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: "/usr/bin/hermes")
original = ["--tui", "--dev", "--profile", "work", "sessions", "browse"]
argv = relaunch_mod.build_relaunch_argv(["--resume", "abc"], original_argv=original)
assert "--tui" in argv
assert "--dev" in argv
assert "--profile" in argv
assert "work" in argv
assert "--resume" in argv
assert "abc" in argv
# The original subcommand should not survive
assert "sessions" not in argv
assert "browse" not in argv
def test_can_disable_preserve(self, monkeypatch):
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: "/usr/bin/hermes")
original = ["--tui", "chat"]
argv = relaunch_mod.build_relaunch_argv(
["--resume", "abc"], preserve_inherited=False, original_argv=original
)
assert "--tui" not in argv
assert argv == ["/usr/bin/hermes", "--resume", "abc"]
class TestRelaunch:
def test_calls_execvp(self, monkeypatch):
calls = []
def fake_execvp(path, argv):
calls.append((path, argv))
raise SystemExit(0)
monkeypatch.setattr(relaunch_mod.os, "execvp", fake_execvp)
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: "/usr/bin/hermes")
with pytest.raises(SystemExit):
relaunch_mod.relaunch(["--resume", "abc"])
assert calls == [("/usr/bin/hermes", ["/usr/bin/hermes", "--resume", "abc"])]
@@ -1998,6 +1998,7 @@ class TestAzureAnthropicEnvVarHint:
assert resolved["api_key"] == "fallback-works"
def test_no_key_anywhere_raises_helpful_error(self, monkeypatch):
"""When nothing resolves, the error message mentions key_env as an option."""
monkeypatch.delenv("AZURE_ANTHROPIC_KEY", raising=False)
@@ -2168,3 +2169,67 @@ class TestTencentTokenhubRuntimeResolution:
assert resolved["base_url"] == "https://explicit-proxy.example.com/v1"
assert resolved["source"] == "explicit"
# ---------------------------------------------------------------------------
# minimax-oauth runtime resolution tests (added by feat/minimax-oauth-provider)
# ---------------------------------------------------------------------------
def test_minimax_oauth_runtime_returns_anthropic_messages_mode(monkeypatch):
"""resolve_runtime_provider for minimax-oauth must return api_mode='anthropic_messages'."""
from hermes_cli.auth import MINIMAX_OAUTH_GLOBAL_INFERENCE
monkeypatch.setattr(rp, "resolve_provider", lambda *a, **k: "minimax-oauth")
monkeypatch.setattr(rp, "_get_model_config", lambda: {"provider": "minimax-oauth"})
monkeypatch.setattr(rp, "load_pool", lambda provider: None)
monkeypatch.setattr(
rp,
"_resolve_named_custom_runtime",
lambda **k: None,
)
monkeypatch.setattr(
rp,
"_resolve_explicit_runtime",
lambda **k: None,
)
fake_creds = {
"provider": "minimax-oauth",
"api_key": "mock-access-token",
"base_url": MINIMAX_OAUTH_GLOBAL_INFERENCE.rstrip("/"),
"source": "oauth",
}
import hermes_cli.auth as auth_mod
monkeypatch.setattr(auth_mod, "resolve_minimax_oauth_runtime_credentials",
lambda **k: fake_creds)
resolved = rp.resolve_runtime_provider(requested="minimax-oauth")
assert resolved["provider"] == "minimax-oauth"
assert resolved["api_mode"] == "anthropic_messages"
assert resolved["api_key"] == "mock-access-token"
def test_minimax_oauth_runtime_uses_inference_base_url(monkeypatch):
"""Base URL returned by resolve_runtime_provider should match the OAuth credentials."""
from hermes_cli.auth import MINIMAX_OAUTH_CN_INFERENCE
monkeypatch.setattr(rp, "resolve_provider", lambda *a, **k: "minimax-oauth")
monkeypatch.setattr(rp, "_get_model_config", lambda: {"provider": "minimax-oauth"})
monkeypatch.setattr(rp, "load_pool", lambda provider: None)
monkeypatch.setattr(rp, "_resolve_named_custom_runtime", lambda **k: None)
monkeypatch.setattr(rp, "_resolve_explicit_runtime", lambda **k: None)
fake_creds = {
"provider": "minimax-oauth",
"api_key": "cn-token",
"base_url": MINIMAX_OAUTH_CN_INFERENCE.rstrip("/"),
"source": "oauth",
}
import hermes_cli.auth as auth_mod
monkeypatch.setattr(auth_mod, "resolve_minimax_oauth_runtime_credentials",
lambda **k: fake_creds)
resolved = rp.resolve_runtime_provider(requested="minimax-oauth")
assert MINIMAX_OAUTH_CN_INFERENCE.rstrip("/") in resolved["base_url"]
@@ -127,6 +127,13 @@ class TestConfigYamlRouting:
or "TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE=True" in env_content
)
def test_terminal_vercel_runtime_goes_to_config_and_env(self, _isolated_hermes_home):
set_config_value("terminal.vercel_runtime", "python3.13")
config = _read_config(_isolated_hermes_home)
env_content = _read_env(_isolated_hermes_home)
assert "vercel_runtime: python3.13" in config
assert "TERMINAL_VERCEL_RUNTIME=python3.13" in env_content
# ---------------------------------------------------------------------------
# Empty / falsy values — regression tests for #4277
+107 -21
View File
@@ -1,5 +1,6 @@
"""Tests for setup.py configuration flows."""
import json
import os
import sys
import types
@@ -29,6 +30,17 @@ def _clear_provider_env(monkeypatch):
monkeypatch.delenv(key, raising=False)
def _clear_vercel_env(monkeypatch):
for key in (
"TERMINAL_VERCEL_RUNTIME",
"VERCEL_OIDC_TOKEN",
"VERCEL_TOKEN",
"VERCEL_PROJECT_ID",
"VERCEL_TEAM_ID",
):
monkeypatch.delenv(key, raising=False)
def _stub_tts(monkeypatch):
"""Stub out TTS prompts so setup_model_provider doesn't block."""
monkeypatch.setattr("hermes_cli.setup.prompt_choice", lambda q, c, d=0: (
@@ -162,12 +174,13 @@ def test_setup_gateway_skips_service_install_when_systemctl_missing(monkeypatch,
"WEBHOOK_ENABLED": "",
}
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(setup_mod, "get_env_value", lambda key: env.get(key, ""))
monkeypatch.setattr(gateway_mod, "get_env_value", lambda key: env.get(key, ""))
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("platform.system", lambda: "Linux")
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(gateway_mod, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_mod, "is_macos", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_installed", lambda: False)
@@ -200,12 +213,13 @@ def test_setup_gateway_in_container_shows_docker_guidance(monkeypatch, capsys):
"WEBHOOK_ENABLED": "",
}
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(setup_mod, "get_env_value", lambda key: env.get(key, ""))
monkeypatch.setattr(gateway_mod, "get_env_value", lambda key: env.get(key, ""))
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("platform.system", lambda: "Linux")
import hermes_cli.gateway as gateway_mod
monkeypatch.setattr(gateway_mod, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_mod, "is_macos", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_installed", lambda: False)
@@ -480,28 +494,91 @@ def test_modal_setup_persists_direct_mode_when_user_chooses_their_own_account(tm
assert config["terminal"]["modal_mode"] == "direct"
def test_resolve_hermes_chat_argv_prefers_which(monkeypatch):
from hermes_cli import setup as setup_mod
def test_vercel_setup_configures_access_token_auth(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_vercel_env(monkeypatch)
monkeypatch.setenv("VERCEL_OIDC_TOKEN", "old-oidc")
monkeypatch.setitem(sys.modules, "vercel", types.ModuleType("vercel"))
config = load_config()
monkeypatch.setattr(setup_mod.shutil, "which", lambda name: "/usr/local/bin/hermes" if name == "hermes" else None)
def fake_prompt_choice(question, choices, default=0):
if question == "Select terminal backend:":
return 5
raise AssertionError(f"Unexpected prompt_choice call: {question}")
assert setup_mod._resolve_hermes_chat_argv() == ["/usr/local/bin/hermes", "chat"]
prompt_values = iter(["python3.13", "yes", "2", "4096", "token", "project", "team"])
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: next(prompt_values))
def test_resolve_hermes_chat_argv_falls_back_to_module(monkeypatch):
from hermes_cli import setup as setup_mod
from hermes_cli.setup import setup_terminal_backend
monkeypatch.setattr(setup_mod.shutil, "which", lambda _name: None)
monkeypatch.setattr(setup_mod.importlib.util, "find_spec", lambda name: object() if name == "hermes_cli" else None)
setup_terminal_backend(config)
assert setup_mod._resolve_hermes_chat_argv() == [sys.executable, "-m", "hermes_cli.main", "chat"]
assert config["terminal"]["backend"] == "vercel_sandbox"
assert config["terminal"]["vercel_runtime"] == "python3.13"
assert config["terminal"]["container_disk"] == 51200
assert os.environ["TERMINAL_VERCEL_RUNTIME"] == "python3.13"
assert "VERCEL_OIDC_TOKEN" not in os.environ
assert os.environ["VERCEL_TOKEN"] == "token"
assert os.environ["VERCEL_PROJECT_ID"] == "project"
assert os.environ["VERCEL_TEAM_ID"] == "team"
def test_offer_launch_chat_execs_fresh_process(monkeypatch):
def test_vercel_setup_prefills_project_and_team_from_link_file(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_vercel_env(monkeypatch)
project_root = tmp_path / "project"
nested = project_root / "app" / "src"
nested.mkdir(parents=True)
vercel_dir = project_root / ".vercel"
vercel_dir.mkdir()
(vercel_dir / "project.json").write_text(
json.dumps({"projectId": "linked-project", "orgId": "linked-team"}),
encoding="utf-8",
)
monkeypatch.chdir(nested)
monkeypatch.setitem(sys.modules, "vercel", types.ModuleType("vercel"))
config = load_config()
config["terminal"]["container_disk"] = 999
def fake_prompt_choice(question, choices, default=0):
if question == "Select terminal backend:":
return 5
raise AssertionError(f"Unexpected prompt_choice call: {question}")
prompt_values = iter(["node24", "no", "1", "5120", "token", "", ""])
defaults = {}
def fake_prompt(message, default="", **kwargs):
defaults[message] = default
value = next(prompt_values)
return value or default
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", fake_prompt)
from hermes_cli.setup import setup_terminal_backend
setup_terminal_backend(config)
assert config["terminal"]["backend"] == "vercel_sandbox"
assert config["terminal"]["container_persistent"] is False
assert config["terminal"]["container_disk"] == 51200
assert "VERCEL_OIDC_TOKEN" not in os.environ
assert os.environ["VERCEL_TOKEN"] == "token"
assert os.environ["VERCEL_PROJECT_ID"] == "linked-project"
assert os.environ["VERCEL_TEAM_ID"] == "linked-team"
assert defaults[" Vercel project ID"] == "linked-project"
assert defaults[" Vercel team ID"] == "linked-team"
def test_offer_launch_chat_relaunches_via_bin(monkeypatch):
from hermes_cli import setup as setup_mod
from hermes_cli import relaunch as relaunch_mod
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *_args, **_kwargs: True)
monkeypatch.setattr(setup_mod, "_resolve_hermes_chat_argv", lambda: ["/usr/local/bin/hermes", "chat"])
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: "/usr/local/bin/hermes")
exec_calls = []
@@ -509,7 +586,7 @@ def test_offer_launch_chat_execs_fresh_process(monkeypatch):
exec_calls.append((path, argv))
raise SystemExit(0)
monkeypatch.setattr(setup_mod.os, "execvp", fake_execvp)
monkeypatch.setattr(relaunch_mod.os, "execvp", fake_execvp)
with pytest.raises(SystemExit):
setup_mod._offer_launch_chat()
@@ -517,13 +594,22 @@ def test_offer_launch_chat_execs_fresh_process(monkeypatch):
assert exec_calls == [("/usr/local/bin/hermes", ["/usr/local/bin/hermes", "chat"])]
def test_offer_launch_chat_manual_fallback_when_unresolvable(monkeypatch, capsys):
def test_offer_launch_chat_falls_back_to_module(monkeypatch):
from hermes_cli import setup as setup_mod
from hermes_cli import relaunch as relaunch_mod
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *_args, **_kwargs: True)
monkeypatch.setattr(setup_mod, "_resolve_hermes_chat_argv", lambda: None)
monkeypatch.setattr(relaunch_mod, "resolve_hermes_bin", lambda: None)
setup_mod._offer_launch_chat()
exec_calls = []
captured = capsys.readouterr()
assert "Run 'hermes chat' manually" in captured.out
def fake_execvp(path, argv):
exec_calls.append((path, argv))
raise SystemExit(0)
monkeypatch.setattr(relaunch_mod.os, "execvp", fake_execvp)
with pytest.raises(SystemExit):
setup_mod._offer_launch_chat()
assert exec_calls == [(sys.executable, [sys.executable, "-m", "hermes_cli.main", "chat"])]
+245
View File
@@ -0,0 +1,245 @@
"""Tests for IRC gateway configuration via `hermes setup gateway` UI.
Covers the full plugin-platform discovery status configure flow so that
a fresh Hermes install (no state, no env vars) can set up IRC through the
interactive setup menus.
"""
import os
import pytest
from gateway.platform_registry import PlatformEntry, platform_registry
def _register_irc_platform(**overrides):
"""Manually register the IRC platform entry as if discover_plugins() found it.
Tests run outside the normal plugin-discovery path, so we inject the entry
directly into the singleton registry and yield its dict shape.
"""
defaults = dict(
name="irc",
label="IRC",
adapter_factory=lambda cfg: None,
check_fn=lambda: bool(os.getenv("IRC_SERVER", "") and os.getenv("IRC_CHANNEL", "")),
validate_config=None,
required_env=["IRC_SERVER", "IRC_CHANNEL", "IRC_NICKNAME"],
install_hint="No extra packages needed (stdlib only)",
setup_fn=lambda: None,
source="plugin",
plugin_name="irc_platform",
allowed_users_env="IRC_ALLOWED_USERS",
allow_all_env="IRC_ALLOW_ALL_USERS",
max_message_length=450,
pii_safe=False,
emoji="💬",
allow_update_command=True,
platform_hint="You are chatting via IRC.",
)
defaults.update(overrides)
entry = PlatformEntry(**defaults)
platform_registry.register(entry)
return {
"key": entry.name,
"label": entry.label,
"emoji": entry.emoji,
"token_var": entry.required_env[0] if entry.required_env else "",
"install_hint": entry.install_hint,
"_registry_entry": entry,
}
def _unregister_irc_platform():
platform_registry.unregister("irc")
# ── Fresh-install discovery ─────────────────────────────────────────────────
class TestIRCFreshInstallDiscovery:
"""IRC appears in the setup menu on a brand-new Hermes install."""
def test_irc_appears_in_all_platforms(self, monkeypatch):
"""When the IRC plugin is registered, _all_platforms() surfaces it."""
import hermes_cli.gateway as gateway_mod
_register_irc_platform()
try:
# Ensure no stale env vars leak in
for key in ("IRC_SERVER", "IRC_CHANNEL", "IRC_NICKNAME"):
monkeypatch.delenv(key, raising=False)
platforms = gateway_mod._all_platforms()
keys = {p["key"] for p in platforms}
assert "irc" in keys
irc_plat = next(p for p in platforms if p["key"] == "irc")
assert irc_plat["label"] == "IRC"
assert irc_plat["emoji"] == "💬"
finally:
_unregister_irc_platform()
def test_irc_status_not_configured_when_fresh(self, monkeypatch):
"""On a fresh install with no env vars, IRC shows 'not configured'."""
import hermes_cli.gateway as gateway_mod
plat = _register_irc_platform()
try:
for key in ("IRC_SERVER", "IRC_CHANNEL", "IRC_NICKNAME"):
monkeypatch.delenv(key, raising=False)
status = gateway_mod._platform_status(plat)
assert status == "not configured"
finally:
_unregister_irc_platform()
def test_irc_status_configured_when_env_set(self, monkeypatch):
"""After the user sets IRC_SERVER and IRC_CHANNEL, status is 'configured'."""
import hermes_cli.gateway as gateway_mod
plat = _register_irc_platform()
try:
monkeypatch.setenv("IRC_SERVER", "irc.libera.chat")
monkeypatch.setenv("IRC_CHANNEL", "#hermes")
monkeypatch.setenv("IRC_NICKNAME", "hermes-bot")
status = gateway_mod._platform_status(plat)
assert status == "configured"
finally:
_unregister_irc_platform()
def test_irc_status_partial_when_only_server_set(self, monkeypatch):
"""If only IRC_SERVER is set, the platform is still not configured."""
import hermes_cli.gateway as gateway_mod
plat = _register_irc_platform()
try:
monkeypatch.delenv("IRC_CHANNEL", raising=False)
monkeypatch.delenv("IRC_NICKNAME", raising=False)
monkeypatch.setenv("IRC_SERVER", "irc.libera.chat")
status = gateway_mod._platform_status(plat)
assert status == "not configured"
finally:
_unregister_irc_platform()
# ── Interactive setup dispatch ──────────────────────────────────────────────
class TestIRCInteractiveSetup:
"""The setup UI dispatches to IRC's interactive_setup() correctly."""
def test_configure_platform_dispatches_to_irc_setup_fn(self, monkeypatch, capsys):
"""_configure_platform() calls the IRC plugin's setup_fn when selected."""
import hermes_cli.gateway as gateway_mod
calls = []
def fake_setup():
calls.append("setup_called")
print("IRC setup complete!")
plat = _register_irc_platform(setup_fn=fake_setup)
try:
gateway_mod._configure_platform(plat)
finally:
_unregister_irc_platform()
assert "setup_called" in calls
out = capsys.readouterr().out
assert "IRC setup complete!" in out
def test_configure_platform_fallback_when_no_setup_fn(self, monkeypatch, capsys):
"""A plugin with no setup_fn falls back to env-var instructions."""
import hermes_cli.gateway as gateway_mod
plat = _register_irc_platform(setup_fn=None)
try:
gateway_mod._configure_platform(plat)
finally:
_unregister_irc_platform()
out = capsys.readouterr().out
assert "IRC" in out
assert "IRC_SERVER" in out
# ── End-to-end fresh-install gateway setup ──────────────────────────────────
class TestIRCGatewaySetupFreshInstall:
"""Simulate the full `hermes setup gateway` experience with IRC present."""
def test_setup_gateway_shows_irc_in_platform_menu(self, monkeypatch, capsys, tmp_path):
"""The gateway setup menu lists IRC among the available platforms."""
import hermes_cli.gateway as gateway_mod
from hermes_cli import setup as setup_mod
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_register_irc_platform()
try:
for key in ("IRC_SERVER", "IRC_CHANNEL", "IRC_NICKNAME"):
monkeypatch.delenv(key, raising=False)
# Sanity-check: IRC must be visible to _all_platforms()
platforms = gateway_mod._all_platforms()
assert any(p["key"] == "irc" for p in platforms), \
f"IRC not in platforms: {[p['key'] for p in platforms]}"
# Capture what prompt_checklist is asked to display
checklist_calls = []
def capture_prompt_checklist(question, choices, pre_selected=None):
checklist_calls.append({"question": question, "choices": choices})
return [] # nothing selected → clean exit
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *a, **kw: False)
monkeypatch.setattr(setup_mod, "prompt_checklist", capture_prompt_checklist)
monkeypatch.setattr(gateway_mod, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_mod, "is_macos", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_installed", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: False)
setup_mod.setup_gateway({})
# Find the platform-selection prompt
platform_prompt = next(
(c for c in checklist_calls if "platform" in c["question"].lower()),
None,
)
assert platform_prompt is not None, \
f"No platform prompt found in {checklist_calls}"
choices_text = "\n".join(platform_prompt["choices"])
assert "IRC" in choices_text
assert "💬" in choices_text
assert "not configured" in choices_text.lower()
finally:
_unregister_irc_platform()
def test_setup_gateway_irc_counts_as_messaging_platform(self, monkeypatch, capsys, tmp_path):
"""When IRC is configured, setup_gateway counts it as a messaging platform."""
import hermes_cli.gateway as gateway_mod
from hermes_cli import setup as setup_mod
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_register_irc_platform()
try:
monkeypatch.setenv("IRC_SERVER", "irc.libera.chat")
monkeypatch.setenv("IRC_CHANNEL", "#hermes")
monkeypatch.setenv("IRC_NICKNAME", "hermes-bot")
monkeypatch.setattr(setup_mod, "prompt_yes_no", lambda *a, **kw: False)
monkeypatch.setattr(setup_mod, "prompt_choice", lambda *a, **kw: 0)
monkeypatch.setattr(gateway_mod, "supports_systemd_services", lambda: False)
monkeypatch.setattr(gateway_mod, "is_macos", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_installed", lambda: False)
monkeypatch.setattr(gateway_mod, "_is_service_running", lambda: False)
setup_mod.setup_gateway({})
out = capsys.readouterr().out
assert "Messaging platforms configured!" in out
finally:
_unregister_irc_platform()
@@ -419,7 +419,12 @@ class TestGetSectionConfigSummary:
return "disc456"
return ""
with patch.object(setup_mod, "get_env_value", side_effect=env_side):
# Also patch gateway module's binding since _platform_status()
# reads from hermes_cli.gateway.get_env_value after the setup
# flows were unified via platform_registry.
import hermes_cli.gateway as gateway_mod
with patch.object(setup_mod, "get_env_value", side_effect=env_side), \
patch.object(gateway_mod, "get_env_value", side_effect=env_side):
result = setup_mod._get_section_config_summary({}, "gateway")
assert "Telegram" in result
assert "Discord" in result
@@ -471,7 +476,9 @@ class TestGetSectionConfigSummary:
def env_side(key):
return "true" if key == "WHATSAPP_ENABLED" else ""
with patch.object(setup_mod, "get_env_value", side_effect=env_side):
import hermes_cli.gateway as gateway_mod
with patch.object(setup_mod, "get_env_value", side_effect=env_side), \
patch.object(gateway_mod, "get_env_value", side_effect=env_side):
result = setup_mod._get_section_config_summary({}, "gateway")
assert result is not None
assert "WhatsApp" in result
@@ -481,7 +488,9 @@ class TestGetSectionConfigSummary:
def env_side(key):
return "http://signal.local" if key == "SIGNAL_HTTP_URL" else ""
with patch.object(setup_mod, "get_env_value", side_effect=env_side):
import hermes_cli.gateway as gateway_mod
with patch.object(setup_mod, "get_env_value", side_effect=env_side), \
patch.object(gateway_mod, "get_env_value", side_effect=env_side):
result = setup_mod._get_section_config_summary({}, "gateway")
assert result is not None
assert "Signal" in result
@@ -529,13 +538,28 @@ class TestGetSectionConfigSummary:
assert result == "gpt-5"
def test_gateway_matches_platform_registry(self):
"""Every platform in _GATEWAY_PLATFORMS should be recognised by its
own env-var sentinel i.e. the summary must not drift from the
"""Every built-in platform should be recognised by its primary
env-var sentinel i.e. the summary must not drift from the
registry used by the setup checklist."""
for label, env_var, _fn in setup_mod._GATEWAY_PLATFORMS:
from hermes_cli.gateway import _PLATFORMS
for plat in _PLATFORMS:
label = plat["label"]
env_var = plat.get("token_var")
if not env_var:
continue
# Some platforms require a specific value shape (e.g. WhatsApp
# needs the literal "true"). Use a sentinel that satisfies every
# real validator _platform_status() currently checks.
def env_side(key, _target=env_var):
return "x" if key == _target else ""
with patch.object(setup_mod, "get_env_value", side_effect=env_side):
if key != _target:
return ""
if _target == "WHATSAPP_ENABLED":
return "true"
return "x"
import hermes_cli.gateway as gateway_mod
with patch.object(setup_mod, "get_env_value", side_effect=env_side), \
patch.object(gateway_mod, "get_env_value", side_effect=env_side):
result = setup_mod._get_section_config_summary({}, "gateway")
expected = setup_mod._gateway_platform_short_label(label)
assert result is not None, f"{label} ({env_var}) not recognised"
+30
View File
@@ -79,3 +79,33 @@ def test_show_status_reports_nous_auth_error(monkeypatch, capsys, tmp_path):
assert "Error: Refresh session has been revoked" in output
assert "Access exp:" in output
assert "Key exp:" in output
def test_show_status_reports_vercel_backend_contract(monkeypatch, capsys, tmp_path):
from hermes_cli import status as status_mod
import hermes_cli.auth as auth_mod
import hermes_cli.gateway as gateway_mod
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("TERMINAL_ENV", "vercel_sandbox")
monkeypatch.setenv("TERMINAL_VERCEL_RUNTIME", "python3.13")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "true")
monkeypatch.setenv("VERCEL_OIDC_TOKEN", "oidc-token")
monkeypatch.setattr(status_mod.importlib.util, "find_spec", lambda name: object() if name == "vercel" else None)
monkeypatch.setattr(status_mod, "load_config", lambda: {"terminal": {"backend": "vercel_sandbox"}}, raising=False)
monkeypatch.setattr(auth_mod, "get_nous_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_codex_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(auth_mod, "get_qwen_auth_status", lambda: {}, raising=False)
monkeypatch.setattr(gateway_mod, "find_gateway_pids", lambda exclude_pids=None: [], raising=False)
status_mod.show_status(SimpleNamespace(all=False, deep=False))
output = capsys.readouterr().out
assert "Backend: vercel_sandbox" in output
assert "Runtime: python3.13" in output
assert "Auth:" in output and "OIDC token via VERCEL_OIDC_TOKEN" in output
assert "Auth detail: mode: OIDC" in output
assert "Auth detail: active env: VERCEL_OIDC_TOKEN" in output
assert "oidc-token" not in output
assert "snapshot filesystem" in output
assert "live processes do not survive" in output
+191 -6
View File
@@ -12,6 +12,7 @@ def _args(**overrides):
"model": None,
"provider": None,
"resume": None,
"toolsets": None,
"tui": True,
"tui_dev": False,
}
@@ -35,7 +36,7 @@ def test_cmd_chat_tui_continue_uses_latest_tui_session(monkeypatch, main_mod):
calls.append(source)
return "20260408_235959_a1b2c3" if source == "tui" else None
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None):
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None, toolsets=None):
captured["resume"] = resume_session_id
raise SystemExit(0)
@@ -62,7 +63,7 @@ def test_cmd_chat_tui_continue_falls_back_to_latest_cli_session(monkeypatch, mai
return "20260408_235959_d4e5f6"
return None
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None):
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None, toolsets=None):
captured["resume"] = resume_session_id
raise SystemExit(0)
@@ -80,7 +81,7 @@ def test_cmd_chat_tui_continue_falls_back_to_latest_cli_session(monkeypatch, mai
def test_cmd_chat_tui_resume_resolves_title_before_launch(monkeypatch, main_mod):
captured = {}
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None):
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None, toolsets=None):
captured["resume"] = resume_session_id
raise SystemExit(0)
@@ -98,12 +99,13 @@ def test_cmd_chat_tui_resume_resolves_title_before_launch(monkeypatch, main_mod)
def test_cmd_chat_tui_passes_model_and_provider(monkeypatch, main_mod):
captured = {}
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None):
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None, toolsets=None):
captured.update(
{
"model": model,
"provider": provider,
"resume": resume_session_id,
"toolsets": toolsets,
"tui_dev": tui_dev,
}
)
@@ -120,11 +122,193 @@ def test_cmd_chat_tui_passes_model_and_provider(monkeypatch, main_mod):
"model": "anthropic/claude-sonnet-4.6",
"provider": "anthropic",
"resume": None,
"toolsets": None,
"tui_dev": False,
}
def test_launch_tui_exports_model_and_provider(monkeypatch, main_mod):
def test_cmd_chat_tui_passes_toolsets(monkeypatch, main_mod):
captured = {}
def fake_launch(resume_session_id=None, tui_dev=False, model=None, provider=None, toolsets=None):
captured["toolsets"] = toolsets
raise SystemExit(0)
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
with pytest.raises(SystemExit):
main_mod.cmd_chat(_args(toolsets="web,terminal"))
assert captured["toolsets"] == "web,terminal"
def test_main_top_level_tui_accepts_toolsets(monkeypatch, main_mod):
captured = {}
import hermes_cli.config as config_mod
monkeypatch.setattr(sys, "argv", ["hermes", "--tui", "--toolsets", "web,terminal"])
monkeypatch.setitem(sys.modules, "hermes_cli.plugins", types.SimpleNamespace(discover_plugins=lambda: None))
monkeypatch.setitem(sys.modules, "tools.mcp_tool", types.SimpleNamespace(discover_mcp_tools=lambda: None))
monkeypatch.setattr(config_mod, "load_config", lambda: {})
monkeypatch.setattr(config_mod, "get_container_exec_info", lambda: None)
monkeypatch.setitem(
sys.modules,
"agent.shell_hooks",
types.SimpleNamespace(register_from_config=lambda _cfg, accept_hooks=False: None),
)
monkeypatch.setattr(main_mod, "cmd_chat", lambda args: captured.update({"toolsets": args.toolsets, "tui": args.tui}))
main_mod.main()
assert captured == {"toolsets": "web,terminal", "tui": True}
def test_main_top_level_oneshot_accepts_toolsets(monkeypatch, main_mod):
captured = {}
import hermes_cli.config as config_mod
monkeypatch.setattr(sys, "argv", ["hermes", "-z", "hello", "--toolsets", "web,terminal"])
monkeypatch.setitem(sys.modules, "hermes_cli.plugins", types.SimpleNamespace(discover_plugins=lambda: None))
monkeypatch.setitem(sys.modules, "tools.mcp_tool", types.SimpleNamespace(discover_mcp_tools=lambda: None))
monkeypatch.setattr(config_mod, "load_config", lambda: {})
monkeypatch.setattr(config_mod, "get_container_exec_info", lambda: None)
monkeypatch.setitem(
sys.modules,
"agent.shell_hooks",
types.SimpleNamespace(register_from_config=lambda _cfg, accept_hooks=False: None),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.oneshot",
types.SimpleNamespace(run_oneshot=lambda prompt, **kwargs: captured.update({"prompt": prompt, **kwargs}) or 0),
)
with pytest.raises(SystemExit) as exc:
main_mod.main()
assert exc.value.code == 0
assert captured == {"prompt": "hello", "model": None, "provider": None, "toolsets": "web,terminal"}
def _stub_plugin_discovery(monkeypatch):
monkeypatch.setitem(
sys.modules,
"hermes_cli.plugins",
types.SimpleNamespace(discover_plugins=lambda: None),
)
def test_oneshot_rejects_invalid_only_toolsets(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
from hermes_cli.oneshot import run_oneshot
assert run_oneshot("hello", toolsets="nope") == 2
err = capsys.readouterr().err
assert "nope" in err
assert "did not contain any valid toolsets" in err
def test_oneshot_filters_invalid_toolsets_before_redirect(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
from hermes_cli.oneshot import _validate_explicit_toolsets
valid, error = _validate_explicit_toolsets("web,nope")
assert valid == ["web"]
assert error is None
assert "nope" in capsys.readouterr().err
def test_oneshot_all_toolsets_means_all_not_configured_cli():
from hermes_cli.oneshot import _validate_explicit_toolsets
valid, error = _validate_explicit_toolsets("all")
assert valid is None
assert error is None
def test_oneshot_all_toolsets_warns_about_ignored_extra_entries(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
from hermes_cli.oneshot import _validate_explicit_toolsets
valid, error = _validate_explicit_toolsets("all,nope")
assert valid is None
assert error is None
assert "ignoring additional entries: nope" in capsys.readouterr().err
def test_oneshot_accepts_plugin_toolset_after_discovery(monkeypatch):
import toolsets
from hermes_cli.oneshot import _validate_explicit_toolsets
discovered = {"ready": False}
original_validate = toolsets.validate_toolset
def fake_validate(name):
return name == "plugin_demo" and discovered["ready"] or original_validate(name)
monkeypatch.setattr(toolsets, "validate_toolset", fake_validate)
monkeypatch.setitem(
sys.modules,
"hermes_cli.plugins",
types.SimpleNamespace(discover_plugins=lambda: discovered.update({"ready": True})),
)
valid, error = _validate_explicit_toolsets("plugin_demo")
assert valid == ["plugin_demo"]
assert error is None
def test_oneshot_rejects_disabled_mcp_toolset(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
import hermes_cli.config as config_mod
from hermes_cli.oneshot import _validate_explicit_toolsets
monkeypatch.setattr(
config_mod,
"read_raw_config",
lambda: {"mcp_servers": {"mcp-off": {"enabled": False}}},
)
valid, error = _validate_explicit_toolsets("mcp-off")
assert valid is None
assert error == "hermes -z: --toolsets did not contain any valid toolsets.\n"
err = capsys.readouterr().err
assert "ignoring disabled MCP servers" in err
assert "mcp-off" in err
def test_oneshot_distinguishes_disabled_mcp_from_unknown(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
import hermes_cli.config as config_mod
from hermes_cli.oneshot import _validate_explicit_toolsets
monkeypatch.setattr(
config_mod,
"read_raw_config",
lambda: {"mcp_servers": {"mcp-off": {"enabled": False}}},
)
valid, error = _validate_explicit_toolsets("web,mcp-off,nope")
assert valid == ["web"]
assert error is None
err = capsys.readouterr().err
assert "ignoring unknown --toolsets entries: nope" in err
assert "ignoring disabled MCP servers" in err
assert "mcp-off" in err
def test_launch_tui_exports_model_provider_and_toolsets(monkeypatch, main_mod):
captured = {}
active_path_during_call = None
@@ -144,13 +328,14 @@ def test_launch_tui_exports_model_and_provider(monkeypatch, main_mod):
monkeypatch.setattr(main_mod.subprocess, "call", fake_call)
with pytest.raises(SystemExit):
main_mod._launch_tui(model="nous/hermes-test", provider="nous")
main_mod._launch_tui(model="nous/hermes-test", provider="nous", toolsets="web, terminal")
env = captured["env"]
assert env["HERMES_MODEL"] == "nous/hermes-test"
assert env["HERMES_INFERENCE_MODEL"] == "nous/hermes-test"
assert env["HERMES_TUI_PROVIDER"] == "nous"
assert env["HERMES_INFERENCE_PROVIDER"] == "nous"
assert env["HERMES_TUI_TOOLSETS"] == "web,terminal"
active_path = Path(env["HERMES_TUI_ACTIVE_SESSION_FILE"])
assert active_path.name.startswith("hermes-tui-active-session-")
assert active_path.suffix == ".json"
+5 -2
View File
@@ -333,7 +333,10 @@ def test_cmd_update_retries_optional_extras_individually_when_all_fails(monkeypa
raise CalledProcessError(returncode=1, cmd=cmd)
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[mcp]", "--quiet"]:
return SimpleNamespace(returncode=0)
return SimpleNamespace(returncode=0)
# Catch-all must include stdout/stderr so consumers that parse
# output (e.g. the dashboard-restart `ps -A` scan added in the
# updater) don't crash on AttributeError.
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(hermes_main.subprocess, "run", fake_run)
@@ -370,7 +373,7 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
return SimpleNamespace(stdout="1\n", stderr="", returncode=0)
if cmd == ["git", "pull", "origin", "main"]:
return SimpleNamespace(stdout="Updating\n", stderr="", returncode=0)
return SimpleNamespace(returncode=0)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(hermes_main.subprocess, "run", fake_run)
+309 -92
View File
@@ -1,19 +1,62 @@
"""Tests for _warn_stale_dashboard_processes — stale dashboard detection.
"""Tests for the stale-dashboard handling run at the end of ``hermes update``.
Ensures ``hermes update`` warns the user when dashboard processes from a
previous version are still running after files on disk have been replaced.
See #16872.
``hermes update`` detects ``hermes dashboard`` processes left over from the
previous version and kills them (SIGTERM + SIGKILL grace, or ``taskkill /F``
on Windows). Without this, the running backend silently serves stale Python
against a freshly-updated JS bundle, producing 401s / empty data.
History:
- #16872 introduced the warn-only helper (``_warn_stale_dashboard_processes``).
- #17049 fixed a Windows wmic UnicodeDecodeError crash on non-UTF-8 locales.
- This file now also covers the kill semantics that replaced the warning.
"""
from __future__ import annotations
import importlib
import os
import sys
from unittest.mock import patch, MagicMock
from unittest.mock import patch, MagicMock, call
import pytest
from hermes_cli.main import _warn_stale_dashboard_processes
from hermes_cli.main import (
_find_stale_dashboard_pids,
_kill_stale_dashboard_processes,
_warn_stale_dashboard_processes, # back-compat alias
)
@pytest.fixture(autouse=True)
def _refresh_bindings_against_live_module():
"""Rebind module-level names to the *current* ``hermes_cli.main``.
Other tests in the suite (notably ``test_env_loader.py`` and
``test_skills_subparser.py``) reload or delete ``hermes_cli.main`` from
``sys.modules``. When that happens on the same xdist worker before we
run, our top-of-file ``from hermes_cli.main import ...`` bindings end
up pointing at the *old* module object. ``patch(\"hermes_cli.main.X\")``
then patches the *new* module, but the function we call still resolves
``_find_stale_dashboard_pids`` via its stale ``__globals__``, so every
patch becomes a no-op and the kill path silently returns early.
Refreshing the bindings (and the patch target) to the live module
object and keeping them consistent makes the tests immune to
ordering within the worker. The fix lives in the test module because
the two pollutants above are load-bearing for their own tests.
"""
global _find_stale_dashboard_pids
global _kill_stale_dashboard_processes
global _warn_stale_dashboard_processes
live = sys.modules.get("hermes_cli.main")
if live is None:
live = importlib.import_module("hermes_cli.main")
_find_stale_dashboard_pids = live._find_stale_dashboard_pids
_kill_stale_dashboard_processes = live._kill_stale_dashboard_processes
_warn_stale_dashboard_processes = live._warn_stale_dashboard_processes
yield
def _ps_line(pid: int, cmd: str) -> str:
@@ -21,11 +64,26 @@ def _ps_line(pid: int, cmd: str) -> str:
return f"{pid:>7} {cmd}"
class TestWarnStaleDashboardProcesses:
"""Unit tests for the stale dashboard process warning."""
def _ps_runner(stdout: str):
"""Build a subprocess.run side_effect that only stubs ps -A calls.
def test_no_warning_when_no_dashboard_running(self, capsys):
"""ps returns no matching processes — no warning should be printed."""
Any other subprocess.run invocation (e.g. taskkill on Windows) is
handed back as a successful no-op. This lets tests exercise the real
scan path without having to re-stub every unrelated subprocess call
made later in ``_kill_stale_dashboard_processes``.
"""
def _side_effect(args, *a, **kw):
if isinstance(args, (list, tuple)) and args and args[0] == "ps":
return MagicMock(returncode=0, stdout=stdout, stderr="")
# Any other subprocess.run (e.g. taskkill) — benign success stub.
return MagicMock(returncode=0, stdout="", stderr="")
return _side_effect
class TestFindStaleDashboardPids:
"""Unit tests for the ps/wmic-based detection step."""
def test_no_matches_returns_empty(self):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
@@ -35,26 +93,18 @@ class TestWarnStaleDashboardProcesses:
+ "\n",
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "dashboard process" not in output
assert _find_stale_dashboard_pids() == []
def test_warning_printed_for_running_dashboard(self, capsys):
"""ps finds a dashboard PID — warning with PID should appear."""
def test_matches_running_dashboard(self):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout=_ps_line(12345, "python3 -m hermes_cli.main dashboard --port 9119") + "\n",
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "1 dashboard process" in output
assert "PID 12345" in output
assert "kill <pid>" in output
assert _find_stale_dashboard_pids() == [12345]
def test_multiple_dashboard_pids(self, capsys):
"""Multiple dashboard processes — all PIDs listed."""
def test_multiple_matches(self):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
@@ -65,15 +115,9 @@ class TestWarnStaleDashboardProcesses:
]) + "\n",
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "3 dashboard process" in output
assert "PID 12345" in output
assert "PID 12346" in output
assert "PID 12347" in output
assert sorted(_find_stale_dashboard_pids()) == [12345, 12346, 12347]
def test_self_pid_excluded(self, capsys):
"""The current process PID should not be reported."""
def test_self_pid_excluded(self):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
@@ -83,41 +127,51 @@ class TestWarnStaleDashboardProcesses:
]) + "\n",
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
# The self PID may still appear inside an unrelated context, so anchor
# the check to "PID <self>" which is how the warning prints.
assert f"PID {os.getpid()}" not in output
assert "PID 12345" in output
pids = _find_stale_dashboard_pids()
assert os.getpid() not in pids
assert 12345 in pids
def test_ps_not_found_silently_ignored(self, capsys):
"""If ps is missing (FileNotFoundError), no crash, no warning."""
def test_ps_not_found_returns_empty(self):
with patch("subprocess.run", side_effect=FileNotFoundError):
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert output == ""
assert _find_stale_dashboard_pids() == []
def test_ps_timeout_silently_ignored(self, capsys):
"""If ps times out, no crash, no warning."""
def test_ps_timeout_returns_empty(self):
import subprocess as sp
with patch("subprocess.run", side_effect=sp.TimeoutExpired("ps", 10)):
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert output == ""
assert _find_stale_dashboard_pids() == []
def test_empty_ps_output_no_warning(self, capsys):
"""ps returns 0 but empty stdout — no warning."""
def test_unrelated_process_containing_word_dashboard_not_matched(self):
"""Guards against greedy pgrep-style matching catching chat sessions
or unrelated processes whose cmdline happens to contain 'dashboard'.
"""
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0, stdout="\n", stderr=""
returncode=0,
stdout="\n".join([
_ps_line(12345, "python3 -m hermes_cli.main dashboard --port 9119"),
_ps_line(22222, "python3 -m hermes_cli.main chat -q 'rewrite my dashboard'"),
_ps_line(33333, "node /opt/grafana/dashboard-server.js"),
]) + "\n",
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "dashboard process" not in output
pids = _find_stale_dashboard_pids()
assert pids == [12345]
def test_invalid_pid_lines_skipped(self, capsys):
"""Malformed ps lines should be skipped gracefully."""
def test_grep_lines_ignored(self):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout="\n".join([
_ps_line(99999, "grep hermes dashboard"),
_ps_line(12345, "hermes dashboard --port 9119"),
]) + "\n",
stderr="",
)
pids = _find_stale_dashboard_pids()
assert 99999 not in pids
assert 12345 in pids
def test_invalid_pid_lines_skipped(self):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
@@ -128,50 +182,213 @@ class TestWarnStaleDashboardProcesses:
]) + "\n",
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "PID 12345" in output
assert "1 dashboard process" in output
pids = _find_stale_dashboard_pids()
assert pids == [12345]
def test_unrelated_process_containing_word_dashboard_not_matched(self, capsys):
"""A process whose cmdline contains 'dashboard' but isn't a hermes
dashboard process must NOT be flagged. This guards against the old
``pgrep -f "hermes.*dashboard"`` greedy regex that matched e.g. a
chat session argv containing both words.
"""
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX kill semantics")
class TestKillStaleDashboardPosix:
"""Kill path on Linux / macOS: SIGTERM then SIGKILL any survivors."""
def test_no_stale_processes_is_a_noop(self, capsys):
with patch("hermes_cli.main._find_stale_dashboard_pids", return_value=[]):
_kill_stale_dashboard_processes()
assert capsys.readouterr().out == ""
def test_sigterm_graceful_exit(self, capsys):
"""Processes that exit on SIGTERM (the probe gets ProcessLookupError)
are reported as stopped and SIGKILL is never sent."""
import signal as _signal
killed_signals: list[tuple[int, int]] = []
def fake_kill(pid, sig):
killed_signals.append((pid, sig))
if sig == 0:
# Probe after SIGTERM → "process gone".
raise ProcessLookupError
# SIGTERM itself: succeed silently.
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[12345, 12346]), \
patch("os.kill", side_effect=fake_kill), \
patch("time.sleep"):
_kill_stale_dashboard_processes()
# Both got SIGTERM.
sigterms = [pid for pid, sig in killed_signals if sig == _signal.SIGTERM]
assert sorted(sigterms) == [12345, 12346]
# No SIGKILL was needed.
assert not any(sig == _signal.SIGKILL for _, sig in killed_signals)
out = capsys.readouterr().out
assert "Stopping 2 dashboard" in out
assert "✓ stopped PID 12345" in out
assert "✓ stopped PID 12346" in out
assert "Restart the dashboard" in out
def test_sigkill_fallback_for_survivors(self, capsys):
"""If a process survives SIGTERM + the grace window, SIGKILL is sent."""
import signal as _signal
sent: list[tuple[int, int]] = []
def fake_kill(pid, sig):
sent.append((pid, sig))
# Simulate stubborn process: probe (sig 0) always succeeds,
# SIGTERM does nothing, SIGKILL is where it "dies".
if sig in (_signal.SIGTERM, 0, _signal.SIGKILL):
return
# Any other signal — also fine.
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[99999]), \
patch("os.kill", side_effect=fake_kill), \
patch("time.sleep"), \
patch("time.monotonic", side_effect=[0.0] + [10.0] * 20):
# monotonic jumps past the 3s deadline on the second read so the
# grace loop exits immediately after one iteration.
_kill_stale_dashboard_processes()
signals_sent = [sig for _, sig in sent]
assert _signal.SIGTERM in signals_sent
assert _signal.SIGKILL in signals_sent
out = capsys.readouterr().out
assert "✓ stopped PID 99999" in out
def test_permission_error_is_reported_not_raised(self, capsys):
"""os.kill raising PermissionError (e.g. another user's process)
must not abort hermes update it's reported as a failure and we
move on."""
def fake_kill(pid, sig):
raise PermissionError("Operation not permitted")
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[12345]), \
patch("os.kill", side_effect=fake_kill), \
patch("time.sleep"):
_kill_stale_dashboard_processes() # must not raise
out = capsys.readouterr().out
assert "✗ failed to stop PID 12345" in out
assert "Operation not permitted" in out
def test_process_already_gone_counts_as_stopped(self, capsys):
"""ProcessLookupError on the initial SIGTERM means the process
already exited between detection and the kill treat as success."""
def fake_kill(pid, sig):
raise ProcessLookupError
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[12345]), \
patch("os.kill", side_effect=fake_kill), \
patch("time.sleep"):
_kill_stale_dashboard_processes()
out = capsys.readouterr().out
assert "✓ stopped PID 12345" in out
assert "failed to stop" not in out
class TestKillStaleDashboardWindows:
"""Kill path on Windows: taskkill /F."""
def test_taskkill_invoked_for_each_pid(self, monkeypatch, capsys):
monkeypatch.setattr(sys, "platform", "win32")
def fake_run(args, *a, **kw):
# taskkill returns 0 on success
return MagicMock(returncode=0, stdout="", stderr="")
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[12345, 12346]), \
patch("subprocess.run", side_effect=fake_run) as mock_run:
_kill_stale_dashboard_processes()
# Each PID triggered a taskkill /PID <n> /F invocation.
taskkill_calls = [
c for c in mock_run.call_args_list
if c.args and isinstance(c.args[0], list) and c.args[0][:1] == ["taskkill"]
]
assert len(taskkill_calls) == 2
assert ["taskkill", "/PID", "12345", "/F"] in [c.args[0] for c in taskkill_calls]
assert ["taskkill", "/PID", "12346", "/F"] in [c.args[0] for c in taskkill_calls]
out = capsys.readouterr().out
assert "✓ stopped PID 12345" in out
assert "✓ stopped PID 12346" in out
def test_taskkill_failure_is_reported(self, monkeypatch, capsys):
monkeypatch.setattr(sys, "platform", "win32")
def fake_run(args, *a, **kw):
return MagicMock(returncode=128, stdout="",
stderr="ERROR: Access is denied.")
with patch("hermes_cli.main._find_stale_dashboard_pids",
return_value=[12345]), \
patch("subprocess.run", side_effect=fake_run):
_kill_stale_dashboard_processes() # must not raise
out = capsys.readouterr().out
assert "✗ failed to stop PID 12345" in out
assert "Access is denied" in out
class TestBackCompatAlias:
"""``_warn_stale_dashboard_processes`` is kept as an alias for the
new kill function so old imports don't break."""
def test_alias_is_the_kill_function(self):
assert _warn_stale_dashboard_processes is _kill_stale_dashboard_processes
class TestWindowsWmicEncoding:
"""Regression tests for #17049 — the Windows wmic branch must not crash
`hermes update` on non-UTF-8 system locales (e.g. cp936 on zh-CN).
"""
def test_wmic_invoked_with_utf8_ignore_errors(self, monkeypatch):
"""The wmic subprocess.run call must pass encoding='utf-8' and
errors='ignore' so the subprocess reader thread cannot raise
UnicodeDecodeError on non-UTF-8 wmic output."""
monkeypatch.setattr(sys, "platform", "win32")
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout="\n".join([
# Legitimate dashboard — should match.
_ps_line(12345, "python3 -m hermes_cli.main dashboard --port 9119"),
# hermes running something else, with "dashboard" as a
# substring of an unrelated arg — should NOT match.
_ps_line(22222, "python3 -m hermes_cli.main chat -q 'rewrite my dashboard'"),
# Completely unrelated process mentioning dashboard.
_ps_line(33333, "node /opt/grafana/dashboard-server.js"),
]) + "\n",
stdout=(
"CommandLine=python -m hermes_cli.main dashboard\n"
"ProcessId=12345\n"
),
stderr="",
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "1 dashboard process" in output
assert "PID 12345" in output
assert "PID 22222" not in output
assert "PID 33333" not in output
_find_stale_dashboard_pids()
def test_grep_lines_ignored(self, capsys):
"""Lines containing 'grep' (from a pipe in ps output) are ignored."""
# The wmic call is the first subprocess.run invocation.
assert mock_run.called, "subprocess.run was not invoked"
wmic_call = mock_run.call_args_list[0]
kwargs = wmic_call.kwargs
assert kwargs.get("encoding") == "utf-8", (
"encoding kwarg must be 'utf-8' so wmic output is decoded "
"deterministically rather than via the implicit reader-thread "
"default that crashes on non-UTF-8 locales (#17049)."
)
assert kwargs.get("errors") == "ignore", (
"errors kwarg must be 'ignore' so undecodable bytes don't take "
"down the reader thread (#17049)."
)
def test_wmic_returns_none_stdout_does_not_crash(self, monkeypatch):
"""If subprocess.run returns successfully but stdout is None — which
is what Python 3.11 leaves behind when the reader thread silently
crashed on UnicodeDecodeError before this fix landed detection
must short-circuit instead of raising AttributeError on
``None.split('\\n')`` and aborting `hermes update` (#17049)."""
monkeypatch.setattr(sys, "platform", "win32")
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout="\n".join([
_ps_line(99999, "grep hermes dashboard"),
_ps_line(12345, "hermes dashboard --port 9119"),
]) + "\n",
stderr="",
returncode=0, stdout=None, stderr=""
)
_warn_stale_dashboard_processes()
output = capsys.readouterr().out
assert "PID 99999" not in output
assert "PID 12345" in output
# Must not raise.
assert _find_stale_dashboard_pids() == []
@@ -453,6 +453,142 @@ def test_list_authenticated_providers_no_duplicate_labels_across_schemas(monkeyp
)
def test_list_authenticated_providers_hides_custom_shadowing_builtin_endpoint(monkeypatch):
"""#16970: a custom_providers entry whose ``base_url`` matches a built-in
provider's endpoint should be hidden. The built-in row already represents
that endpoint with its canonical slug, curated model list, and auth wiring.
Repro: user sets ``DASHSCOPE_API_KEY`` (triggers the built-in ``alibaba``
row pointing at the static ``inference_base_url``) AND defines a
``my-alibaba`` custom provider pointing at the same URL. Before the fix,
the picker showed both rows for one endpoint.
"""
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test")
monkeypatch.setattr(
"agent.models_dev.fetch_models_dev",
lambda: {
"alibaba": {
"name": "Alibaba Cloud (DashScope)",
"env": ["DASHSCOPE_API_KEY"],
}
},
)
monkeypatch.setattr("hermes_cli.providers.HERMES_OVERLAYS", {})
custom_providers = [
{
"name": "my-alibaba",
# Matches PROVIDER_REGISTRY['alibaba'].inference_base_url exactly.
"base_url": "https://dashscope-intl.aliyuncs.com/compatible-mode/v1",
"api_key": "sk-sp-test",
"model": "qwen3.6-plus",
"models": {"qwen3.6-plus": {"context_length": 500000}},
}
]
providers = list_authenticated_providers(
current_provider="my-alibaba",
user_providers={},
custom_providers=custom_providers,
max_models=50,
)
slugs = [p["slug"] for p in providers]
# Built-in alibaba row should be present.
assert "alibaba" in slugs, (
f"Expected built-in alibaba row, got slugs: {slugs}"
)
# Custom shadow row should be hidden — its base_url matches the built-in's.
assert not any("my-alibaba" in s for s in slugs), (
f"Custom my-alibaba should have been dedup'd against the built-in "
f"alibaba endpoint, got slugs: {slugs}"
)
def test_list_authenticated_providers_keeps_custom_with_distinct_endpoint(monkeypatch):
"""Dedup must only apply when the endpoint matches a built-in. A custom
provider on a genuinely distinct endpoint stays visible even if a
built-in is also authenticated."""
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test")
monkeypatch.setattr(
"agent.models_dev.fetch_models_dev",
lambda: {
"alibaba": {
"name": "Alibaba Cloud (DashScope)",
"env": ["DASHSCOPE_API_KEY"],
}
},
)
monkeypatch.setattr("hermes_cli.providers.HERMES_OVERLAYS", {})
custom_providers = [
{
"name": "my-private-relay",
"base_url": "https://relay.example.internal/v1",
"api_key": "sk-relay-test",
"model": "qwen3.6-plus",
"models": {"qwen3.6-plus": {}},
}
]
providers = list_authenticated_providers(
current_provider="my-private-relay",
user_providers={},
custom_providers=custom_providers,
max_models=50,
)
slugs = [p["slug"] for p in providers]
assert any("my-private-relay" in s for s in slugs), (
f"Custom provider on distinct endpoint must stay visible, got: {slugs}"
)
def test_list_authenticated_providers_dedup_honors_base_url_env_override(monkeypatch):
"""The dedup must track the EFFECTIVE endpoint — if DASHSCOPE_BASE_URL
overrides the static inference_base_url, a custom provider pointing at
the overridden URL (not the static one) should still be recognized as
a duplicate."""
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test")
monkeypatch.setenv(
"DASHSCOPE_BASE_URL",
"https://custom-dashscope.example.com/v1",
)
monkeypatch.setattr(
"agent.models_dev.fetch_models_dev",
lambda: {
"alibaba": {
"name": "Alibaba Cloud (DashScope)",
"env": ["DASHSCOPE_API_KEY"],
}
},
)
monkeypatch.setattr("hermes_cli.providers.HERMES_OVERLAYS", {})
custom_providers = [
{
"name": "my-dashscope-override",
# Same URL as DASHSCOPE_BASE_URL env override above.
"base_url": "https://custom-dashscope.example.com/v1",
"api_key": "sk-test",
"model": "qwen3.6-plus",
}
]
providers = list_authenticated_providers(
current_provider="alibaba",
user_providers={},
custom_providers=custom_providers,
max_models=50,
)
slugs = [p["slug"] for p in providers]
assert not any("my-dashscope-override" in s for s in slugs), (
f"Custom entry matching env-overridden built-in endpoint should be "
f"dedup'd, got: {slugs}"
)
# =============================================================================
# Tests for _get_named_custom_provider with providers: dict
# =============================================================================
+48 -11
View File
@@ -29,7 +29,7 @@ class TestReloadEnv:
"""reload_env() adds vars from .env that are not in os.environ."""
env_file = tmp_path / ".env"
env_file.write_text("TEST_RELOAD_VAR=hello123\n")
with patch("hermes_cli.config.get_env_path", return_value=env_file):
with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
os.environ.pop("TEST_RELOAD_VAR", None)
count = reload_env()
assert count >= 1
@@ -40,7 +40,7 @@ class TestReloadEnv:
"""reload_env() updates vars whose value changed on disk."""
env_file = tmp_path / ".env"
env_file.write_text("TEST_RELOAD_VAR=old_value\n")
with patch("hermes_cli.config.get_env_path", return_value=env_file):
with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
os.environ["TEST_RELOAD_VAR"] = "old_value"
# Now change the file
env_file.write_text("TEST_RELOAD_VAR=new_value\n")
@@ -55,7 +55,7 @@ class TestReloadEnv:
env_file.write_text("") # empty .env
# Pick a known key from OPTIONAL_ENV_VARS
known_key = next(iter(OPTIONAL_ENV_VARS.keys()))
with patch("hermes_cli.config.get_env_path", return_value=env_file):
with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
os.environ[known_key] = "stale_value"
count = reload_env()
assert known_key not in os.environ
@@ -65,7 +65,7 @@ class TestReloadEnv:
"""reload_env() preserves non-Hermes env vars even when absent from .env."""
env_file = tmp_path / ".env"
env_file.write_text("")
with patch("hermes_cli.config.get_env_path", return_value=env_file):
with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
os.environ["MY_CUSTOM_UNRELATED_VAR"] = "keep_me"
reload_env()
assert os.environ.get("MY_CUSTOM_UNRELATED_VAR") == "keep_me"
@@ -371,6 +371,12 @@ class TestBuildSchemaFromConfig:
assert entry["type"] == "select"
assert "options" in entry
assert "local" in entry["options"]
assert "vercel_sandbox" in entry["options"]
runtime_entry = CONFIG_SCHEMA["terminal.vercel_runtime"]
assert runtime_entry["type"] == "select"
assert "node24" in runtime_entry["options"]
assert "python3.13" in runtime_entry["options"]
assert len(runtime_entry["options"]) >= 3
def test_empty_prefix_produces_correct_keys(self):
from hermes_cli.web_server import _build_schema_from_config
@@ -671,8 +677,12 @@ class TestNewEndpoints:
assert resp.status_code == 200
assert resp.json()["command"] == "hermes setup"
def test_profiles_create_creates_wrapper_alias_when_safe(self):
from pathlib import Path
def test_profiles_create_creates_wrapper_alias_when_safe(self, monkeypatch, tmp_path):
import hermes_cli.profiles as profiles_mod
wrapper_dir = tmp_path / "bin"
wrapper_dir.mkdir()
monkeypatch.setattr(profiles_mod, "_get_wrapper_dir", lambda: wrapper_dir)
resp = self.client.post(
"/api/profiles",
@@ -680,7 +690,7 @@ class TestNewEndpoints:
)
assert resp.status_code == 200
wrapper_path = Path.home() / ".local" / "bin" / "writer"
wrapper_path = wrapper_dir / "writer"
assert wrapper_path.exists()
assert wrapper_path.read_text() == '#!/bin/sh\nexec hermes -p writer "$@"\n'
@@ -2057,14 +2067,24 @@ class TestPtyWebSocket:
assert b"round-trip-payload" in buf
def test_resize_escape_is_forwarded(self, monkeypatch):
# Resize escape gets intercepted and applied via TIOCSWINSZ,
# then ``tput cols/lines`` reports the new dimensions back.
# Resize escape gets intercepted and applied via TIOCSWINSZ, then the
# child reads the TTY ioctl directly. Avoid tput because CI may not set
# TERM for non-interactive shells.
import sys
winsize_script = (
"import fcntl, struct, termios, time; "
"time.sleep(0.15); "
"rows, cols, *_ = struct.unpack('HHHH', "
"fcntl.ioctl(0, termios.TIOCGWINSZ, b'\\0' * 8)); "
"print(cols); print(rows)"
)
monkeypatch.setattr(
self.ws_module,
"_resolve_chat_argv",
# sleep gives the test time to push the resize before tput runs
# sleep gives the test time to push the resize before the child reads the ioctl.
lambda resume=None, sidecar_url=None: (
["/bin/sh", "-c", "sleep 0.15; tput cols; tput lines"],
[sys.executable, "-c", winsize_script],
None,
None,
),
@@ -2153,13 +2173,30 @@ class TestPtyWebSocket:
def test_pub_broadcasts_to_events_subscribers(self, monkeypatch):
"""Frame written to /api/pub is rebroadcast verbatim to every
/api/events subscriber on the same channel."""
import time
from urllib.parse import urlencode
from hermes_cli import web_server as ws_mod
qs = urlencode({"token": self.token, "channel": "broadcast-test"})
pub_path = f"/api/pub?{qs}"
sub_path = f"/api/events?{qs}"
with self.client.websocket_connect(sub_path) as sub:
# Wait for the subscriber to be registered on the server side.
# websocket_connect returns when ws.accept() completes, but the
# server adds us to ``_event_channels`` in a follow-up await,
# so a publish immediately after connect can race ahead of the
# subscriber registration and the message is dropped.
deadline = time.monotonic() + 5.0
while time.monotonic() < deadline:
if ws_mod._event_channels.get("broadcast-test"):
break
time.sleep(0.01)
else:
raise AssertionError(
"subscriber did not register on channel within 5s"
)
with self.client.websocket_connect(pub_path) as pub:
pub.send_text('{"type":"tool.start","payload":{"tool_id":"t1"}}')
received = sub.receive_text()