Ryan Malloy
4a5035ca52
GIBS capabilities list .svg legend URLs, but the Anthropic API only accepts raster image types. GIBS hosts PNG versions at the same path, so we rewrite .svg → .png and validate content-type on both legend fetch paths (direct URL and WMS GetLegendGraphic fallback).
488 lines
15 KiB
Python
488 lines
15 KiB
Python
"""Tests for GIBSClient using respx mocks — no real HTTP calls."""
|
|
|
|
from io import BytesIO
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from PIL import Image
|
|
|
|
from mcgibs.client import GIBSClient, _LRUCache
|
|
from mcgibs.constants import WMTS_TILE_URL
|
|
|
|
|
|
@respx.mock
|
|
async def test_client_initialize(capabilities_xml):
|
|
"""Loading capabilities populates layer_index with all three sample layers."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
client = GIBSClient()
|
|
await client.initialize()
|
|
|
|
assert len(client.layer_index) == 3
|
|
assert "MODIS_Terra_CorrectedReflectance_TrueColor" in client.layer_index
|
|
assert "AMSR2_Sea_Ice_Concentration_12km_Monthly" in client.layer_index
|
|
assert "AIRS_L3_Surface_Air_Temperature_Daily_Day" in client.layer_index
|
|
|
|
# Spot-check a parsed layer
|
|
modis = client.layer_index["MODIS_Terra_CorrectedReflectance_TrueColor"]
|
|
assert modis.title == "Corrected Reflectance (True Color, Terra/MODIS)"
|
|
assert "image/jpeg" in modis.formats
|
|
assert modis.time is not None
|
|
assert modis.time.start == "2000-02-24"
|
|
assert modis.has_colormap is False
|
|
|
|
await client.close()
|
|
|
|
|
|
@respx.mock
|
|
async def test_client_fetch_layer_metadata(capabilities_xml, layer_metadata_json):
|
|
"""Fetching layer metadata enriches the layer with instrument/platform fields."""
|
|
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)
|
|
)
|
|
|
|
client = GIBSClient()
|
|
await client.initialize()
|
|
|
|
data = await client.fetch_layer_metadata(layer_id)
|
|
|
|
assert data["instrument"] == "MODIS"
|
|
assert data["platform"] == "Terra"
|
|
assert data["ongoing"] is True
|
|
|
|
# Verify the layer_index entry was enriched
|
|
layer = client.get_layer(layer_id)
|
|
assert layer is not None
|
|
assert layer.instrument == "MODIS"
|
|
assert layer.platform == "Terra"
|
|
assert layer.measurement == "Corrected Reflectance"
|
|
assert layer.day_night == "Day"
|
|
|
|
# Second call should use cache
|
|
data2 = await client.fetch_layer_metadata(layer_id)
|
|
assert data2 is data
|
|
|
|
await client.close()
|
|
|
|
|
|
@respx.mock
|
|
async def test_client_fetch_colormap(capabilities_xml, colormap_xml):
|
|
"""Parsing colormap XML produces entries and legend data."""
|
|
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)
|
|
)
|
|
|
|
client = GIBSClient()
|
|
await client.initialize()
|
|
|
|
colormap_set = await client.fetch_colormap("AIRS_L3_Surface_Air_Temperature_Daily_Day")
|
|
|
|
assert colormap_set is not None
|
|
assert len(colormap_set.maps) == 2
|
|
|
|
data_map = colormap_set.data_map
|
|
assert data_map is not None
|
|
assert data_map.title == "Surface Air Temperature"
|
|
assert data_map.units == "K"
|
|
assert len(data_map.entries) == 14
|
|
assert data_map.legend_type == "continuous"
|
|
|
|
# Verify the nodata map
|
|
nodata_map = colormap_set.maps[1]
|
|
assert any(e.nodata for e in nodata_map.entries)
|
|
|
|
# Second fetch should be cached
|
|
cached = await client.fetch_colormap("AIRS_L3_Surface_Air_Temperature_Daily_Day")
|
|
assert cached is colormap_set
|
|
|
|
await client.close()
|
|
|
|
|
|
@respx.mock
|
|
async def test_client_get_wms_image(capabilities_xml):
|
|
"""WMS GetMap returns raw image bytes when content-type is image/*."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
|
|
# Build a tiny valid JPEG to return as the mock response
|
|
buf = BytesIO()
|
|
Image.new("RGB", (10, 10), "blue").save(buf, format="JPEG")
|
|
fake_jpeg = buf.getvalue()
|
|
|
|
respx.get(url__regex=r".*wms\.cgi.*").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
content=fake_jpeg,
|
|
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)
|
|
result = await client.get_wms_image(
|
|
"MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
"2025-06-01",
|
|
bbox,
|
|
)
|
|
|
|
assert isinstance(result, bytes)
|
|
assert len(result) > 0
|
|
# Verify the returned bytes are valid JPEG (starts with FFD8)
|
|
assert result[:2] == b"\xff\xd8"
|
|
|
|
await client.close()
|
|
|
|
|
|
def test_client_build_tile_url():
|
|
"""build_tile_url produces the expected WMTS REST URL format."""
|
|
client = GIBSClient()
|
|
|
|
url = client.build_tile_url(
|
|
layer_id="MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
date="2025-06-01",
|
|
zoom=3,
|
|
row=2,
|
|
col=4,
|
|
tile_matrix_set="250m",
|
|
ext="jpg",
|
|
epsg="4326",
|
|
)
|
|
|
|
assert "MODIS_Terra_CorrectedReflectance_TrueColor" in url
|
|
assert "2025-06-01" in url
|
|
assert "/250m/" in url
|
|
assert "/3/" in url
|
|
assert "/2/" in url
|
|
assert "/4.jpg" in url
|
|
assert "epsg4326" in url
|
|
|
|
# Verify it matches the constant template
|
|
expected = WMTS_TILE_URL.format(
|
|
epsg="4326",
|
|
layer_id="MODIS_Terra_CorrectedReflectance_TrueColor",
|
|
date="2025-06-01",
|
|
tile_matrix_set="250m",
|
|
z=3,
|
|
row=2,
|
|
col=4,
|
|
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='<?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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Legend SVG → PNG rewriting (live API compatibility fix)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_legend_rewrites_svg_to_png(capabilities_xml):
|
|
"""Legend URLs ending in .svg should be rewritten to .png."""
|
|
# Modify the capabilities XML so AMSR2's legend URL ends in .svg
|
|
patched_xml = capabilities_xml.replace(
|
|
"AMSR2_Sea_Ice_Concentration_12km.png",
|
|
"AMSR2_Sea_Ice_Concentration_12km.svg",
|
|
)
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=patched_xml)
|
|
)
|
|
|
|
buf = BytesIO()
|
|
Image.new("RGB", (200, 20), "white").save(buf, format="PNG")
|
|
png_route = respx.get(url__regex=r".*legends/.*\.png").mock(
|
|
return_value=httpx.Response(
|
|
200, content=buf.getvalue(), headers={"content-type": "image/png"}
|
|
)
|
|
)
|
|
|
|
client = GIBSClient()
|
|
await client.initialize()
|
|
|
|
result = await client.get_legend_image("AMSR2_Sea_Ice_Concentration_12km_Monthly")
|
|
assert result is not None
|
|
# Verify the .png URL was requested, not .svg
|
|
assert png_route.call_count == 1
|
|
assert ".png" in str(png_route.calls.last.request.url)
|
|
|
|
await client.close()
|
|
|
|
|
|
@respx.mock
|
|
async def test_legend_rejects_svg_content(capabilities_xml):
|
|
"""If the legend URL returns SVG content, it should be rejected."""
|
|
respx.get(url__regex=r".*WMTSCapabilities\.xml").mock(
|
|
return_value=httpx.Response(200, text=capabilities_xml)
|
|
)
|
|
respx.get(url__regex=r".*legends/.*\.png").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
content=b'<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN"...',
|
|
headers={"content-type": "image/svg+xml"},
|
|
)
|
|
)
|
|
# WMS fallback also returns non-image
|
|
respx.get(url__regex=r".*wms\.cgi.*").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
text="<ServiceException>not supported</ServiceException>",
|
|
headers={"content-type": "application/xml"},
|
|
)
|
|
)
|
|
|
|
client = GIBSClient()
|
|
await client.initialize()
|
|
|
|
result = await client.get_legend_image("AMSR2_Sea_Ice_Concentration_12km_Monthly")
|
|
assert result is None
|
|
|
|
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
|