osint-dashboard/tests/test_chokepoints.py

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