"""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")] # ── Planespotters.net photo lookup ──────────────────────────────────────── def test_normalize_planespotter_photo_prefers_large_thumbnail(): from live_layers import _normalize_planespotter_photo out = _normalize_planespotter_photo({ "id": "1053982", "thumbnail": {"src": "https://t.plnspttrs.net/x_t.jpg", "size": {"width": 200, "height": 141}}, "thumbnail_large": {"src": "https://t.plnspttrs.net/x_280.jpg", "size": {"width": 395, "height": 280}}, "link": "https://www.planespotters.net/photo/1053982/foo", "photographer": "Günther Feniuk", }) assert out["id"] == "1053982" assert out["src"] == "https://t.plnspttrs.net/x_280.jpg" assert out["width"] == 395 assert out["height"] == 280 assert out["photographer"] == "Günther Feniuk" assert "planespotters.net" in out["link"] def test_normalize_planespotter_photo_empty_or_malformed_returns_none(): from live_layers import _normalize_planespotter_photo assert _normalize_planespotter_photo({}) is None assert _normalize_planespotter_photo({"thumbnail": {}}) is None assert _normalize_planespotter_photo(None) is None assert _normalize_planespotter_photo("not-a-dict") is None def test_fetch_planespotters_photo_hex_builds_url_and_normalizes(monkeypatch): import asyncio from live_layers import fetch_planespotters_photo, _cache seen = [] async def fake_get(url, params=None): seen.append(url) return {"photos": [{ "id": "1", "thumbnail": {"src": "https://t.plnspttrs.net/a_t.jpg"}, "thumbnail_large": {"src": "https://t.plnspttrs.net/a_280.jpg"}, "link": "https://www.planespotters.net/photo/1/x", "photographer": "A", }]} monkeypatch.setattr("live_layers._get_json", fake_get) _cache.clear() out = asyncio.run(fetch_planespotters_photo(hex_code="e8027e")) assert out["src"] == "https://t.plnspttrs.net/a_280.jpg" assert seen == ["https://api.planespotters.net/pub/photos/hex/e8027e"] def test_fetch_planespotters_photo_reg_fallback_and_no_result(monkeypatch): import asyncio from live_layers import fetch_planespotters_photo, _cache seen = [] async def fake_get(url, params=None): seen.append(url) return {"photos": []} monkeypatch.setattr("live_layers._get_json", fake_get) _cache.clear() assert asyncio.run(fetch_planespotters_photo(reg="D-ABCD")) is None assert seen == ["https://api.planespotters.net/pub/photos/reg/D-ABCD"] # no hex and no reg → no upstream call at all assert asyncio.run(fetch_planespotters_photo()) is None