Passively record a channel

The record auxiliary can be used to utilize the logging mechanism from a connector. For example the realtime trace from the segger jlink can be recorded during a test run. The record auxiliary can also be used to save the log into a chosen file. It is also able to search for some specific message or regular expression (regex) into the current string or into a specified file/folder.

Usage Examples

To use the auxiliary in your test scripts the auxiliary must be properly defined in the config yaml. Example:

auxiliaries:
  record_aux:
    connectors:
      com: rtt_channel
    config:
      # When is_active is set, it actively polls the connector. It demands if
      # the used connector needs to be polled actively.
      is_active: False # False because rtt_channel has its own receive thread
    type: pykiso.lib.auxiliaries.record_auxiliary:RecordAuxiliary

connectors:
  rtt_channel:
    config:
      chip_name: "STM12345678"
      speed: 4000
      block_address: 0x12345678
      verbose: True
      tx_buffer_idx: 1
      rx_buffer_idx: 1
      # Path relative to this yaml where the RTT logs should be written to.
      # Creates a file named rtt.log
      rtt_log_path: ./
      # RTT channel from where the RTT logs should be read
      rtt_log_buffer_idx: 0
      # Manage RTT log CPU impact by setting logger speed. eg: 100% CPU load
      # default: 1000 lines/s
      rtt_log_speed: null
    type: pykiso.lib.connectors.cc_rtt_segger:CCRttSegger

test_suite_list:
- suite_dir: test_record
  test_filter_pattern: '*.py'
  test_suite_id: 1
auxiliaries:
  record_aux:
    connectors:
      com: example_channel
    config:
      com: CChannel
      is_active: True
      timeout: 0
      log_folder_path: "examples/test_record"
    type: pykiso.lib.auxiliaries.record_auxiliary:RecordAuxiliary

connectors:
  example_channel:
    config: null
    type: pykiso.lib.connectors.cc_raw_loopback:CCLoopback

test_suite_list:
- suite_dir: test_record
  test_filter_pattern: test_recorder_example.py
  test_suite_id: 1

Below find a example for the usage in a test script. It is only necessary to import record auxiliary.

from pykiso.auxiliaries import record_aux

Example test script:

##########################################################################
# Copyright (c) 2010-2022 Robert Bosch GmbH
# This program and the accompanying materials are made available under the
# terms of the Eclipse Public License 2.0 which is available at
# http://www.eclipse.org/legal/epl-2.0.
#
# SPDX-License-Identifier: EPL-2.0
##########################################################################

"""
Record auxiliary test
*********************

:module: test_record

:synopsis: Example test that shows how to record a connector

.. currentmodule:: test_record

"""

import logging
import time

import pykiso

# !!! IMPORTANT !!!
# To start recording the channel which are specified in the yaml file,
# the record_aux must be first imported here.
# The channel recording will then run automatically in the background.
from pykiso.auxiliaries import record_aux


@pykiso.define_test_parameters(suite_id=1, case_id=1, aux_list=[])
class TestWithPowerSupply(pykiso.BasicTest):
    def setUp(self):
        """Hook method from unittest in order to execute code before test case run."""
        logging.info(
            f"--------------- SETUP: {self.test_suite_id}, {self.test_case_id} --------------"
        )

    def test_run(self):
        logging.info(
            f"--------------- RUN: {self.test_suite_id}, {self.test_case_id} ----------------"
        )

        logging.info(
            "Sleep 5 Seconds. Record specified channel from .yaml in the background."
        )
        time.sleep(5)

    def tearDown(self):
        """Hook method from unittest in order to execute code after test case run."""
        logging.info(
            f"--------------- TEARDOWN: {self.test_suite_id}, {self.test_case_id} -----------"
        )
##########################################################################
# Copyright (c) 2010-2022 Robert Bosch GmbH
# This program and the accompanying materials are made available under the
# terms of the Eclipse Public License 2.0 which is available at
# http://www.eclipse.org/legal/epl-2.0.
#
# SPDX-License-Identifier: EPL-2.0
##########################################################################

"""
Record auxiliary test
*********************

:module: test_record

:synopsis: Example test that shows how to record a connector

.. currentmodule:: test_record

"""

import logging
import time

import pykiso
from pykiso.auxiliaries import record_aux

logging = logging.getLogger(__name__)


@pykiso.define_test_parameters(suite_id=1, case_id=1, aux_list=[])
class TestWithPowerSupply(pykiso.BasicTest):
    def generate_new_log(self, msg: bytes):
        return record_aux.channel._cc_send(msg)

    def setUp(self):
        """Hook method from unittest in order to execute code before test case run."""
        logging.info(
            f"--------------- SETUP: {self.test_suite_id}, {self.test_case_id} --------------"
        )

    def test_run(self):
        """"""
        logging.info(
            f"--------------- RUN: {self.test_suite_id}, {self.test_case_id} ----------------"
        )
        header = record_aux.new_log()
        self.assertEqual(header, "Received data :")

        self.generate_new_log(msg=b"log1")
        time.sleep(1)
        new_log = record_aux.new_log()
        self.assertEqual(new_log, "\nlog1")
        logging.info(new_log)

        self.generate_new_log(msg=b"log2")
        time.sleep(1)
        new_log = record_aux.new_log()
        self.assertEqual(new_log, "\nlog2")
        logging.info(new_log)

        logging.info(record_aux.get_data())

        # search regex
        logging.info(record_aux.search_regex_current_string(regex=r"log\d"))

        # clear data and check
        record_aux.clear_buffer()
        logging.info(record_aux.get_data())

        # create a file where it write recorded data. as log is empty, will not return any file
        record_aux.dump_to_file(filename="record_example.txt")

        record_aux.stop_recording()

        logging.info("Sleep 1 Seconds to do something else with the channel.")
        time.sleep(1)

        record_aux.start_recording()
        time.sleep(1)  # Time for the channel to get opened
        header = record_aux.new_log()
        self.assertEqual(header, "Received data :")

        self.generate_new_log(msg=b"log3")
        time.sleep(1)
        new_log = record_aux.new_log()
        self.assertEqual(new_log, "\nlog3")
        logging.info(new_log)

    def tearDown(self):
        """Hook method from unittest in order to execute code after test case run."""
        logging.info(
            f"--------------- TEARDOWN: {self.test_suite_id}, {self.test_case_id} -----------"
        )