Refactor flush_tx_queue for VectorBus

can/interfaces/vector/canlib.py

@@ -878,7 +878,38 @@ def _build_xl_can_tx_event(msg: Message) -> xlclass.XLcanTxEvent:
         return xl_can_tx_event
 
     def flush_tx_buffer(self) -> None:
-        self.xldriver.xlCanFlushTransmitQueue(self.port_handle, self.mask)
+        """
+        Flush the TX buffer of the bus.
+
+        According to XL Driver Library Manual, function xlCanFlushTransmitQueue performs a no-op for
+        Vector devices other than XL family. As a workaround, dummy TX frame with overrun/highprio flag is sent, which
+        clears the TX queue.
+        """
+        if self._can_protocol is CanProtocol.CAN_FD:
+            xl_can_tx_event = xlclass.XLcanTxEvent()
+            xl_can_tx_event.tag = xldefine.XL_CANFD_TX_EventTags.XL_CAN_EV_TAG_TX_MSG
+            xl_can_tx_event.tagData.canMsg.msgFlags |= (
+                xldefine.XL_CANFD_TX_MessageFlags.XL_CAN_TXMSG_FLAG_HIGHPRIO
+            )
+
+            self.xldriver.xlCanTransmitEx(
+                self.port_handle,
+                self.mask,
+                ctypes.c_uint(1),
+                ctypes.c_uint(0),
+                xl_can_tx_event,
+            )
+        else:
+            xl_event = xlclass.XLevent()
+            xl_event.tag = xldefine.XL_EventTags.XL_TRANSMIT_MSG
+            xl_event.tagData.msg.flags |= (
+                xldefine.XL_MessageFlags.XL_CAN_MSG_FLAG_OVERRUN
+                | xldefine.XL_MessageFlags.XL_CAN_MSG_FLAG_WAKEUP
+            )
+
+            self.xldriver.xlCanTransmit(
+                self.port_handle, self.mask, ctypes.c_uint(1), xl_event
+            )
 
     def shutdown(self) -> None:
         super().shutdown()

test/test_vector.py

@@ -610,12 +610,6 @@ def test_receive_fd_non_msg_event() -> None:
     bus.shutdown()
 
 
-def test_flush_tx_buffer_mocked(mock_xldriver) -> None:
-    bus = can.Bus(channel=0, interface="vector", _testing=True)
-    bus.flush_tx_buffer()
-    can.interfaces.vector.canlib.xldriver.xlCanFlushTransmitQueue.assert_called()
-
-
 @pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
 def test_flush_tx_buffer() -> None:
     bus = can.Bus(channel=0, serial=_find_virtual_can_serial(), interface="vector")