"""Tests for EvoScientist.mcp module.""" import textwrap from types import SimpleNamespace import pytest import yaml from EvoScientist.mcp.client import ( _build_connections, _filter_tools, _interpolate_env, _resolve_command, _route_tools, add_mcp_server, edit_mcp_server, load_mcp_config, parse_mcp_add_args, parse_mcp_edit_args, remove_mcp_server, ) # ---- _interpolate_env ---- class TestInterpolateEnv: def test_substitutes_env_var(self, monkeypatch): monkeypatch.setenv("MY_KEY", "secret123") assert _interpolate_env("Bearer ${MY_KEY}") == "Bearer secret123" def test_multiple_vars(self, monkeypatch): monkeypatch.setenv("HOST", "localhost") monkeypatch.setenv("PORT", "8080") assert _interpolate_env("${HOST}:${PORT}") == "localhost:8080" def test_missing_var_returns_empty(self, monkeypatch): monkeypatch.delenv("NONEXISTENT_VAR_XYZ", raising=False) assert _interpolate_env("${NONEXISTENT_VAR_XYZ}") == "" def test_no_vars_unchanged(self): assert _interpolate_env("plain text") == "plain text" def test_empty_string(self): assert _interpolate_env("") == "" # ---- load_mcp_config ---- @pytest.fixture def mcp_config_file(monkeypatch, tmp_path): """Point USER_MCP_CONFIG to a temp file for isolated testing.""" cfg = tmp_path / "mcp.yaml" monkeypatch.setattr("EvoScientist.mcp.client.USER_MCP_CONFIG", cfg) return cfg class TestLoadMcpConfig: def test_missing_file_returns_empty(self, mcp_config_file): # File doesn't exist yet assert load_mcp_config() == {} def test_valid_file_parses(self, mcp_config_file): mcp_config_file.write_text( textwrap.dedent("""\ my-server: transport: stdio command: echo args: ["hello"] """) ) result = load_mcp_config() assert "my-server" in result assert result["my-server"]["transport"] == "stdio" def test_empty_file_returns_empty(self, mcp_config_file): mcp_config_file.write_text("") assert load_mcp_config() == {} def test_comments_only_returns_empty(self, mcp_config_file): mcp_config_file.write_text("# just a comment\n# another comment\n") assert load_mcp_config() == {} def test_env_var_interpolation(self, mcp_config_file, monkeypatch): monkeypatch.setenv("TEST_TOKEN", "tok_abc") mcp_config_file.write_text( textwrap.dedent("""\ my-server: transport: http url: "http://localhost:8080/mcp" headers: Authorization: "Bearer ${TEST_TOKEN}" """) ) result = load_mcp_config() assert result["my-server"]["headers"]["Authorization"] == "Bearer tok_abc" # ---- _build_connections ---- # ---- _resolve_command ---- class TestResolveCommand: def test_absolute_path_returned_as_is(self, tmp_path): """Absolute paths are never modified, even if the file doesn't exist.""" fake = str(tmp_path / "mytool") assert _resolve_command(fake) == fake def test_found_on_path(self): """Commands found via shutil.which are returned as full paths.""" result = _resolve_command("python") assert result.endswith("python") or result.endswith("python3") assert result != "python" # resolved, not the bare name def test_found_in_python_bin(self, tmp_path, monkeypatch): """Falls back to sys.executable's directory when not in PATH.""" # Create a fake executable next to sys.executable bin_dir = tmp_path / "bin" bin_dir.mkdir() fake_exe = bin_dir / "my-mcp-tool" fake_exe.write_text("#!/bin/sh\n") fake_exe.chmod(0o755) monkeypatch.setattr("shutil.which", lambda _: None) monkeypatch.setattr( "EvoScientist.mcp.client.sys.executable", str(bin_dir / "python") ) assert _resolve_command("my-mcp-tool") == str(fake_exe) def test_not_found_returns_original(self, monkeypatch): """Returns the original command when not found anywhere (let OS report the error).""" monkeypatch.setattr("shutil.which", lambda _: None) monkeypatch.setattr( "EvoScientist.mcp.client.sys.executable", "/nonexistent/bin/python" ) assert _resolve_command("unknown-tool-xyz") == "unknown-tool-xyz" def test_build_connections_resolves_command(self, monkeypatch): """_build_connections uses _resolve_command so the full path appears in output.""" monkeypatch.setattr( "EvoScientist.mcp.client._resolve_command", lambda cmd: f"/resolved/{cmd}" ) config = {"srv": {"transport": "stdio", "command": "mytool", "args": []}} conns = _build_connections(config) assert conns["srv"]["command"] == "/resolved/mytool" class TestBuildConnections: def test_stdio_connection(self): config = { "fs": { "transport": "stdio", "command": "npx", "args": ["-y", "server"], } } conns = _build_connections(config) assert "fs" in conns assert conns["fs"]["transport"] == "stdio" assert conns["fs"]["command"].endswith("npx") assert conns["fs"]["args"] == ["-y", "server"] def test_stdio_with_env(self): config = { "fs": { "transport": "stdio", "command": "npx", "args": [], "env": {"FOO": "bar"}, } } conns = _build_connections(config) assert conns["fs"]["env"] == {"FOO": "bar"} def test_http_connection(self): config = { "api": { "transport": "http", "url": "http://localhost:8080/mcp", "headers": {"Authorization": "Bearer xxx"}, } } conns = _build_connections(config) assert conns["api"]["transport"] == "http" assert conns["api"]["url"] == "http://localhost:8080/mcp" assert conns["api"]["headers"]["Authorization"] == "Bearer xxx" def test_sse_connection(self): config = { "sse-srv": { "transport": "sse", "url": "http://localhost:9090/sse", } } conns = _build_connections(config) assert conns["sse-srv"]["transport"] == "sse" assert conns["sse-srv"]["url"] == "http://localhost:9090/sse" def test_websocket_connection(self): config = { "ws": { "transport": "websocket", "url": "ws://localhost:8765", } } conns = _build_connections(config) assert conns["ws"]["transport"] == "websocket" def test_unknown_transport_skipped(self): config = { "bad": { "transport": "carrier_pigeon", "url": "coo://rooftop", } } conns = _build_connections(config) assert conns == {} def test_mixed_transports(self): config = { "a": {"transport": "stdio", "command": "cmd", "args": []}, "b": {"transport": "http", "url": "http://x"}, "c": {"transport": "unknown"}, } conns = _build_connections(config) assert set(conns.keys()) == {"a", "b"} # ---- _filter_tools ---- def _make_tool(name: str): """Create a minimal mock tool with a .name attribute.""" return SimpleNamespace(name=name) class TestFilterTools: def test_none_allowlist_passes_all(self): tools = [_make_tool("a"), _make_tool("b"), _make_tool("c")] assert _filter_tools(tools, None) == tools def test_allowlist_filters(self): tools = [_make_tool("a"), _make_tool("b"), _make_tool("c")] result = _filter_tools(tools, ["a", "c"]) assert [t.name for t in result] == ["a", "c"] def test_empty_allowlist_filters_all(self): tools = [_make_tool("a"), _make_tool("b")] assert _filter_tools(tools, []) == [] def test_allowlist_with_nonexistent_name(self): tools = [_make_tool("a")] result = _filter_tools(tools, ["a", "nonexistent"]) assert [t.name for t in result] == ["a"] def test_empty_tools_list(self): assert _filter_tools([], ["a"]) == [] assert _filter_tools([], None) == [] # Wildcard tests def test_wildcard_star_suffix(self): """Test *_exa pattern matching.""" tools = [ _make_tool("web_search_exa"), _make_tool("get_code_context_exa"), _make_tool("company_research_exa"), _make_tool("unrelated_tool"), ] result = _filter_tools(tools, ["*_exa"]) assert [t.name for t in result] == [ "web_search_exa", "get_code_context_exa", "company_research_exa", ] def test_wildcard_star_prefix(self): """Test read_* pattern matching.""" tools = [ _make_tool("read_file"), _make_tool("read_directory"), _make_tool("read_link"), _make_tool("write_file"), ] result = _filter_tools(tools, ["read_*"]) assert [t.name for t in result] == [ "read_file", "read_directory", "read_link", ] def test_wildcard_star_middle(self): """Test pattern with * in the middle.""" tools = [ _make_tool("get_user_data"), _make_tool("get_admin_data"), _make_tool("get_file"), ] result = _filter_tools(tools, ["get_*_data"]) assert [t.name for t in result] == ["get_user_data", "get_admin_data"] def test_wildcard_star_only(self): """Test * matches everything.""" tools = [_make_tool("a"), _make_tool("b"), _make_tool("c")] result = _filter_tools(tools, ["*"]) assert [t.name for t in result] == ["a", "b", "c"] def test_wildcard_question_mark(self): """Test ? matches single character.""" tools = [ _make_tool("tool_1"), _make_tool("tool_2"), _make_tool("tool_10"), ] result = _filter_tools(tools, ["tool_?"]) assert [t.name for t in result] == ["tool_1", "tool_2"] def test_wildcard_character_class(self): """Test [seq] matches characters in sequence.""" tools = [ _make_tool("tool_a"), _make_tool("tool_b"), _make_tool("tool_c"), _make_tool("tool_d"), ] result = _filter_tools(tools, ["tool_[abc]"]) assert [t.name for t in result] == ["tool_a", "tool_b", "tool_c"] def test_wildcard_character_class_range(self): """Test [0-9] matches digit range.""" tools = [ _make_tool("tool_0"), _make_tool("tool_5"), _make_tool("tool_9"), _make_tool("tool_a"), ] result = _filter_tools(tools, ["tool_[0-9]"]) assert [t.name for t in result] == ["tool_0", "tool_5", "tool_9"] def test_wildcard_negated_character_class(self): """Test [!seq] matches characters not in sequence.""" tools = [ _make_tool("tool_a"), _make_tool("tool_b"), _make_tool("tool_1"), _make_tool("tool_2"), ] result = _filter_tools(tools, ["tool_[!0-9]"]) assert [t.name for t in result] == ["tool_a", "tool_b"] def test_wildcard_mixed_with_exact(self): """Test mixing wildcard and exact patterns.""" tools = [ _make_tool("web_search_exa"), _make_tool("get_code_context_exa"), _make_tool("specific_tool"), _make_tool("another_tool"), ] result = _filter_tools(tools, ["*_exa", "specific_tool"]) assert [t.name for t in result] == [ "web_search_exa", "get_code_context_exa", "specific_tool", ] def test_wildcard_multiple_patterns(self): """Test multiple wildcard patterns.""" tools = [ _make_tool("read_file"), _make_tool("write_file"), _make_tool("delete_file"), _make_tool("search_database"), ] result = _filter_tools(tools, ["read_*", "write_*"]) assert [t.name for t in result] == ["read_file", "write_file"] def test_wildcard_no_match(self): """Test wildcard pattern that doesn't match anything.""" tools = [_make_tool("foo"), _make_tool("bar")] result = _filter_tools(tools, ["baz_*"]) assert result == [] def test_wildcard_overlapping_patterns(self): """Test overlapping patterns don't duplicate results.""" tools = [_make_tool("tool_abc"), _make_tool("tool_xyz")] result = _filter_tools(tools, ["tool_*", "*_abc"]) # Should include each tool only once assert [t.name for t in result] == ["tool_abc", "tool_xyz"] def test_wildcard_complex_pattern(self): """Test complex wildcard pattern.""" tools = [ _make_tool("get_user_info_v1"), _make_tool("get_user_info_v2"), _make_tool("get_admin_info_v1"), _make_tool("set_user_info"), ] result = _filter_tools(tools, ["get_*_info_v?"]) assert [t.name for t in result] == [ "get_user_info_v1", "get_user_info_v2", "get_admin_info_v1", ] def test_exact_match_performance_path(self): """Test exact matching still works (fast path).""" # This test verifies backward compatibility with exact matching tools = [_make_tool("a"), _make_tool("b"), _make_tool("c")] result = _filter_tools(tools, ["a", "c"]) assert [t.name for t in result] == ["a", "c"] # ---- _route_tools ---- class TestRouteTools: def test_default_routes_to_main(self): config = {"srv": {"transport": "stdio"}} server_tools = {"srv": [_make_tool("x")]} result = _route_tools(config, server_tools) assert "main" in result assert [t.name for t in result["main"]] == ["x"] def test_expose_to_named_agent(self): config = {"srv": {"transport": "stdio", "expose_to": ["code-agent"]}} server_tools = {"srv": [_make_tool("x"), _make_tool("y")]} result = _route_tools(config, server_tools) assert "code-agent" in result assert "main" not in result assert [t.name for t in result["code-agent"]] == ["x", "y"] def test_expose_to_multiple_agents(self): config = {"srv": {"transport": "stdio", "expose_to": ["main", "code-agent"]}} server_tools = {"srv": [_make_tool("x")]} result = _route_tools(config, server_tools) assert [t.name for t in result["main"]] == ["x"] assert [t.name for t in result["code-agent"]] == ["x"] def test_tool_filter_applied(self): config = {"srv": {"transport": "stdio", "tools": ["b"]}} server_tools = {"srv": [_make_tool("a"), _make_tool("b"), _make_tool("c")]} result = _route_tools(config, server_tools) assert [t.name for t in result["main"]] == ["b"] def test_multiple_servers(self): config = { "s1": {"transport": "stdio", "expose_to": ["main"]}, "s2": {"transport": "http", "expose_to": ["research-agent"]}, } server_tools = { "s1": [_make_tool("a")], "s2": [_make_tool("b")], } result = _route_tools(config, server_tools) assert [t.name for t in result["main"]] == ["a"] assert [t.name for t in result["research-agent"]] == ["b"] def test_expose_to_string_not_list(self): config = {"srv": {"transport": "stdio", "expose_to": "debug-agent"}} server_tools = {"srv": [_make_tool("x")]} result = _route_tools(config, server_tools) assert "debug-agent" in result def test_empty_server_tools(self): config = {"srv": {"transport": "stdio"}} server_tools = {"srv": []} result = _route_tools(config, server_tools) assert result.get("main", []) == [] # ---- add_mcp_server / remove_mcp_server ---- @pytest.fixture def user_mcp_dir(tmp_path, monkeypatch): """Redirect user MCP config to a temp directory.""" cfg_dir = tmp_path / "config" cfg_dir.mkdir() cfg_file = cfg_dir / "mcp.yaml" monkeypatch.setattr("EvoScientist.mcp.client.USER_CONFIG_DIR", cfg_dir) monkeypatch.setattr("EvoScientist.mcp.client.USER_MCP_CONFIG", cfg_file) return cfg_file class TestAddMcpServer: def test_add_stdio_server(self, user_mcp_dir): entry = add_mcp_server( "fs", "stdio", command="npx", args=["-y", "server", "/tmp"] ) assert entry["transport"] == "stdio" assert entry["command"] == "npx" assert entry["args"] == ["-y", "server", "/tmp"] # Verify persisted data = yaml.safe_load(user_mcp_dir.read_text()) assert "fs" in data def test_add_http_server(self, user_mcp_dir): entry = add_mcp_server( "api", "http", url="http://localhost:8080/mcp", headers={"Authorization": "Bearer tok"}, ) assert entry["url"] == "http://localhost:8080/mcp" assert entry["headers"]["Authorization"] == "Bearer tok" def test_add_sse_server(self, user_mcp_dir): entry = add_mcp_server("sse-srv", "sse", url="http://localhost:9090/sse") assert entry["transport"] == "sse" def test_add_websocket_server(self, user_mcp_dir): entry = add_mcp_server("ws", "websocket", url="ws://localhost:8765") assert entry["transport"] == "websocket" def test_add_with_tools_and_expose_to(self, user_mcp_dir): entry = add_mcp_server( "fs", "stdio", command="npx", args=[], tools=["read_file"], expose_to=["main", "code-agent"], ) assert entry["tools"] == ["read_file"] assert entry["expose_to"] == ["main", "code-agent"] def test_add_replaces_existing(self, user_mcp_dir): add_mcp_server("srv", "stdio", command="old") add_mcp_server("srv", "http", url="http://new") data = yaml.safe_load(user_mcp_dir.read_text()) assert data["srv"]["transport"] == "http" def test_add_invalid_transport_raises(self, user_mcp_dir): with pytest.raises(ValueError, match="Unknown transport"): add_mcp_server("bad", "carrier_pigeon", url="coo://rooftop") def test_stdio_without_command_raises(self, user_mcp_dir): with pytest.raises(ValueError, match="requires a command"): add_mcp_server("bad", "stdio") def test_http_without_url_raises(self, user_mcp_dir): with pytest.raises(ValueError, match="requires a url"): add_mcp_server("bad", "http") def test_add_with_env(self, user_mcp_dir): entry = add_mcp_server( "fs", "stdio", command="npx", args=[], env={"FOO": "bar"} ) assert entry["env"] == {"FOO": "bar"} def test_add_multiple_servers(self, user_mcp_dir): add_mcp_server("a", "stdio", command="cmd1") add_mcp_server("b", "http", url="http://x") data = yaml.safe_load(user_mcp_dir.read_text()) assert "a" in data assert "b" in data class TestRemoveMcpServer: def test_remove_existing(self, user_mcp_dir): add_mcp_server("fs", "stdio", command="npx") assert remove_mcp_server("fs") is True data = yaml.safe_load(user_mcp_dir.read_text()) or {} assert "fs" not in data def test_remove_nonexistent(self, user_mcp_dir): assert remove_mcp_server("nope") is False def test_remove_preserves_others(self, user_mcp_dir): add_mcp_server("a", "stdio", command="cmd1") add_mcp_server("b", "http", url="http://x") remove_mcp_server("a") data = yaml.safe_load(user_mcp_dir.read_text()) assert "a" not in data assert "b" in data # ---- _parse_mcp_add_args (CLI arg parser) ---- class TestParseMcpAddArgs: def test_stdio_basic(self): r = parse_mcp_add_args(["fs", "npx", "-y", "server", "/tmp"]) assert r["name"] == "fs" assert r["transport"] == "stdio" assert r["command"] == "npx" assert r["args"] == ["-y", "server", "/tmp"] def test_http_auto_detected(self): r = parse_mcp_add_args(["api", "http://localhost:8080/mcp"]) assert r["transport"] == "http" assert r["url"] == "http://localhost:8080/mcp" def test_https_auto_detected(self): r = parse_mcp_add_args(["api", "https://example.com/mcp"]) assert r["transport"] == "http" assert r["url"] == "https://example.com/mcp" def test_ws_auto_detected(self): r = parse_mcp_add_args(["ws", "ws://localhost:9090"]) assert r["transport"] == "websocket" def test_explicit_transport_override(self): r = parse_mcp_add_args(["srv", "https://example.com/sse", "--transport", "sse"]) assert r["transport"] == "sse" assert r["url"] == "https://example.com/sse" def test_explicit_transport_short_flag(self): r = parse_mcp_add_args(["srv", "https://x", "-T", "websocket"]) assert r["transport"] == "websocket" def test_tools_flag(self): r = parse_mcp_add_args(["srv", "http://x", "--tools", "a,b"]) assert r["tools"] == ["a", "b"] def test_expose_to_flag(self): r = parse_mcp_add_args(["srv", "http://x", "--expose-to", "main,code-agent"]) assert r["expose_to"] == ["main", "code-agent"] def test_header_flag(self): r = parse_mcp_add_args( ["srv", "http://x", "--header", "Authorization:Bearer tok"] ) assert r["headers"] == {"Authorization": "Bearer tok"} def test_env_flag(self): r = parse_mcp_add_args(["srv", "cmd", "--env", "FOO=bar"]) assert r["env"] == {"FOO": "bar"} def test_too_few_tokens_raises(self): with pytest.raises(ValueError, match="Usage"): parse_mcp_add_args(["fs"]) def test_double_dash_ignored(self): r = parse_mcp_add_args(["srv", "npx", "--", "-y", "pkg"]) assert r["command"] == "npx" assert r["args"] == ["-y", "pkg"] assert "--" not in r["args"] def test_env_ref_flag(self): r = parse_mcp_add_args(["srv", "cmd", "--env-ref", "FOO"]) assert r["env"] == {"FOO": "${FOO}"} def test_env_ref_and_env_combined(self): r = parse_mcp_add_args( ["srv", "cmd", "--env", "DEBUG=true", "--env-ref", "API_KEY"] ) assert r["env"] == {"DEBUG": "true", "API_KEY": "${API_KEY}"} def test_missing_command_or_url_raises(self): with pytest.raises(ValueError, match="command or URL is required"): parse_mcp_add_args(["fs", "--tools", "a"]) # ---- edit_mcp_server ---- class TestEditMcpServer: def test_edit_expose_to(self, user_mcp_dir): add_mcp_server("fs", "stdio", command="npx", args=[]) entry = edit_mcp_server("fs", expose_to=["main", "code-agent"]) assert entry["expose_to"] == ["main", "code-agent"] assert entry["command"] == "npx" # unchanged def test_edit_tools(self, user_mcp_dir): add_mcp_server("fs", "stdio", command="npx", args=[]) entry = edit_mcp_server("fs", tools=["read_file"]) assert entry["tools"] == ["read_file"] def test_edit_clear_tools(self, user_mcp_dir): add_mcp_server("fs", "stdio", command="npx", tools=["read_file"]) entry = edit_mcp_server("fs", tools=None) assert "tools" not in entry def test_edit_url(self, user_mcp_dir): add_mcp_server("api", "http", url="http://old:8080/mcp") entry = edit_mcp_server("api", url="http://new:9090/mcp") assert entry["url"] == "http://new:9090/mcp" assert entry["transport"] == "http" # unchanged def test_edit_nonexistent_raises(self, user_mcp_dir): with pytest.raises(KeyError, match="not found"): edit_mcp_server("nope", tools=["a"]) def test_edit_invalid_transport_raises(self, user_mcp_dir): add_mcp_server("fs", "stdio", command="npx") with pytest.raises(ValueError, match="Unknown transport"): edit_mcp_server("fs", transport="carrier_pigeon") def test_edit_removes_required_field_raises(self, user_mcp_dir): add_mcp_server("fs", "stdio", command="npx") with pytest.raises(ValueError, match="requires a command"): edit_mcp_server("fs", command=None) def test_edit_preserves_unrelated_fields(self, user_mcp_dir): add_mcp_server( "fs", "stdio", command="npx", args=["-y", "srv"], tools=["a"], expose_to=["main"], ) entry = edit_mcp_server("fs", expose_to=["code-agent"]) assert entry["tools"] == ["a"] assert entry["args"] == ["-y", "srv"] assert entry["expose_to"] == ["code-agent"] # ---- parse_mcp_edit_args ---- class TestParseMcpEditArgs: def test_basic_field(self): name, fields = parse_mcp_edit_args(["srv", "--url", "http://new"]) assert name == "srv" assert fields["url"] == "http://new" def test_tools_none_clears(self): _, fields = parse_mcp_edit_args(["srv", "--tools", "none"]) assert fields["tools"] is None def test_expose_to_csv(self): _, fields = parse_mcp_edit_args(["srv", "--expose-to", "main,code-agent"]) assert fields["expose_to"] == ["main", "code-agent"] def test_multiple_fields(self): _, fields = parse_mcp_edit_args(["srv", "--url", "http://x", "--tools", "a,b"]) assert fields["url"] == "http://x" assert fields["tools"] == ["a", "b"] def test_empty_raises(self): with pytest.raises(ValueError, match="Usage"): parse_mcp_edit_args([]) def test_no_fields_raises(self): with pytest.raises(ValueError, match="No fields"): parse_mcp_edit_args(["srv"])