osint-dashboard/tests/test_live_layers.py

"""Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans)."""

from live_layers import (
    MARKER_FIELDS,
    bbox_center_radius_nm,
    clip_fc_to_bbox,
    filter_points_bbox,
    parse_bbox,
    quantize_bbox,
    rainviewer_tile_url,
    slim_alert_properties,
    to_marker,
    transform_adsb_lol,
    transform_ais_frame,
    transform_amtraker,
    transform_nhc_storms,
    transform_wfigs_incidents,
    _cache,
    _ttl_get,
    _wfigs_params,
)

from camera_scraper import parse_caltrans_json


def test_parse_bbox_and_radius_clamps_to_150_nm():
    minlon, minlat, maxlon, maxlat = parse_bbox("-84.5,33.8,-75.4,36.6")
    assert (minlon, minlat, maxlon, maxlat) == (-84.5, 33.8, -75.4, 36.6)
    lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat)
    assert 35.0 < lat < 35.4
    assert -80.1 < lon < -79.8
    assert 1 <= radius <= 150


def test_parse_bbox_rejects_malformed():
    try:
        parse_bbox("1,2,3")
        assert False, "expected ValueError"
    except ValueError:
        pass


def test_transform_adsb_lol_maps_shared_marker_contract():
    payload = {
        "ac": [
            {
                "hex": "a1b2c3",
                "flight": "AAL123  ",
                "r": "N123AA",
                "t": "B738",
                "lat": 35.88,
                "lon": -78.79,
                "alt_baro": 32000,
                "gs": 430.2,
                "track": 87.5,
                "squawk": "1200",
                "emergency": "none",
                "category": "A3",
                "seen_pos": 0.4,
            },
            {"hex": "dead00", "flight": "NOFIX"},  # no coords → drop
        ]
    }
    rows = transform_adsb_lol(payload)
    assert len(rows) == 1
    m = rows[0]
    assert set(MARKER_FIELDS).issubset(m)
    assert m["id"] == "a1b2c3"
    assert m["lat"] == 35.88
    assert m["lon"] == -78.79
    assert m["heading"] == 87.5
    assert m["speed"] == 430.2
    assert m["label"] == "AAL123"
    assert m["extra"]["squawk"] == "1200"
    assert m["extra"]["alt_baro"] == 32000


def test_transform_amtraker_flattens_train_numbers():
    payload = {
        "1": [
            {
                "trainID": "1-9",
                "trainNum": "1",
                "routeName": "Sunset Limited",
                "lat": 29.76,
                "lon": -95.36,
                "heading": 90,
                "velocity": 45.0,
                "late": 12,
                "iconColor": "#ee3a43",
                "stations": [{"name": "Houston", "status": "enroute"}],
            }
        ],
        "5": [
            {
                "trainID": "5-12",
                "trainNum": "5",
                "routeName": "California Zephyr",
                "lat": 40.0,
                "lon": -105.0,
                "heading": "W",
                "lateMin": 5,
                "iconColor": "#005eb8",
            }
        ],
    }
    rows = transform_amtraker(payload)
    assert {r["id"] for r in rows} == {"1-9", "5-12"}
    sunset = next(r for r in rows if r["id"] == "1-9")
    assert sunset["lat"] == 29.76
    assert sunset["label"] == "Sunset Limited #1"
    assert sunset["speed"] == 45.0
    assert sunset["extra"]["late_min"] == 12
    assert sunset["extra"]["iconColor"] == "#ee3a43"


def test_transform_ais_position_report():
    frame = {
        "MessageType": "PositionReport",
        "MetaData": {
            "MMSI": 366912810,
            "ShipName": "EVER GIVEN",
            "latitude": 36.9,
            "longitude": -76.3,
        },
        "Message": {
            "PositionReport": {
                "Sog": 12.4,
                "Cog": 88.0,
                "TrueHeading": 90,
                "NavigationalStatus": 0,
            }
        },
    }
    row = transform_ais_frame(frame)
    assert row is not None
    assert row["id"] == "366912810"
    assert row["lat"] == 36.9
    assert row["lon"] == -76.3
    assert row["label"] == "EVER GIVEN"
    assert row["speed"] == 12.4
    assert row["heading"] == 90
    assert row["extra"]["navstat"] == 0


def test_transform_ais_ignores_non_position():
    assert transform_ais_frame({"MessageType": "Unknown"}) is None


def test_transform_wfigs_incidents_geojson():
    fc = {
        "type": "FeatureCollection",
        "features": [
            {
                "type": "Feature",
                "geometry": {"type": "Point", "coordinates": [-81.3, 28.5]},
                "properties": {
                    "IncidentName": "Foster Bridge",
                    "IncidentSize": 675,
                    "PercentContained": 100,
                    "POOState": "US-FL",
                    "IncidentTypeCategory": "WF",
                    "FireCause": "Human",
                    "FireDiscoveryDateTime": 1750000000000,
                },
            }
        ],
    }
    rows = transform_wfigs_incidents(fc)
    assert len(rows) == 1
    assert rows[0]["label"] == "Foster Bridge"
    assert rows[0]["lat"] == 28.5
    assert rows[0]["lon"] == -81.3
    assert rows[0]["extra"]["acres"] == 675
    assert rows[0]["extra"]["contained"] == 100


def test_filter_points_bbox():
    pts = [
        to_marker("a", 35.0, -78.0, label="in"),
        to_marker("b", 10.0, 20.0, label="out"),
    ]
    kept = filter_points_bbox(pts, -80, 33, -75, 37)
    assert [p["id"] for p in kept] == ["a"]


def test_rainviewer_tile_url():
    url = rainviewer_tile_url(
        host="https://tilecache.rainviewer.com",
        path="/v2/radar/cb581daa2c0f",
    )
    assert url == (
        "https://tilecache.rainviewer.com/v2/radar/cb581daa2c0f/256/{z}/{x}/{y}/2/1_1.png"
    )


def test_transform_nhc_storms():
    payload = {
        "activeStorms": [
            {
                "id": "al042026",
                "name": "Dolly",
                "classification": "TS",
                "latitudeNumeric": 13.6,
                "longitudeNumeric": -38.7,
                "movementDir": 280,
                "movementSpeed": 12,
                "intensity": 35,
            }
        ]
    }
    rows = transform_nhc_storms(payload)
    assert len(rows) == 1
    assert rows[0]["id"] == "al042026"
    assert rows[0]["label"] == "Tropical Storm Dolly"
    assert rows[0]["lat"] == 13.6


def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
    payload = """
    {"data":[
      {"cctv":{
        "location":{
          "latitude":"37.8","longitude":"-122.4",
          "locationName":"I-80 WB","nearbyPlace":"SF",
          "district":"4","route":"80","county":"SF","direction":"W"
        },
        "inService":"true",
        "imageData":{
          "static":{"currentImageURL":"https://cwwp2.dot.ca.gov/data/d4/cctv/image/cam.jpg"},
          "streamingVideoURL":"https://wzmedia.dot.ca.gov/D4/cam.stream/playlist.m3u8"
        }
      }},
      {"cctv":{
        "location":{"latitude":"1","longitude":"2","locationName":"down"},
        "inService":"false",
        "imageData":{"static":{"currentImageURL":"https://example.com/x.jpg"}}
      }}
    ]}
    """
    cams = parse_caltrans_json(payload, "caltrans")
    assert len(cams) == 1
    cam = cams[0]
    assert cam["discovery_source"] == "caltrans"
    assert cam["snapshot_url"].endswith("cam.jpg")
    assert cam["source_url"].endswith("playlist.m3u8")
    assert cam["device_type"] == "hls"
    assert cam["location_lat"] == 37.8
    assert cam["location_lon"] == -122.4
    assert "I-80" in cam["location_name"]
    assert "rtsp://" not in cam["source_url"].lower()
    assert "rtsp://" not in cam["snapshot_url"].lower()


def test_quantize_bbox_stable_under_jitter():
    a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102"))
    b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090"))
    assert a == b
    minlon, minlat, maxlon, maxlat = a
    assert minlon <= -78.7912
    assert minlat <= 35.7700
    assert maxlon >= -78.6101
    assert maxlat >= 35.9102


def test_ttl_get_does_not_block_other_keys():
    import asyncio

    _cache.clear()
    order = []

    async def slow():
        order.append("slow-start")
        await asyncio.sleep(0.2)
        order.append("slow-end")
        return "S"

    async def fast():
        order.append("fast")
        return "F"

    async def run():
        t1 = asyncio.create_task(_ttl_get("slow", 5, slow))
        await asyncio.sleep(0.01)
        t2 = asyncio.create_task(_ttl_get("fast", 5, fast))
        await asyncio.gather(t1, t2)

    asyncio.run(run())
    assert order.index("fast") < order.index("slow-end")
    assert _cache["slow"][1] == "S"
    assert _cache["fast"][1] == "F"
    _cache.clear()


def test_clip_fc_to_bbox_drops_far_features_and_empty_geometry():
    fc = {
        "type": "FeatureCollection",
        "features": [
            {
                "type": "Feature",
                "properties": {"event": "near"},
                "geometry": {"type": "Point", "coordinates": [-78.7, 35.8]},
            },
            {
                "type": "Feature",
                "properties": {"event": "far"},
                "geometry": {"type": "Point", "coordinates": [-120.0, 45.0]},
            },
            {
                "type": "Feature",
                "properties": {"event": "nogeom"},
                "geometry": None,
            },
            {
                "type": "Feature",
                "properties": {"event": "poly-overlap"},
                "geometry": {
                    "type": "Polygon",
                    "coordinates": [[
                        [-79.0, 35.0], [-78.0, 35.0], [-78.0, 36.0],
                        [-79.0, 36.0], [-79.0, 35.0],
                    ]],
                },
            },
        ],
    }
    clipped = clip_fc_to_bbox(fc, -79.0, 35.5, -78.0, 36.0)
    events = [f["properties"]["event"] for f in clipped["features"]]
    assert events == ["near", "poly-overlap"]


def test_slim_alert_properties_keeps_popup_fields_only():
    fat = {
        "event": "Tornado Warning",
        "severity": "Extreme",
        "headline": "TORNADO WARNING",
        "areaDesc": "Wake",
        "wfo": "RAH",
        "source": "nws",
        "parameters": {"WIND": ["70"]},
        "description": "A long narrative " * 40,
        "instruction": "Take shelter.",
        "geocode": {"SAME": ["037183"]},
    }
    slim = slim_alert_properties(fat)
    assert slim == {
        "event": "Tornado Warning",
        "severity": "Extreme",
        "headline": "TORNADO WARNING",
        "areaDesc": "Wake",
        "wfo": "RAH",
        "source": "nws",
    }


def test_wfigs_params_requests_simplified_geometry():
    params = _wfigs_params("-84.5,33.8,-75.4,36.6")
    assert "maxAllowableOffset" in params
    assert float(params["maxAllowableOffset"]) > 0
    assert params["geometryPrecision"] == 5
    assert int(params["resultRecordCount"]) <= 500
    # Envelope is the quantized cell, not the raw pan box.
    geom = params["geometry"]
    assert geom != "-84.5,33.8,-75.4,36.6"


def test_transform_adsb_lol_flags_military_from_dbflags():
    payload = {
        "ac": [
            {
                "hex": "ae01ab",
                "flight": "RCH123  ",
                "r": "04-1234",
                "t": "C17",
                "lat": 35.1,
                "lon": -77.9,
                "alt_baro": 24000,
                "gs": 410,
                "track": 90,
                "squawk": "5101",
                "emergency": "none",
                "category": "A5",
                "dbFlags": 1,
                "baro_rate": 64,
                "alt_geom": 24500,
                "desc": "Boeing C-17A Globemaster III",
                "ownOp": "USAF",
            },
            {
                "hex": "a1b2c3",
                "flight": "AAL123",
                "r": "N123AA",
                "t": "B738",
                "lat": 35.88,
                "lon": -78.79,
                "alt_baro": 32000,
                "gs": 430,
                "track": 87,
                "squawk": "1200",
                "emergency": "none",
                "category": "A3",
            },
        ]
    }
    rows = {r["id"]: r for r in transform_adsb_lol(payload)}
    mil = rows["ae01ab"]["extra"]
    civ = rows["a1b2c3"]["extra"]
    assert mil["role"] == "military"
    assert mil["role_src"] == "dbFlags"
    assert mil["emitter"] == "heavy"
    assert mil["desc"] == "Boeing C-17A Globemaster III"
    assert mil["ownOp"] == "USAF"
    assert mil["vs"] == 64
    assert mil["alt_geom"] == 24500
    assert civ["role"] == "civilian"
    assert civ["emitter"] == "large"


def test_transform_adsb_lol_military_from_icao_type_and_hex():
    payload = {
        "ac": [
            {"hex": "3b76aa", "flight": "FAF123", "t": "F16", "lat": 1, "lon": 2, "category": "A1"},
            {"hex": "ae1234", "flight": "BOXER1", "t": "C172", "lat": 1, "lon": 2, "category": "A1"},
        ]
    }
    rows = {r["id"]: r for r in transform_adsb_lol(payload)}
    assert rows["3b76aa"]["extra"]["role"] == "military"
    assert rows["3b76aa"]["extra"]["role_src"] == "type"
    assert rows["ae1234"]["extra"]["role"] == "military"
    assert rows["ae1234"]["extra"]["role_src"] == "hex"


def test_transform_ais_static_classifies_military_and_cargo():
    mil = transform_ais_frame({
        "MessageType": "ShipStaticData",
        "MetaData": {"MMSI": 338123456, "ShipName": "USNS BOB", "Latitude": 32.7, "Longitude": -117.2},
        "Message": {"ShipStaticData": {
            "Type": 35, "CallSign": "NBXX", "ImoNumber": 0,
            "Destination": "SAN DIEGO", "MaximumStaticDraught": 8.2,
            "Dimension": {"A": 80, "B": 20, "C": 8, "D": 8},
            "Eta": {"Month": 8, "Day": 29, "Hour": 14, "Minute": 0},
        }},
    })
    cargo = transform_ais_frame({
        "MessageType": "ShipStaticData",
        "MetaData": {"MMSI": 477123456, "ShipName": "EVER GIVEN", "Latitude": 36.9, "Longitude": -76.3},
        "Message": {"ShipStaticData": {
            "Type": 70, "CallSign": "VRXX", "ImoNumber": 9811000,
            "Destination": "NORFOLK", "MaximumStaticDraught": 14.5,
            "Dimension": {"A": 200, "B": 150, "C": 20, "D": 20},
        }},
    })
    assert mil is not None and cargo is not None
    assert mil["extra"]["role"] == "military"
    assert mil["extra"]["kind"] == "military"
    assert mil["extra"]["callsign"] == "NBXX"
    assert mil["extra"]["length"] == 100
    assert mil["extra"]["beam"] == 16
    assert mil["extra"]["dest"] == "SAN DIEGO"
    assert mil["extra"]["country"] == "United States"
    assert cargo["extra"]["role"] == "civilian"
    assert cargo["extra"]["kind"] == "cargo"
    assert cargo["extra"]["imo"] == 9811000


def test_transform_ais_position_decodes_navstat():
    row = transform_ais_frame({
        "MessageType": "PositionReport",
        "MetaData": {"MMSI": 366912810, "ShipName": "EVER GIVEN", "latitude": 36.9, "longitude": -76.3},
        "Message": {"PositionReport": {"Sog": 0.1, "Cog": 88.0, "TrueHeading": 90, "NavigationalStatus": 5}},
    })
    assert row is not None
    assert row["extra"]["nav"] == "moored"
    assert row["extra"]["navstat"] == 5


def test_nws_alerts_does_not_send_bbox_param(monkeypatch):
    """api.weather.gov/alerts/active 400s on bbox — clip locally instead."""
    import asyncio

    from live_layers import fetch_weather_alerts, _cache

    seen = []

    async def fake_get(url, params=None):
        seen.append((url, dict(params or {})))
        if "weather.gov" in url:
            return {
                "type": "FeatureCollection",
                "features": [{
                    "type": "Feature",
                    "properties": {"event": "Tornado Warning", "severity": "Extreme"},
                    "geometry": {"type": "Point", "coordinates": [-78.7, 35.8]},
                }],
            }
        return {"type": "FeatureCollection", "features": []}

    monkeypatch.setattr("live_layers._get_json", fake_get)
    _cache.clear()
    fc = asyncio.run(fetch_weather_alerts(None, "-79.0,35.5,-78.0,36.0"))
    nws_calls = [p for u, p in seen if "weather.gov" in u]
    assert nws_calls, "NWS should still be fetched"
    assert "bbox" not in nws_calls[0]
    assert fc.get("nws_ok") is True
    assert len(fc["features"]) == 1


def test_nws_alerts_failure_is_flagged(monkeypatch):
    import asyncio

    from live_layers import fetch_weather_alerts, _cache

    async def fake_get(url, params=None):
        if "weather.gov" in url:
            raise RuntimeError("400 Bad Request")
        return {"type": "FeatureCollection", "features": []}

    monkeypatch.setattr("live_layers._get_json", fake_get)
    _cache.clear()
    fc = asyncio.run(fetch_weather_alerts(None, None))
    assert fc.get("nws_ok") is False


def test_fetch_aircraft_get_path_does_not_persist(monkeypatch):
    """GET /api/aircraft must serve last-known without track/geofence writes."""
    import asyncio

    from live_layers import (
        aircraft_last_known, fetch_aircraft, persist_aircraft_snapshot, _cache,
    )

    aircraft_last_known.clear()
    aircraft_last_known["abc"] = {
        "id": "abc", "lat": 35.8, "lon": -78.7, "heading": 90, "speed": 400,
        "label": "ABC", "extra": {},
    }
    writes = {"n": 0}

    async def boom(*a, **k):
        writes["n"] += 1
        raise AssertionError("GET path must not persist")

    monkeypatch.setattr("tracks.record_position", boom)
    monkeypatch.setattr("geofence.record_and_notify", boom)
    _cache.clear()
    rows = asyncio.run(fetch_aircraft("-79,35,-78,36", persist=False))
    assert writes["n"] == 0
    assert any(r["id"] == "abc" for r in rows)


def test_persist_aircraft_snapshot_writes_tracks(monkeypatch):
    import asyncio

    from live_layers import persist_aircraft_snapshot

    recorded = []

    async def fake_record(kind, marker):
        recorded.append((kind, marker["id"]))
        return True

    async def fake_gf(**kw):
        return 0

    monkeypatch.setattr("tracks.record_position", fake_record)
    monkeypatch.setattr("geofence.record_and_notify", fake_gf)
    monkeypatch.setattr("ws_manager.manager.has_clients", lambda: False)

    rows = [{
        "id": "abc", "lat": 35.8, "lon": -78.7, "heading": 90, "speed": 400,
        "label": "ABC", "extra": {},
    }]
    asyncio.run(persist_aircraft_snapshot(rows))
    assert recorded == [("aircraft", "abc")]