MCP Server Template

""" Unit Tests for Tools Module Tests individual tool components and functionality. """ import json import os from unittest.mock import patch, AsyncMock, Mock import httpx import pytest from src.mcp_server.tools.arithmetic import ArithmeticTools from src.mcp_server.tools.weather import WeatherTools from src.mcp_server.tools.registry import ToolRegistry, get_tool_registry from src.mcp_server.config.settings import reset_settings @pytest.mark.unit def test_arithmetic_tools_creation(): """Test that arithmetic tools can be created.""" tools = ArithmeticTools() assert tools.name == "arithmetic" assert tools.logger is not None @pytest.mark.unit def test_arithmetic_tools_direct_methods(): """Test direct method calls on arithmetic tools.""" tools = ArithmeticTools() # Test add_numbers method result = tools.add_numbers(5, 3) assert result == 8 # Test subtract_numbers method result = tools.subtract_numbers(10, 4) assert result == 6 @pytest.mark.unit def test_arithmetic_tools_with_negative_numbers(): """Test arithmetic tools with negative numbers.""" tools = ArithmeticTools() # Add with negative assert tools.add_numbers(-5, 3) == -2 assert tools.add_numbers(5, -3) == 2 assert tools.add_numbers(-5, -3) == -8 # Subtract with negative assert tools.subtract_numbers(-5, 3) == -8 assert tools.subtract_numbers(5, -3) == 8 assert tools.subtract_numbers(-5, -3) == -2 @pytest.mark.unit def test_arithmetic_tools_with_zero(): """Test arithmetic tools with zero.""" tools = ArithmeticTools() # Add with zero assert tools.add_numbers(0, 5) == 5 assert tools.add_numbers(5, 0) == 5 assert tools.add_numbers(0, 0) == 0 # Subtract with zero assert tools.subtract_numbers(0, 5) == -5 assert tools.subtract_numbers(5, 0) == 5 assert tools.subtract_numbers(0, 0) == 0 @pytest.mark.unit def test_tool_registry_creation(): """Test tool registry creation and basic functionality.""" registry = ToolRegistry() assert len(registry._tool_classes) == 0 assert len(registry._registered_tools) == 0 @pytest.mark.unit def test_tool_registry_registration(): """Test registering tool classes with registry.""" registry = ToolRegistry() registry.register(ArithmeticTools) assert len(registry._tool_classes) == 1 assert ArithmeticTools in registry._tool_classes @pytest.mark.unit def test_tool_registry_invalid_registration(): """Test that invalid tool classes are rejected.""" registry = ToolRegistry() # Try to register a non-BaseTool class class InvalidTool: pass with pytest.raises(TypeError): registry.register(InvalidTool) @pytest.mark.unit def test_global_tool_registry(): """Test that global tool registry is accessible.""" registry = get_tool_registry() assert registry is not None assert isinstance(registry, ToolRegistry) # Should have at least ArithmeticTools registered assert len(registry._tool_classes) >= 1 @pytest.mark.unit def test_tool_info(): """Test tool info generation.""" tools = ArithmeticTools() info = tools.get_info() assert info["name"] == "arithmetic" assert info["class"] == "ArithmeticTools" assert "tools.arithmetic" in info["module"] # Weather Tools Tests @pytest.mark.unit def test_weather_tools_creation(): """Test that weather tools can be created.""" with patch.dict(os.environ, {}, clear=True): reset_settings() tools = WeatherTools() assert tools.name == "weather" assert tools.logger is not None @pytest.mark.unit def test_weather_tools_mock_data_enabled(): """Test weather tools with mock data enabled (default).""" with patch.dict(os.environ, {}, clear=True): reset_settings() tools = WeatherTools() assert tools._use_mock_data() is True @pytest.mark.unit def test_weather_tools_mock_data_disabled(): """Test weather tools with mock data disabled.""" with patch.dict(os.environ, {"USE_MOCK_WEATHER_DATA": "false"}, clear=True): reset_settings() tools = WeatherTools() assert tools._use_mock_data() is False @pytest.mark.unit def test_weather_tools_api_key_configuration(): """Test weather API key configuration.""" # Test with no API key with patch.dict(os.environ, {}, clear=True): reset_settings() tools = WeatherTools() assert tools._get_api_key() is None # Test with MCP_WEATHER_API_KEY with patch.dict(os.environ, {"MCP_WEATHER_API_KEY": "test_key_1"}, clear=True): reset_settings() tools = WeatherTools() assert tools._get_api_key() == "test_key_1" # Test with WEATHER_API_KEY (alternative naming) with patch.dict(os.environ, {"WEATHER_API_KEY": "test_key_2"}, clear=True): reset_settings() tools = WeatherTools() assert tools._get_api_key() == "test_key_2" @pytest.mark.unit def test_weather_tools_mock_data_structure(): """Test the structure of mock weather data.""" with patch.dict(os.environ, {}, clear=True): reset_settings() tools = WeatherTools() location = "Test City" mock_data = tools._get_mock_weather_data(location) # Validate required fields required_fields = [ "location", "temperature", "temperature_unit", "humidity", "humidity_unit", "pressure", "pressure_unit", "wind_speed", "wind_speed_unit", "wind_direction", "wind_direction_unit", "weather", "description", "visibility", "visibility_unit", "source", "timestamp" ] for field in required_fields: assert field in mock_data # Validate data types and values assert mock_data["location"] == location assert isinstance(mock_data["temperature"], (int, float)) assert mock_data["temperature_unit"] == "°C" assert isinstance(mock_data["humidity"], int) assert mock_data["humidity_unit"] == "%" assert isinstance(mock_data["pressure"], (int, float)) assert mock_data["pressure_unit"] == "hPa" assert isinstance(mock_data["wind_speed"], (int, float)) assert mock_data["wind_speed_unit"] == "m/s" assert isinstance(mock_data["wind_direction"], int) assert mock_data["wind_direction_unit"] == "degrees" assert isinstance(mock_data["weather"], str) assert isinstance(mock_data["description"], str) assert isinstance(mock_data["visibility"], int) assert mock_data["visibility_unit"] == "meters" assert mock_data["source"] == "mock_data" assert isinstance(mock_data["timestamp"], str) @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_mock_mode_success(): """Test getting weather data in mock mode.""" with patch.dict(os.environ, {"USE_MOCK_WEATHER_DATA": "true"}, clear=True): reset_settings() tools = WeatherTools() # Create a mock FastMCP instance mock_mcp = Mock() # Register the tools tools.register_with_mcp(mock_mcp) # Get the registered function assert mock_mcp.tool.called get_weather_func = mock_mcp.tool.call_args[0][0] # First positional arg # Test the function result = await get_weather_func("New York") assert result["location"] == "New York" assert result["source"] == "mock_data" assert "temperature" in result assert "humidity" in result @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_empty_location_validation(): """Test validation of empty location inputs.""" with patch.dict(os.environ, {}, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] # Test empty string with pytest.raises(ValueError, match="Location cannot be empty"): await get_weather_func("") # Test whitespace only with pytest.raises(ValueError, match="Location cannot be empty"): await get_weather_func(" ") # Test None (this would be caught by function signature, but test anyway) with pytest.raises(ValueError, match="Location cannot be empty"): await get_weather_func(None) @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_key_missing_error(): """Test error when API key is missing in non-mock mode.""" with patch.dict(os.environ, {"USE_MOCK_WEATHER_DATA": "false"}, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with pytest.raises(ValueError, match="WEATHER_API_KEY environment variable is required"): await get_weather_func("New York") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_success(): """Test successful API call.""" mock_response_data = { "name": "New York", "sys": {"country": "US"}, "main": { "temp": 25.5, "humidity": 70, "pressure": 1015 }, "wind": { "speed": 5.2, "deg": 180 }, "weather": [{ "main": "Clear", "description": "clear sky" }], "visibility": 10000, "dt": 1642249200 } with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.json.return_value = mock_response_data mock_response.raise_for_status.return_value = None mock_client.return_value.__aenter__.return_value.get = AsyncMock(return_value=mock_response) result = await get_weather_func("New York") assert result["location"] == "New York, US" assert result["temperature"] == 25.5 assert result["temperature_unit"] == "°C" assert result["humidity"] == 70 assert result["humidity_unit"] == "%" assert result["pressure"] == 1015 assert result["pressure_unit"] == "hPa" assert result["wind_speed"] == 5.2 assert result["wind_speed_unit"] == "m/s" assert result["wind_direction"] == 180 assert result["wind_direction_unit"] == "degrees" assert result["weather"] == "Clear" assert result["description"] == "clear sky" assert result["visibility"] == 10000 assert result["visibility_unit"] == "meters" assert result["source"] == "openweathermap" assert result["timestamp"] == 1642249200 @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_404_error(): """Test API 404 error handling.""" with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.status_code = 404 mock_client.return_value.__aenter__.return_value.get = AsyncMock( side_effect=httpx.HTTPStatusError( "Not Found", request=Mock(), response=mock_response ) ) with pytest.raises(ValueError, match="Location 'InvalidCity' not found"): await get_weather_func("InvalidCity") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_401_error(): """Test API 401 (unauthorized) error handling.""" with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "invalid_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.status_code = 401 mock_client.return_value.__aenter__.return_value.get = AsyncMock( side_effect=httpx.HTTPStatusError( "Unauthorized", request=Mock(), response=mock_response ) ) with pytest.raises(ValueError, match="Invalid API key provided for weather service"): await get_weather_func("New York") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_timeout_error(): """Test API timeout error handling.""" with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( side_effect=httpx.TimeoutException("Request timed out") ) with pytest.raises(ValueError, match="Weather API request timed out"): await get_weather_func("New York") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_connection_error(): """Test API connection error handling.""" with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( side_effect=httpx.RequestError("Connection failed") ) with pytest.raises(ValueError, match="Weather API request failed: Connection failed"): await get_weather_func("New York") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_invalid_response(): """Test invalid API response handling.""" with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.json.side_effect = json.JSONDecodeError("Invalid JSON", "", 0) mock_response.raise_for_status.return_value = None mock_client.return_value.__aenter__.return_value.get = AsyncMock(return_value=mock_response) with pytest.raises(ValueError, match="Invalid weather API response format"): await get_weather_func("New York") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_missing_fields(): """Test API response with missing required fields.""" incomplete_response = { "name": "New York" # Missing required fields like main, weather, etc. } with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.json.return_value = incomplete_response mock_response.raise_for_status.return_value = None mock_client.return_value.__aenter__.return_value.get = AsyncMock(return_value=mock_response) with pytest.raises(ValueError, match="Invalid weather API response format"): await get_weather_func("New York") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_partial_wind_data(): """Test API response with partial wind data.""" mock_response_data = { "name": "New York", "sys": {"country": "US"}, "main": { "temp": 25.5, "humidity": 70, "pressure": 1015 }, # wind data is optional, test default values "weather": [{ "main": "Clear", "description": "clear sky" }], "visibility": 10000, "dt": 1642249200 } with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.json.return_value = mock_response_data mock_response.raise_for_status.return_value = None mock_client.return_value.__aenter__.return_value.get = AsyncMock(return_value=mock_response) result = await get_weather_func("New York") # Should have default wind values assert result["wind_speed"] == 0 assert result["wind_direction"] == 0 @pytest.mark.unit def test_weather_tools_registry_registration(): """Test that weather tools are properly registered in the global registry.""" registry = get_tool_registry() # WeatherTools should be in the registered tool classes assert WeatherTools in registry._tool_classes @pytest.mark.unit def test_weather_tools_info(): """Test weather tools info generation.""" with patch.dict(os.environ, {}, clear=True): reset_settings() tools = WeatherTools() info = tools.get_info() assert info["name"] == "weather" assert info["class"] == "WeatherTools" assert "tools.weather" in info["module"] @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_whitespace_trimming(): """Test that location input is properly trimmed of whitespace.""" with patch.dict(os.environ, {"USE_MOCK_WEATHER_DATA": "true"}, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] # Test with leading/trailing whitespace result = await get_weather_func(" New York ") assert result["location"] == "New York" # Trimmed @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_http_status_error_generic(): """Test generic HTTP status error handling.""" with patch.dict(os.environ, { "USE_MOCK_WEATHER_DATA": "false", "WEATHER_API_KEY": "test_api_key" }, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with patch('httpx.AsyncClient') as mock_client: mock_response = Mock() mock_response.status_code = 500 mock_client.return_value.__aenter__.return_value.get = AsyncMock( side_effect=httpx.HTTPStatusError( "Internal Server Error", request=Mock(), response=mock_response ) ) with pytest.raises(ValueError, match="Weather API error: HTTP 500"): await get_weather_func("New York") @pytest.mark.unit def test_weather_tools_environment_variables(): """Test various environment variable configurations.""" # Test MCP_USE_MOCK_WEATHER_DATA with different values test_cases = [ ("true", True), ("True", True), ("TRUE", True), ("1", True), ("yes", True), ("on", True), ("false", False), ("False", False), ("FALSE", False), ("0", False), ("no", False), ("off", False), ("invalid", False), # Invalid values default to False ] for env_value, expected in test_cases: with patch.dict(os.environ, {"USE_MOCK_WEATHER_DATA": env_value}, clear=True): reset_settings() tools = WeatherTools() assert tools._use_mock_data() == expected, f"Failed for env_value='{env_value}'" @pytest.mark.unit def test_weather_tools_real_environment_variable(): """Test that weather tools can read from actual environment variables.""" import os # Test reading from actual environment if WEATHER_API_KEY is set if "WEATHER_API_KEY" in os.environ: # Don't reset settings to allow reading from real environment tools = WeatherTools() api_key = tools._get_api_key() # Should be able to read the actual environment variable assert api_key is not None assert api_key == os.environ["WEATHER_API_KEY"] # If USE_MOCK_WEATHER_DATA is not set, should default to True if "USE_MOCK_WEATHER_DATA" not in os.environ: assert tools._use_mock_data() is True else: expected = os.environ["USE_MOCK_WEATHER_DATA"].lower() in ("true", "1", "yes", "on") assert tools._use_mock_data() == expected else: # If no WEATHER_API_KEY in environment, test should pass but note it pytest.skip("WEATHER_API_KEY not found in environment - test requires local environment variable") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_with_real_environment(): """Test weather tools functionality with real environment variables.""" import os # Only run if WEATHER_API_KEY is available and USE_MOCK_WEATHER_DATA is false if "WEATHER_API_KEY" in os.environ and os.environ.get("USE_MOCK_WEATHER_DATA", "true").lower() in ("false", "0", "no", "off"): tools = WeatherTools() # Should not use mock data assert tools._use_mock_data() is False # Should have API key assert tools._get_api_key() is not None mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] # Note: We won't actually call the API in tests to avoid rate limits # Just verify that the configuration is set up correctly print(f"Real environment test - API key configured: {tools._get_api_key() is not None}") print(f"Real environment test - Mock mode: {tools._use_mock_data()}") else: pytest.skip("Real API test requires WEATHER_API_KEY and USE_MOCK_WEATHER_DATA=false") @pytest.mark.unit @pytest.mark.asyncio async def test_weather_tools_api_key_validation(): """Test API key format validation.""" with patch.dict(os.environ, {"USE_MOCK_WEATHER_DATA": "false"}, clear=True): reset_settings() # Test invalid API key length with patch.dict(os.environ, {"WEATHER_API_KEY": "short_key"}, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] with pytest.raises(ValueError, match="Invalid API key format: expected 32 characters"): await get_weather_func("New York") # Test valid length API key (should pass validation, may fail at API level) with patch.dict(os.environ, {"WEATHER_API_KEY": "12345678901234567890123456789012"}, clear=True): reset_settings() tools = WeatherTools() mock_mcp = Mock() tools.register_with_mcp(mock_mcp) get_weather_func = mock_mcp.tool.call_args[0][0] # This should pass API key validation but will likely fail with network/API error # We'll catch any exception, but specifically check it's not a format validation error try: await get_weather_func("New York") except ValueError as e: # Should not be a format validation error assert "Invalid API key format" not in str(e) except Exception: # Other exceptions (network, API response) are expected pass