From daec7f0ebc0bca4695dc66e7a650b63f0a1738f3 Mon Sep 17 00:00:00 2001 From: Rightup Date: Sat, 4 Jul 2026 23:41:33 +0100 Subject: [PATCH] feat: implement neighbor discovery session management and API endpoints --- repeater/handler_helpers/discovery.py | 259 +++++++++++++++ repeater/web/api_endpoints.py | 294 ++++++++++++++++++ repeater/web/auth/middleware.py | 20 ++ repeater/web/openapi.yaml | 157 ++++++++++ tests/test_api_endpoints_core_coverage.py | 183 +++++++++++ ...t_handler_helpers_trace_discovery_login.py | 114 +++++++ 6 files changed, 1027 insertions(+) diff --git a/repeater/handler_helpers/discovery.py b/repeater/handler_helpers/discovery.py index d31c742..42825c6 100644 --- a/repeater/handler_helpers/discovery.py +++ b/repeater/handler_helpers/discovery.py @@ -8,6 +8,10 @@ allowing other nodes to discover repeaters on the mesh network. import asyncio import logging import secrets +import threading +import time +import uuid +from typing import Any, Callable, Optional from openhop_core.node.handlers.control import ControlHandler @@ -22,6 +26,15 @@ logger = logging.getLogger("DiscoveryHelper") # 60s (firmware pending_discover_until = futureMillis(60000)). DEFAULT_DISCOVERY_RESPONSE_JITTER_MS = 2000 +DEFAULT_DISCOVERY_TIMEOUT_SECONDS = 10.0 +DISCOVERY_EVENT_BACKLOG_LIMIT = 512 + +NODE_TYPE_NAMES = { + 1: "Chat Node", + 2: "Repeater", + 3: "Room Server", +} + class DiscoveryHelper: """Helper class for processing discovery requests in the repeater.""" @@ -60,11 +73,257 @@ class DiscoveryHelper: debug_log_fn=debug_log_fn, ) self._pending_tasks = set() + self._sessions: dict[str, dict[str, Any]] = {} + self._sessions_lock = threading.Lock() # Set up the request callback self.control_handler.set_request_callback(self._on_discovery_request) logger.debug("Discovery handler initialized") + def create_session( + self, + *, + timeout: float = DEFAULT_DISCOVERY_TIMEOUT_SECONDS, + filter_mask: int, + since: int = 0, + prefix_only: bool = False, + result_enricher: Optional[Callable[[dict[str, Any]], dict[str, Any]]] = None, + ) -> dict[str, Any]: + """Create a new discovery session and return its public metadata.""" + session_id = uuid.uuid4().hex + tag = secrets.randbits(32) + created_at = time.time() + session = { + "session_id": session_id, + "tag": tag, + "timeout": max(1.0, float(timeout)), + "filter_mask": int(filter_mask) & 0xFF, + "since": max(0, int(since)), + "prefix_only": bool(prefix_only), + "created_at": created_at, + "started_at": None, + "completed_at": None, + "status": "created", + "results": {}, + "events": [], + "next_event_id": 1, + "error": None, + "result_enricher": result_enricher, + } + with self._sessions_lock: + self._sessions[session_id] = session + return self.get_session_snapshot(session_id) or {} + + def get_session_snapshot(self, session_id: str) -> Optional[dict[str, Any]]: + """Return a public snapshot for a discovery session.""" + with self._sessions_lock: + session = self._sessions.get(session_id) + if not session: + return None + return self._public_session_snapshot(session) + + def get_events_since(self, session_id: str, last_event_id: int = 0) -> Optional[dict[str, Any]]: + """Return all session events newer than last_event_id.""" + with self._sessions_lock: + session = self._sessions.get(session_id) + if not session: + return None + events = [event for event in session["events"] if event["id"] > last_event_id] + return { + "events": events, + "status": session["status"], + "completed": session["status"] in {"completed", "timed_out", "error", "cancelled"}, + "latest_event_id": session["next_event_id"] - 1, + } + + async def execute_session(self, session_id: str) -> None: + """Send a discovery request and stream responses into the session.""" + session = self._get_session(session_id) + if not session: + raise ValueError(f"Unknown discovery session: {session_id}") + + if session["status"] != "created": + return + + session["started_at"] = time.time() + session["status"] = "running" + self._emit_event( + session_id, + "started", + { + "session_id": session_id, + "tag": session["tag"], + "timeout": session["timeout"], + "filter_mask": session["filter_mask"], + "since": session["since"], + "prefix_only": session["prefix_only"], + "started_at": session["started_at"], + }, + ) + + try: + from openhop_core.protocol.packet_builder import PacketBuilder + + packet = PacketBuilder.create_discovery_request( + tag=session["tag"], + filter_mask=session["filter_mask"], + since=session["since"], + prefix_only=session["prefix_only"], + ) + + def _response_callback(response_data: dict[str, Any]) -> None: + self._record_response(session_id, response_data) + + self.control_handler.set_response_callback(session["tag"], _response_callback) + + if not self.packet_injector: + raise RuntimeError("No packet injector available") + + success = await self.packet_injector(packet, wait_for_ack=False) + if not success: + raise RuntimeError("Failed to send discovery request") + + logger.info( + "Discovery request sent for session %s tag 0x%08X filter=0x%02X", + session_id, + session["tag"], + session["filter_mask"], + ) + + await asyncio.sleep(session["timeout"]) + self._finish_session(session_id, "completed") + except asyncio.CancelledError: + self._finish_session(session_id, "cancelled") + raise + except Exception as e: + logger.error("Discovery session %s failed: %s", session_id, e, exc_info=True) + self._finish_session(session_id, "error", error=str(e)) + finally: + self.control_handler.clear_response_callback(session["tag"]) + + def start_session_task(self, session_id: str) -> None: + """Schedule a discovery session on the current event loop.""" + task = asyncio.create_task(self.execute_session(session_id)) + self._track_task(task) + + def cleanup_sessions(self, max_age_seconds: int = 120) -> None: + """Remove old completed sessions to keep memory bounded.""" + cutoff = time.time() - max_age_seconds + with self._sessions_lock: + stale_ids = [ + session_id + for session_id, session in self._sessions.items() + if session["status"] in {"completed", "timed_out", "error", "cancelled"} + and (session.get("completed_at") or session.get("created_at", 0)) < cutoff + ] + for session_id in stale_ids: + self._sessions.pop(session_id, None) + + def _get_session(self, session_id: str) -> Optional[dict[str, Any]]: + with self._sessions_lock: + return self._sessions.get(session_id) + + def _record_response(self, session_id: str, response_data: dict[str, Any]) -> None: + with self._sessions_lock: + session = self._sessions.get(session_id) + if not session or session["status"] != "running": + return + + result = dict(response_data) + result["node_type_name"] = NODE_TYPE_NAMES.get( + result.get("node_type"), f"Unknown({result.get('node_type', 0)})" + ) + result["discovered_at"] = time.time() + + enricher = session.get("result_enricher") + if enricher: + try: + result = enricher(result) + except Exception as e: + logger.debug("Discovery result enrichment failed: %s", e) + + result_key = str(result.get("pub_key") or "") + if not result_key: + return + + existing = session["results"].get(result_key) + session["results"][result_key] = result + payload = { + "session_id": session_id, + "tag": session["tag"], + "result": result, + "is_update": existing is not None, + "count": len(session["results"]), + } + self._append_event_unlocked(session, "discovery_result", payload) + + def _finish_session(self, session_id: str, status: str, error: Optional[str] = None) -> None: + with self._sessions_lock: + session = self._sessions.get(session_id) + if not session or session["status"] in {"completed", "timed_out", "error", "cancelled"}: + return + + session["completed_at"] = time.time() + session["status"] = status + session["error"] = error + payload = { + "session_id": session_id, + "tag": session["tag"], + "status": status, + "error": error, + "count": len(session["results"]), + "duration_ms": round( + ( + (session["completed_at"] or session["created_at"]) + - (session["started_at"] or session["created_at"]) + ) + * 1000, + 2, + ), + "completed_at": session["completed_at"], + "results": list(session["results"].values()), + } + event_type = "error" if status == "error" else "completed" + self._append_event_unlocked(session, event_type, payload) + + def _emit_event(self, session_id: str, event_type: str, payload: dict[str, Any]) -> None: + with self._sessions_lock: + session = self._sessions.get(session_id) + if not session: + return + self._append_event_unlocked(session, event_type, payload) + + def _append_event_unlocked( + self, session: dict[str, Any], event_type: str, payload: dict[str, Any] + ) -> None: + event_id = session["next_event_id"] + session["next_event_id"] += 1 + session["events"].append( + { + "id": event_id, + "event": event_type, + "data": payload, + } + ) + if len(session["events"]) > DISCOVERY_EVENT_BACKLOG_LIMIT: + session["events"] = session["events"][-DISCOVERY_EVENT_BACKLOG_LIMIT:] + + def _public_session_snapshot(self, session: dict[str, Any]) -> dict[str, Any]: + return { + "session_id": session["session_id"], + "tag": session["tag"], + "status": session["status"], + "timeout": session["timeout"], + "filter_mask": session["filter_mask"], + "since": session["since"], + "prefix_only": session["prefix_only"], + "created_at": session["created_at"], + "started_at": session["started_at"], + "completed_at": session["completed_at"], + "count": len(session["results"]), + "error": session["error"], + } + def _track_task(self, task: asyncio.Task) -> None: self._pending_tasks.add(task) diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 7be19fa..cc25532 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -122,6 +122,9 @@ POLICY_GROUP_KINDS = { # GET /api/unscoped_flood_policy - Get unscoped flood policy # POST /api/unscoped_flood_policy - Update unscoped flood policy # POST /api/ping_neighbor - Ping a neighbor node +# POST /api/discover_neighbors_start - Start a live repeater discovery session +# GET /api/discover_neighbors_stream?session_id=X - Stream repeater discovery results over SSE +# POST /api/add_discovered_neighbor - Persist a discovered node into the neighbors/adverts table # Identity Management # GET /api/identities - List all identities @@ -303,6 +306,81 @@ class APIEndpoints: end_time = int(time.time()) return end_time - (hours * 3600), end_time + def _encode_sse_event( + self, payload, event_name: Optional[str] = None, event_id: Optional[int] = None + ) -> str: + lines = [] + if event_name: + lines.append(f"event: {event_name}") + if event_id is not None: + lines.append(f"id: {event_id}") + lines.append(f"data: {json.dumps(payload, default=str)}") + return "\n".join(lines) + "\n\n" + + @staticmethod + def _normalize_discovery_node_name(value) -> Optional[str]: + """Normalize placeholder discovery names to None. + + Some peers explicitly advertise "Unknown"-style placeholder names. + Treat those as missing so callers can fall back to hash/prefix labels. + """ + text = str(value or "").strip() + if not text: + return None + + normalized = text.lower() + if normalized in {"unknown", "unknown node"}: + return None + + return text + + def _enrich_discovery_result(self, result: dict) -> dict: + enriched = dict(result) + enriched["node_name"] = self._normalize_discovery_node_name(enriched.get("node_name")) + pub_key = str(enriched.get("pub_key") or "").lower() + if not pub_key: + return enriched + + try: + enriched["node_hash"] = self._fmt_hash(bytes.fromhex(pub_key)) + except ValueError: + enriched["node_hash"] = None + + try: + storage = self._get_storage() + except Exception: + storage = None + + if not storage: + enriched["known_neighbor"] = False + return enriched + + neighbor_info = {} + try: + if hasattr(storage, "get_neighbors"): + neighbor_info = storage.get_neighbors().get(pub_key, {}) or {} + except Exception as exc: + logger.debug("Discovery enrichment could not load neighbors: %s", exc) + + if not neighbor_info: + try: + node_name = storage.get_node_name_by_pubkey(pub_key) + except Exception: + node_name = None + node_name = self._normalize_discovery_node_name(node_name) + enriched["known_neighbor"] = bool(node_name) + if node_name: + enriched["node_name"] = node_name + return enriched + + enriched["known_neighbor"] = True + enriched["node_name"] = self._normalize_discovery_node_name(neighbor_info.get("node_name")) + enriched["contact_type"] = neighbor_info.get("contact_type") + enriched["zero_hop"] = neighbor_info.get("zero_hop") + enriched["last_seen"] = neighbor_info.get("last_seen") + enriched["advert_count"] = neighbor_info.get("advert_count") + return enriched + def _process_counter_data(self, data_points, timestamps_ms): rates = [] prev_value = None @@ -4044,6 +4122,222 @@ class APIEndpoints: logger.error(f"Error pinging neighbor: {e}", exc_info=True) return self._error(str(e)) + @cherrypy.expose + @cherrypy.tools.json_out() + @cherrypy.tools.json_in() + def discover_neighbors_start(self): + + self._set_cors_headers() + + if cherrypy.request.method == "OPTIONS": + return "" + + try: + self._require_post() + data = cherrypy.request.json or {} + timeout = float(data.get("timeout", 5)) + filter_mask = int(data.get("filter_mask", 1 << 2)) + since = int(data.get("since", 0)) + prefix_only = bool(data.get("prefix_only", False)) + + if timeout < 1 or timeout > 60: + return self._error("timeout must be between 1 and 60 seconds") + if filter_mask < 0 or filter_mask > 0xFF: + return self._error("filter_mask must be between 0x00 and 0xFF") + if since < 0: + return self._error("since must be non-negative") + + if self.event_loop is None: + return self._error("Event loop not available") + + discovery_helper = getattr(self.daemon_instance, "discovery_helper", None) + if not discovery_helper: + return self._error("Discovery helper not available") + + discovery_helper.cleanup_sessions() + session = discovery_helper.create_session( + timeout=timeout, + filter_mask=filter_mask, + since=since, + prefix_only=prefix_only, + result_enricher=self._enrich_discovery_result, + ) + + self.event_loop.call_soon_threadsafe( + discovery_helper.start_session_task, session["session_id"] + ) + + return self._success( + session, + message="Discovery session started", + ) + except cherrypy.HTTPError: + raise + except Exception as e: + logger.error("Error starting discovery session: %s", e, exc_info=True) + return self._error(str(e)) + + @cherrypy.expose + def discover_neighbors_stream(self, session_id=None, last_event_id: Optional[str] = None): + self._set_cors_headers() + cherrypy.response.headers["Content-Type"] = "text/event-stream" + cherrypy.response.headers["Cache-Control"] = "no-cache" + cherrypy.response.headers["Connection"] = "keep-alive" + cherrypy.response.headers["X-Accel-Buffering"] = "no" + + discovery_helper = getattr(self.daemon_instance, "discovery_helper", None) + + try: + cursor = int(last_event_id) if last_event_id is not None else 0 + except (TypeError, ValueError): + cursor = 0 + + def generate(): + if not session_id: + yield self._encode_sse_event( + {"type": "error", "error": "Missing session_id"}, + event_name="error", + ) + return + + if not discovery_helper: + yield self._encode_sse_event( + {"type": "error", "error": "Discovery helper not available"}, + event_name="error", + ) + return + + snapshot = discovery_helper.get_session_snapshot(session_id) + if not snapshot: + yield self._encode_sse_event( + {"type": "error", "error": f"Unknown discovery session: {session_id}"}, + event_name="error", + ) + return + + yield self._encode_sse_event( + { + "type": "connected", + "session": snapshot, + }, + event_name="connected", + ) + + current_cursor = cursor + try: + while True: + event_state = discovery_helper.get_events_since(session_id, current_cursor) + if event_state is None: + yield self._encode_sse_event( + { + "type": "error", + "error": f"Unknown discovery session: {session_id}", + }, + event_name="error", + ) + return + + events = event_state.get("events", []) + if events: + for event in events: + current_cursor = max(current_cursor, int(event.get("id", 0))) + yield self._encode_sse_event( + event.get("data", {}), + event_name=event.get("event"), + event_id=event.get("id"), + ) + if event_state.get("completed"): + return + else: + if event_state.get("completed"): + return + yield self._encode_sse_event( + {"type": "keepalive", "session_id": session_id}, + event_name="keepalive", + event_id=current_cursor if current_cursor > 0 else None, + ) + + time.sleep(0.5) + except GeneratorExit: + logger.debug("Discovery SSE stream closed by client for session %s", session_id) + except Exception as exc: + logger.error("Discovery SSE stream error: %s", exc, exc_info=True) + yield self._encode_sse_event( + {"type": "error", "error": str(exc), "session_id": session_id}, + event_name="error", + event_id=current_cursor if current_cursor > 0 else None, + ) + + return generate() + + discover_neighbors_stream._cp_config = {"response.stream": True} + + @cherrypy.expose + @cherrypy.tools.json_out() + @cherrypy.tools.json_in() + @require_auth + def add_discovered_neighbor(self): + + self._set_cors_headers() + + if cherrypy.request.method == "OPTIONS": + return "" + + try: + self._require_post() + data = cherrypy.request.json or {} + pub_key = str(data.get("pub_key") or "").strip().lower() + node_name = self._normalize_discovery_node_name(data.get("node_name")) + node_type = int(data.get("node_type", 0)) + rssi = data.get("rssi") + snr = data.get("response_snr", data.get("snr")) + + if not pub_key: + return self._error("pub_key is required") + if not re.fullmatch(r"[0-9a-f]{16}|[0-9a-f]{64}", pub_key): + return self._error("pub_key must be 8-byte or 32-byte hex") + + contact_type = { + 1: "Chat Node", + 2: "Repeater", + 3: "Room Server", + }.get(node_type, "Unknown") + + advert_record = { + "timestamp": time.time(), + "pubkey": pub_key, + "node_name": node_name, + "is_repeater": node_type == 2, + "route_type": 2, + "contact_type": contact_type, + "latitude": None, + "longitude": None, + "rssi": int(rssi) if rssi is not None else None, + "snr": float(snr) if snr is not None else None, + "is_new_neighbor": True, + "zero_hop": True, + } + + storage = self._get_storage() + storage.record_advert(advert_record) + + enriched = self._enrich_discovery_result( + { + "pub_key": pub_key, + "node_name": node_name, + "node_type": node_type, + "node_type_name": contact_type, + "rssi": advert_record["rssi"], + "response_snr": advert_record["snr"], + } + ) + return self._success(enriched, message="Neighbor added to adverts database") + except cherrypy.HTTPError: + raise + except Exception as e: + logger.error("Error adding discovered neighbor: %s", e, exc_info=True) + return self._error(str(e)) + # ========== Identity Management Endpoints ========== @cherrypy.expose diff --git a/repeater/web/auth/middleware.py b/repeater/web/auth/middleware.py index 54ecc9b..4e7e343 100644 --- a/repeater/web/auth/middleware.py +++ b/repeater/web/auth/middleware.py @@ -39,6 +39,26 @@ def require_auth(func): else: logger.warning("Invalid or expired JWT token") + request_params = getattr(cherrypy.request, "params", None) + if request_params is None: + request_params = {} + + query_token = request_params.get("token") + if query_token: + payload = jwt_handler.verify_jwt(query_token) + + if payload: + cherrypy.request.user = { + "username": payload["sub"], + "client_id": payload["client_id"], + "auth_type": "jwt_query", + } + if hasattr(cherrypy.request, "params") and "token" in cherrypy.request.params: + del cherrypy.request.params["token"] + return func(*args, **kwargs) + else: + logger.warning("Invalid or expired JWT query token") + # Try API token authentication api_key = cherrypy.request.headers.get("X-API-Key", "") if api_key: diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index 7610770..97d0115 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -1600,6 +1600,163 @@ paths: error: type: string + /discover_neighbors_start: + post: + tags: [Network Policy] + summary: Start neighbor discovery session + description: Start a short-lived discovery broadcast session and return a session ID for SSE streaming. + requestBody: + required: false + content: + application/json: + schema: + type: object + properties: + timeout: + type: number + minimum: 1 + maximum: 60 + default: 5 + filter_mask: + type: integer + minimum: 0 + maximum: 255 + default: 4 + since: + type: integer + minimum: 0 + default: 0 + prefix_only: + type: boolean + default: false + responses: + '200': + description: Discovery session started + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + properties: + session_id: + type: string + tag: + type: integer + status: + type: string + timeout: + type: number + filter_mask: + type: integer + since: + type: integer + prefix_only: + type: boolean + created_at: + type: number + started_at: + type: number + nullable: true + completed_at: + type: number + nullable: true + count: + type: integer + error: + type: string + nullable: true + message: + type: string + error: + type: string + + /discover_neighbors_stream: + get: + tags: [Network Policy] + summary: Stream discovery events + description: Stream live neighbor discovery session events via Server-Sent Events (SSE). + security: + - BearerAuth: [] + - ApiKeyAuth: [] + parameters: + - name: session_id + in: query + required: true + schema: + type: string + - name: last_event_id + in: query + required: false + schema: + type: integer + - name: token + in: query + required: false + schema: + type: string + description: JWT token for EventSource clients that cannot send Authorization headers. + responses: + '200': + description: SSE event stream + content: + text/event-stream: + schema: + type: string + + /add_discovered_neighbor: + post: + tags: [Network Policy] + summary: Add discovered node as neighbor + description: Persist a discovered node into the adverts/neighbors store for future management. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [pub_key] + properties: + pub_key: + type: string + description: Hex pubkey (8-byte or 32-byte form). + node_name: + type: string + nullable: true + node_type: + type: integer + rssi: + type: integer + nullable: true + response_snr: + type: number + nullable: true + snr: + type: number + nullable: true + responses: + '200': + description: Node added + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + type: object + message: + type: string + error: + type: string + /policy: get: tags: [Network Policy] diff --git a/tests/test_api_endpoints_core_coverage.py b/tests/test_api_endpoints_core_coverage.py index 0a45a6d..0122b84 100644 --- a/tests/test_api_endpoints_core_coverage.py +++ b/tests/test_api_endpoints_core_coverage.py @@ -25,6 +25,61 @@ def _attach_storage(api, storage): api.daemon_instance = SimpleNamespace(repeater_handler=SimpleNamespace(storage=storage)) +class _FakeDiscoveryHelper: + def __init__(self): + self.cleanup_called = False + self.started_sessions = [] + self._session = { + "session_id": "sess-1", + "tag": 123, + "status": "created", + "timeout": 5.0, + "filter_mask": 0x04, + "since": 0, + "prefix_only": False, + "created_at": 1.0, + "started_at": None, + "completed_at": None, + "count": 0, + "error": None, + } + self._events = [ + {"id": 1, "event": "started", "data": {"session_id": "sess-1"}}, + { + "id": 2, + "event": "discovery_result", + "data": {"session_id": "sess-1", "result": {"pub_key": "aa" * 32}}, + }, + {"id": 3, "event": "completed", "data": {"session_id": "sess-1", "count": 1}}, + ] + + def cleanup_sessions(self): + self.cleanup_called = True + + def create_session(self, **kwargs): + self._session.update( + { + "timeout": kwargs["timeout"], + "filter_mask": kwargs["filter_mask"], + "since": kwargs["since"], + "prefix_only": kwargs["prefix_only"], + } + ) + return dict(self._session) + + def start_session_task(self, session_id): + self.started_sessions.append(session_id) + + def get_session_snapshot(self, session_id): + return dict(self._session) if session_id == self._session["session_id"] else None + + def get_events_since(self, session_id, last_event_id=0): + if session_id != self._session["session_id"]: + return None + events = [event for event in self._events if event["id"] > last_event_id] + return {"events": events, "completed": True, "status": "completed", "latest_event_id": 3} + + @pytest.fixture def cherrypy_ctx(monkeypatch): request = SimpleNamespace(method="GET", params={}, json={}) @@ -148,6 +203,134 @@ def test_success_and_error_helpers(): assert err == {"success": False, "error": "boom"} +def test_discover_neighbors_start_schedules_session(cherrypy_ctx): + request, _ = cherrypy_ctx + request.method = "POST" + request.json = {"timeout": 7, "filter_mask": 0x04, "since": 0, "prefix_only": False} + + helper = _FakeDiscoveryHelper() + loop = MagicMock() + api = _make_api() + api.event_loop = loop + api.daemon_instance = SimpleNamespace(discovery_helper=helper) + + result = api.discover_neighbors_start() + + assert result["success"] is True + assert result["data"]["session_id"] == "sess-1" + assert helper.cleanup_called is True + loop.call_soon_threadsafe.assert_called_once() + + +def test_discover_neighbors_stream_yields_session_events(cherrypy_ctx): + request, response = cherrypy_ctx + request.method = "GET" + + helper = _FakeDiscoveryHelper() + api = _make_api() + api.daemon_instance = SimpleNamespace(discovery_helper=helper) + + stream = api.discover_neighbors_stream(session_id="sess-1") + payload = "".join(list(stream)) + + assert response.headers["Content-Type"] == "text/event-stream" + assert "event: connected" in payload + assert "event: started" in payload + assert "event: discovery_result" in payload + assert "event: completed" in payload + + +def test_add_discovered_neighbor_records_zero_hop_advert(cherrypy_ctx, monkeypatch): + request, _ = cherrypy_ctx + request.method = "POST" + request.headers = {"Authorization": "Bearer test-token"} + request.path_info = "/api/add_discovered_neighbor" + request.json = { + "pub_key": "aa" * 32, + "node_name": "Field Repeater", + "node_type": 2, + "rssi": -71, + "response_snr": 4.5, + } + + jwt_handler = MagicMock() + jwt_handler.verify_jwt.return_value = {"sub": "tester", "client_id": "ui"} + token_manager = MagicMock() + monkeypatch.setattr( + cherrypy, + "config", + {"jwt_handler": jwt_handler, "token_manager": token_manager}, + raising=False, + ) + + storage = MagicMock() + api = _make_api() + api._enrich_discovery_result = lambda result: {**result, "known_neighbor": True} + _attach_storage(api, storage) + + result = api.add_discovered_neighbor() + + assert result["success"] is True + storage.record_advert.assert_called_once() + advert_record = storage.record_advert.call_args.args[0] + assert advert_record["route_type"] == 2 + assert advert_record["zero_hop"] is True + assert advert_record["contact_type"] == "Repeater" + + +def test_add_discovered_neighbor_normalizes_unknown_name(cherrypy_ctx, monkeypatch): + request, _ = cherrypy_ctx + request.method = "POST" + request.headers = {"Authorization": "Bearer test-token"} + request.path_info = "/api/add_discovered_neighbor" + request.json = { + "pub_key": "bb" * 32, + "node_name": "Unknown Node", + "node_type": 2, + } + + jwt_handler = MagicMock() + jwt_handler.verify_jwt.return_value = {"sub": "tester", "client_id": "ui"} + token_manager = MagicMock() + monkeypatch.setattr( + cherrypy, + "config", + {"jwt_handler": jwt_handler, "token_manager": token_manager}, + raising=False, + ) + + storage = MagicMock() + api = _make_api() + api._enrich_discovery_result = lambda result: result + _attach_storage(api, storage) + + result = api.add_discovered_neighbor() + + assert result["success"] is True + advert_record = storage.record_advert.call_args.args[0] + assert advert_record["node_name"] is None + + +def test_enrich_discovery_result_treats_placeholder_name_as_unknown(): + api = _make_api() + + storage = MagicMock() + storage.get_neighbors.return_value = {} + storage.get_node_name_by_pubkey.return_value = "Unknown" + _attach_storage(api, storage) + + result = api._enrich_discovery_result( + { + "pub_key": "cc" * 32, + "node_name": "Unknown Node", + "node_type": 2, + } + ) + + assert result["known_neighbor"] is False + assert result["node_name"] is None + + def test_get_time_range_uses_current_time(monkeypatch): api = _make_api() monkeypatch.setattr("repeater.web.api_endpoints.time.time", lambda: 10_000) diff --git a/tests/test_handler_helpers_trace_discovery_login.py b/tests/test_handler_helpers_trace_discovery_login.py index a923f41..b1329d3 100644 --- a/tests/test_handler_helpers_trace_discovery_login.py +++ b/tests/test_handler_helpers_trace_discovery_login.py @@ -254,6 +254,120 @@ async def test_discovery_response_jitter_disabled_does_not_sleep(): injector.assert_awaited_once() +@pytest.mark.asyncio +async def test_discovery_session_collects_and_completes(): + injector = AsyncMock(return_value=True) + helper = DiscoveryHelper( + local_identity=FakeIdentity(0x42), + packet_injector=injector, + response_jitter_ms=0, + ) + + session = helper.create_session(timeout=0.01, filter_mask=0x04) + session_id = session["session_id"] + + execute_task = asyncio.create_task(helper.execute_session(session_id)) + await asyncio.sleep(0) + + tag = helper.get_session_snapshot(session_id)["tag"] + callback = helper.control_handler._response_callbacks[tag] + callback( + { + "tag": tag, + "node_type": 2, + "inbound_snr": 1.0, + "response_snr": 2.0, + "rssi": -70, + "pub_key": "aa" * 32, + "timestamp": 123.0, + "valid": True, + } + ) + + await execute_task + + snapshot = helper.get_session_snapshot(session_id) + assert snapshot["status"] == "completed" + assert snapshot["count"] == 1 + + event_state = helper.get_events_since(session_id) + event_names = [event["event"] for event in event_state["events"]] + assert "started" in event_names + assert "discovery_result" in event_names + assert "completed" in event_names + + +@pytest.mark.asyncio +async def test_discovery_session_deduplicates_by_pubkey(): + helper = DiscoveryHelper( + local_identity=FakeIdentity(0x42), + packet_injector=AsyncMock(return_value=True), + response_jitter_ms=0, + ) + session = helper.create_session(timeout=1, filter_mask=0x04) + session_id = session["session_id"] + + helper._record_response( + session_id, + { + "tag": session["tag"], + "node_type": 2, + "inbound_snr": 1.0, + "response_snr": 2.0, + "rssi": -70, + "pub_key": "bb" * 32, + "timestamp": 1.0, + }, + ) + helper._record_response( + session_id, + { + "tag": session["tag"], + "node_type": 2, + "inbound_snr": 1.5, + "response_snr": 2.5, + "rssi": -60, + "pub_key": "bb" * 32, + "timestamp": 2.0, + }, + ) + + snapshot = helper.get_session_snapshot(session_id) + assert snapshot["count"] == 0 # session not running yet, responses ignored + + helper._get_session(session_id)["status"] = "running" + helper._record_response( + session_id, + { + "tag": session["tag"], + "node_type": 2, + "inbound_snr": 1.0, + "response_snr": 2.0, + "rssi": -70, + "pub_key": "cc" * 32, + "timestamp": 1.0, + }, + ) + helper._record_response( + session_id, + { + "tag": session["tag"], + "node_type": 2, + "inbound_snr": 1.5, + "response_snr": 2.5, + "rssi": -60, + "pub_key": "cc" * 32, + "timestamp": 2.0, + }, + ) + + snapshot = helper.get_session_snapshot(session_id) + assert snapshot["count"] == 1 + latest_events = helper.get_events_since(session_id)["events"] + result_events = [event for event in latest_events if event["event"] == "discovery_result"] + assert result_events[-1]["data"]["is_update"] is True + + def test_discovery_send_response_without_injector_is_safe(): helper = DiscoveryHelper(local_identity=FakeIdentity(0x42), packet_injector=None)