Implement CanProtocol for SocketcanBus

Author: lumagi

can/interfaces/socketcan/socketcan.py

@@ -30,7 +30,7 @@
 
 
 import can
-from can import BusABC, Message
+from can import BusABC, CanProtocol, Message
 from can.broadcastmanager import (
     LimitedDurationCyclicSendTaskABC,
     ModifiableCyclicTaskABC,
@@ -710,7 +710,12 @@ def __init__(
                 "local_loopback": local_loopback,
             }
         )
-        super().__init__(channel=channel, can_filters=can_filters, **kwargs)
+        super().__init__(
+            channel=channel,
+            can_filters=can_filters,
+            protocol=CanProtocol.CAN_FD if fd else CanProtocol.CAN_20,
+            **kwargs,
+        )
 
     def shutdown(self) -> None:
         """Stops all active periodic tasks and closes the socket."""

test/test_socketcan.py

@@ -9,6 +9,8 @@
 import warnings
 from unittest.mock import patch
 
+from .config import TEST_INTERFACE_SOCKETCAN
+
 import can
 from can.interfaces.socketcan.constants import (
     CAN_BCM_TX_DELETE,
@@ -357,6 +359,16 @@ def test_build_bcm_update_header(self):
         self.assertEqual(can_id, result.can_id)
         self.assertEqual(1, result.nframes)
 
+    @unittest.skipUnless(TEST_INTERFACE_SOCKETCAN, "Only run when vcan0 is available")
+    def test_bus_creation_can(self):
+        bus = can.Bus(interface="socketcan", channel="vcan0", fd=False)
+        self.assertEqual(bus.protocol, can.CanProtocol.CAN_20)
+
+    @unittest.skipUnless(TEST_INTERFACE_SOCKETCAN, "Only run when vcan0 is available")
+    def test_bus_creation_can_fd(self):
+        bus = can.Bus(interface="socketcan", channel="vcan0", fd=True)
+        self.assertEqual(bus.protocol, can.CanProtocol.CAN_FD)
+
     @unittest.skipUnless(IS_LINUX and IS_PYPY, "Only test when run on Linux with PyPy")
     def test_pypy_socketcan_support(self):
         """Wait for PyPy raw CAN socket support