Weather MCP

import httpx import pytest from app.clients.qweather_client import QWeatherClient from app.schemas.weather import ( map_qweather_air_to_result, map_qweather_daily_to_result, map_qweather_hourly_to_result, map_qweather_warning_to_result, ) @pytest.mark.asyncio async def test_qweather_daily_forecast_mapping() -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/geo/v2/city/lookup": return httpx.Response( 200, json={ "code": "200", "location": [{"name": "Guangzhou", "id": "101280101", "lat": "23.13", "lon": "113.26"}], }, ) if request.url.path == "/v7/weather/3d": return httpx.Response( 200, json={ "code": "200", "daily": [ { "fxDate": "2025-12-14", "tempMax": "20", "tempMin": "12", "textDay": "多云", "textNight": "晴", "humidity": "65", "windSpeedDay": "10", "windDirDay": "东风", } ], }, ) raise AssertionError(f"unexpected path: {request.url.path}") transport = httpx.MockTransport(handler) async with QWeatherClient( api_host="https://example.qweatherapi.com", api_key="test_key", transport=transport, ) as client: location, raw = await client.get_forecast_daily(location="广州", days=3, lang="zh", unit="m") result = map_qweather_daily_to_result(location=location, raw=raw) assert result.location == "Guangzhou" assert len(result.forecasts) == 1 assert result.forecasts[0].date == "2025-12-14" assert result.forecasts[0].temp_max == 20.0 @pytest.mark.asyncio async def test_qweather_hourly_forecast_mapping() -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/geo/v2/city/lookup": return httpx.Response( 200, json={ "code": "200", "location": [{"name": "Guangzhou", "id": "101280101", "lat": "23.13", "lon": "113.26"}], }, ) if request.url.path == "/v7/weather/24h": return httpx.Response( 200, json={ "code": "200", "hourly": [ { "fxTime": "2025-12-13T18:00:00+08:00", "temp": "15", "text": "阵雨", "windSpeed": "18", "windDir": "北风", "humidity": "80", } ], }, ) raise AssertionError(f"unexpected path: {request.url.path}") transport = httpx.MockTransport(handler) async with QWeatherClient( api_host="https://example.qweatherapi.com", api_key="test_key", transport=transport, ) as client: location, raw = await client.get_forecast_hourly(location="广州", hours=24, lang="zh", unit="m") result = map_qweather_hourly_to_result(location=location, raw=raw) assert result.location == "Guangzhou" assert len(result.hourly) == 1 assert result.hourly[0].temperature == 15.0 @pytest.mark.asyncio async def test_qweather_weather_alerts_mapping() -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/geo/v2/city/lookup": return httpx.Response( 200, json={ "code": "200", "location": [{"name": "Guangzhou", "id": "101280101", "lat": "23.13", "lon": "113.26"}], }, ) if request.url.path == "/v7/warning/now": return httpx.Response( 200, json={ "code": "200", "warning": [ { "type": "1006", "typeName": "大风", "severity": "Minor", "severityColor": "Blue", "title": "发布大风蓝色预警", "text": "预计将出现大风", "startTime": "2025-12-13T10:30+08:00", "endTime": "2025-12-14T10:30+08:00", } ], }, ) raise AssertionError(f"unexpected path: {request.url.path}") transport = httpx.MockTransport(handler) async with QWeatherClient( api_host="https://example.qweatherapi.com", api_key="test_key", transport=transport, ) as client: location, raw = await client.get_weather_alerts(location="广州", lang="zh") result = map_qweather_warning_to_result(location=location, raw=raw) assert result.location == "Guangzhou" assert len(result.alerts) == 1 assert result.alerts[0].type == "大风" @pytest.mark.asyncio async def test_qweather_air_quality_mapping() -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/geo/v2/city/lookup": return httpx.Response( 200, json={ "code": "200", "location": [{"name": "Guangzhou", "id": "101280101", "lat": "23.13", "lon": "113.26"}], }, ) if request.url.path.startswith("/airquality/v1/current/"): return httpx.Response( 200, json={ "indexes": [ { "code": "us-epa", "aqi": 46, "category": "Good", "primaryPollutant": {"code": "pm2p5", "name": "PM 2.5"}, } ], "pollutants": [ {"code": "pm2p5", "concentration": { "value": 11.0, "unit": "μg/m3"}}, {"code": "pm10", "concentration": { "value": 12.0, "unit": "μg/m3"}}, {"code": "o3", "concentration": { "value": 0.02, "unit": "ppb"}}, {"code": "no2", "concentration": { "value": 6.77, "unit": "ppb"}}, {"code": "so2", "concentration": { "value": 1.0, "unit": "ppb"}}, {"code": "co", "concentration": { "value": 0.25, "unit": "ppm"}}, ], }, ) raise AssertionError(f"unexpected path: {request.url.path}") transport = httpx.MockTransport(handler) async with QWeatherClient( api_host="https://example.qweatherapi.com", api_key="test_key", transport=transport, ) as client: location, raw = await client.get_air_quality_current(location="广州") result = map_qweather_air_to_result(location=location, raw=raw) assert result.location == "Guangzhou" assert result.aqi == 46 assert result.pm2p5 == 11.0