test_vector.py

#!/usr/bin/env python

"""
Test for Vector Interface
"""

import ctypes
import functools
import pickle
import sys
import time
from test.config import IS_WINDOWS
from unittest.mock import Mock

import pytest

import can
from can.interfaces.vector import (
    VectorBusParams,
    VectorCanFdParams,
    VectorCanParams,
    VectorChannelConfig,
    VectorError,
    VectorInitializationError,
    VectorOperationError,
    canlib,
    xlclass,
    xldefine,
)

XLDRIVER_FOUND = canlib.xldriver is not None


@pytest.fixture()
def mock_xldriver() -> None:
    # basic mock for XLDriver
    xldriver_mock = Mock()

    # bus creation functions
    xldriver_mock.xlOpenDriver = Mock()
    xldriver_mock.xlGetApplConfig = Mock(side_effect=xlGetApplConfig)
    xldriver_mock.xlGetChannelIndex = Mock(side_effect=xlGetChannelIndex)
    xldriver_mock.xlOpenPort = Mock(side_effect=xlOpenPort)
    xldriver_mock.xlCanFdSetConfiguration = Mock(return_value=0)
    xldriver_mock.xlCanSetChannelMode = Mock(return_value=0)
    xldriver_mock.xlActivateChannel = Mock(return_value=0)
    xldriver_mock.xlGetSyncTime = Mock(side_effect=xlGetSyncTime)
    xldriver_mock.xlCanSetChannelAcceptance = Mock(return_value=0)
    xldriver_mock.xlCanSetChannelBitrate = Mock(return_value=0)
    xldriver_mock.xlSetNotification = Mock(side_effect=xlSetNotification)
    xldriver_mock.xlCanSetChannelOutput = Mock(return_value=0)

    # bus deactivation functions
    xldriver_mock.xlDeactivateChannel = Mock(return_value=0)
    xldriver_mock.xlClosePort = Mock(return_value=0)
    xldriver_mock.xlCloseDriver = Mock()

    # sender functions
    xldriver_mock.xlCanTransmit = Mock(return_value=0)
    xldriver_mock.xlCanTransmitEx = Mock(return_value=0)

    # various functions
    xldriver_mock.xlCanFlushTransmitQueue = Mock()

    # backup unmodified values
    real_xldriver = canlib.xldriver
    real_waitforsingleobject = canlib.WaitForSingleObject
    real_has_events = canlib.HAS_EVENTS

    # set mock
    canlib.xldriver = xldriver_mock
    canlib.HAS_EVENTS = False

    yield

    # cleanup
    canlib.xldriver = real_xldriver
    canlib.WaitForSingleObject = real_waitforsingleobject
    canlib.HAS_EVENTS = real_has_events


def test_listen_only_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", listen_only=True, _testing=True)
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20

    can.interfaces.vector.canlib.xldriver.xlCanSetChannelOutput.assert_called()
    xlCanSetChannelOutput_args = (
        can.interfaces.vector.canlib.xldriver.xlCanSetChannelOutput.call_args[0]
    )
    assert xlCanSetChannelOutput_args[2] == xldefine.XL_OutputMode.XL_OUTPUT_MODE_SILENT


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_listen_only() -> None:
    bus = can.Bus(
        channel=0,
        serial=_find_virtual_can_serial(),
        interface="vector",
        receive_own_messages=True,
        listen_only=True,
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20

    msg = can.Message(
        arbitration_id=0xC0FFEF, data=[1, 2, 3, 4, 5, 6, 7, 8], is_extended_id=True
    )

    bus.send(msg)

    received_msg = bus.recv()

    assert received_msg.arbitration_id == msg.arbitration_id
    assert received_msg.data == msg.data

    bus.shutdown()


def test_bus_creation_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", _testing=True)
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20

    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION.value
    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelBitrate.assert_not_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation() -> None:
    bus = can.Bus(channel=0, serial=_find_virtual_can_serial(), interface="vector")
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20

    bus.shutdown()

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert bus.channel_masks[0] == xl_channel_config.channelMask
    assert (
        xl_channel_config.busParams.data.can.canOpMode
        & xldefine.XL_CANFD_BusParams_CanOpMode.XL_BUS_PARAMS_CANOPMODE_CAN20
    )

    bus = canlib.VectorBus(channel=0, serial=_find_virtual_can_serial())
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20
    bus.shutdown()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_channel_index() -> None:
    channel_index = 1
    bus = can.Bus(
        channel=0,
        serial=_find_virtual_can_serial(),
        channel_index=channel_index,
        interface="vector",
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20
    assert bus.channel_masks[0] == 1 << channel_index

    bus.shutdown()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_multiple_channels() -> None:
    bus = can.Bus(
        channel="0, 1",
        bitrate=1_000_000,
        serial=_find_virtual_can_serial(),
        interface="vector",
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20
    assert len(bus.channels) == 2
    assert bus.mask == 3

    xl_channel_config_0 = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert xl_channel_config_0.busParams.data.can.bitRate == 1_000_000

    xl_channel_config_1 = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=1
    )
    assert xl_channel_config_1.busParams.data.can.bitRate == 1_000_000

    bus.shutdown()


def test_bus_creation_bitrate_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", bitrate=200_000, _testing=True)
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20

    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION.value
    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelBitrate.assert_called()
    xlCanSetChannelBitrate_args = (
        can.interfaces.vector.canlib.xldriver.xlCanSetChannelBitrate.call_args[0]
    )
    assert xlCanSetChannelBitrate_args[2] == 200_000


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_bitrate() -> None:
    bus = can.Bus(
        channel=0,
        serial=_find_virtual_can_serial(),
        interface="vector",
        bitrate=200_000,
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_20

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert xl_channel_config.busParams.data.can.bitRate == 200_000

    bus.shutdown()


def test_bus_creation_fd_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", fd=True, _testing=True)
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_FD

    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert (
        xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION_V4.value
    )
    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelBitrate.assert_not_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_fd() -> None:
    bus = can.Bus(
        channel=0, serial=_find_virtual_can_serial(), interface="vector", fd=True
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_FD

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert (
        xl_channel_config.interfaceVersion
        == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION_V4
    )
    assert (
        xl_channel_config.busParams.data.canFD.canOpMode
        & xldefine.XL_CANFD_BusParams_CanOpMode.XL_BUS_PARAMS_CANOPMODE_CANFD
    )
    bus.shutdown()


def test_bus_creation_fd_bitrate_timings_mocked(mock_xldriver) -> None:
    bus = can.Bus(
        channel=0,
        interface="vector",
        fd=True,
        bitrate=500_000,
        data_bitrate=2_000_000,
        sjw_abr=16,
        tseg1_abr=127,
        tseg2_abr=32,
        sjw_dbr=6,
        tseg1_dbr=27,
        tseg2_dbr=12,
        _testing=True,
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_FD

    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert (
        xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION_V4.value
    )

    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelBitrate.assert_not_called()

    xlCanFdSetConfiguration_args = (
        can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.call_args[0]
    )
    canFdConf = xlCanFdSetConfiguration_args[2]
    assert canFdConf.arbitrationBitRate == 500000
    assert canFdConf.dataBitRate == 2000000
    assert canFdConf.sjwAbr == 16
    assert canFdConf.tseg1Abr == 127
    assert canFdConf.tseg2Abr == 32
    assert canFdConf.sjwDbr == 6
    assert canFdConf.tseg1Dbr == 27
    assert canFdConf.tseg2Dbr == 12


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_fd_bitrate_timings() -> None:
    bus = can.Bus(
        channel=0,
        serial=_find_virtual_can_serial(),
        interface="vector",
        fd=True,
        bitrate=500_000,
        data_bitrate=2_000_000,
        sjw_abr=16,
        tseg1_abr=127,
        tseg2_abr=32,
        sjw_dbr=6,
        tseg1_dbr=27,
        tseg2_dbr=12,
    )

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert (
        xl_channel_config.interfaceVersion
        == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION_V4
    )
    assert (
        xl_channel_config.busParams.data.canFD.canOpMode
        & xldefine.XL_CANFD_BusParams_CanOpMode.XL_BUS_PARAMS_CANOPMODE_CANFD
    )
    assert xl_channel_config.busParams.data.canFD.arbitrationBitRate == 500_000
    assert xl_channel_config.busParams.data.canFD.sjwAbr == 16
    assert xl_channel_config.busParams.data.canFD.tseg1Abr == 127
    assert xl_channel_config.busParams.data.canFD.tseg2Abr == 32
    assert xl_channel_config.busParams.data.canFD.sjwDbr == 6
    assert xl_channel_config.busParams.data.canFD.tseg1Dbr == 27
    assert xl_channel_config.busParams.data.canFD.tseg2Dbr == 12
    assert xl_channel_config.busParams.data.canFD.dataBitRate == 2_000_000

    bus.shutdown()


def test_bus_creation_timing_8mhz_mocked(mock_xldriver) -> None:
    timing = can.BitTiming.from_bitrate_and_segments(
        f_clock=8_000_000,
        bitrate=125_000,
        tseg1=13,
        tseg2=2,
        sjw=1,
    )
    bus = can.Bus(channel=0, interface="vector", timing=timing, _testing=True)
    assert isinstance(bus, canlib.VectorBus)
    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION.value
    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelParamsC200.assert_called()
    btr0, btr1 = (
        can.interfaces.vector.canlib.xldriver.xlCanSetChannelParamsC200.call_args[0]
    )[2:]
    assert btr0 == timing.btr0
    assert btr1 == timing.btr1


def test_bus_creation_timing_16mhz_mocked(mock_xldriver) -> None:
    timing = can.BitTiming.from_bitrate_and_segments(
        f_clock=16_000_000,
        bitrate=125_000,
        tseg1=13,
        tseg2=2,
        sjw=1,
    )
    bus = can.Bus(channel=0, interface="vector", timing=timing, _testing=True)
    assert isinstance(bus, canlib.VectorBus)
    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION.value
    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelParams.assert_called()
    chip_params = (
        can.interfaces.vector.canlib.xldriver.xlCanSetChannelParams.call_args[0]
    )[2]
    assert chip_params.bitRate == 125_000
    assert chip_params.sjw == 1
    assert chip_params.tseg1 == 13
    assert chip_params.tseg2 == 2
    assert chip_params.sam == 1


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_timing() -> None:
    for f_clock in [8_000_000, 16_000_000]:
        timing = can.BitTiming.from_bitrate_and_segments(
            f_clock=f_clock,
            bitrate=125_000,
            tseg1=13,
            tseg2=2,
            sjw=1,
        )
        bus = can.Bus(
            channel=0,
            serial=_find_virtual_can_serial(),
            interface="vector",
            timing=timing,
        )
        assert isinstance(bus, canlib.VectorBus)
        assert bus.protocol == can.CanProtocol.CAN_20

        xl_channel_config = _find_xl_channel_config(
            serial=_find_virtual_can_serial(), channel=0
        )
        assert xl_channel_config.busParams.data.can.bitRate == 125_000
        assert xl_channel_config.busParams.data.can.sjw == 1
        assert xl_channel_config.busParams.data.can.tseg1 == 13
        assert xl_channel_config.busParams.data.can.tseg2 == 2

        bus.shutdown()


def test_bus_creation_timingfd_mocked(mock_xldriver) -> None:
    timing = can.BitTimingFd.from_bitrate_and_segments(
        f_clock=80_000_000,
        nom_bitrate=500_000,
        nom_tseg1=68,
        nom_tseg2=11,
        nom_sjw=10,
        data_bitrate=2_000_000,
        data_tseg1=10,
        data_tseg2=9,
        data_sjw=8,
    )
    bus = can.Bus(
        channel=0,
        interface="vector",
        timing=timing,
        _testing=True,
    )
    assert isinstance(bus, canlib.VectorBus)
    assert bus.protocol == can.CanProtocol.CAN_FD

    can.interfaces.vector.canlib.xldriver.xlOpenDriver.assert_called()
    can.interfaces.vector.canlib.xldriver.xlGetApplConfig.assert_called()

    can.interfaces.vector.canlib.xldriver.xlOpenPort.assert_called()
    xlOpenPort_args = can.interfaces.vector.canlib.xldriver.xlOpenPort.call_args[0]
    assert (
        xlOpenPort_args[5] == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION_V4.value
    )

    assert xlOpenPort_args[6] == xldefine.XL_BusTypes.XL_BUS_TYPE_CAN.value

    can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCanSetChannelBitrate.assert_not_called()

    xlCanFdSetConfiguration_args = (
        can.interfaces.vector.canlib.xldriver.xlCanFdSetConfiguration.call_args[0]
    )
    canFdConf = xlCanFdSetConfiguration_args[2]
    assert canFdConf.arbitrationBitRate == 500_000
    assert canFdConf.dataBitRate == 2_000_000
    assert canFdConf.sjwAbr == 10
    assert canFdConf.tseg1Abr == 68
    assert canFdConf.tseg2Abr == 11
    assert canFdConf.sjwDbr == 8
    assert canFdConf.tseg1Dbr == 10
    assert canFdConf.tseg2Dbr == 9


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_bus_creation_timingfd() -> None:
    timing = can.BitTimingFd.from_bitrate_and_segments(
        f_clock=80_000_000,
        nom_bitrate=500_000,
        nom_tseg1=68,
        nom_tseg2=11,
        nom_sjw=10,
        data_bitrate=2_000_000,
        data_tseg1=10,
        data_tseg2=9,
        data_sjw=8,
    )
    bus = can.Bus(
        channel=0,
        serial=_find_virtual_can_serial(),
        interface="vector",
        timing=timing,
    )

    assert bus.protocol == can.CanProtocol.CAN_FD

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert (
        xl_channel_config.interfaceVersion
        == xldefine.XL_InterfaceVersion.XL_INTERFACE_VERSION_V4
    )
    assert (
        xl_channel_config.busParams.data.canFD.canOpMode
        & xldefine.XL_CANFD_BusParams_CanOpMode.XL_BUS_PARAMS_CANOPMODE_CANFD
    )
    assert xl_channel_config.busParams.data.canFD.arbitrationBitRate == 500_000
    assert xl_channel_config.busParams.data.canFD.sjwAbr == 10
    assert xl_channel_config.busParams.data.canFD.tseg1Abr == 68
    assert xl_channel_config.busParams.data.canFD.tseg2Abr == 11
    assert xl_channel_config.busParams.data.canFD.sjwDbr == 8
    assert xl_channel_config.busParams.data.canFD.tseg1Dbr == 10
    assert xl_channel_config.busParams.data.canFD.tseg2Dbr == 9
    assert xl_channel_config.busParams.data.canFD.dataBitRate == 2_000_000

    bus.shutdown()


def test_send_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", _testing=True)
    msg = can.Message(
        arbitration_id=0xC0FFEF, data=[1, 2, 3, 4, 5, 6, 7, 8], is_extended_id=True
    )
    bus.send(msg)
    can.interfaces.vector.canlib.xldriver.xlCanTransmit.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCanTransmitEx.assert_not_called()


def test_send_fd_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", fd=True, _testing=True)
    msg = can.Message(
        arbitration_id=0xC0FFEF, data=[1, 2, 3, 4, 5, 6, 7, 8], is_extended_id=True
    )
    bus.send(msg)
    can.interfaces.vector.canlib.xldriver.xlCanTransmit.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanTransmitEx.assert_called()


def test_receive_mocked(mock_xldriver) -> None:
    can.interfaces.vector.canlib.xldriver.xlReceive = Mock(side_effect=xlReceive)
    bus = can.Bus(channel=0, interface="vector", _testing=True)
    bus.recv(timeout=0.05)
    can.interfaces.vector.canlib.xldriver.xlReceive.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_not_called()


def test_receive_fd_mocked(mock_xldriver) -> None:
    can.interfaces.vector.canlib.xldriver.xlCanReceive = Mock(side_effect=xlCanReceive)
    bus = can.Bus(channel=0, interface="vector", fd=True, _testing=True)
    bus.recv(timeout=0.05)
    can.interfaces.vector.canlib.xldriver.xlReceive.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_send_and_receive() -> None:
    bus1 = can.Bus(channel=0, serial=_find_virtual_can_serial(), interface="vector")
    bus2 = can.Bus(channel=0, serial=_find_virtual_can_serial(), interface="vector")

    msg_std = can.Message(
        channel=0, arbitration_id=0xFF, data=list(range(8)), is_extended_id=False
    )
    msg_ext = can.Message(
        channel=0, arbitration_id=0xFFFFFF, data=list(range(8)), is_extended_id=True
    )

    bus1.send(msg_std)
    msg_std_recv = bus2.recv(None)
    assert msg_std.equals(msg_std_recv, timestamp_delta=None)

    bus1.send(msg_ext)
    msg_ext_recv = bus2.recv(None)
    assert msg_ext.equals(msg_ext_recv, timestamp_delta=None)

    bus1.shutdown()
    bus2.shutdown()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_send_and_receive_fd() -> None:
    bus1 = can.Bus(
        channel=0, serial=_find_virtual_can_serial(), fd=True, interface="vector"
    )
    bus2 = can.Bus(
        channel=0, serial=_find_virtual_can_serial(), fd=True, interface="vector"
    )

    msg_std = can.Message(
        channel=0,
        arbitration_id=0xFF,
        data=list(range(64)),
        is_extended_id=False,
        is_fd=True,
    )
    msg_ext = can.Message(
        channel=0,
        arbitration_id=0xFFFFFF,
        data=list(range(64)),
        is_extended_id=True,
        is_fd=True,
    )

    bus1.send(msg_std)
    msg_std_recv = bus2.recv(None)
    assert msg_std.equals(msg_std_recv, timestamp_delta=None)

    bus1.send(msg_ext)
    msg_ext_recv = bus2.recv(None)
    assert msg_ext.equals(msg_ext_recv, timestamp_delta=None)

    bus1.shutdown()
    bus2.shutdown()


def test_receive_non_msg_event_mocked(mock_xldriver) -> None:
    can.interfaces.vector.canlib.xldriver.xlReceive = Mock(
        side_effect=xlReceive_chipstate
    )
    bus = can.Bus(channel=0, interface="vector", _testing=True)
    bus.handle_can_event = Mock()
    bus.recv(timeout=0.05)
    can.interfaces.vector.canlib.xldriver.xlReceive.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_not_called()
    bus.handle_can_event.assert_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_receive_non_msg_event() -> None:
    bus = canlib.VectorBus(
        channel=0, serial=_find_virtual_can_serial(), interface="vector"
    )
    bus.handle_can_event = Mock()
    bus.xldriver.xlCanRequestChipState(bus.port_handle, bus.channel_masks[0])
    bus.recv(timeout=0.5)
    bus.handle_can_event.assert_called()
    bus.shutdown()


def test_receive_fd_non_msg_event_mocked(mock_xldriver) -> None:
    can.interfaces.vector.canlib.xldriver.xlCanReceive = Mock(
        side_effect=xlCanReceive_chipstate
    )
    bus = can.Bus(channel=0, interface="vector", fd=True, _testing=True)
    bus.handle_canfd_event = Mock()
    bus.recv(timeout=0.05)
    can.interfaces.vector.canlib.xldriver.xlReceive.assert_not_called()
    can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_called()
    bus.handle_canfd_event.assert_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_receive_fd_non_msg_event() -> None:
    bus = canlib.VectorBus(
        channel=0, serial=_find_virtual_can_serial(), fd=True, interface="vector"
    )
    bus.handle_canfd_event = Mock()
    bus.xldriver.xlCanRequestChipState(bus.port_handle, bus.channel_masks[0])
    bus.recv(timeout=0.5)
    bus.handle_canfd_event.assert_called()
    bus.shutdown()


def test_flush_tx_buffer_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", _testing=True)
    bus.flush_tx_buffer()
    transmit_args = can.interfaces.vector.canlib.xldriver.xlCanTransmit.call_args[0]

    num_msg = transmit_args[2]
    assert num_msg.value == ctypes.c_uint(1).value

    event = transmit_args[3]
    assert isinstance(event, xlclass.XLevent)
    assert event.tag & xldefine.XL_EventTags.XL_TRANSMIT_MSG
    assert event.tagData.msg.flags & (
        xldefine.XL_MessageFlags.XL_CAN_MSG_FLAG_OVERRUN
        | xldefine.XL_MessageFlags.XL_CAN_MSG_FLAG_WAKEUP
    )


def test_flush_tx_buffer_fd_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", fd=True, _testing=True)
    bus.flush_tx_buffer()
    transmit_args = can.interfaces.vector.canlib.xldriver.xlCanTransmitEx.call_args[0]

    num_msg = transmit_args[2]
    assert num_msg.value == ctypes.c_uint(1).value

    num_msg_sent = transmit_args[3]
    assert num_msg_sent.value == ctypes.c_uint(0).value

    event = transmit_args[4]
    assert isinstance(event, xlclass.XLcanTxEvent)
    assert event.tag & xldefine.XL_CANFD_TX_EventTags.XL_CAN_EV_TAG_TX_MSG
    assert (
        event.tagData.canMsg.msgFlags
        & xldefine.XL_CANFD_TX_MessageFlags.XL_CAN_TXMSG_FLAG_HIGHPRIO
    )


@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")
    bus.flush_tx_buffer()
    bus.shutdown()


def test_shutdown_mocked(mock_xldriver) -> None:
    bus = can.Bus(channel=0, interface="vector", _testing=True)
    bus.shutdown()
    can.interfaces.vector.canlib.xldriver.xlDeactivateChannel.assert_called()
    can.interfaces.vector.canlib.xldriver.xlClosePort.assert_called()
    can.interfaces.vector.canlib.xldriver.xlCloseDriver.assert_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_shutdown() -> None:
    bus = can.Bus(channel=0, serial=_find_virtual_can_serial(), interface="vector")

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert xl_channel_config.isOnBus != 0
    bus.shutdown()

    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=0
    )
    assert xl_channel_config.isOnBus == 0


def test_reset_mocked(mock_xldriver) -> None:
    bus = canlib.VectorBus(channel=0, interface="vector", _testing=True)
    bus.reset()
    can.interfaces.vector.canlib.xldriver.xlDeactivateChannel.assert_called()
    can.interfaces.vector.canlib.xldriver.xlActivateChannel.assert_called()


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_reset() -> None:
    bus = canlib.VectorBus(
        channel=0, serial=_find_virtual_can_serial(), interface="vector"
    )
    bus.reset()
    bus.shutdown()


def test_popup_hw_cfg_mocked(mock_xldriver) -> None:
    canlib.xldriver.xlPopupHwConfig = Mock()
    canlib.VectorBus.popup_vector_hw_configuration(10)
    assert canlib.xldriver.xlPopupHwConfig.called
    args, kwargs = canlib.xldriver.xlPopupHwConfig.call_args
    assert isinstance(args[0], ctypes.c_char_p)
    assert isinstance(args[1], ctypes.c_uint)


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_popup_hw_cfg() -> None:
    with pytest.raises(VectorOperationError):
        canlib.VectorBus.popup_vector_hw_configuration(1)


def test_get_application_config_mocked(mock_xldriver) -> None:
    canlib.xldriver.xlGetApplConfig = Mock()
    canlib.VectorBus.get_application_config(app_name="CANalyzer", app_channel=0)
    assert canlib.xldriver.xlGetApplConfig.called


def test_set_application_config_mocked(mock_xldriver) -> None:
    canlib.xldriver.xlSetApplConfig = Mock()
    canlib.VectorBus.set_application_config(
        app_name="CANalyzer",
        app_channel=0,
        hw_type=xldefine.XL_HardwareType.XL_HWTYPE_VN1610,
        hw_index=0,
        hw_channel=0,
    )
    assert canlib.xldriver.xlSetApplConfig.called


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_set_and_get_application_config() -> None:
    xl_channel_config = _find_xl_channel_config(
        serial=_find_virtual_can_serial(), channel=1
    )
    canlib.VectorBus.set_application_config(
        app_name="python-can::test_vector",
        app_channel=5,
        hw_channel=xl_channel_config.hwChannel,
        hw_index=xl_channel_config.hwIndex,
        hw_type=xldefine.XL_HardwareType(xl_channel_config.hwType),
    )
    hw_type, hw_index, hw_channel = canlib.VectorBus.get_application_config(
        app_name="python-can::test_vector",
        app_channel=5,
    )
    assert hw_type == xldefine.XL_HardwareType(xl_channel_config.hwType)
    assert hw_index == xl_channel_config.hwIndex
    assert hw_channel == xl_channel_config.hwChannel


def test_set_timer_mocked(mock_xldriver) -> None:
    canlib.xldriver.xlSetTimerRate = Mock()
    bus = canlib.VectorBus(channel=0, interface="vector", fd=True, _testing=True)
    bus.set_timer_rate(timer_rate_ms=1)
    assert canlib.xldriver.xlSetTimerRate.called


@pytest.mark.skipif(not XLDRIVER_FOUND, reason="Vector XL API is unavailable")
def test_set_timer() -> None:
    bus = canlib.VectorBus(
        channel=0, serial=_find_virtual_can_serial(), interface="vector"
    )
    bus.handle_can_event = Mock()
    bus.set_timer_rate(timer_rate_ms=1)
    t0 = time.perf_counter()
    while time.perf_counter() - t0 < 0.5:
        bus.recv(timeout=-1)

    # call_count is incorrect when using virtual bus
    # assert bus.handle_can_event.call_count > 498
    # assert bus.handle_can_event.call_count < 502


@pytest.mark.skipif(IS_WINDOWS, reason="Not relevant for Windows.")
def test_called_without_testing_argument() -> None:
    """This tests if an exception is thrown when we are not running on Windows."""
    with pytest.raises(can.CanInterfaceNotImplementedError):
        # do not set the _testing argument, since it would suppress the exception
        can.Bus(channel=0, interface="vector")


def test_vector_error_pickle() -> None:
    for error_type in [
        VectorError,
        VectorInitializationError,
        VectorOperationError,
    ]:
        error_code = 118
        error_string = "XL_ERROR"
        function = "function_name"

        exc = error_type(error_code, error_string, function)

        # pickle and unpickle
        p = pickle.dumps(exc)
        exc_unpickled: VectorError = pickle.loads(p)

        assert str(exc) == str(exc_unpickled)
        assert error_code == exc_unpickled.error_code

        with pytest.raises(error_type):
            raise exc_unpickled


def test_vector_subtype_error_from_generic() -> None:
    for error_type in [VectorInitializationError, VectorOperationError]:
        error_code = 118
        error_string = "XL_ERROR"
        function = "function_name"

        generic = VectorError(error_code, error_string, function)

        # pickle and unpickle
        specific: VectorError = error_type.from_generic(generic)

        assert str(generic) == str(specific)
        assert error_code == specific.error_code

        with pytest.raises(error_type):
            raise specific


def test_iterate_channel_index() -> None:
    channel_mask = 0x23  # 100011
    channels = list(canlib._iterate_channel_index(channel_mask))
    assert channels == [0, 1, 5]


@pytest.mark.skipif(
    sys.byteorder != "little", reason="Test relies on little endian data."
)
def test_get_channel_configs() -> None:
    _original_func = canlib._get_xl_driver_config
    canlib._get_xl_driver_config = _get_predefined_xl_driver_config

    channel_configs = canlib.get_channel_configs()
    assert len(channel_configs) == 12

    canlib._get_xl_driver_config = _original_func


@pytest.mark.skipif(
    sys.byteorder != "little", reason="Test relies on little endian data."
)
def test_detect_available_configs() -> None:
    _original_func = canlib._get_xl_driver_config
    canlib._get_xl_driver_config = _get_predefined_xl_driver_config

    available_configs = canlib.VectorBus._detect_available_configs()

    assert len(available_configs) == 5

    assert available_configs[0]["interface"] == "vector"
    assert available_configs[0]["channel"] == 2
    assert available_configs[0]["serial"] == 1001
    assert available_configs[0]["channel_index"] == 2
    assert available_configs[0]["hw_type"] == xldefine.XL_HardwareType.XL_HWTYPE_VN8900
    assert available_configs[0]["hw_index"] == 0
    assert available_configs[0]["supports_fd"] is True
    assert isinstance(
        available_configs[0]["vector_channel_config"], VectorChannelConfig
    )

    canlib._get_xl_driver_config = _original_func


@pytest.mark.skipif(not IS_WINDOWS, reason="Windows specific test")
def test_winapi_availability() -> None:
    assert canlib.WaitForSingleObject is not None
    assert canlib.INFINITE is not None


def test_vector_channel_config_attributes():
    assert hasattr(VectorChannelConfig, "name")
    assert hasattr(VectorChannelConfig, "hw_type")
    assert hasattr(VectorChannelConfig, "hw_index")
    assert hasattr(VectorChannelConfig, "hw_channel")
    assert hasattr(VectorChannelConfig, "channel_index")
    assert hasattr(VectorChannelConfig, "channel_mask")
    assert hasattr(VectorChannelConfig, "channel_capabilities")
    assert hasattr(VectorChannelConfig, "channel_bus_capabilities")
    assert hasattr(VectorChannelConfig, "is_on_bus")
    assert hasattr(VectorChannelConfig, "bus_params")
    assert hasattr(VectorChannelConfig, "connected_bus_type")
    assert hasattr(VectorChannelConfig, "serial_number")
    assert hasattr(VectorChannelConfig, "article_number")
    assert hasattr(VectorChannelConfig, "transceiver_name")


def test_vector_bus_params_attributes():
    assert hasattr(VectorBusParams, "bus_type")
    assert hasattr(VectorBusParams, "can")
    assert hasattr(VectorBusParams, "canfd")


def test_vector_can_params_attributes():
    assert hasattr(VectorCanParams, "bitrate")
    assert hasattr(VectorCanParams, "sjw")
    assert hasattr(VectorCanParams, "tseg1")
    assert hasattr(VectorCanParams, "tseg2")
    assert hasattr(VectorCanParams, "sam")
    assert hasattr(VectorCanParams, "output_mode")
    assert hasattr(VectorCanParams, "can_op_mode")


def test_vector_canfd_params_attributes():
    assert hasattr(VectorCanFdParams, "bitrate")
    assert hasattr(VectorCanFdParams, "data_bitrate")
    assert hasattr(VectorCanFdParams, "sjw_abr")
    assert hasattr(VectorCanFdParams, "tseg1_abr")
    assert hasattr(VectorCanFdParams, "tseg2_abr")
    assert hasattr(VectorCanFdParams, "sam_abr")
    assert hasattr(VectorCanFdParams, "sjw_dbr")
    assert hasattr(VectorCanFdParams, "tseg1_dbr")
    assert hasattr(VectorCanFdParams, "tseg2_dbr")
    assert hasattr(VectorCanFdParams, "output_mode")
    assert hasattr(VectorCanFdParams, "can_op_mode")