"""Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans).""" import json from live_layers import ( MARKER_FIELDS, bbox_center_radius_nm, clip_fc_to_bbox, filter_points_bbox, parse_bbox, quantize_bbox, pick_sentinel_feature, rainviewer_tile_url, sign_cog_url, sentinel1_tile_url, slim_alert_properties, to_marker, transform_adsb_lol, transform_ais_frame, transform_amtraker, transform_nhc_storms, transform_wfigs_incidents, SENTINEL1_ATTRIBUTION, TITILER_COG_TILES, _cache, _ttl_get, _wfigs_params, ) from camera_scraper import parse_caltrans_json, parse_udot_ibi_page, parse_odot_json, parse_mdot_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() # ── UDOT IBI 511 parser ────────────────────────────────────────────────── def _udot_row(cam_id, lng, lat, **img_overrides): img = { "id": cam_id, "cameraSiteId": cam_id, "imageUrl": f"/map/Cctv/{cam_id}", "disabled": False, "blocked": False, } img.update(img_overrides) return { "id": cam_id, "sourceId": "102771", "source": "ADX", "roadway": "Unknown", "direction": "Unknown", "location": "Freedom Blvd / 200 W @ 1100 N, PVO", "latLng": {"geography": { "coordinateSystemId": 4326, "wellKnownText": f"POINT ({lng} {lat})"}}, "images": [img], } def _udot_page(rows): import json return json.dumps({"draw": 0, "recordsTotal": len(rows), "recordsFiltered": len(rows), "data": rows}) def test_parse_udot_wkt_maps_lng_lat(): cams = parse_udot_ibi_page(_udot_page([_udot_row(112731, -111.66204, 40.24863)])) assert len(cams) == 1 cam = cams[0] # WKT is `POINT (lng lat)` — order must not be swapped. assert cam["location_lat"] == 40.24863 assert cam["location_lon"] == -111.66204 assert cam["discovery_source"] == "udot" assert cam["vendor"] == "UDOT" assert cam["source_url"] == "https://prod-ut.ibi511.com/map/Cctv/112731" assert cam["snapshot_url"] == cam["source_url"] assert "rtsp://" not in cam["source_url"].lower() assert cam["raw"]["udot_id"] == 112731 def test_parse_udot_skips_blocked_and_disabled(): rows = [ _udot_row(1, -111.0, 40.0), _udot_row(2, -111.1, 40.1, blocked=True), _udot_row(3, -111.2, 40.2, disabled=True), ] rows.append(_udot_row(4, -111.3, 40.3)) rows[3]["images"] = [] # no images → drop cams = parse_udot_ibi_page(_udot_page(rows)) assert [c["raw"]["udot_id"] for c in cams] == [1] def test_parse_udot_drops_out_of_bbox(): rows = [ _udot_row(1, -111.0, 40.0), # inside Utah _udot_row(2, -100.0, 40.0), # east of -108.9 _udot_row(3, -120.0, 40.0), # west of -114.2 _udot_row(4, -111.0, 44.0), # north of 42.1 _udot_row(5, -111.0, 30.0), # south of 36.9 ] cams = parse_udot_ibi_page(_udot_page(rows)) assert [c["raw"]["udot_id"] for c in cams] == [1] def test_parse_udot_bad_payload_returns_empty(): import json assert parse_udot_ibi_page("not json") == [] assert parse_udot_ibi_page(json.dumps({"data": None})) == [] assert parse_udot_ibi_page(json.dumps({"data": "nope"})) == [] def test_parse_udot_missing_wkt_skipped(): row = _udot_row(1, -111.0, 40.0) row["latLng"] = {} assert parse_udot_ibi_page(_udot_page([row])) == [] def test_parse_odot_tripcheck_keeps_valid_skips_missing_and_oob(): payload = """ {"features":[ {"attributes":{ "cameraId":277,"filename":"AstoriaUS101_pid392.jpg", "latitude":46.18785,"longitude":-123.85347, "route":"US101 ","title":"US101 at Astoria" }}, {"attributes":{ "cameraId":200,"filename":"","latitude":45.0,"longitude":-122.0, "route":"I-5","title":"missing filename" }}, {"attributes":{ "cameraId":300,"filename":"nocal_pid1.jpg", "latitude":40.0,"longitude":-122.0, "route":"US97","title":"out of bbox" }}, {"attributes":{ "cameraId":400,"filename":"badcoord_pid2.jpg", "latitude":null,"longitude":-122.0, "route":"OR22","title":"null coord" }} ]} """ cams = parse_odot_json(payload, "www.tripcheck.com") assert len(cams) == 1 cam = cams[0] assert cam["discovery_source"] == "odot" assert cam["snapshot_url"] == ( "https://tripcheck.com/RoadCams/cams/AstoriaUS101_pid392.jpg") assert cam["source_url"] == cam["snapshot_url"] assert cam["location_lat"] == 46.18785 assert cam["location_lon"] == -123.85347 assert "US101 at Astoria" in cam["location_name"] assert cam["vendor"] == "ODOT" assert cam["device_type"] == "http" assert "rtsp://" not in cam["snapshot_url"].lower() def test_parse_odot_tripcheck_handles_malformed(): assert parse_odot_json("not json", "www.tripcheck.com") == [] assert parse_odot_json('{"features":null}', "www.tripcheck.com") == [] def test_parse_mdot_extracts_html_fields_and_bbox_filters(): rows = [ # In-bbox, full fields. { "route": "11 Mile", "county": 'Wayne County Go to', "location": " @ Mound NB", "direction": "Traffic closest to camera is traveling north.", "image": 'x', }, # Out of bbox (lat 50) → drop. { "route": "Far", "county": 'Nowhere Go to', "location": "", "image": '', }, # Missing coordinates → drop. { "route": "NoCoords", "county": 'Somewhere Go to', "location": "", "image": '', }, # Missing image → drop. { "route": "NoImage", "county": 'Kent Go to', "location": " @ Division", "image": "", }, # RTSP image src → drop. { "route": "Rtsp", "county": 'Wayne Go to', "location": "", "image": '', }, ] cams = parse_mdot_json(json.dumps(rows), "mdotjboss.state.mi.us") assert len(cams) == 1 cam = cams[0] assert cam["discovery_source"] == "mdot" assert cam["location_lat"] == 42.491304 assert cam["location_lon"] == -83.04479 assert cam["snapshot_url"] == "https://micamerasimages.net/thumbs/semtoc_cam_253.flv.jpg?item=1" assert cam["source_url"] == "https://mdotjboss.state.mi.us/MiDrive/camera/1129" assert cam["device_type"] == "http" assert cam["vendor"] == "MDOT" assert "11 Mile @ Mound NB" in cam["location_name"] assert "Wayne County" in cam["location_name"] def test_parse_mdot_handles_malformed_payload(): assert parse_mdot_json("not json", "mdot") == [] assert parse_mdot_json('{"not": "a list"}', "mdot") == [] assert parse_mdot_json("[]", "mdot") == [] 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, headers=None): seen.append((url, (headers or {}).get("User-Agent", ""))) 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[0][0] == "https://api.planespotters.net/pub/photos/hex/e8027e" assert "@" in seen[0][1] or "http" in seen[0][1] 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, headers=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 def test_planespotters_headers_add_contact_when_ua_is_generic(monkeypatch): import live_layers monkeypatch.setattr(live_layers, "OSINT_USER_AGENT", "osint-dashboard/1.0 (self-hosted)") ua = live_layers._planespotters_headers()["User-Agent"] assert "osint-dashboard" in ua assert "@" in ua # ── Sentinel-1 SAR (Planetary Computer STAC → signed COG template) ──────── def test_sign_cog_url_appends_token(): # PC returns the token pre-encoded as a query string; append verbatim. assert sign_cog_url("https://blob.example/x.tif", "st=s&se=e&sig=x%3D") == \ "https://blob.example/x.tif?st=s&se=e&sig=x%3D" # Existing query string → append with & assert sign_cog_url("https://blob.example/x.tif?foo=1", "st=s&sig=x") == \ "https://blob.example/x.tif?foo=1&st=s&sig=x" def test_sentinel1_tile_url_contains_titiler_rescale_and_cfastie(): signed = "https://blob.example/x.tif?token=secret" url = sentinel1_tile_url(signed) assert url.startswith(TITILER_COG_TILES + "?") assert "WebMercatorQuad/{z}/{x}/{y}?" in url assert "url=https%3A%2F%2Fblob.example%2Fx.tif%3Ftoken%3Dsecret" in url assert "rescale=0%2C500" in url assert "colormap_name=cfastie" in url def test_sentinel1_tile_url_is_same_origin_relative(): # Self-hosted TiTiler: the browser must hit the Pi's nginx vhost, not # titiler.xyz or a raw host:port. The template is a root-relative path. url = sentinel1_tile_url("https://blob.example/x.tif") assert url.startswith("/titiler/cog/tiles/WebMercatorQuad/") assert "://" not in url assert "titiler.xyz" not in url def _stac_feature(assets: dict) -> dict: return { "type": "Feature", "id": "S1A_IW_GRDH_1SDV_20240820T000000", "properties": {"datetime": "2024-08-20T00:00:00Z"}, "assets": assets, } def test_fetch_sentinel1_vv_signed_tile_url(monkeypatch): import asyncio from live_layers import fetch_sentinel1, _cache calls = [] async def fake_post(url, json=None, headers=None): calls.append(("post", url, json)) return {"features": [_stac_feature({ "vv": {"href": "https://blob.example/grd-vv.tif"}, })]} async def fake_get(url, params=None, headers=None): calls.append(("get", url)) return {"token": "sig=abc123"} monkeypatch.setattr("live_layers._post_json", fake_post) monkeypatch.setattr("live_layers._get_json", fake_get) _cache.clear() out = asyncio.run(fetch_sentinel1("-80,35,-79,36")) assert out["id"] == "sentinel-1-sar" assert out["kind"] == "raster" assert out["polarization"] == "vv" assert out["opacity"] == 0.8 assert out["itemId"].startswith("S1A") assert out["attribution"] == SENTINEL1_ATTRIBUTION assert "WebMercatorQuad/{z}/{x}/{y}?" in out["tileUrl"] assert "rescale=0%2C500" in out["tileUrl"] assert "colormap_name=cfastie" in out["tileUrl"] # SAS token "sig=abc123" is appended top-level, then the whole COG URL is # percent-encoded again as a query param (=> sig%3Dabc123). assert "sig%3Dabc123" in out["tileUrl"] # STAC search payload shape post_url, post_json = calls[0][1], calls[0][2] assert post_url.endswith("/api/stac/v1/search") assert post_json["collections"] == ["sentinel-1-grd"] assert post_json["limit"] >= 1 assert post_json["sortby"][0]["direction"] == "desc" assert "bbox" in out def test_fetch_sentinel1_uses_hh_when_vv_missing(monkeypatch): import asyncio from live_layers import fetch_sentinel1, _cache async def fake_post(url, json=None, headers=None): return {"features": [_stac_feature({ "hh": {"href": "https://blob.example/grd-hh.tif"}, })]} async def fake_get(url, params=None, headers=None): return {"token": "tok"} monkeypatch.setattr("live_layers._post_json", fake_post) monkeypatch.setattr("live_layers._get_json", fake_get) _cache.clear() out = asyncio.run(fetch_sentinel1("-80,35,-79,36")) assert out["polarization"] == "hh" assert "url=https%3A%2F%2Fblob.example%2Fgrd-hh.tif" in out["tileUrl"] def test_fetch_sentinel1_none_on_empty_features(monkeypatch): import asyncio from live_layers import fetch_sentinel1, _cache async def fake_post(url, json=None, headers=None): return {"features": []} monkeypatch.setattr("live_layers._post_json", fake_post) _cache.clear() assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None def test_fetch_sentinel1_none_when_no_vv_or_hh(monkeypatch): import asyncio from live_layers import fetch_sentinel1, _cache async def fake_post(url, json=None, headers=None): return {"features": [_stac_feature({"thumbnail": {"href": "https://x"}})]} monkeypatch.setattr("live_layers._post_json", fake_post) _cache.clear() assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None def test_pick_sentinel_feature_prefers_scene_covering_center(): features = [ {"id": "far", "bbox": [10.0, 10.0, 12.0, 12.0]}, {"id": "cover", "bbox": [-80.5, 34.5, -78.5, 36.5]}, {"id": "also-far", "bbox": [-10.0, 0.0, -8.0, 2.0]}, ] picked = pick_sentinel_feature(features, -79.5, 35.5) assert picked["id"] == "cover" def test_pick_sentinel_feature_falls_back_to_first_when_none_cover(): features = [ {"id": "a", "bbox": [10.0, 10.0, 12.0, 12.0]}, {"id": "b", "bbox": [20.0, 20.0, 22.0, 22.0]}, ] assert pick_sentinel_feature(features, -79.5, 35.5)["id"] == "a" assert pick_sentinel_feature([], -79.5, 35.5) is None