diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index e5849c9..eefd70c 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -47,6 +47,9 @@ DEFAULT_ENERGY_SLEEP_SECS = float(6 * 60 * 60) DEFAULT_INGESTOR_HEARTBEAT_SECS = float(60 * 60) """Interval between ingestor heartbeat announcements.""" +DEFAULT_SELF_NODE_REPORT_INTERVAL_SECS = float(60 * 60) +"""Interval between periodic forced self-node re-reports from the daemon.""" + CONNECTION = os.environ.get("CONNECTION") """Optional connection target for the mesh interface. @@ -154,6 +157,7 @@ _INACTIVITY_RECONNECT_SECS = DEFAULT_INACTIVITY_RECONNECT_SECS _ENERGY_ONLINE_DURATION_SECS = DEFAULT_ENERGY_ONLINE_DURATION_SECS _ENERGY_SLEEP_SECS = DEFAULT_ENERGY_SLEEP_SECS _INGESTOR_HEARTBEAT_SECS = DEFAULT_INGESTOR_HEARTBEAT_SECS +_SELF_NODE_REPORT_INTERVAL_SECS = DEFAULT_SELF_NODE_REPORT_INTERVAL_SECS def _debug_log( @@ -209,5 +213,6 @@ __all__ = [ "_ENERGY_ONLINE_DURATION_SECS", "_ENERGY_SLEEP_SECS", "_INGESTOR_HEARTBEAT_SECS", + "_SELF_NODE_REPORT_INTERVAL_SECS", "_debug_log", ] diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index bb1785d..1bb7cbe 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -264,6 +264,7 @@ class _DaemonState: last_inactivity_reconnect: float | None = None ingestor_announcement_sent: bool = False announced_target: bool = False + last_self_node_report: float | None = None # --------------------------------------------------------------------------- @@ -309,6 +310,7 @@ def _try_connect(state: _DaemonState) -> bool: 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 + state.last_self_node_report = None if not state.announced_target and state.resolved_target: config._debug_log( "Using mesh interface", @@ -387,6 +389,7 @@ def _check_energy_saving(state: _DaemonState) -> bool: state.iface = None state.announced_target = False state.initial_snapshot_sent = False + state.last_self_node_report = None state.energy_session_deadline = None _energy_sleep(state, reason) return True @@ -507,11 +510,59 @@ def _check_inactivity_reconnect(state: _DaemonState) -> bool: state.iface = None state.announced_target = False state.initial_snapshot_sent = False + state.last_self_node_report = None state.energy_session_deadline = None state.iface_connected_at = None return True +# --------------------------------------------------------------------------- +# Periodic self-node report helper +# --------------------------------------------------------------------------- + + +def _try_send_self_node(state: _DaemonState) -> None: + """Re-upsert the host self-node when the provider supports it. + + Called once immediately after the initial snapshot and then at most once + per :data:`~data.mesh_ingestor.config._SELF_NODE_REPORT_INTERVAL_SECS`. + This ensures the self-node's protocol and radio metadata are refreshed + even when the ingestor heartbeat races ahead of the first SELF_INFO event + (meshcore) or when the protocol never sends periodic NODEINFO for itself. + + Parameters: + state: Current daemon loop state. + + Returns: + ``None``. Errors are logged and suppressed so a single failure does + not break the main loop. + """ + self_node_fn = getattr(state.provider, "self_node_item", None) + if not callable(self_node_fn): + return + try: + item = self_node_fn(state.iface) + if item is None: + return + node_id, node = item + handlers.upsert_node(node_id, node) + state.last_self_node_report = time.monotonic() + config._debug_log( + "Sent periodic self-node report", + context="daemon.self_node", + severity="info", + node_id=node_id, + ) + except Exception as exc: + config._debug_log( + "Self-node re-report failed", + context="daemon.self_node", + severity="warn", + error_class=exc.__class__.__name__, + error_message=str(exc), + ) + + # --------------------------------------------------------------------------- # Loop iteration helper # --------------------------------------------------------------------------- @@ -540,6 +591,15 @@ def _loop_iteration(state: _DaemonState) -> bool: state.ingestor_announcement_sent = _process_ingestor_heartbeat( state.iface, ingestor_announcement_sent=state.ingestor_announcement_sent ) + # Periodically re-upsert the host self-node so that its protocol and radio + # metadata are corrected after the ingestor heartbeat is registered, and + # kept fresh for protocols (e.g. meshcore) that only emit SELF_INFO once. + _now = time.monotonic() + if state.initial_snapshot_sent and ( + state.last_self_node_report is None + or _now - state.last_self_node_report >= config._SELF_NODE_REPORT_INTERVAL_SECS + ): + _try_send_self_node(state) state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) return False @@ -644,6 +704,7 @@ __all__ = [ "_process_ingestor_heartbeat", "_subscribe_receive_topics", "_try_connect", + "_try_send_self_node", "_try_send_snapshot", "main", ] diff --git a/data/mesh_ingestor/handlers/_state.py b/data/mesh_ingestor/handlers/_state.py index cd79b67..420c5f1 100644 --- a/data/mesh_ingestor/handlers/_state.py +++ b/data/mesh_ingestor/handlers/_state.py @@ -45,6 +45,18 @@ every packet would overwrite the host's profile too aggressively; this window throttles updates to at most once per hour. """ +_host_nodeinfo_last_seen: float | None = None +"""Monotonic timestamp of the last accepted host NODEINFO upsert.""" + +_HOST_NODEINFO_INTERVAL_SECS: int = 60 * 60 +"""Minimum interval (seconds) between accepted host NODEINFO upserts. + +The meshtastic library re-broadcasts the local node's NODEINFO to the mesh +periodically. Accepting every broadcast would overwrite the host node record +too aggressively; this window throttles self-NODEINFO upserts to at most once +per hour. +""" + # --------------------------------------------------------------------------- # Packet receipt tracking # --------------------------------------------------------------------------- @@ -69,10 +81,11 @@ def register_host_node_id(node_id: str | None) -> None: the current host assignment. """ - global _host_node_id, _host_telemetry_last_rx + global _host_node_id, _host_telemetry_last_rx, _host_nodeinfo_last_seen canonical = _canonical_node_id(node_id) _host_node_id = canonical _host_telemetry_last_rx = None + _host_nodeinfo_last_seen = None if canonical: config._debug_log( "Registered host device node id", @@ -128,6 +141,35 @@ def _host_telemetry_suppressed(rx_time: int) -> tuple[bool, int]: return True, int(math.ceil(remaining_secs / 60.0)) +def _host_nodeinfo_suppressed(now: float) -> bool: + """Return ``True`` when a host NODEINFO upsert should be suppressed. + + Self-NODEINFO upserts are throttled to at most once per + :data:`_HOST_NODEINFO_INTERVAL_SECS` to prevent the meshtastic library's + periodic rebroadcast from overwriting the host node record too aggressively. + + Parameters: + now: Current :func:`time.monotonic` value. + + Returns: + ``True`` when the request should be dropped; ``False`` when it should + proceed. + """ + if _host_nodeinfo_last_seen is None: + return False + return (now - _host_nodeinfo_last_seen) < _HOST_NODEINFO_INTERVAL_SECS + + +def _mark_host_nodeinfo_seen(now: float) -> None: + """Record that a host NODEINFO upsert was accepted. + + Parameters: + now: Current :func:`time.monotonic` value from the accepted upsert. + """ + global _host_nodeinfo_last_seen + _host_nodeinfo_last_seen = now + + def last_packet_monotonic() -> float | None: """Return the monotonic timestamp of the most recently processed packet. @@ -147,8 +189,11 @@ def _mark_packet_seen() -> None: __all__ = [ + "_HOST_NODEINFO_INTERVAL_SECS", "_HOST_TELEMETRY_INTERVAL_SECS", + "_host_nodeinfo_suppressed", "_host_telemetry_suppressed", + "_mark_host_nodeinfo_seen", "_mark_host_telemetry_seen", "_mark_packet_seen", "host_node_id", diff --git a/data/mesh_ingestor/handlers/nodeinfo.py b/data/mesh_ingestor/handlers/nodeinfo.py index 8d1640f..f63cfd7 100644 --- a/data/mesh_ingestor/handlers/nodeinfo.py +++ b/data/mesh_ingestor/handlers/nodeinfo.py @@ -76,6 +76,21 @@ def store_nodeinfo_packet(packet: Mapping, decoded: Mapping) -> None: if node_id is None: return + # Throttle self-NODEINFO upserts to at most once per hour. The meshtastic + # library rebroadcasts the local node's NODEINFO periodically; accepting + # every broadcast would overwrite the host node record too aggressively. + if node_id == _state.host_node_id(): + _now = time.monotonic() + if _state._host_nodeinfo_suppressed(_now): + if config.DEBUG: + config._debug_log( + "Suppressed host self-NODEINFO update within throttle window", + context="handlers.store_nodeinfo", + node_id=node_id, + ) + return + _state._mark_host_nodeinfo_seen(_now) + node_payload: dict = {} if user_dict: node_payload["user"] = user_dict diff --git a/data/mesh_ingestor/protocols/meshcore.py b/data/mesh_ingestor/protocols/meshcore.py index dd8360a..293322f 100644 --- a/data/mesh_ingestor/protocols/meshcore.py +++ b/data/mesh_ingestor/protocols/meshcore.py @@ -466,6 +466,8 @@ class _MeshcoreInterface: # which may cause extra upserts after a disconnect — the ON CONFLICT guard # in the Ruby web app ensures those are idempotent and safe. self._synthetic_node_ids: set[str] = set() + self._self_info_payload: dict | None = None + """Most recent SELF_INFO payload received from the device, or ``None``.""" # ------------------------------------------------------------------ # Contact management (called from the asyncio thread) @@ -650,15 +652,15 @@ def _process_self_info( handlers: Module reference for :func:`~data.mesh_ingestor.handlers` functions (passed to avoid circular-import issues). """ + # Cache the payload so node_snapshot_items / self_node_item can use it later. + iface._self_info_payload = payload + pub_key = payload.get("public_key", "") node_id = _meshcore_node_id(pub_key) - if node_id: - iface.host_node_id = node_id - handlers.register_host_node_id(node_id) - handlers.upsert_node(node_id, _self_info_to_node_dict(payload)) - # Capture radio metadata once — never overwrite a previously cached value. - # Mirrors the guard used by interfaces._ensure_radio_metadata for Meshtastic. + # Capture radio metadata BEFORE upserting the node so that + # _apply_radio_metadata_to_nodes finds populated values on the very first + # SELF_INFO. Never overwrite a previously cached value. radio_freq = payload.get("radio_freq") if radio_freq is not None and getattr(config, "LORA_FREQ", None) is None: config.LORA_FREQ = radio_freq @@ -667,6 +669,12 @@ def _process_self_info( ) if modem_preset is not None and getattr(config, "MODEM_PRESET", None) is None: config.MODEM_PRESET = modem_preset + + if node_id: + iface.host_node_id = node_id + handlers.register_host_node_id(node_id) + handlers.upsert_node(node_id, _self_info_to_node_dict(payload)) + config._debug_log( "MeshCore radio metadata captured", context="meshcore.self_info.radio", @@ -1146,9 +1154,37 @@ class MeshcoreProvider: """ return getattr(iface, "host_node_id", None) + def self_node_item(self, iface: object) -> tuple[str, dict] | None: + """Return the ``(node_id, node_dict)`` pair for the host self-node. + + Uses the most recently cached ``SELF_INFO`` payload stored on the + interface. Returns ``None`` when no SELF_INFO has been received yet + or when the public key cannot be mapped to a valid node ID. + + Parameters: + iface: Active :class:`_MeshcoreInterface` instance. + + Returns: + ``(canonical_node_id, node_dict)`` tuple or ``None``. + """ + if not isinstance(iface, _MeshcoreInterface): + return None + payload = getattr(iface, "_self_info_payload", None) + if not payload: + return None + node_id = _meshcore_node_id(payload.get("public_key", "")) + if not node_id: + return None + return node_id, _self_info_to_node_dict(payload) + def node_snapshot_items(self, iface: object) -> list[tuple[str, dict]]: """Return a snapshot of all known MeshCore contacts as node entries. + Includes the host self-node when a ``SELF_INFO`` payload has already + been received, so that the initial snapshot sent by the daemon + covers the local device even when the background event loop delivers + ``SELF_INFO`` before the snapshot is taken. + Parameters: iface: Active :class:`_MeshcoreInterface` instance. Any other object type causes an empty list to be returned. @@ -1159,7 +1195,11 @@ class MeshcoreProvider: """ if not isinstance(iface, _MeshcoreInterface): return [] - return iface.contacts_snapshot() + items: list[tuple[str, dict]] = list(iface.contacts_snapshot()) + self_item = self.self_node_item(iface) + if self_item is not None: + items.append(self_item) + return items __all__ = ["MeshcoreProvider"] diff --git a/tests/test_daemon_unit.py b/tests/test_daemon_unit.py index 1bb800d..5dfe439 100644 --- a/tests/test_daemon_unit.py +++ b/tests/test_daemon_unit.py @@ -1087,3 +1087,230 @@ def test_check_inactivity_reconnect_elapsed_triggers(monkeypatch): # latest_activity = iface_connected_at(0.0); elapsed = 100s > 30s → trigger result = daemon._check_inactivity_reconnect(state) assert result is True + + +# --------------------------------------------------------------------------- +# _try_send_self_node +# --------------------------------------------------------------------------- + + +def test_try_send_self_node_skips_when_no_method(): + """_try_send_self_node does nothing when provider has no self_node_item.""" + + class _NoSelfNode: + pass + + state = _make_state() + state.provider = _NoSelfNode() # type: ignore[assignment] + state.iface = DummyInterface() + # Should not raise; last_self_node_report stays None. + daemon._try_send_self_node(state) + assert state.last_self_node_report is None + + +def test_try_send_self_node_skips_when_item_is_none(monkeypatch): + """_try_send_self_node does nothing when self_node_item returns None.""" + + class _NullSelfNode: + def self_node_item(self, iface): + return None + + upserted = [] + monkeypatch.setattr( + daemon.handlers, "upsert_node", lambda nid, n: upserted.append(nid) + ) + + state = _make_state() + state.provider = _NullSelfNode() # type: ignore[assignment] + state.iface = DummyInterface() + daemon._try_send_self_node(state) + + assert upserted == [] + assert state.last_self_node_report is None + + +def test_try_send_self_node_calls_upsert_and_sets_timestamp(monkeypatch): + """_try_send_self_node upserts the self-node and records the timestamp.""" + + class _GoodSelfNode: + def self_node_item(self, iface): + return "!aabbccdd", {"user": {"longName": "Host"}} + + upserted = [] + monkeypatch.setattr( + daemon.handlers, "upsert_node", lambda nid, n: upserted.append(nid) + ) + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + fixed_time = 5000.0 + monkeypatch.setattr(daemon.time, "monotonic", lambda: fixed_time) + + state = _make_state() + state.provider = _GoodSelfNode() # type: ignore[assignment] + state.iface = DummyInterface() + daemon._try_send_self_node(state) + + assert upserted == ["!aabbccdd"] + assert state.last_self_node_report == fixed_time + + +def test_try_send_self_node_upsert_error_suppressed(monkeypatch): + """_try_send_self_node suppresses upsert errors and does not update timestamp.""" + + class _GoodSelfNode: + def self_node_item(self, iface): + return "!aabbccdd", {} + + def _raise(*_a, **_k): + raise RuntimeError("network error") + + monkeypatch.setattr(daemon.handlers, "upsert_node", _raise) + logged = [] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *a, **kw: logged.append(kw)) + + state = _make_state() + state.provider = _GoodSelfNode() # type: ignore[assignment] + state.iface = DummyInterface() + # Must not raise. + daemon._try_send_self_node(state) + + assert state.last_self_node_report is None + assert any(c.get("context") == "daemon.self_node" for c in logged) + + +def test_try_send_self_node_self_node_item_error_suppressed(monkeypatch): + """_try_send_self_node suppresses errors raised by self_node_item itself.""" + + class _BrokenSelfNode: + def self_node_item(self, iface): + raise RuntimeError("provider error") + + logged = [] + monkeypatch.setattr(daemon.config, "_debug_log", lambda *a, **kw: logged.append(kw)) + + state = _make_state() + state.provider = _BrokenSelfNode() # type: ignore[assignment] + state.iface = DummyInterface() + # Must not raise. + daemon._try_send_self_node(state) + + assert state.last_self_node_report is None + assert any(c.get("context") == "daemon.self_node" for c in logged) + + +# --------------------------------------------------------------------------- +# _loop_iteration — periodic self-node report +# --------------------------------------------------------------------------- + + +def _make_self_node_provider(node_item=("!aabbccdd", {"user": {}})): + """Return a minimal provider stub that exposes ``self_node_item``.""" + + class _SelfNodeProvider: + name = "test" + + def subscribe(self): + return [] + + def node_snapshot_items(self, iface): + return [] + + def self_node_item(self, iface): + return node_item + + return _SelfNodeProvider() + + +def _patch_loop_iteration_common(monkeypatch, *, now=100.0): + """Apply monkeypatches shared by all _loop_iteration self-node tests.""" + monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None) + monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None) + monkeypatch.setattr(daemon.config, "_SELF_NODE_REPORT_INTERVAL_SECS", 3600.0) + monkeypatch.setattr(daemon.time, "monotonic", lambda: now) + monkeypatch.setattr( + daemon, + "_process_ingestor_heartbeat", + lambda iface, **kw: kw.get("ingestor_announcement_sent", False), + ) + + +def test_loop_iteration_triggers_self_node_report_immediately_after_snapshot( + monkeypatch, +): + """Self-node report fires on the first iteration after the initial snapshot.""" + upserted = [] + monkeypatch.setattr( + daemon.handlers, "upsert_node", lambda nid, n: upserted.append(nid) + ) + _patch_loop_iteration_common(monkeypatch) + + state = _make_state() + state.iface = DummyInterface() + state.provider = _make_self_node_provider() # type: ignore[assignment] + state.initial_snapshot_sent = True + state.last_self_node_report = None # never reported before + + daemon._loop_iteration(state) + + assert "!aabbccdd" in upserted + + +def test_loop_iteration_self_node_not_triggered_before_snapshot(monkeypatch): + """Self-node report is NOT triggered before the initial snapshot is sent.""" + upserted = [] + monkeypatch.setattr( + daemon.handlers, "upsert_node", lambda nid, n: upserted.append(nid) + ) + _patch_loop_iteration_common(monkeypatch) + + state = _make_state() + state.iface = DummyInterface() + state.provider = _make_self_node_provider() # type: ignore[assignment] + state.initial_snapshot_sent = False # snapshot not yet sent + + # _loop_iteration will attempt _try_connect because iface is set but + # initial_snapshot_sent is False — prevent real connect by patching snapshot + monkeypatch.setattr(daemon, "_try_send_snapshot", lambda s: True) + + daemon._loop_iteration(state) + + assert "!aabbccdd" not in upserted + + +def test_loop_iteration_self_node_not_retried_within_interval(monkeypatch): + """Self-node report is NOT re-fired within the throttle interval.""" + upserted = [] + monkeypatch.setattr( + daemon.handlers, "upsert_node", lambda nid, n: upserted.append(nid) + ) + _patch_loop_iteration_common(monkeypatch, now=100.0) + + state = _make_state() + state.iface = DummyInterface() + state.provider = _make_self_node_provider() # type: ignore[assignment] + state.initial_snapshot_sent = True + # Simulate a recent report: 100 - 50 = 50 seconds ago < 3600 interval + state.last_self_node_report = 50.0 + + daemon._loop_iteration(state) + + assert "!aabbccdd" not in upserted + + +def test_loop_iteration_self_node_retried_after_interval(monkeypatch): + """Self-node report fires again after the full interval has elapsed.""" + upserted = [] + monkeypatch.setattr( + daemon.handlers, "upsert_node", lambda nid, n: upserted.append(nid) + ) + # now=5000; last_report=1000; elapsed=4000 > 3600 → should fire + _patch_loop_iteration_common(monkeypatch, now=5000.0) + + state = _make_state() + state.iface = DummyInterface() + state.provider = _make_self_node_provider() # type: ignore[assignment] + state.initial_snapshot_sent = True + state.last_self_node_report = 1000.0 # 4000 seconds ago + + daemon._loop_iteration(state) + + assert "!aabbccdd" in upserted diff --git a/tests/test_handlers_unit.py b/tests/test_handlers_unit.py index 2ab6f67..e22811f 100644 --- a/tests/test_handlers_unit.py +++ b/tests/test_handlers_unit.py @@ -39,10 +39,12 @@ def reset_handler_state(): """Reset global handler state between tests.""" _state_mod._host_node_id = None _state_mod._host_telemetry_last_rx = None + _state_mod._host_nodeinfo_last_seen = None _state_mod._last_packet_monotonic = None yield _state_mod._host_node_id = None _state_mod._host_telemetry_last_rx = None + _state_mod._host_nodeinfo_last_seen = None _state_mod._last_packet_monotonic = None @@ -75,6 +77,12 @@ class TestHostNodeId: handlers.register_host_node_id("!aabbccdd") assert _state_mod._host_telemetry_last_rx is None + def test_register_resets_nodeinfo_window(self): + """Registering a new host ID resets the NODEINFO suppression window.""" + _state_mod._host_nodeinfo_last_seen = 12345.0 + handlers.register_host_node_id("!aabbccdd") + assert _state_mod._host_nodeinfo_last_seen is None + def test_register_canonicalises_numeric(self): """Numeric node ID is converted to !xxxxxxxx form.""" handlers.register_host_node_id(0xAABBCCDD) @@ -152,6 +160,51 @@ class TestHostTelemetrySuppressed: assert mins == 1 +# --------------------------------------------------------------------------- +# _state: _host_nodeinfo_suppressed / _mark_host_nodeinfo_seen +# --------------------------------------------------------------------------- + + +class TestHostNodeinfoSuppressed: + """Tests for host NODEINFO suppression logic.""" + + def test_not_suppressed_when_no_previous(self): + """Not suppressed when no previous NODEINFO timestamp is set.""" + assert _state_mod._host_nodeinfo_suppressed(time.monotonic()) is False + + def test_suppressed_within_interval(self): + """Suppressed when within the suppression window.""" + now = time.monotonic() + _state_mod._host_nodeinfo_last_seen = now - 10.0 # 10 seconds ago + assert _state_mod._host_nodeinfo_suppressed(now) is True + + def test_not_suppressed_after_interval(self): + """Not suppressed after the full interval has elapsed.""" + now = time.monotonic() + _state_mod._host_nodeinfo_last_seen = ( + now - _state_mod._HOST_NODEINFO_INTERVAL_SECS - 1.0 + ) + assert _state_mod._host_nodeinfo_suppressed(now) is False + + def test_mark_updates_timestamp(self): + """_mark_host_nodeinfo_seen stores the provided timestamp.""" + now = time.monotonic() + _state_mod._mark_host_nodeinfo_seen(now) + assert _state_mod._host_nodeinfo_last_seen == now + + def test_suppressed_after_mark(self): + """Immediately after marking, a second call is suppressed.""" + now = time.monotonic() + _state_mod._mark_host_nodeinfo_seen(now) + assert _state_mod._host_nodeinfo_suppressed(now + 1.0) is True + + def test_not_suppressed_after_mark_and_full_interval(self): + """After a full interval has elapsed, suppression lifts.""" + long_ago = time.monotonic() - _state_mod._HOST_NODEINFO_INTERVAL_SECS - 5.0 + _state_mod._mark_host_nodeinfo_seen(long_ago) + assert _state_mod._host_nodeinfo_suppressed(time.monotonic()) is False + + # --------------------------------------------------------------------------- # radio: _radio_metadata_fields / _apply_radio_metadata # --------------------------------------------------------------------------- @@ -664,6 +717,91 @@ class TestStoreNodeinfoPacket: q._queue_post_json = original assert sent == [] + def test_host_nodeinfo_not_suppressed_on_first_call(self): + """First NODEINFO from the host node is always forwarded.""" + import data.mesh_ingestor.queue as q + + handlers.register_host_node_id("!aabbccdd") + sent = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append(path) + try: + handlers.store_nodeinfo_packet( + {"id": 1, "rxTime": 100, "fromId": "!aabbccdd"}, + {"user": {"id": "!aabbccdd", "shortName": "AB", "longName": "Alpha"}}, + ) + finally: + q._queue_post_json = original + assert "/api/nodes" in sent + + def test_host_nodeinfo_suppressed_within_window(self): + """Second NODEINFO from the host within the throttle window is dropped.""" + import data.mesh_ingestor.queue as q + + handlers.register_host_node_id("!aabbccdd") + # Simulate a recent upsert so the window is active. + _state_mod._mark_host_nodeinfo_seen(time.monotonic()) + + sent = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append(path) + try: + handlers.store_nodeinfo_packet( + {"id": 2, "rxTime": 200, "fromId": "!aabbccdd"}, + {"user": {"id": "!aabbccdd", "shortName": "AB", "longName": "Alpha"}}, + ) + finally: + q._queue_post_json = original + assert sent == [] + + def test_host_nodeinfo_allowed_after_window_expires(self): + """NODEINFO from the host is forwarded after the throttle window expires.""" + import data.mesh_ingestor.queue as q + + handlers.register_host_node_id("!aabbccdd") + # Place last-seen far in the past so the window has expired. + _state_mod._host_nodeinfo_last_seen = ( + time.monotonic() - _state_mod._HOST_NODEINFO_INTERVAL_SECS - 10.0 + ) + + sent = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append(path) + try: + handlers.store_nodeinfo_packet( + {"id": 3, "rxTime": 300, "fromId": "!aabbccdd"}, + {"user": {"id": "!aabbccdd", "shortName": "AB", "longName": "Alpha"}}, + ) + finally: + q._queue_post_json = original + assert "/api/nodes" in sent + + def test_non_host_nodeinfo_never_suppressed(self): + """NODEINFO from a non-host node is never throttled.""" + import data.mesh_ingestor.queue as q + + handlers.register_host_node_id("!aabbccdd") + # Mark the host as recently seen to activate the throttle. + _state_mod._mark_host_nodeinfo_seen(time.monotonic()) + + sent = [] + original = q._queue_post_json + q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append(path) + try: + handlers.store_nodeinfo_packet( + {"id": 4, "rxTime": 400, "fromId": "!11223344"}, + { + "user": { + "id": "!11223344", + "shortName": "CD", + "longName": "Charlie Delta", + } + }, + ) + finally: + q._queue_post_json = original + assert "/api/nodes" in sent + # --------------------------------------------------------------------------- # store_neighborinfo_packet diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index 7227bf3..1c9b3b7 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -441,6 +441,125 @@ def test_meshcore_node_snapshot_items_with_contacts(monkeypatch): iface.close() +def test_meshcore_node_snapshot_items_includes_self_node_when_cached(monkeypatch): + """node_snapshot_items appends the self-node when _self_info_payload is set.""" + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod, "_run_meshcore", _fake_run_meshcore()) + monkeypatch.setattr(_mod.config, "CONNECTION", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + iface, _, _ = MeshcoreProvider().connect(active_candidate="/dev/ttyUSB0") + + self_pub_key = "deadbeef" + "00" * 28 + iface._self_info_payload = {"public_key": self_pub_key, "name": "SelfNode"} + + items = MeshcoreProvider().node_snapshot_items(iface) + assert len(items) == 1 + node_id, node_dict = items[0] + assert node_id == "!deadbeef" + assert node_dict["user"]["longName"] == "SelfNode" + iface.close() + + +def test_meshcore_node_snapshot_items_excludes_self_node_when_no_payload(monkeypatch): + """node_snapshot_items omits the self-node when no SELF_INFO has been received.""" + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod, "_run_meshcore", _fake_run_meshcore()) + monkeypatch.setattr(_mod.config, "CONNECTION", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + iface, _, _ = MeshcoreProvider().connect(active_candidate="/dev/ttyUSB0") + + assert iface._self_info_payload is None + items = MeshcoreProvider().node_snapshot_items(iface) + assert items == [] + iface.close() + + +def test_meshcore_node_snapshot_items_contacts_and_self(monkeypatch): + """node_snapshot_items includes both contacts and the self-node.""" + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod, "_run_meshcore", _fake_run_meshcore()) + monkeypatch.setattr(_mod.config, "CONNECTION", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + iface, _, _ = MeshcoreProvider().connect(active_candidate="/dev/ttyUSB0") + + contact_pub_key = "aabbccdd" + "00" * 28 + iface._update_contact( + {"public_key": contact_pub_key, "adv_name": "Peer", "last_advert": 1000} + ) + self_pub_key = "deadbeef" + "00" * 28 + iface._self_info_payload = {"public_key": self_pub_key, "name": "Self"} + + items = MeshcoreProvider().node_snapshot_items(iface) + node_ids = {nid for nid, _ in items} + assert "!aabbccdd" in node_ids + assert "!deadbeef" in node_ids + assert len(items) == 2 + iface.close() + + +# --------------------------------------------------------------------------- +# MeshcoreProvider.self_node_item +# --------------------------------------------------------------------------- + + +def test_meshcore_self_node_item_non_interface(): + """self_node_item returns None for any non-_MeshcoreInterface object.""" + assert MeshcoreProvider().self_node_item(object()) is None + + +def test_meshcore_self_node_item_no_payload(monkeypatch): + """self_node_item returns None when no SELF_INFO payload is cached.""" + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod, "_run_meshcore", _fake_run_meshcore()) + monkeypatch.setattr(_mod.config, "CONNECTION", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + iface, _, _ = MeshcoreProvider().connect(active_candidate="/dev/ttyUSB0") + + assert iface._self_info_payload is None + assert MeshcoreProvider().self_node_item(iface) is None + iface.close() + + +def test_meshcore_self_node_item_with_payload(monkeypatch): + """self_node_item returns the correct (node_id, node_dict) when payload cached.""" + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod, "_run_meshcore", _fake_run_meshcore()) + monkeypatch.setattr(_mod.config, "CONNECTION", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + iface, _, _ = MeshcoreProvider().connect(active_candidate="/dev/ttyUSB0") + + pub_key = "deadbeef" + "00" * 28 + iface._self_info_payload = {"public_key": pub_key, "name": "MyHost"} + + result = MeshcoreProvider().self_node_item(iface) + assert result is not None + node_id, node_dict = result + assert node_id == "!deadbeef" + assert node_dict["user"]["longName"] == "MyHost" + assert node_dict["protocol"] == "meshcore" + iface.close() + + +def test_meshcore_self_node_item_empty_key(monkeypatch): + """self_node_item returns None when the cached public_key is empty.""" + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod, "_run_meshcore", _fake_run_meshcore()) + monkeypatch.setattr(_mod.config, "CONNECTION", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + iface, _, _ = MeshcoreProvider().connect(active_candidate="/dev/ttyUSB0") + + # An empty key produces a None node_id from _meshcore_node_id. + iface._self_info_payload = {"public_key": "", "name": "Bad"} + assert MeshcoreProvider().self_node_item(iface) is None + iface.close() + + def test_parse_tcp_target_detects_host_port(): """parse_tcp_target must return (host, port) for host:port strings.""" assert parse_tcp_target("meshnode.local:4403") == ("meshnode.local", 4403) @@ -1419,6 +1538,72 @@ def test_process_self_info_skips_empty_key(): assert registered == [] +def test_process_self_info_caches_payload(): + """_process_self_info must store the payload on iface._self_info_payload.""" + stub = _make_stub_handlers_module() + iface = _MeshcoreInterface(target=None) + payload = {"public_key": "aabbccdd" + "00" * 28, "name": "Host"} + + _process_self_info(payload, iface, stub) + + assert iface._self_info_payload is payload + + +def test_process_self_info_caches_payload_even_when_empty_key(): + """_process_self_info caches the payload even when public_key is empty. + + The payload is cached unconditionally so that radio metadata is always + preserved. self_node_item will still return None for an empty key because + _meshcore_node_id returns None, but the cached payload lets radio metadata + be applied on reconnect without waiting for a second SELF_INFO. + """ + stub = _make_stub_handlers_module() + iface = _MeshcoreInterface(target=None) + payload = {"public_key": "", "name": "Unknown"} + + _process_self_info(payload, iface, stub) + + assert iface._self_info_payload is payload + + +def test_process_self_info_radio_metadata_set_before_upsert(monkeypatch): + """Radio metadata must be written to config BEFORE upsert_node is called. + + Regression test for the ordering bug: previously LORA_FREQ/MODEM_PRESET + were captured after upsert_node, so _apply_radio_metadata_to_nodes found + no values and the first self-node upsert lacked radio metadata. + """ + import data.mesh_ingestor.protocols.meshcore as _mod + + monkeypatch.setattr(_mod.config, "LORA_FREQ", None) + monkeypatch.setattr(_mod.config, "MODEM_PRESET", None) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + + captured_lora_freq_at_upsert: list = [] + captured_modem_preset_at_upsert: list = [] + + def _spy_upsert(node_id, node): + captured_lora_freq_at_upsert.append(_mod.config.LORA_FREQ) + captured_modem_preset_at_upsert.append(_mod.config.MODEM_PRESET) + + stub = _make_stub_handlers_module() + stub.upsert_node = _spy_upsert + + payload = { + "public_key": "aabbccdd" + "00" * 28, + "name": "Host", + "radio_freq": 868.125, + "radio_sf": 8, + "radio_bw": 62.0, + "radio_cr": 8, + } + _process_self_info(payload, _MeshcoreInterface(target=None), stub) + + # Config must have been set before upsert_node was invoked. + assert captured_lora_freq_at_upsert == [pytest.approx(868.125)] + assert captured_modem_preset_at_upsert == ["SF8/BW62/CR8"] + + # --------------------------------------------------------------------------- # _process_self_info — radio metadata capture # ---------------------------------------------------------------------------