meshai/work/tests/test_tle_fetch.py

"""Tests for the native Celestrak TLE fetcher (env.tle_fetch).

Covers epoch parsing, 3-line block parsing, upsert into the shared
sat_tles table, latest-epoch-wins on re-fetch, and malformed-block
tolerance. HTTP is monkeypatched — no network.
"""
from __future__ import annotations

import datetime as _dt

import pytest

from meshai.env.tle_fetch import (
    TLEFetchAdapter,
    parse_tle_block,
    parse_tle_epoch,
)
from meshai.central.tle_handler import get_tle_by_norad
from meshai.config import SatpassConfig
from meshai.persistence import get_db


# -- time-relative TLE fixture generation --------------------------------
#
# These used to be hardcoded 3-line sets with a fixed 2026-06-30..07-02
# epoch. get_tle_by_norad() filters on `epoch >= now() - STALE_DAYS`
# (STALE_DAYS=14, real wall-clock `now`), so a fixed-date fixture ages out
# 14 days after it's written and silently starts failing
# test_tick_upserts_into_sat_tles / test_latest_epoch_wins_on_refetch.
# Deriving the epoch from the real current time at test-collection time
# means these can never expire again.


def _tle_epoch_field(dt: _dt.datetime) -> str:
    """Build a TLE line-1 epoch field (cols 19-32): YYDDD.DDDDDDDD."""
    yy = dt.year % 100
    start_of_year = _dt.datetime(dt.year, 1, 1, tzinfo=_dt.timezone.utc)
    doy_frac = (dt - start_of_year).total_seconds() / 86400.0 + 1.0
    doy_int = int(doy_frac)
    frac = doy_frac - doy_int
    return f"{yy:02d}{doy_int:03d}.{round(frac * 1e8):08d}"


def _tle_checksum(line_without_checksum_digit: str) -> int:
    """Standard TLE mod-10 checksum: sum of digits, '-' counts as 1,
    everything else 0."""
    total = 0
    for ch in line_without_checksum_digit:
        if ch.isdigit():
            total += int(ch)
        elif ch == "-":
            total += 1
    return total % 10


def _iss_tle_block(dt: _dt.datetime, element_set_num: str, revnum: str) -> str:
    """Build a valid 3-line ISS (ZARYA) TLE with the given epoch. All
    other orbital elements are fixed (arbitrary but internally
    consistent) — only epoch/checksum/element-set/revnum vary, mirroring
    the original hand-written fixtures."""
    line1_body = (f"1 25544U 98067A   {_tle_epoch_field(dt)}  .00016717  "
                  f"00000-0  10270-3 0  {element_set_num}")
    line1 = f"{line1_body}{_tle_checksum(line1_body)}"
    line2 = (f"2 25544  51.6400 208.9163 0007417  17.6777  85.6621 "
             f"15.54225995 {revnum}")
    return f"ISS (ZARYA)\n{line1}\n{line2}\n"


# Base epoch: 2 days ago. Comfortably inside the 14-day STALE_DAYS window
# no matter when the suite runs, and never in the future.
_NOW = _dt.datetime.now(_dt.timezone.utc)
_BASE_EPOCH = _NOW - _dt.timedelta(days=2)

# A valid ISS 3-line set at the base epoch.
ISS_TLE = _iss_tle_block(_BASE_EPOCH, "900", "12345")
# Same satellite, a NEWER epoch (1 day after base).
ISS_TLE_NEWER = _iss_tle_block(_BASE_EPOCH + _dt.timedelta(days=1), "901", "12347")
# Same satellite, an OLDER epoch (1 day before base).
ISS_TLE_OLDER = _iss_tle_block(_BASE_EPOCH - _dt.timedelta(days=1), "900", "12343")

# ISO epoch string parse_tle_epoch() will produce for ISS_TLE's line 1 --
# computed via the real parser (not re-derived independently) so
# assertions can't drift from the actual parsing behavior under test.
ISS_TLE_EPOCH_ISO = parse_tle_epoch(ISS_TLE.splitlines()[1])


def _adapter(**overrides) -> TLEFetchAdapter:
    cfg = SatpassConfig(enabled=True, feed_source="native",
                        tle_groups=[], norad_ids=[25544], **overrides)
    return TLEFetchAdapter(cfg)


# -- epoch parsing ------------------------------------------------------------

def test_parse_tle_epoch_iso():
    iso = parse_tle_epoch(
        "1 25544U 98067A   26182.50000000  .00016717  00000-0  10270-3 0  9008")
    assert iso.startswith("2026-07-01T12:00:00")
    assert "+00:00" in iso


def test_parse_tle_epoch_two_digit_year_window():
    # YY=98 -> 1998 (>= 57 maps to 1900s).
    iso = parse_tle_epoch("1 25544U 98067A   98001.00000000  .0  0  0 0  1")
    assert iso.startswith("1998-01-01")


# -- block parsing ------------------------------------------------------------

def test_parse_tle_block_basic():
    recs = parse_tle_block(ISS_TLE)
    assert len(recs) == 1
    r = recs[0]
    assert r["norad_id"] == 25544
    assert r["name"] == "ISS (ZARYA)"
    assert r["line1"].startswith("1 25544")
    assert r["line2"].startswith("2 25544")
    assert r["epoch"] == ISS_TLE_EPOCH_ISO


def test_parse_tle_block_skips_malformed():
    # First triple is garbage (line1 doesn't start with "1 "); second is valid.
    block = (
        "GARBAGE SAT\n"
        "not a real line1\n"
        "also not line2\n"
        + ISS_TLE
    )
    recs = parse_tle_block(block)
    norads = [r["norad_id"] for r in recs]
    assert 25544 in norads
    # The garbage entry must not have produced a record.
    assert all(isinstance(n, int) for n in norads)


# -- fetch + upsert -----------------------------------------------------------

def test_tick_upserts_into_sat_tles(monkeypatch):
    adapter = _adapter()
    monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE)

    changed = adapter.tick(now=1_000_000)
    assert changed is True

    row = get_tle_by_norad(25544)
    assert row is not None
    assert row["name"] == "ISS (ZARYA)"
    assert row["line1"].startswith("1 25544")
    assert row["line2"].startswith("2 25544")
    assert row["epoch"] == ISS_TLE_EPOCH_ISO


def test_latest_epoch_wins_on_refetch(monkeypatch):
    adapter = _adapter()

    monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE)
    assert adapter.tick(now=1_000_000) is True
    first = get_tle_by_norad(25544)["epoch"]

    # A newer epoch replaces it.
    adapter._last_tick = 0  # bypass interval gate for the test
    monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE_NEWER)
    assert adapter.tick(now=2_000_000) is True
    newer = get_tle_by_norad(25544)["epoch"]
    assert newer > first

    # An older epoch is ignored (no write).
    adapter._last_tick = 0
    monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE_OLDER)
    changed = adapter.tick(now=3_000_000)
    assert changed is False
    assert get_tle_by_norad(25544)["epoch"] == newer


def test_malformed_block_does_not_crash_tick(monkeypatch):
    adapter = _adapter()
    monkeypatch.setattr(
        adapter, "_fetch",
        lambda url: "COMPLETE GARBAGE\nno lines here\n")
    # Should complete without raising and write nothing.
    changed = adapter.tick(now=1_000_000)
    assert changed is False
    assert get_tle_by_norad(25544) is None


def test_fetch_error_is_isolated(monkeypatch):
    adapter = _adapter()

    def boom(url):
        raise RuntimeError("HTTP 503")

    monkeypatch.setattr(adapter, "_fetch", boom)
    changed = adapter.tick(now=1_000_000)
    assert changed is False
    assert adapter.health_status["last_error"] is not None
    assert adapter.health_status["consecutive_errors"] == 1


def test_storage_only_no_events(monkeypatch):
    adapter = _adapter()
    monkeypatch.setattr(adapter, "_fetch", lambda url: ISS_TLE)
    adapter.tick(now=1_000_000)
    assert adapter.get_events() == []
    assert adapter.to_event({}) is None


def test_interval_gate_skips_early_ticks(monkeypatch):
    adapter = _adapter(tle_refresh_seconds=21600)
    calls = []
    monkeypatch.setattr(adapter, "_fetch",
                        lambda url: calls.append(url) or ISS_TLE)
    adapter.tick(now=1_000_000.0)    # first tick fetches
    adapter.tick(now=1_000_100.0)    # 100s later, within interval -> skipped
    assert len(calls) == 1


def test_no_targets_configured_is_noop():
    cfg = SatpassConfig(enabled=True, feed_source="native",
                        tle_groups=[], norad_ids=[])
    adapter = TLEFetchAdapter(cfg)
    assert adapter.tick(now=1_000_000) is False


def test_group_and_catnr_urls():
    cfg = SatpassConfig(tle_groups=["weather"], norad_ids=[25544])
    adapter = TLEFetchAdapter(cfg)
    urls = [u for _, u in adapter._targets()]
    assert any("GROUP=weather&FORMAT=tle" in u for u in urls)
    assert any("CATNR=25544&FORMAT=tle" in u for u in urls)
    assert all(u.startswith("https://celestrak.org/NORAD/elements/gp.php") for u in urls)