"""GET /api/place — Nominatim reverse proxy (60s cache, 500 keys, 1 req/s).""" from __future__ import annotations import asyncio import httpx import pytest from main import app from place import cache_key, place_cache, slim_place BASE = "http://test" SAMPLE = { "display_name": "Raleigh, Wake County, North Carolina, United States", "name": "Raleigh", "osm_type": "relation", "osm_id": 123, "address": { "city": "Raleigh", "state": "North Carolina", "country": "United States", "country_code": "us", "tourism": "ignore-me", }, } class _FakeResp: def __init__(self, payload, status=200): self._payload = payload self.status_code = status def raise_for_status(self): if self.status_code >= 400: req = httpx.Request("GET", "https://nominatim.openstreetmap.org/reverse") raise httpx.HTTPStatusError( "upstream", request=req, response=httpx.Response(self.status_code, request=req), ) def json(self): return self._payload class _FakeNominatim: calls: list[dict] = [] def __init__(self, *args, **kwargs): pass async def __aenter__(self): return self async def __aexit__(self, *args): return False async def get(self, url, params=None, headers=None): _FakeNominatim.calls.append({"url": url, "params": params, "headers": headers}) return _FakeResp(SAMPLE) def _nominatim_client(**kwargs): return _FakeNominatim() async def _get(path: str) -> httpx.Response: transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: return await client.get(path) @pytest.fixture(autouse=True) def _reset_place(monkeypatch): place_cache.clear() _FakeNominatim.calls = [] monkeypatch.setattr("place._http_client", _nominatim_client) monkeypatch.setattr("place.NOMINATIM_MIN_INTERVAL", 0.0) monkeypatch.setattr("place._last_req", 0.0) yield place_cache.clear() def test_slim_place_keeps_address_subset(): body = slim_place(35.78, -78.64, SAMPLE) assert body["display_name"].startswith("Raleigh") assert body["name"] == "Raleigh" assert body["address"]["city"] == "Raleigh" assert "tourism" not in body["address"] assert body["attribution"].startswith("© OpenStreetMap") def test_cache_key_quantizes_to_4_decimals(): assert cache_key(35.77961, -78.63821) == cache_key(35.77964, -78.63819) def test_place_requires_lat_lon(): resp = asyncio.run(_get("/api/place")) assert resp.status_code == 422 def test_place_rejects_out_of_range(): assert asyncio.run(_get("/api/place?lat=99&lon=0")).status_code == 422 assert asyncio.run(_get("/api/place?lat=0&lon=200")).status_code == 422 def test_place_reverse_and_cache(): r1 = asyncio.run(_get("/api/place?lat=35.7796&lon=-78.6382")) assert r1.status_code == 200 body = r1.json() assert body["display_name"].startswith("Raleigh") assert body["lat"] == pytest.approx(35.7796, abs=0.001) assert "max-age=60" in (r1.headers.get("cache-control") or "").lower() assert len(_FakeNominatim.calls) == 1 ua = _FakeNominatim.calls[0]["headers"]["User-Agent"] assert "osint-dashboard" in ua.lower() or "@" in ua r2 = asyncio.run(_get("/api/place?lat=35.77961&lon=-78.63821")) assert r2.status_code == 200 assert len(_FakeNominatim.calls) == 1 # cache hit, same 4-decimal key def test_place_cache_cap_500(): from cachetools import TTLCache assert isinstance(place_cache, TTLCache) assert place_cache.maxsize == 500 assert place_cache.ttl == 60