"""Geofence hit detection and WS alert routing (no Redis).""" from __future__ import annotations import asyncio import geofence from geofence import ( matching_geofences, point_in_geojson, validate_polygon_geojson, ) from ws_manager import ConnectionManager NC_BOX = { "type": "Polygon", "coordinates": [[ [-80.0, 35.0], [-78.0, 35.0], [-78.0, 36.0], [-80.0, 36.0], [-80.0, 35.0], ]], } def test_point_inside_polygon_hits(): assert point_in_geojson(-79.0, 35.5, NC_BOX) is True def test_point_outside_polygon_misses(): assert point_in_geojson(-122.4, 37.7, NC_BOX) is False def test_validate_rejects_non_polygon(): try: validate_polygon_geojson({"type": "Point", "coordinates": [-79.0, 35.5]}) assert False, "expected ValueError" except ValueError: pass def test_matching_geofences_only_active_hits(): fences = [ {"id": "a", "name": "NC", "active": True, "geojson": NC_BOX}, {"id": "b", "name": "off", "active": False, "geojson": NC_BOX}, ] hits = matching_geofences(-79.0, 35.5, fences) assert [h["id"] for h in hits] == ["a"] assert matching_geofences(-122.4, 37.7, fences) == [] FENCE_ID = "11111111-1111-1111-1111-111111111111" NC_VIEW = (-80.0, 35.0, -78.0, 36.0) SF_VIEW = (-123.0, 37.0, -121.0, 38.0) def _alert_payload(gid=FENCE_ID): return { "geofence_id": gid, "geofence_name": "NC", "source_kind": "ais", "entity_id": "366123456", "lat": 35.5, "lon": -79.0, } def test_geofence_alert_fans_out_only_to_viewport_clients(): mgr = ConnectionManager() q_nc = mgr.register("nc") q_sf = mgr.register("sf") mgr.set_viewport("nc", NC_VIEW) mgr.set_viewport("sf", SF_VIEW) async def run(): n = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n == 1 msg = q_nc.get_nowait() assert msg["type"] == "geofence_alert" assert msg["payload"]["entity_id"] == "366123456" assert q_sf.empty() asyncio.run(run()) def test_off_viewport_watch_receives_geofence_alert(): mgr = ConnectionManager() q_sf = mgr.register("sf") mgr.set_viewport("sf", SF_VIEW) mgr.set_watched_geofences("sf", [FENCE_ID]) async def run(): n = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n == 1 msg = q_sf.get_nowait() assert msg["type"] == "geofence_alert" assert msg["payload"]["geofence_id"] == FENCE_ID asyncio.run(run()) def test_off_viewport_without_watch_does_not_receive_geofence_alert(): mgr = ConnectionManager() q_sf = mgr.register("sf") mgr.set_viewport("sf", SF_VIEW) async def run(): n = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n == 0 assert q_sf.empty() asyncio.run(run()) def test_on_viewport_receives_geofence_alert_without_watch(): mgr = ConnectionManager() q_nc = mgr.register("nc") mgr.set_viewport("nc", NC_VIEW) async def run(): n = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n == 1 assert q_nc.get_nowait()["type"] == "geofence_alert" asyncio.run(run()) def test_ais_stays_viewport_only_even_when_watching(): mgr = ConnectionManager() q_sf = mgr.register("sf") mgr.set_viewport("sf", SF_VIEW) mgr.set_watched_geofences("sf", [FENCE_ID]) async def run(): n = await mgr.publish_point("ais", {"id": "366123456"}, lat=35.5, lon=-79.0) assert n == 0 assert q_sf.empty() asyncio.run(run()) def test_invalid_watch_uuids_ignored_empty_list_clears(): mgr = ConnectionManager() q = mgr.register("sf") mgr.set_viewport("sf", SF_VIEW) mgr.set_watched_geofences("sf", ["not-a-uuid", FENCE_ID, "also-bad"]) async def run(): n = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n == 1 q.get_nowait() mgr.set_watched_geofences("sf", []) n2 = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n2 == 0 assert q.empty() asyncio.run(run()) def test_unregister_clears_watched_geofences(): mgr = ConnectionManager() q = mgr.register("sf") mgr.set_viewport("sf", SF_VIEW) mgr.set_watched_geofences("sf", [FENCE_ID]) mgr.unregister("sf") q2 = mgr.register("sf") mgr.set_viewport("sf", SF_VIEW) async def run(): n = await mgr.publish_point( "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0, ) assert n == 0 assert q2.empty() asyncio.run(run()) def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch): """FIRMS ingest in the ingester has an empty in-process cache — still ST_Intersects.""" import geofence geofence._cache.clear() geofence._recent_hits.clear() st_called = [] async def fake_st(lon, lat): st_called.append((lon, lat)) return [{ "id": "11111111-1111-1111-1111-111111111111", "name": "NC", "geojson": NC_BOX, "active": True, }] monkeypatch.setattr(geofence, "st_intersects", fake_st) executed: list = [] class FakeSession: async def execute(self, stmt, params=None): executed.append(params or {}) return None async def commit(self): executed.append("commit") async def __aenter__(self): return self async def __aexit__(self, *a): return False monkeypatch.setattr(geofence, "async_session", FakeSession) async def run(): from ws_manager import manager q = manager.register("nc") manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0)) n = await geofence.record_and_notify( source_kind="firms", entity_id="35.5,-79.0,N", lat=35.5, lon=-79.0, payload={"satellite": "N"}, ) msg = None if q.empty() else q.get_nowait() manager.unregister("nc") return n, msg n, msg = asyncio.run(run()) assert st_called == [(-79.0, 35.5)] assert n == 1 assert msg["type"] == "geofence_alert" assert msg["payload"]["source_kind"] == "firms" inserts = [p for p in executed if isinstance(p, dict)] assert inserts and inserts[0]["source_kind"] == "firms" assert "commit" in executed def test_list_alerts_sql_filters(monkeypatch): captured: dict = {} class FakeResult: def mappings(self): return self def all(self): return [] class FakeSession: async def execute(self, stmt, params=None): captured["sql"] = str(stmt) captured["params"] = params return FakeResult() async def __aenter__(self): return self async def __aexit__(self, *a): return False monkeypatch.setattr(geofence, "async_session", FakeSession) from datetime import datetime, timezone since = datetime(2026, 8, 28, tzinfo=timezone.utc) until = datetime(2026, 8, 29, tzinfo=timezone.utc) async def run(): return await geofence.list_alerts( geofence_id=FENCE_ID, since=since, until=until, source_kind="firms", limit=5, ) assert asyncio.run(run()) == [] sql = captured["sql"].lower() assert "geofence_id" in sql assert "created_at >=" in sql assert "created_at <=" in sql assert "source_kind" in sql assert captured["params"]["geofence_id"] == FENCE_ID assert captured["params"]["source_kind"] == "firms" assert captured["params"]["limit"] == 5 def test_alembic_fence_created_index_exists(): from pathlib import Path text = Path(__file__).resolve().parent.parent.joinpath( "alembic/versions/011_geofence_alerts_fence.py", ).read_text() assert "ix_geofence_alerts_fence_created" in text assert "010_bbox_gist" in text def test_snapshot_at_404_when_fence_missing(monkeypatch): geofence._cache.clear() async def boom(): raise RuntimeError("db down") monkeypatch.setattr(geofence, "refresh_cache", boom) async def run(): from datetime import datetime, timezone return await geofence.snapshot_at( FENCE_ID, datetime(2026, 8, 28, 12, 4, tzinfo=timezone.utc), ) assert asyncio.run(run()) is None def test_snapshot_queries_st_intersects(monkeypatch): geofence._cache[:] = [{ "id": FENCE_ID, "name": "NC", "geojson": NC_BOX, "active": True, }] sqls: list[str] = [] class FakeResult: def mappings(self): return self def all(self): return [] class FakeSession: async def execute(self, stmt, params=None): sqls.append(str(stmt)) return FakeResult() async def __aenter__(self): return self async def __aexit__(self, *a): return False monkeypatch.setattr(geofence, "async_session", FakeSession) async def run(): from datetime import datetime, timezone return await geofence.snapshot_at( FENCE_ID, datetime(2026, 8, 28, 12, 4, 30, tzinfo=timezone.utc), ) body = asyncio.run(run()) assert body["aircraft"] == [] assert body["vessels"] == [] assert body["fires"] == [] blob = "\n".join(sqls).lower() assert "st_intersects" in blob assert "aircraft_tracks_1min" in blob assert "vessel_tracks_1min" in blob assert "from fires" in blob