"""Tests for real data clients (OpenMeteo, FAA).
These hit live APIs — marked with pytest.mark for selective running.
"""
import pytest
from mcp_servers.data.real.openmeteo import (
HUB_COORDS,
WMO_CODES,
_interpolate_waypoints,
_interpret_weather,
get_weather_along_route,
get_weather_forecast_hubs,
)
from mcp_servers.data.real.faa import get_airport_status, _parse_status_xml
class TestOpenMeteoHelpers:
def test_interpolate_waypoints_count(self):
origin = (41.97, -87.91)
dest = (37.62, -122.38)
points = _interpolate_waypoints(origin, dest, n=2)
assert len(points) == 2
def test_interpolate_waypoints_between(self):
origin = (0.0, 0.0)
dest = (30.0, 60.0)
points = _interpolate_waypoints(origin, dest, n=2)
for lat, lon in points:
assert 0 < lat < 30
assert 0 < lon < 60
def test_interpret_weather_clear(self):
result = _interpret_weather(0)
assert result["condition"] == "Clear sky"
assert result["is_significant"] is False
def test_interpret_weather_thunderstorm(self):
result = _interpret_weather(95)
assert result["condition"] == "Thunderstorm"
assert result["is_significant"] is True
def test_interpret_weather_unknown(self):
result = _interpret_weather(999)
assert "Unknown" in result["condition"]
def test_hub_coords_complete(self):
for hub in ["ORD", "EWR", "IAH", "SFO", "DEN"]:
assert hub in HUB_COORDS
lat, lon = HUB_COORDS[hub]
assert -90 <= lat <= 90
assert -180 <= lon <= 180
class TestFAAHelpers:
def test_parse_status_xml_ground_stop(self):
xml = """
Ground Stop Programs
SFO
thunderstorms
6:45 pm PDT
"""
result = _parse_status_xml(xml)
assert "SFO" in result
assert len(result["SFO"]) == 1
assert result["SFO"][0]["type"] == "ground_stop"
assert result["SFO"][0]["reason"] == "thunderstorms"
def test_parse_status_xml_gdp(self):
xml = """
Ground Delay Programs
EWR
wind
45 minutes
1 hour
"""
result = _parse_status_xml(xml)
assert "EWR" in result
assert result["EWR"][0]["type"] == "ground_delay_program"
assert result["EWR"][0]["average_delay"] == "45 minutes"
def test_parse_status_xml_empty(self):
xml = ""
result = _parse_status_xml(xml)
assert result == {}
def test_parse_status_xml_multiple_airports(self):
xml = """
Ground Stop Programs
SFOweather
ORDvolume
"""
result = _parse_status_xml(xml)
assert "SFO" in result
assert "ORD" in result
@pytest.mark.live
class TestOpenMeteoLive:
"""Tests that hit the live OpenMeteo API."""
@pytest.mark.asyncio
async def test_weather_along_route(self):
result = await get_weather_along_route("ORD", "SFO")
assert result["origin"] == "ORD"
assert result["destination"] == "SFO"
assert "waypoints" in result
assert "origin" in result["waypoints"]
assert "destination" in result["waypoints"]
# Check a waypoint has weather data
wp = result["waypoints"]["origin"]
assert "temperature_c" in wp
assert "weather" in wp
@pytest.mark.asyncio
async def test_weather_unknown_airport(self):
result = await get_weather_along_route("ORD", "FAKE")
assert "error" in result
@pytest.mark.asyncio
async def test_hub_forecasts(self):
result = await get_weather_forecast_hubs()
assert "hubs" in result
for hub in ["ORD", "EWR", "IAH", "SFO", "DEN"]:
assert hub in result["hubs"]
hub_data = result["hubs"][hub]
if "error" not in hub_data:
assert "forecast" in hub_data
assert len(hub_data["forecast"]) > 0
@pytest.mark.live
class TestFAALive:
"""Tests that hit the live FAA API."""
@pytest.mark.asyncio
async def test_airport_status_known(self):
result = await get_airport_status("ORD")
assert result["airport"] == "ORD"
assert result["source"] == "faa_nasstatus_live"
assert result["status"] in ("normal_operations", "delays_active", "status_unavailable")
@pytest.mark.asyncio
async def test_airport_status_no_delays(self):
"""Small airport unlikely to have delays."""
result = await get_airport_status("BTV")
assert result["airport"] == "BTV"
# Either normal or unavailable, but should not crash
assert "status" in result