mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-26 17:38:36 +00:00
244dabbd9c
Phase 2c found 3 tests that called _run_and_exit_oneshot without mocking _cleanup_oneshot_runtime, causing real cleanup (terminal, browser, MCP, auxiliary) to run in the pytest worker. Add the mock to all three for test isolation. Also remove redundant 'import logging' inside _exit_after_oneshot (already imported at module level, line 729).
1912 lines
58 KiB
Python
1912 lines
58 KiB
Python
from argparse import Namespace
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import types
|
|
|
|
import pytest
|
|
|
|
|
|
def _args(**overrides):
|
|
base = {
|
|
"continue_last": None,
|
|
"model": None,
|
|
"provider": None,
|
|
"resume": None,
|
|
"toolsets": None,
|
|
"tui": True,
|
|
"tui_dev": False,
|
|
}
|
|
base.update(overrides)
|
|
return Namespace(**base)
|
|
|
|
|
|
def _raise_exit(rc):
|
|
raise SystemExit(rc)
|
|
|
|
|
|
@pytest.fixture
|
|
def main_mod(monkeypatch):
|
|
import hermes_cli.main as mod
|
|
|
|
monkeypatch.setattr(mod, "_has_any_provider_configured", lambda: True)
|
|
# Reset the idempotency guard so each test starts fresh.
|
|
monkeypatch.setattr(mod, "_oneshot_cleanup_done", False)
|
|
return mod
|
|
|
|
|
|
def test_cmd_chat_tui_continue_uses_latest_tui_session(monkeypatch, main_mod):
|
|
calls = []
|
|
captured = {}
|
|
|
|
def fake_resolve_last(source="cli"):
|
|
calls.append(source)
|
|
return "20260408_235959_a1b2c3" if source == "tui" else None
|
|
|
|
def fake_launch(
|
|
resume_session_id=None,
|
|
tui_dev=False,
|
|
model=None,
|
|
provider=None,
|
|
toolsets=None,
|
|
**kwargs,
|
|
):
|
|
captured["resume"] = resume_session_id
|
|
raise SystemExit(0)
|
|
|
|
monkeypatch.setattr(main_mod, "_resolve_last_session", fake_resolve_last)
|
|
monkeypatch.setattr(main_mod, "_resolve_session_by_name_or_id", lambda val: val)
|
|
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod.cmd_chat(_args(continue_last=True))
|
|
|
|
assert calls == ["tui"]
|
|
assert captured["resume"] == "20260408_235959_a1b2c3"
|
|
|
|
|
|
def test_cmd_chat_tui_continue_falls_back_to_latest_cli_session(monkeypatch, main_mod):
|
|
calls = []
|
|
captured = {}
|
|
|
|
def fake_resolve_last(source="cli"):
|
|
calls.append(source)
|
|
if source == "tui":
|
|
return None
|
|
if source == "cli":
|
|
return "20260408_235959_d4e5f6"
|
|
return None
|
|
|
|
def fake_launch(
|
|
resume_session_id=None,
|
|
tui_dev=False,
|
|
model=None,
|
|
provider=None,
|
|
toolsets=None,
|
|
**kwargs,
|
|
):
|
|
captured["resume"] = resume_session_id
|
|
raise SystemExit(0)
|
|
|
|
monkeypatch.setattr(main_mod, "_resolve_last_session", fake_resolve_last)
|
|
monkeypatch.setattr(main_mod, "_resolve_session_by_name_or_id", lambda val: val)
|
|
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod.cmd_chat(_args(continue_last=True))
|
|
|
|
assert calls == ["tui", "cli"]
|
|
assert captured["resume"] == "20260408_235959_d4e5f6"
|
|
|
|
|
|
def test_cmd_chat_tui_resume_resolves_title_before_launch(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
def fake_launch(
|
|
resume_session_id=None,
|
|
tui_dev=False,
|
|
model=None,
|
|
provider=None,
|
|
toolsets=None,
|
|
**kwargs,
|
|
):
|
|
captured["resume"] = resume_session_id
|
|
raise SystemExit(0)
|
|
|
|
monkeypatch.setattr(
|
|
main_mod, "_resolve_session_by_name_or_id", lambda val: "20260409_000000_aa11bb"
|
|
)
|
|
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod.cmd_chat(_args(resume="my t0p session"))
|
|
|
|
assert captured["resume"] == "20260409_000000_aa11bb"
|
|
|
|
|
|
def test_cmd_chat_tui_passes_model_and_provider(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
def fake_launch(
|
|
resume_session_id=None,
|
|
tui_dev=False,
|
|
model=None,
|
|
provider=None,
|
|
toolsets=None,
|
|
**kwargs,
|
|
):
|
|
captured.update(
|
|
{
|
|
"model": model,
|
|
"provider": provider,
|
|
"resume": resume_session_id,
|
|
"toolsets": toolsets,
|
|
"tui_dev": tui_dev,
|
|
}
|
|
)
|
|
raise SystemExit(0)
|
|
|
|
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod.cmd_chat(
|
|
_args(model="anthropic/claude-sonnet-4.6", provider="anthropic")
|
|
)
|
|
|
|
assert captured == {
|
|
"model": "anthropic/claude-sonnet-4.6",
|
|
"provider": "anthropic",
|
|
"resume": None,
|
|
"toolsets": None,
|
|
"tui_dev": False,
|
|
}
|
|
|
|
|
|
def test_cmd_chat_tui_passes_toolsets(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
def fake_launch(
|
|
resume_session_id=None,
|
|
tui_dev=False,
|
|
model=None,
|
|
provider=None,
|
|
toolsets=None,
|
|
**kwargs,
|
|
):
|
|
captured["toolsets"] = toolsets
|
|
raise SystemExit(0)
|
|
|
|
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod.cmd_chat(_args(toolsets="web,terminal"))
|
|
|
|
assert captured["toolsets"] == "web,terminal"
|
|
|
|
|
|
def test_cmd_chat_tui_forwards_chat_flags(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
def fake_launch(resume_session_id=None, **kwargs):
|
|
captured["resume_session_id"] = resume_session_id
|
|
captured.update(kwargs)
|
|
raise SystemExit(0)
|
|
|
|
monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod.cmd_chat(
|
|
_args(
|
|
skills=["foo,bar"],
|
|
verbose=True,
|
|
quiet=True,
|
|
query="hello",
|
|
image="/tmp/cat.png",
|
|
worktree=True,
|
|
checkpoints=True,
|
|
pass_session_id=True,
|
|
max_turns=7,
|
|
accept_hooks=True,
|
|
)
|
|
)
|
|
|
|
assert captured["skills"] == ["foo,bar"]
|
|
assert captured["verbose"] is True
|
|
assert captured["quiet"] is True
|
|
assert captured["query"] == "hello"
|
|
assert captured["image"] == "/tmp/cat.png"
|
|
assert captured["worktree"] is True
|
|
assert captured["checkpoints"] is True
|
|
assert captured["pass_session_id"] is True
|
|
assert captured["max_turns"] == 7
|
|
assert captured["accept_hooks"] is True
|
|
|
|
|
|
def test_main_top_level_tui_accepts_toolsets(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
import hermes_cli.config as config_mod
|
|
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "--tui", "--toolsets", "web,terminal"])
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.plugins",
|
|
types.SimpleNamespace(discover_plugins=lambda: None),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.mcp_tool",
|
|
types.SimpleNamespace(discover_mcp_tools=lambda: None),
|
|
)
|
|
monkeypatch.setattr(config_mod, "load_config", lambda: {})
|
|
monkeypatch.setattr(config_mod, "get_container_exec_info", lambda: None)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"agent.shell_hooks",
|
|
types.SimpleNamespace(
|
|
register_from_config=lambda _cfg, accept_hooks=False: None
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"cmd_chat",
|
|
lambda args: captured.update({"toolsets": args.toolsets, "tui": args.tui}),
|
|
)
|
|
|
|
main_mod.main()
|
|
|
|
assert captured == {"toolsets": "web,terminal", "tui": True}
|
|
|
|
|
|
def test_termux_fast_tui_launch_uses_light_parser(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.setattr(
|
|
sys, "argv", ["hermes", "--tui", "--toolsets", "web,terminal"]
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"cmd_chat",
|
|
lambda args: captured.update({"toolsets": args.toolsets, "tui": args.tui}),
|
|
)
|
|
|
|
assert main_mod._try_termux_fast_tui_launch() is True
|
|
assert captured == {"toolsets": "web,terminal", "tui": True}
|
|
|
|
|
|
def test_termux_fast_tui_launch_skips_help(monkeypatch, main_mod):
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "--tui", "--help"])
|
|
|
|
assert main_mod._try_termux_fast_tui_launch() is False
|
|
|
|
|
|
def test_fast_tui_launch_is_termux_only(monkeypatch, main_mod):
|
|
monkeypatch.delenv("TERMUX_VERSION", raising=False)
|
|
monkeypatch.setenv("PREFIX", "/usr")
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "--tui"])
|
|
|
|
assert main_mod._try_termux_fast_tui_launch() is False
|
|
|
|
|
|
def test_termux_fast_cli_launch_chat_uses_light_parser(monkeypatch, main_mod):
|
|
captured = {}
|
|
prepared = []
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.delenv("HERMES_TUI", raising=False)
|
|
monkeypatch.setattr(
|
|
sys, "argv", ["hermes", "chat", "-q", "hello", "--toolsets", "web,terminal"]
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod, "_prepare_agent_startup", lambda args: prepared.append(args.command)
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"cmd_chat",
|
|
lambda args: captured.update(
|
|
{"query": args.query, "toolsets": args.toolsets, "command": args.command}
|
|
),
|
|
)
|
|
|
|
assert main_mod._try_termux_fast_cli_launch() is True
|
|
assert prepared == ["chat"]
|
|
assert captured == {
|
|
"query": "hello",
|
|
"toolsets": "web,terminal",
|
|
"command": "chat",
|
|
}
|
|
|
|
|
|
def test_termux_fast_cli_launch_bare_defers_agent_startup(monkeypatch, main_mod):
|
|
captured = {}
|
|
prepared = []
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.delenv("HERMES_TUI", raising=False)
|
|
monkeypatch.delenv("HERMES_DEFER_AGENT_STARTUP", raising=False)
|
|
monkeypatch.delenv("HERMES_FAST_STARTUP_BANNER", raising=False)
|
|
monkeypatch.setattr(sys, "argv", ["hermes"])
|
|
monkeypatch.setattr(
|
|
main_mod, "_prepare_agent_startup", lambda args: prepared.append(args.command)
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"cmd_chat",
|
|
lambda args: captured.update(
|
|
{
|
|
"query": args.query,
|
|
"command": args.command,
|
|
"compact": getattr(args, "compact", False),
|
|
}
|
|
),
|
|
)
|
|
|
|
assert main_mod._try_termux_fast_cli_launch() is True
|
|
assert prepared == []
|
|
assert captured == {"query": None, "command": None, "compact": True}
|
|
assert os.environ["HERMES_DEFER_AGENT_STARTUP"] == "1"
|
|
assert os.environ["HERMES_FAST_STARTUP_BANNER"] == "1"
|
|
|
|
|
|
def test_termux_fast_cli_launch_oneshot_uses_light_parser(monkeypatch, main_mod):
|
|
captured = {}
|
|
prepared = []
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.delenv("HERMES_TUI", raising=False)
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
"hermes",
|
|
"-z",
|
|
"hello",
|
|
"--model",
|
|
"gpt-test",
|
|
"--provider",
|
|
"openai",
|
|
"--usage-file",
|
|
"usage.json",
|
|
],
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod, "_prepare_agent_startup", lambda args: prepared.append(args.command)
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(
|
|
run_oneshot=lambda prompt, **kwargs: captured.update(
|
|
{"prompt": prompt, **kwargs}
|
|
)
|
|
or 17
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_exit_after_oneshot",
|
|
_raise_exit,
|
|
)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._try_termux_fast_cli_launch()
|
|
|
|
assert exc.value.code == 17
|
|
assert prepared == [None]
|
|
assert captured == {
|
|
"prompt": "hello",
|
|
"model": "gpt-test",
|
|
"provider": "openai",
|
|
"toolsets": None,
|
|
"usage_file": "usage.json",
|
|
}
|
|
|
|
|
|
def test_termux_fast_cli_launch_version_skips_update_check(monkeypatch, main_mod):
|
|
captured = []
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.delenv("HERMES_TUI", raising=False)
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "version"])
|
|
monkeypatch.setattr(
|
|
main_mod, "_print_version_info", lambda *, check_updates: captured.append(check_updates)
|
|
)
|
|
|
|
assert main_mod._try_termux_fast_cli_launch() is True
|
|
assert captured == [False]
|
|
|
|
|
|
def test_termux_ultrafast_version_runs_before_heavy_startup(
|
|
monkeypatch, capsys, main_mod
|
|
):
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.delenv("HERMES_TERMUX_DISABLE_FAST_CLI", raising=False)
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "--version"])
|
|
|
|
assert main_mod._try_termux_ultrafast_version() is True
|
|
|
|
out = capsys.readouterr().out
|
|
assert "Hermes Agent v" in out
|
|
assert "Install directory:" in out
|
|
assert "Python:" in out
|
|
assert "OpenAI SDK:" in out
|
|
|
|
|
|
def test_read_openai_version_fast(monkeypatch, tmp_path, main_mod):
|
|
package_dir = tmp_path / "openai"
|
|
package_dir.mkdir()
|
|
(package_dir / "_version.py").write_text(
|
|
'__version__ = "9.8.7" # x-release-please-version\n',
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(sys, "path", [str(tmp_path)])
|
|
|
|
assert main_mod._read_openai_version_fast() == "9.8.7"
|
|
|
|
|
|
def test_termux_fast_cli_launch_skips_help(monkeypatch, main_mod):
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.delenv("HERMES_TUI", raising=False)
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "chat", "--help"])
|
|
|
|
assert main_mod._try_termux_fast_cli_launch() is False
|
|
|
|
|
|
def test_termux_fast_cli_launch_can_be_disabled(monkeypatch, main_mod):
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.setenv("HERMES_TERMUX_DISABLE_FAST_CLI", "1")
|
|
monkeypatch.delenv("HERMES_TUI", raising=False)
|
|
monkeypatch.setattr(sys, "argv", ["hermes", "version"])
|
|
|
|
assert main_mod._try_termux_fast_cli_launch() is False
|
|
|
|
|
|
def test_termux_bundled_skills_stamp_controls_sync(monkeypatch, tmp_path, main_mod):
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.setattr(main_mod, "get_hermes_home", lambda: tmp_path)
|
|
monkeypatch.setattr(main_mod, "_termux_bundled_skills_fingerprint", lambda: "fp1")
|
|
|
|
assert main_mod._termux_bundled_skills_sync_needed() is True
|
|
main_mod._mark_termux_bundled_skills_synced()
|
|
assert main_mod._termux_bundled_skills_sync_needed() is False
|
|
|
|
monkeypatch.setenv("HERMES_TERMUX_FORCE_SKILLS_SYNC", "1")
|
|
assert main_mod._termux_bundled_skills_sync_needed() is True
|
|
|
|
|
|
def test_termux_skips_bundled_skill_sync_when_stamp_fresh(monkeypatch, tmp_path, main_mod):
|
|
calls = []
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.setattr(main_mod, "get_hermes_home", lambda: tmp_path)
|
|
monkeypatch.setattr(main_mod, "_termux_bundled_skills_fingerprint", lambda: "fp1")
|
|
main_mod._mark_termux_bundled_skills_synced()
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.skills_sync",
|
|
types.SimpleNamespace(sync_skills=lambda quiet: calls.append(quiet)),
|
|
)
|
|
|
|
assert main_mod._sync_bundled_skills_for_startup() is False
|
|
assert calls == []
|
|
|
|
|
|
def test_termux_forced_bundled_skill_sync_runs(monkeypatch, tmp_path, main_mod):
|
|
calls = []
|
|
|
|
monkeypatch.setenv("TERMUX_VERSION", "1")
|
|
monkeypatch.setenv("HERMES_TERMUX_FORCE_SKILLS_SYNC", "1")
|
|
monkeypatch.setattr(main_mod, "get_hermes_home", lambda: tmp_path)
|
|
monkeypatch.setattr(main_mod, "_termux_bundled_skills_fingerprint", lambda: "fp1")
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.skills_sync",
|
|
types.SimpleNamespace(sync_skills=lambda quiet: calls.append(quiet)),
|
|
)
|
|
|
|
assert main_mod._sync_bundled_skills_for_startup() is True
|
|
assert calls == [True]
|
|
|
|
|
|
def test_read_git_revision_fingerprint_resolves_packed_refs(tmp_path, main_mod):
|
|
repo = tmp_path / "repo"
|
|
git_dir = repo / ".git"
|
|
git_dir.mkdir(parents=True)
|
|
(git_dir / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
|
|
packed_sha = "1234567890abcdef1234567890abcdef12345678"
|
|
(git_dir / "packed-refs").write_text(
|
|
"# pack-refs with: peeled fully-peeled sorted\n"
|
|
f"{packed_sha} refs/heads/main\n"
|
|
"abcdef0000000000000000000000000000000000 refs/tags/v1.0\n"
|
|
"^99999999aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
fingerprint = main_mod._read_git_revision_fingerprint(repo)
|
|
|
|
assert fingerprint == f"git:refs/heads/main:{packed_sha}"
|
|
|
|
|
|
def test_read_git_revision_fingerprint_packed_refs_in_worktree_common_dir(
|
|
tmp_path, main_mod
|
|
):
|
|
main_repo = tmp_path / "repo"
|
|
common_git = main_repo / ".git"
|
|
common_git.mkdir(parents=True)
|
|
packed_sha = "fedcba9876543210fedcba9876543210fedcba98"
|
|
(common_git / "packed-refs").write_text(
|
|
f"{packed_sha} refs/heads/main\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
worktree = tmp_path / "wt"
|
|
worktree.mkdir()
|
|
wt_gitdir = common_git / "worktrees" / "wt"
|
|
wt_gitdir.mkdir(parents=True)
|
|
(wt_gitdir / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
|
|
(wt_gitdir / "commondir").write_text("../..\n", encoding="utf-8")
|
|
(worktree / ".git").write_text(f"gitdir: {wt_gitdir}\n", encoding="utf-8")
|
|
|
|
fingerprint = main_mod._read_git_revision_fingerprint(worktree)
|
|
|
|
assert fingerprint == f"git:refs/heads/main:{packed_sha}"
|
|
|
|
|
|
def test_read_git_revision_fingerprint_loose_ref_in_worktree_common_dir(
|
|
tmp_path, main_mod
|
|
):
|
|
"""`git worktree add -b NAME` writes the new branch ref to the common dir,
|
|
not the per-worktree gitdir. The fingerprint must still resolve it."""
|
|
main_repo = tmp_path / "repo"
|
|
common_git = main_repo / ".git"
|
|
common_git.mkdir(parents=True)
|
|
loose_sha = "0123456789abcdef0123456789abcdef01234567"
|
|
(common_git / "refs" / "heads").mkdir(parents=True)
|
|
(common_git / "refs" / "heads" / "feature").write_text(
|
|
loose_sha + "\n", encoding="utf-8"
|
|
)
|
|
|
|
worktree = tmp_path / "wt"
|
|
worktree.mkdir()
|
|
wt_gitdir = common_git / "worktrees" / "wt"
|
|
wt_gitdir.mkdir(parents=True)
|
|
(wt_gitdir / "HEAD").write_text("ref: refs/heads/feature\n", encoding="utf-8")
|
|
(wt_gitdir / "commondir").write_text("../..\n", encoding="utf-8")
|
|
(worktree / ".git").write_text(f"gitdir: {wt_gitdir}\n", encoding="utf-8")
|
|
|
|
fingerprint = main_mod._read_git_revision_fingerprint(worktree)
|
|
|
|
assert fingerprint == f"git:refs/heads/feature:{loose_sha}"
|
|
|
|
|
|
def test_read_git_revision_fingerprint_unresolved_ref_is_stable(tmp_path, main_mod):
|
|
repo = tmp_path / "repo"
|
|
git_dir = repo / ".git"
|
|
git_dir.mkdir(parents=True)
|
|
(git_dir / "HEAD").write_text("ref: refs/heads/missing\n", encoding="utf-8")
|
|
|
|
fingerprint = main_mod._read_git_revision_fingerprint(repo)
|
|
|
|
assert fingerprint == "git:refs/heads/missing:unresolved"
|
|
|
|
|
|
def test_main_top_level_oneshot_accepts_toolsets(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
import hermes_cli.config as config_mod
|
|
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
"hermes",
|
|
"-z",
|
|
"hello",
|
|
"--toolsets",
|
|
"web,terminal",
|
|
"--usage-file",
|
|
"usage.json",
|
|
],
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.plugins",
|
|
types.SimpleNamespace(discover_plugins=lambda: None),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.mcp_tool",
|
|
types.SimpleNamespace(discover_mcp_tools=lambda: None),
|
|
)
|
|
monkeypatch.setattr(config_mod, "load_config", lambda: {})
|
|
monkeypatch.setattr(config_mod, "get_container_exec_info", lambda: None)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"agent.shell_hooks",
|
|
types.SimpleNamespace(
|
|
register_from_config=lambda _cfg, accept_hooks=False: None
|
|
),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(
|
|
run_oneshot=lambda prompt, **kwargs: captured.update(
|
|
{"prompt": prompt, **kwargs}
|
|
)
|
|
or 0
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_exit_after_oneshot",
|
|
_raise_exit,
|
|
)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod.main()
|
|
|
|
assert exc.value.code == 0
|
|
assert captured == {
|
|
"prompt": "hello",
|
|
"model": None,
|
|
"provider": None,
|
|
"toolsets": "web,terminal",
|
|
"usage_file": "usage.json",
|
|
}
|
|
|
|
|
|
def test_exit_after_oneshot_flushes_stdio_and_calls_os_exit(
|
|
monkeypatch, main_mod
|
|
):
|
|
flushed = []
|
|
exits = []
|
|
|
|
class FakeStream:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
|
|
def flush(self):
|
|
flushed.append(self.name)
|
|
|
|
def fake_exit(rc):
|
|
exits.append(rc)
|
|
raise SystemExit(rc)
|
|
|
|
monkeypatch.setattr(main_mod.sys, "stdout", FakeStream("stdout"))
|
|
monkeypatch.setattr(main_mod.sys, "stderr", FakeStream("stderr"))
|
|
monkeypatch.setattr(main_mod.os, "_exit", fake_exit)
|
|
monkeypatch.setattr("logging.shutdown", lambda: None)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._exit_after_oneshot(17)
|
|
|
|
assert exc.value.code == 17
|
|
assert exits == [17]
|
|
assert flushed == ["stdout", "stderr"]
|
|
|
|
|
|
def test_exit_after_oneshot_invokes_logging_shutdown_in_order(
|
|
monkeypatch, main_mod
|
|
):
|
|
events = []
|
|
|
|
class FakeStream:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
|
|
def flush(self):
|
|
events.append(f"flush:{self.name}")
|
|
|
|
def fake_exit(rc):
|
|
events.append(f"exit:{rc}")
|
|
raise SystemExit(rc)
|
|
|
|
monkeypatch.setattr(main_mod.sys, "stdout", FakeStream("stdout"))
|
|
monkeypatch.setattr(main_mod.sys, "stderr", FakeStream("stderr"))
|
|
monkeypatch.setattr("logging.shutdown", lambda: events.append("shutdown"))
|
|
monkeypatch.setattr(main_mod.os, "_exit", fake_exit)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._exit_after_oneshot(0)
|
|
|
|
assert exc.value.code == 0
|
|
assert events == ["flush:stdout", "flush:stderr", "shutdown", "exit:0"]
|
|
|
|
|
|
def test_exit_after_oneshot_exits_even_if_logging_shutdown_raises(
|
|
monkeypatch, main_mod
|
|
):
|
|
exits = []
|
|
|
|
def fake_exit(rc):
|
|
exits.append(rc)
|
|
raise SystemExit(rc)
|
|
|
|
monkeypatch.setattr(
|
|
main_mod.sys, "stdout", types.SimpleNamespace(flush=lambda: None)
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod.sys, "stderr", types.SimpleNamespace(flush=lambda: None)
|
|
)
|
|
monkeypatch.setattr(
|
|
"logging.shutdown",
|
|
lambda: (_ for _ in ()).throw(RuntimeError("shutdown failed")),
|
|
)
|
|
monkeypatch.setattr(main_mod.os, "_exit", fake_exit)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._exit_after_oneshot(1)
|
|
|
|
assert exc.value.code == 1
|
|
assert exits == [1]
|
|
|
|
|
|
def test_exit_after_oneshot_flushes_stderr_when_stdout_flush_fails(
|
|
monkeypatch, main_mod
|
|
):
|
|
flushed = []
|
|
exits = []
|
|
|
|
class BadStdout:
|
|
def flush(self):
|
|
raise BrokenPipeError("pipe closed")
|
|
|
|
class FakeStderr:
|
|
def flush(self):
|
|
flushed.append("stderr")
|
|
|
|
def fake_exit(rc):
|
|
exits.append(rc)
|
|
raise SystemExit(rc)
|
|
|
|
monkeypatch.setattr(main_mod.sys, "stdout", BadStdout())
|
|
monkeypatch.setattr(main_mod.sys, "stderr", FakeStderr())
|
|
monkeypatch.setattr(main_mod.os, "_exit", fake_exit)
|
|
monkeypatch.setattr("logging.shutdown", lambda: None)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._exit_after_oneshot(2)
|
|
|
|
assert exc.value.code == 2
|
|
assert exits == [2]
|
|
assert flushed == ["stderr"]
|
|
|
|
|
|
def test_exit_after_oneshot_normalizes_non_int_exit_code(monkeypatch, main_mod):
|
|
exits = []
|
|
|
|
def fake_exit(rc):
|
|
exits.append(rc)
|
|
raise SystemExit(rc)
|
|
|
|
monkeypatch.setattr(main_mod.os, "_exit", fake_exit)
|
|
monkeypatch.setattr("logging.shutdown", lambda: None)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._exit_after_oneshot(None)
|
|
|
|
assert exc.value.code == 0
|
|
assert exits == [0]
|
|
|
|
|
|
def test_run_and_exit_oneshot_routes_system_exit_to_hard_exit(monkeypatch, main_mod):
|
|
exits = []
|
|
|
|
def fake_run_oneshot(*_args, **_kwargs):
|
|
raise SystemExit(2)
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=fake_run_oneshot),
|
|
)
|
|
monkeypatch.setattr(main_mod, "_cleanup_oneshot_runtime", lambda: None)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exits == [2]
|
|
|
|
|
|
def test_run_and_exit_oneshot_routes_bare_system_exit_to_zero(monkeypatch, main_mod):
|
|
exits = []
|
|
|
|
def fake_run_oneshot(*_args, **_kwargs):
|
|
raise SystemExit
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=fake_run_oneshot),
|
|
)
|
|
monkeypatch.setattr(main_mod, "_cleanup_oneshot_runtime", lambda: None)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exits == [None]
|
|
|
|
|
|
def test_run_and_exit_oneshot_prints_system_exit_message(
|
|
monkeypatch, capsys, main_mod
|
|
):
|
|
exits = []
|
|
|
|
def fake_run_oneshot(*_args, **_kwargs):
|
|
raise SystemExit("fatal")
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=fake_run_oneshot),
|
|
)
|
|
monkeypatch.setattr(main_mod, "_cleanup_oneshot_runtime", lambda: None)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exits == [1]
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
assert captured.err == "fatal\n"
|
|
|
|
|
|
def test_run_and_exit_oneshot_cleans_global_runtime_before_hard_exit(
|
|
monkeypatch, main_mod
|
|
):
|
|
events = []
|
|
|
|
def _mod(name, **attrs):
|
|
fake = types.ModuleType(name)
|
|
for key, value in attrs.items():
|
|
setattr(fake, key, value)
|
|
return fake
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=lambda *_args, **_kwargs: events.append("run") or 0),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.terminal_tool",
|
|
_mod("tools.terminal_tool", cleanup_all_environments=lambda: events.append("terminal")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.async_delegation",
|
|
_mod("tools.async_delegation", interrupt_all=lambda **kw: events.append("delegation")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.browser_tool",
|
|
_mod("tools.browser_tool", _emergency_cleanup_all_sessions=lambda: events.append("browser")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.mcp_tool",
|
|
_mod("tools.mcp_tool", shutdown_mcp_servers=lambda: events.append("mcp")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"agent.auxiliary_client",
|
|
_mod("agent.auxiliary_client", shutdown_cached_clients=lambda: events.append("aux")),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod, "_exit_after_oneshot", lambda rc: events.append(f"exit:{rc}")
|
|
)
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert events == ["run", "terminal", "delegation", "browser", "mcp", "aux", "exit:0"]
|
|
|
|
|
|
def test_run_and_exit_oneshot_still_exits_when_global_cleanup_raises(
|
|
monkeypatch, main_mod
|
|
):
|
|
events = []
|
|
|
|
def _raise_mcp():
|
|
raise RuntimeError("mcp boom")
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=lambda *_args, **_kwargs: 0),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.terminal_tool",
|
|
types.SimpleNamespace(cleanup_all_environments=lambda: events.append("terminal")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.async_delegation",
|
|
types.SimpleNamespace(interrupt_all=lambda **kw: events.append("delegation")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.browser_tool",
|
|
types.SimpleNamespace(_emergency_cleanup_all_sessions=lambda: events.append("browser")),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"tools.mcp_tool",
|
|
types.SimpleNamespace(shutdown_mcp_servers=_raise_mcp),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"agent.auxiliary_client",
|
|
types.SimpleNamespace(shutdown_cached_clients=lambda: events.append("aux")),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod, "_exit_after_oneshot", lambda rc: events.append(f"exit:{rc}")
|
|
)
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert events == ["terminal", "delegation", "browser", "aux", "exit:0"]
|
|
|
|
|
|
def test_run_and_exit_oneshot_hard_exits_when_cleanup_is_interrupted(
|
|
monkeypatch, main_mod
|
|
):
|
|
def _raise_keyboard_interrupt():
|
|
raise KeyboardInterrupt
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=lambda *_args, **_kwargs: 0),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod, "_cleanup_oneshot_runtime", _raise_keyboard_interrupt
|
|
)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", _raise_exit)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exc.value.code == 0
|
|
|
|
|
|
def test_run_and_exit_oneshot_routes_keyboard_interrupt_to_130(
|
|
monkeypatch, main_mod
|
|
):
|
|
exits = []
|
|
|
|
def fake_run_oneshot(*_args, **_kwargs):
|
|
raise KeyboardInterrupt
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=fake_run_oneshot),
|
|
)
|
|
monkeypatch.setattr(main_mod, "_cleanup_oneshot_runtime", lambda: None)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exits == [130]
|
|
|
|
|
|
def test_run_and_exit_oneshot_hard_exits_on_unexpected_exception(
|
|
monkeypatch, main_mod, capsys
|
|
):
|
|
# ``run_oneshot`` is contracted to convert agent failures into an int and
|
|
# only re-raise KeyboardInterrupt / SystemExit. If it ever malfunctions and
|
|
# lets another exception escape, the one-shot path must still hard-exit
|
|
# (rc 1) rather than fall through to interpreter teardown — the exact path
|
|
# that SIGABRTs on AL2023.
|
|
exits = []
|
|
cleaned = []
|
|
|
|
def fake_run_oneshot(*_args, **_kwargs):
|
|
raise RuntimeError("run_oneshot itself blew up")
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=fake_run_oneshot),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod, "_cleanup_oneshot_runtime", lambda: cleaned.append(True)
|
|
)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exits == [1]
|
|
# Global cleanup still runs on the defensive path — resources must not leak
|
|
# just because run_oneshot malfunctioned.
|
|
assert cleaned == [True]
|
|
# The failure is surfaced on stderr, never swallowed silently.
|
|
assert "run_oneshot itself blew up" in capsys.readouterr().err
|
|
|
|
|
|
def test_run_and_exit_oneshot_hard_exits_when_oneshot_import_fails(
|
|
monkeypatch, main_mod, capsys
|
|
):
|
|
import builtins
|
|
|
|
exits = []
|
|
cleaned = []
|
|
real_import = builtins.__import__
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
if name == "hermes_cli.oneshot" and "run_oneshot" in (fromlist or ()):
|
|
raise RuntimeError("oneshot import blew up")
|
|
return real_import(name, globals, locals, fromlist, level)
|
|
|
|
monkeypatch.setattr(builtins, "__import__", fake_import)
|
|
monkeypatch.setattr(
|
|
main_mod, "_cleanup_oneshot_runtime", lambda: cleaned.append(True)
|
|
)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hello")
|
|
|
|
assert exits == [1]
|
|
assert cleaned == [True]
|
|
assert "oneshot import blew up" in capsys.readouterr().err
|
|
|
|
|
|
def test_oneshot_subprocess_exits_without_teardown_abort():
|
|
program = textwrap.dedent(
|
|
"""
|
|
import hermes_cli.oneshot as oneshot
|
|
from hermes_cli.main import _exit_after_oneshot
|
|
|
|
oneshot._run_agent = lambda *args, **kwargs: ("ok", {"final_response": "ok"})
|
|
_exit_after_oneshot(oneshot.run_oneshot("hello"))
|
|
"""
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", program],
|
|
cwd=Path(__file__).resolve().parents[2],
|
|
capture_output=True,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert result.stdout == b"ok\n"
|
|
# Don't demand byte-empty stderr — an import-time warning from the heavy
|
|
# CLI import chain shouldn't fail this. What matters is no crash traceback.
|
|
assert b"Traceback" not in result.stderr
|
|
|
|
|
|
def test_exit_after_oneshot_bypasses_late_atexit_abort():
|
|
program = textwrap.dedent(
|
|
"""
|
|
import atexit
|
|
import os
|
|
import sys
|
|
from hermes_cli.main import _exit_after_oneshot
|
|
|
|
atexit.register(os.abort)
|
|
sys.stdout.write("done\\n")
|
|
_exit_after_oneshot(0)
|
|
"""
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", program],
|
|
cwd=Path(__file__).resolve().parents[2],
|
|
capture_output=True,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert result.stdout == b"done\n"
|
|
|
|
|
|
def test_run_and_exit_oneshot_passes_through_nonzero_return(monkeypatch, main_mod):
|
|
# A non-zero rc from run_oneshot (e.g. provider-without-model → 2, or the
|
|
# empty-response guard → 1) must reach os._exit unchanged.
|
|
exits = []
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.oneshot",
|
|
types.SimpleNamespace(run_oneshot=lambda *a, **k: 2),
|
|
)
|
|
monkeypatch.setattr(main_mod, "_cleanup_oneshot_runtime", lambda: None)
|
|
monkeypatch.setattr(main_mod, "_exit_after_oneshot", lambda rc: exits.append(rc))
|
|
|
|
main_mod._run_and_exit_oneshot("hi")
|
|
|
|
assert exits == [2]
|
|
|
|
|
|
def test_main_oneshot_path_bypasses_late_atexit_abort():
|
|
# End-to-end through the real top-level ``main()`` ``-z`` path: a valid
|
|
# response prints, then a late atexit handler that would abort is bypassed
|
|
# by the hard exit, so the process reports success (#43055).
|
|
program = textwrap.dedent(
|
|
"""
|
|
import atexit
|
|
import os
|
|
import sys
|
|
import types
|
|
|
|
import hermes_cli.main as main_mod
|
|
|
|
sys.argv = ["hermes", "-z", "hello"]
|
|
main_mod._prepare_agent_startup = lambda args: None
|
|
|
|
def _fake_run_oneshot(prompt, **kwargs):
|
|
print("ok")
|
|
return 0
|
|
|
|
sys.modules["hermes_cli.oneshot"] = types.SimpleNamespace(
|
|
run_oneshot=_fake_run_oneshot
|
|
)
|
|
atexit.register(os.abort)
|
|
main_mod.main()
|
|
"""
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", program],
|
|
cwd=Path(__file__).resolve().parents[2],
|
|
capture_output=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert result.stdout == b"ok\n"
|
|
assert b"Traceback" not in result.stderr
|
|
|
|
|
|
def test_oneshot_run_agent_closes_agent_after_chat(monkeypatch):
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
closed = []
|
|
shutdown_messages = []
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **_kwargs):
|
|
self.suppress_status_output = False
|
|
self.stream_delta_callback = object()
|
|
self.tool_gen_callback = object()
|
|
self._session_messages = [{"role": "user", "content": "hello"}]
|
|
|
|
def run_conversation(self, prompt, **_kwargs):
|
|
assert prompt == "hello"
|
|
return {"final_response": "done"}
|
|
|
|
def shutdown_memory_provider(self, messages=None):
|
|
shutdown_messages.append(messages)
|
|
|
|
def close(self):
|
|
closed.append(True)
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules, "run_agent", types.SimpleNamespace(AIAgent=FakeAgent)
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"model": {"default": "gpt-test", "provider": "openai"}},
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
lambda **_kwargs: {
|
|
"api_key": "key",
|
|
"base_url": "https://example.invalid",
|
|
"provider": "openai",
|
|
"api_mode": "chat_completions",
|
|
"credential_pool": None,
|
|
},
|
|
)
|
|
monkeypatch.setattr(oneshot_mod, "_create_session_db_for_oneshot", lambda: None)
|
|
|
|
assert (
|
|
oneshot_mod._run_agent(
|
|
"hello", model="gpt-test", provider="openai", use_config_toolsets=False
|
|
)
|
|
== ("done", {"final_response": "done"})
|
|
)
|
|
assert closed == [True]
|
|
assert shutdown_messages == [[{"role": "user", "content": "hello"}]]
|
|
|
|
|
|
def test_oneshot_run_agent_closes_agent_when_chat_raises(monkeypatch):
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
closed = []
|
|
shutdowns = []
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **_kwargs):
|
|
self.suppress_status_output = False
|
|
self.stream_delta_callback = object()
|
|
self.tool_gen_callback = object()
|
|
|
|
def run_conversation(self, _prompt, **_kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
def shutdown_memory_provider(self, messages=None):
|
|
shutdowns.append(messages)
|
|
|
|
def close(self):
|
|
closed.append(True)
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules, "run_agent", types.SimpleNamespace(AIAgent=FakeAgent)
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"model": {"default": "gpt-test", "provider": "openai"}},
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
lambda **_kwargs: {
|
|
"api_key": "key",
|
|
"base_url": "https://example.invalid",
|
|
"provider": "openai",
|
|
"api_mode": "chat_completions",
|
|
"credential_pool": None,
|
|
},
|
|
)
|
|
monkeypatch.setattr(oneshot_mod, "_create_session_db_for_oneshot", lambda: None)
|
|
|
|
with pytest.raises(RuntimeError, match="boom"):
|
|
oneshot_mod._run_agent(
|
|
"hello", model="gpt-test", provider="openai", use_config_toolsets=False
|
|
)
|
|
assert closed == [True]
|
|
assert shutdowns == [None]
|
|
|
|
|
|
def test_oneshot_run_agent_closes_session_db(monkeypatch):
|
|
# The one-shot exit path hard-exits via os._exit and skips finalizers, so
|
|
# the recall SQLite store it opens must be closed explicitly (checkpointing
|
|
# its WAL) rather than left to interpreter teardown.
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
db_closed = []
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **_kwargs):
|
|
self.suppress_status_output = False
|
|
self.stream_delta_callback = object()
|
|
self.tool_gen_callback = object()
|
|
self._session_messages = []
|
|
|
|
def run_conversation(self, _prompt, **_kwargs):
|
|
return {"final_response": "done"}
|
|
|
|
def shutdown_memory_provider(self, messages=None):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
class FakeSessionDB:
|
|
def close(self):
|
|
db_closed.append(True)
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules, "run_agent", types.SimpleNamespace(AIAgent=FakeAgent)
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"model": {"default": "gpt-test", "provider": "openai"}},
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
lambda **_kwargs: {
|
|
"api_key": "key",
|
|
"base_url": "https://example.invalid",
|
|
"provider": "openai",
|
|
"api_mode": "chat_completions",
|
|
"credential_pool": None,
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
oneshot_mod, "_create_session_db_for_oneshot", lambda: FakeSessionDB()
|
|
)
|
|
|
|
assert (
|
|
oneshot_mod._run_agent(
|
|
"hello", model="gpt-test", provider="openai", use_config_toolsets=False
|
|
)
|
|
== ("done", {"final_response": "done"})
|
|
)
|
|
assert db_closed == [True]
|
|
|
|
|
|
def test_oneshot_run_agent_closes_session_db_when_agent_init_raises(monkeypatch):
|
|
# The recall store is opened before AIAgent is constructed. If construction
|
|
# raises (bad provider/config/model), the store must still be closed — the
|
|
# one-shot exit hard-exits via os._exit and skips finalizers, so an
|
|
# un-closed connection would leave a stale WAL behind.
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
db_closed = []
|
|
|
|
class FakeSessionDB:
|
|
def close(self):
|
|
db_closed.append(True)
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **_kwargs):
|
|
raise RuntimeError("init boom")
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules, "run_agent", types.SimpleNamespace(AIAgent=FakeAgent)
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config",
|
|
lambda: {"model": {"default": "gpt-test", "provider": "openai"}},
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.runtime_provider.resolve_runtime_provider",
|
|
lambda **_kwargs: {
|
|
"api_key": "key",
|
|
"base_url": "https://example.invalid",
|
|
"provider": "openai",
|
|
"api_mode": "chat_completions",
|
|
"credential_pool": None,
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
oneshot_mod, "_create_session_db_for_oneshot", lambda: FakeSessionDB()
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="init boom"):
|
|
oneshot_mod._run_agent(
|
|
"hello", model="gpt-test", provider="openai", use_config_toolsets=False
|
|
)
|
|
|
|
assert db_closed == [True]
|
|
|
|
|
|
def _stub_plugin_discovery(monkeypatch):
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.plugins",
|
|
types.SimpleNamespace(discover_plugins=lambda: None),
|
|
)
|
|
|
|
|
|
def test_oneshot_rejects_invalid_only_toolsets(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
from hermes_cli.oneshot import run_oneshot
|
|
|
|
assert run_oneshot("hello", toolsets="nope") == 2
|
|
err = capsys.readouterr().err
|
|
assert "nope" in err
|
|
assert "did not contain any valid toolsets" in err
|
|
|
|
|
|
def test_oneshot_fails_closed_on_empty_final_response(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
monkeypatch.setattr(oneshot_mod, "_run_agent", lambda *_args, **_kwargs: ("", {}))
|
|
|
|
assert oneshot_mod.run_oneshot("hello") == 1
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
assert "no final response" in captured.err
|
|
|
|
|
|
def test_oneshot_prints_nonempty_final_response(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
monkeypatch.setattr(oneshot_mod, "_run_agent", lambda *_args, **_kwargs: ("done", {}))
|
|
|
|
assert oneshot_mod.run_oneshot("hello") == 0
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "done\n"
|
|
assert captured.err == ""
|
|
|
|
|
|
def test_oneshot_fails_closed_on_agent_exception(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
|
|
def _boom(*_args, **_kwargs):
|
|
raise OSError("not a TTY")
|
|
|
|
monkeypatch.setattr(oneshot_mod, "_run_agent", _boom)
|
|
|
|
assert oneshot_mod.run_oneshot("hello") == 1
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
assert "agent failed" in captured.err
|
|
assert "not a TTY" in captured.err
|
|
|
|
|
|
def test_oneshot_exit_code_when_failed_without_response(monkeypatch):
|
|
from hermes_cli.oneshot import run_oneshot
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.oneshot._run_agent",
|
|
lambda *_a, **_k: ("", {"failed": True, "partial": False}),
|
|
)
|
|
assert run_oneshot("hi") == 2
|
|
|
|
|
|
def test_oneshot_exit_code_zero_when_failed_with_error_text(monkeypatch, capsys):
|
|
from hermes_cli.oneshot import run_oneshot
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.oneshot._run_agent",
|
|
lambda *_a, **_k: (
|
|
"API call failed after 3 retries: HTTP 404: model not found",
|
|
{"failed": True, "partial": False},
|
|
),
|
|
)
|
|
assert run_oneshot("hi") == 0
|
|
assert "HTTP 404" in capsys.readouterr().out
|
|
|
|
|
|
def test_oneshot_reraises_keyboard_interrupt(monkeypatch):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
import hermes_cli.oneshot as oneshot_mod
|
|
import pytest as _pytest
|
|
|
|
def _interrupt(*_args, **_kwargs):
|
|
raise KeyboardInterrupt
|
|
|
|
monkeypatch.setattr(oneshot_mod, "_run_agent", _interrupt)
|
|
|
|
with _pytest.raises(KeyboardInterrupt):
|
|
oneshot_mod.run_oneshot("hello")
|
|
|
|
|
|
def test_oneshot_filters_invalid_toolsets_before_redirect(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
from hermes_cli.oneshot import _validate_explicit_toolsets
|
|
|
|
valid, error = _validate_explicit_toolsets("web,nope")
|
|
|
|
assert valid == ["web"]
|
|
assert error is None
|
|
assert "nope" in capsys.readouterr().err
|
|
|
|
|
|
def test_oneshot_all_toolsets_means_all_not_configured_cli():
|
|
from hermes_cli.oneshot import _validate_explicit_toolsets
|
|
|
|
valid, error = _validate_explicit_toolsets("all")
|
|
|
|
assert valid is None
|
|
assert error is None
|
|
|
|
|
|
def test_oneshot_all_toolsets_warns_about_ignored_extra_entries(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
from hermes_cli.oneshot import _validate_explicit_toolsets
|
|
|
|
valid, error = _validate_explicit_toolsets("all,nope")
|
|
|
|
assert valid is None
|
|
assert error is None
|
|
assert "ignoring additional entries: nope" in capsys.readouterr().err
|
|
|
|
|
|
def test_oneshot_accepts_plugin_toolset_after_discovery(monkeypatch):
|
|
import toolsets
|
|
|
|
from hermes_cli.oneshot import _validate_explicit_toolsets
|
|
|
|
discovered = {"ready": False}
|
|
original_validate = toolsets.validate_toolset
|
|
|
|
def fake_validate(name):
|
|
return name == "plugin_demo" and discovered["ready"] or original_validate(name)
|
|
|
|
monkeypatch.setattr(toolsets, "validate_toolset", fake_validate)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.plugins",
|
|
types.SimpleNamespace(
|
|
discover_plugins=lambda: discovered.update({"ready": True})
|
|
),
|
|
)
|
|
|
|
valid, error = _validate_explicit_toolsets("plugin_demo")
|
|
|
|
assert valid == ["plugin_demo"]
|
|
assert error is None
|
|
|
|
|
|
def test_oneshot_rejects_disabled_mcp_toolset(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
import hermes_cli.config as config_mod
|
|
|
|
from hermes_cli.oneshot import _validate_explicit_toolsets
|
|
|
|
monkeypatch.setattr(
|
|
config_mod,
|
|
"read_raw_config",
|
|
lambda: {"mcp_servers": {"mcp-off": {"enabled": False}}},
|
|
)
|
|
|
|
valid, error = _validate_explicit_toolsets("mcp-off")
|
|
|
|
assert valid is None
|
|
assert error == "hermes -z: --toolsets did not contain any valid toolsets.\n"
|
|
err = capsys.readouterr().err
|
|
assert "ignoring disabled MCP servers" in err
|
|
assert "mcp-off" in err
|
|
|
|
|
|
def test_oneshot_distinguishes_disabled_mcp_from_unknown(monkeypatch, capsys):
|
|
_stub_plugin_discovery(monkeypatch)
|
|
import hermes_cli.config as config_mod
|
|
|
|
from hermes_cli.oneshot import _validate_explicit_toolsets
|
|
|
|
monkeypatch.setattr(
|
|
config_mod,
|
|
"read_raw_config",
|
|
lambda: {"mcp_servers": {"mcp-off": {"enabled": False}}},
|
|
)
|
|
|
|
valid, error = _validate_explicit_toolsets("web,mcp-off,nope")
|
|
|
|
assert valid == ["web"]
|
|
assert error is None
|
|
err = capsys.readouterr().err
|
|
assert "ignoring unknown --toolsets entries: nope" in err
|
|
assert "ignoring disabled MCP servers" in err
|
|
assert "mcp-off" in err
|
|
|
|
|
|
def test_oneshot_wires_session_db_for_recall(monkeypatch):
|
|
"""hermes -z bypasses HermesCLI, but recall still needs SessionDB."""
|
|
from hermes_cli.oneshot import _run_agent
|
|
|
|
captured = {}
|
|
sentinel_db = object()
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **kwargs):
|
|
captured.update(kwargs)
|
|
self.suppress_status_output = False
|
|
self.stream_delta_callback = object()
|
|
self.tool_gen_callback = object()
|
|
|
|
def run_conversation(self, prompt, **_kwargs):
|
|
captured["prompt"] = prompt
|
|
return {"final_response": "ok", "failed": False, "partial": False}
|
|
|
|
class FakeSessionDB:
|
|
def __new__(cls):
|
|
return sentinel_db
|
|
|
|
def mod(name, **attrs):
|
|
module = types.ModuleType(name)
|
|
for key, value in attrs.items():
|
|
setattr(module, key, value)
|
|
return module
|
|
|
|
monkeypatch.setitem(sys.modules, "run_agent", mod("run_agent", AIAgent=FakeAgent))
|
|
monkeypatch.setitem(sys.modules, "hermes_state", mod("hermes_state", SessionDB=FakeSessionDB))
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.config",
|
|
mod("hermes_cli.config", load_config=lambda: {"model": {"default": "m"}}),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.models",
|
|
mod("hermes_cli.models", detect_provider_for_model=lambda *_args, **_kwargs: None),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.runtime_provider",
|
|
mod(
|
|
"hermes_cli.runtime_provider",
|
|
resolve_runtime_provider=lambda **_kwargs: {
|
|
"api_key": "k",
|
|
"base_url": "u",
|
|
"provider": "p",
|
|
"api_mode": "chat_completions",
|
|
"credential_pool": None,
|
|
},
|
|
),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"hermes_cli.tools_config",
|
|
mod("hermes_cli.tools_config", _get_platform_tools=lambda *_args, **_kwargs: {"session_search"}),
|
|
)
|
|
|
|
text, result = _run_agent("recall this")
|
|
assert text == "ok"
|
|
assert not result.get("failed")
|
|
assert captured["session_db"] is sentinel_db
|
|
assert captured["enabled_toolsets"] == ["session_search"]
|
|
assert captured["prompt"] == "recall this"
|
|
|
|
|
|
def test_launch_tui_exports_model_provider_and_toolsets(monkeypatch, main_mod):
|
|
captured = {}
|
|
active_path_during_call = None
|
|
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_make_tui_argv",
|
|
lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
|
|
)
|
|
|
|
def fake_call(argv, cwd=None, env=None):
|
|
nonlocal active_path_during_call
|
|
captured.update({"argv": argv, "cwd": cwd, "env": env})
|
|
active_path_during_call = Path(env["HERMES_TUI_ACTIVE_SESSION_FILE"])
|
|
assert active_path_during_call.exists()
|
|
return 1
|
|
|
|
monkeypatch.setattr(main_mod.subprocess, "call", fake_call)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod._launch_tui(
|
|
model="nous/hermes-test", provider="nous", toolsets="web, terminal"
|
|
)
|
|
|
|
env = captured["env"]
|
|
assert env["HERMES_MODEL"] == "nous/hermes-test"
|
|
assert env["HERMES_INFERENCE_MODEL"] == "nous/hermes-test"
|
|
assert env["HERMES_TUI_PROVIDER"] == "nous"
|
|
assert env["HERMES_INFERENCE_PROVIDER"] == "nous"
|
|
assert env["HERMES_TUI_TOOLSETS"] == "web,terminal"
|
|
active_path = Path(env["HERMES_TUI_ACTIVE_SESSION_FILE"])
|
|
assert active_path.name.startswith("hermes-tui-active-session-")
|
|
assert active_path.suffix == ".json"
|
|
assert active_path_during_call == active_path
|
|
assert not active_path.exists()
|
|
assert env["NODE_ENV"] == "production"
|
|
|
|
|
|
def test_launch_tui_worktree_validates_relative_python_against_final_cwd(
|
|
monkeypatch, main_mod, tmp_path
|
|
):
|
|
import cli as cli_mod
|
|
|
|
parent_cwd = tmp_path / "parent"
|
|
parent_cwd.mkdir()
|
|
worktree = tmp_path / "worktree"
|
|
relative_python = Path(".review-venv") / "bin" / Path(sys.executable).name
|
|
python_path = worktree / relative_python
|
|
python_path.parent.mkdir(parents=True)
|
|
os.link(sys.executable, python_path)
|
|
captured = {}
|
|
|
|
monkeypatch.setenv("HERMES_CWD", str(parent_cwd))
|
|
monkeypatch.setenv("HERMES_PYTHON", str(relative_python))
|
|
monkeypatch.setattr(cli_mod, "_git_repo_root", lambda: None)
|
|
monkeypatch.setattr(cli_mod, "_prune_stale_worktrees", lambda _repo: None)
|
|
monkeypatch.setattr(cli_mod, "_setup_worktree", lambda: {"path": str(worktree)})
|
|
monkeypatch.setattr(cli_mod, "_cleanup_worktree", lambda _info: None)
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_make_tui_argv",
|
|
lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod.subprocess,
|
|
"call",
|
|
lambda argv, cwd=None, env=None: captured.update({"env": env}) or 1,
|
|
)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod._launch_tui(worktree=True)
|
|
|
|
assert captured["env"]["HERMES_CWD"] == str(worktree)
|
|
assert captured["env"]["HERMES_PYTHON"] == str(relative_python)
|
|
|
|
|
|
def test_launch_tui_applies_terminal_backend_config(
|
|
monkeypatch, main_mod, _isolate_hermes_home
|
|
):
|
|
captured = {}
|
|
config_path = Path(os.environ["HERMES_HOME"]) / "config.yaml"
|
|
config_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"terminal:",
|
|
" backend: docker",
|
|
" docker_image: example/hermes-tools:latest",
|
|
" docker_extra_args:",
|
|
" - --network=host",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.delenv("TERMINAL_ENV", raising=False)
|
|
monkeypatch.delenv("TERMINAL_DOCKER_IMAGE", raising=False)
|
|
monkeypatch.delenv("TERMINAL_DOCKER_EXTRA_ARGS", raising=False)
|
|
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_make_tui_argv",
|
|
lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod.subprocess,
|
|
"call",
|
|
lambda argv, cwd=None, env=None: captured.update({"env": env}) or 1,
|
|
)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod._launch_tui()
|
|
|
|
assert captured["env"]["TERMINAL_ENV"] == "docker"
|
|
assert captured["env"]["TERMINAL_DOCKER_IMAGE"] == "example/hermes-tools:latest"
|
|
assert captured["env"]["TERMINAL_DOCKER_EXTRA_ARGS"] == '["--network=host"]'
|
|
|
|
|
|
def test_launch_tui_exit_code_42_relaunches_update(monkeypatch, main_mod):
|
|
from unittest.mock import patch
|
|
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_make_tui_argv",
|
|
lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
|
|
)
|
|
monkeypatch.setattr(main_mod.subprocess, "call", lambda *args, **kwargs: 42)
|
|
|
|
with patch("hermes_cli.relaunch.relaunch") as mock_relaunch:
|
|
with pytest.raises(SystemExit) as exc:
|
|
main_mod._launch_tui()
|
|
|
|
assert exc.value.code == 42
|
|
mock_relaunch.assert_called_once_with(["update"], preserve_inherited=False)
|
|
|
|
|
|
def test_launch_tui_drops_stale_resume_env_without_resume_arg(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
monkeypatch.setenv("HERMES_TUI_RESUME", "stale-missing-session")
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_make_tui_argv",
|
|
lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod.subprocess,
|
|
"call",
|
|
lambda argv, cwd=None, env=None: captured.update({"env": env}) or 1,
|
|
)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod._launch_tui()
|
|
|
|
assert "HERMES_TUI_RESUME" not in captured["env"]
|
|
|
|
|
|
def test_launch_tui_sets_resume_env_from_resume_arg(monkeypatch, main_mod):
|
|
captured = {}
|
|
|
|
monkeypatch.setenv("HERMES_TUI_RESUME", "stale-missing-session")
|
|
monkeypatch.setattr(
|
|
main_mod,
|
|
"_make_tui_argv",
|
|
lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
|
|
)
|
|
monkeypatch.setattr(
|
|
main_mod.subprocess,
|
|
"call",
|
|
lambda argv, cwd=None, env=None: captured.update({"env": env}) or 1,
|
|
)
|
|
|
|
with pytest.raises(SystemExit):
|
|
main_mod._launch_tui(resume_session_id="20260518_000000_goodid")
|
|
|
|
assert captured["env"]["HERMES_TUI_RESUME"] == "20260518_000000_goodid"
|
|
|
|
|
|
def test_make_tui_argv_dev_prebuilds_hermes_ink(monkeypatch, main_mod, tmp_path):
|
|
tui_dir = tmp_path / "ui-tui"
|
|
tsx = tui_dir / "node_modules" / ".bin" / "tsx"
|
|
ink_dir = tui_dir / "packages" / "hermes-ink"
|
|
tsx.parent.mkdir(parents=True)
|
|
ink_dir.mkdir(parents=True)
|
|
tsx.write_text("#!/usr/bin/env node\n", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(main_mod, "_ensure_tui_node", lambda: None)
|
|
monkeypatch.setattr(main_mod, "_tui_need_npm_install", lambda _tui_dir: False)
|
|
monkeypatch.delenv("HERMES_TUI_DIR", raising=False)
|
|
monkeypatch.setattr(main_mod.shutil, "which", lambda bin_name: f"/usr/bin/{bin_name}")
|
|
|
|
calls = []
|
|
|
|
def fake_run(cmd, cwd=None, **_kwargs):
|
|
calls.append((cmd, cwd))
|
|
return types.SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(main_mod.subprocess, "run", fake_run)
|
|
|
|
argv, cwd = main_mod._make_tui_argv(tui_dir, tui_dev=True)
|
|
|
|
assert argv == [str(tsx), "src/entry.tsx"]
|
|
assert cwd == tui_dir
|
|
assert calls == [(["/usr/bin/npm", "run", "build"], str(ink_dir))]
|
|
|
|
|
|
def test_print_tui_exit_summary_includes_resume_and_token_totals(monkeypatch, capsys):
|
|
import hermes_cli.main as main_mod
|
|
|
|
class _FakeDB:
|
|
def get_session(self, session_id):
|
|
assert session_id == "20260409_000001_abc123"
|
|
return {
|
|
"message_count": 2,
|
|
"input_tokens": 10,
|
|
"output_tokens": 6,
|
|
"cache_read_tokens": 2,
|
|
"cache_write_tokens": 2,
|
|
"reasoning_tokens": 1,
|
|
}
|
|
|
|
def get_session_title(self, _session_id):
|
|
return "demo title"
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules, "hermes_state", types.SimpleNamespace(SessionDB=lambda: _FakeDB())
|
|
)
|
|
|
|
main_mod._print_tui_exit_summary("20260409_000001_abc123")
|
|
out = capsys.readouterr().out
|
|
|
|
assert "Resume this session with:" in out
|
|
assert "hermes --tui --resume 20260409_000001_abc123" in out
|
|
assert 'hermes --tui -c "demo title"' in out
|
|
assert "Tokens: 21 (in 10, out 6, cache 4, reasoning 1)" in out
|
|
|
|
|
|
def test_print_tui_exit_summary_prefers_actual_active_session_file(
|
|
monkeypatch, capsys, tmp_path
|
|
):
|
|
import hermes_cli.main as main_mod
|
|
|
|
seen = []
|
|
|
|
class _FakeDB:
|
|
def get_session(self, session_id):
|
|
seen.append(session_id)
|
|
return {
|
|
"message_count": 1,
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
"cache_read_tokens": 0,
|
|
"cache_write_tokens": 0,
|
|
"reasoning_tokens": 0,
|
|
}
|
|
|
|
def get_session_title(self, _session_id):
|
|
return "actual"
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
active = tmp_path / "active.json"
|
|
active.write_text('{"session_id":"actual_session"}', encoding="utf-8")
|
|
monkeypatch.setitem(
|
|
sys.modules, "hermes_state", types.SimpleNamespace(SessionDB=lambda: _FakeDB())
|
|
)
|
|
|
|
main_mod._print_tui_exit_summary("startup_resume", str(active))
|
|
out = capsys.readouterr().out
|
|
|
|
assert seen == ["actual_session"]
|
|
assert "hermes --tui --resume actual_session" in out
|
|
assert "startup_resume" not in out
|