osint-dashboard/tests/test_api_place.py

"""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