Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: merge trc into one file can be disabled #458

Merged
merged 1 commit into from Mar 5, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/whats_new/version_ongoing.rst
@@ -1,2 +1,8 @@
Version ongoing
---------------

PCAN Connector
^^^^^^^^^^^^^^

Merging trc logs into one file can now be turned off by setting the
``merge_trc_logs`` parameter to ``False`` in the PCAN connector
5 changes: 4 additions & 1 deletion src/pykiso/lib/connectors/cc_pcan_can/cc_pcan_can.py
Expand Up @@ -83,6 +83,7 @@ def __init__(
can_filters: list = None,
logging_activated: bool = True,
bus_error_warning_filter: bool = False,
merge_trc_logs: bool = True,
**kwargs,
):
"""Initialize can channel settings.
Expand Down Expand Up @@ -123,6 +124,7 @@ def __init__(
:param logging_activated: boolean used to disable logfile creation
:param bus_error_warning_filter: if True filter the PCAN driver warnings
'Bus error: an error counter' from the logs.
:param merge_trc_logs: if True, merge all traces in one file at the end of the program
"""
super().__init__(**kwargs)
self.interface = interface
Expand Down Expand Up @@ -153,6 +155,7 @@ def __init__(
self.timeout = 1e-6
self.trc_count = 0
self._initialize_trace()
self.merge_trc_logs = merge_trc_logs

if bus_error_warning_filter:
logging.getLogger("can.pcan").addFilter(PcanFilter())
Expand Down Expand Up @@ -493,5 +496,5 @@ def _remove_offset(messages: List[TypedMessage], offset: float) -> List[TypedMes

def shutdown(self) -> None:
"""Destructor method."""
if self.logging_activated:
if self.logging_activated and self.merge_trc_logs:
self._merge_trc()
24 changes: 13 additions & 11 deletions tests/test_cc_pcan_can.py
Expand Up @@ -22,18 +22,9 @@

from pykiso import Message
from pykiso.lib.connectors.cc_pcan_can import cc_pcan_can
from pykiso.lib.connectors.cc_pcan_can.cc_pcan_can import (
CCPCanCan,
PCANBasic,
can,
)
from pykiso.lib.connectors.cc_pcan_can.cc_pcan_can import CCPCanCan, PCANBasic, can
from pykiso.lib.connectors.cc_pcan_can.trc_handler import TRCReaderCanFD
from pykiso.message import (
MessageAckType,
MessageCommandType,
MessageType,
TlvKnownTags,
)
from pykiso.message import MessageAckType, MessageCommandType, MessageType, TlvKnownTags

tlv_dict_to_send = {
TlvKnownTags.TEST_REPORT: "OK",
Expand Down Expand Up @@ -890,7 +881,18 @@ def test_read_trace_messages_with_old_file_version(mocker, trc_files_v1_1, caplo
"Trace merging is not available for trc file version TRCFileVersion.V1_1"
in caplog.text
)
def test_disable_auto_merge(mocker, mock_can_bus, mock_PCANBasic):
mock_merge = mocker.patch.object(CCPCanCan, "_merge_trc")

cc_pcan = CCPCanCan(logging_activated=True, merge_trc_logs=True)
cc_pcan.shutdown()
mock_merge.assert_called_once()

mock_merge = mocker.patch.object(CCPCanCan, "_merge_trc")

cc_pcan = CCPCanCan(logging_activated=True, merge_trc_logs=False)
cc_pcan.shutdown()
mock_merge.assert_not_called()

def test_remove_offset():
class Msg:
Expand Down
19 changes: 8 additions & 11 deletions tests/test_cc_virtual_can.py
@@ -1,15 +1,18 @@
import pytest
import importlib
import logging
import sys
import importlib
from unittest import mock
from pykiso import Message

import can as python_can
from pykiso.lib.connectors import cc_virtual_can
from pykiso.lib.connectors.cc_virtual_can import CCVirtualCan
import pytest
from can.interfaces.udp_multicast.bus import UdpMulticastBus

from pykiso import Message
from pykiso.lib.connectors import cc_virtual_can
from pykiso.lib.connectors.cc_pcan_can.cc_pcan_can import can
from pykiso.lib.connectors.cc_virtual_can import CCVirtualCan

pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="virtual can not supported on windows")


@pytest.fixture
Expand Down Expand Up @@ -42,7 +45,6 @@ def test_import():


def test_constructor(mock_vcan_bus):

vcan_inst = CCVirtualCan(
channel=UdpMulticastBus.DEFAULT_GROUP_IPv4,
receive_own_messages=False,
Expand All @@ -69,7 +71,6 @@ def test_cc_close(
caplog,
mock_vcan_bus,
):

with caplog.at_level(logging.ERROR, logger="pykiso.lib.connectors.cc_pcan_can.log"):
with CCVirtualCan() as vcan_inst:
pass
Expand All @@ -96,7 +97,6 @@ def test_cc_open(mock_vcan_bus, caplog):


def test_cc_send(mock_vcan_bus):

with CCVirtualCan() as vcan:
vcan.bus = mock_vcan_bus.Bus
vcan._cc_send(b"\x10\x36", 0x0A)
Expand All @@ -120,7 +120,6 @@ def test_can_recv(mock_vcan_bus):


def test_can_recv_invalid(mocker, mock_vcan_bus):

mocker.patch("can.interface.Bus.recv", return_value={"msg": None})

with CCVirtualCan() as vcan:
Expand All @@ -131,7 +130,6 @@ def test_can_recv_invalid(mocker, mock_vcan_bus):


def test_can_recv_exception(caplog, mock_vcan_bus, mocker):

mocker.patch("can.interface.Bus.recv", side_effect=Exception())
logging.getLogger("pykiso.lib.connectors.cc_virtual_can.log")
with CCVirtualCan() as can:
Expand All @@ -144,7 +142,6 @@ def test_can_recv_exception(caplog, mock_vcan_bus, mocker):


def test_can_recv_error(caplog, mock_vcan_bus, mocker):

mocker.patch("can.interface.Bus.recv", side_effect=python_can.CanError())
logging.getLogger("pykiso.lib.connectors.cc_virtual_can.log")
with CCVirtualCan() as can:
Expand Down