From a7561432848178fbfad60418af5d9009a9cd101e Mon Sep 17 00:00:00 2001 From: malice Date: Sun, 5 Jul 2026 01:06:29 -0600 Subject: [PATCH] feat(phase4a): peak_compass end-to-end + repair stale satpass test suite (#35) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add azimuth-at-peak-elevation compass to the satpass pipeline (Matt's decision: extend pass_predictor) and repair the 24 stale satpass tests that were written against the pre-async handle_satpass return contract. - pass_predictor.PassInfo gains azimuth_at_peak; _build_pass populates it from the already-computed peak sample - !satpass DM wire + Central consolidated broadcast wire render aosβ†’peakβ†’los - satpass_pending gains a peak_compass column (migration v21, SCHEMA_VERSION 20->21); persisted at ingest, carried through consolidation from the max-elevation observer's row - 24 stale satpass tests repaired to the ingestβ†’consolidate two-call contract (none weakened); satpass suite 112/112 green - full suite 34->10 failures (the 24 were these stale tests) NOTE: peak_compass changes the live Central satpass wire β€” deploy is HELD until the coordinated all-native flip (per Matt). Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/central/pass_predictor.py | 2 + work/meshai/central/satpass_handler.py | 31 +++++-- work/meshai/commands/satpass_cmd.py | 2 + work/meshai/persistence/db.py | 2 +- work/meshai/persistence/migrations/v21.sql | 7 ++ work/tests/test_satpass_broadcast_safety.py | 89 +++++++++++++------ work/tests/test_satpass_compass_fallback.py | 77 ++++++++++------- work/tests/test_satpass_handler.py | 96 +++++++++++++-------- work/tests/test_satpass_wire_fields.py | 48 ++++++++--- 9 files changed, 239 insertions(+), 115 deletions(-) create mode 100644 work/meshai/persistence/migrations/v21.sql diff --git a/work/meshai/central/pass_predictor.py b/work/meshai/central/pass_predictor.py index f0f77ee..3555cf3 100644 --- a/work/meshai/central/pass_predictor.py +++ b/work/meshai/central/pass_predictor.py @@ -46,6 +46,7 @@ class PassInfo: max_elevation: float # Degrees azimuth_at_aos: float # Degrees, clockwise from north azimuth_at_los: float # Degrees, clockwise from north + azimuth_at_peak: float # Degrees, clockwise from north (at max elevation) def compute_passes(line1: str, line2: str, @@ -144,6 +145,7 @@ def _build_pass(samples: list[tuple[datetime, float, float]]) -> PassInfo: max_elevation=samples[peak_idx][1], azimuth_at_aos=samples[0][2] % 360, azimuth_at_los=samples[-1][2] % 360, + azimuth_at_peak=samples[peak_idx][2] % 360, ) diff --git a/work/meshai/central/satpass_handler.py b/work/meshai/central/satpass_handler.py index 61a807b..7367020 100644 --- a/work/meshai/central/satpass_handler.py +++ b/work/meshai/central/satpass_handler.py @@ -190,16 +190,27 @@ def format_pass(*, sat_name: str, max_el: float, aos_compass: str, los_compass: str, broadcast: bool = True, entry_observer: Optional[str] = None, - exit_observer: Optional[str] = None) -> str: + exit_observer: Optional[str] = None, + peak_compass: Optional[str] = None) -> str: """Unified pass formatter with mode switch. broadcast=True: Two-line format with buckets, 12h times, LoRa budget. - πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} + πŸ›°οΈ {name} {bucket}, {aos_compass}β†’[peak_compassβ†’]{los_compass} {duration} min window, {rise}–{set} {AM/PM} {TZ} [tomorrow] [(region)] broadcast=False: Compact DM format with exact degrees. - {name} {HH:MM}–{HH:MM} {TZ} max {el}Β° {aos_compass}β†’{los_compass} + {name} {HH:MM}–{HH:MM} {TZ} max {el}Β° {aos_compass}β†’[peakβ†’]{los_compass} + + peak_compass: compass direction at peak elevation. When provided, the + compass segment renders as aosβ†’peakβ†’los; when None it stays aosβ†’los + (preserving legacy callers that don't thread the peak field). """ + # Compass sweep segment: include the peak point only when supplied. + if peak_compass: + compass_seg = f"{aos_compass}β†’{peak_compass}β†’{los_compass}" + else: + compass_seg = f"{aos_compass}β†’{los_compass}" + if broadcast: bucket = _elevation_bucket(max_el) # Duration in whole minutes @@ -213,7 +224,7 @@ def format_pass(*, sat_name: str, max_el: float, tz = _tz_abbr(aos_epoch) date_lbl = _date_label(aos_epoch) - line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {aos_compass}\u2192{los_compass}" + line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {compass_seg}" # Build time portion time_part = f"{dur_min} min window, {rise_str}\u2013{set_str} {ampm} {tz}{date_lbl}" @@ -235,7 +246,7 @@ def format_pass(*, sat_name: str, max_el: float, tz = _tz_abbr(aos_epoch) return (f"{sat_name} {aos_str}\u2013{los_str} {tz} " f"max {int(max_el)}\u00B0 " - f"{aos_compass}\u2192{los_compass}") + f"{compass_seg}") def _map_severity(max_el: float) -> str: @@ -397,10 +408,10 @@ def handle_satpass(envelope: dict, subject: str, 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, received_at) " - "VALUES (?,?,?,?,?,?,?,?,?,?)", + "aos_at, los_at, aos_compass, los_compass, peak_compass, received_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?)", (consolidated_id, observer, sat_name, norad_id, max_el, - aos_epoch, los_epoch, aos_compass, los_compass, now)) + aos_epoch, los_epoch, aos_compass, los_compass, direction, now)) # Signal consumer to schedule consolidation timer _pending_consolidation_ids.add(consolidated_id) @@ -443,6 +454,8 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None los_epoch = exit_["los_at"] aos_compass = entry["aos_compass"] los_compass = exit_["los_compass"] + # Peak belongs to whoever saw the highest elevation. + peak_compass = best["peak_compass"] entry_obs = entry["observer"] exit_obs = exit_["observer"] @@ -467,6 +480,7 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None wire = format_pass(sat_name=sat_name, max_el=max_el, aos_epoch=aos_epoch, los_epoch=los_epoch, aos_compass=aos_compass, los_compass=los_compass, + peak_compass=peak_compass, entry_observer=entry_obs, exit_observer=exit_obs) # Dry-run gate @@ -586,6 +600,7 @@ CREATE TABLE IF NOT EXISTS satpass_pending ( los_at INTEGER, aos_compass TEXT, los_compass TEXT, + peak_compass TEXT, received_at INTEGER, PRIMARY KEY (consolidated_id, observer) ); diff --git a/work/meshai/commands/satpass_cmd.py b/work/meshai/commands/satpass_cmd.py index 654a5f9..554a3c4 100644 --- a/work/meshai/commands/satpass_cmd.py +++ b/work/meshai/commands/satpass_cmd.py @@ -166,12 +166,14 @@ class SatpassCommand(CommandHandler): from meshai.central.satpass_handler import format_pass az_aos = azimuth_to_compass(p.azimuth_at_aos) az_los = azimuth_to_compass(p.azimuth_at_los) + az_peak = azimuth_to_compass(p.azimuth_at_peak) aos_epoch = int(p.aos_time.timestamp()) los_epoch = int(p.los_time.timestamp()) line = format_pass( sat_name=tle["name"], max_el=p.max_elevation, aos_epoch=aos_epoch, los_epoch=los_epoch, aos_compass=az_aos, los_compass=az_los, + peak_compass=az_peak, broadcast=False, ) all_lines.append(line) diff --git a/work/meshai/persistence/db.py b/work/meshai/persistence/db.py index 82c8f14..d16dbea 100644 --- a/work/meshai/persistence/db.py +++ b/work/meshai/persistence/db.py @@ -30,7 +30,7 @@ logger = logging.getLogger(__name__) DEFAULT_DB_PATH = "/data/meshai.sqlite" MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH" -SCHEMA_VERSION = 20 +SCHEMA_VERSION = 21 SCHEMA_META_TABLE = "schema_meta" MIGRATIONS_DIR = Path(__file__).parent / "migrations" diff --git a/work/meshai/persistence/migrations/v21.sql b/work/meshai/persistence/migrations/v21.sql new file mode 100644 index 0000000..f76f2f5 --- /dev/null +++ b/work/meshai/persistence/migrations/v21.sql @@ -0,0 +1,7 @@ +-- v21: peak_compass column on satpass_pending. +-- Carries the compass direction at a pass's PEAK elevation through the +-- observer-consolidation buffer so the consolidated broadcast wire can +-- render aosβ†’peakβ†’los. Nullable; legacy pending rows (there are none at +-- rest β€” the table is drained per bucket) keep NULL and render aosβ†’los. + +ALTER TABLE satpass_pending ADD COLUMN peak_compass TEXT; diff --git a/work/tests/test_satpass_broadcast_safety.py b/work/tests/test_satpass_broadcast_safety.py index 1507b28..2893394 100644 --- a/work/tests/test_satpass_broadcast_safety.py +++ b/work/tests/test_satpass_broadcast_safety.py @@ -89,6 +89,30 @@ def _clear_handler_flags(): 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 # ══════════════════════════════════════════════════════════════════════ @@ -113,12 +137,14 @@ class TestOptInBirdFilter: _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=1718163120) + "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=1718163180) + "central.sat.pass.iss", data={}, now=1781235000) matching = [r for r in caplog.records if "no norad_ids configured" in r.message] @@ -132,15 +158,15 @@ class TestOptInBirdFilter: # ISS pass should air iss_env = _envelope(norad_id=25544, max_el=65.0) - iss_result = handle_satpass(iss_env, "central.sat.pass.iss", - data={}, now=1718163120) + 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 + # 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 = handle_satpass(noaa_env, "central.sat.pass.noaa", - data={}, now=1718163120) + 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): @@ -201,11 +227,15 @@ class TestRateCap: _enable_satpass_db(norad_ids=[25544], dry_run=False, max_per_hour=0) _clear_handler_flags() - now = 1718200000 + # 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") - handle_satpass(env, "central.sat.pass.iss", data={}, now=now) + 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" @@ -250,11 +280,14 @@ class TestDryRun: with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"): env = _envelope(norad_id=25544, sat_name="ISS", max_el=70.0) - handle_satpass(env, "central.sat.pass.iss", data={}, - now=1718163120) + 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:")] + 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 @@ -264,11 +297,12 @@ class TestDryRun: _enable_satpass_db(norad_ids=[25544], dry_run=False) _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) + 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): @@ -322,8 +356,13 @@ class TestBroadcastWireFormat: lines = wire.split("\n") assert len(lines) == 2 + # No peak_compass passed here \u2192 the sweep stays aos\u2192los (SW\u2192NE). assert lines[0] == "\U0001F6F0\uFE0F ISS high pass, SW\u2192NE" - assert lines[1] == "6 minute window, 8:38\u20138:44 PM MDT" + # line2 renders "min window" + a date qualifier for a pass not + # occurring today (the fixed 2026-06-12 date is always in the past + # relative to run time). + assert lines[1].startswith("6 min window, 8:38\u20138:44 PM MDT") + assert lines[1] == "6 min window, 8:38\u20138:44 PM MDT Fri Jun 12" def test_bucket_overhead_at_60(self): """max_el=60 should be 'overhead'.""" @@ -464,7 +503,7 @@ class TestNoradIdTypeCoercion: _enable_satpass_db(norad_ids=["25544"], dry_run=False) env = _envelope(norad_id=25544, max_el=80.0) - result = handle_satpass(env, "test.subject", data={}, now=1781235000) + 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): @@ -475,13 +514,13 @@ class TestNoradIdTypeCoercion: # int in list, int on wire env_iss = _envelope(norad_id=25544, max_el=65.0) - result_iss = handle_satpass(env_iss, "test.subject", data={}, now=1781235000) + 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 = handle_satpass(env_noaa, "test.subject", data={}, now=1781235000) + 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): @@ -493,7 +532,7 @@ class TestNoradIdTypeCoercion: env = _envelope(norad_id=25544, max_el=80.0) # Must not raise, and the valid entry should still match - result = handle_satpass(env, "test.subject", data={}, now=1781235000) + 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): @@ -513,7 +552,7 @@ class TestNoradIdTypeCoercion: _enable_satpass_db(norad_ids=[25544], dry_run=False) env = _envelope(norad_id=25544, max_el=80.0) - result = handle_satpass(env, "test.subject", data={}, now=1781235000) + 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): @@ -567,7 +606,7 @@ class TestStalenessGuard: 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 = handle_satpass(env, "test.subject", data={}, now=now) + 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): @@ -583,7 +622,7 @@ class TestStalenessGuard: 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 = handle_satpass(env, "test.subject", data={}, now=now) + 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): @@ -601,7 +640,7 @@ class TestStalenessGuard: # Patch los to None by removing los_time from inner data env["data"]["data"]["los_time"] = None - result = handle_satpass(env, "test.subject", data={}, now=now) + 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 diff --git a/work/tests/test_satpass_compass_fallback.py b/work/tests/test_satpass_compass_fallback.py index 4c6713b..1bd8515 100644 --- a/work/tests/test_satpass_compass_fallback.py +++ b/work/tests/test_satpass_compass_fallback.py @@ -145,6 +145,27 @@ def _clear_handler_flags(): delattr(handle_satpass, attr) +def _ingest_and_consolidate(env, subject, *, now, data=None): + """Drive the two-call async satpass contract (ingest β†’ consolidate). + + handle_satpass() ingests the pass and returns None; the consumer then + runs consolidate_satpass_pending(), which yields the (wire, data) to + broadcast (or None). Returns the consolidation result so a test can + assert on the real broadcast 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 + + # ── (a) satpass_predict envelope: raw azimuths β†’ non-empty compass ─── def test_satpass_predict_compass_from_raw_azimuths(): @@ -156,29 +177,26 @@ def test_satpass_predict_compass_from_raw_azimuths(): _clear_handler_flags() now = 1781330520 # well before the AO-27 pass window - wire = handle_satpass( + result = _ingest_and_consolidate( AO27_PREDICT_ENVELOPE, "central.sat.pass.us.id.filer", - data={}, now=now, ) - assert wire is not None, "handler returned None for satpass_predict envelope" + assert result is not None, "handler returned None for satpass_predict envelope" + wire, _ = result lines = wire.split("\n") assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}" - # Line 1 must contain non-empty compass directions - # 163.2Β° β†’ S (8-point compass), 348.7Β° β†’ N - assert "S" in lines[0].split(",")[-1], f"Expected S (from 163.2Β°) in compass portion: {lines[0]!r}" - assert "\u2192" in lines[0], f"Expected β†’ arrow in line 1: {lines[0]!r}" - - # Extract the compass portion: after bucket comma, before newline - # Format: πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} - arrow_idx = lines[0].index("\u2192") - los_part = lines[0][arrow_idx + 1:] - assert los_part != "", f"los_compass is empty in line 1: {lines[0]!r}" - # 348.7Β° β†’ N in 8-point compass - assert los_part == "N", f"Expected N (from 348.7Β°) after arrow: {los_part!r}" + # Line 1 format: name bucket, aosβ†’peakβ†’los. Raw azimuths convert + # to a non-empty 3-point compass sweep. + compass = lines[0].split(", ")[-1] + parts = compass.split("\u2192") + assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}" + # 163.2 -> S, 245.0 -> SW (peak), 348.7 -> N (8-point compass) + assert parts[0] == "S", f"Expected S (from 163.2): {parts!r}" + assert parts[1] == "SW", f"Expected SW (from peak 245.0): {parts!r}" + assert parts[2] == "N", f"Expected N (from 348.7): {parts!r}" # ── (b) n2yo envelope: precomputed _compass strings used as-is ─────── @@ -192,21 +210,22 @@ def test_n2yo_precomputed_compass_unchanged(): _clear_handler_flags() now = 1781065800 # before NOAA-18 pass window - wire = handle_satpass( + result = _ingest_and_consolidate( N2YO_ENVELOPE, "central.sat.pass.us.id.filer", - data={}, now=now, ) - assert wire is not None, "handler returned None for n2yo envelope" + assert result is not None, "handler returned None for n2yo envelope" + wire, _ = result lines = wire.split("\n") - # Must use the precomputed strings: SEβ†’N - assert "SE" in lines[0], f"Expected SE from precomputed _compass: {lines[0]!r}" - # The los_compass from precomputed is "N" - arrow_idx = lines[0].index("\u2192") - los_part = lines[0][arrow_idx + 1:] - assert los_part == "N", f"Expected precomputed los_compass N, got {los_part!r}" + # Must use the precomputed strings verbatim: SEβ†’ENEβ†’N + compass = lines[0].split(", ")[-1] + parts = compass.split("\u2192") + assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}" + assert parts[0] == "SE", f"Expected precomputed aos SE: {parts!r}" + assert parts[1] == "ENE", f"Expected precomputed peak ENE: {parts!r}" + assert parts[2] == "N", f"Expected precomputed los N: {parts!r}" # ── (c) envelope with neither β†’ empty compass, no crash ────────────── @@ -228,16 +247,16 @@ def test_no_compass_no_azimuth_no_crash(): d.pop(key, None) now = 1781330520 - wire = handle_satpass( + result = _ingest_and_consolidate( env, "central.sat.pass.us.id.filer", - data={}, now=now, ) - assert wire is not None, "handler crashed or returned None β€” should produce wire with empty compass" + assert result is not None, "handler crashed or returned None β€” should produce wire with empty compass" + wire, _ = result lines = wire.split("\n") assert len(lines) == 2 - # Arrow should still be present with empty directions: "β†’" or similar - assert "\u2192" in lines[0], f"Expected β†’ in line 1 even with empty compass: {lines[0]!r}" + # Arrow still present even with empty (peak empty -> no peak segment): + assert "\u2192" in lines[0], f"Expected arrow in line 1 even with empty compass: {lines[0]!r}" # No crash = test passes diff --git a/work/tests/test_satpass_handler.py b/work/tests/test_satpass_handler.py index a3efedd..3eaf05e 100644 --- a/work/tests/test_satpass_handler.py +++ b/work/tests/test_satpass_handler.py @@ -53,6 +53,7 @@ def mock_adapter_config(): 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 @@ -63,18 +64,38 @@ def mock_adapter_config(): 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_db, mock_adapter_config): - """A pass with high elevation should broadcast.""" - from meshai.central.satpass_handler import handle_satpass - + 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 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - + result = _ingest_and_consolidate(env, "central.sat.pass.iss", + now=1781235000) assert result is not None - assert "ISS" in result + 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.""" @@ -96,14 +117,12 @@ class TestSatpassHandler: assert result is None - def test_observer_filter_allows_match(self, mock_db, mock_adapter_config): + def test_observer_filter_allows_match(self, mock_adapter_config): """A pass for configured observer should broadcast.""" - from meshai.central.satpass_handler import handle_satpass - mock_adapter_config.observers = ["Boise", "Magic Valley"] env = _envelope(observer="Boise", max_el=45.0) - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - + 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): @@ -116,48 +135,49 @@ class TestSatpassHandler: assert result is None - def test_dedup_blocks_second_broadcast(self, mock_db, mock_adapter_config): + def test_dedup_blocks_second_broadcast(self, mock_adapter_config): """Second pass in same hour bucket should be deduplicated.""" - from meshai.central.satpass_handler import handle_satpass - - # First call returns no existing broadcast - mock_db.execute.return_value.fetchone.return_value = None - env = _envelope(max_el=60.0) - result1 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) + + # First pass consolidates into a real broadcast. + result1 = _ingest_and_consolidate(env, "central.sat.pass.iss", + now=1781235000) assert result1 is not None - # Second call simulates existing broadcast - mock_db.execute.return_value.fetchone.return_value = {"last_broadcast_at": 1718163120} + # Simulate the broadcast committing (marks satpass_events broadcast). + _wire, data1 = result1 + data1["_on_broadcast_committed"](1781235010) - result2 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163180) + # 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_db, mock_adapter_config): + def test_wire_format(self, mock_adapter_config): """Wire format should have 2 lines with correct info.""" - from meshai.central.satpass_handler import handle_satpass - env = _envelope(sat_name="ISS", max_el=75, observer="Boise", - direction="NW-SE", aos_compass="SW", los_compass="NE") - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) + 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 - lines = result.split("\n") + lines = wire.split("\n") assert len(lines) == 2 assert "ISS" in lines[0] assert "overhead" in lines[0] - assert "SW" in lines[0] - assert "NE" in lines[0] - assert "minute window" in lines[1] + assert "SW" in lines[0] # aos_compass + assert "S" in lines[0] # peak_compass (new) + assert "NE" in lines[0] # los_compass + assert "min window" in lines[1] - def test_commit_callback_attached(self, mock_db, mock_adapter_config): - """Broadcast should attach commit callback.""" - from meshai.central.satpass_handler import handle_satpass - - data = {} + 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 = handle_satpass(env, "central.sat.pass.iss", data=data, now=1718163120) - + 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" diff --git a/work/tests/test_satpass_wire_fields.py b/work/tests/test_satpass_wire_fields.py index 5473c29..feb4916 100644 --- a/work/tests/test_satpass_wire_fields.py +++ b/work/tests/test_satpass_wire_fields.py @@ -96,6 +96,26 @@ def _enable_satpass(norad_ids=None): invalidate_cache() +def _ingest_and_consolidate(env, subject, *, now, data=None): + """Drive the two-call async satpass contract (ingest -> consolidate). + + handle_satpass() ingests the pass and returns None; the consumer then + runs consolidate_satpass_pending(), which returns the (wire, data) to + broadcast (or None). Returns the consolidation result. + """ + 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 + + # ── Handler produces correct satpass_events row ───────────────────── def test_noaa18_envelope_produces_satpass_event(): @@ -110,13 +130,13 @@ def test_noaa18_envelope_produces_satpass_event(): del handle_satpass._no_norad_ids_logged now = 1781065800 # before NOAA18 envelope los_time - wire = handle_satpass( + result = _ingest_and_consolidate( NOAA18_ENVELOPE, "central.sat.pass.us.id.filer", - data={}, now=now, ) - assert wire is not None, "handler returned None -- field extraction failed" + assert result is not None, "handler returned None -- field extraction failed" + wire, _ = result conn = get_db() rows = conn.execute( @@ -147,25 +167,26 @@ def test_noaa18_wire_message_format(): if hasattr(handle_satpass, "_no_norad_ids_logged"): del handle_satpass._no_norad_ids_logged - wire = handle_satpass( + result = _ingest_and_consolidate( NOAA18_ENVELOPE, "central.sat.pass.us.id.filer", - data={}, now=1781065800, # before NOAA18 envelope los_time ) - assert wire is not None + assert result is not None + wire, _ = result lines = wire.split("\n") assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}" - # Line 1: satellite name + bucket + compass directions + # Line 1: satellite name + bucket + compass sweep (aos->peak->los) assert "NOAA 18" in lines[0] assert "low pass" in lines[0] # 22.69 < 30 = low pass - assert "SE" in lines[0] # aos_compass - assert "N" in lines[0] # los_compass + assert "SE" in lines[0] # aos_compass + assert "ENE" in lines[0] # peak_compass (new) + assert "N" in lines[0] # los_compass - # Line 2: duration + time window - assert "minute window" in lines[1] + # Line 2: duration + time window ("min window") + assert "min window" in lines[1] def test_missing_norad_id_rejected(): @@ -230,13 +251,12 @@ def test_observer_fallback_to_slug(): del env["data"]["data"]["observer_name"] # observer_slug = "filer" still present - wire = handle_satpass( + result = _ingest_and_consolidate( env, "central.sat.pass.us.id.filer", - data={}, now=1781065800, # before NOAA18 envelope los_time ) - assert wire is not None + assert result is not None # Observer name stored in DB, not in broadcast wire format from meshai.persistence import get_db conn = get_db()