Add Hamilton fix test coverage and graceful shutdown

Tests: 79 total (32 new), covering LRU cache eviction, BBox
validation, _parse_rgb error handling, colormap ID extraction,
client retry logic, WMS response validation, dimension clamping,
geocode caching, classification colormaps, and scientific notation
interval parsing.

Shutdown: register atexit handler to close httpx connection pool
when the MCP server exits, since FastMCP Middleware has no
on_shutdown hook.
This commit is contained in:
Ryan Malloy 2026-02-18 18:19:21 -07:00
parent c2d54e7d59
commit d13ba744d3
5 changed files with 444 additions and 2 deletions

View File

@ -8,6 +8,8 @@ IMPORTANT: Do NOT use `from __future__ import annotations` here.
FastMCP needs runtime types for Literal, list[str], etc. FastMCP needs runtime types for Literal, list[str], etc.
""" """
import asyncio
import atexit
import base64 import base64
import json import json
import logging import logging
@ -44,6 +46,28 @@ def _get_client() -> GIBSClient:
return _client return _client
def _shutdown_client() -> None:
"""Close the GIBS client on interpreter shutdown.
FastMCP Middleware has no on_shutdown hook, so we use atexit to
ensure the httpx connection pool is closed cleanly rather than
relying on garbage collection.
"""
global _client
if _client is not None:
client = _client
_client = None
try:
loop = asyncio.new_event_loop()
loop.run_until_complete(client.close())
loop.close()
except Exception:
pass
atexit.register(_shutdown_client)
# --- Middleware: initialize client on session start --- # --- Middleware: initialize client on session start ---

View File

@ -1,6 +1,10 @@
"""Tests for WMTS GetCapabilities XML parsing and layer search.""" """Tests for WMTS GetCapabilities XML parsing and layer search."""
from mcgibs.capabilities import parse_capabilities, search_layers from mcgibs.capabilities import (
_extract_colormap_id_from_href,
parse_capabilities,
search_layers,
)
def test_parse_capabilities_layer_count(capabilities_xml: str): def test_parse_capabilities_layer_count(capabilities_xml: str):
@ -102,3 +106,38 @@ def test_parse_capabilities_resource_url(capabilities_xml: str):
"{Time}/{TileMatrixSet}/{TileMatrix}/{TileRow}/{TileCol}.jpg" "{Time}/{TileMatrixSet}/{TileMatrix}/{TileRow}/{TileCol}.jpg"
) )
assert layer.resource_url_template == expected assert layer.resource_url_template == expected
# ---------------------------------------------------------------------------
# _extract_colormap_id_from_href (live API colormap ID fix)
# ---------------------------------------------------------------------------
def test_extract_colormap_id_from_href_normal():
"""Extract colormap filename stem from a typical GIBS metadata URL."""
url = "https://gibs.earthdata.nasa.gov/colormaps/v1.3/AMSR2_Sea_Ice_Concentration.xml"
assert _extract_colormap_id_from_href(url) == "AMSR2_Sea_Ice_Concentration"
def test_extract_colormap_id_from_href_trailing_slash():
url = "https://gibs.earthdata.nasa.gov/colormaps/v1.3/Foo_Bar.xml/"
assert _extract_colormap_id_from_href(url) == "Foo_Bar"
def test_extract_colormap_id_from_href_no_xml():
url = "https://gibs.earthdata.nasa.gov/colormaps/v1.3/NoExtension"
assert _extract_colormap_id_from_href(url) is None
def test_extract_colormap_id_from_href_empty():
assert _extract_colormap_id_from_href("") is None
def test_colormap_id_differs_from_layer_id(capabilities_xml: str):
"""AMSR2 layer should have a colormap_id extracted from the metadata href,
which may differ from the layer identifier itself."""
layers = parse_capabilities(capabilities_xml)
amsr2 = layers["AMSR2_Sea_Ice_Concentration_12km_Monthly"]
# The fixture's metadata href contains the actual colormap filename
assert amsr2.has_colormap is True
assert amsr2.colormap_id is not None

View File

@ -3,10 +3,11 @@
from io import BytesIO from io import BytesIO
import httpx import httpx
import pytest
import respx import respx
from PIL import Image from PIL import Image
from mcgibs.client import GIBSClient from mcgibs.client import GIBSClient, _LRUCache
from mcgibs.constants import WMTS_TILE_URL from mcgibs.constants import WMTS_TILE_URL
@ -182,3 +183,237 @@ def test_client_build_tile_url():
ext="jpg", ext="jpg",
) )
assert url == expected assert url == expected
# ---------------------------------------------------------------------------
# _LRUCache (Hamilton fix: bounded cache eviction)
# ---------------------------------------------------------------------------
def test_lru_cache_basic():
"""Cache stores and retrieves values."""
cache = _LRUCache(maxsize=3)
cache.put("a", 1)
cache.put("b", 2)
from mcgibs.client import _SENTINEL
assert cache.get_cached("a") == 1
assert cache.get_cached("b") == 2
assert cache.get_cached("missing") is _SENTINEL
def test_lru_cache_eviction():
"""Oldest entry is evicted when maxsize is exceeded."""
cache = _LRUCache(maxsize=2)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3) # should evict "a"
from mcgibs.client import _SENTINEL
assert cache.get_cached("a") is _SENTINEL
assert cache.get_cached("b") == 2
assert cache.get_cached("c") == 3
def test_lru_cache_access_refreshes():
"""Accessing an entry moves it to the end, so the other is evicted first."""
cache = _LRUCache(maxsize=2)
cache.put("a", 1)
cache.put("b", 2)
cache.get_cached("a") # refresh "a"
cache.put("c", 3) # should evict "b" (oldest unused)
from mcgibs.client import _SENTINEL
assert cache.get_cached("a") == 1 # still present
assert cache.get_cached("b") is _SENTINEL # evicted
assert cache.get_cached("c") == 3
def test_lru_cache_update_existing():
"""Putting an existing key updates the value without increasing size."""
cache = _LRUCache(maxsize=2)
cache.put("a", 1)
cache.put("b", 2)
cache.put("a", 10) # update, not insert
assert len(cache) == 2
assert cache.get_cached("a") == 10
# ---------------------------------------------------------------------------
# WMS non-image response detection (H5 Hamilton fix)
# ---------------------------------------------------------------------------
@respx.mock
async def test_wms_rejects_non_image_response(capabilities_xml):
"""WMS should raise RuntimeError when server returns XML instead of image."""
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,
text='<?xml version="1.0"?><ServiceException>Layer not found</ServiceException>',
headers={"content-type": "application/xml"},
)
)
client = GIBSClient()
await client.initialize()
from mcgibs.models import BBox
bbox = BBox(west=-120.0, south=30.0, east=-110.0, north=40.0)
with pytest.raises(RuntimeError, match="non-image content-type"):
await client.get_wms_image("FakeLayer", "2025-01-01", bbox)
await client.close()
# ---------------------------------------------------------------------------
# Image dimension clamping (H6 Hamilton fix)
# ---------------------------------------------------------------------------
@respx.mock
async def test_wms_clamps_oversized_dimensions(capabilities_xml):
"""Oversized dimensions should be clamped to MAX_IMAGE_DIMENSION."""
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
return_value=httpx.Response(200, text=capabilities_xml)
)
buf = BytesIO()
Image.new("RGB", (10, 10), "red").save(buf, format="JPEG")
route = respx.get(url__regex=r".*wms\.cgi.*").mock(
return_value=httpx.Response(
200, content=buf.getvalue(), headers={"content-type": "image/jpeg"}
)
)
client = GIBSClient()
await client.initialize()
from mcgibs.models import BBox
bbox = BBox(west=-120.0, south=30.0, east=-110.0, north=40.0)
await client.get_wms_image("Test", "2025-01-01", bbox, width=99999, height=99999)
# Verify the request was made with clamped dimensions
request = route.calls.last.request
assert "WIDTH=4096" in str(request.url)
assert "HEIGHT=4096" in str(request.url)
await client.close()
# ---------------------------------------------------------------------------
# Client retry logic (Hamilton fix: exponential backoff)
# ---------------------------------------------------------------------------
@respx.mock
async def test_client_retries_on_capabilities_failure(capabilities_xml):
"""Client retries capabilities loading on HTTP errors, then succeeds."""
call_count = 0
def capabilities_handler(request):
nonlocal call_count
call_count += 1
if call_count < 3:
return httpx.Response(503, text="Service Unavailable")
return httpx.Response(200, text=capabilities_xml)
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(side_effect=capabilities_handler)
# Patch retry delays to avoid slow tests
import mcgibs.client as client_mod
original_delays = client_mod._INIT_RETRY_DELAYS
client_mod._INIT_RETRY_DELAYS = [0.01, 0.01, 0.01]
try:
client = GIBSClient()
await client.initialize()
assert len(client.layer_index) == 3
assert call_count == 3
await client.close()
finally:
client_mod._INIT_RETRY_DELAYS = original_delays
@respx.mock
async def test_client_raises_after_max_retries():
"""Client raises RuntimeError after exhausting all retry attempts."""
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
return_value=httpx.Response(503, text="Service Unavailable")
)
import mcgibs.client as client_mod
original_delays = client_mod._INIT_RETRY_DELAYS
client_mod._INIT_RETRY_DELAYS = [0.01, 0.01, 0.01]
try:
client = GIBSClient()
with pytest.raises(RuntimeError, match="capabilities unavailable"):
await client.initialize()
await client.close()
finally:
client_mod._INIT_RETRY_DELAYS = original_delays
# ---------------------------------------------------------------------------
# Client geocode caching (M7 Hamilton fix)
# ---------------------------------------------------------------------------
@respx.mock
async def test_client_caches_successful_geocode(capabilities_xml):
"""Successful geocode results are cached, preventing duplicate HTTP calls."""
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
return_value=httpx.Response(200, text=capabilities_xml)
)
route = respx.get(url__regex=r".*nominatim.*").mock(
return_value=httpx.Response(
200,
json=[
{
"display_name": "Tokyo, Japan",
"lat": "35.6762",
"lon": "139.6503",
"boundingbox": ["35.5191", "35.8170", "139.5601", "139.9200"],
"osm_type": "relation",
"importance": 0.82,
}
],
),
)
client = GIBSClient()
await client.initialize()
result1 = await client.resolve_place("Tokyo")
result2 = await client.resolve_place("Tokyo")
assert result1 is not None
assert result2 is result1 # exact same object from cache
assert route.call_count == 1 # only one HTTP call
await client.close()
# ---------------------------------------------------------------------------
# Client.http property guard
# ---------------------------------------------------------------------------
def test_client_http_not_initialized():
"""Accessing .http before initialize() raises RuntimeError."""
client = GIBSClient()
with pytest.raises(RuntimeError, match="not initialized"):
_ = client.http

View File

@ -141,3 +141,76 @@ def test_describe_rgb_basic(rgb: tuple[int, int, int], expected_substring: str):
"""Known RGB triples produce color names containing the expected word.""" """Known RGB triples produce color names containing the expected word."""
result = _describe_rgb(rgb) result = _describe_rgb(rgb)
assert expected_substring in result.lower() assert expected_substring in result.lower()
# ---------------------------------------------------------------------------
# _parse_rgb error handling (H3 Hamilton fix)
# ---------------------------------------------------------------------------
def test_parse_rgb_valid():
from mcgibs.colormaps import _parse_rgb
assert _parse_rgb("255,128,0") == (255, 128, 0)
def test_parse_rgb_too_few_parts():
from mcgibs.colormaps import _parse_rgb
assert _parse_rgb("255,128") == (0, 0, 0)
def test_parse_rgb_empty_string():
from mcgibs.colormaps import _parse_rgb
assert _parse_rgb("") == (0, 0, 0)
def test_parse_rgb_non_numeric():
from mcgibs.colormaps import _parse_rgb
assert _parse_rgb("abc,def,ghi") == (0, 0, 0)
# ---------------------------------------------------------------------------
# _parse_interval_value edge cases
# ---------------------------------------------------------------------------
def test_parse_interval_value_bare_number():
"""Bare number without brackets (used in sea ice colormaps)."""
assert _parse_interval_value("42") == (42.0, 42.0)
def test_parse_interval_value_empty_string():
assert _parse_interval_value("") == (None, None)
def test_parse_interval_value_scientific_notation():
low, high = _parse_interval_value("[1.5e2,2.0e2)")
assert low == 150.0
assert high == 200.0
# ---------------------------------------------------------------------------
# explain_colormap: classification colormap
# ---------------------------------------------------------------------------
def test_explain_colormap_classification():
"""Classification colormaps should produce categorical descriptions."""
from mcgibs.models import ColorMap, ColorMapEntry, ColorMapSet
entries = [
ColorMapEntry(rgb=(0, 0, 255), label="Water"),
ColorMapEntry(rgb=(0, 128, 0), label="Forest"),
ColorMapEntry(rgb=(255, 255, 0), label="Desert"),
]
cm = ColorMap(title="Land Cover", legend_type="classification", entries=entries)
cms = ColorMapSet(maps=[cm])
text = explain_colormap(cms)
assert "Land Cover" in text
assert "classification" in text
assert "Water" in text
assert "Forest" in text

71
tests/test_models.py Normal file
View File

@ -0,0 +1,71 @@
"""Tests for mcgibs.models — Pydantic model validation."""
import pytest
from pydantic import ValidationError
from mcgibs.models import BBox, ColorMap, ColorMapEntry, ColorMapSet
# ---------------------------------------------------------------------------
# BBox validation (M3 Hamilton fix)
# ---------------------------------------------------------------------------
def test_bbox_valid():
bbox = BBox(west=-120.0, south=30.0, east=-110.0, north=40.0)
assert bbox.west == -120.0
assert bbox.north == 40.0
def test_bbox_south_greater_than_north():
with pytest.raises(ValidationError, match=r"south.*must be <= north"):
BBox(west=0.0, south=50.0, east=10.0, north=40.0)
def test_bbox_latitude_out_of_range():
with pytest.raises(ValidationError, match="Latitudes must be in"):
BBox(west=0.0, south=-91.0, east=10.0, north=0.0)
def test_bbox_longitude_out_of_range():
with pytest.raises(ValidationError, match="Longitudes must be in"):
BBox(west=-181.0, south=0.0, east=0.0, north=10.0)
def test_bbox_wms_bbox_property():
bbox = BBox(west=-120.0, south=30.0, east=-110.0, north=40.0)
assert bbox.wms_bbox == "-120.0,30.0,-110.0,40.0"
def test_bbox_area_sq_deg():
bbox = BBox(west=0.0, south=0.0, east=10.0, north=10.0)
assert bbox.area_sq_deg == 100.0
def test_bbox_edge_values():
"""Extreme valid values: full globe."""
bbox = BBox(west=-180.0, south=-90.0, east=180.0, north=90.0)
assert bbox.area_sq_deg == 360.0 * 180.0
# ---------------------------------------------------------------------------
# ColorMapSet.data_map property
# ---------------------------------------------------------------------------
def test_data_map_returns_first_non_nodata():
"""data_map should return the first ColorMap that has non-nodata entries."""
nodata_only = ColorMap(
entries=[ColorMapEntry(rgb=(0, 0, 0), nodata=True)],
)
data_map = ColorMap(
title="Real Data",
entries=[ColorMapEntry(rgb=(255, 0, 0), value="[0,1)")],
)
cms = ColorMapSet(maps=[nodata_only, data_map])
assert cms.data_map is data_map
def test_data_map_empty_set():
"""data_map should return None for empty ColorMapSet."""
cms = ColorMapSet(maps=[])
assert cms.data_map is None