"""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())


class _InsertSession:
    """async_session stand-in: fire insert succeeds (rowcount=1)."""

    def __init__(self):
        self.rowcount = 1

    async def execute(self, *a, **k):
        return self

    async def commit(self):
        return None

    async def __aenter__(self):
        return self

    async def __aexit__(self, *a):
        return False


def test_ingest_fire_row_geofence_when_cache_empty(monkeypatch):
    """Ingester process has empty geofence cache; still notify on FIRMS insert."""
    import geofence
    import ingestor
    import live_layers

    geofence._cache.clear()
    live_layers.aircraft_last_known.clear()
    notified = []

    async def fake_record(**kw):
        notified.append(kw)
        return 1

    async def no_markers(*a, **k):
        return []

    monkeypatch.setattr(ingestor, "async_session", _InsertSession)
    monkeypatch.setattr(geofence, "record_and_notify", fake_record)
    monkeypatch.setattr("tracks.recent_markers", no_markers)

    assert asyncio.run(ingest_fire_row(make_fire_msg())) is True
    assert len(notified) == 1
    assert notified[0]["source_kind"] == "firms"
    assert notified[0]["lat"] == 39.45678
    assert notified[0]["lon"] == -121.12345


def test_ingest_fire_row_correlates_from_hypertable_when_last_known_empty(monkeypatch):
    """FIRMS ingester has no ADS-B last-known; still correlate from aircraft_positions."""
    import geofence
    import ingestor
    import live_layers

    geofence._cache.clear()
    live_layers.aircraft_last_known.clear()
    correlated = []

    async def fake_record(**kw):
        return 0

    async def fake_recent(kind, limit=2000):
        assert kind == "aircraft"
        return [{
            "id": "acf001",
            "lat": 39.45,
            "lon": -121.12,
            "extra": {"type": "AT802"},
        }]

    async def fake_corr(fires, aircraft):
        correlated.append((fires, aircraft))
        return aircraft

    monkeypatch.setattr(ingestor, "async_session", _InsertSession)
    monkeypatch.setattr(geofence, "record_and_notify", fake_record)
    monkeypatch.setattr("tracks.recent_markers", fake_recent)
    monkeypatch.setattr("fire_aircraft.correlate_and_notify", fake_corr)

    assert asyncio.run(ingest_fire_row(make_fire_msg())) is True
    assert len(correlated) == 1
    assert correlated[0][1][0]["id"] == "acf001"
    assert correlated[0][0][0]["lat"] == 39.45678


def test_ingest_fire_rows_one_execute_one_commit(monkeypatch):
    """93k FIRMS points must not be 93k commits."""
    from ingestor import ingest_fire_rows

    class _Session:
        def __init__(self):
            self.executes = 0
            self.commits = 0
            self.rowcount = 3

        async def execute(self, *a, **k):
            self.executes += 1
            return self

        async def commit(self):
            self.commits += 1

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

    session = _Session()
    import ingestor
    monkeypatch.setattr(ingestor, "async_session", lambda: session)

    msgs = [
        make_fire_msg(latitude=39.1 + i * 0.01, longitude=-121.1)
        for i in range(3)
    ]

    async def no_corr(*a, **k):
        return []

    monkeypatch.setattr("fire_aircraft.correlate_and_notify", no_corr)
    monkeypatch.setattr("geofence.record_and_notify", no_corr)

    inserted = asyncio.run(ingest_fire_rows(msgs))
    assert inserted == 3
    assert session.executes == 1
    assert session.commits == 1


def test_fire_insert_chunk_stays_under_asyncpg_bind_limit():
    """asyncpg caps bind params at 32767 — a 93k-row INSERT dies."""
    from ingestor import FIRE_INSERT_CHUNK, FIRE_ROW_BIND_PARAMS

    assert FIRE_INSERT_CHUNK * FIRE_ROW_BIND_PARAMS < 32767
    assert FIRE_INSERT_CHUNK >= 500


def test_ingest_fire_rows_chunks_when_over_limit(monkeypatch):
    from ingestor import ingest_fire_rows
    import ingestor

    class _Session:
        def __init__(self):
            self.executes = 0
            self.commits = 0
            self.rowcount = 0

        async def execute(self, *a, **k):
            self.executes += 1
            self.rowcount = 2 if self.executes < 3 else 1
            return self

        async def commit(self):
            self.commits += 1

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

    session = _Session()
    monkeypatch.setattr(ingestor, "async_session", lambda: session)
    monkeypatch.setattr(ingestor, "FIRE_INSERT_CHUNK", 2)

    async def no_corr(*a, **k):
        return []

    monkeypatch.setattr("fire_aircraft.correlate_and_notify", no_corr)
    monkeypatch.setattr("geofence.record_and_notify", no_corr)

    msgs = [
        make_fire_msg(latitude=39.1 + i * 0.01, longitude=-121.1)
        for i in range(5)
    ]
    inserted = asyncio.run(ingest_fire_rows(msgs))
    assert inserted == 5
    assert session.executes == 3
    assert session.commits == 1