logformats_test.py

#!/usr/bin/env python

"""
This test module test the separate reader/writer combinations of the can.io.*
modules by writing some messages to a temporary file and reading it again.
Then it checks if the messages that were read are same ones as the
ones that were written. It also checks that the order of the messages
is correct. The types of messages that are tested differs between the
different writer/reader pairs - e.g., some don't handle error frames and
comments.

TODO: correctly set preserves_channel and adds_default_channel
"""

import locale
import logging
import os
import tempfile
import unittest
from abc import ABCMeta, abstractmethod
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from itertools import zip_longest
from pathlib import Path
from unittest.mock import patch

from parameterized import parameterized

import can.io
from can.io import asc, blf

from .data.example_data import (
    TEST_COMMENTS,
    TEST_MESSAGES_BASE,
    TEST_MESSAGES_CAN_FD,
    TEST_MESSAGES_ERROR_FRAMES,
    TEST_MESSAGES_REMOTE_FRAMES,
    sort_messages,
)
from .message_helper import ComparingMessagesTestCase

logging.basicConfig(level=logging.DEBUG)

try:
    import asammdf
except ModuleNotFoundError:
    asammdf = None


@contextmanager
def override_locale(category: int, locale_str: str) -> None:
    prev_locale = locale.getlocale(category)
    locale.setlocale(category, locale_str)
    yield
    locale.setlocale(category, prev_locale)


class ReaderWriterExtensionTest(unittest.TestCase):
    def _get_suffix_case_variants(self, suffix):
        return [
            suffix.upper(),
            suffix.lower(),
            f"can.msg.ext{suffix}",
            "".join([c.upper() if i % 2 else c for i, c in enumerate(suffix)]),
        ]

    def _test_extension(self, suffix):
        WriterType = can.io.MESSAGE_WRITERS.get(suffix)
        ReaderType = can.io.MESSAGE_READERS.get(suffix)
        for suffix_variant in self._get_suffix_case_variants(suffix):
            tmp_file = tempfile.NamedTemporaryFile(suffix=suffix_variant, delete=False)
            tmp_file.close()
            try:
                if WriterType:
                    with can.Logger(tmp_file.name) as logger:
                        assert type(logger) == WriterType
                if ReaderType:
                    with can.LogReader(tmp_file.name) as player:
                        assert type(player) == ReaderType
            finally:
                os.remove(tmp_file.name)

    def test_extension_matching_asc(self):
        self._test_extension(".asc")

    def test_extension_matching_blf(self):
        self._test_extension(".blf")

    def test_extension_matching_csv(self):
        self._test_extension(".csv")

    def test_extension_matching_db(self):
        self._test_extension(".db")

    def test_extension_matching_log(self):
        self._test_extension(".log")

    def test_extension_matching_txt(self):
        self._test_extension(".txt")

    def test_extension_matching_mf4(self):
        try:
            self._test_extension(".mf4")
        except NotImplementedError:
            if asammdf is not None:
                raise


class ReaderWriterTest(unittest.TestCase, ComparingMessagesTestCase, metaclass=ABCMeta):
    """Tests a pair of writer and reader by writing all data first and
    then reading all data and checking if they could be reconstructed
    correctly. Optionally writes some comments as well.

    .. note::
        This class is prevented from being executed as a test
        case itself by a *del* statement in at the end of the file.
        (Source: `*Wojciech B.* on StackOverlfow <https://stackoverflow.com/a/22836015/3753684>`_)
    """

    def __init__(self, *args, **kwargs):
        unittest.TestCase.__init__(self, *args, **kwargs)
        self._setup_instance()

    @abstractmethod
    def _setup_instance(self):
        """Hook for subclasses."""
        raise NotImplementedError()

    def _setup_instance_helper(
        self,
        writer_constructor,
        reader_constructor,
        binary_file=False,
        check_remote_frames=True,
        check_error_frames=True,
        check_fd=True,
        check_comments=False,
        test_append=False,
        allowed_timestamp_delta=0.0,
        preserves_channel=True,
        adds_default_channel=None,
        assert_file_closed=True,
    ):
        """
        :param Callable writer_constructor: the constructor of the writer class
        :param Callable reader_constructor: the constructor of the reader class
        :param bool binary_file: if True, opens files in binary and not in text mode

        :param bool check_remote_frames: if True, also tests remote frames
        :param bool check_error_frames: if True, also tests error frames
        :param bool check_fd: if True, also tests CAN FD frames
        :param bool check_comments: if True, also inserts comments at some
                                    locations and checks if they are contained anywhere literally
                                    in the resulting file. The locations as selected randomly
                                    but deterministically, which makes the test reproducible.
        :param bool test_append: tests the writer in append mode as well

        :param float or int or None allowed_timestamp_delta: directly passed to :meth:`can.Message.equals`
        :param bool preserves_channel: if True, checks that the channel attribute is preserved
        :param any adds_default_channel: sets this as the channel when not other channel was given
                                         ignored, if *preserves_channel* is True
        """
        # get all test messages
        self.original_messages = list(TEST_MESSAGES_BASE)
        if check_remote_frames:
            self.original_messages += TEST_MESSAGES_REMOTE_FRAMES
        if check_error_frames:
            self.original_messages += TEST_MESSAGES_ERROR_FRAMES
        if check_fd:
            self.original_messages += TEST_MESSAGES_CAN_FD

        # sort them so that for example ASCWriter does not "fix" any messages with timestamp 0.0
        self.original_messages = sort_messages(self.original_messages)

        if check_comments:
            # we check this because of the lack of a common base class
            # we filter for not starts with '__' so we do not get all the builtin
            # methods when logging to the console
            attrs = [
                attr for attr in dir(writer_constructor) if not attr.startswith("__")
            ]
            assert (
                "log_event" in attrs
            ), f"cannot check comments with this writer: {writer_constructor}"

        # get all test comments
        self.original_comments = TEST_COMMENTS if check_comments else ()

        self.writer_constructor = writer_constructor
        self.reader_constructor = reader_constructor
        self.binary_file = binary_file
        self.test_append_enabled = test_append
        self.assert_file_closed = assert_file_closed

        ComparingMessagesTestCase.__init__(
            self,
            allowed_timestamp_delta=allowed_timestamp_delta,
            preserves_channel=preserves_channel,
        )
        # adds_default_channel=adds_default_channel # TODO inlcude in tests

    def setUp(self):
        with tempfile.NamedTemporaryFile("w+", delete=False) as test_file:
            self.test_file_name = test_file.name

    def tearDown(self):
        os.remove(self.test_file_name)
        del self.test_file_name

    def test_path_like_explicit_stop(self):
        """testing with path-like and explicit stop() call"""

        # create writer
        print("writing all messages/comments")
        writer = self.writer_constructor(self.test_file_name)
        self._write_all(writer)
        self._ensure_fsync(writer)
        writer.stop()
        if self.assert_file_closed:
            self.assertTrue(writer.file.closed)

        print("reading all messages")
        reader = self.reader_constructor(self.test_file_name)
        read_messages = list(reader)
        # redundant, but this checks if stop() can be called multiple times
        reader.stop()
        if self.assert_file_closed:
            self.assertTrue(writer.file.closed)

        # check if at least the number of messages matches
        # could use assertCountEqual in later versions of Python and in the other methods
        self.assertEqual(
            len(read_messages),
            len(self.original_messages),
            "the number of written messages does not match the number of read messages",
        )

        self.assertMessagesEqual(self.original_messages, read_messages)
        self.assertIncludesComments(self.test_file_name)

    def test_path_like_context_manager(self):
        """testing with path-like object and context manager"""

        # create writer
        print("writing all messages/comments")
        with self.writer_constructor(self.test_file_name) as writer:
            self._write_all(writer)
            self._ensure_fsync(writer)
            w = writer
        if self.assert_file_closed:
            self.assertTrue(w.file.closed)

        # read all written messages
        print("reading all messages")
        with self.reader_constructor(self.test_file_name) as reader:
            read_messages = list(reader)
            r = reader
        if self.assert_file_closed:
            self.assertTrue(r.file.closed)

        # check if at least the number of messages matches;
        self.assertEqual(
            len(read_messages),
            len(self.original_messages),
            "the number of written messages does not match the number of read messages",
        )

        self.assertMessagesEqual(self.original_messages, read_messages)
        self.assertIncludesComments(self.test_file_name)

    def test_file_like_explicit_stop(self):
        """testing with file-like object and explicit stop() call"""

        # create writer
        print("writing all messages/comments")
        my_file = open(self.test_file_name, "wb" if self.binary_file else "w")
        writer = self.writer_constructor(my_file)
        self._write_all(writer)
        self._ensure_fsync(writer)
        writer.stop()
        if self.assert_file_closed:
            self.assertTrue(my_file.closed)

        print("reading all messages")
        my_file = open(self.test_file_name, "rb" if self.binary_file else "r")
        reader = self.reader_constructor(my_file)
        read_messages = list(reader)
        # redundant, but this checks if stop() can be called multiple times
        reader.stop()
        if self.assert_file_closed:
            self.assertTrue(my_file.closed)

        # check if at least the number of messages matches
        # could use assertCountEqual in later versions of Python and in the other methods
        self.assertEqual(
            len(read_messages),
            len(self.original_messages),
            "the number of written messages does not match the number of read messages",
        )

        self.assertMessagesEqual(self.original_messages, read_messages)
        self.assertIncludesComments(self.test_file_name)

    def test_file_like_context_manager(self):
        """testing with file-like object and context manager"""

        # create writer
        print("writing all messages/comments")
        my_file = open(self.test_file_name, "wb" if self.binary_file else "w")
        with self.writer_constructor(my_file) as writer:
            self._write_all(writer)
            self._ensure_fsync(writer)
            w = writer
        if self.assert_file_closed:
            self.assertTrue(my_file.closed)

        # read all written messages
        print("reading all messages")
        my_file = open(self.test_file_name, "rb" if self.binary_file else "r")
        with self.reader_constructor(my_file) as reader:
            read_messages = list(reader)
            r = reader
        if self.assert_file_closed:
            self.assertTrue(my_file.closed)

        # check if at least the number of messages matches;
        self.assertEqual(
            len(read_messages),
            len(self.original_messages),
            "the number of written messages does not match the number of read messages",
        )

        self.assertMessagesEqual(self.original_messages, read_messages)
        self.assertIncludesComments(self.test_file_name)

    def test_append_mode(self):
        """
        testing append mode with context manager and path-like object
        """
        if not self.test_append_enabled:
            raise unittest.SkipTest("do not test append mode")

        count = len(self.original_messages)
        first_part = self.original_messages[: count // 2]
        second_part = self.original_messages[count // 2 :]

        # write first half
        with self.writer_constructor(self.test_file_name) as writer:
            for message in first_part:
                writer(message)
            self._ensure_fsync(writer)

        # use append mode for second half
        try:
            writer = self.writer_constructor(self.test_file_name, append=True)
        except ValueError as e:
            # maybe "append" is not a formal parameter (this is the case for SqliteWriter)
            try:
                writer = self.writer_constructor(self.test_file_name)
            except TypeError:
                # if it is still a problem, raise the initial error
                raise e
        with writer:
            for message in second_part:
                writer(message)
            self._ensure_fsync(writer)
        with self.reader_constructor(self.test_file_name) as reader:
            read_messages = list(reader)

        self.assertMessagesEqual(self.original_messages, read_messages)

    def _write_all(self, writer):
        """Writes messages and insert comments here and there."""
        # Note: we make no assumptions about the length of original_messages and original_comments
        for msg, comment in zip_longest(
            self.original_messages, self.original_comments, fillvalue=None
        ):
            # msg and comment might be None
            if comment is not None:
                print("writing comment: ", comment)
                writer.log_event(comment)  # we already know that this method exists
            if msg is not None:
                print("writing message: ", msg)
                writer(msg)

    def _ensure_fsync(self, io_handler):
        if hasattr(io_handler, "file") and hasattr(io_handler.file, "fileno"):
            io_handler.file.flush()
            os.fsync(io_handler.file.fileno())

    def assertIncludesComments(self, filename):
        """
        Ensures that all comments are literally contained in the given file.

        :param filename: the path-like object to use
        """
        if self.original_comments:
            # read the entire outout file
            with open(filename, "rb" if self.binary_file else "r") as file:
                output_contents = file.read()
            # check each, if they can be found in there literally
            for comment in self.original_comments:
                self.assertIn(comment, output_contents)


class TestAscFileFormat(ReaderWriterTest):
    """Tests can.ASCWriter and can.ASCReader"""

    FORMAT_START_OF_FILE_DATE = "%a %b %d %I:%M:%S.%f %p %Y"

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.ASCWriter,
            can.ASCReader,
            check_fd=True,
            check_comments=True,
            preserves_channel=False,
            adds_default_channel=0,
        )

    def _get_logfile_location(self, filename: str) -> Path:
        my_dir = Path(__file__).parent
        return my_dir / "data" / filename

    def _read_log_file(self, filename, **kwargs):
        logfile = self._get_logfile_location(filename)
        with can.ASCReader(logfile, **kwargs) as reader:
            return list(reader)

    def test_read_absolute_time(self):
        time_from_file = "Sat Sep 30 10:06:13.191 PM 2017"
        start_time = (
            datetime.strptime(time_from_file, self.FORMAT_START_OF_FILE_DATE)
            .replace(tzinfo=asc._LOCAL_TZ)
            .timestamp()
        )

        expected_messages = [
            can.Message(
                timestamp=2.5010 + start_time,
                arbitration_id=0xC8,
                is_extended_id=False,
                is_rx=False,
                channel=1,
                dlc=8,
                data=[9, 8, 7, 6, 5, 4, 3, 2],
            ),
            can.Message(
                timestamp=17.876708 + start_time,
                arbitration_id=0x6F9,
                is_extended_id=False,
                channel=0,
                dlc=0x8,
                data=[5, 0xC, 0, 0, 0, 0, 0, 0],
            ),
        ]
        actual = self._read_log_file("test_CanMessage.asc", relative_timestamp=False)
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_can_message(self):
        expected_messages = [
            can.Message(
                timestamp=2.5010,
                arbitration_id=0xC8,
                is_extended_id=False,
                is_rx=False,
                channel=1,
                dlc=8,
                data=[9, 8, 7, 6, 5, 4, 3, 2],
            ),
            can.Message(
                timestamp=17.876708,
                arbitration_id=0x6F9,
                is_extended_id=False,
                channel=0,
                dlc=0x8,
                data=[5, 0xC, 0, 0, 0, 0, 0, 0],
            ),
        ]
        actual = self._read_log_file("test_CanMessage.asc")
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_can_remote_message(self):
        expected_messages = [
            can.Message(
                timestamp=2.510001,
                arbitration_id=0x100,
                is_extended_id=False,
                channel=1,
                is_remote_frame=True,
            ),
            can.Message(
                timestamp=2.520002,
                arbitration_id=0x200,
                is_extended_id=False,
                is_rx=False,
                channel=2,
                is_remote_frame=True,
            ),
            can.Message(
                timestamp=2.584921,
                arbitration_id=0x300,
                is_extended_id=False,
                channel=3,
                dlc=8,
                is_remote_frame=True,
            ),
        ]
        actual = self._read_log_file("test_CanRemoteMessage.asc")
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_can_fd_remote_message(self):
        expected_messages = [
            can.Message(
                timestamp=30.300981,
                arbitration_id=0x50005,
                channel=2,
                dlc=5,
                is_rx=False,
                is_fd=True,
                is_remote_frame=True,
                error_state_indicator=True,
            )
        ]
        actual = self._read_log_file("test_CanFdRemoteMessage.asc")
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_can_fd_message(self):
        expected_messages = [
            can.Message(
                timestamp=30.005021,
                arbitration_id=0x300,
                is_extended_id=False,
                channel=0,
                dlc=8,
                data=[0x11, 0xC2, 3, 4, 5, 6, 7, 8],
                is_fd=True,
                bitrate_switch=True,
            ),
            can.Message(
                timestamp=30.005041,
                arbitration_id=0x1C4D80A7,
                channel=1,
                dlc=8,
                is_rx=False,
                data=[0x12, 0xC2, 3, 4, 5, 6, 7, 8],
                is_fd=True,
                error_state_indicator=True,
            ),
            can.Message(
                timestamp=30.005071,
                arbitration_id=0x30A,
                is_extended_id=False,
                channel=2,
                dlc=8,
                data=[1, 2, 3, 4, 5, 6, 7, 8],
                is_fd=True,
                bitrate_switch=True,
                error_state_indicator=True,
            ),
        ]
        actual = self._read_log_file("test_CanFdMessage.asc")
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_can_fd_message_64(self):
        expected_messages = [
            can.Message(
                timestamp=30.506898,
                arbitration_id=0x4EE,
                is_extended_id=False,
                channel=3,
                dlc=64,
                data=[0xA1, 2, 3, 4] + 59 * [0] + [0x64],
                is_fd=True,
                error_state_indicator=True,
            ),
            can.Message(
                timestamp=31.506898,
                arbitration_id=0x1C4D80A7,
                channel=3,
                dlc=64,
                data=[0xB1, 2, 3, 4] + 59 * [0] + [0x64],
                is_fd=True,
                bitrate_switch=True,
            ),
        ]
        actual = self._read_log_file("test_CanFdMessage64.asc")
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_can_and_canfd_error_frames(self):
        expected_messages = [
            can.Message(timestamp=2.501000, channel=0, is_error_frame=True),
            can.Message(timestamp=3.501000, channel=0, is_error_frame=True),
            can.Message(timestamp=4.501000, channel=1, is_error_frame=True),
            can.Message(
                timestamp=30.806898,
                channel=4,
                is_rx=False,
                is_error_frame=True,
                is_fd=True,
            ),
        ]
        actual = self._read_log_file("test_CanErrorFrames.asc")
        self.assertMessagesEqual(actual, expected_messages)

    def test_read_ignore_comments(self):
        _msg_list = self._read_log_file("logfile.asc")

    def test_read_no_triggerblock(self):
        _msg_list = self._read_log_file("issue_1256.asc")

    def test_read_can_dlc_greater_than_8(self):
        _msg_list = self._read_log_file("issue_1299.asc")

    def test_read_error_frame_channel(self):
        # gh-issue 1578
        err_frame = can.Message(is_error_frame=True, channel=4)

        temp_file = tempfile.NamedTemporaryFile("w", delete=False)
        temp_file.close()

        try:
            with can.ASCWriter(temp_file.name) as writer:
                writer.on_message_received(err_frame)

            with can.ASCReader(temp_file.name) as reader:
                msg_list = list(reader)
                assert len(msg_list) == 1
                assert err_frame.equals(
                    msg_list[0], check_channel=True
                ), f"{err_frame!r}!={msg_list[0]!r}"
        finally:
            os.unlink(temp_file.name)

    def test_write_millisecond_handling(self):
        tz = asc._LOCAL_TZ
        now = datetime(
            year=2017,
            month=9,
            day=30,
            hour=15,
            minute=6,
            second=13,
            microsecond=191456,
            tzinfo=tz,
        )

        with patch("can.io.asc.datetime") as mock_datetime:
            mock_datetime.now.return_value = now
            writer = can.ASCWriter(self.test_file_name, tz=tz)

        msg = can.Message(timestamp=now.timestamp(), arbitration_id=0x123, data=b"h")
        writer.on_message_received(msg)
        writer.stop()

        actual_file = Path(self.test_file_name)
        expected_file = self._get_logfile_location("single_frame_us_locale.asc")

        self.assertEqual(expected_file.read_text(), actual_file.read_text())

    def test_write(self):
        tz = asc._LOCAL_TZ
        with patch("can.io.asc.datetime") as mock_datetime:
            now = datetime(
                year=2017,
                month=9,
                day=30,
                hour=15,
                minute=6,
                second=13,
                microsecond=191456,
                tzinfo=tz,
            )
            mock_datetime.now.return_value = now
            writer = can.ASCWriter(self.test_file_name, tz=tz)

        msg = can.Message(
            timestamp=now.timestamp(),
            arbitration_id=0x123,
            data=range(64),
        )

        with writer:
            writer.on_message_received(msg)

        actual_file = Path(self.test_file_name)
        expected_file = self._get_logfile_location("single_frame.asc")

        self.assertEqual(expected_file.read_text(), actual_file.read_text())

    def test_write_timestamps_format_default_is_absolute(self):
        """ASCWriter should write 'timestamps absolute' in the header by default."""
        with can.ASCWriter(self.test_file_name) as writer:
            pass

        content = Path(self.test_file_name).read_text()
        self.assertIn("timestamps absolute", content)

    def test_write_timestamps_format_relative(self):
        """ASCWriter should write 'timestamps relative' when requested."""
        with can.ASCWriter(self.test_file_name, timestamps_format="relative") as writer:
            pass

        content = Path(self.test_file_name).read_text()
        self.assertIn("timestamps relative", content)
        self.assertNotIn("timestamps absolute", content)

    def test_write_timestamps_format_invalid(self):
        """ASCWriter should raise ValueError for an unsupported timestamps_format."""
        with self.assertRaises(ValueError):
            can.ASCWriter(self.test_file_name, timestamps_format="unix")

    def test_write_relative_timestamp_roundtrip(self):
        """Messages written with relative format round-trip to their original timestamps."""
        msgs = [
            can.Message(timestamp=100.0, arbitration_id=0x1, data=b"\x01"),
            can.Message(timestamp=100.3, arbitration_id=0x2, data=b"\x02"),
            can.Message(timestamp=101.0, arbitration_id=0x3, data=b"\x03"),
        ]

        with can.ASCWriter(self.test_file_name, timestamps_format="relative") as writer:
            for m in msgs:
                writer.on_message_received(m)

        with can.ASCReader(self.test_file_name, relative_timestamp=False) as reader:
            result = list(reader)

        self.assertEqual(len(result), len(msgs))
        self.assertAlmostEqual(result[0].timestamp, 100.0, places=3)
        self.assertAlmostEqual(result[1].timestamp, 100.3, places=3)
        self.assertAlmostEqual(result[2].timestamp, 101.0, places=3)

    def test_write_relative_timestamps_are_per_event_deltas(self):
        """With timestamps_format='relative', each written timestamp is a delta from the
        preceding event (not an offset from measurement start)."""
        msgs = [
            can.Message(timestamp=100.0, arbitration_id=0x1, data=b"\x01"),
            can.Message(timestamp=100.3, arbitration_id=0x2, data=b"\x02"),
            can.Message(timestamp=101.0, arbitration_id=0x3, data=b"\x03"),
        ]

        with can.ASCWriter(self.test_file_name, timestamps_format="relative") as writer:
            for m in msgs:
                writer.on_message_received(m)

        with can.ASCReader(self.test_file_name, relative_timestamp=True) as reader:
            result = list(reader)

        self.assertEqual(len(result), len(msgs))
        # msg1: 0.0  (delta from "Start of measurement" at same time)
        # msg2: 0.3  (delta from msg1)
        # msg3: 0.7  (delta from msg2 — NOT 1.0, which would be absolute offset)
        self.assertAlmostEqual(result[0].timestamp, 0.0, places=5)
        self.assertAlmostEqual(result[1].timestamp, 0.3, places=5)
        self.assertAlmostEqual(result[2].timestamp, 0.7, places=5)

    def test_write_absolute_timestamps_are_offsets_from_start(self):
        """With timestamps_format='absolute' (default), messages round-trip to their
        original timestamps when read back with relative_timestamp=False."""
        msgs = [
            can.Message(timestamp=100.0, arbitration_id=0x1, data=b"\x01"),
            can.Message(timestamp=100.3, arbitration_id=0x2, data=b"\x02"),
            can.Message(timestamp=101.0, arbitration_id=0x3, data=b"\x03"),
        ]

        with can.ASCWriter(self.test_file_name, timestamps_format="absolute") as writer:
            for m in msgs:
                writer.on_message_received(m)

        with can.ASCReader(self.test_file_name, relative_timestamp=False) as reader:
            result = list(reader)

        self.assertEqual(len(result), len(msgs))
        # Timestamps are recovered from the triggerblock start time + file offset:
        self.assertAlmostEqual(result[0].timestamp, 100.0, places=3)
        self.assertAlmostEqual(result[1].timestamp, 100.3, places=3)
        self.assertAlmostEqual(result[2].timestamp, 101.0, places=3)

    @parameterized.expand(
        [
            (
                "May 27 04:09:35.000 pm 2014",
                datetime(
                    2014, 5, 27, 16, 9, 35, 0, tzinfo=timezone(timedelta(hours=5))
                ).timestamp(),
            ),
            (
                "Mai 27 04:09:35.000 pm 2014",
                datetime(
                    2014, 5, 27, 16, 9, 35, 0, tzinfo=timezone(timedelta(hours=5))
                ).timestamp(),
            ),
            (
                "Apr 28 10:44:52.480 2022",
                datetime(
                    2022, 4, 28, 10, 44, 52, 480000, tzinfo=timezone(timedelta(hours=5))
                ).timestamp(),
            ),
            (
                "Sep 30 15:06:13.191 2017",
                datetime(
                    2017, 9, 30, 15, 6, 13, 191000, tzinfo=timezone(timedelta(hours=5))
                ).timestamp(),
            ),
            (
                "Sep 30 15:06:13.191 pm 2017",
                datetime(
                    2017, 9, 30, 15, 6, 13, 191000, tzinfo=timezone(timedelta(hours=5))
                ).timestamp(),
            ),
            (
                "Sep 30 15:06:13.191 am 2017",
                datetime(
                    2017, 9, 30, 15, 6, 13, 191000, tzinfo=timezone(timedelta(hours=5))
                ).timestamp(),
            ),
        ]
    )
    def test_datetime_to_timestamp(
        self, datetime_string: str, expected_timestamp: float
    ):
        timestamp = can.ASCReader._datetime_to_timestamp(
            datetime_string, tz=timezone(timedelta(hours=5))
        )
        self.assertAlmostEqual(timestamp, expected_timestamp)


class TestBlfFileFormat(ReaderWriterTest):
    """Tests can.BLFWriter and can.BLFReader.

    Uses log files created by Toby Lorenz:
    https://bitbucket.org/tobylorenz/vector_blf/src/master/src/Vector/BLF/tests/unittests/events_from_binlog/
    """

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.BLFWriter,
            can.BLFReader,
            binary_file=True,
            check_fd=True,
            check_comments=False,
            test_append=True,
            preserves_channel=False,
            adds_default_channel=0,
        )

    def _read_log_file(self, filename):
        logfile = os.path.join(os.path.dirname(__file__), "data", filename)
        with can.BLFReader(logfile) as reader:
            return list(reader)

    def test_can_message(self):
        expected = can.Message(
            timestamp=2459565876.494607,
            arbitration_id=0x4444444,
            is_extended_id=False,
            channel=0x1110,
            dlc=0x33,
            data=[0x55, 0x66, 0x77, 0x88, 0x99, 0xAA, 0xBB, 0xCC],
        )
        actual = self._read_log_file("test_CanMessage.blf")
        self.assertMessagesEqual(actual, [expected] * 2)
        self.assertEqual(actual[0].channel, expected.channel)

    def test_can_message_2(self):
        expected = can.Message(
            timestamp=2459565876.494607,
            arbitration_id=0x4444444,
            is_extended_id=False,
            channel=0x1110,
            dlc=0x33,
            data=[0x55, 0x66, 0x77, 0x88, 0x99, 0xAA, 0xBB, 0xCC],
        )
        actual = self._read_log_file("test_CanMessage2.blf")
        self.assertMessagesEqual(actual, [expected] * 2)
        self.assertEqual(actual[0].channel, expected.channel)

    def test_can_fd_message(self):
        expected = can.Message(
            timestamp=2459565876.494607,
            arbitration_id=0x4444444,
            is_extended_id=False,
            channel=0x1110,
            dlc=64,
            is_fd=True,
            bitrate_switch=True,
            error_state_indicator=True,
            data=range(64),
        )
        actual = self._read_log_file("test_CanFdMessage.blf")
        self.assertMessagesEqual(actual, [expected] * 2)
        self.assertEqual(actual[0].channel, expected.channel)

    def test_can_fd_message_64(self):
        expected = can.Message(
            timestamp=2459565876.494607,
            arbitration_id=0x15555555,
            is_extended_id=False,
            is_remote_frame=True,
            channel=0x10,
            dlc=64,
            is_fd=True,
            is_rx=False,
            bitrate_switch=True,
            error_state_indicator=True,
        )
        actual = self._read_log_file("test_CanFdMessage64.blf")
        self.assertMessagesEqual(actual, [expected] * 2)
        self.assertEqual(actual[0].channel, expected.channel)

    def test_can_error_frame_ext(self):
        expected = can.Message(
            timestamp=2459565876.494607,
            is_error_frame=True,
            arbitration_id=0x19999999,
            is_extended_id=True,
            channel=0x1110,
            dlc=0x66,
            data=[0xCC, 0xDD, 0xEE, 0xFF, 0x11, 0x22, 0x33, 0x44],
        )
        actual = self._read_log_file("test_CanErrorFrameExt.blf")
        self.assertMessagesEqual(actual, [expected] * 2)
        self.assertEqual(actual[0].channel, expected.channel)

    def test_timestamp_to_systemtime(self):
        self.assertAlmostEqual(
            1636485425.999,
            blf.systemtime_to_timestamp(blf.timestamp_to_systemtime(1636485425.998908)),
            places=3,
        )
        self.assertAlmostEqual(
            1636485426.0,
            blf.systemtime_to_timestamp(blf.timestamp_to_systemtime(1636485425.999908)),
            places=3,
        )

    def test_issue_1905(self):
        expected = can.Message(
            timestamp=1735654183.491113,
            channel=6,
            arbitration_id=0x6A9,
            is_extended_id=False,
            is_fd=True,
            bitrate_switch=True,
            error_state_indicator=False,
            dlc=64,
            data=bytearray(
                b"\xff\xff\xff\xff\xff\xff\xff\xff"
                b"\xff\xff\xff\xff\xff\xff\xff\xff"
                b"\xff\xff\xff\xff\xff\xff\xff\xff"
                b"\xff\xff\xff\xff\xff\xff\xff\xff"
                b"\xff\xff\xff\xff\xff\xff\xff\xff"
                b"\xff\xff\xff\xff\xff\xff\xff\xff"
                b"\x00\x00\x00\x00\x00\x00\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00"
            ),
        )
        msgs = self._read_log_file("issue_1905.blf")
        self.assertMessageEqual(expected, msgs[0])


class TestCanutilsFileFormat(ReaderWriterTest):
    """Tests can.CanutilsLogWriter and can.CanutilsLogReader"""

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.CanutilsLogWriter,
            can.CanutilsLogReader,
            check_fd=True,
            test_append=True,
            check_comments=False,
            preserves_channel=False,
            adds_default_channel="vcan0",
        )


class TestCsvFileFormat(ReaderWriterTest):
    """Tests can.CSVWriter and can.CSVReader"""

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.CSVWriter,
            can.CSVReader,
            check_fd=False,
            test_append=True,
            check_comments=False,
            preserves_channel=False,
            adds_default_channel=None,
        )


@unittest.skipIf(asammdf is None, "MF4 is unavailable")
class TestMF4FileFormat(ReaderWriterTest):
    """Tests can.MF4Writer and can.MF4Reader"""

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.MF4Writer,
            can.MF4Reader,
            binary_file=True,
            check_comments=False,
            preserves_channel=False,
            allowed_timestamp_delta=1e-4,
            adds_default_channel=0,
        )


class TestSqliteDatabaseFormat(ReaderWriterTest):
    """Tests can.SqliteWriter and can.SqliteReader"""

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.SqliteWriter,
            can.SqliteReader,
            check_fd=False,
            test_append=True,
            check_comments=False,
            preserves_channel=False,
            adds_default_channel=None,
            assert_file_closed=False,
        )

    @unittest.skip("not implemented")
    def test_file_like_explicit_stop(self):
        pass

    @unittest.skip("not implemented")
    def test_file_like_context_manager(self):
        pass

    def test_read_all(self):
        """
        testing :meth:`can.SqliteReader.read_all` with context manager and path-like object
        """
        # create writer
        print("writing all messages/comments")
        with self.writer_constructor(self.test_file_name) as writer:
            self._write_all(writer)

        # read all written messages
        print("reading all messages")
        with self.reader_constructor(self.test_file_name) as reader:
            read_messages = list(reader.read_all())

        # check if at least the number of messages matches;
        self.assertEqual(
            len(read_messages),
            len(self.original_messages),
            "the number of written messages does not match the number of read messages",
        )

        self.assertMessagesEqual(self.original_messages, read_messages)


class TestPrinter(unittest.TestCase):
    """Tests that can.Printer does not crash.

    TODO test append mode
    """

    messages = (
        TEST_MESSAGES_BASE
        + TEST_MESSAGES_REMOTE_FRAMES
        + TEST_MESSAGES_ERROR_FRAMES
        + TEST_MESSAGES_CAN_FD
    )

    def test_not_crashes_with_stdout(self):
        with can.Printer() as printer:
            for message in self.messages:
                printer(message)

    def test_not_crashes_with_file(self):
        with tempfile.NamedTemporaryFile("w") as temp_file:
            with can.Printer(temp_file) as printer:
                for message in self.messages:
                    printer(message)


class TestTrcFileFormatBase(ReaderWriterTest):
    """
    Base class for Tests with can.TRCWriter and can.TRCReader

    .. note::
        This class is prevented from being executed as a test
        case itself by a *del* statement in at the end of the file.
    """

    def _setup_instance(self):
        super()._setup_instance_helper(
            can.TRCWriter,
            can.TRCReader,
            check_remote_frames=False,
            check_error_frames=False,
            check_fd=False,
            check_comments=False,
            preserves_channel=False,
            allowed_timestamp_delta=0.001,
            adds_default_channel=0,
        )

    def _read_log_file(self, filename, **kwargs):
        logfile = os.path.join(os.path.dirname(__file__), "data", filename)
        with can.TRCReader(logfile, **kwargs) as reader:
            return list(reader)


class TestTrcFileFormatGen(TestTrcFileFormatBase):
    """Generic tests for can.TRCWriter and can.TRCReader with different file versions"""

    def test_can_message(self):
        start_time = 1506809173.191  # 30.09.2017 22:06:13.191.000 as timestamp
        expected_messages = [
            can.Message(
                timestamp=start_time + 2.5010,
                arbitration_id=0xC8,
                is_extended_id=False,
                is_rx=False,
                channel=1,
                dlc=8,
                data=[9, 8, 7, 6, 5, 4, 3, 2],
            ),
            can.Message(
                timestamp=start_time + 17.876708,
                arbitration_id=0x6F9,
                is_extended_id=False,
                channel=0,
                dlc=0x8,
                data=[5, 0xC, 0, 0, 0, 0, 0, 0],
            ),
        ]
        actual = self._read_log_file("test_CanMessage.trc")
        self.assertMessagesEqual(actual, expected_messages)

    @parameterized.expand(
        [
            ("V1_0", "test_CanMessage_V1_0_BUS1.trc", False),
            ("V1_1", "test_CanMessage_V1_1.trc", True),
            ("V1_3", "test_CanMessage_V1_3.trc", True),
            ("V2_0", "test_CanMessage_V2_0_BUS1.trc", True),
            ("V2_1", "test_CanMessage_V2_1.trc", True),
        ]
    )
    def test_can_message_versions(self, name, filename, is_rx_support):
        with self.subTest(name):
            if name == "V1_0":
                # Version 1.0 does not support start time
                start_time = 0
            else:
                start_time = (
                    1639837687.062001  # 18.12.2021 14:28:07.062.001 as timestamp
                )

            def msg_std(timestamp):
                msg = can.Message(
                    timestamp=timestamp + start_time,
                    arbitration_id=0x000,
                    is_extended_id=False,
                    channel=1,
                    dlc=8,
                    data=[0, 0, 0, 0, 0, 0, 0, 0],
                )
                if is_rx_support:
                    msg.is_rx = False
                return msg

            def msg_ext(timestamp):
                msg = can.Message(
                    timestamp=timestamp + start_time,
                    arbitration_id=0x100,
                    is_extended_id=True,
                    channel=1,
                    dlc=8,
                    data=[0, 0, 0, 0, 0, 0, 0, 0],
                )
                if is_rx_support:
                    msg.is_rx = False
                return msg

            def msg_rtr(timestamp):
                msg = can.Message(
                    timestamp=timestamp + start_time,
                    arbitration_id=0x704,
                    is_extended_id=False,
                    is_remote_frame=True,
                    channel=1,
                    dlc=1,
                    data=[],
                )
                if is_rx_support:
                    msg.is_rx = True
                return msg

            expected_messages = [
                msg_ext(17.5354),
                msg_ext(17.7003),
                msg_ext(17.8738),
                msg_std(19.2954),
                msg_std(19.5006),
                msg_std(19.7052),
                msg_ext(20.5927),
                msg_ext(20.7986),
                msg_ext(20.9560),
                msg_ext(21.0971),
                msg_rtr(48.9376),
            ]
            actual = self._read_log_file(filename)
            self.assertMessagesEqual(actual, expected_messages)

    def test_not_supported_version(self):
        with tempfile.NamedTemporaryFile(mode="w") as f:
            with self.assertRaises(NotImplementedError):
                writer = can.TRCWriter(f)
                writer.file_version = can.TRCFileVersion.UNKNOWN
                writer.on_message_received(can.Message())


class TestTrcFileFormatV1_0(TestTrcFileFormatBase):
    """Tests can.TRCWriter and can.TRCReader with file version 1.0"""

    @staticmethod
    def Writer(filename):
        writer = can.TRCWriter(filename)
        writer.file_version = can.TRCFileVersion.V1_0
        return writer

    def _setup_instance(self):
        super()._setup_instance()
        self.writer_constructor = TestTrcFileFormatV1_0.Writer


# this excludes the base class from being executed as a test case itself
del ReaderWriterTest
del TestTrcFileFormatBase


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