"""Tests for hermes_cli.mcp_catalog and hermes_cli.mcp_picker. Manifest parsing, install/uninstall config writes, and picker plumbing are exercised here. Anything that would actually clone a repo or launch an MCP is mocked. """ from __future__ import annotations from pathlib import Path from unittest.mock import patch import pytest import yaml # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _default_mock_probe(monkeypatch): """By default tests run the probe-fails path so install_entry() doesn\'t try to talk to a real MCP server. Individual tests that exercise probe-success behaviour patch ``hermes_cli.mcp_catalog._probe_tools`` themselves. """ # Patch the catalog\'s probe wrapper, not the underlying # mcp_config._probe_single_server (so tests stay decoupled from that # module\'s plumbing). import hermes_cli.mcp_catalog as mc monkeypatch.setattr(mc, "_probe_tools", lambda name: None) @pytest.fixture def catalog_dir(tmp_path, monkeypatch): """Provide an isolated optional-mcps/ directory.""" cat = tmp_path / "optional-mcps" cat.mkdir() monkeypatch.setenv("HERMES_OPTIONAL_MCPS", str(cat)) return cat @pytest.fixture(autouse=True) def _isolate_hermes_home(tmp_path, monkeypatch): """Redirect all config I/O to a temp HERMES_HOME.""" hh = tmp_path / "hermes-home" hh.mkdir() monkeypatch.setenv("HERMES_HOME", str(hh)) monkeypatch.setattr( "hermes_cli.config.get_hermes_home", lambda: hh ) monkeypatch.setattr( "hermes_cli.config.get_config_path", lambda: hh / "config.yaml" ) monkeypatch.setattr( "hermes_cli.config.get_env_path", lambda: hh / ".env" ) # mcp_catalog grabs get_hermes_home() lazily through hermes_constants monkeypatch.setattr( "hermes_constants.get_hermes_home", lambda: hh ) return hh def _write_manifest(catalog_dir: Path, name: str, body: dict) -> Path: entry_dir = catalog_dir / name entry_dir.mkdir(exist_ok=True) path = entry_dir / "manifest.yaml" with open(path, "w") as f: yaml.safe_dump(body, f) return path def _basic_manifest(name: str = "demo", **overrides) -> dict: body = { "manifest_version": 1, "name": name, "description": "Demo MCP", "source": "https://example.com", "transport": { "type": "stdio", "command": "npx", "args": ["-y", "demo-mcp"], }, "auth": {"type": "none"}, } body.update(overrides) return body def _entry(name: str): """Wrapper that asserts entry exists (satisfies type-checker + nicer failure msg).""" from hermes_cli.mcp_catalog import get_entry e = get_entry(name) assert e is not None, f"catalog entry {name!r} missing" return e # --------------------------------------------------------------------------- # Manifest parsing # --------------------------------------------------------------------------- class TestManifestParsing: def test_minimal_valid(self, catalog_dir): _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_catalog import list_catalog entries = list_catalog() assert len(entries) == 1 e = entries[0] assert e.name == "demo" assert e.transport.type == "stdio" assert e.transport.command == "npx" assert e.transport.args == ["-y", "demo-mcp"] assert e.auth.type == "none" assert e.install is None def test_api_key_auth(self, catalog_dir): body = _basic_manifest( auth={ "type": "api_key", "env": [ {"name": "DEMO_KEY", "prompt": "API key", "secret": True}, {"name": "DEMO_URL", "prompt": "Base URL", "secret": False, "required": False}, ], } ) _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import list_catalog e = list_catalog()[0] assert e.auth.type == "api_key" assert len(e.auth.env) == 2 assert e.auth.env[0].name == "DEMO_KEY" assert e.auth.env[0].secret is True assert e.auth.env[1].required is False assert e.auth.env[1].secret is False def test_install_block(self, catalog_dir): body = _basic_manifest( install={ "type": "git", "url": "https://example.com/demo.git", "ref": "v1.0.0", "bootstrap": ["pip install -r requirements.txt"], }, transport={ "type": "stdio", "command": "${INSTALL_DIR}/.venv/bin/python", "args": ["${INSTALL_DIR}/server.py"], }, ) _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import list_catalog e = list_catalog()[0] assert e.install is not None assert e.install.url == "https://example.com/demo.git" assert e.install.ref == "v1.0.0" assert e.install.bootstrap == ["pip install -r requirements.txt"] def test_invalid_manifest_skipped(self, catalog_dir): # Broken: wrong manifest_version _write_manifest(catalog_dir, "bad", { "manifest_version": 99, "name": "bad", "description": "x", "transport": {"type": "stdio", "command": "x"}, }) # Good _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_catalog import list_catalog entries = list_catalog() assert [e.name for e in entries] == ["demo"] def test_missing_transport_command_rejected(self, catalog_dir): body = _basic_manifest() body["transport"] = {"type": "stdio"} # no command _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import list_catalog assert list_catalog() == [] def test_get_entry_strips_official_prefix(self, catalog_dir): _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_catalog import get_entry assert get_entry("demo") is not None assert get_entry("official/demo") is not None assert get_entry("missing") is None # --------------------------------------------------------------------------- # Install flow # --------------------------------------------------------------------------- class TestInstall: def test_install_simple_stdio_writes_config(self, catalog_dir): _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) cfg = load_config() servers = cfg["mcp_servers"] assert "demo" in servers assert servers["demo"]["command"] == "npx" assert servers["demo"]["args"] == ["-y", "demo-mcp"] assert servers["demo"]["enabled"] is True def test_install_with_install_dir_substitution(self, catalog_dir, tmp_path): body = _basic_manifest( install={ "type": "git", "url": "https://example.com/demo.git", "ref": "main", "bootstrap": [], }, transport={ "type": "stdio", "command": "${INSTALL_DIR}/run.sh", "args": ["${INSTALL_DIR}/cfg.json"], }, ) _write_manifest(catalog_dir, "demo", body) # Mock the git clone — return a known directory fake_clone = tmp_path / "fake-clone" fake_clone.mkdir() from hermes_cli import mcp_catalog from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config with patch.object(mcp_catalog, "_do_git_install", return_value=fake_clone): install_entry(_entry("demo"), enable=True) servers = load_config()["mcp_servers"] assert servers["demo"]["command"] == f"{fake_clone}/run.sh" assert servers["demo"]["args"] == [f"{fake_clone}/cfg.json"] def test_install_with_api_key_prompts_and_saves(self, catalog_dir, monkeypatch): body = _basic_manifest( auth={ "type": "api_key", "env": [{"name": "DEMO_KEY", "prompt": "key", "secret": True}], } ) _write_manifest(catalog_dir, "demo", body) from hermes_cli import mcp_catalog monkeypatch.setattr(mcp_catalog, "_prompt_input", lambda *a, **kw: "secret-val") from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import get_env_value, load_config install_entry(_entry("demo"), enable=True) assert get_env_value("DEMO_KEY") == "secret-val" assert "demo" in load_config()["mcp_servers"] def test_install_http_oauth_writes_auth_marker(self, catalog_dir): body = _basic_manifest( transport={"type": "http", "url": "https://mcp.example.com/sse"}, auth={"type": "oauth"}, ) _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] assert server["url"] == "https://mcp.example.com/sse" assert server["auth"] == "oauth" def test_install_required_env_missing_raises(self, catalog_dir, monkeypatch): body = _basic_manifest( auth={ "type": "api_key", "env": [{"name": "MUST", "prompt": "x", "required": True, "secret": False}], } ) _write_manifest(catalog_dir, "demo", body) from hermes_cli import mcp_catalog from hermes_cli.mcp_catalog import install_entry, CatalogError # User hits enter — empty input, no default monkeypatch.setattr(mcp_catalog, "_prompt_input", lambda *a, **kw: "") with pytest.raises(CatalogError): install_entry(_entry("demo"), enable=True) # --------------------------------------------------------------------------- # Uninstall # --------------------------------------------------------------------------- class TestUninstall: def test_uninstall_removes_server_block(self, catalog_dir): _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_catalog import install_entry, uninstall_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) assert "demo" in load_config().get("mcp_servers", {}) assert uninstall_entry("demo") is True assert "demo" not in load_config().get("mcp_servers", {}) def test_uninstall_missing_returns_false(self): from hermes_cli.mcp_catalog import uninstall_entry assert uninstall_entry("nonexistent") is False # --------------------------------------------------------------------------- # Picker (non-TTY paths only — interactive curses is integration-tested) # --------------------------------------------------------------------------- class TestPicker: def test_show_catalog_empty(self, catalog_dir, capsys): from hermes_cli.mcp_picker import show_catalog show_catalog() out = capsys.readouterr().out assert "No MCPs in the catalog or configured" in out def test_show_catalog_lists_entry(self, catalog_dir, capsys): _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_picker import show_catalog show_catalog() out = capsys.readouterr().out assert "demo" in out assert "available" in out def test_install_by_name_unknown(self, catalog_dir, capsys): from hermes_cli.mcp_picker import install_by_name rc = install_by_name("nope") assert rc == 1 assert "not in the catalog" in capsys.readouterr().out def test_install_by_name_success(self, catalog_dir): _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_picker import install_by_name from hermes_cli.config import load_config rc = install_by_name("demo") assert rc == 0 assert "demo" in load_config().get("mcp_servers", {}) def test_run_picker_non_tty_falls_back(self, catalog_dir, capsys, monkeypatch): _write_manifest(catalog_dir, "demo", _basic_manifest()) # Force isatty false import sys as _sys monkeypatch.setattr(_sys.stdin, "isatty", lambda: False) from hermes_cli.mcp_picker import run_picker run_picker() out = capsys.readouterr().out assert "MCP Catalog + configured servers" in out # --------------------------------------------------------------------------- # Shipped catalog (sanity: every manifest in the repo's optional-mcps/ parses) # --------------------------------------------------------------------------- class TestToolSelection: def _make_probed(self, *names): """Return a list of (tool_name, description) tuples for mocking.""" return [(n, f"description of {n}") for n in names] def test_probe_fail_no_default_writes_no_filter(self, catalog_dir): body = _basic_manifest() _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] # No tools.include => all tools active when reachable assert "tools" not in server, server def test_probe_fail_with_default_applies_directly(self, catalog_dir): body = _basic_manifest( tools={"default_enabled": ["a", "b", "c"]}, ) _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] assert server["tools"]["include"] == ["a", "b", "c"] def test_probe_success_non_tty_with_default_filters_to_default( self, catalog_dir, monkeypatch ): body = _basic_manifest( tools={"default_enabled": ["alpha", "gamma"]}, ) _write_manifest(catalog_dir, "demo", body) import hermes_cli.mcp_catalog as mc probed = self._make_probed("alpha", "beta", "gamma", "delta") monkeypatch.setattr(mc, "_probe_tools", lambda name: probed) import sys as _sys monkeypatch.setattr(_sys.stdin, "isatty", lambda: False) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] # Only the manifest defaults that actually exist on the server assert server["tools"]["include"] == ["alpha", "gamma"] def test_probe_success_non_tty_no_default_clears_filter( self, catalog_dir, monkeypatch ): _write_manifest(catalog_dir, "demo", _basic_manifest()) import hermes_cli.mcp_catalog as mc probed = self._make_probed("x", "y") monkeypatch.setattr(mc, "_probe_tools", lambda name: probed) import sys as _sys monkeypatch.setattr(_sys.stdin, "isatty", lambda: False) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] assert "tools" not in server def test_default_enabled_filters_out_unknown_tool_names( self, catalog_dir, monkeypatch ): """If manifest names a tool the server doesn\'t actually expose, it silently drops out — never written into tools.include.""" body = _basic_manifest( tools={"default_enabled": ["real", "ghost"]}, ) _write_manifest(catalog_dir, "demo", body) import hermes_cli.mcp_catalog as mc probed = self._make_probed("real", "other") monkeypatch.setattr(mc, "_probe_tools", lambda name: probed) import sys as _sys monkeypatch.setattr(_sys.stdin, "isatty", lambda: False) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] assert server["tools"]["include"] == ["real"] def test_reinstall_preserves_prior_user_selection( self, catalog_dir, monkeypatch ): """Second install of the same entry uses the user\'s prior tools.include as the pre-check, NOT the manifest default.""" body = _basic_manifest( tools={"default_enabled": ["alpha"]}, ) _write_manifest(catalog_dir, "demo", body) import hermes_cli.mcp_catalog as mc probed = self._make_probed("alpha", "beta", "gamma") monkeypatch.setattr(mc, "_probe_tools", lambda name: probed) import sys as _sys monkeypatch.setattr(_sys.stdin, "isatty", lambda: False) from hermes_cli.mcp_catalog import install_entry from hermes_cli.config import load_config, save_config # First install install_entry(_entry("demo"), enable=True) # Simulate user opening configure and choosing beta+gamma cfg = load_config() cfg["mcp_servers"]["demo"]["tools"]["include"] = ["beta", "gamma"] save_config(cfg) # Reinstall (non-TTY honors prior_selection over manifest default) install_entry(_entry("demo"), enable=True) server = load_config()["mcp_servers"]["demo"] assert server["tools"]["include"] == ["beta", "gamma"], server def test_manifest_invalid_default_enabled_rejected(self, catalog_dir): body = _basic_manifest() body["tools"] = {"default_enabled": "not a list"} _write_manifest(catalog_dir, "demo", body) from hermes_cli.mcp_catalog import list_catalog # Invalid manifests are silently skipped at list_catalog level assert list_catalog() == [] # --------------------------------------------------------------------------- # Forward-compat / diagnostics # --------------------------------------------------------------------------- class TestCatalogDiagnostics: def test_future_manifest_version_skipped_with_diagnostic(self, catalog_dir): """A manifest with a newer manifest_version is skipped, but the skip is reported via catalog_diagnostics so the UI can tell the user.""" body = _basic_manifest() body["manifest_version"] = 999 # Future version _write_manifest(catalog_dir, "futuristic", body) # Plus one valid entry _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.mcp_catalog import list_catalog, catalog_diagnostics entries = list_catalog() assert [e.name for e in entries] == ["demo"] diags = catalog_diagnostics() # At least one future_manifest diagnostic for the futuristic entry future = [d for d in diags if d[1] == "future_manifest"] assert len(future) == 1 assert future[0][0] == "futuristic" def test_invalid_manifest_diagnostic(self, catalog_dir): body = _basic_manifest() body["transport"] = {"type": "unsupported"} _write_manifest(catalog_dir, "broken", body) from hermes_cli.mcp_catalog import list_catalog, catalog_diagnostics entries = list_catalog() assert entries == [] diags = catalog_diagnostics() invalid = [d for d in diags if d[1] == "invalid"] assert len(invalid) == 1 def test_picker_surfaces_future_manifest_warning(self, catalog_dir, capsys, monkeypatch): """The text-dump path should print a warning line for future-manifest entries so users running headless or after `hermes setup` know to update.""" body = _basic_manifest() body["manifest_version"] = 999 _write_manifest(catalog_dir, "futuristic", body) _write_manifest(catalog_dir, "demo", _basic_manifest()) import sys as _sys monkeypatch.setattr(_sys.stdin, "isatty", lambda: False) from hermes_cli.mcp_picker import show_catalog show_catalog() out = capsys.readouterr().out assert "futuristic" in out assert "requires a newer Hermes" in out # --------------------------------------------------------------------------- # Picker — custom (non-catalog) MCP rows # --------------------------------------------------------------------------- class TestCustomMcpRows: def test_custom_mcp_shown_alongside_catalog(self, catalog_dir, capsys): """Servers in mcp_servers that aren't in the catalog show up in the picker text dump with a 'custom' status.""" _write_manifest(catalog_dir, "demo", _basic_manifest()) from hermes_cli.config import load_config, save_config cfg = load_config() cfg.setdefault("mcp_servers", {})["my-custom"] = { "command": "npx", "args": ["-y", "my-custom-mcp"], "enabled": True, } save_config(cfg) from hermes_cli.mcp_picker import show_catalog show_catalog() out = capsys.readouterr().out assert "demo" in out assert "my-custom" in out assert "custom" in out # The status badge def test_custom_mcp_only_no_catalog(self, catalog_dir, capsys): """If the catalog is empty but the user has custom MCPs, they\'re still visible — the picker is the unified surface.""" from hermes_cli.config import load_config, save_config cfg = load_config() cfg.setdefault("mcp_servers", {})["my-custom"] = { "url": "https://mcp.example.com", "enabled": False, } save_config(cfg) from hermes_cli.mcp_picker import show_catalog show_catalog() out = capsys.readouterr().out assert "my-custom" in out # --------------------------------------------------------------------------- # Git install — SHA ref detection # --------------------------------------------------------------------------- class TestGitInstallShaRef: def test_sha_ref_skips_branch_attempt(self, catalog_dir, monkeypatch, tmp_path): """When install.ref is a SHA-shaped hex string, _do_git_install skips the `git clone --branch ` attempt (which would always fail noisily for SHAs) and goes straight to clone + checkout.""" body = _basic_manifest( install={ "type": "git", "url": "https://example.com/x.git", "ref": "abc1234567890abcdef1234567890abcdef12345", # 40-char SHA "bootstrap": [], }, transport={ "type": "stdio", "command": "${INSTALL_DIR}/run.sh", "args": [], }, ) _write_manifest(catalog_dir, "demo", body) from hermes_cli import mcp_catalog from hermes_cli.mcp_catalog import _do_git_install calls = [] class _FakeProc: def __init__(self, returncode): self.returncode = returncode def fake_run(argv, *args, **kwargs): calls.append(list(argv)) # Make every command succeed return _FakeProc(returncode=0) monkeypatch.setattr(mcp_catalog.subprocess, "run", fake_run) monkeypatch.setattr(mcp_catalog.shutil, "which", lambda x: "/usr/bin/git") from hermes_cli.mcp_catalog import get_entry entry = get_entry("demo") assert entry is not None _do_git_install(entry) # Should have called clone (no --branch) then checkout — NOT clone --branch branch_attempts = [c for c in calls if "--branch" in c] assert branch_attempts == [], ( "SHA refs must NOT trigger a --branch clone attempt — that would " "always fail noisily before falling back. Calls were: " + repr(calls) ) # Confirm we DID do plain clone + checkout clone_calls = [c for c in calls if "clone" in c and "--branch" not in c] checkout_calls = [c for c in calls if "checkout" in c] assert len(clone_calls) == 1, calls assert len(checkout_calls) == 1, calls def test_branch_ref_uses_branch_clone(self, catalog_dir, monkeypatch): """When install.ref is a branch/tag (not SHA-shaped), the fast `git clone --depth 1 --branch ` path is used.""" body = _basic_manifest( install={ "type": "git", "url": "https://example.com/x.git", "ref": "v1.0.0", # Tag-shaped "bootstrap": [], }, transport={ "type": "stdio", "command": "${INSTALL_DIR}/run.sh", "args": [], }, ) _write_manifest(catalog_dir, "demo", body) from hermes_cli import mcp_catalog from hermes_cli.mcp_catalog import _do_git_install, get_entry calls = [] class _FakeProc: def __init__(self, returncode): self.returncode = returncode def fake_run(argv, *args, **kwargs): calls.append(list(argv)) return _FakeProc(returncode=0) monkeypatch.setattr(mcp_catalog.subprocess, "run", fake_run) monkeypatch.setattr(mcp_catalog.shutil, "which", lambda x: "/usr/bin/git") _do_git_install(get_entry("demo")) branch_attempts = [c for c in calls if "--branch" in c] assert len(branch_attempts) == 1, calls # --------------------------------------------------------------------------- # Existing tools_config converged to tools.include # --------------------------------------------------------------------------- class TestToolsConfigIncludeMode: def test_configure_mcp_writes_include_not_exclude(self, monkeypatch, tmp_path): """`_configure_mcp_tools_interactive` in tools_config.py must write `tools.include` (whitelist), matching the rest of the codebase. The old behavior wrote `tools.exclude`, which produced inconsistent on-disk shapes depending on which UI the user used last.""" # Build a minimal mcp_servers config + mock probe + checklist cfg = { "_config_version": 23, "mcp_servers": { "demo": { "command": "npx", "args": ["-y", "demo-mcp"], "enabled": True, } }, } import hermes_cli.tools_config as tc # Mock the probe to return three tools monkeypatch.setattr( "tools.mcp_tool.probe_mcp_server_tools", lambda: {"demo": [("a", "desc"), ("b", "desc"), ("c", "desc")]}, ) # Mock the checklist to return just the first tool monkeypatch.setattr( "hermes_cli.curses_ui.curses_checklist", lambda title, labels, pre_selected, **kw: {0}, ) # Mock save_config so we can inspect the write saved = {} def fake_save(config): saved.update(config) monkeypatch.setattr(tc, "save_config", fake_save) tc._configure_mcp_tools_interactive(cfg) # Must have written include, not exclude srv = saved["mcp_servers"]["demo"]["tools"] assert srv.get("include") == ["a"], srv assert "exclude" not in srv, srv class TestShippedCatalog: def test_all_shipped_manifests_parse(self, monkeypatch): """Every manifest in optional-mcps/ must parse cleanly. This is a contract test — CI will fail if a PR adds a malformed manifest. Intentionally NOT a snapshot of catalog names (those are expected to change as PRs land). """ # Use the actual repo's optional-mcps directory (no HERMES_OPTIONAL_MCPS # override) so this test catches real manifests. monkeypatch.delenv("HERMES_OPTIONAL_MCPS", raising=False) from hermes_cli.mcp_catalog import _catalog_root, _parse_manifest root = _catalog_root() if not root.exists(): pytest.skip("optional-mcps/ not present in this checkout") manifests = list(root.glob("*/manifest.yaml")) # Don't assert minimum count — change-detector test rule. Just parse # whatever exists. for m in manifests: entry = _parse_manifest(m) assert entry.name assert entry.description assert entry.transport.type in ("stdio", "http")