Logan CusanoandClaude Opus 5.5
bdbde359a1
- Cap one refresh's NWS zone fetches at 15 s wall clock. They run under the feed lock, so a hung zone endpoint could otherwise stall every /weather/alerts request for minutes; unfinished zones retry next refresh. - Prod c2-core.env.j2 never set NWS_USER_AGENT, so the VM would have sent the contact-less placeholder. Default it to the site URL (overridable via vault_nws_user_agent); warn on the .env.example example.com placeholder too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
240 lines
11 KiB
Python
240 lines
11 KiB
Python
"""
|
|
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
|