Ryan Malloy
f7fad32a9e
Complete implementation of all modules: - constants.py: GIBS API endpoints, projections, TileMatrixSet defs - models.py: Pydantic models for layers, colormaps, geocoding - geo.py: Nominatim geocoding with rate limiting and caching - capabilities.py: WMTS GetCapabilities XML parser with search - colormaps.py: Colormap v1.3 parser with natural-language summaries - client.py: Async GIBS HTTP client wrapping all API interactions - server.py: FastMCP 3.0 tools, resources, and prompts 11 MCP tools, 3 resources, 2 prompts. 47 tests, all passing.
169 lines
4.8 KiB
Python
169 lines
4.8 KiB
Python
"""Tests for mcgibs.geo — geocoding, bbox helpers, and caching."""
|
|
|
|
import httpx
|
|
import respx
|
|
|
|
import mcgibs.geo as _geo_module
|
|
from mcgibs.geo import bbox_from_point, expand_bbox, geocode
|
|
from mcgibs.models import BBox
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
|
|
|
|
TOKYO_HIT = {
|
|
"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,
|
|
}
|
|
|
|
|
|
def _reset_rate_limit() -> None:
|
|
"""Reset the module-level rate-limit timestamp so tests don't stall."""
|
|
_geo_module._last_request_time = 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# geocode() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_geocode_success():
|
|
_reset_rate_limit()
|
|
|
|
respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(200, json=[TOKYO_HIT]),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
cache: dict = {}
|
|
result = await geocode(client, "Tokyo", cache)
|
|
|
|
assert result is not None
|
|
assert result.display_name == "Tokyo, Japan"
|
|
assert result.lat == 35.6762
|
|
assert result.lon == 139.6503
|
|
assert result.osm_type == "relation"
|
|
assert result.importance == 0.82
|
|
|
|
# BBox built from Nominatim's [south, north, west, east] order
|
|
assert result.bbox.south == 35.5191
|
|
assert result.bbox.north == 35.8170
|
|
assert result.bbox.west == 139.5601
|
|
assert result.bbox.east == 139.9200
|
|
|
|
|
|
@respx.mock
|
|
async def test_geocode_no_results():
|
|
_reset_rate_limit()
|
|
|
|
respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(200, json=[]),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
cache: dict = {}
|
|
result = await geocode(client, "xyznonexistent", cache)
|
|
|
|
assert result is None
|
|
|
|
|
|
@respx.mock
|
|
async def test_geocode_caching():
|
|
"""Second call with the same query must be served from cache — no extra HTTP request."""
|
|
_reset_rate_limit()
|
|
|
|
route = respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(200, json=[TOKYO_HIT]),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
cache: dict = {}
|
|
first = await geocode(client, "Tokyo", cache)
|
|
second = await geocode(client, "Tokyo", cache)
|
|
|
|
assert route.call_count == 1
|
|
assert first is not None
|
|
assert second is not None
|
|
assert first.display_name == second.display_name
|
|
|
|
|
|
@respx.mock
|
|
async def test_geocode_http_error():
|
|
"""A 500 response should return None without raising an exception."""
|
|
_reset_rate_limit()
|
|
|
|
respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(500, text="Internal Server Error"),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
cache: dict = {}
|
|
result = await geocode(client, "ServerError", cache)
|
|
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# expand_bbox() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_expand_bbox():
|
|
bbox = BBox(west=10.0, south=20.0, east=20.0, north=30.0)
|
|
expanded = expand_bbox(bbox, factor=0.1)
|
|
|
|
# Width is 10 deg, so 10% padding = 1 deg each side.
|
|
assert expanded.west < bbox.west
|
|
assert expanded.east > bbox.east
|
|
assert expanded.south < bbox.south
|
|
assert expanded.north > bbox.north
|
|
|
|
# Verify exact amounts (width = 10, dlon = 1; height = 10, dlat = 1).
|
|
assert expanded.west == 9.0
|
|
assert expanded.east == 21.0
|
|
assert expanded.south == 19.0
|
|
assert expanded.north == 31.0
|
|
|
|
|
|
def test_expand_bbox_clamping():
|
|
"""Expanding a bbox near the poles must clamp latitude to [-90, 90]."""
|
|
polar = BBox(west=-170.0, south=85.0, east=170.0, north=89.5)
|
|
expanded = expand_bbox(polar, factor=0.5)
|
|
|
|
assert expanded.north <= 90.0
|
|
assert expanded.south >= -90.0
|
|
# Longitude should also stay within bounds.
|
|
assert expanded.west >= -180.0
|
|
assert expanded.east <= 180.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# bbox_from_point() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_bbox_from_point():
|
|
bbox = bbox_from_point(35.0, 139.0, radius_deg=1.0)
|
|
|
|
assert bbox.south == 34.0
|
|
assert bbox.north == 36.0
|
|
assert bbox.west == 138.0
|
|
assert bbox.east == 140.0
|
|
|
|
|
|
def test_bbox_from_point_clamping():
|
|
"""A point near the pole should have its bbox clamped to valid lat range."""
|
|
bbox = bbox_from_point(89.5, 0.0, radius_deg=2.0)
|
|
|
|
assert bbox.north == 90.0 # clamped, not 91.5
|
|
assert bbox.south == 87.5
|
|
assert bbox.west == -2.0
|
|
assert bbox.east == 2.0
|