"""
Tests for Weather MCP Client.
"""
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
# Add src to path for testing
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from weather_mcp.client import WeatherMCPClient # noqa: E402
class TestWeatherMCPClient:
"""Test cases for WeatherMCPClient."""
def test_init_default(self):
"""Test client initialization with defaults."""
client = WeatherMCPClient()
assert client.timeout == 30.0
assert client.server_script is not None
def test_init_custom_timeout(self):
"""Test client initialization with custom timeout."""
client = WeatherMCPClient(timeout=60.0)
assert client.timeout == 60.0
def test_init_with_server_script(self):
"""Test client initialization with custom server script."""
client = WeatherMCPClient(server_script="custom_server.py")
assert client.server_script == "custom_server.py"
assert client.timeout == 30.0
def test_find_server_script_fallback(self):
"""Test server script finding with fallback."""
# Create a mock that returns True for the first candidate
def mock_exists(self):
return str(self) == "src/weather_mcp/server.py"
with patch.object(Path, 'exists', mock_exists):
resolved_path = Path("src/weather_mcp/server.py")
with patch.object(Path, 'resolve', return_value=resolved_path):
client = WeatherMCPClient()
assert client.server_script is not None
assert "server.py" in client.server_script
@pytest.mark.asyncio
async def test_call_tool_server_not_found(self):
"""Test tool call when server script not found."""
client = WeatherMCPClient(server_script="/nonexistent/server.py")
result = await client.call_tool("test_tool", {}, timeout=1.0)
# Should return error
assert "error" in result
assert result["error"] == "server script not found"
assert "trace_id" in result
def test_arun_with_new_loop(self):
"""Test arun function with new event loop."""
# Use a valid server script path to avoid initialization error
client = WeatherMCPClient(server_script="src/weather_mcp/server.py")
async def simple_coro():
return "test_result"
# This should work even without existing event loop
result = client.arun(simple_coro())
assert result == "test_result"
def test_weather_parameters(self):
"""Test weather tool accepts simplified parameters."""
from weather_mcp.tools import WeatherInput # noqa: E402
# Test schema with simplified parameters
weather_input = WeatherInput(
lat=35.6895,
lon=139.6917,
hours=12,
target_date="2025-10-03"
)
assert weather_input.lat == 35.6895
assert weather_input.lon == 139.6917
assert weather_input.hours == 12
assert weather_input.target_date == "2025-10-03"
def test_city_weather_parameters(self):
"""Test city weather tool accepts simplified parameters."""
from weather_mcp.tools import CityWeatherInput # noqa: E402
# Test schema with simplified parameters
city_input = CityWeatherInput(
city="Tokyo",
hours=24,
target_date="2025-10-04"
)
assert city_input.city == "Tokyo"
assert city_input.hours == 24
assert city_input.target_date == "2025-10-04"
def test_historical_weather_parameters(self):
"""Test historical weather query with past date."""
from datetime import datetime, timedelta
from weather_mcp.tools import CityWeatherInput # noqa: E402
# Calculate yesterday's date
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
# Test historical query
city_input = CityWeatherInput(
city="Tokyo",
target_date=yesterday
)
assert city_input.city == "Tokyo"
assert city_input.target_date == yesterday
