chore: uptick

This commit is contained in:
Brooklyn Nicholson
2026-05-02 03:19:39 -05:00
parent 420f68e4e2
commit db884f4646
240 changed files with 25206 additions and 3155 deletions
+35
View File
@@ -5,8 +5,10 @@ from __future__ import annotations
import base64
import json
from datetime import datetime, timezone
from unittest.mock import patch
import pytest
import yaml
def _write_auth_store(tmp_path, payload: dict) -> None:
@@ -589,6 +591,39 @@ def test_logout_clears_stale_active_codex_without_provider_credentials(tmp_path,
assert "provider: auto" in config_text
def test_reset_config_provider_uses_atomic_yaml_write(tmp_path, monkeypatch):
"""Logout config reset should delegate the YAML write atomically."""
hermes_home = tmp_path / "hermes"
hermes_home.mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
config_path = hermes_home / "config.yaml"
original = {
"model": {
"default": "gpt-5.3-codex",
"provider": "openai-codex",
"base_url": "https://chatgpt.com/backend-api/codex",
}
}
config_path.write_text(yaml.safe_dump(original, sort_keys=False), encoding="utf-8")
original_text = config_path.read_text(encoding="utf-8")
from hermes_cli.auth import _reset_config_provider
def _boom(path, data, **kwargs):
assert path == config_path
assert data["model"]["provider"] == "auto"
assert data["model"]["base_url"] == "https://openrouter.ai/api/v1"
assert kwargs["sort_keys"] is False
raise OSError("simulated atomic write failure")
with patch("hermes_cli.auth.atomic_yaml_write", side_effect=_boom) as mock_write:
with pytest.raises(OSError, match="simulated atomic write failure"):
_reset_config_provider()
assert mock_write.call_count == 1
assert config_path.read_text(encoding="utf-8") == original_text
def test_auth_list_does_not_call_mutating_select(monkeypatch, capsys):
from hermes_cli.auth_commands import auth_list_command
@@ -76,6 +76,20 @@ class TestResolveVerifyFallback:
)
assert result is False
def test_string_false_in_auth_state_does_not_disable_tls_verify(self):
import ssl
from hermes_cli.auth import _resolve_verify
result = _resolve_verify(auth_state={"tls": {"insecure": "false"}})
assert result is not False
assert result is True or isinstance(result, ssl.SSLContext)
def test_string_true_in_auth_state_disables_tls_verify(self):
from hermes_cli.auth import _resolve_verify
result = _resolve_verify(auth_state={"tls": {"insecure": "true"}})
assert result is False
def test_no_ca_bundle_returns_true(self, monkeypatch):
from hermes_cli.auth import _resolve_verify
+32 -2
View File
@@ -13,6 +13,7 @@ from hermes_cli.commands import (
SlashCommandAutoSuggest,
SlashCommandCompleter,
_CMD_NAME_LIMIT,
_SLACK_RESERVED_COMMANDS,
_TG_NAME_LIMIT,
_clamp_command_names,
_clamp_telegram_names,
@@ -299,9 +300,19 @@ class TestSlackNativeSlashes:
def test_includes_canonical_commands(self):
names = {n for n, _d, _h in slack_native_slashes()}
# Sample of gateway-available canonical commands
for expected in ("new", "stop", "background", "model", "help", "status"):
for expected in ("new", "stop", "background", "model", "help"):
assert expected in names, f"missing canonical /{expected}"
def test_excludes_slack_reserved_commands(self):
"""Slack built-in commands (e.g. /status, /me, /join) cannot be
registered by apps and must be excluded from the manifest.
Users can still reach them via /hermes <command>."""
names = {n for n, _d, _h in slack_native_slashes()}
for reserved in _SLACK_RESERVED_COMMANDS:
assert reserved not in names, (
f"/{reserved} is a Slack built-in and must not appear in the manifest"
)
def test_includes_aliases_as_first_class_slashes(self):
"""Aliases (/btw, /bg, /reset, /q) must be registered as standalone
slashes — this is the whole point of native-slashes parity."""
@@ -319,6 +330,9 @@ class TestSlackNativeSlashes:
Telegram but not Slack (because of Slack's 50-slash cap), this
test fails loudly so we can curate the list rather than silently
dropping parity.
Slack-reserved built-in commands (e.g. /status) are excluded
from parity checks since they cannot be registered on Slack.
"""
slack_names = {n for n, _d, _h in slack_native_slashes()}
tg_names = {n for n, _d in telegram_bot_commands()}
@@ -329,7 +343,8 @@ class TestSlackNativeSlashes:
slack_norm = {_norm(n) for n in slack_names}
tg_norm = {_norm(n) for n in tg_names}
missing = tg_norm - slack_norm
reserved_norm = {_norm(n) for n in _SLACK_RESERVED_COMMANDS}
missing = (tg_norm - slack_norm) - reserved_norm
assert not missing, (
f"commands on Telegram but missing from Slack native slashes: {sorted(missing)}"
)
@@ -405,6 +420,21 @@ class TestGatewayConfigGate:
joined = "\n".join(lines)
assert "`/verbose" in joined
def test_config_gate_quoted_false_stays_disabled_everywhere(self, tmp_path, monkeypatch):
"""Quoted false must not enable config-gated gateway commands."""
config_file = tmp_path / "config.yaml"
config_file.write_text('display:\n tool_progress_command: "false"\n')
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
lines = gateway_help_lines()
joined = "\n".join(lines)
names = {name for name, _ in telegram_bot_commands()}
mapping = slack_subcommand_map()
assert "`/verbose" not in joined
assert "verbose" not in names
assert "verbose" not in mapping
def test_config_gate_excluded_from_telegram_when_off(self, tmp_path, monkeypatch):
config_file = tmp_path / "config.yaml"
config_file.write_text("display:\n tool_progress_command: false\n")
+358
View File
@@ -0,0 +1,358 @@
"""Tests for hermes_cli/goals.py — persistent cross-turn goals."""
from __future__ import annotations
import json
from unittest.mock import patch, MagicMock
import pytest
# ──────────────────────────────────────────────────────────────────────
# Fixtures
# ──────────────────────────────────────────────────────────────────────
@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
"""Isolated HERMES_HOME so SessionDB.state_meta writes don't clobber the real one."""
from pathlib import Path
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setenv("HERMES_HOME", str(home))
# Bust the goal-module's DB cache for each test so it re-resolves HERMES_HOME.
from hermes_cli import goals
goals._DB_CACHE.clear()
yield home
goals._DB_CACHE.clear()
# ──────────────────────────────────────────────────────────────────────
# _parse_judge_response
# ──────────────────────────────────────────────────────────────────────
class TestParseJudgeResponse:
def test_clean_json_done(self):
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response('{"done": true, "reason": "all good"}')
assert done is True
assert reason == "all good"
def test_clean_json_continue(self):
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response('{"done": false, "reason": "more work needed"}')
assert done is False
assert reason == "more work needed"
def test_json_in_markdown_fence(self):
from hermes_cli.goals import _parse_judge_response
raw = '```json\n{"done": true, "reason": "done"}\n```'
done, reason = _parse_judge_response(raw)
assert done is True
assert "done" in reason
def test_json_embedded_in_prose(self):
"""Some models prefix reasoning before emitting JSON — we extract it."""
from hermes_cli.goals import _parse_judge_response
raw = 'Looking at this... the agent says X. Verdict: {"done": false, "reason": "partial"}'
done, reason = _parse_judge_response(raw)
assert done is False
assert reason == "partial"
def test_string_done_values(self):
from hermes_cli.goals import _parse_judge_response
for s in ("true", "yes", "done", "1"):
done, _ = _parse_judge_response(f'{{"done": "{s}", "reason": "r"}}')
assert done is True
for s in ("false", "no", "not yet"):
done, _ = _parse_judge_response(f'{{"done": "{s}", "reason": "r"}}')
assert done is False
def test_malformed_json_fails_open(self):
"""Non-JSON → not done, with error-ish reason (so judge_goal can map to continue)."""
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response("this is not json at all")
assert done is False
assert reason # non-empty
def test_empty_response(self):
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response("")
assert done is False
assert reason
# ──────────────────────────────────────────────────────────────────────
# judge_goal — fail-open semantics
# ──────────────────────────────────────────────────────────────────────
class TestJudgeGoal:
def test_empty_goal_skipped(self):
from hermes_cli.goals import judge_goal
verdict, _ = judge_goal("", "some response")
assert verdict == "skipped"
def test_empty_response_continues(self):
from hermes_cli.goals import judge_goal
verdict, _ = judge_goal("ship the thing", "")
assert verdict == "continue"
def test_no_aux_client_continues(self):
"""Fail-open: if no aux client, we must return continue, not skipped/done."""
from hermes_cli import goals
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(None, None),
):
verdict, _ = goals.judge_goal("my goal", "my response")
assert verdict == "continue"
def test_api_error_continues(self):
"""Judge exception → fail-open continue (don't wedge progress on judge bugs)."""
from hermes_cli import goals
fake_client = MagicMock()
fake_client.chat.completions.create.side_effect = RuntimeError("boom")
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, reason = goals.judge_goal("goal", "response")
assert verdict == "continue"
assert "judge error" in reason.lower()
def test_judge_says_done(self):
from hermes_cli import goals
fake_client = MagicMock()
fake_client.chat.completions.create.return_value = MagicMock(
choices=[
MagicMock(
message=MagicMock(content='{"done": true, "reason": "achieved"}')
)
]
)
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, reason = goals.judge_goal("goal", "agent response")
assert verdict == "done"
assert reason == "achieved"
def test_judge_says_continue(self):
from hermes_cli import goals
fake_client = MagicMock()
fake_client.chat.completions.create.return_value = MagicMock(
choices=[
MagicMock(
message=MagicMock(content='{"done": false, "reason": "not yet"}')
)
]
)
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, reason = goals.judge_goal("goal", "agent response")
assert verdict == "continue"
assert reason == "not yet"
# ──────────────────────────────────────────────────────────────────────
# GoalManager lifecycle + persistence
# ──────────────────────────────────────────────────────────────────────
class TestGoalManager:
def test_no_goal_initial(self, hermes_home):
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="test-sid-1")
assert mgr.state is None
assert not mgr.is_active()
assert not mgr.has_goal()
assert "No active goal" in mgr.status_line()
def test_set_then_status(self, hermes_home):
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="test-sid-2", default_max_turns=5)
state = mgr.set("port the thing")
assert state.goal == "port the thing"
assert state.status == "active"
assert state.max_turns == 5
assert state.turns_used == 0
assert mgr.is_active()
assert "active" in mgr.status_line().lower()
assert "port the thing" in mgr.status_line()
def test_set_rejects_empty(self, hermes_home):
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="test-sid-3")
with pytest.raises(ValueError):
mgr.set("")
with pytest.raises(ValueError):
mgr.set(" ")
def test_pause_and_resume(self, hermes_home):
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="test-sid-4")
mgr.set("goal text")
mgr.pause(reason="user-paused")
assert mgr.state.status == "paused"
assert not mgr.is_active()
assert mgr.has_goal()
mgr.resume()
assert mgr.state.status == "active"
assert mgr.is_active()
def test_clear(self, hermes_home):
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="test-sid-5")
mgr.set("goal")
mgr.clear()
assert mgr.state is None
assert not mgr.is_active()
def test_persistence_across_managers(self, hermes_home):
"""Key invariant: a second manager on the same session sees the goal.
This is what makes /resume work — each session rebinds its
GoalManager and picks up the saved state.
"""
from hermes_cli.goals import GoalManager
mgr1 = GoalManager(session_id="persist-sid")
mgr1.set("do the thing")
mgr2 = GoalManager(session_id="persist-sid")
assert mgr2.state is not None
assert mgr2.state.goal == "do the thing"
assert mgr2.is_active()
def test_evaluate_after_turn_done(self, hermes_home):
"""Judge says done → status=done, no continuation."""
from hermes_cli import goals
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="eval-sid-1")
mgr.set("ship it")
with patch.object(goals, "judge_goal", return_value=("done", "shipped")):
decision = mgr.evaluate_after_turn("I shipped the feature.")
assert decision["verdict"] == "done"
assert decision["should_continue"] is False
assert decision["continuation_prompt"] is None
assert mgr.state.status == "done"
assert mgr.state.turns_used == 1
def test_evaluate_after_turn_continue_under_budget(self, hermes_home):
from hermes_cli import goals
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="eval-sid-2", default_max_turns=5)
mgr.set("a long goal")
with patch.object(goals, "judge_goal", return_value=("continue", "more work")):
decision = mgr.evaluate_after_turn("made some progress")
assert decision["verdict"] == "continue"
assert decision["should_continue"] is True
assert decision["continuation_prompt"] is not None
assert "a long goal" in decision["continuation_prompt"]
assert mgr.state.status == "active"
assert mgr.state.turns_used == 1
def test_evaluate_after_turn_budget_exhausted(self, hermes_home):
"""When turn budget hits ceiling, auto-pause instead of continuing."""
from hermes_cli import goals
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="eval-sid-3", default_max_turns=2)
mgr.set("hard goal")
with patch.object(goals, "judge_goal", return_value=("continue", "not yet")):
d1 = mgr.evaluate_after_turn("step 1")
assert d1["should_continue"] is True
assert mgr.state.turns_used == 1
assert mgr.state.status == "active"
d2 = mgr.evaluate_after_turn("step 2")
# turns_used is now 2 which equals max_turns → paused
assert d2["should_continue"] is False
assert mgr.state.status == "paused"
assert mgr.state.turns_used == 2
assert "budget" in (mgr.state.paused_reason or "").lower()
def test_evaluate_after_turn_inactive(self, hermes_home):
"""evaluate_after_turn is a no-op when goal isn't active."""
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="eval-sid-4")
d = mgr.evaluate_after_turn("anything")
assert d["verdict"] == "inactive"
assert d["should_continue"] is False
mgr.set("a goal")
mgr.pause()
d2 = mgr.evaluate_after_turn("anything")
assert d2["verdict"] == "inactive"
assert d2["should_continue"] is False
def test_continuation_prompt_shape(self, hermes_home):
"""The continuation prompt must include the goal text verbatim —
and must be safe to inject as a user-role message (prompt-cache
invariants: no system-prompt mutation)."""
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="cont-sid")
mgr.set("port goal command to hermes")
prompt = mgr.next_continuation_prompt()
assert prompt is not None
assert "port goal command to hermes" in prompt
assert prompt.strip() # non-empty
# ──────────────────────────────────────────────────────────────────────
# Smoke: CommandDef is wired
# ──────────────────────────────────────────────────────────────────────
def test_goal_command_in_registry():
from hermes_cli.commands import resolve_command
cmd = resolve_command("goal")
assert cmd is not None
assert cmd.name == "goal"
def test_goal_command_dispatches_in_cli_registry_helpers():
"""goal shows up in autocomplete / help categories alongside other Session cmds."""
from hermes_cli.commands import COMMANDS, COMMANDS_BY_CATEGORY
assert "/goal" in COMMANDS
session_cmds = COMMANDS_BY_CATEGORY.get("Session", {})
assert "/goal" in session_cmds
@@ -71,6 +71,32 @@ class TestSaveModelChoiceAlwaysDict:
class TestProviderPersistsAfterModelSave:
def test_update_config_for_provider_uses_atomic_yaml_write(self, config_home):
"""Provider switches should delegate config writes to atomic_yaml_write."""
from hermes_cli.auth import _update_config_for_provider
config_path = config_home / "config.yaml"
original_text = config_path.read_text(encoding="utf-8")
def _boom(path, data, **kwargs):
assert path == config_path
assert data["model"]["provider"] == "nous"
assert data["model"]["base_url"] == "https://inference.example.com/v1"
assert data["model"]["default"] == "some-old-model"
assert kwargs["sort_keys"] is False
raise OSError("simulated atomic write failure")
with patch("hermes_cli.auth.atomic_yaml_write", side_effect=_boom) as mock_write:
with pytest.raises(OSError, match="simulated atomic write failure"):
_update_config_for_provider(
"nous",
"https://inference.example.com/v1/",
default_model="llama-3.3",
)
assert mock_write.call_count == 1
assert config_path.read_text(encoding="utf-8") == original_text
def test_api_key_provider_saved_when_model_was_string(self, config_home, monkeypatch):
"""_model_flow_api_key_provider must persist the provider even when
config.model started as a plain string."""
+40
View File
@@ -21,6 +21,7 @@ from hermes_cli.plugins import (
get_plugin_command_handler,
get_plugin_commands,
get_pre_tool_call_block_message,
resolve_plugin_command_result,
discover_plugins,
invoke_hook,
)
@@ -1061,6 +1062,45 @@ class TestPluginCommands:
assert mgr._plugin_commands["cmd-b"]["plugin"] == "plugin-b"
class TestPluginCommandResultResolution:
def test_returns_sync_values_unchanged(self):
assert resolve_plugin_command_result("ok") == "ok"
def test_awaits_async_result_without_running_loop(self):
async def _handler():
return "async-ok"
assert resolve_plugin_command_result(_handler()) == "async-ok"
def test_awaits_async_result_with_running_loop(self, monkeypatch):
class _Loop:
pass
async def _handler():
return "threaded-ok"
monkeypatch.setattr("hermes_cli.plugins.asyncio.get_running_loop", lambda: _Loop())
assert resolve_plugin_command_result(_handler()) == "threaded-ok"
def test_running_loop_timeout_does_not_hang_forever(self, monkeypatch):
"""Threaded path must abort a hung async handler instead of blocking the caller."""
import asyncio as _asyncio
class _Loop:
pass
async def _slow_handler():
await _asyncio.sleep(10)
return "should-not-reach"
monkeypatch.setattr("hermes_cli.plugins.asyncio.get_running_loop", lambda: _Loop())
monkeypatch.setattr("hermes_cli.plugins._PLUGIN_COMMAND_AWAIT_TIMEOUT_SECS", 0.1)
import pytest
with pytest.raises(TimeoutError):
resolve_plugin_command_result(_slow_handler())
# ── TestPluginDispatchTool ────────────────────────────────────────────────
@@ -897,6 +897,58 @@ def test_named_custom_provider_does_not_shadow_builtin_provider(monkeypatch):
assert resolved["requested_provider"] == "nous"
def test_named_custom_provider_wins_over_builtin_alias(monkeypatch):
"""A custom_providers entry named after a built-in *alias* (not a canonical
provider name) must win over the built-in. Regression guard for #15743:
when users define ``custom_providers: [{name: kimi, ...}]`` and reference
``provider: kimi``, the built-in alias rewriting (``kimi`` → ``kimi-coding``)
would otherwise hijack the request and send it to the wrong endpoint.
"""
monkeypatch.setattr(
rp,
"load_config",
lambda: {
"custom_providers": [
{
"name": "kimi",
"base_url": "https://my-custom-kimi.example.com/v1",
"api_key": "my-kimi-key",
}
]
},
)
entry = rp._get_named_custom_provider("kimi")
assert entry is not None
assert entry["base_url"] == "https://my-custom-kimi.example.com/v1"
assert entry["api_key"] == "my-kimi-key"
def test_named_custom_provider_skipped_for_canonical_built_in(monkeypatch):
"""Companion to the test above: ``nous`` is a canonical provider name
(``resolve_provider('nous') == 'nous'``), so a custom entry with that name
should NOT be returned — the built-in wins as before.
"""
monkeypatch.setattr(
rp,
"load_config",
lambda: {
"custom_providers": [
{
"name": "nous",
"base_url": "http://localhost:1234/v1",
"api_key": "shadow-key",
}
]
},
)
entry = rp._get_named_custom_provider("nous")
assert entry is None
def test_explicit_openrouter_skips_openai_base_url(monkeypatch):
"""When the user explicitly requests openrouter, OPENAI_BASE_URL
(which may point to a custom endpoint) must not override the
+10 -1
View File
@@ -120,7 +120,16 @@ def test_get_platform_tools_preserves_explicit_empty_selection():
enabled = _get_platform_tools(config, "cli")
assert enabled == set()
# An explicit empty list disables every CONFIGURABLE toolset (web,
# terminal, memory, …). Non-configurable platform toolsets that ride
# along on the platform's default composite (e.g. `kanban`, whose tools
# live in _HERMES_CORE_TOOLS but aren't user-toggleable) are still
# auto-recovered by _get_platform_tools so saving via `hermes tools`
# doesn't silently drop them. The contract this test guards is the
# configurable side: nothing the user could have checked in the TUI
# checklist should reappear here.
configurable = {ts_key for ts_key, _, _ in CONFIGURABLE_TOOLSETS}
assert enabled.isdisjoint(configurable)
def test_apply_toolset_change_from_default_does_not_enable_default_off_toolsets():
@@ -392,6 +392,81 @@ class TestCmdUpdateLaunchdRestart:
captured = capsys.readouterr().out
assert "Restart manually: hermes gateway run" in captured
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_update_restarts_profile_manual_gateways(
self, mock_run, _mock_which, mock_args, capsys, tmp_path, monkeypatch,
):
"""Profile-mapped manual gateways are relaunched automatically after update."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
monkeypatch.setattr(
gateway_cli,
"get_launchd_plist_path",
lambda: tmp_path / "ai.hermes.gateway.plist",
)
mock_run.side_effect = _make_run_side_effect(
commit_count="3",
launchctl_loaded=False,
)
process = gateway_cli.ProfileGatewayProcess(
profile="coder",
path=tmp_path / ".hermes" / "profiles" / "coder",
pid=12345,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[12345]), \
patch.object(gateway_cli, "find_profile_gateway_processes", return_value=[process]), \
patch.object(gateway_cli, "launch_detached_profile_gateway_restart", return_value=True) as restart, \
patch.object(gateway_cli, "_graceful_restart_via_sigusr1", return_value=True) as graceful, \
patch("os.kill") as kill:
cmd_update(mock_args)
captured = capsys.readouterr().out
restart.assert_called_once_with("coder", 12345)
graceful.assert_called_once()
# Graceful drain succeeded — no SIGTERM fallback needed.
kill.assert_not_called()
assert "Restarting manual gateway profile(s): coder" in captured
assert "Restart manually: hermes gateway run" not in captured
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_update_profile_manual_gateway_falls_back_to_sigterm(
self, mock_run, _mock_which, mock_args, capsys, tmp_path, monkeypatch,
):
"""When graceful SIGUSR1 drain fails, manual profile restart falls back to SIGTERM."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: True)
monkeypatch.setattr(
gateway_cli,
"get_launchd_plist_path",
lambda: tmp_path / "ai.hermes.gateway.plist",
)
mock_run.side_effect = _make_run_side_effect(
commit_count="3",
launchctl_loaded=False,
)
process = gateway_cli.ProfileGatewayProcess(
profile="coder",
path=tmp_path / ".hermes" / "profiles" / "coder",
pid=12345,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[12345]), \
patch.object(gateway_cli, "find_profile_gateway_processes", return_value=[process]), \
patch.object(gateway_cli, "launch_detached_profile_gateway_restart", return_value=True) as restart, \
patch.object(gateway_cli, "_graceful_restart_via_sigusr1", return_value=False) as graceful, \
patch("os.kill") as kill:
cmd_update(mock_args)
captured = capsys.readouterr().out
restart.assert_called_once_with("coder", 12345)
graceful.assert_called_once()
# Graceful drain returned False → SIGTERM fallback.
kill.assert_called_once()
assert "Restarting manual gateway profile(s): coder" in captured
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_update_with_systemd_still_restarts_via_systemd(
+167
View File
@@ -0,0 +1,167 @@
"""Tests for `hermes update --yes / -y` — assume yes for interactive prompts.
Covers:
1. argparse parses the flag
2. Config-migration prompt is auto-answered (no input() call) and migrate_config
runs with interactive=False so API-key prompts are skipped
3. Autostash restore prompt is auto-answered (prompt_for_restore == False, no
input() call) and the stash is applied automatically
"""
import subprocess
from types import SimpleNamespace
from unittest.mock import patch
from hermes_cli.main import cmd_update
def _make_run_side_effect(
branch="main", verify_ok=True, commit_count="1", dirty=False
):
"""Minimal subprocess.run side_effect for the update flow."""
def side_effect(cmd, **kwargs):
joined = " ".join(str(c) for c in cmd)
if "rev-parse" in joined and "--abbrev-ref" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout=f"{branch}\n", stderr="")
if "rev-parse" in joined and "--verify" in joined:
return subprocess.CompletedProcess(
cmd, 0 if verify_ok else 128, stdout="", stderr=""
)
if "rev-list" in joined:
return subprocess.CompletedProcess(
cmd, 0, stdout=f"{commit_count}\n", stderr=""
)
# `git status --porcelain` for dirty-tree detection during autostash.
if "status" in joined and "--porcelain" in joined:
out = " M hermes_cli/main.py\n" if dirty else ""
return subprocess.CompletedProcess(cmd, 0, stdout=out, stderr="")
# `git stash list` — return a stash ref when dirty (so _stash_local_changes
# gets something to return). _stash_local_changes_if_needed is what we
# actually patch in tests that exercise restore, so this is a catch-all.
if "stash" in joined and "list" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
return side_effect
class TestUpdateYesConfigMigration:
"""--yes auto-answers the config-migration prompt and skips API-key prompts."""
@patch("hermes_cli.config.migrate_config")
@patch("hermes_cli.config.check_config_version", return_value=(1, 2))
@patch("hermes_cli.config.get_missing_config_fields", return_value=[])
@patch("hermes_cli.config.get_missing_env_vars", return_value=["NEW_KEY"])
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_yes_auto_migrates_without_input(
self,
mock_run,
_mock_which,
_mock_missing_env,
_mock_missing_cfg,
_mock_version,
mock_migrate,
capsys,
):
mock_run.side_effect = _make_run_side_effect(
branch="main", verify_ok=True, commit_count="1"
)
mock_migrate.return_value = {"env_added": [], "config_added": []}
args = SimpleNamespace(yes=True)
with patch("builtins.input") as mock_input:
cmd_update(args)
# Never prompted the user.
mock_input.assert_not_called()
# migrate_config was invoked with interactive=False — API-key prompts
# are suppressed, matching gateway-mode semantics.
assert mock_migrate.call_count == 1
_, kwargs = mock_migrate.call_args
assert kwargs.get("interactive") is False
out = capsys.readouterr().out
assert "--yes: auto-applying config migration" in out
# The "Would you like to configure them now?" prompt text never appears.
assert "Would you like to configure them now?" not in out
@patch("hermes_cli.config.migrate_config")
@patch("hermes_cli.config.check_config_version", return_value=(1, 2))
@patch("hermes_cli.config.get_missing_config_fields", return_value=[])
@patch("hermes_cli.config.get_missing_env_vars", return_value=["NEW_KEY"])
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_no_yes_flag_still_prompts_in_tty(
self,
mock_run,
_mock_which,
_mock_missing_env,
_mock_missing_cfg,
_mock_version,
mock_migrate,
capsys,
):
"""Regression guard: without --yes, the TTY prompt path still fires."""
mock_run.side_effect = _make_run_side_effect(
branch="main", verify_ok=True, commit_count="1"
)
mock_migrate.return_value = {"env_added": [], "config_added": []}
args = SimpleNamespace(yes=False)
with patch("builtins.input", return_value="n") as mock_input, patch(
"hermes_cli.main.sys"
) as mock_sys:
mock_sys.stdin.isatty.return_value = True
mock_sys.stdout.isatty.return_value = True
cmd_update(args)
# The user was actually prompted.
assert mock_input.called
prompts = [c.args[0] if c.args else "" for c in mock_input.call_args_list]
assert any("configure them now" in p for p in prompts)
class TestUpdateYesStashRestore:
"""--yes auto-restores the pre-update autostash without prompting."""
@patch("hermes_cli.main._restore_stashed_changes")
@patch(
"hermes_cli.main._stash_local_changes_if_needed",
return_value="stash@{0}",
)
@patch("hermes_cli.config.check_config_version", return_value=(1, 1))
@patch("hermes_cli.config.get_missing_config_fields", return_value=[])
@patch("hermes_cli.config.get_missing_env_vars", return_value=[])
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_yes_restores_stash_without_prompting(
self,
mock_run,
_mock_which,
_mock_missing_env,
_mock_missing_cfg,
_mock_version,
_mock_stash,
mock_restore,
capsys,
):
# Not on main → cmd_update switches to main → autostash fires.
mock_run.side_effect = _make_run_side_effect(
branch="feature-branch", verify_ok=True, commit_count="1", dirty=True
)
args = SimpleNamespace(yes=True)
cmd_update(args)
# _restore_stashed_changes was called, and called with prompt_user=False
# every time (so the user never sees "Restore local changes now?").
assert mock_restore.called
for call in mock_restore.call_args_list:
assert call.kwargs.get("prompt_user") is False, (
f"Expected prompt_user=False under --yes, got {call.kwargs}"
)
@@ -839,3 +839,148 @@ def test_get_named_custom_provider_transport_resolves_via_display_name(monkeypat
result = rp._get_named_custom_provider("Codex Provider")
assert result is not None
assert result["api_mode"] == "codex_responses"
# =============================================================================
# Regression: user_providers override for private models not listed by /v1/models
# =============================================================================
_REJECTED_VALIDATION = {
"accepted": False,
"persist": False,
"recognized": False,
"message": "not found",
}
def _run_user_provider_override_case(
*,
slug,
name,
base_url,
models,
raw_input,
):
"""Run ``switch_model`` with a private user provider and a rejected API check.
The bug in PR #17964 was that ``user_providers`` was treated like a list,
so private models listed in ``models:`` never triggered the override path.
These tests keep the validation failure in place and prove the config list
still wins for both dict- and list-shaped ``models`` entries.
"""
from unittest.mock import patch
user_providers = {
slug: {
"name": name,
"api": base_url,
"discover_models": False,
"models": models,
}
}
with patch("hermes_cli.model_switch.resolve_alias", return_value=None), \
patch("hermes_cli.model_switch.list_provider_models", return_value=[]), \
patch("hermes_cli.model_switch.normalize_model_for_provider", side_effect=lambda model, provider: model), \
patch("hermes_cli.models.validate_requested_model", return_value=_REJECTED_VALIDATION), \
patch("hermes_cli.models.detect_provider_for_model", return_value=None), \
patch("hermes_cli.model_switch.get_model_info", return_value=None), \
patch("hermes_cli.model_switch.get_model_capabilities", return_value=None), \
patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value={"api_key": "***", "base_url": base_url, "api_mode": "anthropic_messages"}):
return switch_model(
raw_input=raw_input,
current_provider=slug,
current_model="old-model",
current_base_url=base_url,
user_providers=user_providers,
custom_providers=[],
)
@pytest.mark.parametrize(
("slug", "name", "base_url", "models", "raw_input", "expected_model"),
[
(
"kimi-coding",
"Kimi Coding Plan",
"https://api.kimi.com/coding",
{"kimi-k2.6": {}},
"kimi-k2.6",
"kimi-k2.6",
),
(
"kimi-dedicated",
"Kimi Dedicated",
"https://api.kimi.com/v1",
[{"name": "moonshotai/Kimi-K2.6-ACED"}],
"moonshotai/Kimi-K2.6-ACED",
"moonshotai/Kimi-K2.6-ACED",
),
],
ids=["kimi-coding-plan-dict", "kimi-k2-6-aced-list"],
)
def test_user_provider_override_accepts_listed_private_models(
slug,
name,
base_url,
models,
raw_input,
expected_model,
):
"""Private models listed in providers: config should override /v1/models misses.
Covers both config shapes the fix now accepts:
- dict models for the Kimi Coding Plan K2p6 case
- list-of-dicts models for the Kimi-K2.6-ACED dedicated case
"""
result = _run_user_provider_override_case(
slug=slug,
name=name,
base_url=base_url,
models=models,
raw_input=raw_input,
)
assert result.success is True
assert result.new_model == expected_model
assert result.error_message == ""
@pytest.mark.parametrize(
("slug", "name", "base_url", "models", "raw_input"),
[
(
"kimi-coding",
"Kimi Coding Plan",
"https://api.kimi.com/coding",
{"kimi-k2.6": {}},
"kimi-k2.6-mangled",
),
(
"kimi-dedicated",
"Kimi Dedicated",
"https://api.kimi.com/v1",
[{"name": "moonshotai/Kimi-K2.6-ACED"}],
"moonshotai/Kimi-K2.6-ACED!!!",
),
],
ids=["kimi-coding-plan-dict-mangled", "kimi-k2-6-aced-list-mangled"],
)
def test_user_provider_override_rejects_mangled_private_models(
slug,
name,
base_url,
models,
raw_input,
):
"""Malformed model names should fail cleanly, not crash or auto-accept."""
result = _run_user_provider_override_case(
slug=slug,
name=name,
base_url=base_url,
models=models,
raw_input=raw_input,
)
assert result.success is False
assert result.error_message == "not found"