From d13ba744d316269441b93d694fea096068a295a6 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 18 Feb 2026 18:19:21 -0700 Subject: [PATCH] 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. --- src/mcgibs/server.py | 24 ++++ tests/test_capabilities.py | 41 ++++++- tests/test_client.py | 237 ++++++++++++++++++++++++++++++++++++- tests/test_colormaps.py | 73 ++++++++++++ tests/test_models.py | 71 +++++++++++ 5 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 tests/test_models.py diff --git a/src/mcgibs/server.py b/src/mcgibs/server.py index d340707..29101f1 100644 --- a/src/mcgibs/server.py +++ b/src/mcgibs/server.py @@ -8,6 +8,8 @@ IMPORTANT: Do NOT use `from __future__ import annotations` here. FastMCP needs runtime types for Literal, list[str], etc. """ +import asyncio +import atexit import base64 import json import logging @@ -44,6 +46,28 @@ def _get_client() -> GIBSClient: 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 --- diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py index 569c4ea..afe91a5 100644 --- a/tests/test_capabilities.py +++ b/tests/test_capabilities.py @@ -1,6 +1,10 @@ """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): @@ -102,3 +106,38 @@ def test_parse_capabilities_resource_url(capabilities_xml: str): "{Time}/{TileMatrixSet}/{TileMatrix}/{TileRow}/{TileCol}.jpg" ) 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 diff --git a/tests/test_client.py b/tests/test_client.py index e85c5f4..8e3c692 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -3,10 +3,11 @@ from io import BytesIO import httpx +import pytest import respx from PIL import Image -from mcgibs.client import GIBSClient +from mcgibs.client import GIBSClient, _LRUCache from mcgibs.constants import WMTS_TILE_URL @@ -182,3 +183,237 @@ def test_client_build_tile_url(): ext="jpg", ) 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='Layer not found', + 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 diff --git a/tests/test_colormaps.py b/tests/test_colormaps.py index 60dde62..6919b1e 100644 --- a/tests/test_colormaps.py +++ b/tests/test_colormaps.py @@ -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.""" result = _describe_rgb(rgb) 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 diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..b6841cd --- /dev/null +++ b/tests/test_models.py @@ -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