"""Tests for the configuration loader module."""
import json
import shutil
import tempfile
from collections.abc import Callable, Generator
from pathlib import Path
from unittest.mock import patch
import pytest
from mcp.client.stdio import StdioServerParameters
from mcp_proxy.config_loader import load_named_server_configs_from_file
@pytest.fixture
def create_temp_config_file() -> Generator[Callable[[dict], 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:
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]) -> 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],
) -> 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 / "config_example.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(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],
) -> 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, {})
@patch("mcp_proxy.config_loader.logger")
def test_invalid_server_entry_not_dict(
mock_logger: object,
create_temp_config_file: Callable[[dict], 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)
loaded_params = load_named_server_configs_from_file(tmp_config_path, {})
assert len(loaded_params) == 0 # No servers should be loaded
mock_logger.warning.assert_called_with(
"Skipping invalid server config for '%s' in %s. Entry is not a dictionary.",
"server1",
tmp_config_path,
)
@patch("mcp_proxy.config_loader.logger")
def test_server_entry_missing_command(
mock_logger: object,
create_temp_config_file: Callable[[dict], 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)
loaded_params = load_named_server_configs_from_file(tmp_config_path, {})
assert "server_no_command" not in loaded_params
mock_logger.warning.assert_called_with(
"Named server '%s' from config is missing 'command'. Skipping.",
"server_no_command",
)
@patch("mcp_proxy.config_loader.logger")
def test_server_entry_invalid_args_type(
mock_logger: object,
create_temp_config_file: Callable[[dict], 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)
loaded_params = load_named_server_configs_from_file(tmp_config_path, {})
assert "server_invalid_args" not in loaded_params
mock_logger.warning.assert_called_with(
"Named server '%s' from config has invalid 'args' (must be a list). Skipping.",
"server_invalid_args",
)
def test_empty_mcpservers_dict(create_temp_config_file: Callable[[dict], str]) -> None:
"""Test handling of configuration files with empty mcpServers dictionary."""
config_content = {"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]) -> None:
"""Test handling of configuration files with empty JSON objects."""
config_content = {} # 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()