From 5adbe2263e39b622d91cff0c916808f9726f3a5b Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sat, 28 Mar 2026 18:46:21 +0100 Subject: [PATCH] data: resolve circular dependency of deamon.py (#653) * data: resolve circular dependency of deamon.py * address review comments * address review comments * address review comments --- AGENTS.md => CLAUDE.md | 14 +- data/mesh_ingestor/daemon.py | 631 ++++++++++++--------- data/mesh_ingestor/events.py | 51 +- data/mesh_ingestor/node_identity.py | 15 +- data/mesh_ingestor/provider.py | 6 +- data/mesh_ingestor/providers/__init__.py | 1 - data/mesh_ingestor/providers/meshtastic.py | 49 +- tests/test_daemon_unit.py | 388 +++++++++++++ tests/test_events_unit.py | 232 ++++++++ tests/test_node_identity_unit.py | 20 + tests/test_provider_unit.py | 81 ++- 11 files changed, 1154 insertions(+), 334 deletions(-) rename AGENTS.md => CLAUDE.md (65%) create mode 100644 tests/test_events_unit.py diff --git a/AGENTS.md b/CLAUDE.md similarity index 65% rename from AGENTS.md rename to CLAUDE.md index 936e914..0057633 100644 --- a/AGENTS.md +++ b/CLAUDE.md @@ -15,7 +15,7 @@ Run linters for Python (`black`) and Ruby (`rufo`) to ensure consistent code for ## Project Structure & Module Organization The repository splits runtime and ingestion logic. `web/` holds the Sinatra dashboard (Ruby code in `lib/potato_mesh`, views in `views/`, static bundles in `public/`). -`data/` hosts the Python Meshtastic ingestor plus migrations and CLI scripts. API fixtures and end-to-end harnesses live in `tests/`. Dockerfiles and compose files support containerized workflows. +`data/` hosts the Python Meshtastic ingestor plus migrations and CLI scripts. The ingestor is structured as the `data/mesh_ingestor/` package with the following key modules: `daemon.py` (main loop), `handlers.py` (packet processing), `interfaces.py` (interface helpers), `config.py` (env-driven config), `events.py` (TypedDict event schemas), `provider.py` (Provider protocol), `node_identity.py` (canonical node ID utilities), `decode_payload.py` (CLI protobuf decoder), and the `providers/` subpackage (currently `meshtastic.py`). API contracts for all POST ingest routes are documented in `data/mesh_ingestor/CONTRACTS.md`. API fixtures and end-to-end harnesses live in `tests/`. Dockerfiles and compose files support containerized workflows. `matrix/` contains the Rust Matrix bridge; build with `cargo build --release` or `docker build -f matrix/Dockerfile .`, and keep bridge config under `matrix/Config.toml` when running locally. @@ -39,7 +39,17 @@ Install dependencies with `cd app && flutter pub get`; format with `dart format ## Testing Guidelines Ruby specs run with `cd web && bundle exec rspec`, producing SimpleCov output in `coverage/`. Front-end behaviour is verified through Node’s test runner: `cd web && npm test` writes V8 coverage and JUnit XML under `reports/`. -The ingestion layer is guarded by `pytest -q tests/test_mesh.py`; leave fixtures in `tests/` untouched so CI can replay them. New features should ship with matching specs and updated integration checks. +The ingestion layer is tested with `pytest -q tests/`; leave fixtures in `tests/` untouched so CI can replay them. The suite includes both integration tests (`test_mesh.py`) and focused unit tests — `test_events_unit.py` (TypedDict schemas), `test_provider_unit.py` (Provider protocol conformance and `MeshtasticProvider`), `test_node_identity_unit.py` (canonical ID helpers), `test_daemon_unit.py`, `test_serialization_unit.py`, and `test_decode_payload.py`. New features should ship with matching specs and updated integration checks. + +## Adding a New Ingestor Provider +The `data/mesh_ingestor/provider.py` module defines a `@runtime_checkable` `Provider` Protocol with five members: `name` (str), `subscribe()`, `connect(*, active_candidate)`, `extract_host_node_id(iface)`, and `node_snapshot_items(iface)`. To add a new backend (e.g. Reticulum, MeshCore): + +1. Create `data/mesh_ingestor/providers/.py` with a class satisfying the Protocol. +2. Register it in `data/mesh_ingestor/providers/__init__.py`. +3. Pass an instance via `daemon.main(provider=...)` or make it the default in `main()`. +4. Cover the provider with unit tests in `tests/test_provider_unit.py` — at minimum an `isinstance(..., Provider)` conformance check and any retry/error-handling paths. + +Consult `data/mesh_ingestor/CONTRACTS.md` for the canonical event shapes all providers must emit. ## Commit & Pull Request Guidelines Commits should stay imperative and reference issues the way history does (`Add chat log entries... (#408)`). Squash noisy work-in-progress commits before pushing. Pull requests need a concise summary, screenshots or curl traces for UI/API tweaks, and links to tracked issues. Paste the command output for the test suites you ran and mention configuration toggles (`API_TOKEN`, `PRIVATE`) reviewers must set. diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index 4df2fa2..0adb4f2 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -16,6 +16,7 @@ from __future__ import annotations +import dataclasses import inspect import signal import threading @@ -25,7 +26,6 @@ from pubsub import pub from . import config, handlers, ingestors, interfaces from .provider import Provider -from .providers.meshtastic import MeshtasticProvider _RECEIVE_TOPICS = ( "meshtastic.receive", @@ -199,11 +199,6 @@ def _process_ingestor_heartbeat(iface, *, ingestor_announcement_sent: bool) -> b if heartbeat_sent and not ingestor_announcement_sent: return True return ingestor_announcement_sent - iface_cls = getattr(iface_obj, "__class__", None) - if iface_cls is None: - return False - module_name = getattr(iface_cls, "__module__", "") or "" - return "ble_interface" in module_name def _connected_state(candidate) -> bool | None: @@ -245,10 +240,323 @@ def _connected_state(candidate) -> bool | None: return None -def main(existing_interface=None, *, provider: Provider | None = None) -> None: +# --------------------------------------------------------------------------- +# Loop state container +# --------------------------------------------------------------------------- + + +@dataclasses.dataclass +class _DaemonState: + """All mutable state for the :func:`main` daemon loop.""" + + provider: Provider + stop: threading.Event + configured_port: str | None + inactivity_reconnect_secs: float + energy_saving_enabled: bool + energy_online_secs: float + energy_sleep_secs: float + retry_delay: float + last_seen_packet_monotonic: float | None + active_candidate: str | None + + iface: object = None + resolved_target: str | None = None + initial_snapshot_sent: bool = False + energy_session_deadline: float | None = None + iface_connected_at: float | None = None + last_inactivity_reconnect: float | None = None + ingestor_announcement_sent: bool = False + announced_target: bool = False + + +# --------------------------------------------------------------------------- +# Per-iteration helpers (each returns True when the caller should `continue`) +# --------------------------------------------------------------------------- + + +def _advance_retry_delay(current: float) -> float: + """Return the next exponential-backoff retry delay.""" + + if config._RECONNECT_MAX_DELAY_SECS <= 0: + return current + # `current == 0` on the very first call (bootstrap); seed from config. + next_delay = current * 2 if current else config._RECONNECT_INITIAL_DELAY_SECS + return min(next_delay, config._RECONNECT_MAX_DELAY_SECS) + + +def _energy_sleep(state: _DaemonState, reason: str) -> None: + """Sleep for the configured energy-saving interval.""" + + if not state.energy_saving_enabled or state.energy_sleep_secs <= 0: + return + if config.DEBUG: + config._debug_log( + f"energy saving: {reason}; sleeping for {state.energy_sleep_secs:g}s" + ) + state.stop.wait(state.energy_sleep_secs) + + +def _try_connect(state: _DaemonState) -> bool: + """Attempt to establish the mesh interface. + + Returns: + ``True`` when connected and the loop should proceed; ``False`` when + the connection failed and the caller should ``continue``. + """ + + try: + state.iface, state.resolved_target, state.active_candidate = ( + state.provider.connect(active_candidate=state.active_candidate) + ) + handlers.register_host_node_id(state.provider.extract_host_node_id(state.iface)) + ingestors.set_ingestor_node_id(handlers.host_node_id()) + state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) + state.initial_snapshot_sent = False + if not state.announced_target and state.resolved_target: + config._debug_log( + "Using mesh interface", + context="daemon.interface", + severity="info", + target=state.resolved_target, + ) + state.announced_target = True + if state.energy_saving_enabled and state.energy_online_secs > 0: + state.energy_session_deadline = time.monotonic() + state.energy_online_secs + else: + state.energy_session_deadline = None + state.iface_connected_at = time.monotonic() + # Seed the inactivity tracking from the connection time so a + # reconnect is given a full inactivity window even when the + # handler still reports the previous packet timestamp. + state.last_seen_packet_monotonic = state.iface_connected_at + state.last_inactivity_reconnect = None + return True + except interfaces.NoAvailableMeshInterface as exc: + config._debug_log( + "No mesh interface available", + context="daemon.interface", + severity="error", + error_message=str(exc), + ) + _close_interface(state.iface) + raise SystemExit(1) from exc + except Exception as exc: + config._debug_log( + "Failed to create mesh interface", + context="daemon.interface", + severity="warn", + candidate=state.active_candidate or "auto", + error_class=exc.__class__.__name__, + error_message=str(exc), + ) + if state.configured_port is None: + state.active_candidate = None + state.announced_target = False + state.stop.wait(state.retry_delay) + state.retry_delay = _advance_retry_delay(state.retry_delay) + return False + + +def _check_energy_saving(state: _DaemonState) -> bool: + """Disconnect and sleep when energy-saving conditions are met. + + Returns: + ``True`` when the interface was closed and the caller should + ``continue``; ``False`` otherwise. + """ + + if not state.energy_saving_enabled or state.iface is None: + return False + + if ( + state.energy_session_deadline is not None + and time.monotonic() >= state.energy_session_deadline + ): + reason = "disconnected after session" + log_msg = "Energy saving disconnect" + elif ( + _is_ble_interface(state.iface) + and getattr(state.iface, "client", object()) is None + ): + reason = "BLE client disconnected" + log_msg = "Energy saving BLE disconnect" + else: + return False + config._debug_log(log_msg, context="daemon.energy", severity="info") + _close_interface(state.iface) + state.iface = None + state.announced_target = False + state.initial_snapshot_sent = False + state.energy_session_deadline = None + _energy_sleep(state, reason) + return True + + +def _try_send_snapshot(state: _DaemonState) -> bool: + """Send the initial node snapshot via the provider. + + Returns: + ``True`` when the snapshot succeeded (or no nodes exist yet); ``False`` + when a hard error occurred and the caller should ``continue``. + """ + + try: + node_items = state.provider.node_snapshot_items(state.iface) + processed_any = False + for node_id, node in node_items: + processed_any = True + try: + handlers.upsert_node(node_id, node) + except Exception as exc: + config._debug_log( + "Failed to update node snapshot", + context="daemon.snapshot", + severity="warn", + node_id=node_id, + error_class=exc.__class__.__name__, + error_message=str(exc), + ) + if config.DEBUG: + config._debug_log( + "Snapshot node payload", + context="daemon.snapshot", + node=node, + ) + if processed_any: + state.initial_snapshot_sent = True + return True + except Exception as exc: + config._debug_log( + "Snapshot refresh failed", + context="daemon.snapshot", + severity="warn", + error_class=exc.__class__.__name__, + error_message=str(exc), + ) + _close_interface(state.iface) + state.iface = None + state.stop.wait(state.retry_delay) + state.retry_delay = _advance_retry_delay(state.retry_delay) + return False + + +def _check_inactivity_reconnect(state: _DaemonState) -> bool: + """Reconnect when the interface has been silent for too long. + + Returns: + ``True`` when a reconnect was triggered and the caller should + ``continue``; ``False`` otherwise. + """ + + if state.iface is None or state.inactivity_reconnect_secs <= 0: + return False + + now = time.monotonic() + iface_activity = handlers.last_packet_monotonic() + + if ( + iface_activity is not None + and state.iface_connected_at is not None + and iface_activity < state.iface_connected_at + ): + iface_activity = state.iface_connected_at + + if iface_activity is not None and ( + state.last_seen_packet_monotonic is None + or iface_activity > state.last_seen_packet_monotonic + ): + state.last_seen_packet_monotonic = iface_activity + state.last_inactivity_reconnect = None + + latest_activity = iface_activity + if latest_activity is None and state.iface_connected_at is not None: + latest_activity = state.iface_connected_at + if latest_activity is None: + latest_activity = now + + inactivity_elapsed = now - latest_activity + believed_disconnected = ( + _connected_state(getattr(state.iface, "isConnected", None)) is False + ) + + if ( + not believed_disconnected + and inactivity_elapsed < state.inactivity_reconnect_secs + ): + return False + + if ( + state.last_inactivity_reconnect is not None + and now - state.last_inactivity_reconnect < state.inactivity_reconnect_secs + ): + return False + + reason = ( + "disconnected" + if believed_disconnected + else f"no data for {inactivity_elapsed:.0f}s" + ) + config._debug_log( + "Mesh interface inactivity detected", + context="daemon.interface", + severity="warn", + reason=reason, + ) + state.last_inactivity_reconnect = now + _close_interface(state.iface) + state.iface = None + state.announced_target = False + state.initial_snapshot_sent = False + state.energy_session_deadline = None + state.iface_connected_at = None + return True + + +# --------------------------------------------------------------------------- +# Loop iteration helper +# --------------------------------------------------------------------------- + + +def _loop_iteration(state: _DaemonState) -> bool: + """Execute one pass of the daemon main loop. + + Encapsulates the per-iteration ``continue`` decisions so that + :func:`main` stays within the allowed cognitive-complexity budget. + + Returns: + ``True`` when the loop should start the next iteration immediately + (equivalent to a ``continue``); ``False`` when the full pass + completed and the caller should sleep before iterating again. + """ + + if state.iface is None and not _try_connect(state): + return True + if _check_energy_saving(state): + return True + if not state.initial_snapshot_sent and not _try_send_snapshot(state): + return True + if _check_inactivity_reconnect(state): + return True + state.ingestor_announcement_sent = _process_ingestor_heartbeat( + state.iface, ingestor_announcement_sent=state.ingestor_announcement_sent + ) + state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) + return False + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def main(*, provider: Provider | None = None) -> None: """Run the mesh ingestion daemon until interrupted.""" - provider = provider or MeshtasticProvider() + if provider is None: + from .providers.meshtastic import MeshtasticProvider + + provider = MeshtasticProvider() subscribed = provider.subscribe() if subscribed: @@ -259,307 +567,72 @@ def main(existing_interface=None, *, provider: Provider | None = None) -> None: topics=subscribed, ) - iface = existing_interface - resolved_target = None - retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) - - stop = threading.Event() - initial_snapshot_sent = False - energy_session_deadline = None - iface_connected_at: float | None = None - last_seen_packet_monotonic = handlers.last_packet_monotonic() - last_inactivity_reconnect: float | None = None - inactivity_reconnect_secs = max( - 0.0, getattr(config, "_INACTIVITY_RECONNECT_SECS", 0.0) + state = _DaemonState( + provider=provider, + stop=threading.Event(), + configured_port=config.CONNECTION, + inactivity_reconnect_secs=max( + 0.0, getattr(config, "_INACTIVITY_RECONNECT_SECS", 0.0) + ), + energy_saving_enabled=config.ENERGY_SAVING, + energy_online_secs=max(0.0, config._ENERGY_ONLINE_DURATION_SECS), + energy_sleep_secs=max(0.0, config._ENERGY_SLEEP_SECS), + retry_delay=max(0.0, config._RECONNECT_INITIAL_DELAY_SECS), + last_seen_packet_monotonic=handlers.last_packet_monotonic(), + active_candidate=config.CONNECTION, ) - ingestor_announcement_sent = False - - energy_saving_enabled = config.ENERGY_SAVING - energy_online_secs = max(0.0, config._ENERGY_ONLINE_DURATION_SECS) - energy_sleep_secs = max(0.0, config._ENERGY_SLEEP_SECS) - - def _energy_sleep(reason: str) -> None: - if not energy_saving_enabled or energy_sleep_secs <= 0: - return - if config.DEBUG: - config._debug_log( - f"energy saving: {reason}; sleeping for {energy_sleep_secs:g}s" - ) - stop.wait(energy_sleep_secs) def handle_sigterm(*_args) -> None: - stop.set() + state.stop.set() def handle_sigint(signum, frame) -> None: - if stop.is_set(): + if state.stop.is_set(): signal.default_int_handler(signum, frame) return - stop.set() + state.stop.set() if threading.current_thread() == threading.main_thread(): signal.signal(signal.SIGINT, handle_sigint) signal.signal(signal.SIGTERM, handle_sigterm) - target = config.INSTANCE or "(no INSTANCE_DOMAIN configured)" - configured_port = config.CONNECTION - active_candidate = configured_port - announced_target = False config._debug_log( "Mesh daemon starting", context="daemon.main", severity="info", - target=target, - port=configured_port or "auto", + target=config.INSTANCE or "(no INSTANCE_DOMAIN configured)", + port=config.CONNECTION or "auto", channel=config.CHANNEL_INDEX, ) + try: - while not stop.is_set(): - if iface is None: - try: - iface, resolved_target, active_candidate = provider.connect( - active_candidate=active_candidate - ) - handlers.register_host_node_id( - provider.extract_host_node_id(iface) - ) - ingestors.set_ingestor_node_id(handlers.host_node_id()) - retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) - initial_snapshot_sent = False - if not announced_target and resolved_target: - config._debug_log( - "Using mesh interface", - context="daemon.interface", - severity="info", - target=resolved_target, - ) - announced_target = True - if energy_saving_enabled and energy_online_secs > 0: - energy_session_deadline = time.monotonic() + energy_online_secs - else: - energy_session_deadline = None - iface_connected_at = time.monotonic() - # Seed the inactivity tracking from the connection time so a - # reconnect is given a full inactivity window even when the - # handler still reports the previous packet timestamp. - last_seen_packet_monotonic = iface_connected_at - last_inactivity_reconnect = None - except interfaces.NoAvailableMeshInterface as exc: - config._debug_log( - "No mesh interface available", - context="daemon.interface", - severity="error", - error_message=str(exc), - ) - _close_interface(iface) - raise SystemExit(1) from exc - except Exception as exc: - candidate_desc = active_candidate or "auto" - config._debug_log( - "Failed to create mesh interface", - context="daemon.interface", - severity="warn", - candidate=candidate_desc, - error_class=exc.__class__.__name__, - error_message=str(exc), - ) - if configured_port is None: - active_candidate = None - announced_target = False - stop.wait(retry_delay) - if config._RECONNECT_MAX_DELAY_SECS > 0: - retry_delay = min( - ( - retry_delay * 2 - if retry_delay - else config._RECONNECT_INITIAL_DELAY_SECS - ), - config._RECONNECT_MAX_DELAY_SECS, - ) - continue - - if energy_saving_enabled and iface is not None: - if ( - energy_session_deadline is not None - and time.monotonic() >= energy_session_deadline - ): - config._debug_log( - "Energy saving disconnect", - context="daemon.energy", - severity="info", - ) - _close_interface(iface) - iface = None - announced_target = False - initial_snapshot_sent = False - energy_session_deadline = None - _energy_sleep("disconnected after session") - continue - if ( - _is_ble_interface(iface) - and getattr(iface, "client", object()) is None - ): - config._debug_log( - "Energy saving BLE disconnect", - context="daemon.energy", - severity="info", - ) - _close_interface(iface) - iface = None - announced_target = False - initial_snapshot_sent = False - energy_session_deadline = None - _energy_sleep("BLE client disconnected") - continue - - if not initial_snapshot_sent: - try: - node_items = list(provider.node_snapshot_items(iface)) - node_items = _node_items_snapshot(dict(node_items)) - if node_items is None: - config._debug_log( - "Skipping node snapshot due to concurrent modification", - context="daemon.snapshot", - ) - else: - processed_snapshot_item = False - for node_id, node in node_items: - processed_snapshot_item = True - try: - handlers.upsert_node(node_id, node) - except Exception as exc: - config._debug_log( - "Failed to update node snapshot", - context="daemon.snapshot", - severity="warn", - node_id=node_id, - error_class=exc.__class__.__name__, - error_message=str(exc), - ) - if config.DEBUG: - config._debug_log( - "Snapshot node payload", - context="daemon.snapshot", - node=node, - ) - if processed_snapshot_item: - initial_snapshot_sent = True - except Exception as exc: - config._debug_log( - "Snapshot refresh failed", - context="daemon.snapshot", - severity="warn", - error_class=exc.__class__.__name__, - error_message=str(exc), - ) - _close_interface(iface) - iface = None - stop.wait(retry_delay) - if config._RECONNECT_MAX_DELAY_SECS > 0: - retry_delay = min( - ( - retry_delay * 2 - if retry_delay - else config._RECONNECT_INITIAL_DELAY_SECS - ), - config._RECONNECT_MAX_DELAY_SECS, - ) - continue - - if iface is not None and inactivity_reconnect_secs > 0: - now_monotonic = time.monotonic() - iface_activity = handlers.last_packet_monotonic() - if ( - iface_activity is not None - and iface_connected_at is not None - and iface_activity < iface_connected_at - ): - iface_activity = iface_connected_at - if iface_activity is not None and ( - last_seen_packet_monotonic is None - or iface_activity > last_seen_packet_monotonic - ): - last_seen_packet_monotonic = iface_activity - last_inactivity_reconnect = None - - latest_activity = iface_activity - if latest_activity is None and iface_connected_at is not None: - latest_activity = iface_connected_at - if latest_activity is None: - latest_activity = now_monotonic - - inactivity_elapsed = now_monotonic - latest_activity - - connected_attr = getattr(iface, "isConnected", None) - believed_disconnected = False - connected_state = _connected_state(connected_attr) - if connected_state is None: - if callable(connected_attr): - try: - believed_disconnected = not bool(connected_attr()) - except Exception: - believed_disconnected = False - elif connected_attr is not None: - try: - believed_disconnected = not bool(connected_attr) - except Exception: # pragma: no cover - defensive guard - believed_disconnected = False - else: - believed_disconnected = not connected_state - - should_reconnect = believed_disconnected or ( - inactivity_elapsed >= inactivity_reconnect_secs - ) - - if should_reconnect: - if ( - last_inactivity_reconnect is None - or now_monotonic - last_inactivity_reconnect - >= inactivity_reconnect_secs - ): - reason = ( - "disconnected" - if believed_disconnected - else f"no data for {inactivity_elapsed:.0f}s" - ) - config._debug_log( - "Mesh interface inactivity detected", - context="daemon.interface", - severity="warn", - reason=reason, - ) - last_inactivity_reconnect = now_monotonic - _close_interface(iface) - iface = None - announced_target = False - initial_snapshot_sent = False - energy_session_deadline = None - iface_connected_at = None - continue - - ingestor_announcement_sent = _process_ingestor_heartbeat( - iface, ingestor_announcement_sent=ingestor_announcement_sent - ) - - retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) - stop.wait(config.SNAPSHOT_SECS) + while not state.stop.is_set(): + if not _loop_iteration(state): + state.stop.wait(config.SNAPSHOT_SECS) except KeyboardInterrupt: # pragma: no cover - interactive only config._debug_log( "Received KeyboardInterrupt; shutting down", context="daemon.main", severity="info", ) - stop.set() + state.stop.set() finally: - _close_interface(iface) + _close_interface(state.iface) __all__ = [ "_RECEIVE_TOPICS", - "_event_wait_allows_default_timeout", - "_node_items_snapshot", - "_subscribe_receive_topics", - "_is_ble_interface", - "_process_ingestor_heartbeat", + "_advance_retry_delay", + "_loop_iteration", + "_check_energy_saving", + "_check_inactivity_reconnect", "_connected_state", + "_energy_sleep", + "_event_wait_allows_default_timeout", + "_is_ble_interface", + "_node_items_snapshot", + "_process_ingestor_heartbeat", + "_subscribe_receive_topics", + "_try_connect", + "_try_send_snapshot", "main", ] diff --git a/data/mesh_ingestor/events.py b/data/mesh_ingestor/events.py index 3780cc2..2e9dad6 100644 --- a/data/mesh_ingestor/events.py +++ b/data/mesh_ingestor/events.py @@ -27,10 +27,13 @@ from __future__ import annotations from typing import NotRequired, TypedDict -class MessageEvent(TypedDict, total=False): +class _MessageEventRequired(TypedDict): id: int rx_time: int rx_iso: str + + +class MessageEvent(_MessageEventRequired, total=False): from_id: object to_id: object channel: int @@ -48,15 +51,18 @@ class MessageEvent(TypedDict, total=False): modem_preset: str -class PositionEvent(TypedDict, total=False): +class _PositionEventRequired(TypedDict): id: int + rx_time: int + rx_iso: str + + +class PositionEvent(_PositionEventRequired, total=False): node_id: str node_num: int | None num: int | None from_id: str | None to_id: object - rx_time: int - rx_iso: str latitude: float | None longitude: float | None altitude: float | None @@ -78,14 +84,17 @@ class PositionEvent(TypedDict, total=False): modem_preset: str -class TelemetryEvent(TypedDict, total=False): +class _TelemetryEventRequired(TypedDict): id: int + rx_time: int + rx_iso: str + + +class TelemetryEvent(_TelemetryEventRequired, total=False): node_id: str | None node_num: int | None from_id: object to_id: object - rx_time: int - rx_iso: str telemetry_time: int | None channel: int portnum: str | None @@ -102,20 +111,26 @@ class TelemetryEvent(TypedDict, total=False): # evolves over time. -class NeighborEntry(TypedDict, total=False): +class _NeighborEntryRequired(TypedDict): + rx_time: int + rx_iso: str + + +class NeighborEntry(_NeighborEntryRequired, total=False): neighbor_id: str neighbor_num: int | None snr: float | None + + +class _NeighborsSnapshotRequired(TypedDict): + node_id: str rx_time: int rx_iso: str -class NeighborsSnapshot(TypedDict, total=False): - node_id: str +class NeighborsSnapshot(_NeighborsSnapshotRequired, total=False): node_num: int | None neighbors: list[NeighborEntry] - rx_time: int - rx_iso: str node_broadcast_interval_secs: int | None last_sent_by_id: str | None ingestor: str | None @@ -123,14 +138,17 @@ class NeighborsSnapshot(TypedDict, total=False): modem_preset: str -class TraceEvent(TypedDict, total=False): +class _TraceEventRequired(TypedDict): + hops: list[int] + rx_time: int + rx_iso: str + + +class TraceEvent(_TraceEventRequired, total=False): id: int | None request_id: int | None src: int | None dest: int | None - rx_time: int - rx_iso: str - hops: list[int] rssi: int | None snr: float | None elapsed_ms: int | None @@ -161,4 +179,3 @@ __all__ = [ "TelemetryEvent", "TraceEvent", ] - diff --git a/data/mesh_ingestor/node_identity.py b/data/mesh_ingestor/node_identity.py index 2210ce0..29d4010 100644 --- a/data/mesh_ingestor/node_identity.py +++ b/data/mesh_ingestor/node_identity.py @@ -25,8 +25,7 @@ from __future__ import annotations from typing import Final - -_CANONICAL_PREFIX: Final[str] = "!" +CANONICAL_PREFIX: Final[str] = "!" def canonical_node_id(value: object) -> str | None: @@ -48,7 +47,7 @@ def canonical_node_id(value: object) -> str | None: return None if num < 0: return None - return f"{_CANONICAL_PREFIX}{num & 0xFFFFFFFF:08x}" + return f"{CANONICAL_PREFIX}{num & 0xFFFFFFFF:08x}" if not isinstance(value, str): return None @@ -59,13 +58,13 @@ def canonical_node_id(value: object) -> str | None: # Meshtastic special destinations like "^all" are not node ids; callers # that already accept them should keep passing them through unchanged. return trimmed - if trimmed.startswith(_CANONICAL_PREFIX): + if trimmed.startswith(CANONICAL_PREFIX): body = trimmed[1:] elif trimmed.lower().startswith("0x"): body = trimmed[2:] elif trimmed.isdigit(): try: - return f"{_CANONICAL_PREFIX}{int(trimmed, 10) & 0xFFFFFFFF:08x}" + return f"{CANONICAL_PREFIX}{int(trimmed, 10) & 0xFFFFFFFF:08x}" except ValueError: return None else: @@ -74,7 +73,7 @@ def canonical_node_id(value: object) -> str | None: if not body: return None try: - return f"{_CANONICAL_PREFIX}{int(body, 16) & 0xFFFFFFFF:08x}" + return f"{CANONICAL_PREFIX}{int(body, 16) & 0xFFFFFFFF:08x}" except ValueError: return None @@ -96,7 +95,7 @@ def node_num_from_id(node_id: object) -> int | None: trimmed = node_id.strip() if not trimmed: return None - if trimmed.startswith(_CANONICAL_PREFIX): + if trimmed.startswith(CANONICAL_PREFIX): trimmed = trimmed[1:] if trimmed.lower().startswith("0x"): trimmed = trimmed[2:] @@ -110,7 +109,7 @@ def node_num_from_id(node_id: object) -> int | None: __all__ = [ + "CANONICAL_PREFIX", "canonical_node_id", "node_num_from_id", ] - diff --git a/data/mesh_ingestor/provider.py b/data/mesh_ingestor/provider.py index 32ae0f9..f6e11dc 100644 --- a/data/mesh_ingestor/provider.py +++ b/data/mesh_ingestor/provider.py @@ -23,7 +23,8 @@ from __future__ import annotations import enum from collections.abc import Iterable -from typing import Protocol +from typing import Protocol, runtime_checkable + class ProviderCapability(enum.Flag): """Feature flags describing what a provider can supply.""" @@ -33,11 +34,11 @@ class ProviderCapability(enum.Flag): HEARTBEATS = enum.auto() +@runtime_checkable class Provider(Protocol): """Abstract source of mesh observations.""" name: str - capabilities: ProviderCapability def subscribe(self) -> list[str]: """Subscribe to any async receive callbacks and return topic names.""" @@ -62,4 +63,3 @@ __all__ = [ "Provider", "ProviderCapability", ] - diff --git a/data/mesh_ingestor/providers/__init__.py b/data/mesh_ingestor/providers/__init__.py index 4e53443..6cd80dc 100644 --- a/data/mesh_ingestor/providers/__init__.py +++ b/data/mesh_ingestor/providers/__init__.py @@ -23,4 +23,3 @@ from __future__ import annotations from .meshtastic import MeshtasticProvider __all__ = ["MeshtasticProvider"] - diff --git a/data/mesh_ingestor/providers/meshtastic.py b/data/mesh_ingestor/providers/meshtastic.py index 21567ac..80eb399 100644 --- a/data/mesh_ingestor/providers/meshtastic.py +++ b/data/mesh_ingestor/providers/meshtastic.py @@ -16,17 +16,17 @@ from __future__ import annotations -from collections.abc import Iterable +import time -from .. import interfaces -from ..provider import ProviderCapability +from pubsub import pub + +from .. import config, daemon as _daemon, handlers, interfaces class MeshtasticProvider: """Meshtastic ingestion provider (current default).""" name = "meshtastic" - capabilities = ProviderCapability.NODE_SNAPSHOT | ProviderCapability.HEARTBEATS def __init__(self): self._subscribed: list[str] = [] @@ -37,14 +37,15 @@ class MeshtasticProvider: if self._subscribed: return list(self._subscribed) - # Delegate to the historical subscription helper in `daemon.py` so unit - # tests can monkeypatch the subscription mechanism via `daemon.pub`. - from .. import daemon as _daemon # local import avoids module cycles - - topics = _daemon._subscribe_receive_topics() - - self._subscribed = topics - return list(topics) + subscribed = [] + for topic in _daemon._RECEIVE_TOPICS: + try: + pub.subscribe(handlers.on_receive, topic) + subscribed.append(topic) + except Exception as exc: # pragma: no cover + config._debug_log(f"failed to subscribe to {topic!r}: {exc}") + self._subscribed = subscribed + return list(subscribed) def connect( self, *, active_candidate: str | None @@ -56,7 +57,9 @@ class MeshtasticProvider: next_candidate = active_candidate if active_candidate: - iface, resolved_target = interfaces._create_serial_interface(active_candidate) + iface, resolved_target = interfaces._create_serial_interface( + active_candidate + ) else: iface, resolved_target = interfaces._create_default_interface() next_candidate = resolved_target @@ -69,16 +72,20 @@ class MeshtasticProvider: def extract_host_node_id(self, iface: object) -> str | None: return interfaces._extract_host_node_id(iface) - def node_snapshot_items(self, iface: object) -> Iterable[tuple[str, object]]: + def node_snapshot_items(self, iface: object) -> list[tuple[str, object]]: nodes = getattr(iface, "nodes", {}) or {} - items_callable = getattr(nodes, "items", None) - if callable(items_callable): - return list(items_callable()) - if hasattr(nodes, "__iter__") and hasattr(nodes, "__getitem__"): - keys = list(nodes) - return [(key, nodes[key]) for key in keys] + for _ in range(3): + try: + return list(nodes.items()) + except RuntimeError as err: + if "dictionary changed size during iteration" not in str(err): + raise + time.sleep(0) + config._debug_log( + "Skipping node snapshot due to concurrent modification", + context="meshtastic.snapshot", + ) return [] __all__ = ["MeshtasticProvider"] - diff --git a/tests/test_daemon_unit.py b/tests/test_daemon_unit.py index 2924477..e385a97 100644 --- a/tests/test_daemon_unit.py +++ b/tests/test_daemon_unit.py @@ -435,3 +435,391 @@ def test_main_inactivity_reconnect(monkeypatch): daemon.main() assert any(event.is_set() for event in FakeEvent.instances) + + +# --------------------------------------------------------------------------- +# Helper: build a minimal _DaemonState for unit tests +# --------------------------------------------------------------------------- + + +def _make_state(**overrides): + """Return a :class:`daemon._DaemonState` with sensible defaults. + + Any keyword argument is forwarded as a field override via ``setattr`` + after construction, so callers only need to supply fields under test. + """ + state = daemon._DaemonState( + provider=None, # type: ignore[arg-type] + stop=FakeEvent(), # type: ignore[arg-type] + configured_port=None, + inactivity_reconnect_secs=0.0, + energy_saving_enabled=False, + energy_online_secs=0.0, + energy_sleep_secs=0.0, + retry_delay=0.0, + last_seen_packet_monotonic=None, + active_candidate=None, + ) + for key, val in overrides.items(): + setattr(state, key, val) + return state + + +# --------------------------------------------------------------------------- +# _advance_retry_delay +# --------------------------------------------------------------------------- + + +def test_advance_retry_delay_disabled(monkeypatch): + """Returns current delay unchanged when the max is zero.""" + monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 0) + assert daemon._advance_retry_delay(5.0) == 5.0 + + +def test_advance_retry_delay_bootstrap(monkeypatch): + """Seeds from initial config when current delay is zero (first call).""" + monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 60.0) + monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 3.0) + assert daemon._advance_retry_delay(0.0) == 3.0 + + +def test_advance_retry_delay_doubles_and_caps(monkeypatch): + """Doubles current delay and caps at the configured maximum.""" + monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 10.0) + monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 1.0) + assert daemon._advance_retry_delay(3.0) == 6.0 + assert daemon._advance_retry_delay(7.0) == 10.0 + + +# --------------------------------------------------------------------------- +# _energy_sleep +# --------------------------------------------------------------------------- + + +def test_energy_sleep_no_op_when_disabled(): + """No wait issued when energy saving is disabled.""" + state = _make_state(energy_saving_enabled=False, energy_sleep_secs=1.0) + daemon._energy_sleep(state, "reason") + assert not state.stop.wait_calls + + +def test_energy_sleep_no_op_when_zero_secs(): + """No wait issued when sleep duration is zero.""" + state = _make_state(energy_saving_enabled=True, energy_sleep_secs=0.0) + daemon._energy_sleep(state, "reason") + assert not state.stop.wait_calls + + +def test_energy_sleep_emits_debug_log(monkeypatch): + """Debug log is emitted when DEBUG is enabled.""" + state = _make_state(energy_saving_enabled=True, energy_sleep_secs=2.0) + logged = [] + monkeypatch.setattr(daemon.config, "DEBUG", True) + monkeypatch.setattr( + daemon.config, "_debug_log", lambda msg, **_kw: logged.append(msg) + ) + daemon._energy_sleep(state, "wake up") + assert any("wake up" in m for m in logged) + assert state.stop.wait_calls == [2.0] + + +def test_energy_sleep_waits_when_debug_off(monkeypatch): + """Wait is issued for the configured duration when DEBUG is off.""" + state = _make_state(energy_saving_enabled=True, energy_sleep_secs=1.5) + monkeypatch.setattr(daemon.config, "DEBUG", False) + daemon._energy_sleep(state, "reason") + assert state.stop.wait_calls == [1.5] + + +# --------------------------------------------------------------------------- +# _try_connect +# --------------------------------------------------------------------------- + + +def test_try_connect_no_available_interface_raises_system_exit(monkeypatch): + """NoAvailableMeshInterface propagates as SystemExit(1).""" + + class _NoIface: + def connect(self, *, active_candidate): + raise daemon.interfaces.NoAvailableMeshInterface("none") + + def extract_host_node_id(self, iface): + return None + + state = _make_state(active_candidate="serial0", configured_port="serial0") + state.provider = _NoIface() # type: ignore[assignment] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + with pytest.raises(SystemExit): + daemon._try_connect(state) + + +def test_try_connect_generic_failure_resets_candidate(monkeypatch): + """Connect failure in auto-detect mode clears the active candidate.""" + + class _FailProvider: + def connect(self, *, active_candidate): + raise OSError("device busy") + + def extract_host_node_id(self, iface): + return None + + state = _make_state(active_candidate="serial0", configured_port=None) + state.provider = _FailProvider() # type: ignore[assignment] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 0) + monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 0) + + result = daemon._try_connect(state) + assert result is False + assert state.active_candidate is None + assert state.announced_target is False + + +def test_try_connect_sets_energy_session_deadline(monkeypatch): + """Energy-saving deadline is assigned when online duration is positive.""" + + class _OkProvider: + def connect(self, *, active_candidate): + return DummyInterface(), active_candidate, active_candidate + + def extract_host_node_id(self, iface): + return "!host" + + state = _make_state( + active_candidate="serial0", + configured_port="serial0", + energy_saving_enabled=True, + energy_online_secs=30.0, + ) + state.provider = _OkProvider() # type: ignore[assignment] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 0) + monkeypatch.setattr( + daemon.handlers, "register_host_node_id", lambda *_a, **_k: None + ) + monkeypatch.setattr(daemon.handlers, "host_node_id", lambda: "!host") + monkeypatch.setattr( + daemon.ingestors, "set_ingestor_node_id", lambda *_a, **_k: None + ) + + result = daemon._try_connect(state) + assert result is True + assert state.energy_session_deadline is not None + + +# --------------------------------------------------------------------------- +# _check_energy_saving +# --------------------------------------------------------------------------- + + +def test_check_energy_saving_session_expired(monkeypatch): + """Iface is closed and True returned when the session deadline has passed.""" + state = _make_state(energy_saving_enabled=True) + state.iface = DummyInterface() + state.energy_session_deadline = 0.0 + monkeypatch.setattr(daemon.time, "monotonic", lambda: 1.0) + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + + result = daemon._check_energy_saving(state) + assert result is True + assert state.iface is None + assert state.energy_session_deadline is None + + +def test_check_energy_saving_ble_client_disconnected(monkeypatch): + """Iface is closed and True returned when the BLE client reference is gone.""" + state = _make_state(energy_saving_enabled=True) + state.iface = DummyInterface(client_present=False) + state.energy_session_deadline = None + monkeypatch.setattr(daemon, "_is_ble_interface", lambda _: True) + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + + result = daemon._check_energy_saving(state) + assert result is True + assert state.iface is None + + +# --------------------------------------------------------------------------- +# _try_send_snapshot +# --------------------------------------------------------------------------- + + +def test_try_send_snapshot_empty_nodes(): + """Returns True without setting initial_snapshot_sent when no nodes exist.""" + + class _EmptyProvider: + def node_snapshot_items(self, iface): + return [] + + state = _make_state() + state.iface = DummyInterface(nodes={}) + state.provider = _EmptyProvider() # type: ignore[assignment] + + result = daemon._try_send_snapshot(state) + assert result is True + assert state.initial_snapshot_sent is False + + +def test_try_send_snapshot_upsert_failure_is_non_fatal(monkeypatch): + """Upsert errors are logged but do not abort the snapshot pass.""" + + class _OneNodeProvider: + def node_snapshot_items(self, iface): + return [("!node1", {"id": 1})] + + def _raise(*_a, **_k): + raise ValueError("bad node") + + state = _make_state() + state.iface = DummyInterface() + state.provider = _OneNodeProvider() # type: ignore[assignment] + logged = [] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *a, **kw: logged.append(kw)) + monkeypatch.setattr(daemon.config, "DEBUG", False) + monkeypatch.setattr(daemon.handlers, "upsert_node", _raise) + + result = daemon._try_send_snapshot(state) + assert result is True + assert state.initial_snapshot_sent is True + assert any(c.get("context") == "daemon.snapshot" for c in logged) + + +def test_try_send_snapshot_upsert_failure_debug_payload(monkeypatch): + """The node payload is logged when DEBUG is enabled and upsert fails.""" + + class _OneNodeProvider: + def node_snapshot_items(self, iface): + return [("!node1", {"id": 1})] + + def _raise(*_a, **_k): + raise ValueError("bad") + + state = _make_state() + state.iface = DummyInterface() + state.provider = _OneNodeProvider() # type: ignore[assignment] + logged = [] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *a, **kw: logged.append(kw)) + monkeypatch.setattr(daemon.config, "DEBUG", True) + monkeypatch.setattr(daemon.handlers, "upsert_node", _raise) + + daemon._try_send_snapshot(state) + assert any("node" in c for c in logged) + + +def test_try_send_snapshot_outer_exception_resets_iface(monkeypatch): + """An exception from node_snapshot_items resets the interface and returns False.""" + + class _BrokenProvider: + def node_snapshot_items(self, iface): + raise RuntimeError("boom") + + state = _make_state() + state.iface = DummyInterface() + state.provider = _BrokenProvider() # type: ignore[assignment] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 0) + + result = daemon._try_send_snapshot(state) + assert result is False + assert state.iface is None + + +# --------------------------------------------------------------------------- +# _check_inactivity_reconnect (additional branches) +# --------------------------------------------------------------------------- + + +def test_check_inactivity_reconnect_throttles_rapid_reconnects(monkeypatch): + """A reconnect within the inactivity window is suppressed.""" + state = _make_state(inactivity_reconnect_secs=60.0) + state.iface = DummyInterface(is_connected=False) + state.iface_connected_at = 0.0 + state.last_inactivity_reconnect = 1.0 # recent + + monkeypatch.setattr(daemon.time, "monotonic", lambda: 10.0) + monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None) + + assert daemon._check_inactivity_reconnect(state) is False + + +def test_check_inactivity_reconnect_uses_connected_at_when_no_packets(monkeypatch): + """Uses iface_connected_at as the activity baseline when no packets seen.""" + state = _make_state(inactivity_reconnect_secs=60.0) + state.iface = DummyInterface(is_connected=True) + state.iface_connected_at = 5.0 + state.last_inactivity_reconnect = None + + monkeypatch.setattr(daemon.time, "monotonic", lambda: 10.0) + monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None) + + # 10.0 - 5.0 = 5.0 < 60.0 → not triggered + assert daemon._check_inactivity_reconnect(state) is False + + +def test_check_inactivity_reconnect_uses_now_when_no_baseline(monkeypatch): + """Falls back to current time when neither packets nor connected_at is set.""" + state = _make_state(inactivity_reconnect_secs=60.0) + state.iface = DummyInterface(is_connected=True) + state.iface_connected_at = None + state.last_inactivity_reconnect = None + + monkeypatch.setattr(daemon.time, "monotonic", lambda: 10.0) + monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None) + + # latest_activity = now(10.0); inactivity_elapsed = 0.0 < 60.0 → not triggered + assert daemon._check_inactivity_reconnect(state) is False + + +# --------------------------------------------------------------------------- +# _loop_iteration +# --------------------------------------------------------------------------- + + +def test_loop_iteration_connect_fails_returns_true(monkeypatch): + """Returns True (continue) when iface is absent and connect fails.""" + state = _make_state() + state.iface = None + monkeypatch.setattr(daemon, "_try_connect", lambda s: False) + assert daemon._loop_iteration(state) is True + + +def test_loop_iteration_energy_saving_triggers_returns_true(monkeypatch): + """Returns True (continue) when energy saving disconnects the interface.""" + state = _make_state() + state.iface = object() + monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: True) + assert daemon._loop_iteration(state) is True + + +def test_loop_iteration_snapshot_fails_returns_true(monkeypatch): + """Returns True (continue) when the initial snapshot fails.""" + state = _make_state() + state.iface = object() + state.initial_snapshot_sent = False + monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: False) + monkeypatch.setattr(daemon, "_try_send_snapshot", lambda s: False) + assert daemon._loop_iteration(state) is True + + +def test_loop_iteration_inactivity_triggers_returns_true(monkeypatch): + """Returns True (continue) when inactivity reconnect fires.""" + state = _make_state() + state.iface = object() + state.initial_snapshot_sent = True + monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: False) + monkeypatch.setattr(daemon, "_check_inactivity_reconnect", lambda s: True) + assert daemon._loop_iteration(state) is True + + +def test_loop_iteration_full_pass_returns_false(monkeypatch): + """Returns False (sleep) after a complete iteration with no early exits.""" + state = _make_state() + state.iface = object() + state.initial_snapshot_sent = True + monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: False) + monkeypatch.setattr(daemon, "_check_inactivity_reconnect", lambda s: False) + monkeypatch.setattr( + daemon, "_process_ingestor_heartbeat", lambda iface, **_kw: False + ) + monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 0) + assert daemon._loop_iteration(state) is False diff --git a/tests/test_events_unit.py b/tests/test_events_unit.py new file mode 100644 index 0000000..871243f --- /dev/null +++ b/tests/test_events_unit.py @@ -0,0 +1,232 @@ +# 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.11", + "lora_freq": 906875, + "modem_preset": "LONG_FAST", + } + assert hb["version"] == "0.5.11" + 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.11", + } + assert "lora_freq" not in hb diff --git a/tests/test_node_identity_unit.py b/tests/test_node_identity_unit.py index e94a4bd..e1ddd3a 100644 --- a/tests/test_node_identity_unit.py +++ b/tests/test_node_identity_unit.py @@ -52,3 +52,23 @@ def test_node_num_from_id_parses_canonical_and_hex(): assert node_num_from_id(123) == 123 +def test_canonical_node_id_rejects_none_and_empty(): + assert canonical_node_id(None) is None + assert canonical_node_id("") is None + assert canonical_node_id(" ") is None + + +def test_canonical_node_id_rejects_negative(): + assert canonical_node_id(-1) is None + assert canonical_node_id(-0xABCDEF01) is None + + +def test_canonical_node_id_truncates_overflow(): + # Values wider than 32 bits are masked, not rejected. + assert canonical_node_id(0x1_ABCDEF01) == "!abcdef01" + + +def test_node_num_from_id_rejects_none_and_empty(): + assert node_num_from_id(None) is None + assert node_num_from_id("") is None + assert node_num_from_id("not-hex") is None diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index d54de3b..88cdf82 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -26,11 +26,17 @@ if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from data.mesh_ingestor import daemon # noqa: E402 - path setup +from data.mesh_ingestor.provider import Provider # noqa: E402 - path setup from data.mesh_ingestor.providers.meshtastic import ( # noqa: E402 - path setup MeshtasticProvider, ) +def test_meshtastic_provider_satisfies_protocol(): + """MeshtasticProvider must structurally satisfy the Provider Protocol.""" + assert isinstance(MeshtasticProvider(), Provider) + + def test_daemon_main_uses_provider_connect(monkeypatch): calls = {"connect": 0} @@ -40,6 +46,7 @@ def test_daemon_main_uses_provider_connect(monkeypatch): def connect(self, *, active_candidate): # type: ignore[override] calls["connect"] += 1 + # Return a minimal iface and stop immediately via Event. class Iface: nodes = {} @@ -89,13 +96,81 @@ def test_daemon_main_uses_provider_connect(monkeypatch): ), ) - monkeypatch.setattr(daemon.handlers, "register_host_node_id", lambda *_a, **_k: None) + monkeypatch.setattr( + daemon.handlers, "register_host_node_id", lambda *_a, **_k: None + ) monkeypatch.setattr(daemon.handlers, "host_node_id", lambda: "!host") monkeypatch.setattr(daemon.handlers, "upsert_node", lambda *_a, **_k: None) monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None) - monkeypatch.setattr(daemon.ingestors, "set_ingestor_node_id", lambda *_a, **_k: None) - monkeypatch.setattr(daemon.ingestors, "queue_ingestor_heartbeat", lambda *_a, **_k: True) + monkeypatch.setattr( + daemon.ingestors, "set_ingestor_node_id", lambda *_a, **_k: None + ) + monkeypatch.setattr( + daemon.ingestors, "queue_ingestor_heartbeat", lambda *_a, **_k: True + ) daemon.main(provider=FakeProvider()) assert calls["connect"] >= 1 + +def test_node_snapshot_items_retries_on_concurrent_mutation(monkeypatch): + """node_snapshot_items must retry on dict-mutation RuntimeError, not raise.""" + from data.mesh_ingestor.providers.meshtastic import MeshtasticProvider + + attempt = {"n": 0} + + class MutatingNodes: + def items(self): + attempt["n"] += 1 + if attempt["n"] < 3: + raise RuntimeError("dictionary changed size during iteration") + return [("!aabbccdd", {"num": 1})] + + class FakeIface: + nodes = MutatingNodes() + + monkeypatch.setattr("time.sleep", lambda _: None) + result = MeshtasticProvider().node_snapshot_items(FakeIface()) + assert result == [("!aabbccdd", {"num": 1})] + assert attempt["n"] == 3 + + +def test_node_snapshot_items_returns_empty_after_retry_exhaustion(monkeypatch): + """node_snapshot_items returns [] (non-fatal) when all retries fail.""" + from data.mesh_ingestor.providers.meshtastic import MeshtasticProvider + import data.mesh_ingestor.providers.meshtastic as _mod + + class AlwaysMutating: + def items(self): + raise RuntimeError("dictionary changed size during iteration") + + class FakeIface: + nodes = AlwaysMutating() + + monkeypatch.setattr("time.sleep", lambda _: None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + result = MeshtasticProvider().node_snapshot_items(FakeIface()) + assert result == [] + + +def test_meshtastic_subscribe_is_idempotent(monkeypatch): + """Calling subscribe() twice returns the cached list without re-subscribing.""" + import data.mesh_ingestor.providers.meshtastic as _m + + subscribe_calls: list[str] = [] + + monkeypatch.setattr( + _m, + "pub", + types.SimpleNamespace( + subscribe=lambda _h, topic: subscribe_calls.append(topic) + ), + ) + + provider = MeshtasticProvider() + first = provider.subscribe() + second = provider.subscribe() + + assert first == second + # pub.subscribe should only have been called once (first invocation) + assert len(subscribe_calls) == len(first)