Merge pull request 'Weather backend: NWS active alerts + nearest-METAR wind, cached (Refs #237)' (#263) from feat/237-weather-api into main

2026-10-05 03:03:32 -04:00
parent 6d0d3f79ce bdbde359a1
commit f3b21cd432
8 changed files with 719 additions and 1 deletions
@@ -0,0 +1,239 @@
+"""
+server-26#237 — weather backend: NWS active alerts + nearest-METAR wind.
+
+Pins the failure shapes that would otherwise look like "no weather right now":
+an upstream error keeps last-good data AND reports the error, a schema change is
+an error (not an empty layer), and zone-based alerts (geometry null) are never
+dropped. Also pins the NWS contract: User-Agent sent, one upstream call per TTL.
+"""
+import asyncio
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.internal import weather
+from app.internal.auth import require_service_or_firebase_token
+
+client = TestClient(app)
+
+SQUARE = {"type": "Polygon", "coordinates": [[[-74.0, 41.0], [-73.5, 41.0], [-73.5, 41.5], [-74.0, 41.5], [-74.0, 41.0]]]}
+ZONE_POLY = {"type": "Polygon", "coordinates": [[[-75.0, 42.0], [-74.5, 42.0], [-74.5, 42.5], [-75.0, 42.0]]]}
+ZONE_A = "https://api.weather.gov/zones/forecast/NYZ067"
+ZONE_B = "https://api.weather.gov/zones/forecast/NYZ999"
+
+
+def _alert(aid, event, geometry=None, zones=()):
+    return {"id": f"https://api.weather.gov/alerts/{aid}", "type": "Feature", "geometry": geometry,
+            "properties": {"id": aid, "event": event, "severity": "Severe", "urgency": "Immediate",
+                           "headline": f"{event} issued by NWS Upton NY", "onset": "2026-10-05T10:00:00-04:00",
+                           "expires": "2026-10-05T12:00:00-04:00", "ends": None, "senderName": "NWS Upton NY",
+                           "areaDesc": "Westchester", "affectedZones": list(zones)}}
+
+
+class Resp:
+    def __init__(self, body=None, status=200):
+        self._body, self.status_code = body, status
+        self.content = b"" if body is None else b"{}"
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise RuntimeError(f"HTTP {self.status_code}")
+
+    def json(self):
+        return self._body
+
+
+class FakeHttp:
+    """Routes GETs by URL; records every call (url, params, headers)."""
+    def __init__(self, routes):
+        self.routes, self.calls = routes, []
+
+    def __call__(self, **kw):
+        return self
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        pass
+
+    async def get(self, url, params=None, headers=None):
+        self.calls.append((url, params, headers))
+        r = self.routes.get(url)
+        if r is None:
+            return Resp(status=404)
+        return r() if callable(r) else r
+
+
+def setup_function():
+    app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True}
+    weather.alerts = weather._AlertsFeed()
+    weather.zones.clear()
+    weather._wind_cells.clear()
+
+
+def teardown_function():
+    app.dependency_overrides.pop(require_service_or_firebase_token, None)
+
+
+def _nws(features, extra=None):
+    routes = {weather.NWS_ALERTS_URL: Resp({"type": "FeatureCollection", "features": features})}
+    routes.update(extra or {})
+    return FakeHttp(routes)
+
+
+def test_happy_path_normalises_and_sends_user_agent():
+    http = _nws([_alert("a1", "Severe Thunderstorm Warning", SQUARE)])
+    with patch.object(weather.httpx, "AsyncClient", http), \
+         patch.object(weather.settings, "nws_user_agent", "DRB (ops@example.com)"), \
+         patch.object(weather.settings, "nws_area", "ny, ct"):
+        feed = asyncio.run(weather.alerts.get())
+    url, params, headers = http.calls[0]
+    assert params == {"area": "NY,CT"}
+    assert headers["User-Agent"] == "DRB (ops@example.com)" and headers["Accept"] == "application/geo+json"
+    a = feed.alerts[0]
+    assert {k: a[k] for k in ("id", "event", "severity", "urgency", "sender_name", "expires")} == {
+        "id": "a1", "event": "Severe Thunderstorm Warning", "severity": "Severe", "urgency": "Immediate",
+        "sender_name": "NWS Upton NY", "expires": "2026-10-05T12:00:00-04:00"}
+    assert a["geometry"] == SQUARE and a["geometry_source"] == "alert"
+    assert a["bbox"] == [41.0, -74.0, 41.5, -73.5]
+    assert feed.error is None and feed.fetched_at is not None and feed.zone_only == []
+
+
+def test_null_geometry_resolves_zones_or_lands_in_zone_only():
+    http = _nws(
+        [_alert("z1", "Flood Watch", None, [ZONE_A]),          # zone resolves -> polygon
+         _alert("z2", "Heat Advisory", None, [ZONE_B]),        # zone 404s -> zone_only, not dropped
+         _alert("z3", "Wind Advisory", None, ["https://evil.example/zones/x"])],  # never fetched
+        {ZONE_A: Resp({"geometry": ZONE_POLY, "properties": {}})})
+    with patch.object(weather.httpx, "AsyncClient", http):
+        feed = asyncio.run(weather.alerts.get())
+    assert [a["id"] for a in feed.alerts] == ["z1"]
+    assert feed.alerts[0]["geometry"] == ZONE_POLY and feed.alerts[0]["geometry_source"] == "zones"
+    assert feed.alerts[0]["zones_unresolved"] == 0
+    assert sorted(z["id"] for z in feed.zone_only) == ["z2", "z3"]
+    assert feed.zone_only[0]["expires"] and feed.zone_only[0]["event"]
+    assert not any("evil.example" in c[0] for c in http.calls)
+
+    # second refresh: zone cache (24 h) means ZONE_A is not fetched again
+    weather.alerts._next_attempt = 0.0
+    with patch.object(weather.httpx, "AsyncClient", http):
+        asyncio.run(weather.alerts.get())
+    assert sum(1 for c in http.calls if c[0] == ZONE_A) == 1
+
+
+def test_upstream_500_keeps_last_good_and_reports_error():
+    ok = _nws([_alert("a1", "Tornado Warning", SQUARE)])
+    with patch.object(weather.httpx, "AsyncClient", ok):
+        asyncio.run(weather.alerts.get())
+    good_at = weather.alerts.fetched_at
+    weather.alerts._next_attempt = 0.0
+    with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.NWS_ALERTS_URL: Resp(status=500)})):
+        feed = asyncio.run(weather.alerts.get())
+    assert [a["id"] for a in feed.alerts] == ["a1"] and feed.fetched_at == good_at
+    assert "500" in feed.error
+    assert feed._next_attempt - __import__("time").time() <= weather.RETRY_AFTER_FAILURE_S + 1
+
+    r = client.get("/weather/alerts")
+    body = r.json()
+    assert r.status_code == 200 and body["error"] and len(body["alerts"]) == 1
+    assert body["fetched_at"] == good_at and body["credit"] == "National Weather Service"
+
+
+def test_schema_canary_trips_when_no_feature_parses():
+    http = FakeHttp({weather.NWS_ALERTS_URL: Resp({"features": [{"props": {}}, {"properties": {"name": "x"}}]})})
+    with patch.object(weather.httpx, "AsyncClient", http):
+        feed = asyncio.run(weather.alerts.get())
+    assert feed.alerts == [] and "schema" in feed.error
+
+
+def test_empty_feature_list_is_not_an_error():
+    with patch.object(weather.httpx, "AsyncClient", _nws([])):
+        feed = asyncio.run(weather.alerts.get())
+    assert feed.error is None and feed.fetched_at is not None and feed.alerts == []
+
+
+def test_cache_one_upstream_call_per_ttl_even_when_concurrent():
+    http = _nws([_alert("a1", "Tornado Warning", SQUARE)])
+
+    async def many():
+        await asyncio.gather(*(weather.alerts.get() for _ in range(8)))
+        await weather.alerts.get()
+
+    with patch.object(weather.httpx, "AsyncClient", http):
+        asyncio.run(many())
+    assert sum(1 for c in http.calls if c[0] == weather.NWS_ALERTS_URL) == 1
+
+
+def test_endpoint_bbox_filter_and_validation():
+    far = {"type": "Polygon", "coordinates": [[[-79.0, 42.8], [-78.5, 42.8], [-78.5, 43.0], [-79.0, 42.8]]]}  # Buffalo
+    http = _nws([_alert("near", "Flood Warning", SQUARE), _alert("far", "Lake Effect Snow Warning", far),
+                 _alert("zo", "Heat Advisory", None, [ZONE_B])])
+    with patch.object(weather.httpx, "AsyncClient", http):
+        r = client.get("/weather/alerts", params={"south": 40.9, "west": -74.2, "north": 41.2, "east": -73.4})
+        body = r.json()
+        assert r.status_code == 200 and [a["id"] for a in body["alerts"]] == ["near"]
+        assert [z["id"] for z in body["zone_only"]] == ["zo"]   # never bbox-filtered away
+        assert len(client.get("/weather/alerts").json()["alerts"]) == 2
+    assert client.get("/weather/alerts", params={"south": 41.4, "west": -74, "north": 40.9, "east": -73}).status_code == 400
+    assert client.get("/weather/alerts", params={"south": 40.9}).status_code == 400
+
+
+def _metar(icao, lat, lon, wdir, wspd, wgst=None, obs=None):
+    return {"icaoId": icao, "name": f"{icao} airport", "lat": lat, "lon": lon,
+            "obsTime": obs or int(__import__("time").time()) - 600, "reportTime": "2026-10-05T06:48:00.000Z",
+            "wdir": wdir, "wspd": wspd, "wgst": wgst, "rawOb": f"METAR {icao} ..."}
+
+
+def test_wind_picks_nearest_station_and_caches_per_cell():
+    rows = [_metar("KLGA", 40.7794, -73.8803, 310, 6),
+            _metar("KHPN", 41.067, -73.7076, "VRB", 4, 18),
+            _metar("KHPN", 41.067, -73.7076, 200, 3, obs=1)]   # older duplicate report
+    http = FakeHttp({weather.METAR_URL: Resp(rows)})
+    with patch.object(weather.httpx, "AsyncClient", http):
+        r = client.get("/weather/wind", params={"lat": 41.03, "lon": -73.76})   # White Plains
+        body = r.json()
+        assert r.status_code == 200 and body["available"] is True
+        w = body["wind"]
+        assert w["station"] == "KHPN" and w["distance_km"] < 10
+        assert w["wind_dir_deg"] is None and w["variable"] is True and w["wind_kt"] == 4 and w["gust_kt"] == 18
+        assert w["stale"] is False and body["credit"]
+        client.get("/weather/wind", params={"lat": 41.04, "lon": -73.77})   # same cell -> cached
+    assert len(http.calls) == 1
+    params, headers = http.calls[0][1], http.calls[0][2]
+    s, w_, n, e = (float(v) for v in params["bbox"].split(","))
+    assert s < 41.03 < n and w_ < -73.76 < e and params["format"] == "json"
+    assert "User-Agent" in headers
+
+
+def test_wind_unavailable_reasons():
+    with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.METAR_URL: Resp(None, status=204)})):
+        body = client.get("/weather/wind", params={"lat": 30.0, "lon": -50.0}).json()
+    assert body["available"] is False and "no METAR station" in body["reason"] and body["error"] is None
+
+    weather._wind_cells.clear()
+    with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.METAR_URL: Resp(status=503)})):
+        body = client.get("/weather/wind", params={"lat": 41.03, "lon": -73.76}).json()
+    assert body["available"] is False and "unavailable" in body["reason"] and "503" in body["error"]
+
+
+def test_slow_zone_endpoint_cannot_hold_the_feed_lock():
+    """Zone fetches run under the feed lock; a hung zone endpoint must not stall
+    every /weather/alerts request -- the alert lands in zone_only instead."""
+    import time
+
+    class SlowZones(FakeHttp):
+        async def get(self, url, params=None, headers=None):
+            if url.startswith(weather.NWS_ZONE_PREFIX):
+                await asyncio.sleep(5)
+            return await super().get(url, params, headers)
+
+    http = SlowZones({weather.NWS_ALERTS_URL: Resp({"features": [_alert("z1", "Flood Watch", None, [ZONE_A])]}),
+                      ZONE_A: Resp({"geometry": ZONE_POLY})})
+    with patch.object(weather, "ZONE_BUDGET_S", 0.2), patch.object(weather.httpx, "AsyncClient", http):
+        t0 = time.time()
+        feed = asyncio.run(weather.alerts.get())
+    assert time.time() - t0 < 2
+    assert feed.error is None and [z["id"] for z in feed.zone_only] == ["z1"]
+    assert weather.zones.get(ZONE_A) == (False, None)   # not cached as a failure; retried next refresh