Weather MCP

from fastapi.testclient import TestClient import jwt from app.main import create_app from app.core.config import settings 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_mcp_weather_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.get("/mcp/weather?location=Beijing") assert resp.status_code == 401 def test_mcp_weather_ok(monkeypatch): monkeypatch.setattr(settings, "auth_mode", "local_jwt") monkeypatch.setattr(settings, "secret_key", "test_secret") # Avoid real upstream calls in unit tests. from app.api.routes import mcp_weather as mcp_weather_route monkeypatch.setattr(mcp_weather_route, "WeatherCurrentMCP", _StubWeatherCurrentMCP) token = jwt.encode({"sub": "u1"}, "test_secret", algorithm="HS256") app = create_app() client = TestClient(app) resp = client.get( "/mcp/weather?location=%E5%8C%97%E4%BA%AC", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json() assert data["location"] == "Beijing" assert data["condition"] == "多云"