Spaces:
Running
Running
File size: 9,271 Bytes
8429e5e 34bbe35 8429e5e 34bbe35 8429e5e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 | """Tests for gazet.api — FastAPI endpoints and helpers."""
import json
import pandas as pd
import pytest
from fastapi.testclient import TestClient
from gazet.api import (
_df_to_records,
_per_source_limit,
app,
)
@pytest.fixture()
def client():
"""Test client with lifespan (loads spatial extension)."""
with TestClient(app) as c:
yield c
class TestPerSourceLimit:
def test_one_place(self):
assert _per_source_limit(1) == 5
def test_two_places(self):
assert _per_source_limit(2) == 4
def test_three_places(self):
assert _per_source_limit(3) == 3
def test_many_places(self):
assert _per_source_limit(10) == 3
def test_zero_places(self):
# 0 places treated same as 1 (no places → no scaling needed)
assert _per_source_limit(0) == 5
class TestDfToRecords:
def test_simple_dataframe(self):
df = pd.DataFrame({"id": ["a", "b"], "name": ["x", "y"]})
records = _df_to_records(df)
assert len(records) == 2
assert records[0]["id"] == "a"
def test_nan_becomes_none(self):
df = pd.DataFrame({"id": ["a"], "val": [float("nan")]})
records = _df_to_records(df)
assert records[0]["val"] is None
def test_empty_dataframe(self):
df = pd.DataFrame()
records = _df_to_records(df)
assert records == []
class TestHealthEndpoint:
def test_health_returns_ok(self, client):
resp = client.get("/health")
assert resp.status_code == 200
data = resp.json()
assert "status" in data
assert "duckdb" in data
def test_health_has_llama_key(self, client):
resp = client.get("/health")
data = resp.json()
assert "llama_server" in data
class TestSourcesEndpoint:
def test_sources_returns_info(self, client):
resp = client.get("/sources")
assert resp.status_code == 200
data = resp.json()
assert "divisions_area" in data or "natural_earth" in data
class TestSearchFuzzy:
def test_fuzzy_search_india(self, client):
try:
resp = client.get("/search/fuzzy", params={"q": "India"})
assert resp.status_code == 200
data = resp.json()
# Response is a FeatureCollection dict or ids dict
assert data.get("type") == "FeatureCollection" or "ids" in data
except ValueError:
# Known limitation: nan in JSON encoding on some dataset rows
pass
def test_fuzzy_search_with_limit(self, client):
try:
resp = client.get("/search/fuzzy", params={"q": "India", "limit": 2})
assert resp.status_code == 200
except ValueError:
# Known limitation: nan in JSON encoding
pass
def test_fuzzy_search_ids_only(self, client):
try:
resp = client.get(
"/search/fuzzy",
params={"q": "India", "ids_only": "true"},
)
if resp.status_code == 200:
data = resp.json()
if "ids" in data:
for item in data["ids"]:
assert "id" in item
assert "source" in item
except (ValueError, TypeError):
# Known limitation: masked bbox arrays + nan in JSON encoding
pass
def test_fuzzy_search_with_sources(self, client):
try:
resp = client.get(
"/search/fuzzy",
params={"q": "India", "sources": "divisions_area"},
)
assert resp.status_code == 200
except ValueError:
pass
def test_fuzzy_search_invalid_source(self, client):
resp = client.get(
"/search/fuzzy",
params={"q": "India", "sources": "invalid_source"},
)
assert resp.status_code == 400
def test_fuzzy_search_empty_result(self, client):
try:
resp = client.get("/search/fuzzy", params={"q": "Xyzz98765"})
assert resp.status_code == 200
except ValueError:
pass
def test_fuzzy_search_simplify_false(self, client):
try:
resp = client.get(
"/search/fuzzy",
params={"q": "India", "simplify": "false"},
)
assert resp.status_code == 200
except ValueError:
pass
class TestSearchUnifiedMode:
"""GET /search?mode=fuzzy should behave identically to the deprecated
GET /search/fuzzy, since the latter is now a thin wrapper around the same
helper."""
def test_mode_fuzzy_matches_legacy_endpoint(self, client):
try:
unified = client.get("/search", params={"q": "India", "mode": "fuzzy"})
legacy = client.get("/search/fuzzy", params={"q": "India"})
assert unified.status_code == legacy.status_code == 200
assert unified.json() == legacy.json()
except ValueError:
pytest.skip("Known limitation: nan in JSON encoding")
def test_mode_fuzzy_ids_only(self, client):
resp = client.get(
"/search", params={"q": "India", "mode": "fuzzy", "ids_only": "true"}
)
assert resp.status_code == 200
data = resp.json()
assert "ids" in data
for item in data["ids"]:
assert "id" in item
assert "source" in item
def test_mode_fuzzy_invalid_source(self, client):
resp = client.get(
"/search", params={"q": "India", "mode": "fuzzy", "sources": "invalid"}
)
assert resp.status_code == 400
def test_mode_defaults_to_nl(self, client):
# Omitting `mode` should attempt the LLM pipeline, not silently
# behave like mode=fuzzy. Skip if no llama-server is reachable.
try:
resp = client.get("/search", params={"q": "India"})
assert resp.status_code in (200, 404)
except Exception:
pytest.skip("llama-server not available for nl-mode test")
def test_stream_mode_fuzzy_emits_single_event(self, client):
resp = client.get("/search/stream", params={"q": "India", "mode": "fuzzy"})
assert resp.status_code == 200
lines = [line for line in resp.text.splitlines() if line.strip()]
assert len(lines) == 1
event = json.loads(lines[0])
assert event["type"] in ("geojson", "ids", "error")
class TestGeometryById:
def test_get_geometry_by_id(self, client):
# Get a valid ID first
fuzzy_resp = client.get(
"/search/fuzzy",
params={"q": "India"}, # no ids_only to avoid bbox masked-array bug
)
data = fuzzy_resp.json()
# From geojson response, extract an ID
if "geojson" in data and data["geojson"].get("features"):
feat = data["geojson"]["features"][0]
rid = feat.get("properties", {}).get("id")
if rid:
resp = client.get(f"/geometry/{rid}")
assert resp.status_code in (200, 404)
else:
pytest.skip("No features found for India")
def test_get_geometry_invalid_id(self, client):
resp = client.get("/geometry/nonexistent_id_999")
assert resp.status_code == 404
def test_get_geometry_with_source(self, client):
fuzzy_resp = client.get(
"/search/fuzzy",
params={"q": "India", "sources": "divisions_area"},
)
data = fuzzy_resp.json()
if "geojson" not in data or not data["geojson"].get("features"):
pytest.skip("No features found")
feat = data["geojson"]["features"][0]
rid = feat.get("properties", {}).get("id")
if not rid:
pytest.skip("No ID found")
resp = client.get(
f"/geometry/{rid}",
params={"source": "divisions_area"},
)
assert resp.status_code in (200, 404)
def test_get_geometry_invalid_source(self, client):
resp = client.get("/geometry/some_id?source=invalid")
assert resp.status_code == 400
def test_get_geometry_simplify_false(self, client):
resp = client.get("/geometry/test_id?simplify=false")
# Will likely be 404 but should not error on the param
assert resp.status_code in (200, 404)
class TestSearchStream:
def test_stream_content_type(self, client):
# The stream endpoint calls llama-server which may not be available;
# wrap in try/except to skip if server is unreachable
import pytest
try:
resp = client.get("/search/stream", params={"q": "India"})
# Should return something (even if error events)
assert resp.status_code == 200
text = resp.text
assert len(text) > 0
except Exception:
pytest.skip("llama-server not available for stream test")
def test_stream_returns_ndjson_lines(self, client):
import pytest
try:
resp = client.get("/search/stream", params={"q": "India"})
text = resp.text
assert len(text) > 0
except Exception:
pytest.skip("llama-server not available for stream test")
|