88 lines
3.2 KiB
Python
88 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())
|