From e90344435697907595a6f1ae806ba2c97adb0559 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 12 Jun 2026 23:00:03 +0000 Subject: [PATCH] =?UTF-8?q?fix(satpass):=20broadcast=20safety=20controls?= =?UTF-8?q?=20=E2=80=94=20opt-in=20filter,=20rate=20cap,=20dry-run,=20wire?= =?UTF-8?q?=20format?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Incident response for 343 broadcasts in 126s (2026-06-12 22:10 UTC). Five safety controls: 1. OPT-IN BIRD FILTER: norad_ids=[] now means "broadcast nothing" (was "all birds"). Empty list logs once at INFO and suppresses all broadcasts. The !satpass DM command remains ungated β€” it queries any bird in the TLE cache using command_norad_ids as bare-command default. Two paths, two rules. 2. RATE CAP: new satpass.max_broadcasts_per_hour (int, default 4). Excess qualifying passes logged and suppressed. Broadcast path only. 3. DRY-RUN MODE: new satpass.dry_run (bool, default TRUE). Logs exact wire text at INFO prefixed "DRY-RUN would air:" without dispatching. Go-live: enabled=true + dry_run=true β†’ observe β†’ dry_run=false. 4. ELEVATION DEFAULT: min_elevation REGISTRY default already at 30 (confirmed, no change needed). 5. BROADCAST WIRE FORMAT: two-line LoRa-tight format with buckets: πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} {duration} minute window, {rise}–{set} {AM/PM} MDT Buckets: overhead (β‰₯60Β°), high pass (30-59Β°), low pass (<30Β°). DM format keeps exact degrees. One format_pass() function with broadcast= mode switch β€” two callers, one function. Co-Authored-By: Claude Opus 4.6 --- meshai/adapter_config/defaults.py | 12 +- meshai/central/satpass_handler.py | 203 ++++++++--- meshai/commands/satpass_cmd.py | 17 +- tests/test_adapter_config_api.py | 2 +- tests/test_adapter_config_foundation.py | 2 +- tests/test_satpass_broadcast_safety.py | 453 ++++++++++++++++++++++++ tests/test_satpass_handler.py | 32 +- tests/test_satpass_wire_fields.py | 55 ++- 8 files changed, 700 insertions(+), 76 deletions(-) create mode 100644 tests/test_satpass_broadcast_safety.py diff --git a/meshai/adapter_config/defaults.py b/meshai/adapter_config/defaults.py index 7eae654..3198729 100644 --- a/meshai/adapter_config/defaults.py +++ b/meshai/adapter_config/defaults.py @@ -613,13 +613,23 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { ("satpass", "norad_ids"): { "default": [], "type": "json", - "description": "NORAD catalog IDs to include (empty = all).", + "description": "NORAD catalog IDs to broadcast (empty = broadcast nothing, opt-in only).", }, ("satpass", "command_norad_ids"): { "default": [25544], "type": "json", "description": "Default NORAD IDs for bare !satpass command (default: [25544] ISS).", }, + ("satpass", "max_broadcasts_per_hour"): { + "default": 4, + "type": "int", + "description": "Maximum satellite pass broadcasts per hour. Excess qualifying passes are logged and suppressed.", + }, + ("satpass", "dry_run"): { + "default": True, + "type": "bool", + "description": "Dry-run mode: log wire text at INFO with DRY-RUN prefix instead of dispatching. Default true so satpass re-enables inert.", + }, # ================================================================= # DASHBOARD -- UI-only settings persisted for the operator diff --git a/meshai/central/satpass_handler.py b/meshai/central/satpass_handler.py index dd8fc3a..7df7a7a 100644 --- a/meshai/central/satpass_handler.py +++ b/meshai/central/satpass_handler.py @@ -6,7 +6,12 @@ Filter criteria: (a) Pass must be for an observer in adapter_config.satpass.observers (empty list = all observers) (b) Max elevation must meet adapter_config.satpass.min_elevation (default 30) - (c) Optional NORAD ID filter via adapter_config.satpass.norad_ids + (c) Opt-in NORAD ID filter via adapter_config.satpass.norad_ids + (empty list = broadcast NOTHING β€” opt-in only) + +Rate cap: adapter_config.satpass.max_broadcasts_per_hour (default 4). +Dry-run: adapter_config.satpass.dry_run (default True) β€” logs wire text + at INFO with "DRY-RUN would air:" prefix, does not dispatch. Dedup bucketing: canonical event_id = {norad_id}:{observer}:{aos_bucket} where aos_bucket = floor(aos_epoch / 3600) -- one broadcast per satellite @@ -17,10 +22,11 @@ Severity mapping: 3 = priority (>= 45 deg max elevation) <= 2 = routine -Wire format (multi-line, LoRa-tight): - Line 1: satellite emoji {sat_name} Pass -- {max_el} deg max - Line 2: AOS {aos_time} . LOS {los_time} - Line 3: {observer} . {direction} +Broadcast wire format (two lines, LoRa-tight): + Line 1: πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} + Line 2: {duration} minute window, {rise}–{set} {AM/PM} MDT +DM wire format (compact, exact degrees): + {name} {HH:MM}–{HH:MM} {TZ} max {el}Β° {aos_compass}β†’{los_compass} """ from __future__ import annotations @@ -29,12 +35,16 @@ import logging import time from datetime import datetime, timezone from typing import Any, Optional +from zoneinfo import ZoneInfo from meshai.adapter_config import adapter_config from meshai.persistence import get_db logger = logging.getLogger(__name__) +# Mountain time for broadcast display +_TZ = ZoneInfo("America/Boise") + def _now() -> int: return int(time.time()) @@ -73,17 +83,109 @@ def _parse_iso_epoch(s) -> Optional[int]: return None -def _format_time(epoch: Optional[int]) -> str: - """Format epoch to HH:MM local time string.""" +def _elevation_bucket(max_el: float) -> str: + """Map max elevation to human-readable bucket name.""" + if max_el >= 60: + return "overhead" + if max_el >= 30: + return "high pass" + return "low pass" + + +def _format_time_12h(epoch: Optional[int]) -> str: + """Format epoch to h:mm AM/PM in America/Boise.""" if epoch is None: return "?" try: - dt = datetime.fromtimestamp(epoch) + dt = datetime.fromtimestamp(epoch, tz=_TZ) + # Use %-I for no-leading-zero hour on Linux, fall back to %I + try: + return dt.strftime("%-I:%M") + except ValueError: + return dt.strftime("%I:%M").lstrip("0") + except Exception: + return "?" + + +def _format_ampm(epoch: Optional[int]) -> str: + """Return AM or PM for an epoch in America/Boise.""" + if epoch is None: + return "" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) + return dt.strftime("%p") + except Exception: + return "" + + +def _format_time_24h(epoch: Optional[int]) -> str: + """Format epoch to HH:MM local time string (24h).""" + if epoch is None: + return "?" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) return dt.strftime("%H:%M") except Exception: return "?" +def _tz_abbr(epoch: Optional[int]) -> str: + """Return timezone abbreviation for an epoch in America/Boise.""" + if epoch is None: + return "MDT" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) + return dt.strftime("%Z") + except Exception: + return "MDT" + + +def _azimuth_to_compass(az_deg: float) -> str: + """Convert azimuth in degrees to 8-point compass direction.""" + az = az_deg % 360 + dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"] + idx = int((az + 22.5) / 45) % 8 + return dirs[idx] + + +def format_pass(*, sat_name: str, max_el: float, + aos_epoch: Optional[int], los_epoch: Optional[int], + aos_compass: str, los_compass: str, + broadcast: bool = True) -> str: + """Unified pass formatter with mode switch. + + broadcast=True: Two-line format with buckets, 12h times, LoRa budget. + πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} + {duration} minute window, {rise}–{set} {AM/PM} {TZ} + + broadcast=False: Compact DM format with exact degrees. + {name} {HH:MM}–{HH:MM} {TZ} max {el}Β° {aos_compass}β†’{los_compass} + """ + if broadcast: + bucket = _elevation_bucket(max_el) + # Duration in whole minutes + if aos_epoch is not None and los_epoch is not None: + dur_min = max(1, round((los_epoch - aos_epoch) / 60)) + else: + dur_min = 0 + rise_str = _format_time_12h(aos_epoch) + set_str = _format_time_12h(los_epoch) + ampm = _format_ampm(los_epoch) + tz = _tz_abbr(aos_epoch) + + line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {aos_compass}\u2192{los_compass}" + line2 = f"{dur_min} minute window, {rise_str}\u2013{set_str} {ampm} {tz}" + return f"{line1}\n{line2}" + else: + # DM format: compact with exact degrees + aos_str = _format_time_24h(aos_epoch) + los_str = _format_time_24h(los_epoch) + 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}") + + def _map_severity(max_el: float) -> str: """Map max elevation to severity word.""" if max_el >= 60: @@ -102,6 +204,22 @@ def _canonical_id(norad_id: int, observer: str, aos_epoch: int) -> str: return f"{norad_id}:{observer}:{bucket}" +def _check_rate_cap(conn, now: int, max_per_hour: int) -> tuple[bool, int]: + """Check if broadcast rate cap has been reached. + + Returns (allowed, suppressed_count) where suppressed_count is the + number of broadcasts already made in the current hour window. + """ + hour_start = (now // 3600) * 3600 + row = conn.execute( + "SELECT COUNT(*) AS cnt FROM satpass_events " + "WHERE last_broadcast_at >= ? AND last_broadcast_at IS NOT NULL", + (hour_start,), + ).fetchone() + count = row["cnt"] if row else 0 + return (count < max_per_hour, count) + + def handle_satpass(envelope: dict, subject: str, data: Optional[dict] = None, now: Optional[int] = None) -> Optional[str]: @@ -138,6 +256,8 @@ def handle_satpass(envelope: dict, subject: str, aos_iso = d.get("aos_time") los_iso = d.get("los_time") direction = d.get("azimuth_at_peak_compass") or "" + aos_compass = d.get("azimuth_at_aos_compass") or direction or "" + los_compass = d.get("azimuth_at_los_compass") or "" if norad_id is None or max_el is None: logger.debug("satpass_handler: missing norad_id or max_elevation_deg") @@ -156,9 +276,14 @@ def handle_satpass(envelope: dict, subject: str, logger.debug("satpass_handler: observer %r not in configured list", observer) return None - # NORAD ID filter (empty = all) + # OPT-IN NORAD ID filter: empty list = broadcast NOTHING norad_ids = getattr(cfg, "norad_ids", []) or [] - if norad_ids and norad_id not in norad_ids: + if not norad_ids: + if not getattr(handle_satpass, "_no_norad_ids_logged", False): + logger.info("satpass: no norad_ids configured; pass broadcasts disabled") + handle_satpass._no_norad_ids_logged = True + return None + if norad_id not in norad_ids: logger.debug("satpass_handler: norad_id %d not in configured list", norad_id) return None @@ -197,40 +322,38 @@ def handle_satpass(envelope: dict, subject: str, subject=subject, handled=0, table_name="satpass_events", table_pk=event_id) - if row is None: - # First time seeing this pass bucket - _upsert_satpass(conn, event_id=event_id, norad_id=norad_id, - sat_name=sat_name, observer=observer, - max_elevation=max_el, aos_at=aos_epoch, - los_at=los_epoch, payload_json=payload_json, - first_seen_at=now, set_last_broadcast=False) - wire = _render(sat_name, max_el, aos_epoch, los_epoch, observer, direction) - _attach_commit(data, event_id=event_id, event_log_row_id=log_id) - return wire + if row is not None and row["last_broadcast_at"] is not None: + # Already broadcast this pass bucket + return None - if row["last_broadcast_at"] is None: - # Seen but not yet broadcast - wire = _render(sat_name, max_el, aos_epoch, los_epoch, observer, direction) - _attach_commit(data, event_id=event_id, event_log_row_id=log_id) - return wire + # Rate cap check + max_per_hour = int(getattr(cfg, "max_broadcasts_per_hour", 4)) + allowed, count = _check_rate_cap(conn, now, max_per_hour) + if not allowed: + logger.info("satpass: rate cap reached, suppressing 1 passes") + return None - # Already broadcast this pass bucket - return None + # Build wire message + _upsert_satpass(conn, event_id=event_id, norad_id=norad_id, + sat_name=sat_name, observer=observer, + max_elevation=max_el, aos_at=aos_epoch, + los_at=los_epoch, payload_json=payload_json, + first_seen_at=now, set_last_broadcast=False) + 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, + broadcast=True, + ) + # Dry-run gate: log but don't dispatch + dry_run = getattr(cfg, "dry_run", True) + if dry_run: + logger.info("DRY-RUN would air: %s", wire) + return None -def _render(sat_name: str, max_el: float, aos_epoch: Optional[int], - los_epoch: Optional[int], observer: str, direction: str) -> str: - """Render wire format message.""" - aos_str = _format_time(aos_epoch) - los_str = _format_time(los_epoch) - - line1 = f"\U0001F6F0\uFE0F {sat_name} Pass \u2014 {int(max_el)}\u00B0 max" - line2 = f"AOS {aos_str} \u00B7 LOS {los_str}" - line3 = f"{observer}" - if direction: - line3 += f" \u00B7 {direction}" - - return "\n".join([line1, line2, line3]) + _attach_commit(data, event_id=event_id, event_log_row_id=log_id) + return wire def _upsert_satpass(conn, *, event_id, norad_id, sat_name, observer, diff --git a/meshai/commands/satpass_cmd.py b/meshai/commands/satpass_cmd.py index 03f3b08..654a5f9 100644 --- a/meshai/commands/satpass_cmd.py +++ b/meshai/commands/satpass_cmd.py @@ -163,16 +163,17 @@ class SatpassCommand(CommandHandler): continue for p in passes: - aos_local = p.aos_time.astimezone(_TZ) - los_local = p.los_time.astimezone(_TZ) - tz_abbr = aos_local.strftime("%Z") - aos_str = aos_local.strftime("%H:%M") - los_str = los_local.strftime("%H:%M") + 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) - line = (f"{tle['name']} {aos_str}\u2013{los_str} {tz_abbr} " - f"max {int(p.max_elevation)}\u00B0 " - f"{az_aos}\u2192{az_los}") + 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, + broadcast=False, + ) all_lines.append(line) if not all_lines: diff --git a/tests/test_adapter_config_api.py b/tests/test_adapter_config_api.py index e659815..564d6d5 100644 --- a/tests/test_adapter_config_api.py +++ b/tests/test_adapter_config_api.py @@ -36,7 +36,7 @@ def test_list_returns_all_59_keys(client): # 14 adapters with at least one key (itd_511 has zero -- not in the # grouped dict because the SQL only returns rows that exist). total = sum(len(v) for v in body.values()) - assert total == 90 + assert total == 92 def test_list_grouped_by_adapter(client): diff --git a/tests/test_adapter_config_foundation.py b/tests/test_adapter_config_foundation.py index 443f59d..8227ab8 100644 --- a/tests/test_adapter_config_foundation.py +++ b/tests/test_adapter_config_foundation.py @@ -75,7 +75,7 @@ def test_adapter_config_type_check_constrains_vocabulary(fresh_db): def test_registry_at_59_entries(): """v0.6-3a.1 trim: 43 CONFIG-only keys (was 77 in v0.6-3a draft).""" - assert len(REGISTRY) == 90, ( + assert len(REGISTRY) == 92, ( f"REGISTRY drift guard; got {len(REGISTRY)}. " f"If a sentence template / emoji / heuristic snuck in, it belongs in CODE not config." ) diff --git a/tests/test_satpass_broadcast_safety.py b/tests/test_satpass_broadcast_safety.py new file mode 100644 index 0000000..39268bd --- /dev/null +++ b/tests/test_satpass_broadcast_safety.py @@ -0,0 +1,453 @@ +"""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 + 4. Elevation default: REGISTRY min_elevation = 30 + 5. Broadcast wire format: two-line, buckets, byte budget +""" +from __future__ import annotations + +import json +import logging +import time +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +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) + + +# ══════════════════════════════════════════════════════════════════════ +# 1. OPT-IN BIRD FILTER +# ══════════════════════════════════════════════════════════════════════ + +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() + + 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) + handle_satpass(_envelope(norad_id=25544, max_el=80.0, + aos="2026-06-12T04:32:00Z"), + "central.sat.pass.iss", data={}, now=1718163180) + + 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 = handle_satpass(iss_env, "central.sat.pass.iss", + data={}, now=1718163120) + assert iss_result is not None, "ISS pass should broadcast" + + # NOAA-18 pass should be rejected + 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) + 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.""" + # This test verifies the DM command path doesn't use the + # broadcast norad_ids filter. The command_norad_ids is separate. + from meshai.adapter_config.defaults import REGISTRY + # Verify command_norad_ids is a separate key + assert ("satpass", "command_norad_ids") in REGISTRY + assert ("satpass", "norad_ids") in REGISTRY + assert REGISTRY[("satpass", "command_norad_ids")]["default"] == [25544] + assert REGISTRY[("satpass", "norad_ids")]["default"] == [] + + +# ══════════════════════════════════════════════════════════════════════ +# 2. RATE CAP +# ══════════════════════════════════════════════════════════════════════ + +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 = 1718200000 + 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) + + 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, + # it uses pass_predictor directly. Verify they're separate paths. + import inspect + from meshai.commands import satpass_cmd + src = inspect.getsource(satpass_cmd) + assert "max_broadcasts_per_hour" not in src + assert "_check_rate_cap" not in src + + +# ══════════════════════════════════════════════════════════════════════ +# 3. DRY-RUN MODE +# ══════════════════════════════════════════════════════════════════════ + +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) + handle_satpass(env, "central.sat.pass.iss", data={}, + now=1718163120) + + 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() + + 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 not None, "dry_run=False should dispatch" + 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 + spec = REGISTRY[("satpass", "dry_run")] + assert spec["default"] is True + assert spec["type"] == "bool" + + +# ══════════════════════════════════════════════════════════════════════ +# 4. ELEVATION DEFAULT +# ══════════════════════════════════════════════════════════════════════ + +class TestElevationDefault: + """min_elevation default in REGISTRY must be 30.""" + + def test_registry_min_elevation_default_30(self): + from meshai.adapter_config.defaults import REGISTRY + spec = REGISTRY[("satpass", "min_elevation")] + assert spec["default"] == 30 + + +# ══════════════════════════════════════════════════════════════════════ +# 5. BROADCAST WIRE FORMAT +# ══════════════════════════════════════════════════════════════════════ + +class TestBroadcastWireFormat: + """Two-line format, buckets, byte budget.""" + + def test_exact_two_line_example(self): + """Formatter produces the exact two-line example from spec.""" + from meshai.central.satpass_handler import format_pass + + # ISS high pass, SWβ†’NE, 6 minute window, 8:38–8:44 PM MDT + # We need epoch values that produce 8:38 PM and 8:44 PM MDT + from zoneinfo import ZoneInfo + tz = ZoneInfo("America/Boise") + aos_dt = datetime(2026, 6, 12, 20, 38, 0, tzinfo=tz) + los_dt = datetime(2026, 6, 12, 20, 44, 0, tzinfo=tz) + aos_epoch = int(aos_dt.timestamp()) + los_epoch = int(los_dt.timestamp()) + + wire = format_pass( + sat_name="ISS", max_el=55.0, + aos_epoch=aos_epoch, los_epoch=los_epoch, + aos_compass="SW", los_compass="NE", + broadcast=True, + ) + + lines = wire.split("\n") + assert len(lines) == 2 + + assert lines[0] == "\U0001F6F0\uFE0F ISS high pass, SW\u2192NE" + assert lines[1] == "6 minute window, 8:38\u20138:44 PM MDT" + + 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 zoneinfo import ZoneInfo + + tz = ZoneInfo("America/Boise") + + # Test with longest plausible satellite name + test_cases = [ + ("ISS", 75.0, "SW", "NE"), + ("NOAA 18", 45.0, "SE", "NW"), + ("AMATEUR-SAT-1", 35.0, "S", "N"), + ("SO-50", 88.0, "NE", "SW"), + ] + + for sat_name, max_el, aos_c, los_c in test_cases: + aos_dt = datetime(2026, 6, 12, 20, 38, 0, tzinfo=tz) + los_dt = datetime(2026, 6, 12, 20, 44, 0, tzinfo=tz) + wire = format_pass( + sat_name=sat_name, max_el=max_el, + aos_epoch=int(aos_dt.timestamp()), + los_epoch=int(los_dt.timestamp()), + aos_compass=aos_c, los_compass=los_c, + broadcast=True, + ) + byte_len = len(wire.encode("utf-8")) + assert byte_len <= 120, ( + f"Broadcast for {sat_name} is {byte_len} bytes, exceeds 120: " + f"{wire!r}" + ) + + 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 zoneinfo import ZoneInfo + + tz = ZoneInfo("America/Boise") + aos_dt = datetime(2026, 6, 12, 20, 38, 0, tzinfo=tz) + los_dt = datetime(2026, 6, 12, 20, 44, 0, tzinfo=tz) + + wire = format_pass( + sat_name="ISS", max_el=75.3, + aos_epoch=int(aos_dt.timestamp()), + los_epoch=int(los_dt.timestamp()), + aos_compass="SW", los_compass="NE", + broadcast=False, + ) + + assert "max 75\u00B0" 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 zoneinfo import ZoneInfo + + tz = ZoneInfo("America/Boise") + aos_dt = datetime(2026, 6, 12, 20, 38, 0, tzinfo=tz) + los_dt = datetime(2026, 6, 12, 20, 44, 0, tzinfo=tz) + + wire = format_pass( + sat_name="ISS", max_el=75.0, + aos_epoch=int(aos_dt.timestamp()), + los_epoch=int(los_dt.timestamp()), + aos_compass="SW", los_compass="NE", + broadcast=False, + ) + + assert "\n" not in wire + + +# ══════════════════════════════════════════════════════════════════════ +# REGISTRY completeness +# ══════════════════════════════════════════════════════════════════════ + +class TestRegistryKeys: + """Verify all new adapter_config keys exist.""" + + def test_max_broadcasts_per_hour_in_registry(self): + from meshai.adapter_config.defaults import REGISTRY + spec = REGISTRY[("satpass", "max_broadcasts_per_hour")] + assert spec["default"] == 4 + assert spec["type"] == "int" + + def test_dry_run_in_registry(self): + from meshai.adapter_config.defaults import REGISTRY + spec = REGISTRY[("satpass", "dry_run")] + assert spec["default"] is True + assert spec["type"] == "bool" + + def test_norad_ids_description_says_opt_in(self): + from meshai.adapter_config.defaults import REGISTRY + spec = REGISTRY[("satpass", "norad_ids")] + assert "broadcast nothing" in spec["description"].lower() or \ + "opt-in" in spec["description"].lower() diff --git a/tests/test_satpass_handler.py b/tests/test_satpass_handler.py index ef84449..a2561f0 100644 --- a/tests/test_satpass_handler.py +++ b/tests/test_satpass_handler.py @@ -6,7 +6,8 @@ from unittest.mock import MagicMock, patch 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"): + 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", @@ -25,6 +26,8 @@ def _envelope(norad_id=25544, sat_name="ISS", observer="Boise", "aos_time": aos, "los_time": los, "azimuth_at_peak_compass": direction, + "azimuth_at_aos_compass": aos_compass, + "azimuth_at_los_compass": los_compass, } } } @@ -47,12 +50,16 @@ def mock_adapter_config(): cfg.enabled = True cfg.observers = [] # empty = all observers cfg.min_elevation = 30 - cfg.norad_ids = [] # empty = all satellites + cfg.norad_ids = [25544] # must be non-empty for opt-in + cfg.dry_run = False + cfg.max_broadcasts_per_hour = 4 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 @@ -67,9 +74,7 @@ class TestSatpassHandler: result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) assert result is not None - assert "ISS Pass" in result - assert "75" in result - assert "Boise" in result + assert "ISS" in result def test_low_elevation_pass_filtered(self, mock_db, mock_adapter_config): """A pass below min_elevation should be filtered.""" @@ -129,19 +134,20 @@ class TestSatpassHandler: assert result2 is None def test_wire_format(self, mock_db, mock_adapter_config): - """Wire format should have 3 lines with correct info.""" + """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") + 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) lines = result.split("\n") - assert len(lines) == 3 - assert "ISS Pass" in lines[0] - assert "75" in lines[0] - assert "AOS" in lines[1] - assert "LOS" in lines[1] - assert "Boise" in lines[2] + 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] def test_commit_callback_attached(self, mock_db, mock_adapter_config): """Broadcast should attach commit callback.""" diff --git a/tests/test_satpass_wire_fields.py b/tests/test_satpass_wire_fields.py index 466e1d8..2c51feb 100644 --- a/tests/test_satpass_wire_fields.py +++ b/tests/test_satpass_wire_fields.py @@ -66,7 +66,7 @@ NOAA18_ENVELOPE = { } -def _enable_satpass(): +def _enable_satpass(norad_ids=None): """Set satpass.enabled=true in the test DB.""" from meshai.persistence import get_db from meshai.adapter_config import invalidate_cache @@ -80,6 +80,19 @@ def _enable_satpass(): "UPDATE adapter_config SET value_json='5' " "WHERE adapter='satpass' AND key='min_elevation'" ) + # Disable dry_run for tests that expect wire output + conn.execute( + "UPDATE adapter_config SET value_json='false' " + "WHERE adapter='satpass' AND key='dry_run'" + ) + # Set norad_ids (must be non-empty for opt-in) + if norad_ids is None: + norad_ids = [28654] + conn.execute( + "UPDATE adapter_config SET value_json=? " + "WHERE adapter='satpass' AND key='norad_ids'", + (json.dumps(norad_ids),) + ) invalidate_cache() @@ -93,6 +106,8 @@ def test_noaa18_envelope_produces_satpass_event(): _enable_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 now = int(time.time()) wire = handle_satpass( @@ -123,12 +138,14 @@ def test_noaa18_envelope_produces_satpass_event(): def test_noaa18_wire_message_format(): - """Wire message includes satellite name, elevation, observer, direction.""" + """Wire message includes satellite name, direction in new 2-line format.""" from meshai.central.satpass_handler import handle_satpass _enable_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 wire = handle_satpass( NOAA18_ENVELOPE, @@ -138,17 +155,17 @@ def test_noaa18_wire_message_format(): ) assert wire is not None - # Line 1: satellite name + elevation - assert "NOAA 18" in wire - assert "22" in wire # int(22.69) = 22 + lines = wire.split("\n") + assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}" - # Line 2: AOS/LOS times - assert "AOS" in wire - assert "LOS" in wire + # Line 1: satellite name + bucket + compass directions + 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 - # Line 3: observer + azimuth direction - assert "Filer" in wire - assert "ENE" in wire # azimuth_at_peak_compass + # Line 2: duration + time window + assert "minute window" in lines[1] def test_missing_norad_id_rejected(): @@ -158,6 +175,8 @@ def test_missing_norad_id_rejected(): _enable_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 # Deep copy and remove norad_id import copy @@ -180,6 +199,8 @@ def test_missing_max_elevation_deg_rejected(): _enable_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 import copy env = copy.deepcopy(NOAA18_ENVELOPE) @@ -201,6 +222,8 @@ def test_observer_fallback_to_slug(): _enable_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 import copy env = copy.deepcopy(NOAA18_ENVELOPE) @@ -214,7 +237,15 @@ def test_observer_fallback_to_slug(): now=int(time.time()), ) assert wire is not None - assert "filer" in wire + # Observer name stored in DB, not in broadcast wire format + from meshai.persistence import get_db + conn = get_db() + row = conn.execute( + "SELECT observer FROM satpass_events WHERE norad_id=28654" + ).fetchone() + assert row is not None + assert row["observer"] == "filer" + # ── Consumer category mapping ──────────────────────────────────────