a82b62a011
Curated static catalogue of 13 conflict theatres (Ukraine, Gaza, Sudan,
Myanmar, DRC, Yemen, Syria, Lebanon, Sahel, Somalia, Red Sea, Taiwan
Strait, Korean DMZ) with war|high|elevated severity and short factual
descriptions. GET /api/conflicts returns { zones: [{id,label,severity,
lat,lon,description,eventCount,lastUpdated}], timestamp } where
eventCount rolls up pre-existing geocoded news/GDELT//api/news/map rows
inside each zone bbox (0 when the tables are empty). Adds the conflicts
entry to overlay_catalog(). No upstream scraping, no jittered coords.
132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
"""GET /api/conflicts — curated conflict-zone catalog + event-count roll-up.
|
|
|
|
No outbound HTTP: event counts come from geocoded rows already (or not) in the
|
|
DB, and the API tests monkeypatch ``main._fetch_geocoded_points`` so no database
|
|
is required for the contract checks.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from conflicts import SEVERITIES, conflict_zones, zone_event_stats
|
|
from live_layers import overlay_catalog
|
|
from main import app
|
|
|
|
BASE = "http://test"
|
|
|
|
|
|
def _get(path: str, monkeypatch=None, points=None) -> httpx.Response:
|
|
import asyncio
|
|
|
|
async def run() -> httpx.Response:
|
|
if monkeypatch is not None:
|
|
async def fake():
|
|
return points or []
|
|
|
|
monkeypatch.setattr("main._fetch_geocoded_points", fake)
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
|
|
return await client.get(path)
|
|
|
|
return asyncio.run(run())
|
|
|
|
|
|
# ── Catalog shape ──────────────────────────────────────────────────────
|
|
|
|
|
|
def test_catalog_length():
|
|
zones = conflict_zones()
|
|
assert len(zones) == 13
|
|
|
|
|
|
def test_catalog_severity_enum():
|
|
zones = conflict_zones()
|
|
sevs = {z["severity"] for z in zones}
|
|
assert sevs.issubset(SEVERITIES)
|
|
# All three tiers are represented.
|
|
assert sevs == SEVERITIES
|
|
|
|
|
|
def test_catalog_fields_factual_and_complete():
|
|
zones = conflict_zones()
|
|
ids = [z["id"] for z in zones]
|
|
assert len(set(ids)) == len(ids) # unique ids
|
|
for z in zones:
|
|
assert z["label"]
|
|
assert z["description"].strip()
|
|
assert -90.0 <= z["lat"] <= 90.0
|
|
assert -180.0 <= z["lon"] <= 180.0
|
|
# internal-only bbox is well-formed: (min_lat, min_lon, max_lat, max_lon)
|
|
min_lat, min_lon, max_lat, max_lon = z["bbox"]
|
|
assert min_lat <= max_lat and min_lon <= max_lon
|
|
assert min_lat <= z["lat"] <= max_lat and min_lon <= z["lon"] <= max_lon
|
|
|
|
|
|
def test_overlay_catalog_has_conflicts():
|
|
entry = overlay_catalog()["conflicts"]
|
|
assert entry["kind"] == "points"
|
|
assert entry["endpoint"] == "/api/conflicts"
|
|
|
|
|
|
# ── Pure counting ──────────────────────────────────────────────────────
|
|
|
|
TS1 = datetime(2026, 8, 30, 12, 0, tzinfo=timezone.utc)
|
|
TS2 = datetime(2026, 8, 30, 13, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_zone_event_stats_counts_and_picks_latest():
|
|
bbox = (40.0, 20.0, 52.0, 40.0) # roughly Ukraine
|
|
points = [
|
|
(50.45, 30.52, TS1), # inside
|
|
(48.0, 25.0, TS2), # inside, later
|
|
(0.0, -60.0, TS1), # outside
|
|
(15.0, 45.0, TS2), # outside (lat ok, lon out)
|
|
]
|
|
count, latest = zone_event_stats(points, bbox)
|
|
assert count == 2
|
|
assert latest == TS2
|
|
|
|
|
|
def test_zone_event_stats_empty_bbox():
|
|
count, latest = zone_event_stats([], (0.0, 0.0, 1.0, 1.0))
|
|
assert count == 0
|
|
assert latest is None
|
|
|
|
|
|
# ── API contract (mocked map items, no DB) ─────────────────────────────
|
|
|
|
|
|
def test_conflicts_returns_catalog_with_mocked_counts(monkeypatch):
|
|
points = [
|
|
(50.45, 30.52, TS1), # Ukraine
|
|
(25.03, 121.56, TS2), # Taiwan Strait
|
|
(0.0, -60.0, TS1), # nowhere
|
|
]
|
|
resp = _get("/api/conflicts", monkeypatch=monkeypatch, points=points)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "zones" in body and "timestamp" in body
|
|
by_id = {z["id"]: z for z in body["zones"]}
|
|
assert len(body["zones"]) == 13
|
|
|
|
zone = by_id["ukraine"]
|
|
assert zone["eventCount"] == 1
|
|
assert zone["lastUpdated"] == TS1.isoformat().replace("+00:00", "Z")
|
|
assert zone["severity"] == "war"
|
|
|
|
assert by_id["taiwan_strait"]["eventCount"] == 1
|
|
assert by_id["gaza"]["eventCount"] == 0
|
|
# exact per-zone key contract the frontend consumes
|
|
assert set(zone.keys()) == {
|
|
"id", "label", "severity", "lat", "lon",
|
|
"description", "eventCount", "lastUpdated",
|
|
}
|
|
|
|
|
|
def test_conflicts_empty_db_yields_zero_counts(monkeypatch):
|
|
resp = _get("/api/conflicts", monkeypatch=monkeypatch, points=[])
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert all(z["eventCount"] == 0 for z in body["zones"])
|
|
assert all(z["lastUpdated"] is None for z in body["zones"])
|