matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

fix(satpass): broadcast safety controls — opt-in filter, rate cap, dry-run, wire format

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Claude 2026-06-12 23:00:03 +00:00
commit e903444356
8 changed files with 704 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/test_adapter_config_api.py
View file

@ -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):

2
tests/test_adapter_config_foundation.py
View file

@ -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."
)

453
tests/test_satpass_broadcast_safety.py Normal file
View file

@ -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()

32
tests/test_satpass_handler.py
View file

@ -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."""

55
tests/test_satpass_wire_fields.py
View file

@ -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 ──────────────────────────────────────