zephyr/scripts/pylib/twister/twisterlib/harness.py

# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations

import json
import logging
import os
import platform
import re
import shlex
import shutil
import subprocess
import sys
import threading
import time
import xml.etree.ElementTree as ET
from collections import OrderedDict
from enum import Enum

from pytest import ExitCode
from twisterlib.constants import SUPPORTED_SIMS_IN_PYTEST
from twisterlib.environment import PYTEST_PLUGIN_INSTALLED, ZEPHYR_BASE
from twisterlib.error import ConfigurationError, StatusAttributeError
from twisterlib.handlers import Handler, terminate_process
from twisterlib.reports import ReportStatus
from twisterlib.statuses import TwisterStatus
from twisterlib.testinstance import TestInstance

logger = logging.getLogger('twister')
logger.setLevel(logging.DEBUG)

_WINDOWS = platform.system() == 'Windows'


class Harness:
    GCOV_START = "GCOV_COVERAGE_DUMP_START"
    GCOV_END = "GCOV_COVERAGE_DUMP_END"
    FAULT = "ZEPHYR FATAL ERROR"
    RUN_PASSED = "PROJECT EXECUTION SUCCESSFUL"
    RUN_FAILED = "PROJECT EXECUTION FAILED"
    run_id_pattern = r"RunID: (?P<run_id>.*)"

    def __init__(self):
        self._status = TwisterStatus.NONE
        self.reason = None
        self.type = None
        self.regex = []
        self.matches = OrderedDict()
        self.ordered = True
        self.id = None
        self.fail_on_fault = True
        self.fault = False
        self.capture_coverage = False
        self.next_pattern = 0
        self.record = None
        self.record_pattern = None
        self.record_as_json = None
        self.recording = []
        self.ztest = False
        self.detected_suite_names = []
        self.run_id = None
        self.started_suites = {}
        self.started_cases = {}
        self.matched_run_id = False
        self.run_id_exists = False
        self.instance: TestInstance | None = None
        self.testcase_output = ""
        self._match = False


    @property
    def trace(self) -> bool:
        return self.instance.handler.options.verbose > 2

    @property
    def status(self) -> TwisterStatus:
        return self._status

    @status.setter
    def status(self, value : TwisterStatus) -> None:
        # Check for illegal assignments by value
        try:
            key = value.name if isinstance(value, Enum) else value
            self._status = TwisterStatus[key]
        except KeyError as err:
            raise StatusAttributeError(self.__class__, value) from err

    def configure(self, instance):
        self.instance = instance
        config = instance.testsuite.harness_config
        self.id = instance.testsuite.id
        self.run_id = instance.run_id
        if instance.testsuite.ignore_faults:
            self.fail_on_fault = False

        if config:
            self.type = config.get('type', None)
            self.regex = config.get('regex', [])
            self.ordered = config.get('ordered', True)
            self.record = config.get('record', {})
            if self.record:
                self.record_pattern = re.compile(self.record.get("regex", ""))
                self.record_as_json = self.record.get("as_json")

    def build(self):
        pass

    def get_testcase_name(self):
        """
        Get current TestCase name.
        """
        return self.id

    def translate_record(self, record: dict) -> dict:
        if self.record_as_json:
            for k in self.record_as_json:
                if k not in record:
                    continue
                try:
                    record[k] = json.loads(record[k]) if record[k] else {}
                except json.JSONDecodeError as parse_error:
                    logger.warning(f"HARNESS:{self.__class__.__name__}: recording JSON failed:"
                                   f" {parse_error} for '{k}':'{record[k]}'")
                    # Don't set the Harness state to failed for recordings.
                    record[k] = { 'ERROR': { 'msg': str(parse_error), 'doc': record[k] } }
        return record

    def parse_record(self, line) -> re.Match:
        match = None
        if self.record_pattern:
            match = self.record_pattern.search(line)
            if match:
                rec = self.translate_record(
                    { k:v.strip() for k,v in match.groupdict(default="").items() }
                )
                self.recording.append(rec)
        return match
    #

    def process_test(self, line):

        self.parse_record(line)

        runid_match = re.search(self.run_id_pattern, line)
        if runid_match:
            run_id = runid_match.group("run_id")
            self.run_id_exists = True
            if run_id == str(self.run_id):
                self.matched_run_id = True

        if self.RUN_PASSED in line:
            if self.fault:
                self.status = TwisterStatus.FAIL
                self.reason = "Fault detected while running test"
            else:
                self.status = TwisterStatus.PASS

        if self.RUN_FAILED in line:
            self.status = TwisterStatus.FAIL
            self.reason = "Testsuite failed"

        if self.fail_on_fault and line == self.FAULT:
            self.fault = True

        if self.GCOV_START in line:
            self.capture_coverage = True
        elif self.GCOV_END in line:
            self.capture_coverage = False

class Robot(Harness):

    is_robot_test = True

    def configure(self, instance):
        super().configure(instance)
        self.instance = instance

        config = instance.testsuite.harness_config
        if config:
            self.path = config.get('robot_testsuite', None)
            self.option = config.get('robot_option', None)

    def handle(self, line):
        ''' Test cases that make use of this harness care about results given
            by Robot Framework which is called in run_robot_test(), so works of this
            handle is trying to give a PASS or FAIL to avoid timeout, nothing
            is writen into handler.log
        '''
        self.instance.status = TwisterStatus.PASS
        tc = self.instance.get_case_or_create(self.id)
        tc.status = TwisterStatus.PASS

    def run_robot_test(self, command, handler):
        start_time = time.time()
        env = os.environ.copy()

        if self.option:
            if isinstance(self.option, list):
                for option in self.option:
                    for v in str(option).split():
                        command.append(f'{v}')
            else:
                for v in str(self.option).split():
                    command.append(f'{v}')

        if self.path is None:
            raise PytestHarnessException('The parameter robot_testsuite is mandatory')

        if isinstance(self.path, list):
            for suite in self.path:
                command.append(os.path.join(handler.sourcedir, suite))
        else:
            command.append(os.path.join(handler.sourcedir, self.path))

        with subprocess.Popen(command, stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT, cwd=self.instance.build_dir, env=env) as renode_test_proc:
            out, _ = renode_test_proc.communicate()

            self.instance.execution_time = time.time() - start_time

            if renode_test_proc.returncode == 0:
                self.instance.status = TwisterStatus.PASS
                # all tests in one Robot file are treated as a single test case,
                # so its status should be set accordingly to the instance status
                # please note that there should be only one testcase in testcases list
                self.instance.testcases[0].status = TwisterStatus.PASS
            else:
                logger.error(
                    f"Robot test failure: {handler.sourcedir} for {self.instance.platform.name}"
                )
                self.instance.status = TwisterStatus.FAIL
                self.instance.testcases[0].status = TwisterStatus.FAIL

            if out:
                with open(os.path.join(self.instance.build_dir, handler.log), 'w') as log:
                    log_msg = out.decode(sys.getdefaultencoding())
                    log.write(log_msg)

class Console(Harness):

    def get_testcase_name(self):
        '''
        Get current TestCase name.

        Console Harness id has only TestSuite id without TestCase name suffix.
        Only the first TestCase name might be taken if available when a Ztest with
        a single test case is configured to use this harness type for simplified
        output parsing instead of the Ztest harness as Ztest suite should do.
        '''
        if self.instance and len(self.instance.testcases) == 1:
            return self.instance.testcases[0].name
        return super().get_testcase_name()

    def configure(self, instance):
        super().configure(instance)
        if self.regex is None or len(self.regex) == 0:
            self.status = TwisterStatus.FAIL
            tc = self.instance.set_case_status_by_name(
                self.get_testcase_name(),
                TwisterStatus.FAIL,
                f"HARNESS:{self.__class__.__name__}:no regex patterns configured."
            )
            raise ConfigurationError(self.instance.name, tc.reason)
        if self.type == "one_line":
            self.pattern = re.compile(self.regex[0])
            self.patterns_expected = 1
        elif self.type == "multi_line":
            self.patterns = []
            for r in self.regex:
                self.patterns.append(re.compile(r))
            self.patterns_expected = len(self.patterns)
        else:
            self.status = TwisterStatus.FAIL
            tc = self.instance.set_case_status_by_name(
                self.get_testcase_name(),
                TwisterStatus.FAIL,
                f"HARNESS:{self.__class__.__name__}:incorrect type={self.type}"
            )
            raise ConfigurationError(self.instance.name, tc.reason)
        #

    def handle(self, line):
        if self.type == "one_line":
            if self.pattern.search(line):
                logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED:"
                             f"'{self.pattern.pattern}'")
                self.next_pattern += 1
                self.status = TwisterStatus.PASS
        elif self.type == "multi_line" and self.ordered:
            if (self.next_pattern < len(self.patterns) and
                self.patterns[self.next_pattern].search(line)):
                logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED("
                             f"{self.next_pattern + 1}/{self.patterns_expected}):"
                             f"'{self.patterns[self.next_pattern].pattern}'")
                self.next_pattern += 1
                if self.next_pattern >= len(self.patterns):
                    self.status = TwisterStatus.PASS
        elif self.type == "multi_line" and not self.ordered:
            for i, pattern in enumerate(self.patterns):
                r = self.regex[i]
                if pattern.search(line) and r not in self.matches:
                    self.matches[r] = line
                    logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED("
                                 f"{len(self.matches)}/{self.patterns_expected}):"
                                 f"'{pattern.pattern}'")
            if len(self.matches) == len(self.regex):
                self.status = TwisterStatus.PASS
        else:
            logger.error("Unknown harness_config type")

        if self.fail_on_fault and self.FAULT in line:
            self.fault = True

        if self.GCOV_START in line:
            self.capture_coverage = True
        elif self.GCOV_END in line:
            self.capture_coverage = False

        self.process_test(line)
        # Reset the resulting test state to FAIL when not all of the patterns were
        # found in the output, but just ztest's 'PROJECT EXECUTION SUCCESSFUL'.
        # It might happen because of the pattern sequence diverged from the
        # test code, the test platform has console issues, or even some other
        # test image was executed.
        # TODO: Introduce explicit match policy type to reject
        # unexpected console output, allow missing patterns, deny duplicates.
        if self.status == TwisterStatus.PASS and \
           self.ordered and \
           self.next_pattern < self.patterns_expected:
            logger.error(f"HARNESS:{self.__class__.__name__}: failed with"
                         f" {self.next_pattern} of {self.patterns_expected}"
                         f" expected ordered patterns.")
            self.status = TwisterStatus.FAIL
            self.reason = "patterns did not match (ordered)"
        if self.status == TwisterStatus.PASS and \
           not self.ordered and \
           len(self.matches) < self.patterns_expected:
            logger.error(f"HARNESS:{self.__class__.__name__}: failed with"
                         f" {len(self.matches)} of {self.patterns_expected}"
                         f" expected unordered patterns.")
            self.status = TwisterStatus.FAIL
            self.reason = "patterns did not match (unordered)"

        tc = self.instance.get_case_or_create(self.get_testcase_name())
        if self.status == TwisterStatus.PASS:
            tc.status = TwisterStatus.PASS
        else:
            tc.status = TwisterStatus.FAIL


class PytestHarnessException(Exception):
    """General exception for pytest."""


class Pytest(Harness):

    def configure(self, instance: TestInstance):
        super().configure(instance)
        self.running_dir = instance.build_dir
        self.source_dir = instance.testsuite.source_dir
        self.report_file = os.path.join(self.running_dir, 'report.xml')
        self.pytest_log_file_path = os.path.join(self.running_dir, 'twister_harness.log')
        self.reserved_dut = None
        self._output = []

    def pytest_run(self, timeout):
        try:
            cmd = self.generate_command()
            self.run_command(cmd, timeout)
        except PytestHarnessException as pytest_exception:
            logger.error(str(pytest_exception))
            self.status = TwisterStatus.FAIL
            self.instance.reason = str(pytest_exception)
        finally:
            self.instance.record(self.recording)
            self._update_test_status()
            if self.reserved_dut:
                self.instance.handler.make_dut_available(self.reserved_dut)

    def generate_command(self):
        config = self.instance.testsuite.harness_config
        handler: Handler = self.instance.handler
        pytest_root = config.get('pytest_root', ['pytest']) if config else ['pytest']
        pytest_args_yaml = config.get('pytest_args', []) if config else []
        pytest_dut_scope = config.get('pytest_dut_scope', None) if config else None
        command = [
            'pytest',
            '--twister-harness',
            '-s', '-v',
            f'--build-dir={self.running_dir}',
            f'--junit-xml={self.report_file}',
            '--log-file-level=DEBUG',
            '--log-file-format=%(asctime)s.%(msecs)03d:%(levelname)s:%(name)s: %(message)s',
            f'--log-file={self.pytest_log_file_path}',
            f'--platform={self.instance.platform.name}'
        ]
        command.extend([os.path.normpath(os.path.join(
            self.source_dir, os.path.expanduser(os.path.expandvars(src)))) for src in pytest_root])

        if pytest_dut_scope:
            command.append(f'--dut-scope={pytest_dut_scope}')

        # Always pass output from the pytest test and the test image up to Twister log.
        command.extend([
            '--log-cli-level=DEBUG',
            '--log-cli-format=%(levelname)s: %(message)s'
        ])

        # Use the test timeout as the base timeout for pytest
        base_timeout = handler.get_test_timeout()
        command.append(f'--base-timeout={base_timeout}')

        if handler.type_str == 'device':
            command.extend(
                self._generate_parameters_for_hardware(handler)
            )
        elif handler.type_str in SUPPORTED_SIMS_IN_PYTEST:
            command.append(f'--device-type={handler.type_str}')
        elif handler.type_str == 'build':
            command.append('--device-type=custom')
        else:
            raise PytestHarnessException(
                f'Support for handler {handler.type_str} not implemented yet'
            )

        if handler.type_str != 'device':
            for fixture in handler.options.fixture:
                command.append(f'--twister-fixture={fixture}')

        if handler.options.extra_test_args and handler.type_str == 'native':
            command.append(f'--extra-test-args={shlex.join(handler.options.extra_test_args)}')

        command.extend(pytest_args_yaml)

        if handler.options.pytest_args:
            command.extend(handler.options.pytest_args)

        return command

    def _generate_parameters_for_hardware(self, handler: Handler):
        command = ['--device-type=hardware']
        hardware = handler.get_hardware()
        if not hardware:
            raise PytestHarnessException('Hardware is not available')
        # update the instance with the device id to have it in the summary report
        self.instance.dut = hardware.id

        self.reserved_dut = hardware
        if hardware.serial_pty:
            command.append(f'--device-serial-pty={hardware.serial_pty}')
        else:
            command.extend([
                f'--device-serial={hardware.serial}',
                f'--device-serial-baud={hardware.baud}'
            ])

        if hardware.flash_timeout:
            command.append(f'--flash-timeout={hardware.flash_timeout}')

        options = handler.options
        if runner := hardware.runner or options.west_runner:
            command.append(f'--runner={runner}')

        if hardware.runner_params:
            for param in hardware.runner_params:
                command.append(f'--runner-params={param}')

        if options.west_flash and options.west_flash != []:
            command.append(f'--west-flash-extra-args={options.west_flash}')

        if board_id := hardware.probe_id or hardware.id:
            command.append(f'--device-id={board_id}')

        if hardware.product:
            command.append(f'--device-product={hardware.product}')

        if hardware.pre_script:
            command.append(f'--pre-script={hardware.pre_script}')

        if hardware.post_flash_script:
            command.append(f'--post-flash-script={hardware.post_flash_script}')

        if hardware.post_script:
            command.append(f'--post-script={hardware.post_script}')

        if hardware.flash_before:
            command.append(f'--flash-before={hardware.flash_before}')

        for fixture in hardware.fixtures:
            command.append(f'--twister-fixture={fixture}')

        return command

    def run_command(self, cmd, timeout):
        cmd, env = self._update_command_with_env_dependencies(cmd)
        with subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            env=env
        ) as proc:
            try:
                reader_t = threading.Thread(target=self._output_reader, args=(proc,), daemon=True)
                reader_t.start()
                reader_t.join(timeout)
                if reader_t.is_alive():
                    terminate_process(proc)
                    logger.warning('Timeout has occurred. Can be extended in testspec file. '
                                   f'Currently set to {timeout} seconds.')
                    self.instance.reason = 'Pytest timeout'
                    self.status = TwisterStatus.FAIL
                proc.wait(timeout)
            except subprocess.TimeoutExpired:
                self.status = TwisterStatus.FAIL
                proc.kill()

        if proc.returncode in (ExitCode.INTERRUPTED, ExitCode.USAGE_ERROR, ExitCode.INTERNAL_ERROR):
            self.status = TwisterStatus.ERROR
            self.instance.reason = f'Pytest error - return code {proc.returncode}'
            with open(self.pytest_log_file_path, 'w') as log_file:
                log_file.write(shlex.join(cmd) + '\n\n')
                log_file.write('\n'.join(self._output))

    @staticmethod
    def _update_command_with_env_dependencies(cmd):
        '''
        If python plugin wasn't installed by pip, then try to indicate it to
        pytest by update PYTHONPATH and append -p argument to pytest command.
        '''
        env = os.environ.copy()
        if not PYTEST_PLUGIN_INSTALLED:
            cmd.extend(['-p', 'twister_harness.plugin'])
            pytest_plugin_path = os.path.join(
                ZEPHYR_BASE,
                'scripts',
                'pylib',
                'pytest-twister-harness',
                'src'
            )
            env['PYTHONPATH'] = pytest_plugin_path + os.pathsep + env.get('PYTHONPATH', '')
            if _WINDOWS:
                cmd_append_python_path = f'set PYTHONPATH={pytest_plugin_path};%PYTHONPATH% && '
            else:
                cmd_append_python_path = (
                    f'export PYTHONPATH={pytest_plugin_path}:${{PYTHONPATH}} && '
                )
        else:
            cmd_append_python_path = ''
        cmd_to_print = cmd_append_python_path + shlex.join(cmd)
        logger.debug(f'Running pytest command: {cmd_to_print}')

        return cmd, env

    def _output_reader(self, proc):
        self._output = []
        while proc.stdout.readable() and proc.poll() is None:
            line = proc.stdout.readline().decode().strip()
            if not line:
                continue
            self._output.append(line)
            logger.debug(f'PYTEST: {line}')
            self.parse_record(line)
        proc.communicate()

    def _update_test_status(self):
        if self.status == TwisterStatus.NONE:
            self.instance.testcases = []
            try:
                self._parse_report_file(self.report_file)
            except Exception as e:
                logger.error(f'Error when parsing file {self.report_file}: {e}')
                self.status = TwisterStatus.FAIL
            finally:
                if not self.instance.testcases:
                    self.instance.init_cases()

        self.instance.status = self.status if self.status != TwisterStatus.NONE else \
                               TwisterStatus.FAIL
        if self.instance.status in [TwisterStatus.ERROR, TwisterStatus.FAIL]:
            self.instance.reason = self.instance.reason or 'Pytest failed'
            self.instance.add_missing_case_status(TwisterStatus.BLOCK, self.instance.reason)

    def _parse_report_file(self, report):
        tree = ET.parse(report)
        root = tree.getroot()

        if (elem_ts := root.find('testsuite')) is not None:
            if elem_ts.get('failures') != '0':
                self.status = TwisterStatus.FAIL
                self.instance.reason = (
                    f"{elem_ts.get('failures')}/{elem_ts.get('tests')} pytest scenario(s) failed"
                )
            elif elem_ts.get('errors') != '0':
                self.status = TwisterStatus.ERROR
                self.instance.reason = 'Error during pytest execution'
            elif elem_ts.get('skipped') == elem_ts.get('tests'):
                self.status = TwisterStatus.SKIP
            else:
                self.status = TwisterStatus.PASS
            self.instance.execution_time = float(elem_ts.get('time'))

            for elem_tc in elem_ts.findall('testcase'):
                tc = self.instance.add_testcase(f"{self.id}.{elem_tc.get('name')}")
                tc.duration = float(elem_tc.get('time'))
                elem = elem_tc.find('*')
                if elem is None:
                    tc.status = TwisterStatus.PASS
                else:
                    if elem.tag == ReportStatus.SKIP:
                        tc.status = TwisterStatus.SKIP
                    elif elem.tag == ReportStatus.FAIL:
                        tc.status = TwisterStatus.FAIL
                    else:
                        tc.status = TwisterStatus.ERROR
                    tc.reason = elem.get('message')
                    tc.output = elem.text
        else:
            self.status = TwisterStatus.SKIP
            self.instance.reason = 'No tests collected'


class Gtest(Harness):
    ANSI_ESCAPE = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
    _NAME_PATTERN = "[a-zA-Z_][a-zA-Z0-9_]*"
    _SUITE_TEST_NAME_PATTERN = f"(?P<suite_name>{_NAME_PATTERN})\\.(?P<test_name>{_NAME_PATTERN})"
    TEST_START_PATTERN = f".*\\[ RUN      \\] {_SUITE_TEST_NAME_PATTERN}"
    TEST_PASS_PATTERN = f".*\\[       OK \\] {_SUITE_TEST_NAME_PATTERN}"
    TEST_SKIP_PATTERN = f".*\\[ DISABLED \\] {_SUITE_TEST_NAME_PATTERN}"
    TEST_FAIL_PATTERN = f".*\\[  FAILED  \\] {_SUITE_TEST_NAME_PATTERN}"
    FINISHED_PATTERN = (
        ".*(?:\\[==========\\] Done running all tests\\.|"
        + "\\[----------\\] Global test environment tear-down)"
    )

    def __init__(self):
        super().__init__()
        self.tc = None
        self.has_failures = False

    def handle(self, line):
        # Strip the ANSI characters, they mess up the patterns
        non_ansi_line = self.ANSI_ESCAPE.sub('', line)

        if self.status != TwisterStatus.NONE:
            return

        # Check if we started running a new test
        test_start_match = re.search(self.TEST_START_PATTERN, non_ansi_line)
        if test_start_match:
            # Add the suite name
            suite_name = test_start_match.group("suite_name")
            if suite_name not in self.detected_suite_names:
                self.detected_suite_names.append(suite_name)

            # Generate the internal name of the test
            name = "{}.{}.{}".format(self.id, suite_name, test_start_match.group("test_name"))

            # Assert that we don't already have a running test
            assert (
                self.tc is None
            ), f"gTest error, {self.tc} didn't finish"

            # Check that the instance doesn't exist yet (prevents re-running)
            tc = self.instance.get_case_by_name(name)
            assert tc is None, f"gTest error, {tc} running twice"

            # Create the test instance and set the context
            tc = self.instance.get_case_or_create(name)
            self.tc = tc
            self.tc.status = TwisterStatus.STARTED
            self.testcase_output += line + "\n"
            self._match = True

        # Check if the test run finished
        finished_match = re.search(self.FINISHED_PATTERN, non_ansi_line)
        if finished_match:
            tc = self.instance.get_case_or_create(self.id)
            if self.has_failures or self.tc is not None:
                self.status = TwisterStatus.FAIL
                tc.status = TwisterStatus.FAIL
            else:
                self.status = TwisterStatus.PASS
                tc.status = TwisterStatus.PASS
            return

        # Check if the individual test finished
        state, name = self._check_result(non_ansi_line)
        if state == TwisterStatus.NONE or name is None:
            # Nothing finished, keep processing lines
            return

        # Get the matching test and make sure it's the same as the current context
        tc = self.instance.get_case_by_name(name)
        assert (
            tc is not None and tc == self.tc
        ), f"gTest error, mismatched tests. Expected {self.tc} but got {tc}"

        # Test finished, clear the context
        self.tc = None

        # Update the status of the test
        tc.status = state
        if tc.status == TwisterStatus.FAIL:
            self.has_failures = True
            tc.output = self.testcase_output
        self.testcase_output = ""
        self._match = False

    def _check_result(self, line):
        test_pass_match = re.search(self.TEST_PASS_PATTERN, line)
        if test_pass_match:
            return TwisterStatus.PASS, \
                   "{}.{}.{}".format(
                        self.id, test_pass_match.group("suite_name"),
                        test_pass_match.group("test_name")
                    )
        test_skip_match = re.search(self.TEST_SKIP_PATTERN, line)
        if test_skip_match:
            return TwisterStatus.SKIP, \
                   "{}.{}.{}".format(
                       self.id, test_skip_match.group("suite_name"),
                       test_skip_match.group("test_name")
                    )
        test_fail_match = re.search(self.TEST_FAIL_PATTERN, line)
        if test_fail_match:
            return TwisterStatus.FAIL, \
                   "{}.{}.{}".format(
                       self.id, test_fail_match.group("suite_name"),
                       test_fail_match.group("test_name")
                    )
        return None, None


class Test(Harness):
    __test__ = False  # for pytest to skip this class when collects tests

    test_suite_start_pattern = re.compile(r"Running TESTSUITE (?P<suite_name>\S*)")
    test_suite_end_pattern = re.compile(
        r"TESTSUITE (?P<suite_name>\S*)\s+(?P<suite_status>succeeded|failed)"
    )
    test_case_start_pattern = re.compile(r"START - (test_)?([a-zA-Z0-9_-]+)")
    test_case_end_pattern = re.compile(
        r".*(PASS|FAIL|SKIP) - (test_)?(\S*) in (\d*[.,]?\d*) seconds"
    )
    test_suite_summary_pattern = re.compile(
        r"SUITE (?P<suite_status>\S*) - .* \[(?P<suite_name>\S*)\]:"
        r" .* duration = (\d*[.,]?\d*) seconds"
    )
    test_case_summary_pattern = re.compile(
        r" - (PASS|FAIL|SKIP) - \[([^\.]*).(test_)?(\S*)\] duration = (\d*[.,]?\d*) seconds"
    )


    def get_testcase(self, tc_name, phase, ts_name=None):
        """ Search a Ztest case among detected in the test image binary
            expecting the same test names as already known from the ELF.
            Track suites and cases unexpectedly found in the log.
        """
        ts_names = self.started_suites.keys()
        if ts_name:
            if ts_name not in self.instance.testsuite.ztest_suite_names:
                logger.warning(f"On {phase}: unexpected Ztest suite '{ts_name}' "
                               f"not present among: {self.instance.testsuite.ztest_suite_names}")
            if ts_name not in self.detected_suite_names:
                if self.trace:
                    logger.debug(f"On {phase}: detected new Ztest suite '{ts_name}'")
                self.detected_suite_names.append(ts_name)
            ts_names = [ ts_name ] if ts_name in ts_names else []

        # First, try to match the test case ID to the first running Ztest suite with this test name.
        for ts_name_ in ts_names:
            if self.started_suites[ts_name_]['count'] < (0 if phase == 'TS_SUM' else 1):
                continue
            tc_fq_id = self.instance.compose_case_name(f"{ts_name_}.{tc_name}")
            if tc := self.instance.get_case_by_name(tc_fq_id):
                if self.trace:
                    logger.debug(f"On {phase}: Ztest case '{tc_name}' matched to '{tc_fq_id}")
                return tc
        logger.debug(
            f"On {phase}: Ztest case '{tc_name}' is not known"
            f" in {self.started_suites} running suite(s)."
        )
        tc_id = self.instance.compose_case_name(tc_name)
        return self.instance.get_case_or_create(tc_id)

    def start_suite(self, suite_name):
        if suite_name not in self.detected_suite_names:
            self.detected_suite_names.append(suite_name)
        if suite_name not in self.instance.testsuite.ztest_suite_names:
            logger.warning(f"Unexpected Ztest suite '{suite_name}'")
        if suite_name in self.started_suites:
            if self.started_suites[suite_name]['count'] > 0:
                logger.warning(f"Already STARTED '{suite_name}':{self.started_suites[suite_name]}")
            elif self.trace:
                logger.debug(f"START suite '{suite_name}'")
            self.started_suites[suite_name]['count'] += 1
            self.started_suites[suite_name]['repeat'] += 1
        else:
            self.started_suites[suite_name] = { 'count': 1, 'repeat': 0 }

    def end_suite(self, suite_name, phase='', suite_status=None):
        if suite_name in self.started_suites:
            if phase == 'TS_SUM' and self.started_suites[suite_name]['count'] == 0:
                return
            if self.started_suites[suite_name]['count'] < 1:
                logger.error(
                    f"Already ENDED {phase} suite '{suite_name}':{self.started_suites[suite_name]}"
                )
            elif self.trace:
                logger.debug(f"END {phase} suite '{suite_name}':{self.started_suites[suite_name]}")
            self.started_suites[suite_name]['count'] -= 1
        elif suite_status == 'SKIP':
            self.start_suite(suite_name)  # register skipped suites at their summary end
            self.started_suites[suite_name]['count'] -= 1
        else:
            logger.warning(f"END {phase} suite '{suite_name}' without START detected")

    def start_case(self, tc_name):
        if tc_name in self.started_cases:
            if self.started_cases[tc_name]['count'] > 0:
                logger.warning(f"Already STARTED '{tc_name}':{self.started_cases[tc_name]}")
            self.started_cases[tc_name]['count'] += 1
        else:
            self.started_cases[tc_name] = { 'count': 1 }

    def end_case(self, tc_name, phase=''):
        if tc_name in self.started_cases:
            if phase == 'TS_SUM' and self.started_cases[tc_name]['count'] == 0:
                return
            if self.started_cases[tc_name]['count'] < 1:
                logger.error(
                    f"Already ENDED {phase} case '{tc_name}':{self.started_cases[tc_name]}"
                )
            elif self.trace:
                logger.debug(f"END {phase} case '{tc_name}':{self.started_cases[tc_name]}")
            self.started_cases[tc_name]['count'] -= 1
        elif phase != 'TS_SUM':
            logger.warning(f"END {phase} case '{tc_name}' without START detected")


    def handle(self, line):
        testcase_match = None
        if self._match:
            self.testcase_output += line + "\n"

        if test_suite_start_match := re.search(self.test_suite_start_pattern, line):
            self.start_suite(test_suite_start_match.group("suite_name"))
        elif test_suite_end_match := re.search(self.test_suite_end_pattern, line):
            suite_name=test_suite_end_match.group("suite_name")
            self.end_suite(suite_name, 'TS_END')
        elif testcase_match := re.search(self.test_case_start_pattern, line):
            tc_name = testcase_match.group(2)
            tc = self.get_testcase(tc_name, 'TC_START')
            self.start_case(tc.name)
            # Mark the test as started, if something happens here, it is mostly
            # due to this tests, for example timeout. This should in this case
            # be marked as failed and not blocked (not run).
            tc.status = TwisterStatus.STARTED
            if not self._match:
                self.testcase_output += line + "\n"
                self._match = True
        # some testcases are skipped based on predicates and do not show up
        # during test execution, however they are listed in the summary. Parse
        # the summary for status and use that status instead.
        elif result_match := self.test_case_end_pattern.match(line):
            matched_status = result_match.group(1)
            tc_name = result_match.group(3)
            tc = self.get_testcase(tc_name, 'TC_END')
            self.end_case(tc.name)
            tc.status = TwisterStatus[matched_status]
            if tc.status == TwisterStatus.SKIP:
                tc.reason = "ztest skip"
            tc.duration = float(result_match.group(4))
            if tc.status == TwisterStatus.FAIL:
                tc.output = self.testcase_output
            self.testcase_output = ""
            self._match = False
            self.ztest = True
        elif test_suite_summary_match := self.test_suite_summary_pattern.match(line):
            suite_name=test_suite_summary_match.group("suite_name")
            suite_status=test_suite_summary_match.group("suite_status")
            self._match = False
            self.ztest = True
            self.end_suite(suite_name, 'TS_SUM', suite_status=suite_status)
        elif test_case_summary_match := self.test_case_summary_pattern.match(line):
            matched_status = test_case_summary_match.group(1)
            suite_name = test_case_summary_match.group(2)
            tc_name = test_case_summary_match.group(4)
            tc = self.get_testcase(tc_name, 'TS_SUM', suite_name)
            self.end_case(tc.name, 'TS_SUM')
            tc.status = TwisterStatus[matched_status]
            if tc.status == TwisterStatus.SKIP:
                tc.reason = "ztest skip"
            tc.duration = float(test_case_summary_match.group(5))
            if tc.status == TwisterStatus.FAIL:
                tc.output = self.testcase_output
            self.testcase_output = ""
            self._match = False
            self.ztest = True

        self.process_test(line)

        if not self.ztest and self.status != TwisterStatus.NONE:
            logger.debug(f"not a ztest and no state for {self.id}")
            tc = self.instance.get_case_or_create(self.id)
            if self.status == TwisterStatus.PASS:
                tc.status = TwisterStatus.PASS
            else:
                tc.status = TwisterStatus.FAIL
                tc.reason = "Test failure"


class Ztest(Test):
    pass


class Bsim(Harness):

    def build(self):
        """
        Copying the application executable to BabbleSim's bin directory enables
        running multidevice bsim tests after twister has built them.
        """

        if self.instance is None:
            return

        original_exe_path: str = os.path.join(self.instance.build_dir, 'zephyr', 'zephyr.exe')
        if not os.path.exists(original_exe_path):
            logger.warning('Cannot copy bsim exe - cannot find original executable.')
            return

        bsim_out_path: str = os.getenv('BSIM_OUT_PATH', '')
        if not bsim_out_path:
            logger.warning('Cannot copy bsim exe - BSIM_OUT_PATH not provided.')
            return

        new_exe_name: str = self.instance.testsuite.harness_config.get('bsim_exe_name', '')
        if new_exe_name:
            new_exe_name = f'bs_{self.instance.platform.name}_{new_exe_name}'
        else:
            new_exe_name = self.instance.name
            new_exe_name = f'bs_{new_exe_name}'

        new_exe_name = new_exe_name.replace(os.path.sep, '_').replace('.', '_').replace('@', '_')

        new_exe_path: str = os.path.join(bsim_out_path, 'bin', new_exe_name)
        logger.debug(f'Copying executable from {original_exe_path} to {new_exe_path}')
        shutil.copy(original_exe_path, new_exe_path)


class HarnessImporter:

    @staticmethod
    def get_harness(harness_name):
        thismodule = sys.modules[__name__]
        try:
            if harness_name:
                harness_class = getattr(thismodule, harness_name)
            else:
                harness_class = thismodule.Test
            return harness_class()
        except AttributeError as e:
            logger.debug(f"harness {harness_name} not implemented: {e}")
            return None