Ryan Malloy
d163ade9a4
Address 20 findings from safety-critical review: CRITICAL: - C1: Replace ET.fromstring with defusedxml across all XML parsers - C2: Fix client init failure leaving half-initialized state; clean up HTTP client on startup failure so next connection can retry HIGH: - H1: Replace unbounded dicts with LRU caches (maxsize=500) - H2: Move Nominatim rate limiter from module globals to per-instance state on GIBSClient, eliminating shared mutable state - H3: Validate _parse_rgb input, return (0,0,0) on malformed data - H4: Add exponential backoff retry for capabilities loading - H5: Invert WMS error detection to verify image content-type - H6: Clamp image dimensions to 4096 max to prevent OOM MEDIUM: - M1: Convert images to RGB mode in compare_dates for RGBA safety - M2: Narrow DescribeDomains XML matching to TimeDomain elements - M3: Add BBox model_validator for coordinate range validation - M4: Add ET.ParseError to colormap fetch exception handling - M5: Replace bare except Exception with specific types in server - M6: Catch ValueError from _resolve_bbox in imagery tools for consistent error returns - M7: Only cache successful geocoding lookups (no negative caching) LOW: - L3: Derive USER_AGENT version from package __version__ - L5: Remove unused start_date/end_date params from check_layer_dates
152 lines
4.5 KiB
Python
152 lines
4.5 KiB
Python
"""Tests for mcgibs.geo — geocoding and bbox helpers."""
|
|
|
|
import httpx
|
|
import respx
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# geocode() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_geocode_success():
|
|
respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(200, json=[TOKYO_HIT]),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
result = await geocode(client, "Tokyo")
|
|
|
|
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():
|
|
respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(200, json=[]),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
result = await geocode(client, "xyznonexistent")
|
|
|
|
assert result is None
|
|
|
|
|
|
@respx.mock
|
|
async def test_geocode_repeated_calls():
|
|
"""Each geocode() call makes an HTTP request (caching is caller's responsibility)."""
|
|
route = respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(200, json=[TOKYO_HIT]),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
first = await geocode(client, "Tokyo")
|
|
second = await geocode(client, "Tokyo")
|
|
|
|
# geocode() no longer caches — GIBSClient.resolve_place() handles that
|
|
assert route.call_count == 2
|
|
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."""
|
|
respx.get(NOMINATIM_URL).mock(
|
|
return_value=httpx.Response(500, text="Internal Server Error"),
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
result = await geocode(client, "ServerError")
|
|
|
|
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
|