diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index 6c417a3..ce0eaba 100644 --- a/repeater/data_acquisition/mqtt_handler.py +++ b/repeater/data_acquisition/mqtt_handler.py @@ -1058,11 +1058,22 @@ class MeshCoreToMqttPusher: ) continue result = conn.publish("neighbors", message, retain=False, qos=1) + # This is by far the largest payload the node emits and it is not + # size-capped, so an oversized or queue-full rejection is a real + # outcome. Check paho's rc rather than reporting a publish that + # never left the client as a success. + rc = getattr(result, "rc", None) + if result is None or (rc is not None and rc != mqtt.MQTT_ERR_SUCCESS): + logger.warning( + f"Neighbors publish rejected by {conn.broker['name']} " + f"(rc={rc}, bytes={len(message.encode('utf-8'))})" + ) + continue results.append((conn.broker["name"], result)) _trace(f"Published to {conn.broker['name']} -- neighbors") if not results: - logger.warning("No connected broker opted into the neighbors topic") + logger.warning("Neighbors table was not published to any broker") return results diff --git a/repeater/handler_helpers/mesh_cli.py b/repeater/handler_helpers/mesh_cli.py index 1ecf87e..2d783d4 100644 --- a/repeater/handler_helpers/mesh_cli.py +++ b/repeater/handler_helpers/mesh_cli.py @@ -1,3 +1,4 @@ +import concurrent.futures import logging from pathlib import Path from typing import Any, Callable, Dict, Optional @@ -1324,9 +1325,6 @@ class MeshCLI: if not publisher.enabled(): return "Err - neighbors publishing is disabled (no broker opted in)" - if publisher.status().get("phase") == "active": - return "Err - neighbors cycle already active" - import asyncio loop = self._event_loop @@ -1340,10 +1338,21 @@ class MeshCLI: return "Error: Event loop not available" try: - loop.call_soon_threadsafe( - lambda: asyncio.create_task(publisher.run_cycle(trigger="manual")) - ) - return "OK - neighbor scope discovery started" + # trigger_cycle owns the "already running" check and tracks the task + # so shutdown can cancel it. Doing it here instead would race: this + # runs on the CLI thread, the cycle starts on the event loop. + started = concurrent.futures.Future() + + def _start(): + try: + started.set_result(publisher.trigger_cycle()) + except Exception as exc: # pragma: no cover - defensive + started.set_exception(exc) + + loop.call_soon_threadsafe(_start) + if started.result(timeout=5): + return "OK - neighbor scope discovery started" + return "Err - neighbors cycle already active" except Exception as e: logger.error(f"discover.scopes failed: {e}", exc_info=True) return f"Error: {e}" diff --git a/repeater/handler_helpers/neighbor_scopes.py b/repeater/handler_helpers/neighbor_scopes.py index cab3510..181658d 100644 --- a/repeater/handler_helpers/neighbor_scopes.py +++ b/repeater/handler_helpers/neighbor_scopes.py @@ -127,6 +127,9 @@ class NeighborScopeHelper: self.local_identity = local_identity self.packet_injector = packet_injector self.airtime_manager = airtime_manager + # Held by reference so a live config update is visible; re-read at the + # start of every sweep rather than cached at construction. + self.config = config if config is not None else {} self._pending: Optional[_PendingQuery] = None self._sweep_lock = asyncio.Lock() @@ -135,13 +138,17 @@ class NeighborScopeHelper: self._max_sweep_seconds = DEFAULT_MAX_SWEEP_SECONDS self._duty_cycle_abort_seconds = DEFAULT_DUTY_CYCLE_ABORT_SECONDS self._direct_tx_delay_factor = 0.5 - self.refresh_config(config or {}) + self.refresh_config(self.config) # ------------------------------------------------------------------ # Configuration # ------------------------------------------------------------------ - def refresh_config(self, config: dict) -> None: - """Re-read the tunables so a live config update takes effect next sweep.""" + def refresh_config(self, config: Optional[dict] = None) -> None: + """Re-read the tunables. Called at construction and before each sweep.""" + if config is None: + config = self.config + else: + self.config = config neighbors_cfg = (config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {} self._response_timeout_override = _as_float( @@ -216,6 +223,9 @@ class NeighborScopeHelper: raise RuntimeError("neighbor scope sweep already active") async with self._sweep_lock: + # Pick up live config edits (the mesh CLI can change + # delays.direct_tx_delay_factor, which sizes the response window). + self.refresh_config(self.config) deadline = time.monotonic() + self._max_sweep_seconds timeout = self.response_timeout() logger.info( @@ -276,30 +286,35 @@ class NeighborScopeHelper: ) self._pending = pending + # One finally for the whole method: the injector await spends most of its + # wall time in the engine TX path, so a shutdown cancel lands there. A + # CancelledError escaping with _pending still set would leave a dead query + # matching (and hiding from the companion bridges) every later RESPONSE. try: - # Resolves only once the packet is actually on air (or has failed): - # the engine awaits dispatcher.send_packet under its TX lock and - # defers local TX until the duty cycle allows. This is the firmware's - # logTx / logTxFail boundary, which is where the response deadline is - # armed -- hence the wait_for below, and not a moment earlier. - sent = await self.packet_injector(packet, wait_for_ack=False) - except Exception as e: - self._pending = None - logger.warning(f"Scope request send failed for {target.pubkey[:8]}: {e}") - return ScopeResult(STATUS_SEND_FAILED) + try: + # Resolves only once the packet is actually on air (or has failed): + # the engine awaits dispatcher.send_packet under its TX lock and + # defers local TX until the duty cycle allows. This is the firmware's + # logTx / logTxFail boundary, which is where the response deadline is + # armed -- hence the wait_for below, and not a moment earlier. + sent = await self.packet_injector(packet, wait_for_ack=False) + except asyncio.CancelledError: + raise + except Exception as e: + logger.warning(f"Scope request send failed for {target.pubkey[:8]}: {e}") + return ScopeResult(STATUS_SEND_FAILED) - if not sent: - self._pending = None - logger.debug(f"Scope request not transmitted for {target.pubkey[:8]}") - return ScopeResult(STATUS_SEND_FAILED) + if not sent: + logger.debug(f"Scope request not transmitted for {target.pubkey[:8]}") + return ScopeResult(STATUS_SEND_FAILED) - try: - scopes = await asyncio.wait_for(pending.future, timeout) + try: + scopes = await asyncio.wait_for(pending.future, timeout) + except asyncio.TimeoutError: + logger.debug(f"Scope query timed out for {target.pubkey[:8]}") + return ScopeResult(STATUS_TIMEOUT) logger.debug(f"Scope response from {target.pubkey[:8]}: '{scopes}'") return ScopeResult(STATUS_RESPONDED, scopes) - except asyncio.TimeoutError: - logger.debug(f"Scope query timed out for {target.pubkey[:8]}") - return ScopeResult(STATUS_TIMEOUT) finally: self._pending = None @@ -384,7 +399,11 @@ class NeighborScopeHelper: if int.from_bytes(plaintext[:4], "little") != pending.tag: return False - scopes = bytes(plaintext[8:]).decode("utf-8", errors="replace").rstrip("\x00").strip() + # The responder builds this field as a C string and the block cipher + # zero-pads the tail, so stop at the first NUL exactly as the firmware + # reader does -- rstrip alone would let "DEN\x00junk" through. + raw_scopes = bytes(plaintext[8:]).split(b"\x00", 1)[0] + scopes = raw_scopes.decode("utf-8", errors="replace").strip() if not pending.future.done(): pending.future.set_result(scopes) return True diff --git a/repeater/neighbors_publisher.py b/repeater/neighbors_publisher.py index dc38d40..4c6438b 100644 --- a/repeater/neighbors_publisher.py +++ b/repeater/neighbors_publisher.py @@ -53,6 +53,10 @@ DEFAULT_MAX_NEIGHBOR_AGE_SECONDS = 86400.0 # How often the loop wakes to re-evaluate its schedule. _TICK_SECONDS = 30.0 +# Delay before retrying a cycle that failed or published nothing, instead of +# waiting out the full interval. +RETRY_DELAY_SECONDS = 900.0 + # Phases reported by status(), mirroring the firmware's NeighborsPhase. PHASE_DISABLED = "disabled" PHASE_SCHEDULED = "scheduled" @@ -112,6 +116,7 @@ class NeighborsPublisher: self._self_scopes_fn = self_scopes_fn self._task: Optional[asyncio.Task] = None + self._manual_task: Optional[asyncio.Task] = None self._running = False self._active = False # Discovery responses collected during the current cycle, keyed by pubkey. @@ -179,12 +184,39 @@ class NeighborsPublisher: self._task = asyncio.create_task(self._run_loop(), name="neighbors-publisher") logger.info("Neighbors publisher started (interval %.1fh)", self.interval_seconds / 3600.0) + def trigger_cycle(self) -> bool: + """Start a manual cycle, tracked so shutdown can cancel it. + + A cycle runs for minutes (a discovery window plus one serialized scope + query per neighbour), so an untracked task would keep transmitting while + the daemon tears down. Returns False when a cycle is already running. + """ + if self._active or (self._manual_task is not None and not self._manual_task.done()): + return False + self._manual_task = asyncio.create_task( + self.run_cycle(trigger="manual"), name="neighbors-manual-cycle" + ) + self._manual_task.add_done_callback(self._on_manual_task_done) + return True + + def _on_manual_task_done(self, task: asyncio.Task) -> None: + if self._manual_task is task: + self._manual_task = None + try: + task.result() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Manual neighbors cycle failed: {e}", exc_info=True) + async def stop(self) -> None: self._running = False - task = self._task + tasks = [t for t in (self._task, self._manual_task) if t and not t.done()] self._task = None - if task and not task.done(): + self._manual_task = None + for task in tasks: task.cancel() + for task in tasks: try: await task except asyncio.CancelledError: @@ -261,7 +293,7 @@ class NeighborsPublisher: except Exception as e: logger.error(f"Neighbors publisher cycle failed: {e}", exc_info=True) self._last_result = f"error: {e}" - self._reschedule() + self._reschedule(retry=True) await asyncio.sleep(_TICK_SECONDS) except asyncio.CancelledError: logger.debug("Neighbors publisher loop cancelled") @@ -285,8 +317,16 @@ class NeighborsPublisher: await self.run_cycle(trigger="periodic") - def _reschedule(self) -> None: - self._next_publish_at = time.monotonic() + self.interval_seconds + def _reschedule(self, *, retry: bool = False) -> None: + """Arm the next cycle. + + A cycle that produced no publish retries on a short delay instead of + burning the whole interval: the usual cause is a broker that was briefly + unreachable or rejected the payload, and waiting a day to find out + otherwise is not useful. + """ + delay = min(RETRY_DELAY_SECONDS, self.interval_seconds) if retry else self.interval_seconds + self._next_publish_at = time.monotonic() + delay # ------------------------------------------------------------------ # Cycle @@ -299,6 +339,7 @@ class NeighborsPublisher: self._active = True started = time.monotonic() self._discovery_seen = {} + published = False try: await self._refresh_neighbor_table() targets = self._snapshot_neighbors() @@ -316,15 +357,17 @@ class NeighborsPublisher: self._last_result = ( f"ok ({len(targets)} neighbours, {responded} with scopes)" if published - else "publish failed (no connected broker)" + else "publish failed (broker unreachable or rejected the payload)" ) self._last_publish_at = time.time() logger.info( - "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes", + "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes, " + "published=%s", trigger, time.monotonic() - started, len(targets), responded, + published, ) return { "success": True, @@ -334,7 +377,7 @@ class NeighborsPublisher: } finally: self._active = False - self._reschedule() + self._reschedule(retry=not published) async def _refresh_neighbor_table(self) -> None: """Stage 1: zero-hop node discovery, awaited to completion. diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index c1dbf6f..8b89225 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -2354,16 +2354,36 @@ class APIEndpoints: if not isinstance(raw, dict): return None, "neighbors must be an object" + # Reject unknown keys rather than accepting them silently: a typo would + # otherwise return success while changing nothing the runtime reads. + known_keys = { + "enabled", + "interval_hours", + "discovery_timeout_seconds", + "scope_response_timeout_seconds", + "max_sweep_seconds", + "duty_cycle_abort_seconds", + "max_neighbors", + "max_neighbor_age_seconds", + } + unknown = sorted(set(raw) - known_keys) + if unknown: + return None, f"Unknown neighbors settings: {', '.join(unknown)}" + settings = {} if "enabled" in raw: - settings["enabled"] = bool(raw["enabled"]) + if not isinstance(raw["enabled"], bool): + return None, "neighbors.enabled must be true or false" + settings["enabled"] = raw["enabled"] if "interval_hours" in raw: - try: - interval = int(raw["interval_hours"]) - except (TypeError, ValueError): + interval = raw["interval_hours"] + if isinstance(interval, bool) or not isinstance(interval, (int, float)): return None, "neighbors.interval_hours must be a number" + if float(interval) != int(interval): + return None, "neighbors.interval_hours must be a whole number of hours" + interval = int(interval) if interval < MIN_INTERVAL_HOURS or interval > MAX_INTERVAL_HOURS: return ( None, @@ -2408,6 +2428,24 @@ class APIEndpoints: return None, "neighbors.max_neighbor_age_seconds must be at least 60" settings["max_neighbor_age_seconds"] = max_age + if "max_sweep_seconds" in raw: + try: + max_sweep = float(raw["max_sweep_seconds"]) + except (TypeError, ValueError): + return None, "neighbors.max_sweep_seconds must be a number" + if max_sweep < 30 or max_sweep > 7200: + return None, "neighbors.max_sweep_seconds must be between 30 and 7200" + settings["max_sweep_seconds"] = max_sweep + + if "duty_cycle_abort_seconds" in raw: + try: + abort_after = float(raw["duty_cycle_abort_seconds"]) + except (TypeError, ValueError): + return None, "neighbors.duty_cycle_abort_seconds must be a number" + if abort_after < 0 or abort_after > 600: + return None, "neighbors.duty_cycle_abort_seconds must be between 0 and 600" + settings["duty_cycle_abort_seconds"] = abort_after + return settings, None @cherrypy.expose @@ -2467,6 +2505,18 @@ class APIEndpoints: brokers = data["brokers"] if not isinstance(brokers, list): return self._error("brokers must be a list") + + # The rebuild below is a strict field whitelist, so any key a + # client omits is reset to its default. For neighbors that would + # silently switch the feature off whenever a UI that predates it + # saves an unrelated MQTT setting, so fall back to the stored + # value per broker name instead of to False. + stored_neighbors_by_name = { + str(existing.get("name")): bool(existing.get("neighbors", False)) + for existing in (self.config.get("mqtt_brokers", {}) or {}).get("brokers", []) + if isinstance(existing, dict) and existing.get("name") + } + validated = [] for i, b in enumerate(brokers): if not isinstance(b, dict): @@ -2500,7 +2550,11 @@ class APIEndpoints: "retain_status": bool(b.get("retain_status", False)), # Opt-in per broker; brokers that do not expect the # neighbors topic can reject it and drop the connection. - "neighbors": bool(b.get("neighbors", False)), + "neighbors": bool( + b["neighbors"] + if "neighbors" in b + else stored_neighbors_by_name.get(str(b["name"]).strip(), False) + ), "tls": { "enabled": bool( b.get("tls", {}).get("enabled", True if port == 443 else False) diff --git a/tests/test_mqtt_neighbors.py b/tests/test_mqtt_neighbors.py index af34fe1..f3158cf 100644 --- a/tests/test_mqtt_neighbors.py +++ b/tests/test_mqtt_neighbors.py @@ -42,6 +42,11 @@ from repeater.neighbors_publisher import ( class _FakePacket: """Minimal Packet stand-in for the response-matching path.""" + do_not_retransmit = False + + def mark_do_not_retransmit(self): + self.do_not_retransmit = True + def __init__(self, payload: bytes): self.payload = bytearray(payload) @@ -352,6 +357,215 @@ async def test_response_with_no_query_pending_is_ignored(): assert await helper.process_response_packet(packet) is False +@pytest.mark.asyncio +@pytest.mark.parametrize( + "payload", + [ + b"", + b"\x01", + b"\x01\x02", # header only, no ciphertext + b"\x01\x02\x03\x04\x05", # too short to carry a MAC + block + bytes(64), # right shape, garbage contents + ], +) +async def test_malformed_response_never_raises_or_matches(payload): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + # Truncated/garbage payloads must be rejected quietly, not blow up the + # router thread that offers every RESPONSE to this matcher. + malformed = _FakePacket(payload) + assert await helper.process_response_packet(malformed) is False + return True + + helper = _helper_with_injector( + local, + injector, + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}}, + ) + results = await helper.sweep([target]) + + assert results[target.pubkey].status == STATUS_TIMEOUT + + +@pytest.mark.asyncio +async def test_scopes_are_truncated_at_the_first_nul(): + """The responder builds a C string; the cipher zero-pads the tail.""" + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + response = _make_response_packet(peer, local, helper._pending.tag, "DEN\x00junk") + asyncio.get_running_loop().call_soon( + lambda: asyncio.ensure_future(helper.process_response_packet(response)) + ) + return True + + helper = _helper_with_injector(local, injector) + results = await helper.sweep([target]) + + assert results[target.pubkey].scopes == "DEN" + + +@pytest.mark.asyncio +async def test_cancelling_a_sweep_mid_send_clears_the_pending_query(): + """A leaked pending query would keep hiding RESPONSE packets from companions. + + The injector await is where a shutdown cancel lands: it covers the engine's + TX-delay and duty-cycle deferral, which is most of a query's wall time. + """ + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + await asyncio.sleep(30) # still "transmitting" when the cancel arrives + return True + + helper = _helper_with_injector(local, injector) + task = asyncio.create_task(helper.sweep([target])) + await asyncio.sleep(0.05) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert helper._pending is None + assert helper.active is False + + # A response arriving afterwards must not be consumed by the dead query. + stale = _make_response_packet(peer, local, 1, "DEN") + assert await helper.process_response_packet(stale) is False + + +@pytest.mark.asyncio +async def test_concurrent_sweep_is_rejected(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def injector(packet, wait_for_ack=False): + await asyncio.sleep(0.2) + return True + + helper = _helper_with_injector(local, injector) + first = asyncio.create_task(helper.sweep([target])) + await asyncio.sleep(0.05) + + with pytest.raises(RuntimeError): + await helper.sweep([target]) + + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + + +@pytest.mark.asyncio +async def test_sweep_rereads_live_config(): + """delays.direct_tx_delay_factor is live-updatable and sizes the window.""" + local = LocalIdentity() + config = {"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 5}}} + helper = _helper_with_injector(local, None, config=config) + assert helper.response_timeout() == 5 + + config["mqtt_brokers"]["neighbors"]["scope_response_timeout_seconds"] = 11 + await helper.sweep([NeighborSnapshot(pubkey="aa" * 32, last_seen=time.time())]) + + assert helper.response_timeout() == 11 + + +# ==================================================================== +# Router integration +# ==================================================================== +class _StubRouter: + """Exercises the real PacketRouter RESPONSE branch against a stub daemon.""" + + def __init__(self, scope_helper): + self.fanned_out = [] + self.recorded = [] + self.daemon = SimpleNamespace( + neighbor_scope_helper=scope_helper, + local_hash=0x11, + repeater_handler=None, + ) + + async def _fan_out_to_bridges(self, packet, bridges, context=""): + self.fanned_out.append((packet, dict(bridges), context)) + return (bool(bridges), False) + + def _companion_bridges_for_packet(self, packet, metadata): + return {0x11: object()} + + def _record_for_ui(self, packet, metadata): + self.recorded.append(packet) + + +async def _route_response(router_stub, packet): + from repeater.packet_router import PacketRouter + + return await PacketRouter._route_packet(router_stub, packet) + + +@pytest.mark.asyncio +async def test_router_consumes_only_a_matching_scope_response(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + routed = {} + + async def injector(packet, wait_for_ack=False): + response = _make_response_packet(peer, local, helper._pending.tag, "DEN") + stub = _StubRouter(helper) + await _route_response(stub, response) + routed["consumed"] = stub + return True + + helper = _helper_with_injector(local, injector) + results = await helper.sweep([target]) + + stub = routed["consumed"] + assert results[target.pubkey].status == STATUS_RESPONDED + # Consumed: not retransmitted, recorded, and never offered to a bridge. + assert stub.fanned_out == [] + assert stub.recorded + + +@pytest.mark.asyncio +async def test_router_still_delivers_unrelated_responses_to_companions(): + """A companion's login reply must not be swallowed by an active sweep. + + Same 1-byte dest hash, same instant, different sender: the matcher must + decline it so the companion bridge still sees it. + """ + local = LocalIdentity() + peer = LocalIdentity() + stranger = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + routed = {} + + async def injector(packet, wait_for_ack=False): + foreign = _make_response_packet(stranger, local, helper._pending.tag, "NOPE") + foreign.payload[0] = 0x11 # collide with the companion's dest hash + stub = _StubRouter(helper) + await _route_response(stub, foreign) + routed["stub"] = stub + return True + + helper = _helper_with_injector( + local, + injector, + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}}, + ) + results = await helper.sweep([target]) + + stub = routed["stub"] + assert results[target.pubkey].status == STATUS_TIMEOUT + assert len(stub.fanned_out) == 1 + assert stub.fanned_out[0][2] == "RESPONSE" + + # ==================================================================== # Publish gating # ==================================================================== @@ -624,3 +838,180 @@ async def test_cycle_rejects_reentry_while_active(): result = await publisher.run_cycle() assert result["success"] is False + + +@pytest.mark.asyncio +async def test_failed_publish_retries_sooner_than_a_full_interval(): + """A rejected payload must not cost a whole 24h interval.""" + from repeater.neighbors_publisher import RETRY_DELAY_SECONDS + + handler = SimpleNamespace( + has_neighbors_brokers=lambda: True, + has_connected_neighbors_brokers=lambda: True, + publish_neighbors=lambda payload: [], # nothing reached a broker + node_name="n", + public_key="AB" * 32, + ) + publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=None) + + result = await publisher.run_cycle(trigger="test") + + assert result["published"] is False + assert "publish failed" in publisher._last_result + assert publisher.status()["secs_until_next"] <= RETRY_DELAY_SECONDS + + +def test_enricher_records_only_full_key_repeaters_and_never_self(): + local = LocalIdentity() + publisher = _publisher({"mqtt_brokers": {}}, local_identity=local) + publisher._discovery_seen = {} + + publisher._enrich_discovery_result({"pub_key": "aa" * 32, "node_type": 2, "response_snr": 7.5}) + publisher._enrich_discovery_result({"pub_key": "bb" * 32, "node_type": 1}) # chat node + publisher._enrich_discovery_result({"pub_key": "cc", "node_type": 2}) # prefix only + publisher._enrich_discovery_result( + {"pub_key": local.get_public_key().hex(), "node_type": 2} + ) # self + + assert set(publisher._discovery_seen) == {"aa" * 32} + assert publisher._discovery_seen["aa" * 32]["snr"] == 7.5 + + +def test_enricher_persists_discovery_results_to_storage(): + storage = SimpleNamespace(record_advert=MagicMock(), get_neighbors=lambda: {}) + publisher = _publisher({"mqtt_brokers": {}}, storage=storage) + publisher._discovery_seen = {} + + publisher._enrich_discovery_result( + {"pub_key": "aa" * 32, "node_type": 2, "response_snr": 3.0, "rssi": -90} + ) + + record = storage.record_advert.call_args.args[0] + assert record["pubkey"] == "aa" * 32 + assert record["is_repeater"] is True + assert record["zero_hop"] is True + assert record["snr"] == 3.0 + + +# ==================================================================== +# Config validation +# ==================================================================== +def _validate(raw): + from repeater.web.api_endpoints import APIEndpoints + + return APIEndpoints._validate_neighbors_settings(raw) + + +@pytest.mark.parametrize( + "raw,expect_error", + [ + ({"interval_hours": 24}, False), + ({"interval_hours": MIN_INTERVAL_HOURS}, False), + ({"interval_hours": MAX_INTERVAL_HOURS}, False), + ({"interval_hours": MIN_INTERVAL_HOURS - 1}, True), + ({"interval_hours": MAX_INTERVAL_HOURS + 1}, True), + ({"interval_hours": 12.5}, True), # silently truncating would be worse + ({"interval_hours": "24"}, True), + ({"enabled": True}, False), + ({"enabled": "false"}, True), # would coerce to True + ({"discovery_timeout_seconds": 60}, False), + ({"discovery_timeout_seconds": 1}, True), + ({"scope_response_timeout_seconds": 0}, False), + ({"max_neighbors": 0}, True), + ({"max_neighbor_age_seconds": 10}, True), + ({"max_sweep_seconds": 900}, False), + ({"duty_cycle_abort_seconds": 30}, False), + ({"max_sweep_secondz": 5}, True), # typo must not report success + ("not a dict", True), + ], +) +def test_neighbors_settings_validation(raw, expect_error): + settings, error = _validate(raw) + assert (error is not None) is expect_error + if not expect_error: + assert settings + + +def _api_with_stored_brokers(monkeypatch, brokers, neighbors_block=None): + import cherrypy + + from repeater.web.api_endpoints import APIEndpoints + + request = SimpleNamespace(method="POST", params={}, json={}) + response = SimpleNamespace(headers={}, status=200) + monkeypatch.setattr(cherrypy, "request", request, raising=False) + monkeypatch.setattr(cherrypy, "response", response, raising=False) + + api = APIEndpoints.__new__(APIEndpoints) + api.config = {"mqtt_brokers": {"brokers": brokers, "neighbors": neighbors_block or {}}} + api.daemon_instance = None + api.send_advert_func = None + api.event_loop = None + api.stats_getter = None + api._config_path = "/tmp/test-config.yaml" + api.config_manager = MagicMock() + api.config_manager.update_and_save.return_value = {"success": True, "saved": True} + return api, request + + +def test_broker_neighbors_flag_survives_a_save_that_omits_it(monkeypatch): + """A UI that predates the feature must not silently disable it. + + The broker rebuild is a strict field whitelist, so a client that never learned + about `neighbors` would otherwise reset every broker to False on any unrelated + MQTT save — turning the feature off with no error. + """ + api, request = _api_with_stored_brokers( + monkeypatch, + [{"name": "keeper", "neighbors": True}, {"name": "plain", "neighbors": False}], + ) + + request.json = { + "email": "someone@example.com", + "brokers": [ + {"name": "keeper", "host": "h", "port": 1883, "format": "letsmesh"}, + {"name": "plain", "host": "h", "port": 1883, "format": "letsmesh"}, + ], + } + assert api.update_mqtt_config()["success"] is True + + saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"] + by_name = {b["name"]: b["neighbors"] for b in saved["brokers"]} + assert by_name == {"keeper": True, "plain": False} + + +def test_broker_neighbors_flag_can_be_turned_off_explicitly(monkeypatch): + api, request = _api_with_stored_brokers(monkeypatch, [{"name": "keeper", "neighbors": True}]) + + request.json = { + "brokers": [ + {"name": "keeper", "host": "h", "port": 1883, "format": "letsmesh", "neighbors": False} + ] + } + assert api.update_mqtt_config()["success"] is True + + saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"] + assert saved["brokers"][0]["neighbors"] is False + + +def test_partial_neighbors_post_keeps_unmentioned_settings(monkeypatch): + api, request = _api_with_stored_brokers( + monkeypatch, [], neighbors_block={"interval_hours": 48, "max_neighbors": 8} + ) + + request.json = {"neighbors": {"enabled": False}} + assert api.update_mqtt_config()["success"] is True + + saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"] + assert saved["neighbors"] == {"interval_hours": 48, "max_neighbors": 8, "enabled": False} + + +def test_invalid_neighbors_interval_is_rejected_by_the_endpoint(monkeypatch): + api, request = _api_with_stored_brokers(monkeypatch, []) + + request.json = {"neighbors": {"interval_hours": 1}} + out = api.update_mqtt_config() + + assert out["success"] is False + assert "between 12 and 336" in out["error"] + api.config_manager.update_and_save.assert_not_called()