Foxxy Bridge

# # Copyright (C) 2024 Billy Bryant # Portions copyright (C) 2024 Sergey Parfenyuk (original MIT-licensed author) # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <https://www.gnu.org/licenses/>. # # MIT License attribution: Portions of this file were originally licensed # under the MIT License by Sergey Parfenyuk (2024). # """Tests for the configuration loader module.""" import json import shutil import tempfile from collections.abc import Callable, Generator from pathlib import Path from typing import Any import pytest from mcp.client.stdio import StdioServerParameters from mcp_foxxy_bridge.config.config_loader import load_named_server_configs_from_file @pytest.fixture def create_temp_config_file() -> Generator[Callable[[dict[str, Any]], str], None, None]: """Creates a temporary JSON config file and returns its path.""" temp_files: list[str] = [] def _create_temp_config_file(config_content: dict[str, Any]) -> str: with tempfile.NamedTemporaryFile( mode="w", delete=False, suffix=".json", ) as tmp_config: json.dump(config_content, tmp_config) temp_files.append(tmp_config.name) return tmp_config.name yield _create_temp_config_file for f_path in temp_files: path = Path(f_path) if path.exists(): path.unlink() def test_load_valid_config(create_temp_config_file: Callable[[dict[str, Any]], str]) -> None: """Test loading a valid configuration file.""" config_content = { "mcpServers": { "server1": { "command": "echo", "args": ["hello"], "env": {"FOO": "bar"}, "enabled": True, }, "server2": { "command": "cat", "args": ["file.txt"], }, }, } tmp_config_path = create_temp_config_file(config_content) base_env = {"PASSED": "env_value"} base_env_with_added_env = {"PASSED": "env_value", "FOO": "bar"} loaded_params = load_named_server_configs_from_file(tmp_config_path, base_env) assert "server1" in loaded_params assert loaded_params["server1"].command == "echo" assert loaded_params["server1"].args == ["hello"] assert loaded_params["server1"].env == base_env_with_added_env # Env is a copy, check if it contains base_env items assert "server2" in loaded_params assert loaded_params["server2"].command == "cat" assert loaded_params["server2"].args == ["file.txt"] assert loaded_params["server2"].env == base_env def test_load_config_with_not_enabled_server( create_temp_config_file: Callable[[dict[str, Any]], str], ) -> None: """Test loading a configuration with disabled servers.""" config_content = { "mcpServers": { "explicitly_enabled_server": {"command": "true_command", "enabled": True}, # No 'enabled' flag, defaults to True "implicitly_enabled_server": {"command": "another_true_command"}, "not_enabled_server": {"command": "false_command", "enabled": False}, }, } tmp_config_path = create_temp_config_file(config_content) loaded_params = load_named_server_configs_from_file(tmp_config_path, {}) assert "explicitly_enabled_server" in loaded_params assert loaded_params["explicitly_enabled_server"].command == "true_command" assert "implicitly_enabled_server" in loaded_params assert loaded_params["implicitly_enabled_server"].command == "another_true_command" assert "not_enabled_server" not in loaded_params def test_file_not_found() -> None: """Test handling of non-existent configuration files.""" with pytest.raises(FileNotFoundError): load_named_server_configs_from_file("non_existent_file.json", {}) def test_json_decode_error() -> None: """Test handling of invalid JSON in configuration files.""" # Create a file with invalid JSON content with tempfile.NamedTemporaryFile( mode="w", delete=False, suffix=".json", ) as tmp_config: tmp_config.write("this is not json {") tmp_config_path = tmp_config.name # Use try/finally to ensure cleanup try: with pytest.raises(json.JSONDecodeError): load_named_server_configs_from_file(tmp_config_path, {}) finally: path = Path(tmp_config_path) if path.exists(): path.unlink() def test_load_example_fetch_config_if_uvx_exists() -> None: """Test loading the example fetch configuration if uvx is available.""" if not shutil.which("uvx"): pytest.skip("uvx command not found in PATH, skipping test for example config.") # Assuming the test is run from the root of the repository example_config_path = Path(__file__).parent.parent / "docs" / "examples" / "basic-config.json" if not example_config_path.exists(): pytest.fail( f"Example config file not found at expected path: {example_config_path}", ) base_env = {"EXAMPLE_ENV": "true"} loaded_params = load_named_server_configs_from_file(str(example_config_path), base_env) assert "fetch" in loaded_params fetch_param = loaded_params["fetch"] assert isinstance(fetch_param, StdioServerParameters) assert fetch_param.command == "uvx" assert fetch_param.args == ["mcp-server-fetch"] assert fetch_param.env == base_env # The 'timeout' and 'transportType' fields from the config are currently ignored by the loader, # so no need to assert them on StdioServerParameters. def test_invalid_config_format_missing_mcpservers( create_temp_config_file: Callable[[dict[str, Any]], str], ) -> None: """Test handling of configuration files missing the mcpServers key.""" config_content = {"some_other_key": "value"} tmp_config_path = create_temp_config_file(config_content) with pytest.raises(ValueError, match="Missing 'mcpServers' key"): load_named_server_configs_from_file(tmp_config_path, {}) def test_invalid_server_entry_not_dict( create_temp_config_file: Callable[[dict[str, Any]], str], ) -> None: """Test handling of server entries that are not dictionaries.""" config_content = {"mcpServers": {"server1": "not_a_dict"}} tmp_config_path = create_temp_config_file(config_content) # Schema validation should now catch this and raise an exception with pytest.raises(ValueError, match="Invalid configuration"): load_named_server_configs_from_file(tmp_config_path, {}) def test_server_entry_missing_command( create_temp_config_file: Callable[[dict[str, Any]], str], ) -> None: """Test handling of server entries missing the command field.""" config_content = {"mcpServers": {"server_no_command": {"args": ["arg1"]}}} tmp_config_path = create_temp_config_file(config_content) # Should raise ValueError due to schema validation failure with pytest.raises(ValueError, match="Invalid configuration"): load_named_server_configs_from_file(tmp_config_path, {}) def test_server_entry_invalid_args_type( create_temp_config_file: Callable[[dict[str, Any]], str], ) -> None: """Test handling of server entries with invalid args type.""" config_content = { "mcpServers": { "server_invalid_args": {"command": "mycmd", "args": "not_a_list"}, }, } tmp_config_path = create_temp_config_file(config_content) # Should raise ValueError due to schema validation failure with pytest.raises(ValueError, match="Invalid configuration"): load_named_server_configs_from_file(tmp_config_path, {}) def test_empty_mcpservers_dict(create_temp_config_file: Callable[[dict[str, Any]], str]) -> None: """Test handling of configuration files with empty mcpServers dictionary.""" config_content: dict[str, Any] = {"mcpServers": {}} tmp_config_path = create_temp_config_file(config_content) loaded_params = load_named_server_configs_from_file(tmp_config_path, {}) assert len(loaded_params) == 0 def test_config_file_is_empty_json_object(create_temp_config_file: Callable[[dict[str, Any]], str]) -> None: """Test handling of configuration files with empty JSON objects.""" config_content: dict[str, Any] = {} # Empty JSON object tmp_config_path = create_temp_config_file(config_content) with pytest.raises(ValueError, match="Missing 'mcpServers' key"): load_named_server_configs_from_file(tmp_config_path, {}) def test_config_file_is_empty_string() -> None: """Test handling of configuration files with empty content.""" # Create a file with an empty string with tempfile.NamedTemporaryFile( mode="w", delete=False, suffix=".json", ) as tmp_config: tmp_config.write("") # Empty content tmp_config_path = tmp_config.name try: with pytest.raises(json.JSONDecodeError): load_named_server_configs_from_file(tmp_config_path, {}) finally: path = Path(tmp_config_path) if path.exists(): path.unlink()