Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -602,16 +602,27 @@ def _get_connection_error_guidance(self, error: Exception, device: str) -> str:
str: Guidance message specific to the error type, or empty string if no specific guidance.

"""
# Check for permission denied errors on Linux
if isinstance(error, PermissionError) and os_name == "posix" and "/dev/" in device:
# PySerial can wrap the underlying PermissionError in SerialException before
# the MAVLink factory wraps it again in ConnectionError.
permission_denied = isinstance(error, PermissionError) or "permission denied" in str(error).lower()
if permission_denied and os_name == "posix" and "/dev/" in device:
return _(
"Permission denied accessing the serial port. This is common on Linux systems.\n"
"To fix this issue, add your user to the 'dialout' group with the following command:\n"
" sudo adduser $USER dialout\n"
"Then log out and log back in for the changes to take effect."
)

# Add more specific guidance for other error types as needed
device_busy = "device or resource busy" in str(error).lower()
if device_busy and os_name == "posix" and "/dev/" in device:
return _(
"The serial port is already in use by another application.\n"
"Close other ground-control or serial-monitoring applications, then try again.\n"
"To find the process using the port, run:\n"
" lsof {device}\n"
"or:\n"
" fuser {device}"
).format(device=device)

return ""

Expand Down Expand Up @@ -873,12 +884,17 @@ def create_connection_with_retry( # pylint: disable=too-many-arguments, too-man
# Select a supported autopilot
error = self._select_supported_autopilot(detected_vehicles)
if error:
self.disconnect()
return error

# Retrieve autopilot version and banner information
return self._retrieve_autopilot_version_and_banner(timeout)
error = self._retrieve_autopilot_version_and_banner(timeout)
if error:
self.disconnect()
return error

except (ConnectionError, SerialException, PermissionError, ConnectionRefusedError) as e:
self.disconnect()
if log_errors:
logging_warning(_("Connection failed: %s"), e)
logging_error(_("Failed to connect after %d attempts."), retries)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ def create( # pylint: disable=too-many-arguments, too-many-positional-arguments
progress_callback=progress_callback,
autoreconnect=True,
)
except PermissionError:
raise
except (OSError, TimeoutError, ValueError) as exc:
# Preserve the root cause in a ConnectionError so callers can display
# actionable information to the user.
Expand Down
89 changes: 89 additions & 0 deletions tests/test_backend_flightcontroller_connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -923,6 +923,25 @@ def create( # pylint: disable=too-many-arguments, too-many-positional-arguments
assert "Permission denied" in error
assert "/dev/ttyACM0" in error

def test_permission_error_is_preserved_by_mavlink_factory(self) -> None:
"""
The MAVLink factory preserves PermissionError for connection guidance.

GIVEN: PyMAVLink raises PermissionError while opening a device
WHEN: The system MAVLink factory creates a connection
THEN: The original PermissionError should be raised unchanged
"""
factory = SystemMavlinkConnectionFactory()

with (
patch(
"ardupilot_methodic_configurator.backend_flightcontroller_factory_mavlink.mavutil.mavlink_connection",
side_effect=PermissionError(13, "Permission denied"),
),
pytest.raises(PermissionError),
):
factory.create(device="/dev/ttyACM0", baudrate=115200)

def test_connection_with_invalid_device_string(self) -> None:
"""
Connection handles empty device strings gracefully.
Expand Down Expand Up @@ -1485,6 +1504,47 @@ def test_permission_error_on_windows_returns_empty_guidance(self) -> None:

assert guidance == ""

def test_wrapped_permission_error_on_linux_dev_path_returns_guidance(self) -> None:
"""
A wrapped PySerial permission error returns Linux guidance.

GIVEN: A ConnectionError containing PySerial's permission-denied message
WHEN: _get_connection_error_guidance is called on Linux
THEN: A non-empty guidance string about the dialout group should be returned
"""
connection = FlightControllerConnection(info=FlightControllerInfo())
error = ConnectionError(
"/dev/ttyACM1: [Errno 13] could not open port /dev/ttyACM1: [Errno 13] Permission denied: '/dev/ttyACM1'"
)

with patch(
"ardupilot_methodic_configurator.backend_flightcontroller_connection.os_name",
"posix",
):
guidance = connection._get_connection_error_guidance(error, "/dev/ttyACM1")

assert "dialout" in guidance

def test_busy_serial_port_on_linux_dev_path_returns_guidance(self) -> None:
"""
A busy Linux serial port returns guidance for finding its owner.

GIVEN: A connection error reports that a /dev/ serial device is busy
WHEN: _get_connection_error_guidance is called on Linux
THEN: Guidance should explain how to identify the process using it
"""
connection = FlightControllerConnection(info=FlightControllerInfo())
error = ConnectionError("Device or resource busy")

with patch(
"ardupilot_methodic_configurator.backend_flightcontroller_connection.os_name",
"posix",
):
guidance = connection._get_connection_error_guidance(error, "/dev/ttyACM1")

assert "lsof /dev/ttyACM1" in guidance
assert "fuser /dev/ttyACM1" in guidance

def test_non_permission_error_returns_empty_guidance(self) -> None:
"""
Non-PermissionError exceptions return empty guidance.
Expand Down Expand Up @@ -1740,6 +1800,35 @@ def test_create_connection_udp_device_logs_without_baudrate(self) -> None:
assert any("udp:127.0.0.1:14550" in c for c in info_calls)
assert result != "" # error from no heartbeat

def test_create_connection_closes_master_when_no_supported_autopilot(self) -> None:
"""
Failed connection attempts release the serial port.

GIVEN: A MAVLink connection is created but no supported autopilot is found
WHEN: create_connection_with_retry returns the connection error
THEN: The created master connection should be closed
"""
master = Mock()

class MasterFactory(MavlinkConnectionFactory): # pylint: disable=too-few-public-methods, missing-class-docstring
def create( # pylint: disable=too-many-arguments, too-many-positional-arguments
self, device, baudrate=115200, timeout=5.0, retries=3, progress_callback=None
) -> Mock: # type: ignore[override]
return master

connection = FlightControllerConnection(
info=FlightControllerInfo(),
mavlink_connection_factory=MasterFactory(),
)
connection.comport = mavutil.SerialPort(device="/dev/ttyACM0", description="Test")

with patch.object(connection, "_detect_vehicles_from_heartbeats", return_value={}):
result = connection.create_connection_with_retry(progress_callback=None, retries=1, timeout=1)

assert result == "No MAVLink heartbeat received, connection failed."
master.close.assert_called_once()
assert connection.master is None

def test_create_connection_with_retry_raises_connection_error_when_master_is_none(self) -> None:
"""
create_connection_with_retry raises ConnectionError when factory returns None.
Expand Down