tests: drivers: can: host: apply ruff format and fixups

henrikbrixandersen · jhedberg · commit 6a95243e59cc · 2025-10-31T22:40:58.000+02:00
Apply ruff format and fixups to the CAN host test suite pytest files.

Signed-off-by: Henrik Brix Andersen &lt;henrik@brixandersen.dk&gt;
diff --git a/.ruff-excludes.toml b/.ruff-excludes.toml
@@ -1088,17 +1088,6 @@
     "I001",     # https://docs.astral.sh/ruff/rules/unsorted-imports
     "UP031",    # https://docs.astral.sh/ruff/rules/printf-string-formatting
 ]
-"./tests/drivers/can/host/pytest/can_shell.py" = [
-    "I001",     # https://docs.astral.sh/ruff/rules/unsorted-imports
-    "SIM401",   # https://docs.astral.sh/ruff/rules/if-else-block-instead-of-dict-get
-    "UP006",    # https://docs.astral.sh/ruff/rules/non-pep585-annotation
-    "UP035",    # https://docs.astral.sh/ruff/rules/deprecated-import
-    "UP045",    # https://docs.astral.sh/ruff/rules/non-pep604-annotation-optional
-]
-"./tests/drivers/can/host/pytest/test_can.py" = [
-    "I001",     # https://docs.astral.sh/ruff/rules/unsorted-imports
-    "UP039",    # https://docs.astral.sh/ruff/rules/unnecessary-class-parentheses
-]
 "./tests/kernel/timer/timer_behavior/pytest/saleae_logic2.py" = [
     "B905",     # https://docs.astral.sh/ruff/rules/zip-without-explicit-strict
     "I001",     # https://docs.astral.sh/ruff/rules/unsorted-imports
@@ -1481,9 +1470,6 @@ exclude = [
     "./tests/boot/with_mcumgr/pytest/test_downgrade_prevention.py",
     "./tests/boot/with_mcumgr/pytest/test_upgrade.py",
     "./tests/boot/with_mcumgr/pytest/west_sign_wrapper.py",
-    "./tests/drivers/can/host/pytest/can_shell.py",
-    "./tests/drivers/can/host/pytest/conftest.py",
-    "./tests/drivers/can/host/pytest/test_can.py",
     "./tests/kernel/timer/timer_behavior/pytest/saleae_logic2.py",
     "./tests/kernel/timer/timer_behavior/pytest/test_timer.py",
     "./tests/lib/devicetree/memory_region_flags/pytest/test_memory_region_flags.py",
diff --git a/tests/drivers/can/host/pytest/can_shell.py b/tests/drivers/can/host/pytest/can_shell.py
@@ -6,25 +6,30 @@
 Zephyr CAN shell module support for providing a python-can bus interface for testing.
 """
 
-import re
 import logging
-from typing import Optional, Tuple
+import re
 
 from can import BusABC, CanProtocol, Message
 from can.exceptions import CanInitializationError, CanOperationError
 from can.typechecking import CanFilters
-
 from twister_harness import DeviceAdapter, Shell
 
 logger = logging.getLogger(__name__)
 
-class CanShellBus(BusABC): # pylint: disable=abstract-method
+
+class CanShellBus(BusABC):  # pylint: disable=abstract-method
     """
     A CAN interface using the Zephyr CAN shell module.
     """
 
-    def __init__(self, dut: DeviceAdapter, shell: Shell, channel: str,
-                 can_filters: Optional[CanFilters] = None, **kwargs) -> None:
+    def __init__(
+        self,
+        dut: DeviceAdapter,
+        shell: Shell,
+        channel: str,
+        can_filters: CanFilters | None = None,
+        **kwargs,
+    ) -> None:
         self._dut = dut
         self._shell = shell
         self._device = channel
@@ -76,7 +81,7 @@ def _start(self):
     def _stop(self):
         self._shell.exec_command(f'can stop {self._device}')
 
-    def send(self, msg: Message, timeout: Optional[float] = None) -> None:
+    def send(self, msg: Message, timeout: float | None = None) -> None:
         logger.debug('sending: %s', msg)
 
         cmd = f'can send {self._device}'
@@ -146,7 +151,7 @@ def _remove_all_filters(self) -> None:
         for filter_id in self._filter_ids[:]:
             self._remove_filter(filter_id)
 
-    def _apply_filters(self, filters: Optional[CanFilters]) -> None:
+    def _apply_filters(self, filters: CanFilters | None) -> None:
         self._remove_all_filters()
 
         if filters:
@@ -155,19 +160,23 @@ def _apply_filters(self, filters: Optional[CanFilters]) -> None:
             # Accept all frames if no hardware filters provided
             filters = [
                 {'can_id': 0x0, 'can_mask': 0x0},
-                {'can_id': 0x0, 'can_mask': 0x0, 'extended': True}
+                {'can_id': 0x0, 'can_mask': 0x0, 'extended': True},
             ]
             self._is_filtered = False
 
         for can_filter in filters:
             can_id = can_filter['can_id']
             can_mask = can_filter['can_mask']
-            extended = can_filter['extended'] if 'extended' in can_filter else False
+            extended = can_filter.get('extended', False)
             self._add_filter(can_id, can_mask, extended)
 
-    def _recv_internal(self, timeout: Optional[float]) -> Tuple[Optional[Message], bool]:
-        frame_regex = r'.*' + re.escape(self._device) + \
-            r'\s+(?P<brs>\S)(?P<esi>\S)\s+(?P<can_id>\d+)\s+\[(?P<dlc>\d+)\]\s*(?P<data>[a-z0-9 ]*)'
+    def _recv_internal(self, timeout: float | None) -> tuple[Message | None, bool]:
+        frame_regex = (
+            r'.*'
+            + re.escape(self._device)
+            + r'\s+(?P<brs>\S)(?P<esi>\S)\s+(?P<can_id>\d+)\s+'
+            + r'\[(?P<dlc>\d+)\]\s*(?P<data>[a-z0-9 ]*)'
+        )
         lines = self._dut.readlines_until(regex=frame_regex, timeout=timeout)
         msg = None
 
@@ -182,10 +191,17 @@ def _recv_internal(self, timeout: Optional[float]) -> Tuple[Optional[Message], b
                 brs = m.group('brs') == 'B'
                 esi = m.group('esi') == 'P'
                 data = bytearray.fromhex(m.group('data'))
-                msg = Message(arbitration_id=can_id,is_extended_id=ext,
-                              data=data, dlc=dlc,
-                              is_fd=fd, bitrate_switch=brs, error_state_indicator=esi,
-                              channel=self._device, check=True)
+                msg = Message(
+                    arbitration_id=can_id,
+                    is_extended_id=ext,
+                    data=data,
+                    dlc=dlc,
+                    is_fd=fd,
+                    bitrate_switch=brs,
+                    error_state_indicator=esi,
+                    channel=self._device,
+                    check=True,
+                )
                 logger.debug('received: %s', msg)
 
         return msg, self._is_filtered
diff --git a/tests/drivers/can/host/pytest/conftest.py b/tests/drivers/can/host/pytest/conftest.py
@@ -16,10 +16,15 @@
 
 logger = logging.getLogger(__name__)
 
+
 def pytest_addoption(parser) -> None:
     """Add local parser options to pytest."""
-    parser.addoption('--can-context', default=None,
-                     help='Configuration context to use for python-can (default: None)')
+    parser.addoption(
+        '--can-context',
+        default=None,
+        help='Configuration context to use for python-can (default: None)',
+    )
+
 
 @pytest.fixture(name='context', scope='session')
 def fixture_context(request, dut: DeviceAdapter) -> str:
@@ -35,6 +40,7 @@ def fixture_context(request, dut: DeviceAdapter) -> str:
     logger.info('using python-can configuration context "%s"', ctx)
     return ctx
 
+
 @pytest.fixture(name='chosen', scope='module')
 def fixture_chosen(shell: Shell) -> str:
     """Return the name of the zephyr,canbus devicetree chosen device."""
@@ -51,6 +57,7 @@ def fixture_chosen(shell: Shell) -> str:
     pytest.fail('zephyr,canbus chosen device not found or not ready')
     return None
 
+
 @pytest.fixture
 def can_dut(dut: DeviceAdapter, shell: Shell, chosen: str) -> BusABC:
     """Return DUT CAN bus."""
@@ -59,9 +66,10 @@ def can_dut(dut: DeviceAdapter, shell: Shell, chosen: str) -> BusABC:
     bus.shutdown()
     dut.clear_buffer()
 
+
 @pytest.fixture
 def can_host(context: str) -> BusABC:
     """Return host CAN bus."""
-    bus = Bus(config_context = context)
+    bus = Bus(config_context=context)
     yield bus
     bus.shutdown()
diff --git a/tests/drivers/can/host/pytest/test_can.py b/tests/drivers/can/host/pytest/test_can.py
@@ -7,50 +7,55 @@
 """
 
 import logging
-import pytest
 
 import can
+import pytest
 from can import BusABC, CanProtocol
 
 # RX/TX timeout in seconds
 TIMEOUT = 1.0
 
 logger = logging.getLogger(__name__)
 
-@pytest.mark.parametrize('msg', [
-    pytest.param(
-        can.Message(arbitration_id=0x10,
-                    is_extended_id=False),
-        id='std_id_dlc_0'
-    ),
-    pytest.param(
-        can.Message(arbitration_id=0x20,
-                    data=[0xaa, 0xbb, 0xcc, 0xdd],
-                    is_extended_id=False),
-        id='std_id_dlc_4'
-    ),
-    pytest.param(
-        can.Message(arbitration_id=0x30,
-                    data=[0xee, 0xff, 0xee, 0xff, 0xee, 0xff, 0xee, 0xff],
-                    is_extended_id=True),
-        id='ext_id_dlc_8'
-    ),
-    pytest.param(
-        can.Message(arbitration_id=0x40,
-                    data=[0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09,
-                          0x10, 0x11],
-                    is_fd=True, is_extended_id=False),
-        id='std_id_fdf_dlc_9'
-    ),
-    pytest.param(
-        can.Message(arbitration_id=0x50,
-                    data=[0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09,
-                          0x10, 0x11],
-                    is_fd=True, bitrate_switch=True, is_extended_id=False),
-        id='std_id_fdf_brs_dlc_9'
-    ),
-])
-class TestCanRxTx():
+
+@pytest.mark.parametrize(
+    'msg',
+    [
+        pytest.param(can.Message(arbitration_id=0x10, is_extended_id=False), id='std_id_dlc_0'),
+        pytest.param(
+            can.Message(arbitration_id=0x20, data=[0xAA, 0xBB, 0xCC, 0xDD], is_extended_id=False),
+            id='std_id_dlc_4',
+        ),
+        pytest.param(
+            can.Message(
+                arbitration_id=0x30,
+                data=[0xEE, 0xFF, 0xEE, 0xFF, 0xEE, 0xFF, 0xEE, 0xFF],
+                is_extended_id=True,
+            ),
+            id='ext_id_dlc_8',
+        ),
+        pytest.param(
+            can.Message(
+                arbitration_id=0x40,
+                data=[0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x10, 0x11],
+                is_fd=True,
+                is_extended_id=False,
+            ),
+            id='std_id_fdf_dlc_9',
+        ),
+        pytest.param(
+            can.Message(
+                arbitration_id=0x50,
+                data=[0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x10, 0x11],
+                is_fd=True,
+                bitrate_switch=True,
+                is_extended_id=False,
+            ),
+            id='std_id_fdf_brs_dlc_9',
+        ),
+    ],
+)
+class TestCanRxTx:
     """
     Class for testing CAN RX/TX between Zephyr DUT and host.
     """
@@ -64,8 +69,7 @@ def check_rx(tx: can.Message, rx: can.Message) -> None:
         if rx is None:
             pytest.fail('no message received')
 
-        if not tx.equals(rx, timestamp_delta=None, check_channel=False,
-                         check_direction=False):
+        if not tx.equals(rx, timestamp_delta=None, check_channel=False, check_direction=False):
             pytest.fail(f'rx message "{rx}" not equal to tx message "{tx}"')
 
     @staticmethod
