Merge branch 'main' into rewbs/tool-use-charge-to-subscription

This commit is contained in:
Ben Barclay 2026-04-02 11:00:35 +11:00
commit a2e56d044b
175 changed files with 18848 additions and 3772 deletions

View file

@ -40,6 +40,119 @@ class TestFindMigrationScript:
assert claw_mod._find_migration_script() is None
# ---------------------------------------------------------------------------
# _find_openclaw_dirs
# ---------------------------------------------------------------------------
class TestFindOpenclawDirs:
"""Test discovery of OpenClaw directories."""
def test_finds_openclaw_dir(self, tmp_path):
openclaw = tmp_path / ".openclaw"
openclaw.mkdir()
with patch("pathlib.Path.home", return_value=tmp_path):
found = claw_mod._find_openclaw_dirs()
assert openclaw in found
def test_finds_legacy_dirs(self, tmp_path):
clawdbot = tmp_path / ".clawdbot"
clawdbot.mkdir()
moldbot = tmp_path / ".moldbot"
moldbot.mkdir()
with patch("pathlib.Path.home", return_value=tmp_path):
found = claw_mod._find_openclaw_dirs()
assert len(found) == 2
assert clawdbot in found
assert moldbot in found
def test_returns_empty_when_none_exist(self, tmp_path):
with patch("pathlib.Path.home", return_value=tmp_path):
found = claw_mod._find_openclaw_dirs()
assert found == []
# ---------------------------------------------------------------------------
# _scan_workspace_state
# ---------------------------------------------------------------------------
class TestScanWorkspaceState:
"""Test scanning for workspace state files."""
def test_finds_root_state_files(self, tmp_path):
(tmp_path / "todo.json").write_text("{}")
(tmp_path / "sessions").mkdir()
findings = claw_mod._scan_workspace_state(tmp_path)
descs = [desc for _, desc in findings]
assert any("todo.json" in d for d in descs)
assert any("sessions" in d for d in descs)
def test_finds_workspace_state_files(self, tmp_path):
ws = tmp_path / "workspace"
ws.mkdir()
(ws / "todo.json").write_text("{}")
(ws / "sessions").mkdir()
findings = claw_mod._scan_workspace_state(tmp_path)
descs = [desc for _, desc in findings]
assert any("workspace/todo.json" in d for d in descs)
assert any("workspace/sessions" in d for d in descs)
def test_ignores_hidden_dirs(self, tmp_path):
scan_dir = tmp_path / "scan_target"
scan_dir.mkdir()
hidden = scan_dir / ".git"
hidden.mkdir()
(hidden / "todo.json").write_text("{}")
findings = claw_mod._scan_workspace_state(scan_dir)
assert len(findings) == 0
def test_empty_dir_returns_empty(self, tmp_path):
scan_dir = tmp_path / "scan_target"
scan_dir.mkdir()
findings = claw_mod._scan_workspace_state(scan_dir)
assert findings == []
# ---------------------------------------------------------------------------
# _archive_directory
# ---------------------------------------------------------------------------
class TestArchiveDirectory:
"""Test directory archival (rename)."""
def test_renames_to_pre_migration(self, tmp_path):
source = tmp_path / ".openclaw"
source.mkdir()
(source / "test.txt").write_text("data")
archive_path = claw_mod._archive_directory(source)
assert archive_path == tmp_path / ".openclaw.pre-migration"
assert archive_path.is_dir()
assert not source.exists()
assert (archive_path / "test.txt").read_text() == "data"
def test_adds_timestamp_when_archive_exists(self, tmp_path):
source = tmp_path / ".openclaw"
source.mkdir()
# Pre-existing archive
(tmp_path / ".openclaw.pre-migration").mkdir()
archive_path = claw_mod._archive_directory(source)
assert ".pre-migration-" in archive_path.name
assert archive_path.is_dir()
assert not source.exists()
def test_dry_run_does_not_rename(self, tmp_path):
source = tmp_path / ".openclaw"
source.mkdir()
archive_path = claw_mod._archive_directory(source, dry_run=True)
assert archive_path == tmp_path / ".openclaw.pre-migration"
assert source.is_dir() # Still exists
# ---------------------------------------------------------------------------
# claw_command routing
# ---------------------------------------------------------------------------
@ -56,11 +169,24 @@ class TestClawCommand:
claw_mod.claw_command(args)
mock.assert_called_once_with(args)
def test_routes_to_cleanup(self):
args = Namespace(claw_action="cleanup", source=None, dry_run=False, yes=False)
with patch.object(claw_mod, "_cmd_cleanup") as mock:
claw_mod.claw_command(args)
mock.assert_called_once_with(args)
def test_routes_clean_alias(self):
args = Namespace(claw_action="clean", source=None, dry_run=False, yes=False)
with patch.object(claw_mod, "_cmd_cleanup") as mock:
claw_mod.claw_command(args)
mock.assert_called_once_with(args)
def test_shows_help_for_no_action(self, capsys):
args = Namespace(claw_action=None)
claw_mod.claw_command(args)
captured = capsys.readouterr()
assert "migrate" in captured.out
assert "cleanup" in captured.out
# ---------------------------------------------------------------------------
@ -168,6 +294,7 @@ class TestCmdMigrate:
patch.object(claw_mod, "_load_migration_module", return_value=fake_mod),
patch.object(claw_mod, "get_config_path", return_value=config_path),
patch.object(claw_mod, "prompt_yes_no", return_value=True),
patch.object(claw_mod, "_offer_source_archival"),
):
claw_mod._cmd_migrate(args)
@ -175,6 +302,75 @@ class TestCmdMigrate:
assert "Migration Results" in captured.out
assert "Migration complete!" in captured.out
def test_execute_offers_archival_on_success(self, tmp_path, capsys):
"""After successful migration, _offer_source_archival should be called."""
openclaw_dir = tmp_path / ".openclaw"
openclaw_dir.mkdir()
fake_mod = ModuleType("openclaw_to_hermes")
fake_mod.resolve_selected_options = MagicMock(return_value={"soul"})
fake_migrator = MagicMock()
fake_migrator.migrate.return_value = {
"summary": {"migrated": 3, "skipped": 0, "conflict": 0, "error": 0},
"items": [
{"kind": "soul", "status": "migrated", "destination": str(tmp_path / "SOUL.md")},
],
}
fake_mod.Migrator = MagicMock(return_value=fake_migrator)
args = Namespace(
source=str(openclaw_dir),
dry_run=False, preset="full", overwrite=False,
migrate_secrets=False, workspace_target=None,
skill_conflict="skip", yes=True,
)
with (
patch.object(claw_mod, "_find_migration_script", return_value=tmp_path / "s.py"),
patch.object(claw_mod, "_load_migration_module", return_value=fake_mod),
patch.object(claw_mod, "get_config_path", return_value=tmp_path / "config.yaml"),
patch.object(claw_mod, "save_config"),
patch.object(claw_mod, "load_config", return_value={}),
patch.object(claw_mod, "_offer_source_archival") as mock_archival,
):
claw_mod._cmd_migrate(args)
mock_archival.assert_called_once_with(openclaw_dir, True)
def test_dry_run_skips_archival(self, tmp_path, capsys):
"""Dry run should not offer archival."""
openclaw_dir = tmp_path / ".openclaw"
openclaw_dir.mkdir()
fake_mod = ModuleType("openclaw_to_hermes")
fake_mod.resolve_selected_options = MagicMock(return_value=set())
fake_migrator = MagicMock()
fake_migrator.migrate.return_value = {
"summary": {"migrated": 2, "skipped": 0, "conflict": 0, "error": 0},
"items": [],
"preset": "full",
}
fake_mod.Migrator = MagicMock(return_value=fake_migrator)
args = Namespace(
source=str(openclaw_dir),
dry_run=True, preset="full", overwrite=False,
migrate_secrets=False, workspace_target=None,
skill_conflict="skip", yes=False,
)
with (
patch.object(claw_mod, "_find_migration_script", return_value=tmp_path / "s.py"),
patch.object(claw_mod, "_load_migration_module", return_value=fake_mod),
patch.object(claw_mod, "get_config_path", return_value=tmp_path / "config.yaml"),
patch.object(claw_mod, "save_config"),
patch.object(claw_mod, "load_config", return_value={}),
patch.object(claw_mod, "_offer_source_archival") as mock_archival,
):
claw_mod._cmd_migrate(args)
mock_archival.assert_not_called()
def test_execute_cancelled_by_user(self, tmp_path, capsys):
openclaw_dir = tmp_path / ".openclaw"
openclaw_dir.mkdir()
@ -290,6 +486,172 @@ class TestCmdMigrate:
assert call_kwargs["migrate_secrets"] is True
# ---------------------------------------------------------------------------
# _offer_source_archival
# ---------------------------------------------------------------------------
class TestOfferSourceArchival:
"""Test the post-migration archival offer."""
def test_archives_with_auto_yes(self, tmp_path, capsys):
source = tmp_path / ".openclaw"
source.mkdir()
(source / "workspace").mkdir()
(source / "workspace" / "todo.json").write_text("{}")
claw_mod._offer_source_archival(source, auto_yes=True)
captured = capsys.readouterr()
assert "Archived" in captured.out
assert not source.exists()
assert (tmp_path / ".openclaw.pre-migration").is_dir()
def test_skips_when_user_declines(self, tmp_path, capsys):
source = tmp_path / ".openclaw"
source.mkdir()
with patch.object(claw_mod, "prompt_yes_no", return_value=False):
claw_mod._offer_source_archival(source, auto_yes=False)
captured = capsys.readouterr()
assert "Skipped" in captured.out
assert source.is_dir() # Still exists
def test_noop_when_source_missing(self, tmp_path, capsys):
claw_mod._offer_source_archival(tmp_path / "nonexistent", auto_yes=True)
captured = capsys.readouterr()
assert captured.out == "" # No output
def test_shows_state_files(self, tmp_path, capsys):
source = tmp_path / ".openclaw"
source.mkdir()
ws = source / "workspace"
ws.mkdir()
(ws / "todo.json").write_text("{}")
with patch.object(claw_mod, "prompt_yes_no", return_value=False):
claw_mod._offer_source_archival(source, auto_yes=False)
captured = capsys.readouterr()
assert "todo.json" in captured.out
def test_handles_archive_error(self, tmp_path, capsys):
source = tmp_path / ".openclaw"
source.mkdir()
with patch.object(claw_mod, "_archive_directory", side_effect=OSError("permission denied")):
claw_mod._offer_source_archival(source, auto_yes=True)
captured = capsys.readouterr()
assert "Could not archive" in captured.out
# ---------------------------------------------------------------------------
# _cmd_cleanup
# ---------------------------------------------------------------------------
class TestCmdCleanup:
"""Test the cleanup command handler."""
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=[]):
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "No OpenClaw directories found" in captured.out
def test_dry_run_lists_dirs(self, tmp_path, capsys):
openclaw = tmp_path / ".openclaw"
openclaw.mkdir()
ws = openclaw / "workspace"
ws.mkdir()
(ws / "todo.json").write_text("{}")
args = Namespace(source=None, dry_run=True, yes=False)
with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]):
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "Would archive" in captured.out
assert openclaw.is_dir() # Not actually archived
def test_archives_with_yes(self, tmp_path, capsys):
openclaw = tmp_path / ".openclaw"
openclaw.mkdir()
(openclaw / "workspace").mkdir()
(openclaw / "workspace" / "todo.json").write_text("{}")
args = Namespace(source=None, dry_run=False, yes=True)
with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]):
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "Archived" in captured.out
assert "Cleaned up 1" in captured.out
assert not openclaw.exists()
assert (tmp_path / ".openclaw.pre-migration").is_dir()
def test_skips_when_user_declines(self, tmp_path, capsys):
openclaw = tmp_path / ".openclaw"
openclaw.mkdir()
args = Namespace(source=None, dry_run=False, yes=False)
with (
patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]),
patch.object(claw_mod, "prompt_yes_no", return_value=False),
):
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "Skipped" in captured.out
assert openclaw.is_dir()
def test_explicit_source(self, tmp_path, capsys):
custom_dir = tmp_path / "my-openclaw"
custom_dir.mkdir()
(custom_dir / "todo.json").write_text("{}")
args = Namespace(source=str(custom_dir), dry_run=False, yes=True)
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "Archived" in captured.out
assert not custom_dir.exists()
def test_shows_workspace_details(self, tmp_path, capsys):
openclaw = tmp_path / ".openclaw"
openclaw.mkdir()
ws = openclaw / "workspace"
ws.mkdir()
(ws / "todo.json").write_text("{}")
(ws / "SOUL.md").write_text("# Soul")
args = Namespace(source=None, dry_run=True, yes=False)
with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw]):
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "workspace/" in captured.out
assert "todo.json" in captured.out
def test_handles_multiple_dirs(self, tmp_path, capsys):
openclaw = tmp_path / ".openclaw"
openclaw.mkdir()
clawdbot = tmp_path / ".clawdbot"
clawdbot.mkdir()
args = Namespace(source=None, dry_run=False, yes=True)
with patch.object(claw_mod, "_find_openclaw_dirs", return_value=[openclaw, clawdbot]):
claw_mod._cmd_cleanup(args)
captured = capsys.readouterr()
assert "Cleaned up 2" in captured.out
assert not openclaw.exists()
assert not clawdbot.exists()
# ---------------------------------------------------------------------------
# _print_migration_report
# ---------------------------------------------------------------------------

View file

@ -12,10 +12,13 @@ from hermes_cli.commands import (
SUBCOMMANDS,
SlashCommandAutoSuggest,
SlashCommandCompleter,
_TG_NAME_LIMIT,
_clamp_telegram_names,
gateway_help_lines,
resolve_command,
slack_subcommand_map,
telegram_bot_commands,
telegram_menu_commands,
)
@ -504,3 +507,83 @@ class TestGhostText:
def test_no_suggestion_for_non_slash(self):
assert _suggestion("hello") is None
# ---------------------------------------------------------------------------
# Telegram command name clamping (32-char limit)
# ---------------------------------------------------------------------------
class TestClampTelegramNames:
"""Tests for _clamp_telegram_names() — 32-char enforcement + collision."""
def test_short_names_unchanged(self):
entries = [("help", "Show help"), ("status", "Show status")]
result = _clamp_telegram_names(entries, set())
assert result == entries
def test_long_name_truncated(self):
long = "a" * 40
result = _clamp_telegram_names([(long, "desc")], set())
assert len(result) == 1
assert result[0][0] == "a" * _TG_NAME_LIMIT
assert result[0][1] == "desc"
def test_collision_with_reserved_gets_digit_suffix(self):
# The truncated form collides with a reserved name
prefix = "x" * _TG_NAME_LIMIT
long_name = "x" * 40
result = _clamp_telegram_names([(long_name, "d")], reserved={prefix})
assert len(result) == 1
name = result[0][0]
assert len(name) == _TG_NAME_LIMIT
assert name == "x" * (_TG_NAME_LIMIT - 1) + "0"
def test_collision_between_entries_gets_incrementing_digits(self):
# Two long names that truncate to the same 32-char prefix
base = "y" * 40
entries = [(base + "_alpha", "d1"), (base + "_beta", "d2")]
result = _clamp_telegram_names(entries, set())
assert len(result) == 2
assert result[0][0] == "y" * _TG_NAME_LIMIT
assert result[1][0] == "y" * (_TG_NAME_LIMIT - 1) + "0"
def test_collision_with_reserved_and_entries_skips_taken_digits(self):
prefix = "z" * _TG_NAME_LIMIT
digit0 = "z" * (_TG_NAME_LIMIT - 1) + "0"
# Reserve both the plain truncation and digit-0
reserved = {prefix, digit0}
long_name = "z" * 50
result = _clamp_telegram_names([(long_name, "d")], reserved)
assert len(result) == 1
assert result[0][0] == "z" * (_TG_NAME_LIMIT - 1) + "1"
def test_all_digits_exhausted_drops_entry(self):
prefix = "w" * _TG_NAME_LIMIT
# Reserve the plain truncation + all 10 digit slots
reserved = {prefix} | {"w" * (_TG_NAME_LIMIT - 1) + str(d) for d in range(10)}
long_name = "w" * 50
result = _clamp_telegram_names([(long_name, "d")], reserved)
assert result == []
def test_exact_32_chars_not_truncated(self):
name = "a" * _TG_NAME_LIMIT
result = _clamp_telegram_names([(name, "desc")], set())
assert result[0][0] == name
def test_duplicate_short_name_deduplicated(self):
entries = [("foo", "d1"), ("foo", "d2")]
result = _clamp_telegram_names(entries, set())
assert len(result) == 1
assert result[0] == ("foo", "d1")
class TestTelegramMenuCommands:
"""Integration: telegram_menu_commands enforces the 32-char limit."""
def test_all_names_within_limit(self):
menu, _ = telegram_menu_commands(max_commands=100)
for name, _desc in menu:
assert 1 <= len(name) <= _TG_NAME_LIMIT, (
f"Command '{name}' is {len(name)} chars (limit {_TG_NAME_LIMIT})"
)

View file

@ -271,7 +271,7 @@ class TestGatewaySystemServiceRouting:
)
run_calls = []
monkeypatch.setattr(gateway_cli, "run_gateway", lambda verbose=False, replace=False: run_calls.append((verbose, replace)))
monkeypatch.setattr(gateway_cli, "run_gateway", lambda verbose=0, quiet=False, replace=False: run_calls.append((verbose, quiet, replace)))
monkeypatch.setattr(gateway_cli, "kill_gateway_processes", lambda force=False: 0)
try:
@ -339,6 +339,102 @@ class TestDetectVenvDir:
assert result is None
class TestSystemUnitHermesHome:
"""HERMES_HOME in system units must reference the target user, not root."""
def test_system_unit_uses_target_user_home_not_calling_user(self, monkeypatch):
# Simulate sudo: Path.home() returns /root, target user is alice
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.delenv("HERMES_HOME", raising=False)
monkeypatch.setattr(
gateway_cli, "_system_service_identity",
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
)
monkeypatch.setattr(
gateway_cli, "_build_user_local_paths",
lambda home, existing: [],
)
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
assert 'HERMES_HOME=/home/alice/.hermes' in unit
assert '/root/.hermes' not in unit
def test_system_unit_remaps_profile_to_target_user(self, monkeypatch):
# Simulate sudo with a profile: HERMES_HOME was resolved under root
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
monkeypatch.setattr(
gateway_cli, "_system_service_identity",
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
)
monkeypatch.setattr(
gateway_cli, "_build_user_local_paths",
lambda home, existing: [],
)
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
assert 'HERMES_HOME=/home/alice/.hermes/profiles/coder' in unit
assert '/root/' not in unit
def test_system_unit_preserves_custom_hermes_home(self, monkeypatch):
# Custom HERMES_HOME not under any user's home — keep as-is
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/opt/hermes-shared")
monkeypatch.setattr(
gateway_cli, "_system_service_identity",
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
)
monkeypatch.setattr(
gateway_cli, "_build_user_local_paths",
lambda home, existing: [],
)
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
assert 'HERMES_HOME=/opt/hermes-shared' in unit
def test_user_unit_unaffected_by_change(self):
# User-scope units should still use the calling user's HERMES_HOME
unit = gateway_cli.generate_systemd_unit(system=False)
hermes_home = str(gateway_cli.get_hermes_home().resolve())
assert f'HERMES_HOME={hermes_home}' in unit
class TestHermesHomeForTargetUser:
"""Unit tests for _hermes_home_for_target_user()."""
def test_remaps_default_home(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.delenv("HERMES_HOME", raising=False)
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/home/alice/.hermes"
def test_remaps_profile_path(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/home/alice/.hermes/profiles/coder"
def test_keeps_custom_path(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/opt/hermes")
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/opt/hermes"
def test_noop_when_same_user(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/home/alice")))
monkeypatch.delenv("HERMES_HOME", raising=False)
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/home/alice/.hermes"
class TestGeneratedUnitUsesDetectedVenv:
def test_systemd_unit_uses_dot_venv_when_detected(self, tmp_path, monkeypatch):
dot_venv = tmp_path / ".venv"

View file

@ -0,0 +1,54 @@
from types import SimpleNamespace
from unittest.mock import patch
from hermes_cli.config import (
format_managed_message,
get_managed_system,
recommended_update_command,
)
from hermes_cli.main import cmd_update
from tools.skills_hub import OptionalSkillSource
def test_get_managed_system_homebrew(monkeypatch):
monkeypatch.setenv("HERMES_MANAGED", "homebrew")
assert get_managed_system() == "Homebrew"
assert recommended_update_command() == "brew upgrade hermes-agent"
def test_format_managed_message_homebrew(monkeypatch):
monkeypatch.setenv("HERMES_MANAGED", "homebrew")
message = format_managed_message("update Hermes Agent")
assert "managed by Homebrew" in message
assert "brew upgrade hermes-agent" in message
def test_recommended_update_command_defaults_to_hermes_update(monkeypatch):
monkeypatch.delenv("HERMES_MANAGED", raising=False)
assert recommended_update_command() == "hermes update"
def test_cmd_update_blocks_managed_homebrew(monkeypatch, capsys):
monkeypatch.setenv("HERMES_MANAGED", "homebrew")
with patch("hermes_cli.main.subprocess.run") as mock_run:
cmd_update(SimpleNamespace())
assert not mock_run.called
captured = capsys.readouterr()
assert "managed by Homebrew" in captured.err
assert "brew upgrade hermes-agent" in captured.err
def test_optional_skill_source_honors_env_override(monkeypatch, tmp_path):
optional_dir = tmp_path / "optional-skills"
optional_dir.mkdir()
monkeypatch.setenv("HERMES_OPTIONAL_SKILLS", str(optional_dir))
source = OptionalSkillSource()
assert source._optional_dir == optional_dir

View file

@ -0,0 +1,52 @@
"""Tests for credential exclusion during profile export.
Profile exports should NEVER include auth.json or .env these contain
API keys, OAuth tokens, and credential pool data. Users share exported
profiles; leaking credentials in the archive is a security issue.
"""
import tarfile
from pathlib import Path
from hermes_cli.profiles import export_profile, _DEFAULT_EXPORT_EXCLUDE_ROOT
class TestCredentialExclusion:
def test_auth_json_in_default_exclude_set(self):
"""auth.json must be in the default export exclusion set."""
assert "auth.json" in _DEFAULT_EXPORT_EXCLUDE_ROOT
def test_dotenv_in_default_exclude_set(self):
""".env must be in the default export exclusion set."""
assert ".env" in _DEFAULT_EXPORT_EXCLUDE_ROOT
def test_named_profile_export_excludes_auth(self, tmp_path, monkeypatch):
"""Named profile export must not contain auth.json or .env."""
profiles_root = tmp_path / "profiles"
profile_dir = profiles_root / "testprofile"
profile_dir.mkdir(parents=True)
# Create a profile with credentials
(profile_dir / "config.yaml").write_text("model: gpt-4\n")
(profile_dir / "auth.json").write_text('{"tokens": {"access": "sk-secret"}}')
(profile_dir / ".env").write_text("OPENROUTER_API_KEY=sk-secret-key\n")
(profile_dir / "SOUL.md").write_text("I am helpful.\n")
(profile_dir / "memories").mkdir()
(profile_dir / "memories" / "MEMORY.md").write_text("# Memories\n")
monkeypatch.setattr("hermes_cli.profiles._get_profiles_root", lambda: profiles_root)
monkeypatch.setattr("hermes_cli.profiles.get_profile_dir", lambda n: profile_dir)
monkeypatch.setattr("hermes_cli.profiles.validate_profile_name", lambda n: None)
output = tmp_path / "export.tar.gz"
result = export_profile("testprofile", str(output))
# Check archive contents
with tarfile.open(result, "r:gz") as tf:
names = tf.getnames()
assert any("config.yaml" in n for n in names), "config.yaml should be in export"
assert any("SOUL.md" in n for n in names), "SOUL.md should be in export"
assert not any("auth.json" in n for n in names), "auth.json must NOT be in export"
assert not any(".env" in n for n in names), ".env must NOT be in export"

View file

@ -6,6 +6,7 @@ and shell completion generation.
"""
import json
import io
import os
import tarfile
from pathlib import Path
@ -449,10 +450,187 @@ class TestExportImport:
with pytest.raises(FileExistsError):
import_profile(str(archive_path), name="coder")
def test_import_rejects_traversal_archive_member(self, profile_env, tmp_path):
archive_path = tmp_path / "export" / "evil.tar.gz"
archive_path.parent.mkdir(parents=True, exist_ok=True)
escape_path = tmp_path / "escape.txt"
with tarfile.open(archive_path, "w:gz") as tf:
info = tarfile.TarInfo("../../escape.txt")
data = b"pwned"
info.size = len(data)
tf.addfile(info, io.BytesIO(data))
with pytest.raises(ValueError, match="Unsafe archive member path"):
import_profile(str(archive_path), name="coder")
assert not escape_path.exists()
assert not get_profile_dir("coder").exists()
def test_import_rejects_absolute_archive_member(self, profile_env, tmp_path):
archive_path = tmp_path / "export" / "evil-abs.tar.gz"
archive_path.parent.mkdir(parents=True, exist_ok=True)
absolute_target = tmp_path / "abs-escape.txt"
with tarfile.open(archive_path, "w:gz") as tf:
info = tarfile.TarInfo(str(absolute_target))
data = b"pwned"
info.size = len(data)
tf.addfile(info, io.BytesIO(data))
with pytest.raises(ValueError, match="Unsafe archive member path"):
import_profile(str(archive_path), name="coder")
assert not absolute_target.exists()
assert not get_profile_dir("coder").exists()
def test_export_nonexistent_raises(self, profile_env, tmp_path):
with pytest.raises(FileNotFoundError):
export_profile("nonexistent", str(tmp_path / "out.tar.gz"))
# ---------------------------------------------------------------
# Default profile export / import
# ---------------------------------------------------------------
def test_export_default_creates_valid_archive(self, profile_env, tmp_path):
"""Exporting the default profile produces a valid tar.gz."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("model: test")
output = tmp_path / "export" / "default.tar.gz"
output.parent.mkdir(parents=True, exist_ok=True)
result = export_profile("default", str(output))
assert Path(result).exists()
assert tarfile.is_tarfile(str(result))
def test_export_default_includes_profile_data(self, profile_env, tmp_path):
"""Profile data files end up in the archive (credentials excluded)."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("model: test")
(default_dir / ".env").write_text("KEY=val")
(default_dir / "SOUL.md").write_text("Be nice.")
mem_dir = default_dir / "memories"
mem_dir.mkdir(exist_ok=True)
(mem_dir / "MEMORY.md").write_text("remember this")
output = tmp_path / "export" / "default.tar.gz"
output.parent.mkdir(parents=True, exist_ok=True)
export_profile("default", str(output))
with tarfile.open(str(output), "r:gz") as tf:
names = tf.getnames()
assert "default/config.yaml" in names
assert "default/.env" not in names # credentials excluded
assert "default/SOUL.md" in names
assert "default/memories/MEMORY.md" in names
def test_export_default_excludes_infrastructure(self, profile_env, tmp_path):
"""Repo checkout, worktrees, profiles, databases are excluded."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("ok")
# Create dirs/files that should be excluded
for d in ("hermes-agent", ".worktrees", "profiles", "bin",
"image_cache", "logs", "sandboxes", "checkpoints"):
sub = default_dir / d
sub.mkdir(exist_ok=True)
(sub / "marker.txt").write_text("excluded")
for f in ("state.db", "gateway.pid", "gateway_state.json",
"processes.json", "errors.log", ".hermes_history",
"active_profile", ".update_check", "auth.lock"):
(default_dir / f).write_text("excluded")
output = tmp_path / "export" / "default.tar.gz"
output.parent.mkdir(parents=True, exist_ok=True)
export_profile("default", str(output))
with tarfile.open(str(output), "r:gz") as tf:
names = tf.getnames()
# Config is present
assert "default/config.yaml" in names
# Infrastructure excluded
excluded_prefixes = [
"default/hermes-agent", "default/.worktrees", "default/profiles",
"default/bin", "default/image_cache", "default/logs",
"default/sandboxes", "default/checkpoints",
]
for prefix in excluded_prefixes:
assert not any(n.startswith(prefix) for n in names), \
f"Expected {prefix} to be excluded but found it in archive"
excluded_files = [
"default/state.db", "default/gateway.pid",
"default/gateway_state.json", "default/processes.json",
"default/errors.log", "default/.hermes_history",
"default/active_profile", "default/.update_check",
"default/auth.lock",
]
for f in excluded_files:
assert f not in names, f"Expected {f} to be excluded"
def test_export_default_excludes_pycache_at_any_depth(self, profile_env, tmp_path):
"""__pycache__ dirs are excluded even inside nested directories."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("ok")
nested = default_dir / "skills" / "my-skill" / "__pycache__"
nested.mkdir(parents=True)
(nested / "cached.pyc").write_text("bytecode")
output = tmp_path / "export" / "default.tar.gz"
output.parent.mkdir(parents=True, exist_ok=True)
export_profile("default", str(output))
with tarfile.open(str(output), "r:gz") as tf:
names = tf.getnames()
assert not any("__pycache__" in n for n in names)
def test_import_default_without_name_raises(self, profile_env, tmp_path):
"""Importing a default export without --name gives clear guidance."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("ok")
archive = tmp_path / "export" / "default.tar.gz"
archive.parent.mkdir(parents=True, exist_ok=True)
export_profile("default", str(archive))
with pytest.raises(ValueError, match="Cannot import as 'default'"):
import_profile(str(archive))
def test_import_default_with_explicit_default_name_raises(self, profile_env, tmp_path):
"""Explicitly importing as 'default' is also rejected."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("ok")
archive = tmp_path / "export" / "default.tar.gz"
archive.parent.mkdir(parents=True, exist_ok=True)
export_profile("default", str(archive))
with pytest.raises(ValueError, match="Cannot import as 'default'"):
import_profile(str(archive), name="default")
def test_import_default_export_with_new_name_roundtrip(self, profile_env, tmp_path):
"""Export default → import under a different name → data preserved."""
default_dir = get_profile_dir("default")
(default_dir / "config.yaml").write_text("model: opus")
mem_dir = default_dir / "memories"
mem_dir.mkdir(exist_ok=True)
(mem_dir / "MEMORY.md").write_text("important fact")
archive = tmp_path / "export" / "default.tar.gz"
archive.parent.mkdir(parents=True, exist_ok=True)
export_profile("default", str(archive))
imported = import_profile(str(archive), name="backup")
assert imported.is_dir()
assert (imported / "config.yaml").read_text() == "model: opus"
assert (imported / "memories" / "MEMORY.md").read_text() == "important fact"
# ===================================================================
# TestProfileIsolation

View file

@ -1,12 +1,13 @@
"""Tests for set_config_value — verifying secrets route to .env and config to config.yaml."""
import argparse
import os
from pathlib import Path
from unittest.mock import patch, call
import pytest
from hermes_cli.config import set_config_value
from hermes_cli.config import set_config_value, config_command
@pytest.fixture(autouse=True)
@ -125,3 +126,42 @@ class TestConfigYamlRouting:
"TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE=true" in env_content
or "TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE=True" in env_content
)
# ---------------------------------------------------------------------------
# Empty / falsy values — regression tests for #4277
# ---------------------------------------------------------------------------
class TestFalsyValues:
"""config set should accept empty strings and falsy values like '0'."""
def test_empty_string_routes_to_env(self, _isolated_hermes_home):
"""Blanking an API key should write an empty value to .env."""
set_config_value("OPENROUTER_API_KEY", "")
env_content = _read_env(_isolated_hermes_home)
assert "OPENROUTER_API_KEY=" in env_content
def test_empty_string_routes_to_config(self, _isolated_hermes_home):
"""Blanking a config key should write an empty string to config.yaml."""
set_config_value("model", "")
config = _read_config(_isolated_hermes_home)
assert "model: ''" in config or "model: \"\"" in config
def test_zero_routes_to_config(self, _isolated_hermes_home):
"""Setting a config key to '0' should write 0 to config.yaml."""
set_config_value("verbose", "0")
config = _read_config(_isolated_hermes_home)
assert "verbose: 0" in config
def test_config_command_rejects_missing_value(self):
"""config set with no value arg (None) should still exit."""
args = argparse.Namespace(config_command="set", key="model", value=None)
with pytest.raises(SystemExit):
config_command(args)
def test_config_command_accepts_empty_string(self, _isolated_hermes_home):
"""config set KEY '' should not exit — it should set the value."""
args = argparse.Namespace(config_command="set", key="model", value="")
config_command(args)
config = _read_config(_isolated_hermes_home)
assert "model" in config

View file

@ -1,8 +1,10 @@
"""Tests for setup_model_provider — verifies the delegation to
select_provider_and_model() and config dict sync."""
import json
import sys
import types
from hermes_cli.auth import _update_config_for_provider, get_active_provider
from hermes_cli.auth import get_active_provider
from hermes_cli.config import load_config, save_config
from hermes_cli.setup import setup_model_provider
@ -25,249 +27,201 @@ def _clear_provider_env(monkeypatch):
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: (
_maybe_keep_current_tts(q, c) if _maybe_keep_current_tts(q, c) is not None
else d
))
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *a, **kw: False)
def test_nous_oauth_setup_keeps_current_model_when_syncing_disk_provider(
tmp_path, monkeypatch
):
def _write_model_config(tmp_path, provider, base_url="", model_name="test-model"):
"""Simulate what a _model_flow_* function writes to disk."""
cfg = load_config()
m = cfg.get("model")
if not isinstance(m, dict):
m = {"default": m} if m else {}
cfg["model"] = m
m["provider"] = provider
if base_url:
m["base_url"] = base_url
if model_name:
m["default"] = model_name
save_config(cfg)
def test_setup_delegates_to_select_provider_and_model(tmp_path, monkeypatch):
"""setup_model_provider calls select_provider_and_model and syncs config."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 1 # Nous Portal
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
assert choices[-1] == "Keep current (anthropic/claude-opus-4.6)"
return len(choices) - 1
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_select():
_write_model_config(tmp_path, "custom", "http://localhost:11434/v1", "qwen3.5:32b")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
def _fake_login_nous(*args, **kwargs):
auth_path = tmp_path / "auth.json"
auth_path.write_text(json.dumps({"active_provider": "nous", "providers": {}}))
_update_config_for_provider("nous", "https://inference.example.com/v1")
monkeypatch.setattr("hermes_cli.auth._login_nous", _fake_login_nous)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["gemini-3-flash"],
)
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "custom"
assert reloaded["model"]["base_url"] == "http://localhost:11434/v1"
assert reloaded["model"]["default"] == "qwen3.5:32b"
def test_setup_syncs_openrouter_from_disk(tmp_path, monkeypatch):
"""When select_provider_and_model saves OpenRouter config to disk,
the wizard's config dict picks it up."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
assert isinstance(config.get("model"), str) # fresh install
def fake_select():
_write_model_config(tmp_path, "openrouter", model_name="anthropic/claude-opus-4.6")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "openrouter"
def test_setup_syncs_nous_from_disk(tmp_path, monkeypatch):
"""Nous OAuth writes config to disk; wizard config dict must pick it up."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
def fake_select():
_write_model_config(tmp_path, "nous", "https://inference.example.com/v1", "gemini-3-flash")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "nous"
assert reloaded["model"]["base_url"] == "https://inference.example.com/v1"
assert reloaded["model"]["default"] == "anthropic/claude-opus-4.6"
def test_custom_setup_clears_active_oauth_provider(tmp_path, monkeypatch):
def test_setup_custom_providers_synced(tmp_path, monkeypatch):
"""custom_providers written by select_provider_and_model must survive."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
auth_path = tmp_path / "auth.json"
auth_path.write_text(json.dumps({"active_provider": "nous", "providers": {}}))
_stub_tts(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 3
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_select():
_write_model_config(tmp_path, "custom", "http://localhost:8080/v1", "llama3")
cfg = load_config()
cfg["custom_providers"] = [{"name": "Local", "base_url": "http://localhost:8080/v1"}]
save_config(cfg)
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
# _model_flow_custom uses builtins.input (URL, key, model, context_length)
input_values = iter([
"https://custom.example/v1",
"custom-api-key",
"custom/model",
"", # context_length (blank = auto-detect)
])
monkeypatch.setattr("builtins.input", lambda _prompt="": next(input_values))
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.main._save_custom_provider", lambda *args, **kwargs: None)
monkeypatch.setattr(
"hermes_cli.models.probe_api_models",
lambda api_key, base_url: {"models": ["m"], "probed_url": base_url + "/models"},
)
setup_model_provider(config)
# Core assertion: switching to custom endpoint clears OAuth provider
assert get_active_provider() is None
# _model_flow_custom writes config via its own load/save cycle
reloaded = load_config()
if isinstance(reloaded.get("model"), dict):
assert reloaded["model"].get("provider") == "custom"
assert reloaded["model"].get("default") == "custom/model"
def test_codex_setup_uses_runtime_access_token_for_live_model_list(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("OPENROUTER_API_KEY", "or-test-key")
_clear_provider_env(monkeypatch)
monkeypatch.setenv("OPENROUTER_API_KEY", "or-test-key")
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 2 # OpenAI Codex
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
return 0
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.auth._login_openai_codex", lambda *args, **kwargs: None)
monkeypatch.setattr(
"hermes_cli.auth.resolve_codex_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://chatgpt.com/backend-api/codex",
"api_key": "codex-access-token",
},
)
captured = {}
def _fake_get_codex_model_ids(access_token=None):
captured["access_token"] = access_token
return ["gpt-5.2-codex", "gpt-5.2"]
monkeypatch.setattr(
"hermes_cli.codex_models.get_codex_model_ids",
_fake_get_codex_model_ids,
)
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert reloaded.get("custom_providers") == [{"name": "Local", "base_url": "http://localhost:8080/v1"}]
assert captured["access_token"] == "codex-access-token"
def test_setup_cancel_preserves_existing_config(tmp_path, monkeypatch):
"""When the user cancels provider selection, existing config is preserved."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
# Pre-set a provider
_write_model_config(tmp_path, "openrouter", model_name="gpt-4o")
config = load_config()
assert config["model"]["provider"] == "openrouter"
def fake_select():
pass # user cancelled — nothing written to disk
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "openrouter"
assert reloaded["model"]["default"] == "gpt-4o"
def test_setup_exception_in_select_gracefully_handled(tmp_path, monkeypatch):
"""If select_provider_and_model raises, setup continues with existing config."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
def fake_select():
raise RuntimeError("something broke")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
# Should not raise
setup_model_provider(config)
def test_setup_keyboard_interrupt_gracefully_handled(tmp_path, monkeypatch):
"""KeyboardInterrupt during provider selection is handled."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
def fake_select():
raise KeyboardInterrupt()
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
def test_codex_setup_uses_runtime_access_token_for_live_model_list(tmp_path, monkeypatch):
"""Codex model list fetching uses the runtime access token."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("OPENROUTER_API_KEY", "or-test-key")
_clear_provider_env(monkeypatch)
monkeypatch.setenv("OPENROUTER_API_KEY", "or-test-key")
config = load_config()
_stub_tts(monkeypatch)
def fake_select():
_write_model_config(tmp_path, "openai-codex", "https://api.openai.com/v1", "gpt-4o")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "openai-codex"
assert reloaded["model"]["default"] == "gpt-5.2-codex"
assert reloaded["model"]["base_url"] == "https://chatgpt.com/backend-api/codex"
def test_nous_setup_sets_managed_openai_tts_when_unconfigured(tmp_path, monkeypatch, capsys):
monkeypatch.setenv("HERMES_ENABLE_NOUS_MANAGED_TOOLS", "1")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 1
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
return len(choices) - 1
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
def _fake_login_nous(*args, **kwargs):
auth_path = tmp_path / "auth.json"
auth_path.write_text(json.dumps({"active_provider": "nous", "providers": {"nous": {"access_token": "nous-token"}}}))
_update_config_for_provider("nous", "https://inference.example.com/v1")
monkeypatch.setattr("hermes_cli.auth._login_nous", _fake_login_nous)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["gemini-3-flash"],
)
setup_model_provider(config)
out = capsys.readouterr().out
assert config["tts"]["provider"] == "openai"
assert "Nous subscription enables managed web tools" in out
assert "OpenAI TTS via your Nous subscription" in out
def test_nous_setup_preserves_existing_tts_provider(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
config["tts"] = {"provider": "elevenlabs"}
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 1
if question == "Configure vision:":
return len(choices) - 1
if question == "Select default model:":
return len(choices) - 1
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr(
"hermes_cli.auth._login_nous",
lambda *args, **kwargs: (tmp_path / "auth.json").write_text(
json.dumps({"active_provider": "nous", "providers": {"nous": {"access_token": "nous-token"}}})
),
)
monkeypatch.setattr(
"hermes_cli.auth.resolve_nous_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://inference.example.com/v1",
"api_key": "nous-key",
},
)
monkeypatch.setattr(
"hermes_cli.auth.fetch_nous_models",
lambda *args, **kwargs: ["gemini-3-flash"],
)
setup_model_provider(config)
assert config["tts"]["provider"] == "elevenlabs"
def test_modal_setup_can_use_nous_subscription_without_modal_creds(tmp_path, monkeypatch, capsys):

View file

@ -1,4 +1,9 @@
"""Regression tests for interactive setup provider/model persistence."""
"""Regression tests for interactive setup provider/model persistence.
Since setup_model_provider delegates to select_provider_and_model()
from hermes_cli.main, these tests mock the delegation point and verify
that the setup wizard correctly syncs config from disk after the call.
"""
from __future__ import annotations
@ -14,19 +19,6 @@ def _maybe_keep_current_tts(question, choices):
return len(choices) - 1
def _read_env(home):
env_path = home / ".env"
data = {}
if not env_path.exists():
return data
for line in env_path.read_text().splitlines():
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
data[k] = v
return data
def _clear_provider_env(monkeypatch):
for key in (
"HERMES_INFERENCE_PROVIDER",
@ -45,419 +37,375 @@ def _clear_provider_env(monkeypatch):
monkeypatch.delenv(key, raising=False)
def _stub_tts(monkeypatch):
monkeypatch.setattr("hermes_cli.setup.prompt_choice", lambda q, c, d=0: (
_maybe_keep_current_tts(q, c) if _maybe_keep_current_tts(q, c) is not None
else d
))
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *a, **kw: False)
def _write_model_config(provider, base_url="", model_name="test-model"):
"""Simulate what a _model_flow_* function writes to disk."""
cfg = load_config()
m = cfg.get("model")
if not isinstance(m, dict):
m = {"default": m} if m else {}
cfg["model"] = m
m["provider"] = provider
if base_url:
m["base_url"] = base_url
else:
m.pop("base_url", None)
if model_name:
m["default"] = model_name
m.pop("api_mode", None)
save_config(cfg)
def test_setup_keep_current_custom_from_config_does_not_fall_through(tmp_path, monkeypatch):
"""Keep-current custom should not fall through to the generic model menu."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
save_env_value("OPENAI_BASE_URL", "https://example.invalid/v1")
save_env_value("OPENAI_API_KEY", "custom-key")
_stub_tts(monkeypatch)
# Pre-set custom provider
_write_model_config("custom", "http://localhost:8080/v1", "local-model")
config = load_config()
config["model"] = {
"default": "custom/model",
"provider": "custom",
"base_url": "https://example.invalid/v1",
}
save_config(config)
assert config["model"]["provider"] == "custom"
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
assert choices[-1] == "Keep current (Custom: https://example.invalid/v1)"
return len(choices) - 1
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError("Model menu should not appear for keep-current custom")
def fake_select():
pass # user chose "cancel" or "keep current"
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "custom"
assert reloaded["model"]["default"] == "custom/model"
assert reloaded["model"]["base_url"] == "https://example.invalid/v1"
assert reloaded["model"]["base_url"] == "http://localhost:8080/v1"
def test_setup_custom_endpoint_saves_working_v1_base_url(tmp_path, monkeypatch):
def test_setup_keep_current_config_provider_uses_provider_specific_model_menu(
tmp_path, monkeypatch
):
"""Keeping current provider preserves the config on disk."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
_write_model_config("zai", "https://open.bigmodel.cn/api/paas/v4", "glm-5")
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 3 # Custom endpoint
if question == "Configure vision:":
return len(choices) - 1 # Skip
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_select():
pass # keep current
# _model_flow_custom uses builtins.input (URL, key, model, context_length)
input_values = iter([
"http://localhost:8000",
"local-key",
"llm",
"", # context_length (blank = auto-detect)
])
monkeypatch.setattr("builtins.input", lambda _prompt="": next(input_values))
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
monkeypatch.setattr("hermes_cli.main._save_custom_provider", lambda *args, **kwargs: None)
monkeypatch.setattr(
"hermes_cli.models.probe_api_models",
lambda api_key, base_url: {
"models": ["llm"],
"probed_url": "http://localhost:8000/v1/models",
"resolved_base_url": "http://localhost:8000/v1",
"suggested_base_url": "http://localhost:8000/v1",
"used_fallback": True,
},
)
setup_model_provider(config)
env = _read_env(tmp_path)
# _model_flow_custom saves env vars and config to disk
assert env.get("OPENAI_BASE_URL") == "http://localhost:8000/v1"
assert env.get("OPENAI_API_KEY") == "local-key"
# The model config is saved as a dict by _model_flow_custom
reloaded = load_config()
model_cfg = reloaded.get("model", {})
if isinstance(model_cfg, dict):
assert model_cfg.get("provider") == "custom"
assert model_cfg.get("default") == "llm"
def test_setup_keep_current_config_provider_uses_provider_specific_model_menu(tmp_path, monkeypatch):
"""Keep-current should respect config-backed providers, not fall back to OpenRouter."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
config["model"] = {
"default": "claude-opus-4-6",
"provider": "anthropic",
}
save_config(config)
captured = {"provider_choices": None, "model_choices": None}
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
captured["provider_choices"] = list(choices)
assert choices[-1] == "Keep current (Anthropic)"
return len(choices) - 1
if question == "Configure vision:":
assert question == "Configure vision:"
assert choices[-1] == "Skip for now"
return len(choices) - 1
if question == "Select default model:":
captured["model_choices"] = list(choices)
return len(choices) - 1 # keep current model
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.models.provider_model_ids", lambda provider: [])
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
save_config(config)
assert captured["provider_choices"] is not None
assert captured["model_choices"] is not None
assert captured["model_choices"][0] == "claude-opus-4-6"
assert "anthropic/claude-opus-4.6 (recommended)" not in captured["model_choices"]
def test_setup_keep_current_anthropic_can_configure_openai_vision_default(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
config["model"] = {
"default": "claude-opus-4-6",
"provider": "anthropic",
}
save_config(config)
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
assert choices[-1] == "Keep current (Anthropic)"
return len(choices) - 1
if question == "Configure vision:":
return 1
if question == "Select vision model:":
assert choices[-1] == "Use default (gpt-4o-mini)"
return len(choices) - 1
if question == "Select default model:":
assert choices[-1] == "Keep current (claude-opus-4-6)"
return len(choices) - 1
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr(
"hermes_cli.setup.prompt",
lambda message, *args, **kwargs: "sk-openai" if "OpenAI API key" in message else "",
)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.models.provider_model_ids", lambda provider: [])
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
env = _read_env(tmp_path)
assert env.get("OPENAI_API_KEY") == "sk-openai"
assert env.get("OPENAI_BASE_URL") == "https://api.openai.com/v1"
assert env.get("AUXILIARY_VISION_MODEL") == "gpt-4o-mini"
def test_setup_copilot_uses_gh_auth_and_saves_provider(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
assert choices[14] == "GitHub Copilot (uses GITHUB_TOKEN or gh auth token)"
return 14
if question == "Select default model:":
assert "gpt-4.1" in choices
assert "gpt-5.4" in choices
return choices.index("gpt-5.4")
if question == "Select reasoning effort:":
assert "low" in choices
assert "high" in choices
return choices.index("high")
if question == "Configure vision:":
return len(choices) - 1
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_prompt(message, *args, **kwargs):
raise AssertionError(f"Unexpected prompt call: {message}")
def fake_get_auth_status(provider_id):
if provider_id == "copilot":
return {"logged_in": True}
return {"logged_in": False}
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", fake_prompt)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.auth.get_auth_status", fake_get_auth_status)
monkeypatch.setattr(
"hermes_cli.auth.resolve_api_key_provider_credentials",
lambda provider_id: {
"provider": provider_id,
"api_key": "gh-cli-token",
"base_url": "https://api.githubcopilot.com",
"source": "gh auth token",
},
)
monkeypatch.setattr(
"hermes_cli.models.fetch_github_model_catalog",
lambda api_key: [
{
"id": "gpt-4.1",
"capabilities": {"type": "chat", "supports": {}},
"supported_endpoints": ["/chat/completions"],
},
{
"id": "gpt-5.4",
"capabilities": {"type": "chat", "supports": {"reasoning_effort": ["low", "medium", "high"]}},
"supported_endpoints": ["/responses"],
},
],
)
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
save_config(config)
env = _read_env(tmp_path)
reloaded = load_config()
assert env.get("GITHUB_TOKEN") is None
assert reloaded["model"]["provider"] == "copilot"
assert reloaded["model"]["base_url"] == "https://api.githubcopilot.com"
assert reloaded["model"]["default"] == "gpt-5.4"
assert reloaded["model"]["api_mode"] == "codex_responses"
assert reloaded["agent"]["reasoning_effort"] == "high"
def test_setup_copilot_acp_uses_model_picker_and_saves_provider(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
assert choices[15] == "GitHub Copilot ACP (spawns `copilot --acp --stdio`)"
return 15
if question == "Select default model:":
assert "gpt-4.1" in choices
assert "gpt-5.4" in choices
return choices.index("gpt-5.4")
if question == "Configure vision:":
return len(choices) - 1
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_prompt(message, *args, **kwargs):
raise AssertionError(f"Unexpected prompt call: {message}")
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt", fake_prompt)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.auth.get_auth_status", lambda provider_id: {"logged_in": provider_id == "copilot-acp"})
monkeypatch.setattr(
"hermes_cli.auth.resolve_api_key_provider_credentials",
lambda provider_id: {
"provider": "copilot",
"api_key": "gh-cli-token",
"base_url": "https://api.githubcopilot.com",
"source": "gh auth token",
},
)
monkeypatch.setattr(
"hermes_cli.models.fetch_github_model_catalog",
lambda api_key: [
{
"id": "gpt-4.1",
"capabilities": {"type": "chat", "supports": {}},
"supported_endpoints": ["/chat/completions"],
},
{
"id": "gpt-5.4",
"capabilities": {"type": "chat", "supports": {"reasoning_effort": ["low", "medium", "high"]}},
"supported_endpoints": ["/responses"],
},
],
)
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert reloaded["model"]["provider"] == "copilot-acp"
assert reloaded["model"]["base_url"] == "acp://copilot"
assert reloaded["model"]["default"] == "gpt-5.4"
assert reloaded["model"]["api_mode"] == "chat_completions"
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "zai"
def test_setup_switch_custom_to_codex_clears_custom_endpoint_and_updates_config(tmp_path, monkeypatch):
"""Switching from custom to Codex should clear custom endpoint overrides."""
def test_setup_same_provider_rotation_strategy_saved_for_multi_credential_pool(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
save_env_value("OPENROUTER_API_KEY", "or-key")
save_env_value("OPENAI_BASE_URL", "https://example.invalid/v1")
save_env_value("OPENAI_API_KEY", "sk-custom")
save_env_value("OPENROUTER_API_KEY", "sk-or")
# Pre-write config so the pool step sees provider="openrouter"
_write_model_config("openrouter", "", "anthropic/claude-opus-4.6")
config = load_config()
config["model"] = {
"default": "custom/model",
"provider": "custom",
"base_url": "https://example.invalid/v1",
}
save_config(config)
class _Entry:
def __init__(self, label):
self.label = label
class _Pool:
def entries(self):
return [_Entry("primary"), _Entry("secondary")]
def fake_select():
pass # no-op — config already has provider set
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 2 # OpenAI Codex
if question == "Select default model:":
if "rotation strategy" in question:
return 1 # round robin
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
return default
def fake_prompt_yes_no(question, default=True):
return False
# Patch directly on the module objects to ensure local imports pick them up.
import hermes_cli.main as _main_mod
import hermes_cli.setup as _setup_mod
import agent.credential_pool as _pool_mod
import agent.auxiliary_client as _aux_mod
monkeypatch.setattr(_main_mod, "select_provider_and_model", fake_select)
# NOTE: _stub_tts overwrites prompt_choice, so set our mock AFTER it.
_stub_tts(monkeypatch)
monkeypatch.setattr(_setup_mod, "prompt_choice", fake_prompt_choice)
monkeypatch.setattr(_setup_mod, "prompt_yes_no", fake_prompt_yes_no)
monkeypatch.setattr(_setup_mod, "prompt", lambda *args, **kwargs: "")
monkeypatch.setattr(_pool_mod, "load_pool", lambda provider: _Pool())
monkeypatch.setattr(_aux_mod, "get_available_vision_backends", lambda: [])
setup_model_provider(config)
# The pool has 2 entries, so the strategy prompt should fire
strategy = config.get("credential_pool_strategies", {}).get("openrouter")
assert strategy == "round_robin", f"Expected round_robin but got {strategy}"
def test_setup_same_provider_fallback_can_add_another_credential(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
save_env_value("OPENROUTER_API_KEY", "or-key")
# Pre-write config so the pool step sees provider="openrouter"
_write_model_config("openrouter", "", "anthropic/claude-opus-4.6")
config = load_config()
pool_sizes = iter([1, 2])
add_calls = []
class _Entry:
def __init__(self, label):
self.label = label
class _Pool:
def __init__(self, size):
self._size = size
def entries(self):
return [_Entry(f"cred-{idx}") for idx in range(self._size)]
def fake_load_pool(provider):
return _Pool(next(pool_sizes))
def fake_auth_add_command(args):
add_calls.append(args.provider)
def fake_select():
pass # no-op — config already has provider set
def fake_prompt_choice(question, choices, default=0):
if question == "Select same-provider rotation strategy:":
return 0
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
return default
yes_no_answers = iter([True, False])
def fake_prompt_yes_no(question, default=True):
if question == "Add another credential for same-provider fallback?":
return next(yes_no_answers)
return False
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
_stub_tts(monkeypatch)
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", fake_prompt_yes_no)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("agent.credential_pool.load_pool", fake_load_pool)
monkeypatch.setattr("hermes_cli.auth_commands.auth_add_command", fake_auth_add_command)
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
assert add_calls == ["openrouter"]
assert config.get("credential_pool_strategies", {}).get("openrouter") == "fill_first"
def test_setup_pool_step_shows_manual_vs_auto_detected_counts(tmp_path, monkeypatch, capsys):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
save_env_value("OPENROUTER_API_KEY", "or-key")
# Pre-write config so the pool step sees provider="openrouter"
_write_model_config("openrouter", "", "anthropic/claude-opus-4.6")
config = load_config()
class _Entry:
def __init__(self, label, source):
self.label = label
self.source = source
class _Pool:
def entries(self):
return [
_Entry("primary", "manual"),
_Entry("secondary", "manual"),
_Entry("OPENROUTER_API_KEY", "env:OPENROUTER_API_KEY"),
]
def fake_select():
pass # no-op — config already has provider set
def fake_prompt_choice(question, choices, default=0):
if "rotation strategy" in question:
return 0
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
return default
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
_stub_tts(monkeypatch)
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("agent.credential_pool.load_pool", lambda provider: _Pool())
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
out = capsys.readouterr().out
assert "Current pooled credentials for openrouter: 3 (2 manual, 1 auto-detected from env/shared auth)" in out
def test_setup_copilot_acp_skips_same_provider_pool_step(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
config = load_config()
def fake_prompt_choice(question, choices, default=0):
if question == "Select your inference provider:":
return 15 # GitHub Copilot ACP
if question == "Select default model:":
return 0
if question == "Configure vision:":
return len(choices) - 1
tts_idx = _maybe_keep_current_tts(question, choices)
if tts_idx is not None:
return tts_idx
raise AssertionError(f"Unexpected prompt_choice call: {question}")
def fake_prompt_yes_no(question, default=True):
if question == "Add another credential for same-provider fallback?":
raise AssertionError("same-provider pool prompt should not appear for copilot-acp")
return False
monkeypatch.setattr("hermes_cli.setup.prompt_choice", fake_prompt_choice)
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", fake_prompt_yes_no)
monkeypatch.setattr("hermes_cli.setup.prompt", lambda *args, **kwargs: "")
monkeypatch.setattr("hermes_cli.setup.prompt_yes_no", lambda *args, **kwargs: False)
monkeypatch.setattr("hermes_cli.auth.get_active_provider", lambda: None)
monkeypatch.setattr("hermes_cli.auth.detect_external_credentials", lambda: [])
monkeypatch.setattr("hermes_cli.auth._login_openai_codex", lambda *args, **kwargs: None)
monkeypatch.setattr(
"hermes_cli.auth.resolve_codex_runtime_credentials",
lambda *args, **kwargs: {
"base_url": "https://chatgpt.com/backend-api/codex",
"api_key": "codex-...oken",
},
)
monkeypatch.setattr(
"hermes_cli.codex_models.get_codex_model_ids",
lambda **kwargs: ["openai/gpt-5.3-codex", "openai/gpt-5-codex-mini"],
)
monkeypatch.setattr("agent.auxiliary_client.get_available_vision_backends", lambda: [])
setup_model_provider(config)
assert config.get("credential_pool_strategies", {}) == {}
def test_setup_copilot_uses_gh_auth_and_saves_provider(tmp_path, monkeypatch):
"""Copilot provider saves correctly through delegation."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
def fake_select():
_write_model_config("copilot", "https://models.github.ai/inference/v1", "gpt-4o")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
env = _read_env(tmp_path)
reloaded = load_config()
assert env.get("OPENAI_BASE_URL") == ""
assert env.get("OPENAI_API_KEY") == ""
assert reloaded["model"]["provider"] == "openai-codex"
assert reloaded["model"]["default"] == "openai/gpt-5.3-codex"
assert reloaded["model"]["base_url"] == "https://chatgpt.com/backend-api/codex"
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "copilot"
def test_setup_summary_marks_codex_auth_as_vision_available(tmp_path, monkeypatch, capsys):
def test_setup_copilot_acp_uses_model_picker_and_saves_provider(tmp_path, monkeypatch):
"""Copilot ACP provider saves correctly through delegation."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
(tmp_path / "auth.json").write_text(
'{"active_provider":"openai-codex","providers":{"openai-codex":{"tokens":{"access_token": "***", "refresh_token": "***"}}}}'
)
config = load_config()
monkeypatch.setattr("shutil.which", lambda _name: None)
def fake_select():
_write_model_config("copilot-acp", "", "claude-sonnet-4")
_print_setup_summary(load_config(), tmp_path)
output = capsys.readouterr().out
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
assert "Vision (image analysis)" in output
assert "missing run 'hermes setup' to configure" not in output
assert "Mixture of Agents" in output
assert "missing OPENROUTER_API_KEY" in output
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "copilot-acp"
def test_setup_switch_custom_to_codex_clears_custom_endpoint_and_updates_config(
tmp_path, monkeypatch
):
"""Switching from custom to codex updates config correctly."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
# Start with custom
_write_model_config("custom", "http://localhost:11434/v1", "qwen3.5:32b")
config = load_config()
assert config["model"]["provider"] == "custom"
def fake_select():
_write_model_config("openai-codex", "https://api.openai.com/v1", "gpt-4o")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert isinstance(reloaded["model"], dict)
assert reloaded["model"]["provider"] == "openai-codex"
assert reloaded["model"]["default"] == "gpt-4o"
def test_setup_switch_preserves_non_model_config(tmp_path, monkeypatch):
"""Provider switch preserves other config sections (terminal, display, etc.)."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_clear_provider_env(monkeypatch)
_stub_tts(monkeypatch)
config = load_config()
config["terminal"]["timeout"] = 999
save_config(config)
config = load_config()
def fake_select():
_write_model_config("openrouter", model_name="gpt-4o")
monkeypatch.setattr("hermes_cli.main.select_provider_and_model", fake_select)
setup_model_provider(config)
save_config(config)
reloaded = load_config()
assert reloaded["terminal"]["timeout"] == 999
assert reloaded["model"]["provider"] == "openrouter"
def test_setup_summary_marks_anthropic_auth_as_vision_available(tmp_path, monkeypatch, capsys):

View file

@ -25,6 +25,8 @@ def _make_run_side_effect(
verify_ok=True,
commit_count="3",
systemd_active=False,
system_service_active=False,
system_restart_rc=0,
launchctl_loaded=False,
):
"""Build a subprocess.run side_effect that simulates git + service commands."""
@ -45,14 +47,23 @@ def _make_run_side_effect(
if "rev-list" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout=f"{commit_count}\n", stderr="")
# systemctl --user is-active
# systemctl is-active — distinguish --user from system scope
if "systemctl" in joined and "is-active" in joined:
if systemd_active:
return subprocess.CompletedProcess(cmd, 0, stdout="active\n", stderr="")
return subprocess.CompletedProcess(cmd, 3, stdout="inactive\n", stderr="")
if "--user" in joined:
if systemd_active:
return subprocess.CompletedProcess(cmd, 0, stdout="active\n", stderr="")
return subprocess.CompletedProcess(cmd, 3, stdout="inactive\n", stderr="")
else:
# System-level check (no --user)
if system_service_active:
return subprocess.CompletedProcess(cmd, 0, stdout="active\n", stderr="")
return subprocess.CompletedProcess(cmd, 3, stdout="inactive\n", stderr="")
# systemctl --user restart
# systemctl restart — distinguish --user from system scope
if "systemctl" in joined and "restart" in joined:
if "--user" not in joined and system_service_active:
stderr = "" if system_restart_rc == 0 else "Failed to restart: Permission denied"
return subprocess.CompletedProcess(cmd, system_restart_rc, stdout="", stderr=stderr)
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
# launchctl list ai.hermes.gateway
@ -393,3 +404,91 @@ class TestCmdUpdateLaunchdRestart:
assert "Stopped gateway" not in captured
assert "Gateway restarted" not in captured
assert "Gateway restarted via launchd" not in captured
# ---------------------------------------------------------------------------
# cmd_update — system-level systemd service detection
# ---------------------------------------------------------------------------
class TestCmdUpdateSystemService:
"""cmd_update detects system-level gateway services where --user fails."""
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_update_detects_system_service_and_restarts(
self, mock_run, _mock_which, mock_args, capsys, monkeypatch,
):
"""When user systemd is inactive but a system service exists, restart via system scope."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
mock_run.side_effect = _make_run_side_effect(
commit_count="3",
systemd_active=False,
system_service_active=True,
)
with patch("gateway.status.get_running_pid", return_value=12345), \
patch("gateway.status.remove_pid_file"):
cmd_update(mock_args)
captured = capsys.readouterr().out
assert "system gateway service" in captured.lower()
assert "Gateway restarted (system service)" in captured
# Verify systemctl restart (no --user) was called
restart_calls = [
c for c in mock_run.call_args_list
if "restart" in " ".join(str(a) for a in c.args[0])
and "systemctl" in " ".join(str(a) for a in c.args[0])
and "--user" not in " ".join(str(a) for a in c.args[0])
]
assert len(restart_calls) == 1
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_update_system_service_restart_failure_shows_sudo_hint(
self, mock_run, _mock_which, mock_args, capsys, monkeypatch,
):
"""When system service restart fails (e.g. no root), show sudo hint."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
mock_run.side_effect = _make_run_side_effect(
commit_count="3",
systemd_active=False,
system_service_active=True,
system_restart_rc=1,
)
with patch("gateway.status.get_running_pid", return_value=12345), \
patch("gateway.status.remove_pid_file"):
cmd_update(mock_args)
captured = capsys.readouterr().out
assert "sudo systemctl restart" in captured
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_user_service_takes_priority_over_system(
self, mock_run, _mock_which, mock_args, capsys, monkeypatch,
):
"""When both user and system services are active, user wins."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "is_linux", lambda: True)
mock_run.side_effect = _make_run_side_effect(
commit_count="3",
systemd_active=True,
system_service_active=True,
)
with patch("gateway.status.get_running_pid", return_value=12345), \
patch("gateway.status.remove_pid_file"), \
patch("os.kill"):
cmd_update(mock_args)
captured = capsys.readouterr().out
# Should restart via user service, not system
assert "Gateway restarted." in captured
assert "(system service)" not in captured