Weather MCP

from fastapi.testclient import TestClient import jwt from app.core.config import settings from app.main import create_app class _StubWeatherCurrentMCP: async def execute(self, input_data, ctx): return { "location": "Beijing", "observed_at": "2025-12-13T15:00:00+08:00", "temperature": 12.0, "condition": "多云", "humidity": 55, "wind_speed": 5.4, "wind_dir": "西北风", "raw": {"code": "200"}, } def test_weather_current_tool_requires_auth(monkeypatch): monkeypatch.setattr(settings, "auth_mode", "local_jwt") monkeypatch.setattr(settings, "secret_key", "test_secret") app = create_app() client = TestClient(app) resp = client.post("/tools/weather.current", json={"location": "北京"}) assert resp.status_code == 401 def test_weather_current_tool_ok(monkeypatch): monkeypatch.setattr(settings, "auth_mode", "local_jwt") monkeypatch.setattr(settings, "secret_key", "test_secret") # Avoid upstream calls; route imports WeatherCurrentMCP from app.mcp.weather_mcp. from app import api as api_pkg # noqa: F401 from app.api import weather as weather_route monkeypatch.setattr(weather_route, "WeatherCurrentMCP", _StubWeatherCurrentMCP) token = jwt.encode({"sub": "u1"}, "test_secret", algorithm="HS256") app = create_app() client = TestClient(app) resp = client.post( "/tools/weather.current", json={"location": "北京"}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json() assert data["location"] == "Beijing" assert data["condition"] == "多云"