matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

fix(coverage): NWS carries its alert polygon; gate fails closed for weather

Browse source 
The real LA leak was a zone-only advisory with no polygon and no centroid —
the fail-open gate kept it. Now NWS attaches the full GeoJSON alert geometry
(Polygon/MultiPolygon) to the event, and the coverage gate drops weather
alerts it cannot locate (fail-closed, matching Central), while staying
fail-open for other categories. Removes the old buggy adapter-level
_in_coverage heuristic (the gate supersedes it).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-06 18:39:44 +00:00
commit 94154b5549
5 changed files with 385 additions and 118 deletions
Download patch file Download diff file Expand all files Collapse all files

41
work/meshai/coverage_area.py
View file

@ -168,19 +168,27 @@ def areas_from_config(coverage_cfg) -> list[MonitoringArea]:
# event_in_areas — convenience wrapper over build_geom_json + classify
# ---------------------------------------------------------------------------
def event_in_areas(event: Any, areas: list[MonitoringArea]) -> bool:
"""Return True if the event's geometry intersects any configured area.
def classify_event_areas(event: Any, areas: list[MonitoringArea]) -> str:
"""Classify an event against configured areas, returning the 3-state result.
Extracts the event's geometry (same priority chain as ``event_in_areas``)
and returns the raw ``classify_geom_areas`` verdict so callers can
distinguish "out-of-bounds" from "can't be located":
'null-geom' -- no geometry / bbox / centroid at all
'invalid-geom' -- geometry present but Shapely could not evaluate it
'no-area' -- areas list empty (coverage effectively off)
'in-bounds' -- intersects at least one configured area
'out-of-bounds' -- lies entirely outside every configured area
The coverage gate uses this to fail CLOSED (drop) on 'null-geom' /
'invalid-geom' for weather-alert categories, while other adapters keep
the fail-OPEN behaviour.
Geometry extraction priority (mirrors Central's archive chain):
1. event.data["geometry"] — full GeoJSON dict (richest, preferred)
2. event.data["bbox"] — [west, south, east, north]
3. (event.lon, event.lat) — Point centroid [lon, lat] in GeoJSON order
Fail-open semantics (matching Central):
'null-geom' (no geometry at all) → True (keep)
'invalid-geom' (parse/Shapely error) → True (keep)
'no-area' (areas list empty) → True (keep)
'out-of-bounds' → False (drop)
"""
data: dict[str, Any] = getattr(event, "data", None) or {}
geo: dict[str, Any] = {}
@ -199,5 +207,18 @@ def event_in_areas(event: Any, areas: list[MonitoringArea]) -> bool:
geo["centroid"] = [lon, lat] # GeoJSON coordinate order: [lon, lat]
geom_json = build_geom_json(geo if geo else None)
result = classify_geom_areas(geom_json, areas)
return result != "out-of-bounds"
return classify_geom_areas(geom_json, areas)
def event_in_areas(event: Any, areas: list[MonitoringArea]) -> bool:
"""Return True if the event's geometry intersects any configured area.
Back-compat bool wrapper over ``classify_event_areas``. Retains the
original fail-OPEN semantics (matching Central):
'null-geom' (no geometry at all) → True (keep)
'invalid-geom' (parse/Shapely error) → True (keep)
'no-area' (areas list empty) → True (keep)
'in-bounds' → True (keep)
'out-of-bounds' → False (drop)
"""
return classify_event_areas(event, areas) != "out-of-bounds"

51
work/meshai/env/nws.py vendored
View file

@ -24,7 +24,8 @@ class NWSAlertsAdapter:
derived_areas = coverage["areas"]
if not derived_areas:
# bbox overlaps no state — keep config areas so the API call
# remains well-formed, but retain the bbox for geometry filter.
# remains well-formed. (The pipeline coverage gate now does the
# geometry filtering; this path is only a fetch-scope hint.)
logger.debug(
"NWS coverage: derived area list is empty (bbox outside all states); "
"falling back to config areas for API query"
@ -32,10 +33,8 @@ class NWSAlertsAdapter:
self._areas = config.areas or ["ID"]
else:
self._areas = derived_areas
self._coverage_bbox = coverage["bbox"]
else:
self._areas = config.areas or ["ID"]
self._coverage_bbox = None
self._user_agent = config.user_agent or "(meshai, ops@example.com)"
self._severity_min = config.severity_min or "moderate"
self._tick_interval = config.tick_seconds or 60
@ -134,6 +133,9 @@ class NWSAlertsAdapter:
"references": raw.get("references") or [],
"category": category,
"headline": raw.get("headline", ""),
# RAW GeoJSON alert geometry (Polygon/MultiPolygon/None). Read by
# the pipeline coverage gate as Event.data["geometry"].
"geometry": raw.get("geometry"),
}
return make_event(
@ -172,24 +174,6 @@ class NWSAlertsAdapter:
self._last_tick = now
return self._fetch()
def _in_coverage(self, event: dict) -> bool:
"""Return True iff the event should be kept under the coverage bbox filter.
Rules:
- If no coverage bbox is set, always keep.
- If the event has a numeric lat/lon centroid, drop it when outside the box.
- If the event has no centroid (lat/lon absent or None), keep it — the
state area= filter already scopes it; we can't box-filter without coords.
"""
if not self._coverage_bbox:
return True
lat = event.get("lat")
lon = event.get("lon")
if lat is None or lon is None:
return True # no geometry — pass through
from meshai.coverage import point_in_bbox
return point_in_bbox(lat, lon, self._coverage_bbox)
def _fetch(self) -> bool:
"""Fetch alerts from NWS API.
@ -284,14 +268,27 @@ class NWSAlertsAdapter:
"references": props.get("references") or [],
}
# Try to get centroid from geometry
# Attach the RAW GeoJSON alert geometry (Polygon / MultiPolygon /
# None) to the event. This is the authoritative field the pipeline
# coverage gate intersects against configured areas. Zone-only NWS
# alerts have geometry=None; the fail-closed gate drops those.
geom = feature.get("geometry")
event["geometry"] = geom
# Compute a best-effort centroid (fallback / nice-to-have; the
# geometry above is authoritative). Handle Polygon and MultiPolygon.
if geom and geom.get("coordinates"):
try:
coords = geom["coordinates"]
if geom.get("type") == "Polygon" and coords:
# Compute centroid of first ring
gtype = geom.get("type")
ring = None
if gtype == "Polygon" and coords:
# Outer ring of the polygon
ring = coords[0]
elif gtype == "MultiPolygon" and coords:
# Outer ring of the first polygon
ring = coords[0][0]
if ring:
lat_sum = sum(c[1] for c in ring)
lon_sum = sum(c[0] for c in ring)
event["lat"] = lat_sum / len(ring)
@ -299,12 +296,6 @@ class NWSAlertsAdapter:
except Exception:
pass
# Geometry filter: drop alerts whose centroid falls outside the
# coverage bbox. Alerts with no centroid are kept (state area=
# filter already scopes them; we can't box-filter without coords).
if not self._in_coverage(event):
continue
new_events.append(event)
# Check if data changed

51
work/meshai/notifications/pipeline/coverage_filter.py
View file

@ -5,9 +5,13 @@ same base interface (a callable ``.handle(event)``), constructed from config,
inserted in the pipeline chain adjacent to ToggleFilter.
Events that lie entirely outside ALL configured bounding boxes are dropped.
Events with no parseable geometry, or events whose source is in
``excluded_adapters``, are kept (fail-open, matching Central's choke-point
semantics).
Events whose source is in ``excluded_adapters`` are always kept.
Unlocatable events (no parseable geometry/centroid) are handled by category:
weather-alert categories (``FAIL_CLOSED_CATEGORIES``) are DROPPED — fail-closed,
matching Central's NWS behaviour and closing the zone-only-advisory leak — while
all other categories are KEPT (fail-open, matching Central's choke-point
semantics; quake/fire events always carry a centroid anyway).
When coverage is disabled (``coverage.enabled`` is False) or no areas are
configured (``areas`` is empty), the filter is a no-op and passes everything.
@ -17,7 +21,19 @@ import logging
from typing import Callable
from meshai.notifications.events import Event
from meshai.coverage_area import MonitoringArea, event_in_areas
from meshai.coverage_area import MonitoringArea, classify_event_areas
# Weather-alert categories (from env/nws.py::_derive_category) for which an
# event that cannot be geographically located is DROPPED (fail-closed),
# matching Central's NWS behaviour. The real LA leak was a zone-only Heat
# Advisory with no polygon and no centroid — fail-open kept it; this drops it.
FAIL_CLOSED_CATEGORIES = {
"weather_warning",
"weather_watch",
"weather_advisory",
"weather_statement",
}
class CoverageFilter:
@ -70,9 +86,16 @@ class CoverageFilter:
self._next(event)
return
if event_in_areas(event, self._areas):
verdict = classify_event_areas(event, self._areas)
# Positively in-bounds → keep. 'no-area' can't occur here (guarded by
# the `not self._areas` no-op above) but is treated as keep for safety.
if verdict in ("in-bounds", "no-area"):
self._next(event)
else:
return
# Positively out-of-bounds → drop, regardless of category.
if verdict == "out-of-bounds":
self._logger.debug(
"DROPPED event %s — out of all coverage areas "
"(source=%s category=%s)",
@ -80,3 +103,19 @@ class CoverageFilter:
event.source,
event.category,
)
return
# Unlocatable ('null-geom' / 'invalid-geom'): fail CLOSED for weather
# alerts (drop — matches Central), fail OPEN for everything else (keep).
if event.category in FAIL_CLOSED_CATEGORIES:
self._logger.debug(
"DROPPED event %s — unlocatable weather alert (%s), failing "
"closed (source=%s category=%s)",
event.id,
verdict,
event.source,
event.category,
)
return
self._next(event)

272
work/tests/test_coverage_area.py
View file

@ -15,6 +15,7 @@ Idaho reference area (magic-valley to eastern border, generous):
from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
import pytest
@ -22,11 +23,16 @@ from meshai.coverage_area import (
MonitoringArea,
areas_from_config,
build_geom_json,
classify_event_areas,
classify_geom_areas,
event_in_areas,
)
from meshai.env.nws import NWSAlertsAdapter
from meshai.notifications.events import make_event
from meshai.notifications.pipeline.coverage_filter import CoverageFilter
from meshai.notifications.pipeline.coverage_filter import (
CoverageFilter,
FAIL_CLOSED_CATEGORIES,
)
from meshai.config import Coverage
@ -515,3 +521,267 @@ class TestCoverageFilterPipelineWiring:
)
bus.emit(id_event)
assert len(received_by_tee) == 1, "Idaho event must reach _tee"
# ===========================================================================
# Idaho + MultiPolygon geometry fixtures (for the fail-closed / geometry work)
# ===========================================================================
# A polygon firmly inside the Idaho box (centroid ~ 42.9N, -114.2W → Twin Falls)
ID_POLYGON_COORDS = [
[-115.0, 42.5],
[-113.0, 42.5],
[-113.0, 43.5],
[-115.0, 43.5],
[-115.0, 42.5],
]
ID_POLYGON_GEOM = {"type": "Polygon", "coordinates": [ID_POLYGON_COORDS]}
# A MultiPolygon whose FIRST polygon's outer ring is inside Idaho; a second,
# far-away polygon (in Kansas) exercises the "first polygon wins" centroid rule.
KS_POLYGON_COORDS = [
[-98.0, 38.0],
[-97.0, 38.0],
[-97.0, 39.0],
[-98.0, 39.0],
[-98.0, 38.0],
]
ID_MULTIPOLYGON_GEOM = {
"type": "MultiPolygon",
"coordinates": [[ID_POLYGON_COORDS], [KS_POLYGON_COORDS]],
}
# ===========================================================================
# classify_event_areas — 3-state classification helper
# ===========================================================================
class TestClassifyEventAreas:
def test_idaho_polygon_in_bounds(self):
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="ID advisory", data={"geometry": ID_POLYGON_GEOM},
)
assert classify_event_areas(event, IDAHO_AREAS) == "in-bounds"
def test_ca_polygon_out_of_bounds(self):
ca_geom = {"type": "Polygon", "coordinates": [CA_POLYGON_COORDS]}
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="CA advisory", data={"geometry": ca_geom},
)
assert classify_event_areas(event, IDAHO_AREAS) == "out-of-bounds"
def test_no_geometry_no_centroid_is_null_geom(self):
"""The exact zone-only-alert shape: geometry=None, no lat/lon."""
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="Heat Advisory (zone-only)", data={"geometry": None},
)
assert classify_event_areas(event, IDAHO_AREAS) == "null-geom"
def test_multipolygon_in_bounds(self):
event = make_event(
source="nws", category="weather_warning", severity="priority",
title="ID multipolygon warning", data={"geometry": ID_MULTIPOLYGON_GEOM},
)
assert classify_event_areas(event, IDAHO_AREAS) == "in-bounds"
# ===========================================================================
# CoverageFilter — fail-CLOSED for weather, fail-OPEN for everything else
# ===========================================================================
class TestCoverageFilterFailClosed:
def _make_filter(self, areas=None, excluded=None):
received: list = []
flt = CoverageFilter(
next_handler=received.append,
areas=areas if areas is not None else IDAHO_AREAS,
enabled=True,
excluded_adapters=set(excluded or []),
)
return flt, received
def test_fail_closed_categories_are_the_nws_categories(self):
"""Guard: the fail-closed set is exactly env/nws._derive_category's outputs."""
assert FAIL_CLOSED_CATEGORIES == {
"weather_warning", "weather_watch",
"weather_advisory", "weather_statement",
}
# -- THE REAL LA LEAK: zone-only advisory, no polygon, no centroid --------
def test_zone_only_advisory_no_geometry_dropped_failclosed(self):
"""REAL repro of the leak: a zone-only NWS Heat Advisory with NO polygon
and NO centroid, under an Idaho-only coverage area, is now DROPPED
(fail-closed). Previously the fail-open gate KEPT it → wrong-region leak.
"""
flt, received = self._make_filter()
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="Heat Advisory issued for Los Angeles County",
# zone-only: geometry is explicitly None, and there is NO lat/lon
data={"geometry": None},
)
assert event.lat is None and event.lon is None # truly unlocatable
flt.handle(event)
assert len(received) == 0, (
"zone-only weather advisory with no geometry must be DROPPED "
"(fail-closed) — this is the exact LA leak"
)
def test_zone_only_statement_no_geometry_dropped_failclosed(self):
"""A weather_statement (Special Weather Statement) with no geometry is
also fail-closed — all four NWS categories are covered."""
flt, received = self._make_filter()
event = make_event(
source="nws", category="weather_statement", severity="routine",
title="Special Weather Statement", data={"geometry": None},
)
flt.handle(event)
assert len(received) == 0
# -- out-of-bounds real polygon (CA) -------------------------------------
def test_ca_polygon_advisory_dropped_out_of_bounds(self):
flt, received = self._make_filter()
ca_geom = {"type": "Polygon", "coordinates": [CA_POLYGON_COORDS]}
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="CA advisory", data={"geometry": ca_geom},
)
flt.handle(event)
assert len(received) == 0, "out-of-bounds CA polygon must be dropped"
# -- in-bounds real polygon (ID) -----------------------------------------
def test_idaho_polygon_advisory_kept_in_bounds(self):
flt, received = self._make_filter()
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="ID advisory", data={"geometry": ID_POLYGON_GEOM},
)
flt.handle(event)
assert len(received) == 1, "in-bounds Idaho polygon must be kept"
def test_idaho_multipolygon_advisory_kept_in_bounds(self):
flt, received = self._make_filter()
event = make_event(
source="nws", category="weather_advisory", severity="routine",
title="ID multipolygon advisory", data={"geometry": ID_MULTIPOLYGON_GEOM},
)
flt.handle(event)
assert len(received) == 1, "in-bounds Idaho MultiPolygon must be kept"
# -- fail-OPEN preserved for non-weather ---------------------------------
def test_non_weather_no_geometry_kept_fail_open(self):
"""A non-weather event (quake) with no geometry AND no centroid is still
KEPT (fail-open unchanged) — proves we did not over-tighten other
adapters. Only the FAIL_CLOSED_CATEGORIES are dropped when unlocatable.
"""
flt, received = self._make_filter()
event = make_event(
source="usgs", category="earthquake", severity="priority",
title="Quake with no geometry", # no lat/lon, no data geometry
)
assert event.lat is None and event.lon is None
assert event.category not in FAIL_CLOSED_CATEGORIES
flt.handle(event)
assert len(received) == 1, "non-weather unlocatable event must be KEPT (fail-open)"
def test_swpc_no_geometry_kept_fail_open(self):
"""SWPC (space weather, global) with no geometry stays fail-open."""
flt, received = self._make_filter()
event = make_event(
source="swpc", category="kp_index", severity="priority",
title="Geomagnetic storm — no location",
)
flt.handle(event)
assert len(received) == 1
# ===========================================================================
# NWS adapter — attaches raw alert geometry + MultiPolygon centroid, and the
# geometry flows env-event → to_event → Event.data["geometry"]
# ===========================================================================
def _nws_adapter():
config = MagicMock()
config.areas = ["ID"]
config.user_agent = "(test, test@example.com)"
config.severity_min = "moderate"
config.tick_seconds = 60
return NWSAlertsAdapter(config)
def _mock_urlopen_with_features(features):
"""Return a context-manager mock whose .read() yields a geo+json payload."""
payload = json.dumps({"features": features}).encode("utf-8")
resp = MagicMock()
resp.read.return_value = payload
resp.__enter__.return_value = resp
resp.__exit__.return_value = False
return resp
class TestNWSGeometryAttach:
def _fetch_one(self, geometry):
adapter = _nws_adapter()
feature = {
"properties": {
"id": "urn:oid:nws-test-1",
"event": "Heat Advisory",
"severity": "Moderate",
"headline": "Heat Advisory",
"description": "Hot.",
"areaDesc": "Test County",
},
"geometry": geometry,
}
resp = _mock_urlopen_with_features([feature])
with patch("meshai.env.nws.urlopen", return_value=resp):
adapter._fetch()
events = adapter.get_events()
assert len(events) == 1
return adapter, events[0]
def test_polygon_geometry_attached_to_event(self):
"""The parsed env event carries event['geometry'] = the feature geometry."""
_, event = self._fetch_one(ID_POLYGON_GEOM)
assert event["geometry"] == ID_POLYGON_GEOM
def test_polygon_centroid_computed(self):
_, event = self._fetch_one(ID_POLYGON_GEOM)
# centroid of the ID polygon ring ~ (42.9N, -114.2W) — inside Idaho
assert 42.0 <= event["lat"] <= 44.0
assert -117.0 <= event["lon"] <= -111.0
def test_multipolygon_geometry_attached_and_centroid_computed(self):
"""MultiPolygon: geometry attached verbatim + centroid from first
polygon's outer ring (must land in Idaho, not the far-away 2nd polygon).
"""
_, event = self._fetch_one(ID_MULTIPOLYGON_GEOM)
assert event["geometry"] == ID_MULTIPOLYGON_GEOM
# first polygon is the Idaho one → centroid inside Idaho
assert 42.0 <= event["lat"] <= 44.0
assert -117.0 <= event["lon"] <= -111.0
def test_zone_only_geometry_is_none(self):
"""A zone-only alert (no geometry) yields event['geometry'] is None and
NO centroid — the exact unlocatable shape the gate fails closed on."""
_, event = self._fetch_one(None)
assert event["geometry"] is None
assert "lat" not in event and "lon" not in event
def test_to_event_carries_geometry_into_data(self):
"""Trace: env event['geometry'] → to_event → Event.data['geometry']."""
adapter, event = self._fetch_one(ID_POLYGON_GEOM)
pipeline_event = adapter.to_event(event)
assert pipeline_event.data["geometry"] == ID_POLYGON_GEOM
def test_to_event_carries_none_geometry_for_zone_only(self):
adapter, event = self._fetch_one(None)
pipeline_event = adapter.to_event(event)
assert pipeline_event.data["geometry"] is None

88
work/tests/test_coverage_wiring_2c.py
View file

@ -1,12 +1,16 @@
"""Tests for Phase 2c coverage-bbox wiring: nws + traffic.
"""Tests for Phase 2c coverage wiring: nws + traffic.
Verifies that:
1. traffic: coverage wins (9 grid corridors derived from bbox); None falls back
to config.corridors.
2. nws: coverage wins for areas + coverage_bbox; None falls back to config.areas
and leaves _coverage_bbox as None.
3. nws geometry filter: alerts inside bbox are kept; outside are dropped; alerts
with no centroid (lat/lon None) are always kept.
2. nws: coverage wins for the ``area=`` fetch-scope (``_areas``); None falls back
to config.areas.
NOTE: the old adapter-level ``_in_coverage`` / ``_coverage_bbox`` geometry
heuristic was removed — the pipeline coverage gate
(``notifications/pipeline/coverage_filter.py``) supersedes it and fails CLOSED
for zone-only weather alerts (see test_coverage_area.py). ``_areas`` remains as
a fetch-scope optimization only.
"""
from __future__ import annotations
@ -113,24 +117,12 @@ def test_nws_coverage_areas():
assert adapter._areas == cov["areas"]
def test_nws_coverage_bbox_set():
"""When coverage is provided, _coverage_bbox is set from coverage['bbox']."""
from meshai.env.nws import NWSAlertsAdapter
cov = _cov("nws")
assert cov is not None
cfg = _nws_cfg()
adapter = NWSAlertsAdapter(cfg, coverage=cov)
assert adapter._coverage_bbox == cov["bbox"]
assert adapter._coverage_bbox == IDAHO_BOX
def test_nws_fallback_to_config():
"""When coverage=None, _areas falls back to config.areas and _coverage_bbox is None."""
"""When coverage=None, _areas falls back to config.areas."""
from meshai.env.nws import NWSAlertsAdapter
cfg = _nws_cfg(areas=["ID", "OR"])
adapter = NWSAlertsAdapter(cfg, coverage=None)
assert adapter._areas == ["ID", "OR"]
assert adapter._coverage_bbox is None
def test_nws_fallback_config_areas_default():
@ -140,11 +132,14 @@ def test_nws_fallback_config_areas_default():
cfg.areas = None # force the falsy case
adapter = NWSAlertsAdapter(cfg, coverage=None)
assert adapter._areas == ["ID"]
assert adapter._coverage_bbox is None
def test_nws_empty_areas_fallback_for_api():
"""When derived areas is empty, config areas are used for the API call but bbox is kept."""
"""When derived areas is empty, config areas are used for the API call.
(The old _coverage_bbox geometry filter was removed; the pipeline coverage
gate now does geographic filtering. _areas is a fetch-scope hint only.)
"""
from meshai.env.nws import NWSAlertsAdapter
# Synthesise a coverage dict whose areas list is empty (bbox outside all states)
empty_areas_cov = {"areas": [], "bbox": [-1.0, 0.0, 1.0, 1.0]}
@ -152,54 +147,5 @@ def test_nws_empty_areas_fallback_for_api():
adapter = NWSAlertsAdapter(cfg, coverage=empty_areas_cov)
# API area= query falls back to config areas
assert adapter._areas == ["ID"]
# but the geometry filter bbox is still set
assert adapter._coverage_bbox == [-1.0, 0.0, 1.0, 1.0]
# ===========================================================================
# nws geometry filter (_in_coverage helper)
# ===========================================================================
def _make_nws_with_bbox(bbox=None):
"""Create a NWSAlertsAdapter with a specific coverage_bbox, no live config needed."""
from meshai.env.nws import NWSAlertsAdapter
cfg = _nws_cfg()
if bbox is not None:
cov = {"areas": ["ID"], "bbox": bbox}
return NWSAlertsAdapter(cfg, coverage=cov)
return NWSAlertsAdapter(cfg, coverage=None)
def test_nws_in_coverage_inside_bbox_kept():
"""Alert whose centroid is inside the bbox passes the filter."""
adapter = _make_nws_with_bbox(IDAHO_BOX)
# Twin Falls is inside the Idaho box
event = {"lat": 42.56, "lon": -114.47}
assert adapter._in_coverage(event) is True
def test_nws_in_coverage_outside_bbox_dropped():
"""Alert whose centroid is outside the bbox is filtered out."""
adapter = _make_nws_with_bbox(IDAHO_BOX)
# Seattle (47.6, -122.3) is well outside south-central Idaho box
event = {"lat": 47.6, "lon": -122.3}
assert adapter._in_coverage(event) is False
def test_nws_in_coverage_no_coords_kept():
"""Alert with no lat/lon is always kept (can't filter without coords)."""
adapter = _make_nws_with_bbox(IDAHO_BOX)
event = {} # no lat/lon keys at all
assert adapter._in_coverage(event) is True
event_none = {"lat": None, "lon": None}
assert adapter._in_coverage(event_none) is True
def test_nws_in_coverage_no_bbox_always_passes():
"""When no coverage bbox is set, _in_coverage always returns True."""
adapter = _make_nws_with_bbox(bbox=None)
assert adapter._coverage_bbox is None
# Even an "outside" coord passes when there is no bbox filter
event = {"lat": 47.6, "lon": -122.3}
assert adapter._in_coverage(event) is True
# The removed geometry filter no longer stashes a bbox on the adapter.
assert not hasattr(adapter, "_coverage_bbox")