# 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 :mod:`data.mesh_ingestor.events`.""" from __future__ import annotations 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)) from data.mesh_ingestor.events import ( # noqa: E402 - path setup IngestorHeartbeat, MessageEvent, NeighborEntry, NeighborsSnapshot, PositionEvent, TelemetryEvent, TraceEvent, ) def test_message_event_schema(): assert MessageEvent.__required_keys__ == frozenset({"id", "rx_time", "rx_iso"}) assert "text" in MessageEvent.__optional_keys__ assert "from_id" in MessageEvent.__optional_keys__ assert "snr" in MessageEvent.__optional_keys__ assert "rssi" in MessageEvent.__optional_keys__ def test_message_event_requires_id_rx_time_rx_iso(): event: MessageEvent = { "id": 1, "rx_time": 1700000000, "rx_iso": "2023-11-14T00:00:00Z", } assert event["id"] == 1 assert event["rx_time"] == 1700000000 assert event["rx_iso"] == "2023-11-14T00:00:00Z" def test_message_event_accepts_optional_fields(): event: MessageEvent = { "id": 2, "rx_time": 1700000001, "rx_iso": "2023-11-14T00:00:01Z", "text": "hello", "from_id": "!aabbccdd", "snr": 4.5, "rssi": -90, } assert event["text"] == "hello" assert event["snr"] == pytest.approx(4.5) def test_position_event_schema(): assert PositionEvent.__required_keys__ == frozenset({"id", "rx_time", "rx_iso"}) assert "latitude" in PositionEvent.__optional_keys__ assert "longitude" in PositionEvent.__optional_keys__ assert "node_id" in PositionEvent.__optional_keys__ def test_position_event_required_fields(): event: PositionEvent = { "id": 10, "rx_time": 1700000002, "rx_iso": "2023-11-14T00:00:02Z", } assert event["id"] == 10 def test_position_event_optional_fields(): event: PositionEvent = { "id": 11, "rx_time": 1700000003, "rx_iso": "2023-11-14T00:00:03Z", "latitude": 37.7749, "longitude": -122.4194, "altitude": 10.0, "node_id": "!aabbccdd", } assert event["latitude"] == pytest.approx(37.7749) def test_telemetry_event_schema(): assert TelemetryEvent.__required_keys__ == frozenset({"id", "rx_time", "rx_iso"}) assert "payload_b64" in TelemetryEvent.__optional_keys__ assert "snr" in TelemetryEvent.__optional_keys__ def test_telemetry_event_required_fields(): event: TelemetryEvent = { "id": 20, "rx_time": 1700000004, "rx_iso": "2023-11-14T00:00:04Z", } assert event["id"] == 20 def test_telemetry_event_optional_fields(): event: TelemetryEvent = { "id": 21, "rx_time": 1700000005, "rx_iso": "2023-11-14T00:00:05Z", "channel": 0, "payload_b64": "AAEC", "snr": 3.0, } assert event["payload_b64"] == "AAEC" def test_neighbor_entry_schema(): assert NeighborEntry.__required_keys__ == frozenset({"rx_time", "rx_iso"}) assert "neighbor_id" in NeighborEntry.__optional_keys__ assert "snr" in NeighborEntry.__optional_keys__ def test_neighbor_entry_required_fields(): entry: NeighborEntry = {"rx_time": 1700000006, "rx_iso": "2023-11-14T00:00:06Z"} assert entry["rx_time"] == 1700000006 def test_neighbor_entry_optional_fields(): entry: NeighborEntry = { "rx_time": 1700000007, "rx_iso": "2023-11-14T00:00:07Z", "neighbor_id": "!11223344", "snr": 6.0, } assert entry["neighbor_id"] == "!11223344" def test_neighbors_snapshot_schema(): assert NeighborsSnapshot.__required_keys__ == frozenset( {"node_id", "rx_time", "rx_iso"} ) assert "neighbors" in NeighborsSnapshot.__optional_keys__ assert "node_broadcast_interval_secs" in NeighborsSnapshot.__optional_keys__ def test_neighbors_snapshot_required_fields(): snap: NeighborsSnapshot = { "node_id": "!aabbccdd", "rx_time": 1700000008, "rx_iso": "2023-11-14T00:00:08Z", } assert snap["node_id"] == "!aabbccdd" def test_neighbors_snapshot_optional_fields(): snap: NeighborsSnapshot = { "node_id": "!aabbccdd", "rx_time": 1700000009, "rx_iso": "2023-11-14T00:00:09Z", "neighbors": [], "node_broadcast_interval_secs": 900, } assert snap["node_broadcast_interval_secs"] == 900 def test_trace_event_schema(): assert TraceEvent.__required_keys__ == frozenset({"hops", "rx_time", "rx_iso"}) assert "elapsed_ms" in TraceEvent.__optional_keys__ assert "snr" in TraceEvent.__optional_keys__ def test_trace_event_required_fields(): event: TraceEvent = { "hops": [1, 2, 3], "rx_time": 1700000010, "rx_iso": "2023-11-14T00:00:10Z", } assert event["hops"] == [1, 2, 3] def test_trace_event_optional_fields(): event: TraceEvent = { "hops": [4, 5], "rx_time": 1700000011, "rx_iso": "2023-11-14T00:00:11Z", "elapsed_ms": 42, "snr": 2.5, } assert event["elapsed_ms"] == 42 def test_ingestor_heartbeat_schema(): # IngestorHeartbeat uses total=True with NotRequired fields. Under # `from __future__ import annotations` the TypedDict metaclass cannot # evaluate the annotation strings at class creation time, so # NotRequired keys appear in __required_keys__ rather than # __optional_keys__. Verify the four always-present keys are included. always_required = {"node_id", "start_time", "last_seen_time", "version"} assert always_required <= IngestorHeartbeat.__required_keys__ def test_ingestor_heartbeat_all_fields(): hb: IngestorHeartbeat = { "node_id": "!aabbccdd", "start_time": 1700000000, "last_seen_time": 1700000012, "version": "0.5.12", "lora_freq": 906875, "modem_preset": "LONG_FAST", } assert hb["version"] == "0.5.12" assert hb["lora_freq"] == 906875 def test_ingestor_heartbeat_without_optional_fields(): hb: IngestorHeartbeat = { "node_id": "!aabbccdd", "start_time": 1700000000, "last_seen_time": 1700000013, "version": "0.5.12", } assert "lora_freq" not in hb