mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
e53e8a782c
Server names with non-env-safe characters (dots, slashes, spaces)
produced invalid env-var keys like MCP_MY.SERVER_API_KEY or
MCP_GITHUB/MCP_API_KEY, breaking .env writes and ${VAR} header
substitution. _env_key_for_server now replaces any character outside
[A-Za-z0-9_] with an underscore.
Co-authored-by: Hermes Agent <agent@nousresearch.com>
1013 lines
36 KiB
Python
1013 lines
36 KiB
Python
"""
|
|
Tests for hermes_cli.mcp_config — ``hermes mcp`` subcommands.
|
|
|
|
These tests mock the MCP server connection layer so they run without
|
|
any actual MCP servers or API keys.
|
|
"""
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def _set_interactive_stdin(monkeypatch, *, is_tty: bool = True) -> None:
|
|
from unittest.mock import MagicMock
|
|
|
|
mock_stdin = MagicMock()
|
|
mock_stdin.isatty.return_value = is_tty
|
|
monkeypatch.setattr("tools.mcp_oauth.sys.stdin", mock_stdin)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_config(tmp_path, monkeypatch):
|
|
"""Redirect all config I/O to a temp directory."""
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.get_hermes_home", lambda: tmp_path
|
|
)
|
|
config_path = tmp_path / "config.yaml"
|
|
env_path = tmp_path / ".env"
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.get_config_path", lambda: config_path
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.get_env_path", lambda: env_path
|
|
)
|
|
return tmp_path
|
|
|
|
|
|
def _make_args(**kwargs):
|
|
"""Build a minimal argparse.Namespace."""
|
|
defaults = {
|
|
"name": "test-server",
|
|
"url": None,
|
|
"mcp_command": None,
|
|
"args": None,
|
|
"auth": None,
|
|
"preset": None,
|
|
"env": None,
|
|
"mcp_action": None,
|
|
}
|
|
defaults.update(kwargs)
|
|
return argparse.Namespace(**defaults)
|
|
|
|
|
|
def _seed_config(tmp_path: Path, mcp_servers: dict):
|
|
"""Write a config.yaml with the given mcp_servers."""
|
|
import yaml
|
|
|
|
config = {"mcp_servers": mcp_servers, "_config_version": 9}
|
|
config_path = tmp_path / "config.yaml"
|
|
with open(config_path, "w") as f:
|
|
yaml.safe_dump(config, f)
|
|
|
|
|
|
class FakeTool:
|
|
"""Mimics an MCP tool object returned by the SDK."""
|
|
|
|
def __init__(self, name: str, description: str = ""):
|
|
self.name = name
|
|
self.description = description
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: cmd_mcp_list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMcpList:
|
|
def test_list_empty_config(self, tmp_path, capsys):
|
|
from hermes_cli.mcp_config import cmd_mcp_list
|
|
|
|
cmd_mcp_list()
|
|
out = capsys.readouterr().out
|
|
assert "No MCP servers configured" in out
|
|
|
|
def test_list_with_servers(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {
|
|
"ink": {
|
|
"url": "https://mcp.ml.ink/mcp",
|
|
"enabled": True,
|
|
"tools": {"include": ["create_service", "get_service"]},
|
|
},
|
|
"github": {
|
|
"command": "npx",
|
|
"args": ["@mcp/github"],
|
|
"enabled": False,
|
|
},
|
|
})
|
|
from hermes_cli.mcp_config import cmd_mcp_list
|
|
|
|
cmd_mcp_list()
|
|
out = capsys.readouterr().out
|
|
assert "ink" in out
|
|
assert "github" in out
|
|
assert "2 selected" in out # ink has 2 in include
|
|
assert "disabled" in out # github is disabled
|
|
|
|
def test_list_enabled_default_true(self, tmp_path, capsys):
|
|
"""Server without explicit enabled key defaults to enabled."""
|
|
_seed_config(tmp_path, {
|
|
"myserver": {"url": "https://example.com/mcp"},
|
|
})
|
|
from hermes_cli.mcp_config import cmd_mcp_list
|
|
|
|
cmd_mcp_list()
|
|
out = capsys.readouterr().out
|
|
assert "myserver" in out
|
|
assert "enabled" in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: cmd_mcp_remove
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMcpRemove:
|
|
def test_remove_existing_server(self, tmp_path, capsys, monkeypatch):
|
|
_seed_config(tmp_path, {
|
|
"myserver": {"url": "https://example.com/mcp"},
|
|
})
|
|
monkeypatch.setattr("builtins.input", lambda _: "y")
|
|
from hermes_cli.mcp_config import cmd_mcp_remove
|
|
|
|
cmd_mcp_remove(_make_args(name="myserver"))
|
|
|
|
out = capsys.readouterr().out
|
|
assert "Removed" in out
|
|
|
|
# Verify config updated
|
|
from hermes_cli.config import load_config
|
|
|
|
config = load_config()
|
|
assert "myserver" not in config.get("mcp_servers", {})
|
|
|
|
def test_remove_nonexistent(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {})
|
|
from hermes_cli.mcp_config import cmd_mcp_remove
|
|
|
|
cmd_mcp_remove(_make_args(name="ghost"))
|
|
out = capsys.readouterr().out
|
|
assert "not found" in out
|
|
|
|
def test_remove_cleans_oauth_tokens(self, tmp_path, capsys, monkeypatch):
|
|
_seed_config(tmp_path, {
|
|
"oauth-srv": {"url": "https://example.com/mcp", "auth": "oauth"},
|
|
})
|
|
monkeypatch.setattr("builtins.input", lambda _: "y")
|
|
# Also patch get_hermes_home in the mcp_config module namespace
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config.get_hermes_home", lambda: tmp_path
|
|
)
|
|
|
|
# Create a fake token file
|
|
token_dir = tmp_path / "mcp-tokens"
|
|
token_dir.mkdir()
|
|
token_file = token_dir / "oauth-srv.json"
|
|
token_file.write_text("{}")
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_remove
|
|
|
|
cmd_mcp_remove(_make_args(name="oauth-srv"))
|
|
assert not token_file.exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: cmd_mcp_add
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMcpAdd:
|
|
def test_add_no_transport(self, capsys):
|
|
"""Must specify --url or --command."""
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(name="bad"))
|
|
out = capsys.readouterr().out
|
|
assert "Must specify" in out
|
|
|
|
def test_add_http_server_all_tools(self, tmp_path, capsys, monkeypatch):
|
|
"""Add an HTTP server, accept all tools."""
|
|
fake_tools = [
|
|
FakeTool("create_service", "Deploy from repo"),
|
|
FakeTool("list_services", "List all services"),
|
|
]
|
|
|
|
def mock_probe(name, config, **kw):
|
|
return [(t.name, t.description) for t in fake_tools]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
# No auth, accept all tools
|
|
inputs = iter(["n", ""]) # no auth needed, enable all
|
|
monkeypatch.setattr("builtins.input", lambda _: next(inputs))
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(name="ink", url="https://mcp.ml.ink/mcp"))
|
|
out = capsys.readouterr().out
|
|
assert "Saved" in out
|
|
assert "2/2 tools" in out
|
|
|
|
# Verify config written
|
|
from hermes_cli.config import load_config
|
|
|
|
config = load_config()
|
|
assert "ink" in config.get("mcp_servers", {})
|
|
assert config["mcp_servers"]["ink"]["url"] == "https://mcp.ml.ink/mcp"
|
|
|
|
def test_add_stdio_server(self, tmp_path, capsys, monkeypatch):
|
|
"""Add a stdio server."""
|
|
fake_tools = [FakeTool("search", "Search repos")]
|
|
|
|
def mock_probe(name, config, **kw):
|
|
return [(t.name, t.description) for t in fake_tools]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
inputs = iter([""]) # accept all tools
|
|
monkeypatch.setattr("builtins.input", lambda _: next(inputs))
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(
|
|
name="github",
|
|
mcp_command="npx",
|
|
args=["@mcp/github"],
|
|
))
|
|
out = capsys.readouterr().out
|
|
assert "Saved" in out
|
|
|
|
from hermes_cli.config import load_config
|
|
|
|
config = load_config()
|
|
srv = config["mcp_servers"]["github"]
|
|
assert srv["command"] == "npx"
|
|
assert srv["args"] == ["@mcp/github"]
|
|
|
|
def test_add_connection_failure_save_disabled(
|
|
self, tmp_path, capsys, monkeypatch
|
|
):
|
|
"""Failed connection → option to save as disabled."""
|
|
|
|
def mock_probe_fail(name, config, **kw):
|
|
raise ConnectionError("Connection refused")
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe_fail
|
|
)
|
|
inputs = iter(["n", "y"]) # no auth, yes save disabled
|
|
monkeypatch.setattr("builtins.input", lambda _: next(inputs))
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(name="broken", url="https://bad.host/mcp"))
|
|
out = capsys.readouterr().out
|
|
assert "disabled" in out
|
|
|
|
from hermes_cli.config import load_config
|
|
|
|
config = load_config()
|
|
assert config["mcp_servers"]["broken"]["enabled"] is False
|
|
|
|
def test_add_stdio_server_with_env(self, tmp_path, capsys, monkeypatch):
|
|
"""Stdio servers can persist explicit environment variables."""
|
|
fake_tools = [FakeTool("search", "Search repos")]
|
|
|
|
def mock_probe(name, config, **kw):
|
|
assert config["env"] == {
|
|
"MY_API_KEY": "secret123",
|
|
"DEBUG": "true",
|
|
}
|
|
return [(t.name, t.description) for t in fake_tools]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
monkeypatch.setattr("builtins.input", lambda _: "")
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(
|
|
name="github",
|
|
mcp_command="npx",
|
|
args=["@mcp/github"],
|
|
env=["MY_API_KEY=secret123", "DEBUG=true"],
|
|
))
|
|
out = capsys.readouterr().out
|
|
assert "Saved" in out
|
|
|
|
from hermes_cli.config import load_config
|
|
|
|
config = load_config()
|
|
srv = config["mcp_servers"]["github"]
|
|
assert srv["env"] == {
|
|
"MY_API_KEY": "secret123",
|
|
"DEBUG": "true",
|
|
}
|
|
|
|
def test_add_stdio_server_rejects_invalid_env_name(self, capsys):
|
|
"""Invalid environment variable names are rejected up front."""
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(
|
|
name="github",
|
|
mcp_command="npx",
|
|
args=["@mcp/github"],
|
|
env=["BAD-NAME=value"],
|
|
))
|
|
out = capsys.readouterr().out
|
|
assert "Invalid --env variable name" in out
|
|
|
|
def test_add_http_server_rejects_env_flag(self, capsys):
|
|
"""The --env flag is only valid for stdio transports."""
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(
|
|
name="ink",
|
|
url="https://mcp.ml.ink/mcp",
|
|
env=["DEBUG=true"],
|
|
))
|
|
out = capsys.readouterr().out
|
|
assert "only supported for stdio MCP servers" in out
|
|
|
|
def test_add_preset_fills_transport(self, tmp_path, capsys, monkeypatch):
|
|
"""A preset fills in command/args when no explicit transport given."""
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._MCP_PRESETS",
|
|
{"testmcp": {"command": "npx", "args": ["-y", "test-mcp-server"], "display_name": "Test MCP"}},
|
|
)
|
|
fake_tools = [FakeTool("do_thing", "Does a thing")]
|
|
|
|
def mock_probe(name, config, **kw):
|
|
assert name == "myserver"
|
|
assert config["command"] == "npx"
|
|
assert config["args"] == ["-y", "test-mcp-server"]
|
|
assert "env" not in config
|
|
return [(t.name, t.description) for t in fake_tools]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
monkeypatch.setattr("builtins.input", lambda _: "")
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
from hermes_cli.config import read_raw_config
|
|
|
|
cmd_mcp_add(_make_args(name="myserver", preset="testmcp"))
|
|
out = capsys.readouterr().out
|
|
assert "Saved" in out
|
|
|
|
config = read_raw_config()
|
|
srv = config["mcp_servers"]["myserver"]
|
|
assert srv["command"] == "npx"
|
|
assert srv["args"] == ["-y", "test-mcp-server"]
|
|
assert "env" not in srv
|
|
|
|
def test_preset_does_not_override_explicit_command(self, tmp_path, capsys, monkeypatch):
|
|
"""Explicit transports win over presets."""
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._MCP_PRESETS",
|
|
{"testmcp": {"command": "npx", "args": ["-y", "test-mcp-server"], "display_name": "Test MCP"}},
|
|
)
|
|
fake_tools = [FakeTool("search", "Search repos")]
|
|
|
|
def mock_probe(name, config, **kw):
|
|
assert config["command"] == "uvx"
|
|
assert config["args"] == ["custom-server"]
|
|
assert "env" not in config
|
|
return [(t.name, t.description) for t in fake_tools]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
monkeypatch.setattr("builtins.input", lambda _: "")
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
from hermes_cli.config import read_raw_config
|
|
|
|
cmd_mcp_add(_make_args(
|
|
name="custom",
|
|
preset="testmcp",
|
|
mcp_command="uvx",
|
|
args=["custom-server"],
|
|
))
|
|
out = capsys.readouterr().out
|
|
assert "Saved" in out
|
|
|
|
config = read_raw_config()
|
|
srv = config["mcp_servers"]["custom"]
|
|
assert srv["command"] == "uvx"
|
|
assert srv["args"] == ["custom-server"]
|
|
assert "env" not in srv
|
|
|
|
def test_unknown_preset_rejected(self, capsys):
|
|
"""An unknown preset name is rejected with a clear error."""
|
|
from hermes_cli.mcp_config import cmd_mcp_add
|
|
|
|
cmd_mcp_add(_make_args(name="foo", preset="nonexistent"))
|
|
out = capsys.readouterr().out
|
|
assert "Unknown MCP preset" in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: cmd_mcp_test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMcpTest:
|
|
def test_test_not_found(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {})
|
|
from hermes_cli.mcp_config import cmd_mcp_test
|
|
|
|
cmd_mcp_test(_make_args(name="ghost"))
|
|
out = capsys.readouterr().out
|
|
assert "not found" in out
|
|
|
|
def test_test_success(self, tmp_path, capsys, monkeypatch):
|
|
_seed_config(tmp_path, {
|
|
"ink": {"url": "https://mcp.ml.ink/mcp"},
|
|
})
|
|
|
|
def mock_probe(name, config, **kw):
|
|
return [("create_service", "Deploy"), ("list_services", "List all")]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
from hermes_cli.mcp_config import cmd_mcp_test
|
|
|
|
cmd_mcp_test(_make_args(name="ink"))
|
|
out = capsys.readouterr().out
|
|
assert "Connected" in out
|
|
assert "Tools discovered: 2" in out
|
|
|
|
def test_probe_uses_configured_connect_timeout(self, monkeypatch):
|
|
"""OAuth-capable probes must not hard-code a short 30s timeout."""
|
|
import asyncio
|
|
from hermes_cli import mcp_config
|
|
import tools.mcp_tool as mcp_tool
|
|
|
|
captured = {}
|
|
|
|
class FakeServer:
|
|
_tools = []
|
|
|
|
async def shutdown(self):
|
|
captured["shutdown"] = True
|
|
|
|
async def fake_connect(name, config):
|
|
return FakeServer()
|
|
|
|
def fake_run_on_mcp_loop(coro, timeout):
|
|
captured["outer_timeout"] = timeout
|
|
return asyncio.run(coro)
|
|
|
|
async def fake_wait_for(awaitable, timeout):
|
|
captured["inner_timeout"] = timeout
|
|
return await awaitable
|
|
|
|
monkeypatch.setattr(mcp_tool, "_ensure_mcp_loop", lambda: None)
|
|
monkeypatch.setattr(mcp_tool, "_stop_mcp_loop_if_idle", lambda: None)
|
|
monkeypatch.setattr(mcp_tool, "_connect_server", fake_connect)
|
|
monkeypatch.setattr(mcp_tool, "_run_on_mcp_loop", fake_run_on_mcp_loop)
|
|
monkeypatch.setattr(mcp_config.asyncio, "wait_for", fake_wait_for)
|
|
|
|
assert mcp_config._probe_single_server(
|
|
"supabase", {"connect_timeout": 300}
|
|
) == []
|
|
assert captured["inner_timeout"] == 300.0
|
|
assert captured["outer_timeout"] == 310.0
|
|
assert captured["shutdown"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: env var interpolation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEnvVarInterpolation:
|
|
def test_interpolate_simple(self, monkeypatch):
|
|
monkeypatch.setenv("MY_KEY", "secret123")
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
result = _interpolate_env_vars("Bearer ${MY_KEY}")
|
|
assert result == "Bearer secret123"
|
|
|
|
def test_interpolate_missing_var(self, monkeypatch):
|
|
monkeypatch.delenv("MISSING_VAR", raising=False)
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
result = _interpolate_env_vars("Bearer ${MISSING_VAR}")
|
|
assert result == "Bearer ${MISSING_VAR}"
|
|
|
|
def test_interpolate_nested_dict(self, monkeypatch):
|
|
monkeypatch.setenv("API_KEY", "abc")
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
result = _interpolate_env_vars({
|
|
"url": "https://example.com",
|
|
"headers": {"Authorization": "Bearer ${API_KEY}"},
|
|
})
|
|
assert result["headers"]["Authorization"] == "Bearer abc"
|
|
assert result["url"] == "https://example.com"
|
|
|
|
def test_interpolate_list(self, monkeypatch):
|
|
monkeypatch.setenv("ARG1", "hello")
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
result = _interpolate_env_vars(["${ARG1}", "static"])
|
|
assert result == ["hello", "static"]
|
|
|
|
def test_interpolate_non_string(self):
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
assert _interpolate_env_vars(42) == 42
|
|
assert _interpolate_env_vars(True) is True
|
|
assert _interpolate_env_vars(None) is None
|
|
|
|
def test_interpolate_cursor_env_prefix(self, monkeypatch):
|
|
"""Cursor-style ${env:VAR} resolves the same secret as ${VAR}."""
|
|
monkeypatch.setenv("MY_KEY", "secret123")
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
assert _interpolate_env_vars("Bearer ${env:MY_KEY}") == "Bearer secret123"
|
|
|
|
def test_interpolate_cursor_env_prefix_missing(self, monkeypatch):
|
|
"""An unset ${env:VAR} keeps its literal placeholder, like ${VAR}."""
|
|
monkeypatch.delenv("MISSING_VAR", raising=False)
|
|
from tools.mcp_tool import _interpolate_env_vars
|
|
|
|
assert _interpolate_env_vars("Bearer ${env:MISSING_VAR}") == "Bearer ${env:MISSING_VAR}"
|
|
|
|
def test_env_ref_name_strips_prefix(self):
|
|
from tools.mcp_tool import _env_ref_name
|
|
|
|
assert _env_ref_name("env:API_KEY") == "API_KEY"
|
|
assert _env_ref_name("API_KEY") == "API_KEY"
|
|
assert _env_ref_name(" env:API_KEY ") == "API_KEY"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: probe-path env resolution (#37792)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestProbeEnvResolution:
|
|
"""The probe path must resolve ``${ENV}`` before connecting, so the
|
|
discovery probe behaves like runtime tool loading. Regression for #37792
|
|
where `hermes mcp add --auth header` sent a literal
|
|
``Authorization: Bearer ${MCP_X_API_KEY}`` and got 401."""
|
|
|
|
def test_resolve_interpolates_header(self, monkeypatch):
|
|
from hermes_cli.mcp_config import _resolve_mcp_server_config
|
|
|
|
monkeypatch.setenv("MCP_N8N_API_KEY", "jwt-token-xyz")
|
|
resolved = _resolve_mcp_server_config({
|
|
"url": "http://localhost:5678/mcp-server/http",
|
|
"headers": {"Authorization": "Bearer ${MCP_N8N_API_KEY}"},
|
|
})
|
|
assert resolved["headers"]["Authorization"] == "Bearer jwt-token-xyz"
|
|
|
|
def test_resolve_leaves_unset_var_literal(self, monkeypatch):
|
|
from hermes_cli.mcp_config import _resolve_mcp_server_config
|
|
|
|
monkeypatch.delenv("MCP_UNSET_API_KEY", raising=False)
|
|
resolved = _resolve_mcp_server_config({
|
|
"headers": {"Authorization": "Bearer ${MCP_UNSET_API_KEY}"},
|
|
})
|
|
# Unresolved placeholder stays literal (no crash) — matches
|
|
# _interpolate_env_vars semantics.
|
|
assert resolved["headers"]["Authorization"] == "Bearer ${MCP_UNSET_API_KEY}"
|
|
|
|
def test_probe_resolves_before_connect(self, monkeypatch):
|
|
"""_probe_single_server must pass the RESOLVED config to _connect_server."""
|
|
import hermes_cli.mcp_config as mc
|
|
|
|
monkeypatch.setenv("MCP_N8N_API_KEY", "jwt-token-xyz")
|
|
|
|
seen = {}
|
|
|
|
class _FakeTool:
|
|
name = "do_thing"
|
|
description = "a tool"
|
|
|
|
class _FakeServer:
|
|
_tools = [_FakeTool()]
|
|
|
|
async def shutdown(self):
|
|
return None
|
|
|
|
async def _fake_connect(name, config):
|
|
seen["config"] = config
|
|
return _FakeServer()
|
|
|
|
monkeypatch.setattr("tools.mcp_tool._connect_server", _fake_connect)
|
|
|
|
tools = mc._probe_single_server("n8n", {
|
|
"url": "http://localhost:5678/mcp-server/http",
|
|
"headers": {"Authorization": "Bearer ${MCP_N8N_API_KEY}"},
|
|
})
|
|
|
|
assert tools == [("do_thing", "a tool")]
|
|
assert seen["config"]["headers"]["Authorization"] == "Bearer jwt-token-xyz"
|
|
|
|
|
|
class TestProbeCapabilityGating:
|
|
"""The ``details`` probe must not fire prompts/list or resources/list at
|
|
servers that either disabled them in config or never advertised them.
|
|
|
|
Regression for the Unreal MCP server case: it answers
|
|
``Call to unknown method "prompts/list"``, so an unconditional probe logged
|
|
a hard error and ``tools.prompts: false`` (the documented workaround) had no
|
|
effect because the probe never consulted config or capabilities.
|
|
"""
|
|
|
|
class _FakeTool:
|
|
name = "do_thing"
|
|
description = "a tool"
|
|
|
|
class _Caps:
|
|
def __init__(self, prompts=None, resources=None):
|
|
self.prompts = prompts
|
|
self.resources = resources
|
|
|
|
class _InitResult:
|
|
def __init__(self, caps):
|
|
self.capabilities = caps
|
|
|
|
def _make_server(self, called, caps):
|
|
outer = self
|
|
|
|
class _Result(list):
|
|
@property
|
|
def prompts(self):
|
|
return self
|
|
|
|
@property
|
|
def resources(self):
|
|
return self
|
|
|
|
class _Session:
|
|
async def list_prompts(self_inner):
|
|
called.append("prompts")
|
|
return _Result()
|
|
|
|
async def list_resources(self_inner):
|
|
called.append("resources")
|
|
return _Result()
|
|
|
|
class _FakeServer:
|
|
_tools = [outer._FakeTool()]
|
|
session = _Session()
|
|
initialize_result = outer._InitResult(caps)
|
|
|
|
async def shutdown(self_inner):
|
|
return None
|
|
|
|
return _FakeServer()
|
|
|
|
def _run_probe(self, monkeypatch, config, caps):
|
|
import hermes_cli.mcp_config as mc
|
|
|
|
called: list[str] = []
|
|
|
|
async def _fake_connect(name, cfg):
|
|
return self._make_server(called, caps)
|
|
|
|
monkeypatch.setattr("tools.mcp_tool._connect_server", _fake_connect)
|
|
details: dict = {}
|
|
mc._probe_single_server("srv", config, details=details)
|
|
return called, details
|
|
|
|
def test_config_disables_prompts_probe(self, monkeypatch):
|
|
# Server advertises both, but user turned prompts off.
|
|
caps = self._Caps(prompts=object(), resources=object())
|
|
called, details = self._run_probe(
|
|
monkeypatch, {"url": "http://x/mcp", "tools": {"prompts": False}}, caps
|
|
)
|
|
assert "prompts" not in called
|
|
assert "resources" in called
|
|
|
|
def test_unadvertised_capability_not_probed(self, monkeypatch):
|
|
# Unreal case: no prompts capability advertised → never call it.
|
|
caps = self._Caps(prompts=None, resources=None)
|
|
called, _ = self._run_probe(monkeypatch, {"url": "http://x/mcp"}, caps)
|
|
assert called == []
|
|
|
|
def test_advertised_and_enabled_is_probed(self, monkeypatch):
|
|
caps = self._Caps(prompts=object(), resources=object())
|
|
called, details = self._run_probe(monkeypatch, {"url": "http://x/mcp"}, caps)
|
|
assert set(called) == {"prompts", "resources"}
|
|
|
|
def test_missing_capability_info_falls_back_to_probe(self, monkeypatch):
|
|
# No initialize_result captured → preserve legacy always-try behaviour.
|
|
called, _ = self._run_probe(monkeypatch, {"url": "http://x/mcp"}, None)
|
|
assert set(called) == {"prompts", "resources"}
|
|
|
|
|
|
class TestStripBearerPrefix:
|
|
"""Pasted tokens that already include ``Bearer `` would otherwise produce
|
|
``Bearer Bearer <jwt>`` once the header template adds its own prefix."""
|
|
|
|
def test_bare_token_unchanged(self):
|
|
from hermes_cli.mcp_config import _strip_bearer_prefix
|
|
|
|
assert _strip_bearer_prefix("eyJabc123") == "eyJabc123"
|
|
|
|
def test_strips_bearer_prefix(self):
|
|
from hermes_cli.mcp_config import _strip_bearer_prefix
|
|
|
|
assert _strip_bearer_prefix("Bearer eyJabc123") == "eyJabc123"
|
|
|
|
def test_strips_case_insensitive_and_whitespace(self):
|
|
from hermes_cli.mcp_config import _strip_bearer_prefix
|
|
|
|
assert _strip_bearer_prefix("bearer eyJabc123") == "eyJabc123"
|
|
assert _strip_bearer_prefix(" Bearer eyJabc123 ") == "eyJabc123"
|
|
|
|
def test_does_not_strip_without_space(self):
|
|
from hermes_cli.mcp_config import _strip_bearer_prefix
|
|
|
|
# "BearerToken" is a token that happens to start with "Bearer", not a prefix.
|
|
assert _strip_bearer_prefix("BearerToken") == "BearerToken"
|
|
|
|
def test_non_string_passthrough(self):
|
|
from hermes_cli.mcp_config import _strip_bearer_prefix
|
|
|
|
assert _strip_bearer_prefix(None) is None # type: ignore[arg-type]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: config helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestConfigHelpers:
|
|
def test_save_and_load_mcp_server(self, tmp_path):
|
|
from hermes_cli.mcp_config import _save_mcp_server, _get_mcp_servers
|
|
|
|
_save_mcp_server("mysvr", {"url": "https://example.com/mcp"})
|
|
servers = _get_mcp_servers()
|
|
assert "mysvr" in servers
|
|
assert servers["mysvr"]["url"] == "https://example.com/mcp"
|
|
|
|
def test_remove_mcp_server(self, tmp_path):
|
|
from hermes_cli.mcp_config import (
|
|
_save_mcp_server,
|
|
_remove_mcp_server,
|
|
_get_mcp_servers,
|
|
)
|
|
|
|
_save_mcp_server("s1", {"command": "test"})
|
|
_save_mcp_server("s2", {"command": "test2"})
|
|
result = _remove_mcp_server("s1")
|
|
assert result is True
|
|
assert "s1" not in _get_mcp_servers()
|
|
assert "s2" in _get_mcp_servers()
|
|
|
|
def test_remove_nonexistent(self, tmp_path):
|
|
from hermes_cli.mcp_config import _remove_mcp_server
|
|
|
|
assert _remove_mcp_server("ghost") is False
|
|
|
|
def test_env_key_for_server(self):
|
|
from hermes_cli.mcp_config import _env_key_for_server
|
|
|
|
assert _env_key_for_server("ink") == "MCP_INK_API_KEY"
|
|
assert _env_key_for_server("my-server") == "MCP_MY_SERVER_API_KEY"
|
|
assert _env_key_for_server("my.server") == "MCP_MY_SERVER_API_KEY"
|
|
assert _env_key_for_server("github/mcp") == "MCP_GITHUB_MCP_API_KEY"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDispatcher:
|
|
def test_no_action_shows_list(self, tmp_path, capsys):
|
|
from hermes_cli.mcp_config import mcp_command
|
|
|
|
_seed_config(tmp_path, {})
|
|
mcp_command(_make_args(mcp_action=None))
|
|
out = capsys.readouterr().out
|
|
assert "Commands:" in out or "No MCP servers" in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Task 7 consolidation — cmd_mcp_remove evicts manager cache,
|
|
# cmd_mcp_login forces re-auth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMcpRemoveEvictsManager:
|
|
def test_remove_evicts_in_memory_provider(self, tmp_path, capsys, monkeypatch):
|
|
"""After cmd_mcp_remove, the MCPOAuthManager no longer caches the provider."""
|
|
_seed_config(tmp_path, {
|
|
"oauth-srv": {"url": "https://example.com/mcp", "auth": "oauth"},
|
|
})
|
|
monkeypatch.setattr("builtins.input", lambda _: "y")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config.get_hermes_home", lambda: tmp_path
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
_set_interactive_stdin(monkeypatch)
|
|
|
|
from tools.mcp_oauth_manager import get_manager, reset_manager_for_tests
|
|
reset_manager_for_tests()
|
|
|
|
mgr = get_manager()
|
|
mgr.get_or_build_provider(
|
|
"oauth-srv", "https://example.com/mcp", None,
|
|
)
|
|
assert "oauth-srv" in mgr._entries
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_remove
|
|
cmd_mcp_remove(_make_args(name="oauth-srv"))
|
|
|
|
assert "oauth-srv" not in mgr._entries
|
|
|
|
|
|
class TestMcpLogin:
|
|
def test_login_rejects_unknown_server(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {})
|
|
from hermes_cli.mcp_config import cmd_mcp_login
|
|
cmd_mcp_login(_make_args(name="ghost"))
|
|
out = capsys.readouterr().out
|
|
assert "not found" in out
|
|
|
|
def test_login_rejects_non_oauth_server(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {
|
|
"srv": {"url": "https://example.com/mcp", "auth": "header"},
|
|
})
|
|
from hermes_cli.mcp_config import cmd_mcp_login
|
|
cmd_mcp_login(_make_args(name="srv"))
|
|
out = capsys.readouterr().out
|
|
assert "not configured for OAuth" in out
|
|
|
|
def test_login_rejects_stdio_server(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {
|
|
"srv": {"command": "npx", "args": ["some-server"]},
|
|
})
|
|
from hermes_cli.mcp_config import cmd_mcp_login
|
|
cmd_mcp_login(_make_args(name="srv"))
|
|
out = capsys.readouterr().out
|
|
assert "no URL" in out or "not an OAuth" in out
|
|
|
|
def test_login_false_success_no_token(self, tmp_path, capsys, monkeypatch):
|
|
"""Probe lists tools without auth (Google Drive), but no token landed.
|
|
|
|
The server allows tools/list without auth (DCR 400'd), so the probe
|
|
succeeds yet no OAuth token exists. Login must NOT claim success — it
|
|
should warn and point the user at pre-registered client_id config.
|
|
"""
|
|
_seed_config(tmp_path, {
|
|
"googledrive": {
|
|
"url": "https://drivemcp.googleapis.com/mcp/v1",
|
|
"auth": "oauth",
|
|
},
|
|
})
|
|
# Probe returns tools even though auth never completed.
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server",
|
|
lambda name, cfg, connect_timeout=30: [
|
|
("search_files", "d"), ("read_file_content", "d"),
|
|
],
|
|
)
|
|
# No token file is created → _oauth_tokens_present() returns False.
|
|
from hermes_cli.mcp_config import cmd_mcp_login
|
|
|
|
cmd_mcp_login(_make_args(name="googledrive"))
|
|
out = capsys.readouterr().out
|
|
|
|
assert "no OAuth token was obtained" in out
|
|
assert "Authenticated" not in out
|
|
assert "client_id" in out
|
|
|
|
def test_login_genuine_success_with_token(self, tmp_path, capsys, monkeypatch):
|
|
"""Probe lists tools AND a token exists → report real success."""
|
|
_seed_config(tmp_path, {
|
|
"realserver": {"url": "https://mcp.example.com/mcp", "auth": "oauth"},
|
|
})
|
|
token_dir = tmp_path / "mcp-tokens"
|
|
|
|
# cmd_mcp_login wipes tokens before probing, then the real OAuth flow
|
|
# writes a fresh token during the probe. Simulate that: the mocked
|
|
# probe drops a token file, mirroring a successful authorization.
|
|
seen = {}
|
|
|
|
def mock_probe(name, cfg, connect_timeout=30):
|
|
seen["connect_timeout"] = connect_timeout
|
|
token_dir.mkdir(exist_ok=True)
|
|
(token_dir / "realserver.json").write_text('{"access_token": "x"}')
|
|
return [("a", "d"), ("b", "d"), ("c", "d")]
|
|
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._probe_single_server", mock_probe
|
|
)
|
|
|
|
from hermes_cli.mcp_config import cmd_mcp_login
|
|
|
|
cmd_mcp_login(_make_args(name="realserver"))
|
|
out = capsys.readouterr().out
|
|
|
|
assert "Authenticated — 3 tool(s) available" in out
|
|
assert "no OAuth token" not in out
|
|
# The login path must grant a human enough time to finish the browser
|
|
# OAuth round-trip — far longer than the 30s probe default.
|
|
assert seen["connect_timeout"] >= 180
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: cmd_mcp_reauth (GH#36767)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMcpReauth:
|
|
def test_reauth_all_visits_only_oauth_servers_in_order(
|
|
self, tmp_path, capsys, monkeypatch
|
|
):
|
|
"""--all re-auths every oauth server (skipping non-oauth), serially."""
|
|
_seed_config(tmp_path, {
|
|
"gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
|
|
"jira": {"url": "https://jira.example.com/mcp", "auth": "oauth"},
|
|
"localstdio": {"command": "foo"}, # no url / no oauth → skipped
|
|
"apikey": {"url": "https://k.example.com/mcp", "headers": {"x": "y"}},
|
|
})
|
|
visited = []
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._reauth_oauth_server",
|
|
lambda name, cfg: visited.append(name) or True,
|
|
)
|
|
from hermes_cli.mcp_config import cmd_mcp_reauth
|
|
|
|
cmd_mcp_reauth(_make_args(name=None, all=True))
|
|
out = capsys.readouterr().out
|
|
|
|
assert visited == ["gh", "jira"]
|
|
assert "Re-authenticated 2/2 server(s)" in out
|
|
|
|
def test_reauth_all_reports_partial_failures(self, tmp_path, capsys, monkeypatch):
|
|
"""A server that fails to re-auth is counted but doesn't abort the rest."""
|
|
_seed_config(tmp_path, {
|
|
"a": {"url": "https://a.example.com/mcp", "auth": "oauth"},
|
|
"b": {"url": "https://b.example.com/mcp", "auth": "oauth"},
|
|
})
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._reauth_oauth_server",
|
|
lambda name, cfg: name == "a", # only 'a' succeeds
|
|
)
|
|
from hermes_cli.mcp_config import cmd_mcp_reauth
|
|
|
|
cmd_mcp_reauth(_make_args(name=None, all=True))
|
|
out = capsys.readouterr().out
|
|
|
|
assert "Re-authenticated 1/2 server(s)" in out
|
|
|
|
def test_reauth_all_no_oauth_servers(self, tmp_path, capsys, monkeypatch):
|
|
_seed_config(tmp_path, {"localstdio": {"command": "foo"}})
|
|
called = []
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._reauth_oauth_server",
|
|
lambda name, cfg: called.append(name) or True,
|
|
)
|
|
from hermes_cli.mcp_config import cmd_mcp_reauth
|
|
|
|
cmd_mcp_reauth(_make_args(name=None, all=True))
|
|
out = capsys.readouterr().out
|
|
|
|
assert "No OAuth-based MCP servers found" in out
|
|
assert called == []
|
|
|
|
def test_reauth_single_server(self, tmp_path, capsys, monkeypatch):
|
|
_seed_config(tmp_path, {
|
|
"gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
|
|
})
|
|
visited = []
|
|
monkeypatch.setattr(
|
|
"hermes_cli.mcp_config._reauth_oauth_server",
|
|
lambda name, cfg: visited.append(name) or True,
|
|
)
|
|
from hermes_cli.mcp_config import cmd_mcp_reauth
|
|
|
|
cmd_mcp_reauth(_make_args(name="gh", all=False))
|
|
assert visited == ["gh"]
|
|
|
|
def test_reauth_requires_name_or_all(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {
|
|
"gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
|
|
})
|
|
from hermes_cli.mcp_config import cmd_mcp_reauth
|
|
|
|
cmd_mcp_reauth(_make_args(name=None, all=False))
|
|
out = capsys.readouterr().out
|
|
assert "Specify a server name" in out
|
|
|
|
def test_reauth_unknown_server(self, tmp_path, capsys):
|
|
_seed_config(tmp_path, {
|
|
"gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
|
|
})
|
|
from hermes_cli.mcp_config import cmd_mcp_reauth
|
|
|
|
cmd_mcp_reauth(_make_args(name="ghost", all=False))
|
|
out = capsys.readouterr().out
|
|
assert "not found" in out
|