Ryan Malloy
ac6977a105
- query_point: reverse-maps pixel RGB through colormap to recover exact data values at geographic coordinates - get_time_series: fetches imagery across evenly-spaced dates for temporal analysis (up to 12 frames) - Auto-detect polar stereographic projection (EPSG:3413/3031) for high-latitude bounding boxes - Add progress reporting to all HTTP-calling tools - Add quantitative_snapshot and seasonal_timelapse prompts - Update README with 3 new conversational examples - 92 tests passing
553 lines
19 KiB
Python
553 lines
19 KiB
Python
"""FastMCP integration tests — tool calls against the real server, HTTP mocked."""
|
|
|
|
import json
|
|
from io import BytesIO
|
|
|
|
import httpx
|
|
import respx
|
|
from fastmcp import Client
|
|
from PIL import Image as PILImage
|
|
|
|
import mcgibs.server as server_module
|
|
from mcgibs.client import GIBSClient
|
|
from mcgibs.server import mcp
|
|
|
|
|
|
async def _init_mock_client(capabilities_xml: str) -> GIBSClient:
|
|
"""Create and initialize a GIBSClient with mocked capabilities."""
|
|
client = GIBSClient()
|
|
await client.initialize()
|
|
return client
|
|
|
|
|
|
@respx.mock
|
|
async def test_search_tool(capabilities_xml):
|
|
"""search_gibs_layers finds the sea ice layer by keyword."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("search_gibs_layers", {"query": "sea ice"})
|
|
text = result.content[0].text
|
|
assert "AMSR2" in text
|
|
assert "Sea Ice" in text
|
|
# Should not match unrelated layers
|
|
assert "Surface Air Temperature" not in text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_get_layer_info_tool(capabilities_xml, layer_metadata_json):
|
|
"""get_layer_info returns enriched JSON for the true color layer."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
layer_id = "MODIS_Terra_CorrectedReflectance_TrueColor"
|
|
respx.get(url__regex=rf".*layer-metadata.*/{layer_id}\.json").mock(
|
|
return_value=httpx.Response(200, text=layer_metadata_json)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("get_layer_info", {"layer_id": layer_id})
|
|
text = result.content[0].text
|
|
info = json.loads(text)
|
|
|
|
assert info["identifier"] == layer_id
|
|
assert info["instrument"] == "MODIS"
|
|
assert info["platform"] == "Terra"
|
|
assert info["ongoing"] is True
|
|
assert "time" in info
|
|
assert info["time"]["start"] == "2000-02-24"
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_resolve_place_tool(capabilities_xml):
|
|
"""resolve_place geocodes a place name via mocked Nominatim."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
nominatim_response = [
|
|
{
|
|
"display_name": "Tokyo, Japan",
|
|
"lat": "35.6762",
|
|
"lon": "139.6503",
|
|
"boundingbox": ["35.5190", "35.8178", "138.9428", "139.9200"],
|
|
"osm_type": "relation",
|
|
"importance": 0.82,
|
|
}
|
|
]
|
|
respx.get(url__regex=r".*nominatim.*search.*").mock(
|
|
return_value=httpx.Response(200, json=nominatim_response)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("resolve_place", {"place": "Tokyo"})
|
|
text = result.content[0].text
|
|
data = json.loads(text)
|
|
|
|
assert data["display_name"] == "Tokyo, Japan"
|
|
assert abs(data["lat"] - 35.6762) < 0.01
|
|
assert abs(data["lon"] - 139.6503) < 0.01
|
|
assert "bbox" in data
|
|
assert data["bbox"]["west"] < data["bbox"]["east"]
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_build_tile_url_tool(capabilities_xml):
|
|
"""build_tile_url returns a properly formatted WMTS tile URL."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"build_tile_url",
|
|
{
|
|
"layer_id": "MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
"date": "2025-06-01",
|
|
"zoom": 3,
|
|
"row": 2,
|
|
"col": 4,
|
|
},
|
|
)
|
|
url = result.content[0].text
|
|
|
|
assert "MODIS_Terra_CorrectedReflectance_TrueColor" in url
|
|
assert "2025-06-01" in url
|
|
assert "epsg4326" in url
|
|
# True color layer uses JPEG and 250m matrix set
|
|
assert url.endswith(".jpg")
|
|
assert "/250m/" in url
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_list_tools(capabilities_xml):
|
|
"""All expected tools are registered on the MCP server."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
tool_names = {t.name for t in tools}
|
|
|
|
expected = {
|
|
"search_gibs_layers",
|
|
"get_layer_info",
|
|
"list_measurements",
|
|
"check_layer_dates",
|
|
"get_imagery",
|
|
"compare_dates",
|
|
"get_imagery_composite",
|
|
"explain_layer_colormap",
|
|
"get_legend",
|
|
"resolve_place",
|
|
"build_tile_url",
|
|
"query_point",
|
|
"get_time_series",
|
|
}
|
|
|
|
for name in expected:
|
|
assert name in tool_names, f"Missing tool: {name}"
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Image() return type tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_fake_jpeg() -> bytes:
|
|
"""Create a tiny valid JPEG for mocking WMS responses."""
|
|
buf = BytesIO()
|
|
PILImage.new("RGB", (10, 10), "blue").save(buf, format="JPEG")
|
|
return buf.getvalue()
|
|
|
|
|
|
@respx.mock
|
|
async def test_get_imagery_returns_image_content(capabilities_xml):
|
|
"""get_imagery returns TextContent + ImageContent (not raw dicts)."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
respx.get(url__regex=r".*wms\.cgi.*").mock(
|
|
return_value=httpx.Response(
|
|
200, content=_make_fake_jpeg(), headers={"content-type": "image/jpeg"}
|
|
)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"get_imagery",
|
|
{
|
|
"layer_id": "MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
"date": "2025-06-01",
|
|
"bbox": [-120.0, 30.0, -110.0, 40.0],
|
|
},
|
|
)
|
|
assert len(result.content) == 2
|
|
assert result.content[0].type == "text"
|
|
assert "Corrected Reflectance" in result.content[0].text
|
|
assert result.content[1].type == "image"
|
|
assert result.content[1].mimeType == "image/jpeg"
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_get_legend_returns_image_content(capabilities_xml):
|
|
"""get_legend returns TextContent + ImageContent for layers with legends."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
fake_png = BytesIO()
|
|
PILImage.new("RGB", (200, 20), "white").save(fake_png, format="PNG")
|
|
respx.get(url__regex=r".*legends/.*\.png").mock(
|
|
return_value=httpx.Response(
|
|
200, content=fake_png.getvalue(), headers={"content-type": "image/png"}
|
|
)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"get_legend",
|
|
{"layer_id": "AMSR2_Sea_Ice_Concentration_12km_Monthly"},
|
|
)
|
|
assert len(result.content) == 2
|
|
assert result.content[0].type == "text"
|
|
assert result.content[1].type == "image"
|
|
assert result.content[1].mimeType == "image/png"
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dynamic resource tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_list_resources(capabilities_xml):
|
|
"""Server exposes all expected resources including dynamic templates."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
templates = await client.list_resource_templates()
|
|
template_uris = {t.uriTemplate for t in templates}
|
|
|
|
assert "gibs://layer/{layer_id}" in template_uris
|
|
assert "gibs://colormap/{layer_id}" in template_uris
|
|
assert "gibs://dates/{layer_id}" in template_uris
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Prompt tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_list_prompts(capabilities_xml):
|
|
"""All expected prompts are registered on the MCP server."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
prompts = await client.list_prompts()
|
|
prompt_names = {p.name for p in prompts}
|
|
|
|
expected = {
|
|
"investigate_event",
|
|
"earth_overview",
|
|
"satellite_snapshot",
|
|
"climate_monitor",
|
|
"layer_deep_dive",
|
|
"multi_layer_story",
|
|
"polar_watch",
|
|
"quantitative_snapshot",
|
|
"seasonal_timelapse",
|
|
}
|
|
|
|
for name in expected:
|
|
assert name in prompt_names, f"Missing prompt: {name}"
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_satellite_snapshot_prompt(capabilities_xml):
|
|
"""satellite_snapshot prompt includes the place name and expected tool references."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt(
|
|
"satellite_snapshot", {"place": "Tokyo", "date": "2025-06-01"}
|
|
)
|
|
text = result.messages[0].content.text
|
|
|
|
assert "Tokyo" in text
|
|
assert "resolve_place" in text
|
|
assert "get_imagery" in text or "get_imagery_composite" in text
|
|
assert "MODIS_Terra_CorrectedReflectance_TrueColor" in text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_colormap_resource(capabilities_xml, colormap_xml):
|
|
"""gibs://colormap/{layer_id} returns colormap explanation text."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
respx.get(url__regex=r".*colormaps/v1\.3/.*\.xml").mock(
|
|
return_value=httpx.Response(200, text=colormap_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource(
|
|
"gibs://colormap/AIRS_L3_Surface_Air_Temperature_Daily_Day"
|
|
)
|
|
text = result[0].text
|
|
assert "Surface Air Temperature" in text
|
|
assert "K" in text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# query_point tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_fake_png(rgb: tuple[int, int, int], size: int = 3) -> bytes:
|
|
"""Create a tiny PNG with uniform color for mocking point queries."""
|
|
buf = BytesIO()
|
|
PILImage.new("RGBA", (size, size), (*rgb, 255)).save(buf, format="PNG")
|
|
return buf.getvalue()
|
|
|
|
|
|
@respx.mock
|
|
async def test_query_point(capabilities_xml, colormap_xml):
|
|
"""query_point reverse-maps pixel RGB through colormap to data value."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
# Return a 3x3 PNG with rgb (255,100,50) → maps to [290,300) K
|
|
respx.get(url__regex=r".*wms\.cgi.*").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
content=_make_fake_png((255, 100, 50)),
|
|
headers={"content-type": "image/png"},
|
|
)
|
|
)
|
|
respx.get(url__regex=r".*colormaps/v1\.3/.*\.xml").mock(
|
|
return_value=httpx.Response(200, text=colormap_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"query_point",
|
|
{
|
|
"layer_id": "AIRS_L3_Surface_Air_Temperature_Daily_Day",
|
|
"date": "2025-06-01",
|
|
"lat": 35.67,
|
|
"lon": 139.65,
|
|
},
|
|
)
|
|
data = json.loads(result.content[0].text)
|
|
|
|
assert data["lat"] == 35.67
|
|
assert data["lon"] == 139.65
|
|
assert data["quality"] == "exact" # exact RGB match
|
|
assert "value" in data
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_query_point_no_colormap(capabilities_xml):
|
|
"""query_point returns error for layers without colormaps."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"query_point",
|
|
{
|
|
"layer_id": "MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
"date": "2025-06-01",
|
|
"lat": 35.67,
|
|
"lon": 139.65,
|
|
},
|
|
)
|
|
text = result.content[0].text
|
|
assert "no colormap" in text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_time_series tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_get_time_series(capabilities_xml):
|
|
"""get_time_series returns multiple dated images."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
respx.get(url__regex=r".*wms\.cgi.*").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
content=_make_fake_jpeg(),
|
|
headers={"content-type": "image/jpeg"},
|
|
)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"get_time_series",
|
|
{
|
|
"layer_id": "MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
"start_date": "2025-01-01",
|
|
"end_date": "2025-06-01",
|
|
"bbox": [-120.0, 30.0, -110.0, 40.0],
|
|
"steps": 3,
|
|
},
|
|
)
|
|
# 3 steps → 3 text labels + 3 images = 6 content items
|
|
texts = [c for c in result.content if c.type == "text"]
|
|
images = [c for c in result.content if c.type == "image"]
|
|
|
|
assert len(texts) == 3
|
|
assert len(images) == 3
|
|
assert "2025-01-01" in texts[0].text
|
|
assert "2025-06-01" in texts[-1].text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New prompt tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_quantitative_snapshot_prompt(capabilities_xml):
|
|
"""quantitative_snapshot prompt includes locations and tool references."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt(
|
|
"quantitative_snapshot",
|
|
{
|
|
"layer_id": "AIRS_L3_Surface_Air_Temperature_Daily_Day",
|
|
"locations": "Tokyo, Sydney",
|
|
"date": "2025-06-01",
|
|
},
|
|
)
|
|
text = result.messages[0].content.text
|
|
|
|
assert "Tokyo" in text
|
|
assert "Sydney" in text
|
|
assert "query_point" in text
|
|
assert "explain_layer_colormap" in text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|
|
|
|
|
|
@respx.mock
|
|
async def test_seasonal_timelapse_prompt(capabilities_xml):
|
|
"""seasonal_timelapse prompt includes dates and get_time_series reference."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
server_module._client = await _init_mock_client(capabilities_xml)
|
|
try:
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt(
|
|
"seasonal_timelapse",
|
|
{
|
|
"layer_id": "MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
"location": "Amazon Rainforest",
|
|
"start_date": "2025-01-01",
|
|
"end_date": "2025-12-01",
|
|
},
|
|
)
|
|
text = result.messages[0].content.text
|
|
|
|
assert "Amazon Rainforest" in text
|
|
assert "get_time_series" in text
|
|
assert "2025-01-01" in text
|
|
assert "2025-12-01" in text
|
|
finally:
|
|
await server_module._client.close()
|
|
server_module._client = None
|