From f6c1cfc4546415f4a3b18031147821b29fac57e2 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Tue, 1 Sep 2026 01:24:36 -0400 Subject: [PATCH] feat(geofence): watch list, filtered hit log, fence snapshot Off-viewport clients that send {type:watch_geofences,ids} on /ws/live still receive geofence_alert; AIS/ADS-B stay viewport-only. GET /api/geofence-alerts accepts geofence_id/since/until/source_kind. GET /api/geofences/{id}/at returns CAGG+FIRMS inside the fence at T (404 if missing, empty lists if DB down). DELETE missing fences 404s. --- alembic/versions/011_geofence_alerts_fence.py | 24 ++ app/geofence.py | 151 +++++++++++ app/main.py | 92 +++++-- app/ws_manager.py | 38 ++- tests/test_geofence.py | 249 +++++++++++++++++- tests/test_phase2_api.py | 108 ++++++++ 6 files changed, 625 insertions(+), 37 deletions(-) create mode 100644 alembic/versions/011_geofence_alerts_fence.py diff --git a/alembic/versions/011_geofence_alerts_fence.py b/alembic/versions/011_geofence_alerts_fence.py new file mode 100644 index 0000000..4b29f74 --- /dev/null +++ b/alembic/versions/011_geofence_alerts_fence.py @@ -0,0 +1,24 @@ +"""geofence_alerts (geofence_id, created_at DESC) for fence-scoped hit log + +Revision ID: 011_geofence_alerts_fence +Revises: 010_bbox_gist +Create Date: 2026-09-01 +""" + +from alembic import op + +revision = "011_geofence_alerts_fence" +down_revision = "010_bbox_gist" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute( + "CREATE INDEX IF NOT EXISTS ix_geofence_alerts_fence_created " + "ON geofence_alerts (geofence_id, created_at DESC)" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS ix_geofence_alerts_fence_created") diff --git a/app/geofence.py b/app/geofence.py index 398c73b..f010a3f 100644 --- a/app/geofence.py +++ b/app/geofence.py @@ -319,3 +319,154 @@ async def record_and_notify( except Exception: pass return sent + + +async def list_alerts( + *, + geofence_id: str | None = None, + since: datetime | None = None, + until: datetime | None = None, + source_kind: str | None = None, + limit: int = 100, +) -> list[dict]: + """Filterable hit log. Empty list if the DB is down — never raises.""" + where = ["TRUE"] + params: dict[str, Any] = {"limit": int(limit)} + if geofence_id: + where.append("geofence_id = CAST(:geofence_id AS uuid)") + params["geofence_id"] = geofence_id + if since is not None: + where.append("created_at >= :since") + params["since"] = since + if until is not None: + where.append("created_at <= :until") + params["until"] = until + if source_kind: + where.append("source_kind = :source_kind") + params["source_kind"] = source_kind + sql = f""" + SELECT id::text, geofence_id::text, source_kind, entity_id, + lat, lon, payload, created_at + FROM geofence_alerts + WHERE {' AND '.join(where)} + ORDER BY created_at DESC + LIMIT :limit + """ + try: + async with async_session() as session: + rows = (await session.execute(text(sql), params)).mappings().all() + out = [] + for r in rows: + item = dict(r) + if item.get("created_at") is not None: + item["created_at"] = item["created_at"].isoformat() + out.append(item) + return out + except Exception: + return [] + + +async def get_geofence(gid: str) -> dict | None: + current = next((f for f in _cache if f["id"] == gid), None) + if current is not None: + return current + try: + await refresh_cache() + except Exception: + return None + return next((f for f in _cache if f["id"] == gid), None) + + +def _marker_from_track(row) -> dict: + from live_layers import to_marker + + extra = {"bucket": row["bucket"].isoformat() if row.get("bucket") else None, "dvr": True} + return to_marker( + row["id"], row["lat"], row["lon"], + heading=row.get("heading"), speed=row.get("speed"), + label=row.get("label") or row["id"], + extra=extra, + ) + + +async def _cagg_inside(gid: str, kind: str, bucket: datetime, limit: int = 2000) -> list[dict]: + table = "aircraft_tracks_1min" if kind == "aircraft" else "vessel_tracks_1min" + id_col = "hex" if kind == "aircraft" else "mmsi" + sql = f""" + SELECT {id_col} AS id, lat, lon, heading, speed, label, bucket + FROM {table} + WHERE bucket = :bucket + AND ST_Intersects( + (SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)), + ST_SetSRID(ST_MakePoint(lon, lat), 4326) + ) + LIMIT :limit + """ + try: + async with async_session() as session: + rows = (await session.execute( + text(sql), {"bucket": bucket, "gid": gid, "limit": limit}, + )).mappings().all() + return [ + _marker_from_track(r) + for r in rows + if r["lat"] is not None and r["lon"] is not None + ] + except Exception: + return [] + + +async def _fires_inside(gid: str, ts: datetime, limit: int = 2000) -> list[dict]: + from tracks import minute_bucket + + bucket = minute_bucket(ts) + t1 = bucket + timedelta(minutes=1) + sql = """ + SELECT latitude, longitude, brightness, confidence, acq_time, satellite, + instrument, bright_ti5, frp, daynight + FROM fires + WHERE acq_time >= :t0 AND acq_time < :t1 + AND ST_Intersects( + (SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)), + ST_SetSRID(ST_MakePoint(longitude, latitude), 4326) + ) + LIMIT :limit + """ + try: + async with async_session() as session: + rows = (await session.execute( + text(sql), + {"t0": bucket, "t1": t1, "gid": gid, "limit": limit}, + )).mappings().all() + out = [] + for r in rows: + item = dict(r) + if item.get("acq_time") is not None: + item["acq_time"] = item["acq_time"].isoformat() + out.append(item) + return out + except Exception: + return [] + + +async def snapshot_at(gid: str, ts: datetime) -> dict | None: + """Positions inside the fence at time T. None if the fence is missing. + + Does not persist or notify. Empty lists if track/fire queries fail. + """ + fence = await get_geofence(gid) + if fence is None: + return None + from tracks import minute_bucket + + bucket = minute_bucket(ts) + aircraft = await _cagg_inside(gid, "aircraft", bucket) + vessels = await _cagg_inside(gid, "vessel", bucket) + fires = await _fires_inside(gid, ts) + return { + "geofence_id": gid, + "timestamp": ts.isoformat(), + "aircraft": aircraft, + "vessels": vessels, + "fires": fires, + } diff --git a/app/main.py b/app/main.py index 57c9d33..049d96c 100644 --- a/app/main.py +++ b/app/main.py @@ -873,7 +873,14 @@ async def trigger_social_ingest(query: str = "", max_items: int = 50): @app.websocket("/ws/live") async def live_ws(ws: WebSocket): - """Viewport-filtered AIS/ADS-B fan-out. Client sends {type:viewport,bbox}.""" + """Viewport-filtered AIS/ADS-B fan-out. + + Client JSON: + {"type":"viewport","bbox":"minlon,minlat,maxlon,maxlat"} + {"type":"watch_geofences","ids":["", ...]} — empty list = none + geofence_alert delivers if the point is in-viewport OR geofence_id is watched. + AIS/ADS-B/fire_aircraft stay viewport-only. + """ from ws_manager import manager client_id = str(id(ws)) @@ -899,6 +906,10 @@ async def live_ws(ws: WebSocket): manager.set_viewport(client_id, parse_bbox(str(data["bbox"]))) except ValueError: continue + elif data.get("type") == "watch_geofences": + ids = data.get("ids") or [] + if isinstance(ids, list): + manager.set_watched_geofences(client_id, [str(x) for x in ids]) except WebSocketDisconnect: pass finally: @@ -1757,33 +1768,68 @@ async def api_update_geofence(gid: str, payload: GeofenceUpdate): @app.delete("/api/geofences/{gid}", status_code=204) async def api_delete_geofence(gid: str): from geofence import delete_geofence - await delete_geofence(gid) + ok = await delete_geofence(gid) + if not ok: + raise HTTPException(404, "geofence not found") return None -@app.get("/api/geofence-alerts") -async def api_geofence_alerts(limit: int = Query(100, ge=1, le=500)): - from sqlalchemy import text as sql_text +@app.get("/api/geofences/{gid}/at") +async def api_geofence_at( + gid: str, + timestamp: str = Query(..., description="ISO-8601 instant for the 1-minute DVR bucket"), +): + """Aircraft/vessels/fires inside this fence at time T. Never writes.""" + from geofence import snapshot_at + from tracks import parse_timestamp + try: - async with async_session() as session: - rows = (await session.execute(sql_text( - """ - SELECT id::text, geofence_id::text, source_kind, entity_id, - lat, lon, payload, created_at - FROM geofence_alerts - ORDER BY created_at DESC - LIMIT :limit - """ - ), {"limit": limit})).mappings().all() - out = [] - for r in rows: - item = dict(r) - if item.get("created_at") is not None: - item["created_at"] = item["created_at"].isoformat() - out.append(item) - return out + ts = parse_timestamp(timestamp) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + if ts is None: + raise HTTPException(422, "timestamp required") + try: + body = await snapshot_at(gid, ts) except Exception: - return [] + body = { + "geofence_id": gid, + "timestamp": ts.isoformat(), + "aircraft": [], + "vessels": [], + "fires": [], + } + if body is None: + raise HTTPException(404, "geofence not found") + return body + + +@app.get("/api/geofence-alerts") +async def api_geofence_alerts( + geofence_id: UUID | None = Query(None), + since: str | None = Query(None, description="ISO-8601 inclusive lower bound"), + until: str | None = Query(None, description="ISO-8601 inclusive upper bound"), + source_kind: str | None = Query(None, description="firms|ais|adsb"), + limit: int = Query(100, ge=1, le=500), +): + """Hit log for drawn fences. Not /api/alerts (entity/keyword).""" + from geofence import list_alerts + from tracks import parse_timestamp + + if source_kind is not None and source_kind not in ("firms", "ais", "adsb"): + raise HTTPException(422, "source_kind must be one of: firms, ais, adsb") + try: + since_ts = parse_timestamp(since) if since else None + until_ts = parse_timestamp(until) if until else None + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + return await list_alerts( + geofence_id=str(geofence_id) if geofence_id else None, + since=since_ts, + until=until_ts, + source_kind=source_kind, + limit=limit, + ) @app.get("/api/fire-aircraft") diff --git a/app/ws_manager.py b/app/ws_manager.py index f91657c..9c3c59d 100644 --- a/app/ws_manager.py +++ b/app/ws_manager.py @@ -8,10 +8,18 @@ from __future__ import annotations import asyncio from typing import Any +from uuid import UUID BBox = tuple[float, float, float, float] # minlon, minlat, maxlon, maxlat +def _uuid_str(value: object) -> str | None: + try: + return str(UUID(str(value))) + except (ValueError, TypeError, AttributeError): + return None + + def point_in_bbox(lon: float, lat: float, bbox: BBox | None) -> bool: """True if (lon, lat) sits inside an axis-aligned viewport.""" if bbox is None: @@ -26,6 +34,7 @@ class ConnectionManager: def __init__(self) -> None: self._queues: dict[str, asyncio.Queue] = {} self._viewports: dict[str, BBox] = {} + self._watched: dict[str, set[str]] = {} def register(self, client_id: str, maxsize: int = 256) -> asyncio.Queue: q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) @@ -35,6 +44,21 @@ class ConnectionManager: def unregister(self, client_id: str) -> None: self._queues.pop(client_id, None) self._viewports.pop(client_id, None) + self._watched.pop(client_id, None) + + def set_watched_geofences(self, client_id: str, ids: list[str]) -> None: + """Watch these fence UUIDs so geofence_alert delivers off-viewport. + + Invalid UUIDs are ignored. Empty list = watch none (viewport-only). + """ + if client_id not in self._queues: + return + watched: set[str] = set() + for raw in ids: + uid = _uuid_str(raw) + if uid is not None: + watched.add(uid) + self._watched[client_id] = watched def set_viewport(self, client_id: str, bbox: BBox) -> None: if client_id in self._queues: @@ -59,13 +83,21 @@ class ConnectionManager: ) -> int: """Enqueue `{type, payload}` for clients whose viewport contains the point. - Drops the oldest queued message if a client's buffer is full so a slow - tab cannot stall ingest. Returns the number of clients that got a copy. + kind=geofence_alert also delivers when payload.geofence_id is in the + client's watch set (even if the point is off-viewport). Other kinds + stay viewport-only. Drops the oldest queued message if a client's + buffer is full. Returns the number of clients that got a copy. """ msg = {"type": kind, "payload": payload} sent = 0 + gid = _uuid_str(payload.get("geofence_id")) if kind == "geofence_alert" else None for client_id, queue in list(self._queues.items()): - if not point_in_bbox(lon, lat, self._viewports.get(client_id)): + in_view = point_in_bbox(lon, lat, self._viewports.get(client_id)) + if kind == "geofence_alert": + watching = gid is not None and gid in self._watched.get(client_id, set()) + if not in_view and not watching: + continue + elif not in_view: continue if queue.full(): try: diff --git a/tests/test_geofence.py b/tests/test_geofence.py index df1d59d..10f533a 100644 --- a/tests/test_geofence.py +++ b/tests/test_geofence.py @@ -51,23 +51,33 @@ def test_matching_geofences_only_active_hits(): 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", (-80.0, 35.0, -78.0, 36.0)) - mgr.set_viewport("sf", (-123.0, 37.0, -121.0, 38.0)) + mgr.set_viewport("nc", NC_VIEW) + mgr.set_viewport("sf", SF_VIEW) async def run(): - payload = { - "geofence_id": "a", - "geofence_name": "NC", - "source_kind": "ais", - "entity_id": "366123456", - "lat": 35.5, - "lon": -79.0, - } - n = await mgr.publish_point("geofence_alert", payload, lat=35.5, lon=-79.0) + 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" @@ -77,6 +87,109 @@ def test_geofence_alert_fans_out_only_to_viewport_clients(): 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 @@ -134,3 +247,117 @@ def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch): 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 diff --git a/tests/test_phase2_api.py b/tests/test_phase2_api.py index e52548b..2afead2 100644 --- a/tests/test_phase2_api.py +++ b/tests/test_phase2_api.py @@ -17,6 +17,9 @@ async def _req(method: str, path: str, **kw) -> httpx.Response: return await client.request(method, path, **kw) +FENCE_ID = "11111111-1111-1111-1111-111111111111" + + def test_geofence_post_rejects_point(): resp = asyncio.run(_req( "POST", "/api/geofences", @@ -25,6 +28,111 @@ def test_geofence_post_rejects_point(): assert resp.status_code == 422 +def test_delete_geofence_404_when_missing(monkeypatch): + async def missing(_gid: str) -> bool: + return False + + monkeypatch.setattr("geofence.delete_geofence", missing) + resp = asyncio.run(_req("DELETE", f"/api/geofences/{FENCE_ID}")) + assert resp.status_code == 404 + + +def test_geofence_alerts_passes_filters(monkeypatch): + seen = {} + + async def fake_list(**kwargs): + seen.update(kwargs) + return [{"id": "a", "geofence_id": FENCE_ID, "source_kind": "ais"}] + + monkeypatch.setattr("geofence.list_alerts", fake_list) + resp = asyncio.run(_req( + "GET", "/api/geofence-alerts", + params={ + "geofence_id": FENCE_ID, + "since": "2026-08-28T00:00:00Z", + "until": "2026-08-29T00:00:00Z", + "source_kind": "ais", + "limit": 10, + }, + )) + assert resp.status_code == 200 + assert resp.json()[0]["source_kind"] == "ais" + assert seen["geofence_id"] == FENCE_ID + assert seen["source_kind"] == "ais" + assert seen["limit"] == 10 + assert seen["since"] is not None + assert seen["until"] is not None + + +def test_geofence_alerts_rejects_bad_source_kind(): + resp = asyncio.run(_req( + "GET", "/api/geofence-alerts", params={"source_kind": "camera"}, + )) + assert resp.status_code == 422 + + +def test_geofence_at_404_when_missing(monkeypatch): + async def no_snap(gid: str, ts): + return None + + monkeypatch.setattr("geofence.snapshot_at", no_snap) + resp = asyncio.run(_req( + "GET", f"/api/geofences/{FENCE_ID}/at", + params={"timestamp": "2026-08-28T12:04:00Z"}, + )) + assert resp.status_code == 404 + + +def test_geofence_at_empty_lists_when_db_down(monkeypatch): + async def empty_snap(gid: str, ts): + return { + "geofence_id": gid, + "timestamp": ts.isoformat(), + "aircraft": [], + "vessels": [], + "fires": [], + } + + monkeypatch.setattr("geofence.snapshot_at", empty_snap) + resp = asyncio.run(_req( + "GET", f"/api/geofences/{FENCE_ID}/at", + params={"timestamp": "2026-08-28T12:04:00Z"}, + )) + assert resp.status_code == 200 + body = resp.json() + assert body["geofence_id"] == FENCE_ID + assert body["aircraft"] == [] + assert body["vessels"] == [] + assert body["fires"] == [] + assert "timestamp" in body + + +def test_geofence_at_does_not_notify(monkeypatch): + called = {"notify": 0} + + async def empty_snap(gid: str, ts): + return { + "geofence_id": gid, + "timestamp": ts.isoformat(), + "aircraft": [], + "vessels": [], + "fires": [], + } + + async def boom(**_kw): + called["notify"] += 1 + raise AssertionError("GET /at must not record_and_notify") + + monkeypatch.setattr("geofence.snapshot_at", empty_snap) + monkeypatch.setattr("geofence.record_and_notify", boom) + resp = asyncio.run(_req( + "GET", f"/api/geofences/{FENCE_ID}/at", + params={"timestamp": "2026-08-28T12:04:00Z"}, + )) + assert resp.status_code == 200 + assert called["notify"] == 0 + + def test_geofences_list_does_not_collide_with_alerts(): resp = asyncio.run(_req("GET", "/api/geofences")) assert resp.status_code == 200