mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-06-23 11:31:47 +02:00
Better no-connection errors
This commit is contained in:
+36
-2
@@ -2,17 +2,20 @@ import asyncio
|
||||
import glob
|
||||
import logging
|
||||
import platform
|
||||
import re
|
||||
from collections import OrderedDict
|
||||
from contextlib import asynccontextmanager, nullcontext
|
||||
from pathlib import Path
|
||||
|
||||
from meshcore import MeshCore
|
||||
from serial.serialutil import SerialException
|
||||
|
||||
from app.config import settings
|
||||
from app.keystore import clear_keys
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
MAX_FRONTEND_RECONNECT_ERROR_BROADCASTS = 3
|
||||
_SERIAL_PORT_ERROR_RE = re.compile(r"could not open port (?P<port>.+?):")
|
||||
|
||||
|
||||
class RadioOperationError(RuntimeError):
|
||||
@@ -69,6 +72,36 @@ def detect_serial_devices() -> list[str]:
|
||||
return devices
|
||||
|
||||
|
||||
def _extract_serial_port_from_error(exc: Exception) -> str | None:
|
||||
"""Best-effort extraction of a serial port path from a pyserial error."""
|
||||
message = str(exc)
|
||||
match = _SERIAL_PORT_ERROR_RE.search(message)
|
||||
if match:
|
||||
return match.group("port")
|
||||
return None
|
||||
|
||||
|
||||
def _format_reconnect_failure(exc: Exception) -> tuple[str, str, bool]:
|
||||
"""Return log message, frontend detail, and whether to log a traceback."""
|
||||
if settings.connection_type == "serial":
|
||||
if isinstance(exc, RuntimeError) and str(exc).startswith("No MeshCore radio found"):
|
||||
message = (
|
||||
"Could not find a MeshCore radio on any serial port. "
|
||||
"Did the radio get disconnected or change serial ports?"
|
||||
)
|
||||
return (message, message, False)
|
||||
|
||||
if isinstance(exc, SerialException):
|
||||
port = settings.serial_port or _extract_serial_port_from_error(exc) or "the serial port"
|
||||
message = (
|
||||
f"Could not connect to serial port {port}. "
|
||||
"Did the radio get disconnected or change serial ports?"
|
||||
)
|
||||
return (message, message, False)
|
||||
|
||||
return (f"Reconnection failed: {exc}", str(exc), True)
|
||||
|
||||
|
||||
async def test_serial_device(port: str, baudrate: int, timeout: float = 3.0) -> bool:
|
||||
"""Test if a MeshCore radio responds on the given serial port."""
|
||||
mc = None
|
||||
@@ -592,8 +625,9 @@ class RadioManager:
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.warning("Reconnection failed: %s", e, exc_info=True)
|
||||
self._broadcast_reconnect_error_if_needed(str(e))
|
||||
log_message, frontend_detail, include_traceback = _format_reconnect_failure(e)
|
||||
logger.warning(log_message, exc_info=include_traceback)
|
||||
self._broadcast_reconnect_error_if_needed(frontend_detail)
|
||||
return False
|
||||
|
||||
async def start_connection_monitor(self) -> None:
|
||||
|
||||
@@ -6,6 +6,7 @@ import asyncio
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from serial.serialutil import SerialException
|
||||
|
||||
|
||||
class TestRadioManagerConnect:
|
||||
@@ -509,6 +510,43 @@ class TestReconnectLock:
|
||||
for call in mock_broadcast_error.call_args_list:
|
||||
assert call.args == ("Reconnection failed", "radio unavailable")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reconnect_serial_missing_port_logs_clean_message_without_traceback(self):
|
||||
"""Missing serial ports should log a concise operator-facing warning."""
|
||||
from app.radio import RadioManager
|
||||
|
||||
rm = RadioManager()
|
||||
rm.connect = AsyncMock(
|
||||
side_effect=SerialException(
|
||||
2,
|
||||
"could not open port /dev/serial/by-id/test-radio: "
|
||||
"[Errno 2] No such file or directory: '/dev/serial/by-id/test-radio'",
|
||||
)
|
||||
)
|
||||
|
||||
with (
|
||||
patch("app.radio.settings") as mock_settings,
|
||||
patch("app.radio.logger") as mock_logger,
|
||||
patch("app.websocket.broadcast_health"),
|
||||
patch("app.websocket.broadcast_error") as mock_broadcast_error,
|
||||
):
|
||||
mock_settings.connection_type = "serial"
|
||||
mock_settings.serial_port = "/dev/serial/by-id/test-radio"
|
||||
|
||||
result = await rm.reconnect(broadcast_on_success=False)
|
||||
|
||||
assert result is False
|
||||
mock_logger.warning.assert_called_once_with(
|
||||
"Could not connect to serial port /dev/serial/by-id/test-radio. "
|
||||
"Did the radio get disconnected or change serial ports?",
|
||||
exc_info=False,
|
||||
)
|
||||
assert mock_broadcast_error.call_args.args == (
|
||||
"Reconnection failed",
|
||||
"Could not connect to serial port /dev/serial/by-id/test-radio. "
|
||||
"Did the radio get disconnected or change serial ports?",
|
||||
)
|
||||
|
||||
|
||||
class TestManualDisconnectCleanup:
|
||||
"""Tests for manual disconnect teardown behavior."""
|
||||
|
||||
Reference in New Issue
Block a user