627990efde
Coherent merge of two coordinated features on the shared working tree: FIRMS fire heatmap (backend, t_6e404c14): - app/fire_sources.py: fetch FIRMS VIIRS area CSV (free MAP_KEY) -> NATS events.fire - fires hypertable (TimescaleDB, 1-day chunks) with natural-key PK (latitude, longitude, acq_time, satellite); idempotent ON CONFLICT DO NOTHING - alembic/versions/002_fires.py; GET /api/fires?bbox=&since= (JSON only) - POST /api/ingest/fires; ~15 min poll loop (FIRMS_INTERVAL=900) in ingester - env-driven config (FIRMS_MAP_KEY/DATASET/BBOX/INTERVAL); docs/firms.md covers the zero-cost GIBS VIIRS_SNPP_Thermal_Anomalies_375m_All tile alternative - 18 tests (parser, mapping, idempotency, API contract) verified vs real TimescaleDB+PostGIS (localhost/osint-dashboard-pg image) API keys page (frontend, t_4433cff2): - app/keystore.py: api_keys table (self-creating), FIRMS/GEMINI/TELEGRAM registry with format validation, ****last4 masking, get_api_key() - GET/POST/DELETE /api/keys (never returns full values); Keys tab in index.html DB_NULL_POOL env switch in app/database.py enables a NullPool for tests / short-lived processes that open a fresh event loop per unit.
87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
"""Unit + integration tests for fire message routing and idempotent storage.
|
|
|
|
The row-mapping logic (_fire_row_from_msg) is pure and tested everywhere; the
|
|
DB-backed idempotency tests are `integration` and auto-skip without a live DB.
|
|
"""
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
|
|
from conftest import make_fire_msg, requires_db
|
|
|
|
from ingestor import _fire_row_from_msg, ingest_fire_row, ingest_event
|
|
|
|
|
|
# ── Row mapping (pure, no DB) ────────────────────────────────────────────
|
|
|
|
def test_fire_row_from_msg_maps_fields():
|
|
row = _fire_row_from_msg(make_fire_msg())
|
|
assert row is not None
|
|
assert row["latitude"] == 39.45678
|
|
assert row["longitude"] == -121.12345
|
|
assert row["brightness"] == 341.40
|
|
assert row["confidence"] == "h"
|
|
assert row["satellite"] == "N"
|
|
assert row["acq_time"] == datetime(2026, 8, 24, 18, 10, tzinfo=timezone.utc)
|
|
assert row["frp"] == 12.4
|
|
assert row["daynight"] == "D"
|
|
|
|
|
|
def test_fire_row_from_msg_rejects_missing_natural_key():
|
|
for key in ("latitude", "longitude", "acq_time", "satellite"):
|
|
msg = make_fire_msg(**{key: None})
|
|
assert _fire_row_from_msg(msg) is None, f"{key}=None should be dropped"
|
|
|
|
|
|
def test_fire_row_from_msg_rejects_bad_timestamp():
|
|
msg = make_fire_msg(acq_time="not-a-timestamp")
|
|
assert _fire_row_from_msg(msg) is None
|
|
|
|
|
|
def test_ingest_event_routes_fire_away_from_events(monkeypatch):
|
|
"""source_type='fire' must hit the fire path, never the generic events insert."""
|
|
calls = {"fire": 0, "events": 0}
|
|
|
|
async def fake_fire_row(msg):
|
|
calls["fire"] += 1
|
|
return True
|
|
|
|
async def boom(*a, **k): # the generic events insert must not be reached
|
|
calls["events"] += 1
|
|
raise AssertionError("fire message leaked into the events insert path")
|
|
|
|
import ingestor
|
|
monkeypatch.setattr(ingestor, "ingest_fire_row", fake_fire_row)
|
|
monkeypatch.setattr(ingestor.events_table, "insert", boom)
|
|
|
|
asyncio.run(ingest_event(make_fire_msg()))
|
|
assert calls["fire"] == 1
|
|
assert calls["events"] == 0
|
|
|
|
|
|
# ── Idempotent storage against a live database ───────────────────────────
|
|
|
|
@requires_db
|
|
def test_ingest_fire_row_idempotent(clean_fires):
|
|
async def run():
|
|
# First insert persists.
|
|
assert await ingest_fire_row(make_fire_msg()) is True
|
|
# Same natural key on a later 15-min poll -> silently ignored.
|
|
assert await ingest_fire_row(make_fire_msg(brightness=999.0)) is False
|
|
# Same point/time but a different satellite is a distinct detection.
|
|
assert await ingest_fire_row(make_fire_msg(satellite="N20")) is True
|
|
# Same point but a different acquisition time is a distinct detection.
|
|
assert await ingest_fire_row(
|
|
make_fire_msg(acq_time="2026-08-24T19:10:00+00:00")
|
|
) is True
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
@requires_db
|
|
def test_ingest_fire_row_drops_malformed(clean_fires):
|
|
async def run():
|
|
assert await ingest_fire_row(make_fire_msg(latitude=None)) is False
|
|
assert await ingest_fire_row(make_fire_msg(acq_time="garbage")) is False
|
|
|
|
asyncio.run(run())
|