"""Unit tests for the NASA FIRMS CSV parser / timestamp normalization.""" import asyncio from datetime import datetime, timezone from fire_sources import normalize_acq_time, parse_firms_csv, ingest_fires # Grounded against real FIRMS VIIRS area-CSV output. SAMPLE_CSV = """latitude,longitude,bright_ti4,scan,track,acq_date,acq_time,satellite,instrument,confidence,version,bright_ti5,frp,daynight -16.28359,29.40531,295.78,0.50,0.66,2025-06-06,1,N20,VIIRS,n,2.0NRT,284.11,1.17,N -16.28190,29.40279,303.31,0.50,0.66,2025-06-06,1,N20,VIIRS,n,2.0NRT,284.43,0.67,N -14.98900,28.36286,341.04,0.41,0.60,2025-06-06,1,N20,VIIRS,n,2.0NRT,279.77,4.59,N 15.17397,-11.28343,341.40,0.45,0.47,2025-06-06,1425,N20,VIIRS,l,2.0NRT,315.18,7.91,D 15.45870,-11.13616,339.25,0.46,0.47,2025-06-06,1425,N20,VIIRS,l,2.0NRT,312.05,9.24,D """ def test_normalize_acq_time_single_digit_hour(): # acq_time "1" (HHMM int) -> 00:01 UTC dt = normalize_acq_time("2025-06-06", "1") assert dt == datetime(2025, 6, 6, 0, 1, tzinfo=timezone.utc) def test_normalize_acq_time_four_digit(): dt = normalize_acq_time("2025-06-06", 1425) assert dt == datetime(2025, 6, 6, 14, 25, tzinfo=timezone.utc) def test_normalize_acq_time_bad_values(): assert normalize_acq_time(None, 100) is None assert normalize_acq_time("2025-06-06", None) is None assert normalize_acq_time("2025-06-06", "") is None assert normalize_acq_time("not-a-date", 100) is None def test_parse_firms_csv_happy_path(): points = parse_firms_csv(SAMPLE_CSV) assert len(points) == 5 first = points[0] assert first["latitude"] == -16.28359 assert first["longitude"] == 29.40531 assert first["brightness"] == 295.78 assert first["confidence"] == "n" assert first["satellite"] == "N20" assert first["instrument"] == "VIIRS" assert first["frp"] == 1.17 assert first["daynight"] == "N" assert first["acq_time"] == "2025-06-06T00:01:00+00:00" # late-day acquisition (acq_time 1425) parses to 14:25 UTC assert points[3]["acq_time"] == "2025-06-06T14:25:00+00:00" assert points[3]["confidence"] == "l" def test_parse_firms_csv_skips_legend_line(): # FIRMS occasionally prepends a legend/info line before the real header. with_legend = ( "Active Fire Data from VIIRS (S-NPP) — near real time\n" + SAMPLE_CSV ) points = parse_firms_csv(with_legend) assert len(points) == 5 assert points[0]["latitude"] == -16.28359 def test_parse_firms_csv_empty_and_garbage(): assert parse_firms_csv("") == [] assert parse_firms_csv("not a csv at all\njust text\n") == [] # Header present but a data row that is too short / has junk lat-lon junk = SAMPLE_CSV.splitlines()[0] + "\n1,2\n" assert parse_firms_csv(junk) == [] def test_ingest_fires_idles_without_map_key(monkeypatch): # No key anywhere -> returns 0 without attempting a network call. monkeypatch.setenv("FIRMS_MAP_KEY", "") monkeypatch.setattr( "fire_sources.get_api_key", lambda name: _async_none() ) assert asyncio.run(ingest_fires()) == 0 def test_ingest_fires_uses_keystore_key(monkeypatch): # Key saved via the dashboard Keys page (Postgres) is picked up. from upstream_cache import firms_cache firms_cache.clear() monkeypatch.setenv("FIRMS_MAP_KEY", "") monkeypatch.setattr( "fire_sources.get_api_key", lambda name: _async_return("a" * 32), ) captured = {} class FakeResp: status_code = 200 text = SAMPLE_CSV def raise_for_status(self): pass class FakeClient: def __init__(self, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *exc): return False async def get(self, url): captured.setdefault("urls", []).append(url) captured["url"] = url return FakeResp() monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient) published = [] async def fake_publish(points): published.extend(points) return len(points) monkeypatch.setattr("fire_sources.persist_hotspots", fake_publish) assert asyncio.run(ingest_fires()) == 10 # NOAA-20 + NOAA-21 dual-write assert "a" * 32 in captured["url"] assert any("VIIRS_NOAA20_NRT" in u for u in captured["urls"]) assert any("VIIRS_NOAA21_NRT" in u for u in captured["urls"]) def _reset_firms_poll_state(): from upstream_cache import firms_cache import fire_sources firms_cache.clear() if hasattr(fire_sources, "_csv_digest"): fire_sources._csv_digest.clear() if hasattr(fire_sources, "_seen_ids"): fire_sources._seen_ids.clear() def _fake_firms_http(monkeypatch, bodies_by_call: list[str] | None = None, body: str = SAMPLE_CSV): hits = {"n": 0} class FakeResp: def __init__(self, text): self.text = text def raise_for_status(self): pass class FakeClient: def __init__(self, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *exc): return False async def get(self, url): idx = hits["n"] hits["n"] += 1 if bodies_by_call is not None: text = bodies_by_call[min(idx, len(bodies_by_call) - 1)] else: text = body return FakeResp(text) monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32) monkeypatch.setattr("fire_sources.FIRMS_DATASETS", ["VIIRS_NOAA20_NRT"]) monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient) return hits def test_ingest_fires_skips_unchanged_csv(monkeypatch): """Same FIRMS CSV must not be re-parsed into a 100k-row ON CONFLICT insert.""" _reset_firms_poll_state() hits = _fake_firms_http(monkeypatch) persisted = [] async def fake_persist(points): persisted.append(len(points)) return len(points) monkeypatch.setattr("fire_sources.persist_hotspots", fake_persist) assert asyncio.run(ingest_fires()) == 5 assert persisted == [5] firms_cache_hits = hits["n"] persisted.clear() assert asyncio.run(ingest_fires()) == 0 assert persisted == [] # TTL cache may skip HTTP; either way we must not persist again. assert hits["n"] >= firms_cache_hits def test_ingest_fires_persists_only_new_hotspots(monkeypatch): """When the CSV grows, persist the delta — not the whole 2-day dump.""" _reset_firms_poll_state() extra = ( SAMPLE_CSV + "16.00000,-12.00000,340.00,0.40,0.40,2025-06-06,1500,N20,VIIRS,h,2.0NRT,310.00,8.00,D\n" ) hits = _fake_firms_http(monkeypatch, bodies_by_call=[SAMPLE_CSV, extra]) persisted = [] async def fake_persist(points): persisted.append([p["latitude"] for p in points]) return len(points) monkeypatch.setattr("fire_sources.persist_hotspots", fake_persist) from upstream_cache import firms_cache assert asyncio.run(ingest_fires()) == 5 firms_cache.clear() # force the next poll to see the grown CSV persisted.clear() assert asyncio.run(ingest_fires()) == 1 assert persisted == [[16.0]] assert hits["n"] == 2 def _async_return(value): async def inner(): return value return inner() def _async_none(): return _async_return(None)