tests(backend_flightcontroller): Improved test coverage

Author: amilcarlucas

.codespell-exclude-file

@@ -4,3 +4,4 @@
                 "Version": "LAF"
              datas= [],
                ardupilot_methodic_configuratorAny.datas,
+    critical_param_prefixes = ["FRAME", "BATT", "COMPASS", "SERVO", "MOT"]

ardupilot_methodic_configurator/backend_flightcontroller_connection.py

@@ -27,6 +27,14 @@
 from serial.serialutil import SerialException
 
 from ardupilot_methodic_configurator import _
+from ardupilot_methodic_configurator.backend_flightcontroller_mavlink_factory import (
+    MavlinkConnectionFactory,
+    SystemMavlinkConnectionFactory,
+)
+from ardupilot_methodic_configurator.backend_flightcontroller_serial import (
+    SerialPortDiscovery,
+    SystemSerialPortDiscovery,
+)
 from ardupilot_methodic_configurator.data_model_flightcontroller_info import FlightControllerInfo
 
 if TYPE_CHECKING:
@@ -83,7 +91,7 @@ def close(self) -> None:
 ]
 
 
-class FlightControllerConnection:
+class FlightControllerConnection:  # pylint: disable=too-many-instance-attributes
     """
     Manages flight controller connection establishment and lifecycle.
 
@@ -108,11 +116,13 @@ class FlightControllerConnection:
         "udp:127.0.0.1:14550",
     ]
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         info: FlightControllerInfo,
         baudrate: int = DEFAULT_BAUDRATE,
         network_ports: Optional[list[str]] = None,
+        serial_port_discovery: Optional[SerialPortDiscovery] = None,
+        mavlink_connection_factory: Optional[MavlinkConnectionFactory] = None,
     ) -> None:
         """
         Initialize the connection manager.
@@ -121,6 +131,8 @@ def __init__(
             info: Flight controller information object to populate
             baudrate: Default baud rate for serial connections
             network_ports: Optional list of network ports to try (overrides defaults)
+            serial_port_discovery: Optional serial port discovery service
+            mavlink_connection_factory: Optional MAVLink connection factory service
 
         """
         self.info = info
@@ -129,6 +141,10 @@ def __init__(
         self._baudrate = baudrate
         self._network_ports = network_ports or self.DEFAULT_NETWORK_PORTS
         self._connection_tuples: list[tuple[str, str]] = []
+        self._serial_port_discovery: SerialPortDiscovery = serial_port_discovery or SystemSerialPortDiscovery()
+        self._mavlink_connection_factory: MavlinkConnectionFactory = (
+            mavlink_connection_factory or SystemMavlinkConnectionFactory()
+        )
 
     def discover_connections(self) -> None:
         """
@@ -137,7 +153,7 @@ def discover_connections(self) -> None:
         Populates the list of available serial ports and network ports
         that can be used to connect to a flight controller.
         """
-        comports = self.get_serial_ports()
+        comports = self._serial_port_discovery.get_available_ports()
         netports = self.get_network_ports()
         # list of tuples with the first element being the port name and the second element being the port description
         self._connection_tuples = [(port.device, port.description) for port in comports] + [(port, port) for port in netports]
@@ -314,9 +330,9 @@ def _create_mavlink_connection(  # pylint: disable=too-many-arguments, too-many-
             mavutil.mavlink_connection: The MAVLink connection object
 
         """
-        return mavutil.mavlink_connection(
+        return self._mavlink_connection_factory.create(
             device=device,
-            baud=baudrate,
+            baudrate=baudrate,
             timeout=timeout,
             retries=retries,
             progress_callback=progress_callback,

ardupilot_methodic_configurator/backend_flightcontroller_mavlink_factory.py

@@ -0,0 +1,123 @@
+"""
+MAVLink connection factory service for flight controller connections.
+
+This file is part of ArduPilot Methodic Configurator. https://github.com/ArduPilot/MethodicConfigurator
+
+SPDX-FileCopyrightText: 2024-2025 Amilcar do Carmo Lucas <amilcar.lucas@iav.de>
+
+SPDX-License-Identifier: GPL-3.0-or-later
+"""
+
+from typing import Optional, Protocol
+
+from pymavlink import mavutil
+
+
+class MavlinkConnectionFactory(Protocol):  # pylint: disable=too-few-public-methods
+    """Protocol for creating MAVLink connections."""
+
+    def create(  # pylint: disable=too-many-arguments, too-many-positional-arguments
+        self,
+        device: str,
+        baudrate: int,
+        timeout: float = 5.0,
+        retries: int = 3,
+        progress_callback: Optional[object] = None,
+    ) -> Optional[mavutil.mavlink_connection]:  # pyright: ignore[reportGeneralTypeIssues]
+        """Create a MAVLink connection."""
+        ...  # pylint: disable=unnecessary-ellipsis
+
+
+class SystemMavlinkConnectionFactory:  # pylint: disable=too-few-public-methods
+    """Real implementation using PyMAVLink library."""
+
+    def create(  # pylint: disable=too-many-arguments, too-many-positional-arguments
+        self,
+        device: str,
+        baudrate: int,
+        timeout: float = 5.0,
+        retries: int = 3,
+        progress_callback: Optional[object] = None,
+    ) -> Optional[mavutil.mavlink_connection]:  # pyright: ignore[reportGeneralTypeIssues]
+        """Create connection using actual PyMAVLink library."""
+        try:
+            return mavutil.mavlink_connection(
+                device=device,
+                baud=baudrate,
+                timeout=timeout,
+                retries=retries,
+                progress_callback=progress_callback,
+            )
+        except (OSError, TimeoutError, ValueError):
+            # OSError: connection failures (file not found, permission denied, etc.)
+            # TimeoutError: connection timeout
+            # ValueError: invalid parameter values
+            return None
+
+
+class FakeMavlinkConnectionFactory:
+    """Mock implementation for testing without actual hardware."""
+
+    def __init__(self) -> None:
+        """Initialize mock factory."""
+        self._connections: dict[str, FakeMavlinkConnection] = {}
+
+    def create(  # pylint: disable=too-many-arguments, too-many-positional-arguments
+        self,
+        device: str,
+        baudrate: int,
+        timeout: float = 5.0,  # noqa: ARG002 # pylint: disable=unused-argument
+        retries: int = 3,
+        progress_callback: Optional[object] = None,
+    ) -> Optional["FakeMavlinkConnection"]:
+        """Create a fake MAVLink connection for testing."""
+        conn = FakeMavlinkConnection(device, baudrate)
+        conn.retries = retries
+        conn.progress_callback = progress_callback
+        return conn
+
+    def get_connection(self, device: str) -> Optional["FakeMavlinkConnection"]:
+        """Get a previously created fake connection."""
+        return self._connections.get(device)
+
+
+class FakeMavlinkConnection:
+    """Fake MAVLink connection for testing."""
+
+    retries: int
+    progress_callback: Optional[object]
+
+    def __init__(self, device: str, baudrate: int) -> None:
+        """Initialize fake connection."""
+        self.device = device
+        self.baudrate = baudrate
+        self.connected = True
+        self._message_queue: list[object] = []
+
+    def recv_match(
+        self,
+        blocking: bool = True,  # noqa: ARG002  # pylint: disable=unused-argument
+        timeout: Optional[float] = None,  # noqa: ARG002  # pylint: disable=unused-argument
+    ) -> Optional[object]:
+        """Receive a matched message from queue."""
+        # Note: blocking and timeout parameters are accepted for API compatibility
+        # but not used in fake implementation
+        if self._message_queue:
+            return self._message_queue.pop(0)
+        return None
+
+    def mav_send(self, msg: object) -> None:
+        """Send a MAVLink message (no-op for fake)."""
+        # Note: msg parameter is accepted for API compatibility but not used in fake
+
+    def close(self) -> None:
+        """Close connection."""
+        self.connected = False
+
+    def add_message(self, msg: object) -> None:
+        """Add a message to the queue for testing."""
+        self._message_queue.append(msg)
+
+    def clear_messages(self) -> None:
+        """Clear all queued messages."""
+        self._message_queue.clear()

ardupilot_methodic_configurator/backend_flightcontroller_params.py

@@ -23,7 +23,7 @@
 from ardupilot_methodic_configurator import _
 from ardupilot_methodic_configurator.backend_flightcontroller_mavftp_utils import create_mavftp
 from ardupilot_methodic_configurator.data_model_flightcontroller_info import FlightControllerInfo
-from ardupilot_methodic_configurator.data_model_par_dict import ParDict
+from ardupilot_methodic_configurator.data_model_par_dict import ParDict, validate_param_name
 
 # Type hint for connection manager to avoid circular imports
 if TYPE_CHECKING:
@@ -242,12 +242,24 @@ def set_param(self, param_name: str, param_value: float) -> tuple[bool, str]:
 
         Returns:
             tuple[bool, str]: (True, "") if command sent successfully,
-                             (False, error_message) if no connection available
+                             (False, error_message) if no connection available or invalid parameters
 
         """
         if self.master is None:
             return False, _("No flight controller connection available")
 
+        # Validate parameter name using ArduPilot standards
+        is_valid_name, name_error = validate_param_name(param_name)
+        if not is_valid_name:
+            logging_error(name_error)
+            return False, name_error
+
+        # Validate parameter value
+        if not isinstance(param_value, (int, float)):
+            error_msg = _("Invalid parameter value type: %s (expected numeric)") % type(param_value).__name__
+            logging_error(error_msg)
+            return False, error_msg
+
         self.master.param_set_send(param_name, param_value)
         # Update local cache optimistically
         self.fc_parameters[param_name] = param_value
@@ -276,12 +288,18 @@ def fetch_param(self, param_name: str, timeout: int = 5) -> Optional[float]:
             timeout: Timeout in seconds to wait for the response. Default is 5
 
         Returns:
-            float: The value of the parameter, or None if not found or timeout occurred
+            float: The value of the parameter
 
         """
         if self.master is None:
             return None
 
+        # Validate parameter name using ArduPilot standards
+        is_valid_name, name_error = validate_param_name(param_name)
+        if not is_valid_name:
+            logging_error(name_error)
+            raise ValueError(name_error)
+
         # Send PARAM_REQUEST_READ message
         self.master.mav.param_request_read_send(
             self.master.target_system,

ardupilot_methodic_configurator/backend_flightcontroller_serial.py

@@ -0,0 +1,88 @@
+"""
+Serial port discovery service for flight controller connections.
+
+This file is part of ArduPilot Methodic Configurator. https://github.com/ArduPilot/MethodicConfigurator
+
+SPDX-FileCopyrightText: 2024-2025 Amilcar do Carmo Lucas <amilcar.lucas@iav.de>
+
+SPDX-License-Identifier: GPL-3.0-or-later
+"""
+
+from typing import Protocol
+
+import serial.tools.list_ports
+import serial.tools.list_ports_common
+
+
+class SerialPortDiscovery(Protocol):
+    """Protocol for discovering and managing serial ports."""
+
+    def get_available_ports(
+        self,
+    ) -> list[serial.tools.list_ports_common.ListPortInfo]:
+        """Get list of available serial ports."""
+        ...  # pylint: disable=unnecessary-ellipsis
+
+    def get_port_description(self, device: str) -> str:
+        """Get description for a serial port device."""
+        ...  # pylint: disable=unnecessary-ellipsis
+
+
+class SystemSerialPortDiscovery:
+    """Real implementation using PySerial for hardware discovery."""
+
+    def get_available_ports(
+        self,
+    ) -> list[serial.tools.list_ports_common.ListPortInfo]:
+        """Get actual serial ports from system."""
+        return list(serial.tools.list_ports.comports())
+
+    def get_port_description(self, device: str) -> str:
+        """Get port description from system."""
+        for port in serial.tools.list_ports.comports():
+            if port.device == device:
+                return str(port.description)
+        return device
+
+
+class FakeSerialPortDiscovery:
+    """Mock implementation for testing without physical hardware."""
+
+    def __init__(self) -> None:
+        """Initialize with empty port list."""
+        self._ports: list[serial.tools.list_ports_common.ListPortInfo] = []
+
+    def get_available_ports(
+        self,
+    ) -> list[serial.tools.list_ports_common.ListPortInfo]:
+        """Return configured fake ports."""
+        return self._ports.copy()
+
+    def get_port_description(self, device: str) -> str:
+        """Return fake port description."""
+        for port in self._ports:
+            if port.device == device:
+                return str(port.description)
+        return f"Test Port {device}"
+
+    def add_port(
+        self,
+        device: str,
+        description: str = "Test Serial Port",
+        manufacturer: str = "Test Manufacturer",
+    ) -> None:
+        """Add a fake port for testing."""
+        port_info = _create_mock_port_info(device, description, manufacturer)
+        self._ports.append(port_info)
+
+    def clear_ports(self) -> None:
+        """Clear all configured fake ports."""
+        self._ports.clear()
+
+
+def _create_mock_port_info(device: str, description: str, manufacturer: str) -> serial.tools.list_ports_common.ListPortInfo:
+    """Create a mock ListPortInfo object for testing."""
+    port_info = serial.tools.list_ports_common.ListPortInfo(device)
+    port_info.description = description
+    port_info.manufacturer = manufacturer
+    return port_info

ardupilot_methodic_configurator/data_model_par_dict.py

@@ -24,6 +24,36 @@
 PARAM_NAME_MAX_LEN = 16
 
 
+def validate_param_name(param_name: str) -> tuple[bool, str]:
+    """
+    Validate parameter name according to ArduPilot standards.
+
+    Args:
+        param_name: The parameter name to validate
+
+    Returns:
+        tuple[bool, str]: (is_valid, error_message)
+            is_valid: True if valid, False otherwise
+            error_message: Description of validation error, empty if valid
+
+    """
+    # Check if parameter name is provided and is a string
+    if not param_name or not isinstance(param_name, str):
+        return False, _("Parameter name cannot be empty")
+
+    # Check if parameter name exceeds maximum length
+    if len(param_name) > PARAM_NAME_MAX_LEN:
+        msg = _("Parameter name too long (max %d characters): %s") % (PARAM_NAME_MAX_LEN, param_name)
+        return False, msg
+
+    # Check if parameter name matches the required format
+    if not re.match(PARAM_NAME_REGEX, param_name):
+        msg = _("Invalid parameter name format (must start with capital letter, contain only A-Z, 0-9, _): %s") % param_name
+        return False, msg
+
+    return True, ""
+
+
 def is_within_tolerance(x: float, y: float, atol: float = 1e-08, rtol: float = 1e-04) -> bool:
     """
     Determines if the absolute difference between two numbers is within a specified tolerance.