Ryan Malloy
db6ed2f550
Replace hand-rolled base64 dicts with FastMCP's Image() helper in
get_imagery, compare_dates, get_imagery_composite, and get_legend.
FastMCP handles base64 encoding and MIME type automatically.
Add Context progress reporting to imagery tools so MCP clients can
display fetch status (resolving location, fetching imagery, etc).
Add dynamic resources:
gibs://colormap/{layer_id} — colormap explanation text
gibs://dates/{layer_id} — date range + live DescribeDomains
Remove unused base64 import.
302 lines
10 KiB
Python
302 lines
10 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",
|
|
}
|
|
|
|
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
|
|
|
|
|
|
@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
|