mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
6b81590c55
Systematic prune per AGENTS.md test policy, one pass over every major test tree (gateway, hermes_cli, tools, agent, run_agent, plugins, cli, cron, tui_gateway, honcho/openviking, root-level): - DELETE: source-reading tests (read_text/getsource on prod files), change-detector tests (exact catalog counts, model-name snapshots, config version literals), mock-echo tests (assert a mock returns what it was told), assertion-free/trivial tests, near-duplicate parametrizations (boundaries + one representative kept), async/sync twin duplicates, cosmetic within-file variations. - KEEP (mandatory): security/redaction/approval guards, message-role alternation invariants, prompt-caching/deterministic-call-id invariants, issue-number regression tests (deduped), E2E tests. - 6 test files deleted outright (script-style/no-assert or fully redundant); conftest.py, fakes/, fixtures/ untouched. - tests/acp/conftest.py added: autouse fixture stubs the live models.dev/GitHub/Copilot/Anthropic inventory fetches that ACP server tests performed on every session create — test_server.py 147s → 3.4s, and the tests are now genuinely hermetic. - Sleep-based slowness shrunk where safe (codex_ttfb_watchdog, compression_concurrent_fork, etc.); no wall-clock assertion tightened. Verification: full hermetic suite via scripts/run_tests.sh — 2439 files, 31,130 tests passed, 0 failed, 0 flaky retries, 315s wall (baseline: 583s wall, 13,564s subprocess CPU).
340 lines
11 KiB
Python
340 lines
11 KiB
Python
"""Unit tests for ``hermes_cli.proxy_cli`` command handlers.
|
|
|
|
These tests cover the user-facing CLI surface that was previously
|
|
uncovered. We mock the iron_proxy module's side-effect functions
|
|
(install / start / stop / discover) and exercise the dispatch +
|
|
return-code logic plus the small amount of presentation logic in
|
|
each handler (e.g. --from-bitwarden's fail-loud path).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from agent.proxy_sources import iron_proxy as ip
|
|
from hermes_cli import proxy_cli
|
|
|
|
|
|
@pytest.fixture
|
|
def hermes_home(tmp_path, monkeypatch):
|
|
"""Point HERMES_HOME at a temp dir so the wizard doesn't touch the
|
|
operator's real config. Also blanks any provider env vars so we
|
|
don't accidentally read a real key."""
|
|
|
|
home = tmp_path / "hermes"
|
|
home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
for key in list(os.environ):
|
|
if key.endswith("_API_KEY") or key in (
|
|
"BWS_ACCESS_TOKEN", "ANTHROPIC_API_KEY",
|
|
"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
return home
|
|
|
|
|
|
def _args(**overrides):
|
|
ns = argparse.Namespace(
|
|
force=False,
|
|
tunnel_port=None,
|
|
from_bitwarden=False,
|
|
no_bitwarden=False,
|
|
rotate_tokens=False,
|
|
restart=None,
|
|
show_tokens=False,
|
|
)
|
|
for k, v in overrides.items():
|
|
setattr(ns, k, v)
|
|
return ns
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cmd_install
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cmd_install_success_returns_0(hermes_home, monkeypatch):
|
|
monkeypatch.setattr(ip, "install_iron_proxy", lambda **kw: hermes_home / "iron-proxy")
|
|
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "v0.39.0-test")
|
|
rc = proxy_cli.cmd_install(_args())
|
|
assert rc == 0
|
|
|
|
|
|
def test_cmd_install_failure_returns_1(hermes_home, monkeypatch):
|
|
def boom(**kw):
|
|
raise RuntimeError("download failed")
|
|
monkeypatch.setattr(ip, "install_iron_proxy", boom)
|
|
rc = proxy_cli.cmd_install(_args())
|
|
assert rc == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cmd_setup — --from-bitwarden fail-loud paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_cmd_setup_from_bitwarden_refuses_on_empty_vault(hermes_home, monkeypatch):
|
|
"""If BW returns {} (empty vault / scoped wrong / unreachable), fail
|
|
loud rather than silently writing credential_source: bitwarden."""
|
|
|
|
from hermes_cli.config import load_config, save_config
|
|
|
|
cfg = load_config()
|
|
cfg.setdefault("secrets", {})["bitwarden"] = {
|
|
"enabled": True,
|
|
"project_id": "test-proj",
|
|
"access_token_env": "BWS_ACCESS_TOKEN",
|
|
}
|
|
save_config(cfg)
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "bwsk-test-token")
|
|
|
|
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: hermes_home / "iron-proxy")
|
|
monkeypatch.setattr(ip, "iron_proxy_version", lambda b: "test")
|
|
monkeypatch.setattr(
|
|
ip, "ensure_ca_cert",
|
|
lambda **kw: (hermes_home / "ca.crt", hermes_home / "ca.key"),
|
|
)
|
|
|
|
# Mock fetch_bitwarden_secrets to return an empty dict (empty vault).
|
|
fake_bw = MagicMock()
|
|
fake_bw.fetch_bitwarden_secrets = lambda **kw: ({}, [])
|
|
monkeypatch.setattr("agent.secret_sources.bitwarden", fake_bw, raising=False)
|
|
import sys
|
|
sys.modules["agent.secret_sources.bitwarden"] = fake_bw
|
|
|
|
rc = proxy_cli.cmd_setup(_args(from_bitwarden=True))
|
|
assert rc == 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_cmd_start_passes_bitwarden_refresh_flag_when_credential_source_is_bitwarden(
|
|
hermes_home, monkeypatch,
|
|
):
|
|
"""When credential_source=bitwarden, cmd_start must wire
|
|
refresh_secrets_from_bitwarden=True into start_proxy. That's what
|
|
delivers the rotation promise the docs make."""
|
|
|
|
from hermes_cli.config import load_config, save_config
|
|
cfg = load_config()
|
|
cfg.setdefault("proxy", {})["enabled"] = True
|
|
cfg["proxy"]["credential_source"] = "bitwarden"
|
|
cfg.setdefault("secrets", {})["bitwarden"] = {
|
|
"enabled": True,
|
|
"project_id": "test-proj-id",
|
|
"access_token_env": "BWS_ACCESS_TOKEN",
|
|
}
|
|
save_config(cfg)
|
|
# v3: cmd_start now pre-checks BWS access token + project_id before
|
|
# calling start_proxy. Provide both so we get to the rotation
|
|
# wire-up code path.
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "bwsk-test-access-token")
|
|
|
|
captured: dict = {}
|
|
def fake_start_proxy(**kw):
|
|
captured.update(kw)
|
|
s = ip.ProxyStatus()
|
|
s.pid = 4242
|
|
s.listening = True
|
|
s.tunnel_port = 9090
|
|
return s
|
|
monkeypatch.setattr(ip, "start_proxy", fake_start_proxy)
|
|
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
|
|
|
|
rc = proxy_cli.cmd_start(_args())
|
|
assert rc == 0
|
|
assert captured.get("refresh_secrets_from_bitwarden") is True
|
|
assert captured.get("bitwarden_config") is not None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cmd_stop, cmd_status, cmd_disable, cmd_config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cmd_restart
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_cmd_restart_propagates_start_failure(hermes_home, monkeypatch):
|
|
monkeypatch.setattr(ip, "stop_proxy", lambda: True)
|
|
monkeypatch.setattr(proxy_cli, "cmd_start", lambda args: 1)
|
|
rc = proxy_cli.cmd_restart(_args())
|
|
assert rc == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _load_env_file_into_environ — setup discovers keys kept only in ~/.hermes/.env
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_cmd_status_returns_0(hermes_home, monkeypatch):
|
|
monkeypatch.setattr(ip, "get_status", lambda: ip.ProxyStatus())
|
|
monkeypatch.setattr(ip, "load_mappings", lambda: [])
|
|
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
|
|
rc = proxy_cli.cmd_status(_args())
|
|
assert rc == 0
|
|
|
|
|
|
def test_cmd_disable_uses_public_status_pid_not_private_read_pid(
|
|
hermes_home, monkeypatch,
|
|
):
|
|
"""cmd_disable must read status.pid (which incorporates the _pid_alive
|
|
check) — NOT ip._read_pid() directly (which would fire a spurious
|
|
'still running' warning for a stale pidfile from a crashed run)."""
|
|
|
|
from hermes_cli.config import load_config, save_config
|
|
|
|
cfg = load_config()
|
|
cfg.setdefault("proxy", {})["enabled"] = True
|
|
save_config(cfg)
|
|
|
|
# Pidfile exists but the process is dead. Old code would have warned
|
|
# "still running"; the new code reads status.pid which returns None
|
|
# because _pid_alive is False, so no spurious warning.
|
|
state = ip._proxy_state_dir()
|
|
(state / "iron-proxy.pid").write_text("99999")
|
|
# _pid_alive returns False → status.pid is None.
|
|
monkeypatch.setattr(ip, "_pid_alive", lambda pid: False)
|
|
|
|
# If cmd_disable reads _read_pid() directly (old path), this test
|
|
# would still pass — but reading status.pid is the correct
|
|
# API. Sentinel: confirm _read_pid is NOT called from cmd_disable.
|
|
read_pid_calls = []
|
|
real_read_pid = ip._read_pid
|
|
def tracked_read_pid(*a, **kw):
|
|
read_pid_calls.append((a, kw))
|
|
return real_read_pid(*a, **kw)
|
|
monkeypatch.setattr(ip, "_read_pid", tracked_read_pid)
|
|
|
|
rc = proxy_cli.cmd_disable(_args())
|
|
assert rc == 0
|
|
# cmd_disable should call get_status() (which may internally call
|
|
# _read_pid), but should NOT call _read_pid from its own body.
|
|
# Hard to assert directly without source-introspection — the meatier
|
|
# assertion is that no "still running" message fired with a stale
|
|
# pidfile. That's covered by inspecting return code + config
|
|
# mutation only.
|
|
from hermes_cli.config import load_config as _lc
|
|
cfg2 = _lc()
|
|
assert cfg2["proxy"]["enabled"] is False
|
|
|
|
|
|
def test_cmd_config_returns_0_when_present(hermes_home, monkeypatch):
|
|
fake = ip.ProxyStatus()
|
|
fake.config_path = hermes_home / "proxy.yaml"
|
|
monkeypatch.setattr(ip, "get_status", lambda: fake)
|
|
rc = proxy_cli.cmd_config(_args())
|
|
assert rc == 0
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Argparse wiring — dest='egress_command' regression
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_register_cli_uses_egress_command_dest():
|
|
"""The subparser dest must be 'egress_command' to stay disjoint from
|
|
the inbound OAuth 'hermes proxy' subparser (dest='proxy_command').
|
|
A future grep-and-refactor on proxy_command should not hit this
|
|
subparser by accident."""
|
|
|
|
parser = argparse.ArgumentParser(prog="hermes egress")
|
|
proxy_cli.register_cli(parser)
|
|
# Parse a no-op invocation and confirm the attribute name.
|
|
args = parser.parse_args(["install"])
|
|
assert hasattr(args, "egress_command")
|
|
assert not hasattr(args, "proxy_command")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# v4 round: credential_source=bitwarden with secrets.bitwarden disabled
|
|
# must NOT silently degrade to host-env secrets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cmd_start_refuses_when_bitwarden_mode_but_disabled(hermes_home, monkeypatch):
|
|
"""config keeps credential_source: bitwarden but secrets.bitwarden.enabled
|
|
later flips to false — cmd_start must refuse, not silently start on
|
|
host env (the silent-degrade class strict mode is meant to close)."""
|
|
|
|
from hermes_cli.config import load_config, save_config
|
|
cfg = load_config()
|
|
cfg.setdefault("proxy", {})["enabled"] = True
|
|
cfg["proxy"]["credential_source"] = "bitwarden"
|
|
cfg.setdefault("secrets", {})["bitwarden"] = {"enabled": False}
|
|
save_config(cfg)
|
|
|
|
def must_not_call(**kw):
|
|
pytest.fail("start_proxy must not run when bitwarden mode is broken")
|
|
monkeypatch.setattr(ip, "start_proxy", must_not_call)
|
|
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
|
|
|
|
rc = proxy_cli.cmd_start(_args())
|
|
assert rc == 1
|
|
|
|
|
|
|
|
|
|
def test_cmd_setup_audit_log_failure_is_warning_not_abort(hermes_home, monkeypatch):
|
|
"""On the pinned v0.39 the daemon never writes audit.log, so a
|
|
pre-create failure must not abort the wizard."""
|
|
|
|
monkeypatch.setattr(ip, "find_iron_proxy", lambda **kw: hermes_home / "iron-proxy")
|
|
monkeypatch.setattr(ip, "discover_provider_mappings", lambda **kw: [
|
|
ip.TokenMapping(
|
|
proxy_token="hermes-proxy-deadbeef",
|
|
real_env_name="OPENROUTER_API_KEY",
|
|
upstream_hosts=("openrouter.ai",),
|
|
),
|
|
])
|
|
monkeypatch.setattr(ip, "discover_uncovered_providers", lambda **kw: [])
|
|
|
|
def boom(path):
|
|
raise RuntimeError("could not pre-create audit log (synthetic)")
|
|
monkeypatch.setattr(ip, "ensure_audit_log", boom)
|
|
|
|
rc = proxy_cli.cmd_setup(_args())
|
|
assert rc == 0
|