meshai/work/tests/test_satpass_handler.py

"""v0.7 satpass_handler tests."""

import pytest
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",
              aos_compass="SW", los_compass="NE"):
    """Build a CloudEvents envelope for a satellite pass."""
    return {
        "specversion": "1.0",
        "type": "central.sat.pass",
        "source": "central",
        "id": f"pass-{norad_id}-{aos}",
        "data": {
            "adapter": "n2yo_visualpasses",
            "category": "pass.n2yo_visualpasses",
            "severity": 0,
            "data": {
                "norad_id": norad_id,
                "satellite_name": sat_name,
                "observer_name": observer,
                "max_elevation_deg": max_el,
                "aos_time": aos,
                "los_time": los,
                "azimuth_at_peak_compass": direction,
                "azimuth_at_aos_compass": aos_compass,
                "azimuth_at_los_compass": los_compass,
            }
        }
    }


@pytest.fixture
def mock_db():
    """Mock database connection."""
    conn = MagicMock()
    conn.execute.return_value.fetchone.return_value = None
    conn.execute.return_value.lastrowid = 1
    with patch("meshai.central.satpass_handler.get_db", return_value=conn):
        yield conn


@pytest.fixture
def mock_adapter_config():
    """Mock adapter_config.satpass."""
    cfg = MagicMock()
    cfg.enabled = True
    cfg.observers = []  # empty = all observers
    cfg.min_elevation = 30
    cfg.norad_ids = [25544]  # must be non-empty for opt-in
    cfg.dry_run = False
    cfg.max_broadcasts_per_hour = 4
    cfg.max_aos_horizon_hours = 24
    with patch("meshai.central.satpass_handler.adapter_config") as mock:
        mock.satpass = cfg
        from meshai.central.satpass_handler import handle_satpass
        if hasattr(handle_satpass, "_disabled_logged"):
            del handle_satpass._disabled_logged
        if hasattr(handle_satpass, "_no_norad_ids_logged"):
            del handle_satpass._no_norad_ids_logged
        yield cfg


def _ingest_and_consolidate(env, subject, *, now, data=None):
    """Drive the two-call async satpass contract (ingest -> consolidate).

    handle_satpass() ingests the pass into satpass_pending and ALWAYS
    returns None; the consumer then runs consolidate_satpass_pending(),
    which returns the (wire, data) to broadcast (or None if suppressed).
    Returns the consolidation result so a test can assert on the real wire.
    """
    from meshai.central.satpass_handler import (
        handle_satpass, consolidate_satpass_pending,
        drain_pending_consolidation_ids)
    drain_pending_consolidation_ids()
    assert handle_satpass(
        env, subject, data=data if data is not None else {}, now=now) is None
    for cid in drain_pending_consolidation_ids():
        res = consolidate_satpass_pending(cid)
        if res is not None:
            return res
    return None


class TestSatpassHandler:
    """Tests for handle_satpass function."""

    def test_high_elevation_pass_broadcasts(self, mock_adapter_config):
        """A pass with high elevation should broadcast (via consolidation)."""
        env = _envelope(max_el=75.0)
        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
                                         now=1781235000)
        assert result is not None
        wire, _ = result
        assert "ISS" in wire

    def test_low_elevation_pass_filtered(self, mock_db, mock_adapter_config):
        """A pass below min_elevation should be filtered."""
        from meshai.central.satpass_handler import handle_satpass

        mock_adapter_config.min_elevation = 30
        env = _envelope(max_el=25.0)
        result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)

        assert result is None

    def test_observer_filter_blocks_mismatch(self, mock_db, mock_adapter_config):
        """A pass for non-configured observer should be filtered."""
        from meshai.central.satpass_handler import handle_satpass

        mock_adapter_config.observers = ["Magic Valley"]
        env = _envelope(observer="Boise")
        result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)

        assert result is None

    def test_observer_filter_allows_match(self, mock_adapter_config):
        """A pass for configured observer should broadcast."""
        mock_adapter_config.observers = ["Boise", "Magic Valley"]
        env = _envelope(observer="Boise", max_el=45.0)
        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
                                         now=1781235000)
        assert result is not None

    def test_norad_id_filter(self, mock_db, mock_adapter_config):
        """NORAD ID filter should block non-matching satellites."""
        from meshai.central.satpass_handler import handle_satpass

        mock_adapter_config.norad_ids = [25544]  # ISS only
        env = _envelope(norad_id=12345, max_el=60.0)
        result = handle_satpass(env, "central.sat.pass.other", data={}, now=1718163120)

        assert result is None

    def test_dedup_blocks_second_broadcast(self, mock_adapter_config):
        """Second pass in same hour bucket should be deduplicated."""
        env = _envelope(max_el=60.0)

        # First pass consolidates into a real broadcast.
        result1 = _ingest_and_consolidate(env, "central.sat.pass.iss",
                                          now=1781235000)
        assert result1 is not None

        # Simulate the broadcast committing (marks satpass_events broadcast).
        _wire, data1 = result1
        data1["_on_broadcast_committed"](1781235010)

        # Second pass in the same (norad, aos-hour) bucket must be deduped.
        result2 = _ingest_and_consolidate(env, "central.sat.pass.iss",
                                          now=1781235060)
        assert result2 is None

    def test_wire_format(self, mock_adapter_config):
        """Wire is a single clean line: short name, degrees, collapsed compass."""
        env = _envelope(sat_name="ISS", max_el=75, observer="Boise",
                        direction="S", aos_compass="SW", los_compass="NE")
        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
                                         now=1781235000)
        assert result is not None
        wire, _ = result

        assert "\n" not in wire          # single line now
        assert "ISS" in wire
        assert "max 75°" in wire    # numeric elevation, not a bucket word
        assert "overhead" not in wire
        assert "min window" not in wire
        assert "SW→S→NE" in wire  # aos -> peak -> los, collapsed

    def test_commit_callback_attached(self, mock_adapter_config):
        """Broadcast should attach commit callback (on the consolidation data)."""
        env = _envelope(max_el=60.0)
        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
                                         now=1781235000)
        assert result is not None
        _wire, data = result
        assert "_on_broadcast_committed" in data
        assert "_broadcast_audit" in data
        assert data["_broadcast_audit"]["table"] == "satpass_events"

    def test_wrong_adapter_ignored(self, mock_db, mock_adapter_config):
        """Envelope with wrong adapter should be ignored."""
        from meshai.central.satpass_handler import handle_satpass

        env = _envelope()
        env["data"]["adapter"] = "some_other_adapter"
        result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)

        assert result is None


# ============================================================================
# Budget-fit SAFETY CAP: a pathologically long satellite name must not push
# the broadcast string past 140 chars. Calls format_pass directly (bypasses
# the broken consolidation path).
# ============================================================================

def test_format_pass_worst_case_fits_140():
    from meshai.central.satpass_handler import format_pass
    wire = format_pass(
        sat_name=("NOAA-19 EXPERIMENTAL SUPER LONG SATELLITE DESIGNATION "
                  "PAYLOAD REVISION X PROTOTYPE FLIGHT MODEL SERIAL 00042"),
        max_el=78.0,
        aos_epoch=1719900000, los_epoch=1719900780,
        aos_compass="NNW", los_compass="SSE",
        entry_observer="Treasure Valley Observatory West Ridge Site",
        exit_observer="Magic Valley Observatory East Rim Station",
        broadcast=True,
    )
    assert len(wire) <= 140, f"{len(wire)} chars:\n{wire!r}"