From e90d30accbcc02b6a0ef0308af952f7dcb06d7e2 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Fri, 17 Jul 2026 21:37:23 +0000 Subject: [PATCH] chore(central-ripout 2b): update satpass/tle test suite for the relocation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Repoints every remaining test import at the new env.satellite.* modules and removes/adapts coverage for the Central envelope-ingest path deleted in this pass (handle_satpass, consolidate_satpass_pending, handle_tle, and the filtering/coercion/staleness logic that lived only inside them): - test_satpass_native.py, test_tle_fetch.py, test_satpass_command.py: import-path updates only (pass_predictor, tle_store). Also dropped test_satpass_command.py's TestTLEUpsert.test_returns_none_always (handle_tle-specific contract, no longer applicable) and rewrote its two latest-wins tests to call upsert_tle directly — same behavior under test, now exercised through the still-live primitive instead of the dead wrapper. - test_satpass_native.py: deleted test_central_consolidate_feeds_shared_gate_merged (spied on consolidate_satpass_pending, which no longer exists). The merge-across-observers logic it guarded is native-side (_consolidate) and already covered by test_two_observers_consolidate_to_one_broadcast. - test_satpass_handler.py: gutted to the one test that calls format_pass directly (test_format_pass_worst_case_fits_140); the rest exercised handle_satpass's observer/norad/elevation filters, which have no live equivalent (the native adapter filters at the config level, not per-envelope) and is redundant with test_satpass_native.py's dedup/wire coverage via the real SatpassAdapter path. - test_satpass_broadcast_safety.py: kept every test that calls format_pass or gate_consolidated_pass-adjacent REGISTRY checks directly (wire format, clean-format rules, REGISTRY defaults); deleted TestNoradIdTypeCoercion and TestStalenessGuard (handle_satpass-only logic, no live equivalent) and the 6 _elevation_bucket tests (_elevation_bucket itself was dead before this pass too — zero callers anywhere but its own tests, already superseded by the numeric "max NN°" wire format per its own docstring). - test_satpass_persisted_timer.py: dropped test_due_at_persisted_on_normal_ingest (handle_satpass-only); kept the two schema/migration tests, which don't touch satpass_handler. - Deleted outright (tested ONLY the dead Central envelope-ingest path, no live equivalent to port to): test_satpass_event_path.py, test_satpass_compass_fallback.py, test_satpass_wire_fields.py. Full suite: 2059 passed, 0 failed (was 0 failed on main pre-change). Satpass/TLE subset (99 tests across 8 files) verified green in isolation. Co-Authored-By: Claude Opus 4.8 (1M context) --- work/tests/test_satpass_broadcast_safety.py | 497 ++------------------ work/tests/test_satpass_command.py | 88 ++-- work/tests/test_satpass_handler.py | 207 +------- work/tests/test_satpass_native.py | 72 +-- work/tests/test_satpass_persisted_timer.py | 86 +--- work/tests/test_tle_fetch.py | 2 +- 6 files changed, 99 insertions(+), 853 deletions(-) diff --git a/work/tests/test_satpass_broadcast_safety.py b/work/tests/test_satpass_broadcast_safety.py index 45a0bd7..07eb828 100644 --- a/work/tests/test_satpass_broadcast_safety.py +++ b/work/tests/test_satpass_broadcast_safety.py @@ -1,174 +1,41 @@ """Tests for satpass broadcast safety controls. -Covers all five incident-response requirements: - 1. Opt-in bird filter: empty norad_ids broadcasts nothing - 2. Rate cap: max_broadcasts_per_hour suppresses excess - 3. Dry-run mode: logs wire text, dispatches nothing +Covers the incident-response requirements that live in +`meshai.env.satellite.pass_format` (formatting rules) and REGISTRY defaults: + 3. Dry-run default: REGISTRY dry_run default is True 4. Elevation default: REGISTRY min_elevation = 30 - 5. Broadcast wire format: two-line, buckets, byte budget + 5. Broadcast wire format: single-line, numeric degrees, byte budget + 6. Clean format: short names, degrees, compass collapse, friendly observers + +The Central envelope-ingest path (`handle_satpass`, +`consolidate_satpass_pending`, and the opt-in-bird-filter / rate-cap / +dry-run / staleness-guard / norad-id-coercion logic that lived INSIDE +`handle_satpass`) was retired with the Central NATS consumer and deleted +2026-07. That logic has no live equivalent in the native path -- the native +SatpassAdapter (env/satpass.py) does its own norad/observer filtering at the +adapter level (config-driven, not per-envelope) and its own imminence gate +instead of a staleness guard; both are covered by tests/test_satpass_native.py. +Dedup, rate-cap, and dry-run behavior of the shared +`gate_consolidated_pass` gate are also exercised live via the native adapter +in tests/test_satpass_native.py (`test_second_tick_does_not_rebroadcast_after_commit`, +imminence tests, etc). Tests that only exercised the dead ingest path's +filters (opt-in bird filter mechanics, rate-cap-via-handle_satpass, +dry-run-via-handle_satpass, norad-id string/int coercion, staleness guard) +were deleted rather than ported, since their behavior no longer exists +anywhere to test. """ from __future__ import annotations -import json -import logging -import time -from datetime import datetime, timezone -from unittest.mock import MagicMock, patch +from datetime import datetime import pytest -# ── Helpers ────────────────────────────────────────────────────────── - -def _envelope(norad_id=25544, sat_name="ISS", observer="Boise", - max_el=75.0, aos="2026-06-12T03:32:00Z", - los="2026-06-12T03:38:00Z", - aos_compass="SW", los_compass="NE"): - """Build a CloudEvents envelope for a satellite pass.""" - return { - "specversion": "1.0", - "type": "central.sat.pass", - "source": "central", - "id": f"pass-{norad_id}-{aos}", - "data": { - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "severity": 0, - "data": { - "norad_id": norad_id, - "satellite_name": sat_name, - "observer_name": observer, - "max_elevation_deg": max_el, - "aos_time": aos, - "los_time": los, - "azimuth_at_peak_compass": "S", - "azimuth_at_aos_compass": aos_compass, - "azimuth_at_los_compass": los_compass, - "duration_s": 360, - } - } - } - - -def _enable_satpass_db(norad_ids=None, dry_run=False, max_per_hour=100): - """Set satpass config in the test DB.""" - from meshai.persistence import get_db - from meshai.adapter_config import invalidate_cache - 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(dry_run),) - ) - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='max_broadcasts_per_hour'", - (json.dumps(max_per_hour),) - ) - if norad_ids is not None: - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='norad_ids'", - (json.dumps(norad_ids),) - ) - conn.execute( - "UPDATE adapter_config SET value_json='5' " - "WHERE adapter='satpass' AND key='min_elevation'" - ) - invalidate_cache() - - -def _clear_handler_flags(): - from meshai.central.satpass_handler import handle_satpass - for attr in ("_disabled_logged", "_no_norad_ids_logged"): - if hasattr(handle_satpass, attr): - delattr(handle_satpass, attr) - - -def _ingest_and_consolidate(env, subject, *, now, data=None): - """Drive the two-call async satpass contract. - - handle_satpass() ingests the pass into satpass_pending and ALWAYS - returns None (the immediate-broadcast suppression); the consumer then - fires consolidate_satpass_pending() on its 5s timer, which returns the - (wire_string, data_dict) to actually broadcast, or None if suppressed. - This helper runs both halves and returns the consolidation result so a - test can assert on the real broadcast decision + wire. - """ - from meshai.central.satpass_handler import ( - handle_satpass, consolidate_satpass_pending, - drain_pending_consolidation_ids) - drain_pending_consolidation_ids() # clear any cross-test leakage - ingest = handle_satpass( - env, subject, data=data if data is not None else {}, now=now) - assert ingest is None, "handle_satpass must suppress the immediate broadcast" - for cid in drain_pending_consolidation_ids(): - res = consolidate_satpass_pending(cid) - if res is not None: - return res - return None - - # ══════════════════════════════════════════════════════════════════════ -# 1. OPT-IN BIRD FILTER +# 1. OPT-IN BIRD FILTER — REGISTRY-only surviving check # ══════════════════════════════════════════════════════════════════════ class TestOptInBirdFilter: - """empty norad_ids = broadcast nothing; norad_ids=[25544] = ISS only.""" - - def test_empty_norad_ids_broadcasts_nothing(self): - """norad_ids=[] must broadcast nothing.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[], dry_run=False) - _clear_handler_flags() - - env = _envelope(norad_id=25544, max_el=80.0) - result = handle_satpass(env, "central.sat.pass.iss", data={}, - now=1718163120) - assert result is None - - def test_empty_norad_ids_logs_once(self, caplog): - """Empty norad_ids logs info message once.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[], dry_run=False) - _clear_handler_flags() - - # now near the 2026 envelope window so the AOS-horizon guard does - # not short-circuit before the opt-in norad_ids check that logs. - with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"): - handle_satpass(_envelope(norad_id=25544, max_el=80.0), - "central.sat.pass.iss", data={}, now=1781235000) - handle_satpass(_envelope(norad_id=25544, max_el=80.0, - aos="2026-06-12T04:32:00Z"), - "central.sat.pass.iss", data={}, now=1781235000) - - matching = [r for r in caplog.records - if "no norad_ids configured" in r.message] - assert len(matching) == 1, "Should log exactly once" - - def test_norad_ids_25544_airs_iss_only(self): - """norad_ids=[25544] must air ISS passes, reject others.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=False) - _clear_handler_flags() - - # ISS pass should air - iss_env = _envelope(norad_id=25544, max_el=65.0) - iss_result = _ingest_and_consolidate(iss_env, "central.sat.pass.iss", - now=1781235000) - assert iss_result is not None, "ISS pass should broadcast" - - # NOAA-18 pass should be rejected (opt-in norad_ids excludes it) - noaa_env = _envelope(norad_id=28654, sat_name="NOAA 18", max_el=65.0, - aos="2026-06-12T05:00:00Z") - noaa_result = _ingest_and_consolidate(noaa_env, "central.sat.pass.noaa", - now=1781235000) - assert noaa_result is None, "NOAA-18 pass should be rejected" - def test_dm_command_not_gated_by_norad_ids(self): """!satpass DM replies about any bird in TLE cache, regardless of broadcast norad_ids being empty.""" @@ -183,66 +50,13 @@ class TestOptInBirdFilter: # ══════════════════════════════════════════════════════════════════════ -# 2. RATE CAP +# 2. RATE CAP — DM-path isolation check # ══════════════════════════════════════════════════════════════════════ class TestRateCap: - """max_broadcasts_per_hour suppresses excess.""" - - def test_cap_suppresses_pass_n_plus_1(self): - """After max_broadcasts_per_hour broadcasts, next one is suppressed.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - - _enable_satpass_db(norad_ids=[25544, 28654, 99999, 99998, 99997], - dry_run=False, max_per_hour=4) - _clear_handler_flags() - - now = 1718200000 - hour_start = (now // 3600) * 3600 - conn = get_db() - - # Pre-insert 4 broadcast records in current hour window - for i in range(4): - eid = f"prefill:{i}:dummy:{hour_start // 3600}" - conn.execute( - "INSERT INTO satpass_events(event_id, norad_id, sat_name, " - "observer, max_elevation, aos_at, los_at, first_seen_at, " - "last_broadcast_at) VALUES (?,?,?,?,?,?,?,?,?)", - (eid, 99990 + i, f"SAT-{i}", "Boise", 60.0, - now - 300, now + 300, now - 600, hour_start + i) - ) - - # 5th broadcast should be suppressed - env = _envelope(norad_id=25544, max_el=70.0, - aos="2026-06-12T10:00:00Z") - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=now) - assert result is None, "5th broadcast should be suppressed by rate cap" - - def test_cap_logged_on_suppress(self, caplog): - """Rate cap suppression must log at INFO.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - - _enable_satpass_db(norad_ids=[25544], dry_run=False, max_per_hour=0) - _clear_handler_flags() - - # now near the 2026 envelope window; max_per_hour=0 => the - # consolidation step always trips the rate cap and logs. - now = 1781258400 # 2026-06-12T10:00:00Z (just before this pass) - with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"): - env = _envelope(norad_id=25544, max_el=70.0, - aos="2026-06-12T10:05:00Z", - los="2026-06-12T10:11:00Z") - result = _ingest_and_consolidate(env, "central.sat.pass.iss", now=now) - assert result is None, "rate cap should suppress the broadcast" - - matching = [r for r in caplog.records if "rate cap reached" in r.message] - assert len(matching) >= 1, "Rate cap suppression should be logged" - def test_cap_does_not_apply_to_dm_path(self): """Rate cap is broadcast-only, never affects DM replies.""" - # The DM command (satpass_cmd.py) does not call handle_satpass, + # The DM command (satpass_cmd.py) does not call gate_consolidated_pass, # it uses pass_predictor directly. Verify they're separate paths. import inspect from meshai.commands import satpass_cmd @@ -252,59 +66,10 @@ class TestRateCap: # ══════════════════════════════════════════════════════════════════════ -# 3. DRY-RUN MODE +# 3. DRY-RUN MODE — REGISTRY default # ══════════════════════════════════════════════════════════════════════ class TestDryRun: - """dry_run=True logs wire text, dispatches nothing.""" - - def test_dry_run_dispatches_nothing(self): - """dry_run=True must return None (no dispatch).""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=True) - _clear_handler_flags() - - data = {} - env = _envelope(norad_id=25544, max_el=70.0) - result = handle_satpass(env, "central.sat.pass.iss", data=data, - now=1718163120) - assert result is None, "dry_run should suppress dispatch" - assert "_on_broadcast_committed" not in data, \ - "dry_run should not attach commit callback" - - def test_dry_run_logs_wire_text(self, caplog): - """dry_run=True must log the exact wire text at INFO.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=True) - _clear_handler_flags() - - with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"): - env = _envelope(norad_id=25544, sat_name="ISS", max_el=70.0) - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is None, "dry_run should suppress dispatch" - - # dry-run logging now happens in the consolidation step; the message - # is "DRY-RUN would air (consolidated, N observers): ". - matching = [r for r in caplog.records - if r.message.startswith("DRY-RUN would air")] - assert len(matching) == 1, "Should log DRY-RUN wire text once" - assert "ISS" in matching[0].message - - def test_dry_run_false_dispatches(self): - """dry_run=False must dispatch normally.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=False) - _clear_handler_flags() - - env = _envelope(norad_id=25544, max_el=70.0) - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is not None, "dry_run=False should dispatch" - # commit callback now rides on the consolidation result's data dict. - wire, data = result - assert "_on_broadcast_committed" in data - def test_dry_run_default_is_true(self): """REGISTRY default for dry_run must be True.""" from meshai.adapter_config.defaults import REGISTRY @@ -331,11 +96,11 @@ class TestElevationDefault: # ══════════════════════════════════════════════════════════════════════ class TestBroadcastWireFormat: - """Two-line format, buckets, byte budget.""" + """Single-line format, numeric degrees, byte budget.""" def test_exact_single_line_example(self): """Formatter produces the exact single-line target format.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass # ISS, SW->NE, 6-minute window, rises 8:38 PM MDT, max 55 deg. from zoneinfo import ZoneInfo @@ -353,46 +118,16 @@ class TestBroadcastWireFormat: ) # Single line: absolute local rise time, numeric max elevation, and a - # date qualifier ("Fri Jun 12" \u2014 the fixed date is always in the past). + # date qualifier ("Fri Jun 12" — the fixed date is always in the past). assert "\n" not in wire assert wire == ( - "\U0001F6F0\uFE0F ISS 8:38 PM MDT Fri Jun 12, " - "max 55\u00B0 SW\u2192NE (6 min)" + "\U0001F6F0️ ISS 8:38 PM MDT Fri Jun 12, " + "max 55° SW→NE (6 min)" ) - def test_bucket_overhead_at_60(self): - """max_el=60 should be 'overhead'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(60.0) == "overhead" - - def test_bucket_overhead_at_90(self): - """max_el=90 should be 'overhead'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(90.0) == "overhead" - - def test_bucket_high_pass_at_59(self): - """max_el=59 should be 'high pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(59.0) == "high pass" - - def test_bucket_high_pass_at_30(self): - """max_el=30 should be 'high pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(30.0) == "high pass" - - def test_bucket_low_pass_at_29(self): - """max_el=29 should be 'low pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(29.0) == "low pass" - - def test_bucket_low_pass_at_10(self): - """max_el=10 should be 'low pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(10.0) == "low pass" - def test_broadcast_byte_length_under_budget(self): """Broadcast wire message must be <= 120 bytes UTF-8.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") @@ -423,7 +158,7 @@ class TestBroadcastWireFormat: def test_dm_format_has_exact_degrees(self): """DM format must include exact degree number, not bucket.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") @@ -438,13 +173,13 @@ class TestBroadcastWireFormat: broadcast=False, ) - assert "max 75\u00B0" in wire + assert "max 75°" in wire assert "overhead" not in wire assert "high pass" not in wire def test_dm_format_single_line(self): """DM format is a single line.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") @@ -488,164 +223,6 @@ class TestRegistryKeys: "opt-in" in spec["description"].lower() -class TestNoradIdTypeCoercion: - """norad_ids may arrive as strings from the GUI or ints from code. - The handler must accept both shapes forever.""" - - def test_string_norad_ids_matches_int_wire(self): - """norad_ids=["25544"] must match wire norad_id 25544 (int).""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["25544"], dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - result = _ingest_and_consolidate(env, "test.subject", now=1781235000) - assert result is not None, "string norad_id should match int wire" - - def test_mixed_int_and_string_norad_ids(self): - """norad_ids=[25544, "22825"] must match both NORAD IDs.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544, "22825"], dry_run=False) - - # int in list, int on wire - env_iss = _envelope(norad_id=25544, max_el=65.0) - result_iss = _ingest_and_consolidate(env_iss, "test.subject", now=1781235000) - assert result_iss is not None, "int norad_id in mixed list should match" - - # string in list, int on wire - env_noaa = _envelope(norad_id=22825, sat_name="NOAA 15", max_el=65.0, - aos="2026-06-12T05:32:00Z", los="2026-06-12T05:38:00Z") - result_noaa = _ingest_and_consolidate(env_noaa, "test.subject", now=1781235000) - assert result_noaa is not None, "string norad_id in mixed list should match int wire" - - def test_garbage_entries_skipped_without_crash(self): - """Non-numeric entries in norad_ids must be silently skipped.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["25544", "not_a_number", "", None, "abc123"], - dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - # Must not raise, and the valid entry should still match - result = _ingest_and_consolidate(env, "test.subject", now=1781235000) - assert result is not None, "valid entry should match despite garbage siblings" - - def test_all_garbage_norad_ids_matches_nothing(self): - """If every entry is garbage, allow_set is empty and nothing matches.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["abc", "", "xyz"], dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - result = handle_satpass(env, "test.subject", data={}, now=1781235000) - assert result is None, "all-garbage norad_ids should match nothing" - - def test_pure_int_norad_ids_still_works(self): - """norad_ids=[25544] (pure int) must continue to work.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - result = _ingest_and_consolidate(env, "test.subject", now=1781235000) - assert result is not None, "pure int norad_id should still match" - - def test_string_norad_id_rejects_non_matching(self): - """norad_ids=["25544"] must NOT match wire norad_id 99999.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["25544"], dry_run=False) - - env = _envelope(norad_id=99999, max_el=80.0) - result = handle_satpass(env, "test.subject", data={}, now=1781235000) - assert result is None, "non-matching norad_id should be rejected" - - -class TestStalenessGuard: - """Reject passes whose window already ended; allow ongoing/future/None.""" - - def test_past_pass_rejected(self): - """Pass with los 10 min in the past produces no wire, no broadcast mark.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos = "2026-06-12T02:00:00Z" # well in the past - los_epoch = now - 600 # 10 min ago - los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - env = _envelope(norad_id=25544, max_el=80.0, aos=aos, los=los) - result = handle_satpass(env, "test.subject", data={}, now=now) - assert result is None, "past pass should produce no wire" - - # Verify no broadcast mark in DB - conn = get_db() - row = conn.execute( - "SELECT last_broadcast_at FROM satpass_events WHERE norad_id=25544" - ).fetchone() - assert row is None or row["last_broadcast_at"] is None, \ - "past pass should not create a broadcast-marked DB row" - - def test_ongoing_pass_broadcasts(self): - """Ongoing pass (aos -2 min, los +5 min) must produce wire.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos_epoch = now - 120 # started 2 min ago - los_epoch = now + 300 # ends in 5 min - aos = datetime.fromtimestamp(aos_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - env = _envelope(norad_id=25544, max_el=70.0, aos=aos, los=los) - result = _ingest_and_consolidate(env, "test.subject", now=now) - assert result is not None, "ongoing pass (los in future) should broadcast" - - def test_future_pass_broadcasts(self): - """Future pass (aos and los both in future) must produce wire.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos_epoch = now + 600 # starts in 10 min - los_epoch = now + 1200 # ends in 20 min - aos = datetime.fromtimestamp(aos_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - env = _envelope(norad_id=25544, max_el=55.0, aos=aos, los=los) - result = _ingest_and_consolidate(env, "test.subject", now=now) - assert result is not None, "future pass should broadcast" - - def test_none_los_falls_through(self): - """los_epoch=None must not be rejected by staleness guard.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos_epoch = now + 600 - aos = datetime.fromtimestamp(aos_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - # Build envelope with no los_time - env = _envelope(norad_id=25544, max_el=60.0, aos=aos, los="") - # Patch los to None by removing los_time from inner data - env["data"]["data"]["los_time"] = None - - result = _ingest_and_consolidate(env, "test.subject", now=now) - # Should not be rejected by staleness guard — falls through to - # normal handling (wire produced or other filter applies) - # We just verify it does NOT crash and is not rejected as stale - # It may still produce wire or be filtered by something else, - # but the staleness guard specifically must not block it. - # Since all other filters pass, wire should be produced. - assert result is not None, "None los_epoch should fall through staleness guard" - - # ══════════════════════════════════════════════════════════════════════ # 6. CLEAN FORMAT: short names, degrees, compass collapse, friendly obs # ══════════════════════════════════════════════════════════════════════ @@ -655,7 +232,7 @@ class TestCleanBroadcastFormat: @staticmethod def _wire(**kw): - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") base = dict( diff --git a/work/tests/test_satpass_command.py b/work/tests/test_satpass_command.py index 4bac7ce..c17f46f 100644 --- a/work/tests/test_satpass_command.py +++ b/work/tests/test_satpass_command.py @@ -67,11 +67,17 @@ def _seed_tle(conn, *, norad_id, name, line1, line2, epoch, updated_at=None): class TestTLEUpsert: - """T1: TLE upsert latest-wins on epoch.""" + """T1: TLE upsert latest-wins on epoch. + + Exercises `upsert_tle` directly — the shared latest-wins primitive used + by every writer of `sat_tles` (native env.tle_fetch is the only one + left; the Central envelope ingest path that used to call it via + `tle_handler.handle_tle` was retired with Central). + """ def test_newer_epoch_updates(self): _enable_satpass() - from meshai.central.tle_handler import handle_tle + from meshai.env.satellite.tle_store import upsert_tle conn = get_db() now = int(time.time()) @@ -79,20 +85,9 @@ class TestTLEUpsert: _seed_tle(conn, norad_id=25544, name="ISS", line1="OLD1", line2="OLD2", epoch="2024-06-10T00:00:00Z") - # Send newer TLE - env = { - "data": { - "adapter": "celestrak_tle", - "data": { - "norad_id": 25544, - "satellite_name": "ISS (ZARYA)", - "tle_line1": "NEW1", - "tle_line2": "NEW2", - "epoch": "2024-06-15T00:00:00Z", - }, - } - } - handle_tle(env, "central.sat.tle.25544", now=now) + # Upsert a newer TLE + upsert_tle(conn, 25544, "ISS (ZARYA)", "NEW1", "NEW2", + "2024-06-15T00:00:00Z", now=now) row = conn.execute("SELECT line1, line2 FROM sat_tles WHERE norad_id=25544").fetchone() assert row["line1"] == "NEW1" @@ -100,7 +95,7 @@ class TestTLEUpsert: def test_older_epoch_skipped(self): _enable_satpass() - from meshai.central.tle_handler import handle_tle + from meshai.env.satellite.tle_store import upsert_tle conn = get_db() now = int(time.time()) @@ -108,49 +103,20 @@ class TestTLEUpsert: _seed_tle(conn, norad_id=25544, name="ISS", line1="CURRENT1", line2="CURRENT2", epoch="2024-06-15T00:00:00Z") - # Send older TLE — should be skipped - env = { - "data": { - "adapter": "celestrak_tle", - "data": { - "norad_id": 25544, - "satellite_name": "ISS (ZARYA)", - "tle_line1": "OLD1", - "tle_line2": "OLD2", - "epoch": "2024-06-10T00:00:00Z", - }, - } - } - handle_tle(env, "central.sat.tle.25544", now=now) + # Upsert an older TLE — should be skipped + written = upsert_tle(conn, 25544, "ISS (ZARYA)", "OLD1", "OLD2", + "2024-06-10T00:00:00Z", now=now) + assert written is False row = conn.execute("SELECT line1 FROM sat_tles WHERE norad_id=25544").fetchone() assert row["line1"] == "CURRENT1", "older epoch should not overwrite" - def test_returns_none_always(self): - """TLE handler is storage-only, never returns wire.""" - _enable_satpass() - from meshai.central.tle_handler import handle_tle - env = { - "data": { - "adapter": "celestrak_tle", - "data": { - "norad_id": 99999, - "satellite_name": "TEST", - "tle_line1": "L1", - "tle_line2": "L2", - "epoch": "2024-06-15T00:00:00Z", - }, - } - } - result = handle_tle(env, "central.sat.tle.99999") - assert result is None - class TestTLEStaleness: """T2: 14-day staleness exclusion at read time.""" def test_fresh_tle_returned(self): - from meshai.central.tle_handler import get_tle_by_norad + from meshai.env.satellite.tle_store import get_tle_by_norad conn = get_db() # Seed with recent epoch recent = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat() @@ -161,7 +127,7 @@ class TestTLEStaleness: assert tle["norad_id"] == 25544 def test_stale_tle_excluded(self): - from meshai.central.tle_handler import get_tle_by_norad + from meshai.env.satellite.tle_store import get_tle_by_norad conn = get_db() # Seed with 15-day old epoch stale = (datetime.now(timezone.utc) - timedelta(days=15)).isoformat() @@ -171,7 +137,7 @@ class TestTLEStaleness: assert tle is None, "stale TLE (>14 days) should be excluded" def test_search_excludes_stale(self): - from meshai.central.tle_handler import search_tle_by_name + from meshai.env.satellite.tle_store import search_tle_by_name conn = get_db() stale = (datetime.now(timezone.utc) - timedelta(days=15)).isoformat() _seed_tle(conn, norad_id=25544, name="ISS (ZARYA)", line1=ISS_LINE1, @@ -185,7 +151,7 @@ class TestPassPredictor: def test_iss_produces_passes(self): """ISS TLE for Boise should produce at least one pass in 24h.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes # Use a fixed time near the TLE epoch for best accuracy start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, @@ -194,7 +160,7 @@ class TestPassPredictor: def test_pass_max_elevation_reasonable(self): """Max elevation should be between min_el and 90°.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -204,7 +170,7 @@ class TestPassPredictor: def test_pass_aos_before_los(self): """AOS should be before LOS for every pass.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -214,7 +180,7 @@ class TestPassPredictor: def test_pass_duration_reasonable(self): """Pass durations should be positive; 30s step may merge adjacent passes.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -232,7 +198,7 @@ class TestPassPredictor: over Boise (43.6°N, 51.6° inclination orbit). We assert that at least one pass in 24h exceeds 30° — a conservative threshold. """ - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -243,7 +209,7 @@ class TestPassPredictor: def test_azimuth_range(self): """Azimuths should be in [0, 360) range.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -252,7 +218,7 @@ class TestPassPredictor: assert 0 <= p.azimuth_at_los < 360, f"LOS azimuth {p.azimuth_at_los} out of range" def test_compass_conversion(self): - from meshai.central.pass_predictor import azimuth_to_compass + from meshai.env.satellite.pass_predictor import azimuth_to_compass assert azimuth_to_compass(0) == "N" assert azimuth_to_compass(45) == "NE" assert azimuth_to_compass(90) == "E" @@ -442,7 +408,7 @@ class TestReplyFormat: def test_line_format_matches_spec(self): """Lines should match 'NAME HH:MM–HH:MM TZ max XX° DIR→DIR'.""" - from meshai.central.pass_predictor import compute_passes, azimuth_to_compass, PassInfo + from meshai.env.satellite.pass_predictor import compute_passes, azimuth_to_compass, PassInfo from meshai.commands.satpass_cmd import SatpassCommand from zoneinfo import ZoneInfo diff --git a/work/tests/test_satpass_handler.py b/work/tests/test_satpass_handler.py index 104c8e9..7d30233 100644 --- a/work/tests/test_satpass_handler.py +++ b/work/tests/test_satpass_handler.py @@ -1,204 +1,29 @@ -"""v0.7 satpass_handler tests.""" +"""v0.7 satpass_handler tests. -import pytest -from unittest.mock import MagicMock, patch +The Central envelope-ingest path (`handle_satpass`, +`consolidate_satpass_pending`, and the observer/norad/elevation/staleness +filtering that lived inside them) was retired with the Central NATS +consumer and deleted 2026-07 when the still-live wire-formatting + gate +code (`format_pass`, `gate_consolidated_pass`) moved to +`meshai.env.satellite.pass_format`. That live code's coverage now lives in +tests/test_satpass_native.py (dedup, wire format, commit callback, via the +native SatpassAdapter path) and tests/test_satpass_broadcast_safety.py +(format_pass formatting rules called directly). What remains here is the +one test that calls `format_pass` directly and doesn't fit either of those +files' focus. +""" - -def _envelope(norad_id=25544, sat_name="ISS", observer="Boise", - max_el=75.0, aos="2026-06-12T03:32:00Z", - los="2026-06-12T03:38:00Z", direction="NW-SE", - aos_compass="SW", los_compass="NE"): - """Build a CloudEvents envelope for a satellite pass.""" - return { - "specversion": "1.0", - "type": "central.sat.pass", - "source": "central", - "id": f"pass-{norad_id}-{aos}", - "data": { - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "severity": 0, - "data": { - "norad_id": norad_id, - "satellite_name": sat_name, - "observer_name": observer, - "max_elevation_deg": max_el, - "aos_time": aos, - "los_time": los, - "azimuth_at_peak_compass": direction, - "azimuth_at_aos_compass": aos_compass, - "azimuth_at_los_compass": los_compass, - } - } - } - - -@pytest.fixture -def mock_db(): - """Mock database connection.""" - conn = MagicMock() - conn.execute.return_value.fetchone.return_value = None - conn.execute.return_value.lastrowid = 1 - with patch("meshai.central.satpass_handler.get_db", return_value=conn): - yield conn - - -@pytest.fixture -def mock_adapter_config(): - """Mock adapter_config.satpass.""" - cfg = MagicMock() - cfg.enabled = True - cfg.observers = [] # empty = all observers - cfg.min_elevation = 30 - cfg.norad_ids = [25544] # must be non-empty for opt-in - cfg.dry_run = False - cfg.max_broadcasts_per_hour = 4 - cfg.max_aos_horizon_hours = 24 - with patch("meshai.central.satpass_handler.adapter_config") as mock: - mock.satpass = cfg - from meshai.central.satpass_handler import handle_satpass - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - yield cfg - - -def _ingest_and_consolidate(env, subject, *, now, data=None): - """Drive the two-call async satpass contract (ingest -> consolidate). - - handle_satpass() ingests the pass into satpass_pending and ALWAYS - returns None; the consumer then runs consolidate_satpass_pending(), - which returns the (wire, data) to broadcast (or None if suppressed). - Returns the consolidation result so a test can assert on the real wire. - """ - from meshai.central.satpass_handler import ( - handle_satpass, consolidate_satpass_pending, - drain_pending_consolidation_ids) - drain_pending_consolidation_ids() - assert handle_satpass( - env, subject, data=data if data is not None else {}, now=now) is None - for cid in drain_pending_consolidation_ids(): - res = consolidate_satpass_pending(cid) - if res is not None: - return res - return None - - -class TestSatpassHandler: - """Tests for handle_satpass function.""" - - def test_high_elevation_pass_broadcasts(self, mock_adapter_config): - """A pass with high elevation should broadcast (via consolidation).""" - env = _envelope(max_el=75.0) - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is not None - wire, _ = result - assert "ISS" in wire - - def test_low_elevation_pass_filtered(self, mock_db, mock_adapter_config): - """A pass below min_elevation should be filtered.""" - from meshai.central.satpass_handler import handle_satpass - - mock_adapter_config.min_elevation = 30 - env = _envelope(max_el=25.0) - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - - assert result is None - - def test_observer_filter_blocks_mismatch(self, mock_db, mock_adapter_config): - """A pass for non-configured observer should be filtered.""" - from meshai.central.satpass_handler import handle_satpass - - mock_adapter_config.observers = ["Magic Valley"] - env = _envelope(observer="Boise") - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - - assert result is None - - def test_observer_filter_allows_match(self, mock_adapter_config): - """A pass for configured observer should broadcast.""" - mock_adapter_config.observers = ["Boise", "Magic Valley"] - env = _envelope(observer="Boise", max_el=45.0) - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is not None - - def test_norad_id_filter(self, mock_db, mock_adapter_config): - """NORAD ID filter should block non-matching satellites.""" - from meshai.central.satpass_handler import handle_satpass - - mock_adapter_config.norad_ids = [25544] # ISS only - env = _envelope(norad_id=12345, max_el=60.0) - result = handle_satpass(env, "central.sat.pass.other", data={}, now=1718163120) - - assert result is None - - def test_dedup_blocks_second_broadcast(self, mock_adapter_config): - """Second pass in same hour bucket should be deduplicated.""" - env = _envelope(max_el=60.0) - - # First pass consolidates into a real broadcast. - result1 = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result1 is not None - - # Simulate the broadcast committing (marks satpass_events broadcast). - _wire, data1 = result1 - data1["_on_broadcast_committed"](1781235010) - - # Second pass in the same (norad, aos-hour) bucket must be deduped. - result2 = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235060) - assert result2 is None - - def test_wire_format(self, mock_adapter_config): - """Wire is a single clean line: short name, degrees, collapsed compass.""" - env = _envelope(sat_name="ISS", max_el=75, observer="Boise", - direction="S", aos_compass="SW", los_compass="NE") - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is not None - wire, _ = result - - assert "\n" not in wire # single line now - assert "ISS" in wire - assert "max 75°" in wire # numeric elevation, not a bucket word - assert "overhead" not in wire - assert "min window" not in wire - assert "SW→S→NE" in wire # aos -> peak -> los, collapsed - - def test_commit_callback_attached(self, mock_adapter_config): - """Broadcast should attach commit callback (on the consolidation data).""" - env = _envelope(max_el=60.0) - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is not None - _wire, data = result - assert "_on_broadcast_committed" in data - assert "_broadcast_audit" in data - assert data["_broadcast_audit"]["table"] == "satpass_events" - - def test_wrong_adapter_ignored(self, mock_db, mock_adapter_config): - """Envelope with wrong adapter should be ignored.""" - from meshai.central.satpass_handler import handle_satpass - - env = _envelope() - env["data"]["adapter"] = "some_other_adapter" - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - - assert result is None +from __future__ import annotations # ============================================================================ # Budget-fit SAFETY CAP: a pathologically long satellite name must not push # the broadcast string past 140 chars. Calls format_pass directly (bypasses -# the broken consolidation path). +# the consolidation path entirely). # ============================================================================ def test_format_pass_worst_case_fits_140(): - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass wire = format_pass( sat_name=("NOAA-19 EXPERIMENTAL SUPER LONG SATELLITE DESIGNATION " "PAYLOAD REVISION X PROTOTYPE FLIGHT MODEL SERIAL 00042"), diff --git a/work/tests/test_satpass_native.py b/work/tests/test_satpass_native.py index 72f6e2d..8a35ef1 100644 --- a/work/tests/test_satpass_native.py +++ b/work/tests/test_satpass_native.py @@ -1,13 +1,12 @@ """Tests for the native SGP4 satpass adapter (env.satpass). The native adapter computes every observer for a satellite in ONE tick, so it -consolidates in-memory and gates synchronously via the SHARED -`satpass_handler.gate_consolidated_pass` — with NO `satpass_pending` buffer and -NO Central consumer/timer. These tests monkeypatch `compute_passes` and +consolidates in-memory and gates synchronously via +`env.satellite.pass_format.gate_consolidated_pass` — with no buffer table and +no consumer/timer. These tests monkeypatch `compute_passes` and `get_observers` (no SGP4 / no network) and seed a fresh `sat_tles` row, then exercise: multi-observer consolidation, cross-tick dedup via `satpass_events`, -resilient empty cases, the precomposed aos→peak→los wire, and a guard that the -Central path still feeds the shared gate the correctly-merged consolidation. +resilient empty cases, and the precomposed aos→peak→los wire. """ from __future__ import annotations @@ -18,8 +17,8 @@ import pytest from meshai.env.satpass import SatpassAdapter from meshai.config import SatpassConfig -from meshai.central.pass_predictor import PassInfo -from meshai.central.tle_handler import upsert_tle +from meshai.env.satellite.pass_predictor import PassInfo +from meshai.env.satellite.tle_store import upsert_tle from meshai.persistence import get_db @@ -85,7 +84,7 @@ def _adapter(**overrides) -> SatpassAdapter: def _patch_predictor(monkeypatch, fake): - monkeypatch.setattr("meshai.central.pass_predictor.compute_passes", fake) + monkeypatch.setattr("meshai.env.satellite.pass_predictor.compute_passes", fake) def _patch_observers(monkeypatch, observers): @@ -253,63 +252,6 @@ def test_emitted_event_renders_aos_peak_los(monkeypatch): assert composed.startswith("\U0001F6F0") # satellite emoji -# ══════════════════════════════════════════════════════════════════════ -# 5. SHARED-GATE EXTRACTION GUARD (Central path still feeds the gate right) -# ══════════════════════════════════════════════════════════════════════ - -def test_central_consolidate_feeds_shared_gate_merged(monkeypatch): - """consolidate_satpass_pending must merge observers and hand the SAME - shared gate a correctly-consolidated dict (earliest AOS / latest LOS / - max-el observer / entry+exit / observer_list). Spying the gate proves the - Central path routes through the extracted, source-agnostic function.""" - import meshai.central.satpass_handler as sh - - conn = get_db() - cid = f"25544:{T0 // 3600}" - # Two pending rows for the same canonical id (boise entry, twin exit+peak). - rows = [ - (cid, "boise", "ISS", 25544, 40.0, T0, T0 + 300, "SW", "NW", "E", T0, T0 + 5), - (cid, "twin", "ISS", 25544, 70.0, T0 + 60, T0 + 400, "W", "NE", "S", T0, T0 + 5), - ] - for r in rows: - conn.execute( - "INSERT OR REPLACE INTO satpass_pending(" - "consolidated_id, observer, sat_name, norad_id, max_elevation, " - "aos_at, los_at, aos_compass, los_compass, peak_compass, " - "received_at, due_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", r) - - captured = {} - - def _spy(consolidated, *, now): - captured["c"] = consolidated - captured["now"] = now - return None # short-circuit: no broadcast side effects - - monkeypatch.setattr(sh, "gate_consolidated_pass", _spy) - - result = sh.consolidate_satpass_pending(cid) - assert result is None - - c = captured["c"] - assert c["consolidated_id"] == cid - assert c["norad_id"] == 25544 - assert c["max_elevation"] == 70.0 # twin (higher) - assert c["aos_epoch"] == T0 # boise earliest AOS - assert c["los_epoch"] == T0 + 400 # twin latest LOS - assert c["aos_compass"] == "SW" # boise - assert c["los_compass"] == "NE" # twin - assert c["peak_compass"] == "S" # twin (max-el observer) - assert c["entry_observer"] == "boise" - assert c["exit_observer"] == "twin" - assert c["observer_list"] == "boise,twin" - - # Pending buffer is drained regardless of the gate's decision. - left = conn.execute( - "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?", - (cid,)).fetchone() - assert left["n"] == 0 - - # ══════════════════════════════════════════════════════════════════════ # 6. IMMINENCE BROADCAST TRIGGER (satpass's "just received" analog) # ══════════════════════════════════════════════════════════════════════ diff --git a/work/tests/test_satpass_persisted_timer.py b/work/tests/test_satpass_persisted_timer.py index d38d2bf..8e5fee8 100644 --- a/work/tests/test_satpass_persisted_timer.py +++ b/work/tests/test_satpass_persisted_timer.py @@ -1,4 +1,4 @@ -"""Tests for the satpass persisted-timer `due_at` column. +"""Tests for the satpass persisted-timer `due_at` column (schema only). Pending satellite-pass consolidations used to be scheduled only as in-memory asyncio TimerHandles, so a restart orphaned any satpass_pending rows: the row @@ -11,62 +11,24 @@ 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) +live is affected. + +The ingest path that wrote `due_at` (`central.satpass_handler.handle_satpass`) +was itself dead -- its only caller was the already-retired Central consumer +-- and was deleted in the same pass that relocated the still-live satellite +code to `env.satellite` (2026-07). Its due_at-persistence test is gone with +it; the `satpass_pending` table (and its due_at/peak_compass columns) are +now unused by any live code path, but the migrations that created them are +left in place (schema changes are out of scope for that pass). What remains +here: - 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 ─────────────────────────────────────────────── @@ -94,29 +56,3 @@ def test_v22_migration_applies_and_adds_due_at_column(tmp_path, monkeypatch): 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 diff --git a/work/tests/test_tle_fetch.py b/work/tests/test_tle_fetch.py index 86b599e..134a844 100644 --- a/work/tests/test_tle_fetch.py +++ b/work/tests/test_tle_fetch.py @@ -15,7 +15,7 @@ from meshai.env.tle_fetch import ( parse_tle_block, parse_tle_epoch, ) -from meshai.central.tle_handler import get_tle_by_norad +from meshai.env.satellite.tle_store import get_tle_by_norad from meshai.config import SatpassConfig from meshai.persistence import get_db