potato-mesh/tests/test_waypoint_unit.py

# Copyright © 2025-26 l5yth & contributors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Unit tests for the ``WAYPOINT_APP`` handler (SPEC W1/W2, acceptance WP-A1).

Covers the decode → normalise → queue path of
:func:`data.mesh_ingestor.handlers.waypoint.store_waypoint_packet`, the
``locked_to`` canonicalisation helper, and the ``store_packet_dict``
dispatch seam that routes waypoint packets to the handler.
"""

from __future__ import annotations

import base64
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

import data.mesh_ingestor.config as config
import data.mesh_ingestor.handlers as handlers
import data.mesh_ingestor.handlers.generic as generic_mod
import data.mesh_ingestor.handlers.waypoint as waypoint_mod
import data.mesh_ingestor.queue as q


@pytest.fixture()
def sent(monkeypatch):
    """Capture queued POSTs as ``(path, payload, priority)`` tuples."""

    captured = []
    monkeypatch.setattr(
        q,
        "_queue_post_json",
        lambda path, payload, *, priority, **kw: captured.append(
            (path, payload, priority)
        ),
    )
    return captured


def _make_packet(**extra):
    """Build a minimal WAYPOINT_APP packet with a decoded waypoint section."""

    pkt = {
        "id": 555_001,
        "rxTime": 1_700_000_000,
        "fromId": "!3769b133",
        "decoded": {
            "portnum": "WAYPOINT_APP",
            "waypoint": {
                "id": 41_206,
                "name": "Tempelhofer Feld",
                "description": "See further",
                "icon": 0x2708,
                "latitude": 52.4751642,
                "longitude": 13.4029586,
                "expire": 1_700_010_000,
                "lockedTo": 0x3769B133,
            },
        },
    }
    pkt.update(extra)
    return pkt


class TestCanonicalLockedTo:
    """Tests for :func:`waypoint._canonical_locked_to` (C3 mapping)."""

    def test_maps_node_num_to_canonical_id(self):
        """A non-zero node number maps to the canonical ``!%08x`` id."""
        assert waypoint_mod._canonical_locked_to(0x3769B133) == "!3769b133"

    def test_normalises_canonical_string(self):
        """An already-canonical id string is normalised to lowercase."""
        assert waypoint_mod._canonical_locked_to("!3769B133") == "!3769b133"

    def test_zero_means_unlocked(self):
        """The Meshtastic ``0`` sentinel yields ``None`` (not locked)."""
        assert waypoint_mod._canonical_locked_to(0) is None
        assert waypoint_mod._canonical_locked_to("0") is None

    def test_nil_and_garbage_yield_none(self):
        """``None`` and unparseable references are treated as unlocked."""
        assert waypoint_mod._canonical_locked_to(None) is None
        assert waypoint_mod._canonical_locked_to("not a node") is None


class TestStoreWaypointPacket:
    """Tests for :func:`handlers.store_waypoint_packet`."""

    def test_queues_full_waypoint_payload(self, sent):
        """A valid packet queues the documented contract shape (WP-A1)."""
        pkt = _make_packet()
        handlers.store_waypoint_packet(pkt, pkt["decoded"])

        assert len(sent) == 1
        path, payload, priority = sent[0]
        assert path == "/api/waypoints"
        assert priority == q._WAYPOINT_POST_PRIORITY
        assert payload["id"] == 41_206
        assert payload["node_id"] == "!3769b133"
        assert payload["node_num"] == 0x3769B133
        assert payload["from_id"] == "!3769b133"
        assert payload["rx_time"] == 1_700_000_000
        assert payload["rx_iso"].startswith("2023-")
        assert payload["name"] == "Tempelhofer Feld"
        assert payload["description"] == "See further"
        assert payload["icon"] == 0x2708
        assert payload["latitude"] == pytest.approx(52.4751642)
        assert payload["longitude"] == pytest.approx(13.4029586)
        assert payload["expire"] == 1_700_010_000
        assert payload["locked_to"] == "!3769b133"
        assert payload["protocol"] == config.PROTOCOL

    def test_ignores_packet_without_waypoint_id(self, sent):
        """A packet whose waypoint section lacks an id is dropped (ignored)."""
        pkt = _make_packet()
        pkt["decoded"]["waypoint"].pop("id")
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert sent == []

    def test_ignores_packet_without_author(self, sent):
        """A packet without a resolvable author node id is dropped."""
        pkt = _make_packet()
        pkt.pop("fromId")
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert sent == []

    def test_tolerates_missing_waypoint_section(self, sent):
        """A decoded mapping without a waypoint section is dropped, not raised."""
        handlers.store_waypoint_packet({"fromId": "!3769b133"}, {})
        assert sent == []

    def test_integer_scaled_coordinates(self, sent):
        """latitudeI/longitudeI (1e-7 degrees) convert to float degrees."""
        pkt = _make_packet()
        section = pkt["decoded"]["waypoint"]
        section.pop("latitude")
        section.pop("longitude")
        section["latitudeI"] = int(52.5 * 1e7)
        section["longitude_i"] = int(13.4 * 1e7)
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        payload = sent[0][1]
        assert payload["latitude"] == pytest.approx(52.5)
        assert payload["longitude"] == pytest.approx(13.4)

    def test_collapses_null_island_sentinel(self, sent):
        """The paired (0, 0) no-fix sentinel collapses to None/None (#782)."""
        pkt = _make_packet()
        pkt["decoded"]["waypoint"]["latitude"] = 0.0
        pkt["decoded"]["waypoint"]["longitude"] = 0.0
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        payload = sent[0][1]
        assert payload["latitude"] is None
        assert payload["longitude"] is None

    def test_expire_zero_means_never(self, sent):
        """``expire = 0`` is normalised to None (never expires, W5)."""
        pkt = _make_packet()
        pkt["decoded"]["waypoint"]["expire"] = 0
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert sent[0][1]["expire"] is None

    def test_absent_expire_and_lock_are_none(self, sent):
        """Absent expire/lockedTo fields are omitted as None."""
        pkt = _make_packet()
        pkt["decoded"]["waypoint"].pop("expire")
        pkt["decoded"]["waypoint"].pop("lockedTo")
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        payload = sent[0][1]
        assert payload["expire"] is None
        assert payload["locked_to"] is None

    def test_blank_name_and_description_become_none(self, sent):
        """Empty strings are normalised to None, non-strings coerced to str."""
        pkt = _make_packet()
        pkt["decoded"]["waypoint"]["name"] = ""
        pkt["decoded"]["waypoint"]["description"] = 123
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        payload = sent[0][1]
        assert payload["name"] is None
        assert payload["description"] == "123"

    def test_uncoercible_rx_time_falls_back_to_now(self, sent):
        """A non-numeric rxTime falls back to the current clock."""
        pkt = _make_packet(rxTime="not-a-time")
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert sent[0][1]["rx_time"] > 1_700_000_000

    def test_packet_protocol_stamp_wins(self, sent):
        """An explicit per-packet protocol stamp overrides the config default."""
        pkt = _make_packet(protocol="meshcore")
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert sent[0][1]["protocol"] == "meshcore"

    def test_radio_metadata_and_rf_fields(self, sent, monkeypatch):
        """RF fields ride along and radio metadata is applied when configured."""
        monkeypatch.setattr(config, "LORA_FREQ", 868)
        monkeypatch.setattr(config, "MODEM_PRESET", "MediumFast")
        pkt = _make_packet(snr=-8.5, rssi=-90, hopLimit=3)
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        payload = sent[0][1]
        assert payload["snr"] == pytest.approx(-8.5)
        assert payload["rssi"] == -90
        assert payload["hop_limit"] == 3
        assert payload["lora_freq"] == 868
        assert payload["modem_preset"] == "MediumFast"

    def test_payload_bytes_are_base64_encoded(self, sent):
        """Raw decoded payload bytes are forwarded Base64-encoded."""
        pkt = _make_packet()
        pkt["decoded"]["payload"] = b"\x01\x02"
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert sent[0][1]["payload_b64"] == base64.b64encode(b"\x01\x02").decode(
            "ascii"
        )

    def test_debug_log_branch(self, sent, monkeypatch):
        """The DEBUG branch emits a structured debug log line."""
        logged = []
        monkeypatch.setattr(config, "DEBUG", True)
        monkeypatch.setattr(
            config, "_debug_log", lambda msg, **kw: logged.append((msg, kw))
        )
        pkt = _make_packet()
        handlers.store_waypoint_packet(pkt, pkt["decoded"])
        assert len(sent) == 1
        assert any("waypoint" in msg.lower() for msg, _ in logged)


class TestStorePacketDictWaypointDispatch:
    """Dispatch seam: ``store_packet_dict`` routes waypoints to the handler."""

    def _capture(self, monkeypatch):
        routed = []
        monkeypatch.setattr(
            generic_mod,
            "store_waypoint_packet",
            lambda packet, decoded: routed.append((packet, decoded)),
        )
        return routed

    def test_routes_by_string_portnum(self, sent, monkeypatch):
        """A WAYPOINT_APP portnum string routes to the waypoint handler."""
        routed = self._capture(monkeypatch)
        handlers.store_packet_dict({"decoded": {"portnum": "WAYPOINT_APP"}})
        assert len(routed) == 1

    def test_routes_by_decoded_section(self, sent, monkeypatch):
        """A decoded ``waypoint`` section routes even without a portnum."""
        routed = self._capture(monkeypatch)
        handlers.store_packet_dict({"decoded": {"waypoint": {"id": 1}}})
        assert len(routed) == 1

    def test_routes_by_integer_portnum(self, sent, monkeypatch):
        """The numeric WAYPOINT_APP portnum (8) routes via the enum lookup."""
        routed = self._capture(monkeypatch)
        candidates = generic_mod._portnum_candidates("WAYPOINT_APP")
        if not candidates:
            pytest.skip("meshtastic protobuf enums unavailable")
        handlers.store_packet_dict({"decoded": {"portnum": next(iter(candidates))}})
        assert len(routed) == 1

    def test_non_waypoint_packet_is_not_routed(self, sent, monkeypatch):
        """A plain text message never reaches the waypoint handler."""
        routed = self._capture(monkeypatch)
        handlers.store_packet_dict(
            {"fromId": "!3769b133", "decoded": {"portnum": "TEXT_MESSAGE_APP"}}
        )
        assert routed == []