meshai/work/tests/test_satpass_persisted_timer.py
malice 9f06930a0d
chore: excise the dead Central NATS consumer path (-11,328 LOC) (#144)
* chore: excise the dead Central NATS consumer path

Central was retired and its database dropped 2026-07-15; its NATS broker no
longer exists. Verified against the live CT108 deployment: all 12 adapters
run feed_source=native, zero on central, and central.enabled is False
(default, never overridden). The consumer and its handlers were unreachable.

Removed:
- central/consumer.py and 6 dead handlers (nws, quake, swpc, nwis, avy,
  incident) -- their handle_* entrypoints were reachable only from the
  consumer's dispatch
- the Central wiring in main.py (init, guarded start, retry loop, stop path)
- the dead config surface: CentralConsumerConfig, EnvironmentalConfig.central,
  adapter_config ("central","severity_thresholds") and its display block
- the nats-py dependency (consumer.py was its only importer)
- 19 test files that exercised only the dead path

KEPT -- these live under central/ but are imported directly by native
adapters, and deleting them would break production:
- wfigs_handler.py: firms_handler._handle_pass_boundary() calls its _render()
  on the live FIRMS growth-fire path (env/firms.py -> ingest_hotspot_pixel)
- firms_handler, satpass_handler, tle_handler: split files whose handle_*
  entrypoints are dead but whose engines are live. Left intact; splitting
  them is separate work.
- pass_predictor, budget, idaho_gauge_sites: fully live.

The usgs_quake keys global_mag_floor / regional_mag_floor / regional_centroid
/ regional_radius_mi / broadcast_pager_alerts are NOT removed despite comments
labelling them "CENTRAL-PATH ONLY" -- notifications/gating/quake.py reads them
unconditionally in the native path. Those comments are corrected separately.

Test-count note: the suite drops ~425 tests. Most were migration PARITY tests
whose sole purpose was proving the native rewrite byte-matched the Central
handler (golden byte-parity, cross-source identity, gate-sequence replay).
With the handler deleted there is nothing left to compare against, so they
cannot exist. Native-only tests were kept and reworked where a test reached
for a central symbol incidentally. This is a real coverage loss, accepted
deliberately: the parity harness proved the refactor faithful, and git
history preserves the originals.

Suite: 1984 passed, 6 failed -- the same 6 pre-existing failures as main
(stale SCHEMA_VERSION x3, expired TLE fixtures x2, one order-dependent),
all being fixed on fix/green-test-suite. No new failures.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(nws): restore native-only golden coverage for the wire formatter

Commit ca751fb5 deleted the Central nws_handler parity harness along with
the handler itself, which took the ONLY tests that pinned formatters.nws
.format()'s literal wire output. Gate-sequence and schema-conformance
tests already survived natively; the formatter's actual rendered text did
not have any native-only regression net.

Add TestFormatterGolden to test_nws_refactor.py: 3 real-fixture cases plus
6 hand-built pathological cases mined from the deleted test_nws_handler.py
(SVR path-sampling, the "no dangling separator" regression, TOR on-ground
vs radar-indicated, FFW flood-cause detection). Every literal was verified
by temporarily restoring the pre-excision central.nws_handler._render()
from git history (ca751fb5^) in a throwaway, uncommitted script, confirming
byte-identical output against the current native format() for all 37 real
fixtures (nws/ + nws_last/) and all 9 pathological cases, then pinning the
confirmed-matching string as the literal -- not a blind snapshot of
current behavior.

quake/swpc/avalanche/hydro/incident/fire were checked and already carry
equivalent native-only golden coverage (added directly in ca751fb5), so no
changes were needed there.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-17 14:03:40 -06:00

122 lines
5.3 KiB
Python

"""Tests for the satpass persisted-timer `due_at` column.
Pending satellite-pass consolidations used to be scheduled only as in-memory
asyncio TimerHandles, so a restart orphaned any satpass_pending rows: the row
survived but its timer did not, and it was never consolidated/broadcast. The
fix persisted a durable `due_at` on each pending row and added a startup
sweep (`CentralConsumer._sweep_pending_satpass`) that reconstructed a timer
for every pending consolidated_id off its persisted due_at.
The Central NATS consumer (and `_sweep_pending_satpass` with it) was retired
2026-07 -- the sweep's tests are gone with it. The native satpass path
(env/satpass.py) never used the satpass_pending buffer or this sweep in the
first place (it consolidates in-memory within a single tick), so nothing
live is affected. What remains here:
- `due_at` is persisted on the normal ingest path (satpass_handler.py,
still live -- shared by both paths historically, now native-only)
- SCHEMA_VERSION == 26 and the v22 migration (which added the due_at
column) still applies cleanly on a fresh DB
"""
from __future__ import annotations
import json
import time
import pytest
from meshai.persistence import get_db, init_db, SCHEMA_VERSION
from meshai.adapter_config import invalidate_cache
# ── helpers ───────────────────────────────────────────────────────────
def _enable_satpass_db(norad_ids=(25544,), dry_run=True):
"""Enable satpass and set opt-in norad_ids in the test DB."""
conn = get_db()
conn.execute("UPDATE adapter_config SET value_json='true' "
"WHERE adapter='satpass' AND key='enabled'")
conn.execute("UPDATE adapter_config SET value_json=? "
"WHERE adapter='satpass' AND key='dry_run'",
(json.dumps(bool(dry_run)),))
conn.execute("UPDATE adapter_config SET value_json=? "
"WHERE adapter='satpass' AND key='norad_ids'",
(json.dumps(list(norad_ids)),))
invalidate_cache()
def _ingest_envelope(norad_id=25544, observer="Boise", max_el=72.5,
aos="2026-06-12T03:32:00Z", los="2026-06-12T03:38:00Z"):
return {
"specversion": "1.0",
"type": "central.sat.pass",
"source": "central",
"id": f"pass-{norad_id}-{aos}",
"data": {
"adapter": "n2yo_visualpasses",
"category": "pass.n2yo_visualpasses",
"data": {
"norad_id": norad_id,
"satellite_name": "ISS",
"observer_name": observer,
"max_elevation_deg": max_el,
"aos_time": aos,
"los_time": los,
"azimuth_at_peak_compass": "S",
"azimuth_at_aos_compass": "SW",
"azimuth_at_los_compass": "NE",
},
},
}
# ── schema / migration ───────────────────────────────────────────────
def test_schema_version_is_current():
# SCHEMA_VERSION is derived from the highest vN.sql in migrations/, so
# this just guards against the derivation returning something bogus
# (e.g. 0, which would mean the migrations dir wasn't found). The v22
# migration (this file's focus) must always be <= the current version.
assert SCHEMA_VERSION >= 22
def test_v22_migration_applies_and_adds_due_at_column(tmp_path, monkeypatch):
"""Fresh DB migrates cleanly and satpass_pending has due_at (v22 column)."""
from meshai.persistence import close_thread_connection
from meshai.persistence import db as persistence_db
db = str(tmp_path / "fresh-v22.sqlite")
monkeypatch.setenv("MESHAI_DB_PATH", db)
persistence_db._initialised.clear()
close_thread_connection()
conn = init_db()
row = conn.execute("SELECT value FROM schema_meta WHERE key='version'").fetchone()
assert int(row["value"]) == SCHEMA_VERSION
cols = {r["name"] for r in conn.execute("PRAGMA table_info(satpass_pending)")}
assert "due_at" in cols
close_thread_connection()
persistence_db._initialised.discard(db)
# ── due_at persisted on normal ingest ────────────────────────────────
def test_due_at_persisted_on_normal_ingest():
"""handle_satpass writes due_at = received_at + CONSOLIDATION_DELAY."""
from meshai.central.satpass_handler import (
handle_satpass, CONSOLIDATION_DELAY, _parse_iso_epoch)
_enable_satpass_db(norad_ids=[25544], dry_run=True)
for attr in ("_disabled_logged", "_no_norad_ids_logged"):
if hasattr(handle_satpass, attr):
delattr(handle_satpass, attr)
env = _ingest_envelope()
aos_epoch = _parse_iso_epoch("2026-06-12T03:32:00Z")
now = aos_epoch - 300 # inside horizon, before los
assert handle_satpass(env, "central.sat.pass.iss", now=now) is None
conn = get_db()
row = conn.execute(
"SELECT received_at, due_at FROM satpass_pending "
"WHERE norad_id=25544").fetchone()
assert row is not None, "ingest did not write a pending row"
assert row["due_at"] is not None
assert row["due_at"] == row["received_at"] + CONSOLIDATION_DELAY
assert row["due_at"] == now + CONSOLIDATION_DELAY