ef8b877d21
- GET /api/map/chokepoints: static five-strait fly-to catalog (Hormuz, Bab el-Mandeb, Suez, Malacca, Taiwan) with bbox/center/zoom/vesselapi. No VesselAPI calls; only Hormuz flags vesselapi=true. - GET /api/vessels?src=aisstream|vesselapi|all (default all) filters the union store by provider so a Hormuz view skips ~5k CONUS AISStream rows. - Tests: span validation of all five boxes, catalog 200 shape, src filter.
122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""Tests for the chokepoint preset catalog + vessels ``src=`` filter.
|
|
|
|
- Span: every catalog box passes VesselAPI's ``|dLat|+|dLon| <= 4`` validator.
|
|
- Catalog: ``GET /api/map/chokepoints`` returns 200 with the documented shape.
|
|
- Vessels filter: ``GET /api/vessels?src=`` narrows the union store by provider.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import httpx
|
|
|
|
from chokepoints import chokepoints
|
|
from live_layers import fetch_vessels, vessel_last_known
|
|
from main import app
|
|
from vesselapi import validate_bbox_span
|
|
|
|
BASE = "http://test"
|
|
|
|
|
|
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)
|
|
|
|
|
|
# ── Span validation (VesselAPI rule) ──────────────────────────────────────
|
|
|
|
def test_all_catalog_boxes_within_span() -> None:
|
|
for preset in chokepoints():
|
|
minlat, minlon, maxlat, maxlon = (float(p) for p in preset["bbox"].split(","))
|
|
dlat = abs(maxlat - minlat)
|
|
dlon = abs(maxlon - minlon)
|
|
assert dlat + dlon <= 4.0, preset["id"]
|
|
validate_bbox_span(minlat, minlon, maxlat, maxlon) # no raise
|
|
|
|
|
|
# ── Catalog API contract ──────────────────────────────────────────────────
|
|
|
|
def test_chokepoints_catalog_shape() -> None:
|
|
resp = asyncio.run(_get("/api/map/chokepoints"))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert set(body) == {"chokepoints"}
|
|
rows = body["chokepoints"]
|
|
assert [r["id"] for r in rows] == [
|
|
"hormuz", "bab_el_mandeb", "suez", "malacca", "taiwan",
|
|
]
|
|
for r in rows:
|
|
assert set(r) == {"id", "title", "bbox", "center", "zoom", "vesselapi"}
|
|
assert isinstance(r["center"], list) and len(r["center"]) == 2
|
|
assert r["zoom"] == 9
|
|
assert isinstance(r["vesselapi"], bool)
|
|
# bbox is minlat,minlon,maxlat,maxlon
|
|
minlat, minlon, maxlat, maxlon = (float(p) for p in r["bbox"].split(","))
|
|
assert minlat < maxlat and minlon < maxlon
|
|
|
|
|
|
def test_only_hormuz_is_vesselapi() -> None:
|
|
rows = chokepoints()
|
|
by_id = {r["id"]: r for r in rows}
|
|
assert by_id["hormuz"]["vesselapi"] is True
|
|
for cid in ("bab_el_mandeb", "suez", "malacca", "taiwan"):
|
|
assert by_id[cid]["vesselapi"] is False
|
|
|
|
|
|
# ── Vessels src= filter (mocked store) ────────────────────────────────────
|
|
|
|
def _seed_store() -> None:
|
|
vessel_last_known.clear()
|
|
vessel_last_known["422050100"] = {
|
|
"id": "422050100", "lat": 26.5, "lon": 56.3, "label": "HORMUZ STAR",
|
|
"extra": {"src": "vesselapi", "mmsi": "422050100"},
|
|
}
|
|
vessel_last_known["366001230"] = {
|
|
"id": "366001230", "lat": 35.0, "lon": -79.0, "label": "CONUS SHIP",
|
|
"extra": {"src": "aisstream", "mmsi": "366001230"},
|
|
}
|
|
vessel_last_known["366001231"] = {
|
|
"id": "366001231", "lat": 36.0, "lon": -78.0, "label": "CONUS SHIP 2",
|
|
"extra": {"src": "aisstream", "mmsi": "366001231"},
|
|
}
|
|
|
|
|
|
def test_fetch_vessels_src_filters() -> None:
|
|
_seed_store()
|
|
assert {v["id"] for v in asyncio.run(fetch_vessels(None, src="vesselapi"))} == {"422050100"}
|
|
assert {v["id"] for v in asyncio.run(fetch_vessels(None, src="aisstream"))} == {
|
|
"366001230", "366001231",
|
|
}
|
|
assert len(asyncio.run(fetch_vessels(None, src="all"))) == 3
|
|
assert len(asyncio.run(fetch_vessels(None))) == 3 # default all
|
|
|
|
|
|
def test_vessels_src_query_param(monkeypatch) -> None:
|
|
_seed_store()
|
|
|
|
async def _fake_fetch(bbox, limit, src=None):
|
|
rows = [
|
|
{"id": k, **{kk: v[kk] for kk in ("lat", "lon", "label", "extra")}}
|
|
for k, v in vessel_last_known.items()
|
|
]
|
|
if src and src != "all":
|
|
rows = [r for r in rows if (r.get("extra") or {}).get("src") == src]
|
|
return rows
|
|
|
|
monkeypatch.setattr("main.fetch_vessels", _fake_fetch)
|
|
|
|
body = asyncio.run(_get("/api/vessels?src=vesselapi")).json()
|
|
assert [r["id"] for r in body] == ["422050100"]
|
|
|
|
body = asyncio.run(_get("/api/vessels?src=aisstream")).json()
|
|
assert {r["id"] for r in body} == {"366001230", "366001231"}
|
|
|
|
body = asyncio.run(_get("/api/vessels?src=all")).json()
|
|
assert len(body) == 3
|
|
|
|
|
|
def test_vessels_src_rejects_bad_value() -> None:
|
|
resp = asyncio.run(_get("/api/vessels?src=marine-traffic"))
|
|
assert resp.status_code == 422
|