test_pcan.py

"""
Test for PCAN Interface
"""

import ctypes
import struct
import unittest
from unittest import mock
from unittest.mock import Mock, patch

import pytest
from parameterized import parameterized

import can
from can import BusState, CanProtocol
from can.exceptions import CanInitializationError
from can.interfaces.pcan import PcanBus, PcanError
from can.interfaces.pcan.basic import *


class TestPCANBus(unittest.TestCase):
    def setUp(self) -> None:
        patcher = mock.patch("can.interfaces.pcan.pcan.PCANBasic", spec=True)
        self.MockPCANBasic = patcher.start()
        self.addCleanup(patcher.stop)
        self.mock_pcan = self.MockPCANBasic.return_value
        self.mock_pcan.Initialize.return_value = PCAN_ERROR_OK
        self.mock_pcan.InitializeFD = Mock(return_value=PCAN_ERROR_OK)
        self.mock_pcan.SetValue = Mock(return_value=PCAN_ERROR_OK)
        self.mock_pcan.GetValue = self._mockGetValue
        self.PCAN_API_VERSION_SIM = "4.2"
        self.bus = None

    def tearDown(self) -> None:
        if self.bus:
            self.bus.shutdown()
            self.bus = None

    def _mockGetValue(self, channel, parameter):
        """
        This method is used as mock for GetValue method of PCANBasic object.
        Only a subset of parameters are supported.
        """
        if parameter == PCAN_API_VERSION:
            return PCAN_ERROR_OK, self.PCAN_API_VERSION_SIM.encode("ascii")
        elif parameter == PCAN_RECEIVE_EVENT:
            return PCAN_ERROR_OK, int.from_bytes(PCAN_RECEIVE_EVENT, "big")
        raise NotImplementedError(
            f"No mock return value specified for parameter {parameter}"
        )

    def test_bus_creation(self) -> None:
        self.bus = can.Bus(interface="pcan")

        self.assertIsInstance(self.bus, PcanBus)
        self.assertEqual(self.bus.protocol, CanProtocol.CAN_20)

        with pytest.deprecated_call():
            self.assertFalse(self.bus.fd)

        self.MockPCANBasic.assert_called_once()
        self.mock_pcan.Initialize.assert_called_once()
        self.mock_pcan.InitializeFD.assert_not_called()

    def test_bus_creation_state_error(self) -> None:
        with self.assertRaises(ValueError):
            can.Bus(interface="pcan", state=BusState.STOPPED)

    @parameterized.expand([("f_clock", 80_000_000), ("f_clock_mhz", 80)])
    def test_bus_creation_fd(self, clock_param: str, clock_val: int) -> None:
        self.bus = can.Bus(
            interface="pcan",
            fd=True,
            nom_brp=1,
            nom_tseg1=129,
            nom_tseg2=30,
            nom_sjw=1,
            data_brp=1,
            data_tseg1=9,
            data_tseg2=6,
            data_sjw=1,
            channel="PCAN_USBBUS1",
            **{clock_param: clock_val},
        )

        self.assertIsInstance(self.bus, PcanBus)
        self.assertEqual(self.bus.protocol, CanProtocol.CAN_FD)

        with pytest.deprecated_call():
            self.assertTrue(self.bus.fd)

        self.MockPCANBasic.assert_called_once()
        self.mock_pcan.Initialize.assert_not_called()
        self.mock_pcan.InitializeFD.assert_called_once()

        # Retrieve second argument of first call
        bitrate_arg = self.mock_pcan.InitializeFD.call_args[0][-1]

        self.assertTrue(f"{clock_param}={clock_val}".encode("ascii") in bitrate_arg)
        self.assertTrue(b"nom_brp=1" in bitrate_arg)
        self.assertTrue(b"nom_tseg1=129" in bitrate_arg)
        self.assertTrue(b"nom_tseg2=30" in bitrate_arg)
        self.assertTrue(b"nom_sjw=1" in bitrate_arg)
        self.assertTrue(b"data_brp=1" in bitrate_arg)
        self.assertTrue(b"data_tseg1=9" in bitrate_arg)
        self.assertTrue(b"data_tseg2=6" in bitrate_arg)
        self.assertTrue(b"data_sjw=1" in bitrate_arg)

    def test_api_version_low(self) -> None:
        self.PCAN_API_VERSION_SIM = "1.0"
        with self.assertLogs("can.pcan", level="WARNING") as cm:
            self.bus = can.Bus(interface="pcan")
            found_version_warning = False
            for i in cm.output:
                if "version" in i and "pcan" in i:
                    found_version_warning = True
            self.assertTrue(
                found_version_warning,
                f"No warning was logged for incompatible api version {cm.output}",
            )

    def test_api_version_read_fail(self) -> None:
        self.mock_pcan.GetValue = Mock(return_value=(PCAN_ERROR_ILLOPERATION, None))
        with self.assertRaises(CanInitializationError):
            self.bus = can.Bus(interface="pcan")

    def test_issue1642(self) -> None:
        self.PCAN_API_VERSION_SIM = "1, 3, 0, 50"
        with self.assertLogs("can.pcan", level="WARNING") as cm:
            self.bus = can.Bus(interface="pcan")
            found_version_warning = False
            for i in cm.output:
                if "version" in i and "pcan" in i:
                    found_version_warning = True
            self.assertTrue(
                found_version_warning,
                f"No warning was logged for incompatible api version {cm.output}",
            )

    @parameterized.expand(
        [
            ("no_error", PCAN_ERROR_OK, PCAN_ERROR_OK, "some ok text 1"),
            ("one_error", PCAN_ERROR_UNKNOWN, PCAN_ERROR_OK, "some ok text 2"),
            (
                "both_errors",
                PCAN_ERROR_UNKNOWN,
                PCAN_ERROR_UNKNOWN,
                "An error occurred. Error-code's text (8h) couldn't be retrieved",
            ),
        ]
    )
    def test_get_formatted_error(self, name, status1, status2, expected_result: str):
        with self.subTest(name):
            self.bus = can.Bus(interface="pcan")
            self.mock_pcan.GetErrorText = Mock(
                side_effect=[
                    (status1, expected_result.encode("utf-8", errors="replace")),
                    (status2, expected_result.encode("utf-8", errors="replace")),
                ]
            )

            complete_text = self.bus._get_formatted_error(PCAN_ERROR_BUSHEAVY)

            self.assertEqual(complete_text, expected_result)

    def test_status(self) -> None:
        self.bus = can.Bus(interface="pcan")
        self.bus.status()
        self.mock_pcan.GetStatus.assert_called_once_with(PCAN_USBBUS1)

    @parameterized.expand(
        [("no_error", PCAN_ERROR_OK, True), ("error", PCAN_ERROR_UNKNOWN, False)]
    )
    def test_status_is_ok(self, name, status, expected_result) -> None:
        with self.subTest(name):
            self.mock_pcan.GetStatus = Mock(return_value=status)
            self.bus = can.Bus(interface="pcan")
            self.assertEqual(self.bus.status_is_ok(), expected_result)
            self.mock_pcan.GetStatus.assert_called_once_with(PCAN_USBBUS1)

    @parameterized.expand(
        [("no_error", PCAN_ERROR_OK, True), ("error", PCAN_ERROR_UNKNOWN, False)]
    )
    def test_reset(self, name, status, expected_result) -> None:
        with self.subTest(name):
            self.mock_pcan.Reset = Mock(return_value=status)
            self.bus = can.Bus(interface="pcan", fd=True)
            self.assertEqual(self.bus.reset(), expected_result)
            self.mock_pcan.Reset.assert_called_once_with(PCAN_USBBUS1)

    @parameterized.expand(
        [("no_error", PCAN_ERROR_OK, 1), ("error", PCAN_ERROR_UNKNOWN, None)]
    )
    def test_get_device_number(self, name, status, expected_result) -> None:
        with self.subTest(name):
            self.bus = can.Bus(interface="pcan", fd=True)
            # Mock GetValue after creation of bus to use first mock of
            # GetValue in constructor
            self.mock_pcan.GetValue = Mock(return_value=(status, 1))

            self.assertEqual(self.bus.get_device_number(), expected_result)
            self.mock_pcan.GetValue.assert_called_once_with(
                PCAN_USBBUS1, PCAN_DEVICE_NUMBER
            )

    @parameterized.expand(
        [("no_error", PCAN_ERROR_OK, True), ("error", PCAN_ERROR_UNKNOWN, False)]
    )
    def test_set_device_number(self, name, status, expected_result) -> None:
        with self.subTest(name):
            self.bus = can.Bus(interface="pcan")
            self.mock_pcan.SetValue = Mock(return_value=status)
            self.assertEqual(self.bus.set_device_number(3), expected_result)
            # check last SetValue call
            self.assertEqual(
                self.mock_pcan.SetValue.call_args_list[-1][0],
                (PCAN_USBBUS1, PCAN_DEVICE_NUMBER, 3),
            )

    def test_recv(self):
        data = (ctypes.c_ubyte * 8)(*[x for x in range(8)])
        msg = TPCANMsg(ID=0xC0FFEF, LEN=8, MSGTYPE=PCAN_MESSAGE_EXTENDED, DATA=data)

        timestamp = TPCANTimestamp()
        self.mock_pcan.Read = Mock(return_value=(PCAN_ERROR_OK, msg, timestamp))
        self.bus = can.Bus(interface="pcan")

        recv_msg = self.bus.recv()
        self.assertEqual(recv_msg.arbitration_id, msg.ID)
        self.assertEqual(recv_msg.dlc, msg.LEN)
        self.assertEqual(recv_msg.is_extended_id, True)
        self.assertEqual(recv_msg.is_fd, False)
        self.assertSequenceEqual(recv_msg.data, msg.DATA)
        self.assertEqual(recv_msg.timestamp, 0)

    def test_recv_fd(self):
        data = (ctypes.c_ubyte * 64)(*[x for x in range(64)])
        msg = TPCANMsgFD(
            ID=0xC0FFEF,
            DLC=64,
            MSGTYPE=(PCAN_MESSAGE_EXTENDED.value | PCAN_MESSAGE_FD.value),
            DATA=data,
        )

        timestamp = TPCANTimestampFD()

        self.mock_pcan.ReadFD = Mock(return_value=(PCAN_ERROR_OK, msg, timestamp))

        self.bus = can.Bus(interface="pcan", fd=True)

        recv_msg = self.bus.recv()
        self.assertEqual(recv_msg.arbitration_id, msg.ID)
        self.assertEqual(recv_msg.dlc, msg.DLC)
        self.assertEqual(recv_msg.is_extended_id, True)
        self.assertEqual(recv_msg.is_fd, True)
        self.assertSequenceEqual(recv_msg.data, msg.DATA)
        self.assertEqual(recv_msg.timestamp, 0)

    @pytest.mark.timeout(3.0)
    @patch("select.select", return_value=([], [], []))
    def test_recv_no_message(self, mock_select):
        self.mock_pcan.Read = Mock(return_value=(PCAN_ERROR_QRCVEMPTY, None, None))
        self.bus = can.Bus(interface="pcan")
        self.assertEqual(self.bus.recv(timeout=0.5), None)

    def test_send(self) -> None:
        self.mock_pcan.Write = Mock(return_value=PCAN_ERROR_OK)
        self.bus = can.Bus(interface="pcan")
        msg = can.Message(
            arbitration_id=0xC0FFEF, data=[1, 2, 3, 4, 5, 6, 7, 8], is_extended_id=True
        )
        self.bus.send(msg)
        self.mock_pcan.Write.assert_called_once()
        self.mock_pcan.WriteFD.assert_not_called()

    def test_send_fd(self) -> None:
        self.mock_pcan.WriteFD = Mock(return_value=PCAN_ERROR_OK)
        self.bus = can.Bus(interface="pcan", fd=True)
        msg = can.Message(
            arbitration_id=0xC0FFEF, data=[1, 2, 3, 4, 5, 6, 7, 8], is_extended_id=True
        )
        self.bus.send(msg)
        self.mock_pcan.Write.assert_not_called()
        self.mock_pcan.WriteFD.assert_called_once()

    @parameterized.expand(
        [
            (
                "standart",
                (False, False, False, False, False, False),
                PCAN_MESSAGE_STANDARD,
            ),
            (
                "extended",
                (True, False, False, False, False, False),
                PCAN_MESSAGE_EXTENDED,
            ),
            ("remote", (False, True, False, False, False, False), PCAN_MESSAGE_RTR),
            ("error", (False, False, True, False, False, False), PCAN_MESSAGE_ERRFRAME),
            ("fd", (False, False, False, True, False, False), PCAN_MESSAGE_FD),
            (
                "bitrate_switch",
                (False, False, False, False, True, False),
                PCAN_MESSAGE_BRS,
            ),
            (
                "error_state_indicator",
                (False, False, False, False, False, True),
                PCAN_MESSAGE_ESI,
            ),
        ]
    )
    def test_send_type(self, name, msg_type, expected_value) -> None:
        with self.subTest(name):
            (
                is_extended_id,
                is_remote_frame,
                is_error_frame,
                is_fd,
                bitrate_switch,
                error_state_indicator,
            ) = msg_type

            self.mock_pcan.Write = Mock(return_value=PCAN_ERROR_OK)

            self.bus = can.Bus(interface="pcan")
            msg = can.Message(
                arbitration_id=0xC0FFEF,
                data=[1, 2, 3, 4, 5, 6, 7, 8],
                is_extended_id=is_extended_id,
                is_remote_frame=is_remote_frame,
                is_error_frame=is_error_frame,
                bitrate_switch=bitrate_switch,
                error_state_indicator=error_state_indicator,
                is_fd=is_fd,
            )
            self.bus.send(msg)
            # self.mock_m_objPCANBasic.Write.assert_called_once()
            CANMsg = self.mock_pcan.Write.call_args_list[0][0][1]
            self.assertEqual(CANMsg.MSGTYPE, expected_value.value)

    def test_send_error(self) -> None:
        self.mock_pcan.Write = Mock(return_value=PCAN_ERROR_BUSHEAVY)
        self.bus = can.Bus(interface="pcan")
        msg = can.Message(
            arbitration_id=0xC0FFEF, data=[1, 2, 3, 4, 5, 6, 7, 8], is_extended_id=True
        )

        with self.assertRaises(PcanError):
            self.bus.send(msg)

    @parameterized.expand([("on", True), ("off", False)])
    def test_flash(self, name, flash) -> None:
        with self.subTest(name):
            self.bus = can.Bus(interface="pcan")
            self.bus.flash(flash)
            call_list = self.mock_pcan.SetValue.call_args_list
            last_call_args_list = call_list[-1][0]
            self.assertEqual(
                last_call_args_list, (PCAN_USBBUS1, PCAN_CHANNEL_IDENTIFYING, flash)
            )

    def test_shutdown(self) -> None:
        self.bus = can.Bus(interface="pcan")
        self.bus.shutdown()
        self.mock_pcan.Uninitialize.assert_called_once_with(PCAN_USBBUS1)

    @parameterized.expand(
        [
            ("active", BusState.ERROR_ACTIVE, PCAN_PARAMETER_OFF),
            ("passive", BusState.ERROR_PASSIVE, PCAN_PARAMETER_ON),
        ]
    )
    def test_state(self, name, bus_state: BusState, expected_parameter) -> None:
        with self.subTest(name):
            self.bus = can.Bus(interface="pcan")

            self.bus.state = bus_state
            call_list = self.mock_pcan.SetValue.call_args_list
            last_call_args_list = call_list[-1][0]
            self.assertEqual(
                last_call_args_list,
                (PCAN_USBBUS1, PCAN_LISTEN_ONLY, expected_parameter),
            )

    def test_state_constructor(self):
        for state in [BusState.ERROR_ACTIVE, BusState.ERROR_PASSIVE]:
            bus = can.Bus(interface="pcan", state=state)
            assert bus.state == state

    def test_detect_available_configs(self) -> None:
        if platform.system() == "Darwin":
            self.mock_pcan.GetValue = Mock(
                return_value=(PCAN_ERROR_OK, PCAN_CHANNEL_AVAILABLE)
            )
            configs = PcanBus._detect_available_configs()
            self.assertEqual(len(configs), 50)
        else:
            value = (TPCANChannelInformation * 1).from_buffer_copy(
                struct.pack("HBBI33sII", 81, 5, 0, 1, b"PCAN-USB FD", 1122867, 1)
            )
            self.mock_pcan.GetValue = Mock(return_value=(PCAN_ERROR_OK, value))
            configs = PcanBus._detect_available_configs()
            assert len(configs) == 1
            assert configs[0]["interface"] == "pcan"
            assert configs[0]["channel"] == "PCAN_USBBUS1"
            assert configs[0]["supports_fd"]
            assert configs[0]["controller_number"] == 0
            assert configs[0]["device_features"] == 1
            assert configs[0]["device_id"] == 1122867
            assert configs[0]["device_name"] == "PCAN-USB FD"
            assert configs[0]["device_type"] == 5
            assert configs[0]["channel_condition"] == 1

    @parameterized.expand([("valid", PCAN_ERROR_OK, "OK"), ("invalid", 0x00005, None)])
    def test_status_string(self, name, status, expected_result) -> None:
        with self.subTest(name):
            self.bus = can.Bus(interface="pcan")
            self.mock_pcan.GetStatus = Mock(return_value=status)
            self.assertEqual(self.bus.status_string(), expected_result)
            self.mock_pcan.GetStatus.assert_called()

    @parameterized.expand([(0x0, "error"), (0x42, "PCAN_USBBUS8")])
    def test_constructor_with_device_id(self, dev_id, expected_result):
        def get_value_side_effect(handle, param):
            if param == PCAN_API_VERSION:
                return PCAN_ERROR_OK, self.PCAN_API_VERSION_SIM.encode("ascii")

            if handle in (PCAN_USBBUS8, PCAN_USBBUS14):
                return 0, 0x42
            else:
                return PCAN_ERROR_ILLHW, 0x0

        self.mock_pcan.GetValue = Mock(side_effect=get_value_side_effect)

        if expected_result == "error":
            with self.assertRaises(ValueError):
                can.Bus(interface="pcan", device_id=dev_id)
        else:
            self.bus = can.Bus(interface="pcan", device_id=dev_id)
            self.assertEqual(expected_result, self.bus.channel_info)

    def test_bus_creation_auto_reset(self):
        self.bus = can.Bus(interface="pcan", auto_reset=True)
        self.assertIsInstance(self.bus, PcanBus)
        self.MockPCANBasic.assert_called_once()

    def test_auto_reset_init_fault(self):
        self.mock_pcan.SetValue = Mock(return_value=PCAN_ERROR_INITIALIZE)
        with self.assertRaises(CanInitializationError):
            self.bus = can.Bus(interface="pcan", auto_reset=True)

    def test_peak_fd_bus_constructor_regression(self):
        # Tests that the following issue has been fixed:
        # https://github.com/hardbyte/python-can/issues/1458
        params = {
            "interface": "pcan",
            "fd": True,
            "f_clock": 80000000,
            "nom_brp": 1,
            "nom_tseg1": 129,
            "nom_tseg2": 30,
            "nom_sjw": 1,
            "data_brp": 1,
            "data_tseg1": 9,
            "data_tseg2": 6,
            "data_sjw": 1,
            "channel": "PCAN_USBBUS1",
        }

        can.Bus(**params)

    def test_constructor_bit_timing(self):
        timing = can.BitTiming.from_registers(f_clock=8_000_000, btr0=0x47, btr1=0x2F)
        bus = can.Bus(interface="pcan", channel="PCAN_USBBUS1", timing=timing)

        bitrate_arg = self.mock_pcan.Initialize.call_args[0][1]
        self.assertEqual(bitrate_arg.value, 0x472F)
        self.assertEqual(bus.protocol, CanProtocol.CAN_20)

    def test_constructor_bit_timing_fd(self):
        timing = can.BitTimingFd(
            f_clock=40_000_000,
            nom_brp=1,
            nom_tseg1=129,
            nom_tseg2=30,
            nom_sjw=1,
            data_brp=1,
            data_tseg1=9,
            data_tseg2=6,
            data_sjw=1,
        )
        bus = can.Bus(interface="pcan", channel="PCAN_USBBUS1", timing=timing)
        self.assertEqual(bus.protocol, CanProtocol.CAN_FD)

        bitrate_arg = self.mock_pcan.InitializeFD.call_args[0][-1]

        self.assertTrue(b"f_clock=40000000" in bitrate_arg)
        self.assertTrue(b"nom_brp=1" in bitrate_arg)
        self.assertTrue(b"nom_tseg1=129" in bitrate_arg)
        self.assertTrue(b"nom_tseg2=30" in bitrate_arg)
        self.assertTrue(b"nom_sjw=1" in bitrate_arg)
        self.assertTrue(b"data_brp=1" in bitrate_arg)
        self.assertTrue(b"data_tseg1=9" in bitrate_arg)
        self.assertTrue(b"data_tseg2=6" in bitrate_arg)
        self.assertTrue(b"data_sjw=1" in bitrate_arg)


if __name__ == "__main__":
    unittest.main()