# vim: set syntax=python ts=4 :
#
# Copyright (c) 20180-2022 Intel Corporation
# Copyright 2022 NXP
# SPDX-License-Identifier: Apache-2.0

import logging
import multiprocessing
import os
import pickle
import queue
import re
import shutil
import subprocess
import sys
import time
import traceback
import yaml
from multiprocessing import Lock, Process, Value
from multiprocessing.managers import BaseManager
from typing import List
from packaging import version

from colorama import Fore
from domains import Domains
from twisterlib.cmakecache import CMakeCache
from twisterlib.environment import canonical_zephyr_base
from twisterlib.error import BuildError, ConfigurationError

import elftools
from elftools.elf.elffile import ELFFile
from elftools.elf.sections import SymbolTableSection

if version.parse(elftools.__version__) < version.parse('0.24'):
    sys.exit("pyelftools is out of date, need version 0.24 or later")

# Job server only works on Linux for now.
if sys.platform == 'linux':
    from twisterlib.jobserver import GNUMakeJobClient, GNUMakeJobServer, JobClient

from twisterlib.log_helper import log_command
from twisterlib.testinstance import TestInstance
from twisterlib.environment import TwisterEnv
from twisterlib.testsuite import TestSuite
from twisterlib.platform import Platform
from twisterlib.testplan import change_skip_to_error_if_integration
from twisterlib.harness import HarnessImporter, Pytest

try:
    from yaml import CSafeLoader as SafeLoader
except ImportError:
    from yaml import SafeLoader

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


class ExecutionCounter(object):
    def __init__(self, total=0):
        '''
        Most of the stats are at test instance level
        Except that "_cases" and "_skipped_cases" are for cases of ALL test instances

        total complete = done + skipped_filter
        total = yaml test scenarios * applicable platforms
        complete perctenage = (done + skipped_filter) / total
        pass rate = passed / (total - skipped_configs)
        '''
        # instances that go through the pipeline
        # updated by report_out()
        self._done = Value('i', 0)

        # iteration
        self._iteration = Value('i', 0)

        # instances that actually executed and passed
        # updated by report_out()
        self._passed = Value('i', 0)

        # static filter + runtime filter + build skipped
        # updated by update_counting_before_pipeline() and report_out()
        self._skipped_configs = Value('i', 0)

        # cmake filter + build skipped
        # updated by report_out()
        self._skipped_runtime = Value('i', 0)

        # staic filtered at yaml parsing time
        # updated by update_counting_before_pipeline()
        self._skipped_filter = Value('i', 0)

        # updated by update_counting_before_pipeline() and report_out()
        self._skipped_cases = Value('i', 0)

        # updated by report_out() in pipeline
        self._error = Value('i', 0)
        self._failed = Value('i', 0)

        # initialized to number of test instances
        self._total = Value('i', total)

        # updated in report_out
        self._cases = Value('i', 0)
        self.lock = Lock()

    def summary(self):
        print("--------------------------------")
        print(f"Total test suites: {self.total}") # actually test instances
        print(f"Total test cases: {self.cases}")
        print(f"Executed test cases: {self.cases - self.skipped_cases}")
        print(f"Skipped test cases: {self.skipped_cases}")
        print(f"Completed test suites: {self.done}")
        print(f"Passing test suites: {self.passed}")
        print(f"Failing test suites: {self.failed}")
        print(f"Skipped test suites: {self.skipped_configs}")
        print(f"Skipped test suites (runtime): {self.skipped_runtime}")
        print(f"Skipped test suites (filter): {self.skipped_filter}")
        print(f"Errors: {self.error}")
        print("--------------------------------")

    @property
    def cases(self):
        with self._cases.get_lock():
            return self._cases.value

    @cases.setter
    def cases(self, value):
        with self._cases.get_lock():
            self._cases.value = value

    @property
    def skipped_cases(self):
        with self._skipped_cases.get_lock():
            return self._skipped_cases.value

    @skipped_cases.setter
    def skipped_cases(self, value):
        with self._skipped_cases.get_lock():
            self._skipped_cases.value = value

    @property
    def error(self):
        with self._error.get_lock():
            return self._error.value

    @error.setter
    def error(self, value):
        with self._error.get_lock():
            self._error.value = value

    @property
    def iteration(self):
        with self._iteration.get_lock():
            return self._iteration.value

    @iteration.setter
    def iteration(self, value):
        with self._iteration.get_lock():
            self._iteration.value = value

    @property
    def done(self):
        with self._done.get_lock():
            return self._done.value

    @done.setter
    def done(self, value):
        with self._done.get_lock():
            self._done.value = value

    @property
    def passed(self):
        with self._passed.get_lock():
            return self._passed.value

    @passed.setter
    def passed(self, value):
        with self._passed.get_lock():
            self._passed.value = value

    @property
    def skipped_configs(self):
        with self._skipped_configs.get_lock():
            return self._skipped_configs.value

    @skipped_configs.setter
    def skipped_configs(self, value):
        with self._skipped_configs.get_lock():
            self._skipped_configs.value = value

    @property
    def skipped_filter(self):
        with self._skipped_filter.get_lock():
            return self._skipped_filter.value

    @skipped_filter.setter
    def skipped_filter(self, value):
        with self._skipped_filter.get_lock():
            self._skipped_filter.value = value

    @property
    def skipped_runtime(self):
        with self._skipped_runtime.get_lock():
            return self._skipped_runtime.value

    @skipped_runtime.setter
    def skipped_runtime(self, value):
        with self._skipped_runtime.get_lock():
            self._skipped_runtime.value = value

    @property
    def failed(self):
        with self._failed.get_lock():
            return self._failed.value

    @failed.setter
    def failed(self, value):
        with self._failed.get_lock():
            self._failed.value = value

    @property
    def total(self):
        with self._total.get_lock():
            return self._total.value

class CMake:
    config_re = re.compile('(CONFIG_[A-Za-z0-9_]+)[=]\"?([^\"]*)\"?$')
    dt_re = re.compile('([A-Za-z0-9_]+)[=]\"?([^\"]*)\"?$')

    def __init__(self, testsuite: TestSuite, platform: Platform, source_dir, build_dir, jobserver):

        self.cwd = None
        self.capture_output = True

        self.defconfig = {}
        self.cmake_cache = {}

        self.instance = None
        self.testsuite = testsuite
        self.platform = platform
        self.source_dir = source_dir
        self.build_dir = build_dir
        self.log = "build.log"

        self.default_encoding = sys.getdefaultencoding()
        self.jobserver = jobserver

    def parse_generated(self, filter_stages=[]):
        self.defconfig = {}
        return {}

    def run_build(self, args=[]):

        logger.debug("Building %s for %s" % (self.source_dir, self.platform.name))

        cmake_args = []
        cmake_args.extend(args)
        cmake = shutil.which('cmake')
        cmd = [cmake] + cmake_args
        kwargs = dict()

        if self.capture_output:
            kwargs['stdout'] = subprocess.PIPE
            # CMake sends the output of message() to stderr unless it's STATUS
            kwargs['stderr'] = subprocess.STDOUT

        if self.cwd:
            kwargs['cwd'] = self.cwd

        start_time = time.time()
        if sys.platform == 'linux':
            p = self.jobserver.popen(cmd, **kwargs)
        else:
            p = subprocess.Popen(cmd, **kwargs)
        logger.debug(f'Running {" ".join(cmd)}')

        out, _ = p.communicate()

        ret = {}
        duration = time.time() - start_time
        self.instance.build_time += duration
        if p.returncode == 0:
            msg = f"Finished building {self.source_dir} for {self.platform.name} in {duration:.2f} seconds"
            logger.debug(msg)

            self.instance.status = "passed"
            if not self.instance.run:
                self.instance.add_missing_case_status("skipped", "Test was built only")
            ret = {"returncode": p.returncode}

            if out:
                log_msg = out.decode(self.default_encoding)
                with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
                    log.write(log_msg)
            else:
                return None
        else:
            # A real error occurred, raise an exception
            log_msg = ""
            if out:
                log_msg = out.decode(self.default_encoding)
                with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
                    log.write(log_msg)

            if log_msg:
                overflow_found = re.findall("region `(FLASH|ROM|RAM|ICCM|DCCM|SRAM|dram\\d_\\d_seg)' overflowed by", log_msg)
                imgtool_overflow_found = re.findall(r"Error: Image size \(.*\) \+ trailer \(.*\) exceeds requested size", log_msg)
                if overflow_found and not self.options.overflow_as_errors:
                    logger.debug("Test skipped due to {} Overflow".format(overflow_found[0]))
                    self.instance.status = "skipped"
                    self.instance.reason = "{} overflow".format(overflow_found[0])
                    change_skip_to_error_if_integration(self.options, self.instance)
                elif imgtool_overflow_found and not self.options.overflow_as_errors:
                    self.instance.status = "skipped"
                    self.instance.reason = "imgtool overflow"
                    change_skip_to_error_if_integration(self.options, self.instance)
                else:
                    self.instance.status = "error"
                    self.instance.reason = "Build failure"

            ret = {
                "returncode": p.returncode
            }

        return ret

    def run_cmake(self, args="", filter_stages=[]):

        if not self.options.disable_warnings_as_errors:
            warnings_as_errors = 'y'
            gen_defines_args = "--edtlib-Werror"
        else:
            warnings_as_errors = 'n'
            gen_defines_args = ""

        warning_command = 'CONFIG_COMPILER_WARNINGS_AS_ERRORS'
        if self.testsuite.sysbuild:
            warning_command = 'SB_' + warning_command

        logger.debug("Running cmake on %s for %s" % (self.source_dir, self.platform.name))
        cmake_args = [
            f'-B{self.build_dir}',
            f'-DTC_RUNID={self.instance.run_id}',
            f'-D{warning_command}={warnings_as_errors}',
            f'-DEXTRA_GEN_DEFINES_ARGS={gen_defines_args}',
            f'-G{self.env.generator}'
        ]

        # If needed, run CMake using the package_helper script first, to only run
        # a subset of all cmake modules. This output will be used to filter
        # testcases, and the full CMake configuration will be run for
        # testcases that should be built
        if filter_stages:
            cmake_filter_args = [
                f'-DMODULES={",".join(filter_stages)}',
                f'-P{canonical_zephyr_base}/cmake/package_helper.cmake',
            ]

        if self.testsuite.sysbuild and not filter_stages:
            logger.debug("Building %s using sysbuild" % (self.source_dir))
            source_args = [
                f'-S{canonical_zephyr_base}/share/sysbuild',
                f'-DAPP_DIR={self.source_dir}'
            ]
        else:
            source_args = [
                f'-S{self.source_dir}'
            ]
        cmake_args.extend(source_args)

        cmake_args.extend(args)

        cmake_opts = ['-DBOARD={}'.format(self.platform.name)]
        cmake_args.extend(cmake_opts)

        if self.instance.testsuite.required_snippets:
            cmake_opts = ['-DSNIPPET={}'.format(';'.join(self.instance.testsuite.required_snippets))]
            cmake_args.extend(cmake_opts)

        cmake = shutil.which('cmake')
        cmd = [cmake] + cmake_args

        if filter_stages:
            cmd += cmake_filter_args

        kwargs = dict()

        log_command(logger, "Calling cmake", cmd)

        if self.capture_output:
            kwargs['stdout'] = subprocess.PIPE
            # CMake sends the output of message() to stderr unless it's STATUS
            kwargs['stderr'] = subprocess.STDOUT

        if self.cwd:
            kwargs['cwd'] = self.cwd

        start_time = time.time()
        if sys.platform == 'linux':
            p = self.jobserver.popen(cmd, **kwargs)
        else:
            p = subprocess.Popen(cmd, **kwargs)
        out, _ = p.communicate()

        duration = time.time() - start_time
        self.instance.build_time += duration

        if p.returncode == 0:
            filter_results = self.parse_generated(filter_stages)
            msg = f"Finished running cmake {self.source_dir} for {self.platform.name} in {duration:.2f} seconds"
            logger.debug(msg)
            ret = {
                    'returncode': p.returncode,
                    'filter': filter_results
                    }
        else:
            self.instance.status = "error"
            self.instance.reason = "Cmake build failure"

            for tc in self.instance.testcases:
                tc.status = self.instance.status

            logger.error("Cmake build failure: %s for %s" % (self.source_dir, self.platform.name))
            ret = {"returncode": p.returncode}

        if out:
            os.makedirs(self.build_dir, exist_ok=True)
            with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
                log_msg = out.decode(self.default_encoding)
                log.write(log_msg)

        return ret


class FilterBuilder(CMake):

    def __init__(self, testsuite: TestSuite, platform: Platform, source_dir, build_dir, jobserver):
        super().__init__(testsuite, platform, source_dir, build_dir, jobserver)

        self.log = "config-twister.log"

    def parse_generated(self, filter_stages=[]):

        if self.platform.name == "unit_testing":
            return {}

        if self.testsuite.sysbuild and not filter_stages:
            # Load domain yaml to get default domain build directory
            domain_path = os.path.join(self.build_dir, "domains.yaml")
            domains = Domains.from_file(domain_path)
            logger.debug("Loaded sysbuild domain data from %s" % (domain_path))
            self.instance.domains = domains
            domain_build = domains.get_default_domain().build_dir
            cmake_cache_path = os.path.join(domain_build, "CMakeCache.txt")
            defconfig_path = os.path.join(domain_build, "zephyr", ".config")
            edt_pickle = os.path.join(domain_build, "zephyr", "edt.pickle")
        else:
            cmake_cache_path = os.path.join(self.build_dir, "CMakeCache.txt")
            # .config is only available after kconfig stage in cmake. If only dt based filtration is required
            # package helper call won't produce .config
            if not filter_stages or "kconfig" in filter_stages:
                defconfig_path = os.path.join(self.build_dir, "zephyr", ".config")
            # dt is compiled before kconfig, so edt_pickle is available regardless of choice of filter stages
            edt_pickle = os.path.join(self.build_dir, "zephyr", "edt.pickle")


        if not filter_stages or "kconfig" in filter_stages:
            with open(defconfig_path, "r") as fp:
                defconfig = {}
                for line in fp.readlines():
                    m = self.config_re.match(line)
                    if not m:
                        if line.strip() and not line.startswith("#"):
                            sys.stderr.write("Unrecognized line %s\n" % line)
                        continue
                    defconfig[m.group(1)] = m.group(2).strip()

            self.defconfig = defconfig

        cmake_conf = {}
        try:
            cache = CMakeCache.from_file(cmake_cache_path)
        except FileNotFoundError:
            cache = {}

        for k in iter(cache):
            cmake_conf[k.name] = k.value

        self.cmake_cache = cmake_conf

        filter_data = {
            "ARCH": self.platform.arch,
            "PLATFORM": self.platform.name
        }
        filter_data.update(os.environ)
        if not filter_stages or "kconfig" in filter_stages:
            filter_data.update(self.defconfig)
        filter_data.update(self.cmake_cache)

        if self.testsuite.sysbuild and self.env.options.device_testing:
            # Verify that twister's arguments support sysbuild.
            # Twister sysbuild flashing currently only works with west, so
            # --west-flash must be passed. Additionally, erasing the DUT
            # before each test with --west-flash=--erase will inherently not
            # work with sysbuild.
            if self.env.options.west_flash is None:
                logger.warning("Sysbuild test will be skipped. " +
                    "West must be used for flashing.")
                return {os.path.join(self.platform.name, self.testsuite.name): True}
            elif "--erase" in self.env.options.west_flash:
                logger.warning("Sysbuild test will be skipped, " +
                    "--erase is not supported with --west-flash")
                return {os.path.join(self.platform.name, self.testsuite.name): True}

        if self.testsuite and self.testsuite.filter:
            try:
                if os.path.exists(edt_pickle):
                    with open(edt_pickle, 'rb') as f:
                        edt = pickle.load(f)
                else:
                    edt = None
                ret = expr_parser.parse(self.testsuite.filter, filter_data, edt)

            except (ValueError, SyntaxError) as se:
                sys.stderr.write(
                    "Failed processing %s\n" % self.testsuite.yamlfile)
                raise se

            if not ret:
                return {os.path.join(self.platform.name, self.testsuite.name): True}
            else:
                return {os.path.join(self.platform.name, self.testsuite.name): False}
        else:
            self.platform.filter_data = filter_data
            return filter_data


class ProjectBuilder(FilterBuilder):

    def __init__(self, instance: TestInstance, env: TwisterEnv, jobserver, **kwargs):
        super().__init__(instance.testsuite, instance.platform, instance.testsuite.source_dir, instance.build_dir, jobserver)

        self.log = "build.log"
        self.instance = instance
        self.filtered_tests = 0
        self.options = env.options
        self.env = env
        self.duts = None

    def log_info(self, filename, inline_logs, log_testcases=False):
        filename = os.path.abspath(os.path.realpath(filename))
        if inline_logs:
            logger.info("{:-^100}".format(filename))

            try:
                with open(filename) as fp:
                    data = fp.read()
            except Exception as e:
                data = "Unable to read log data (%s)\n" % (str(e))

            logger.error(data)

            logger.info("{:-^100}".format(filename))

            if log_testcases:
                for tc in self.instance.testcases:
                    if not tc.reason:
                        continue
                    logger.info(
                        f"\n{str(tc.name).center(100, '_')}\n"
                        f"{tc.reason}\n"
                        f"{100*'_'}\n"
                        f"{tc.output}"
                    )
        else:
            logger.error("see: " + Fore.YELLOW + filename + Fore.RESET)

    def log_info_file(self, inline_logs):
        build_dir = self.instance.build_dir
        h_log = "{}/handler.log".format(build_dir)
        b_log = "{}/build.log".format(build_dir)
        v_log = "{}/valgrind.log".format(build_dir)
        d_log = "{}/device.log".format(build_dir)
        pytest_log = "{}/twister_harness.log".format(build_dir)

        if os.path.exists(v_log) and "Valgrind" in self.instance.reason:
            self.log_info("{}".format(v_log), inline_logs)
        elif os.path.exists(pytest_log) and os.path.getsize(pytest_log) > 0:
            self.log_info("{}".format(pytest_log), inline_logs, log_testcases=True)
        elif os.path.exists(h_log) and os.path.getsize(h_log) > 0:
            self.log_info("{}".format(h_log), inline_logs)
        elif os.path.exists(d_log) and os.path.getsize(d_log) > 0:
            self.log_info("{}".format(d_log), inline_logs)
        else:
            self.log_info("{}".format(b_log), inline_logs)


    def process(self, pipeline, done, message, lock, results):
        op = message.get('op')

        self.instance.setup_handler(self.env)

        if op == "filter":
            ret = self.cmake(filter_stages=self.instance.filter_stages)
            if self.instance.status in ["failed", "error"]:
                pipeline.put({"op": "report", "test": self.instance})
            else:
                # Here we check the dt/kconfig filter results coming from running cmake
                if self.instance.name in ret['filter'] and ret['filter'][self.instance.name]:
                    logger.debug("filtering %s" % self.instance.name)
                    self.instance.status = "filtered"
                    self.instance.reason = "runtime filter"
                    results.skipped_runtime += 1
                    self.instance.add_missing_case_status("skipped")
                    pipeline.put({"op": "report", "test": self.instance})
                else:
                    pipeline.put({"op": "cmake", "test": self.instance})

        # The build process, call cmake and build with configured generator
        elif op == "cmake":
            ret = self.cmake()
            if self.instance.status in ["failed", "error"]:
                pipeline.put({"op": "report", "test": self.instance})
            elif self.options.cmake_only:
                if self.instance.status is None:
                    self.instance.status = "passed"
                pipeline.put({"op": "report", "test": self.instance})
            else:
                # Here we check the runtime filter results coming from running cmake
                if self.instance.name in ret['filter'] and ret['filter'][self.instance.name]:
                    logger.debug("filtering %s" % self.instance.name)
                    self.instance.status = "filtered"
                    self.instance.reason = "runtime filter"
                    results.skipped_runtime += 1
                    self.instance.add_missing_case_status("skipped")
                    pipeline.put({"op": "report", "test": self.instance})
                else:
                    pipeline.put({"op": "build", "test": self.instance})

        elif op == "build":
            logger.debug("build test: %s" % self.instance.name)
            ret = self.build()
            if not ret:
                self.instance.status = "error"
                self.instance.reason = "Build Failure"
                pipeline.put({"op": "report", "test": self.instance})
            else:
                # Count skipped cases during build, for example
                # due to ram/rom overflow.
                if  self.instance.status == "skipped":
                    results.skipped_runtime += 1
                    self.instance.add_missing_case_status("skipped", self.instance.reason)

                if ret.get('returncode', 1) > 0:
                    self.instance.add_missing_case_status("blocked", self.instance.reason)
                    pipeline.put({"op": "report", "test": self.instance})
                else:
                    logger.debug(f"Determine test cases for test instance: {self.instance.name}")
                    try:
                        self.determine_testcases(results)
                        pipeline.put({"op": "gather_metrics", "test": self.instance})
                    except BuildError as e:
                        logger.error(str(e))
                        self.instance.status = "error"
                        self.instance.reason = str(e)
                        pipeline.put({"op": "report", "test": self.instance})

        elif op == "gather_metrics":
            ret = self.gather_metrics(self.instance)
            if not ret or ret.get('returncode', 1) > 0:
                self.instance.status = "error"
                self.instance.reason = "Build Failure at gather_metrics."
                pipeline.put({"op": "report", "test": self.instance})
            elif self.instance.run and self.instance.handler.ready:
                pipeline.put({"op": "run", "test": self.instance})
            else:
                pipeline.put({"op": "report", "test": self.instance})

        # Run the generated binary using one of the supported handlers
        elif op == "run":
            logger.debug("run test: %s" % self.instance.name)
            self.run()
            logger.debug(f"run status: {self.instance.name} {self.instance.status}")
            try:
                # to make it work with pickle
                self.instance.handler.thread = None
                self.instance.handler.duts = None
                pipeline.put({
                    "op": "report",
                    "test": self.instance,
                    "status": self.instance.status,
                    "reason": self.instance.reason
                    }
                )
            except RuntimeError as e:
                logger.error(f"RuntimeError: {e}")
                traceback.print_exc()

        # Report results and output progress to screen
        elif op == "report":
            with lock:
                done.put(self.instance)
                self.report_out(results)

            if not self.options.coverage:
                if self.options.prep_artifacts_for_testing:
                    pipeline.put({"op": "cleanup", "mode": "device", "test": self.instance})
                elif self.options.runtime_artifact_cleanup == "pass" and self.instance.status == "passed":
                    pipeline.put({"op": "cleanup", "mode": "passed", "test": self.instance})
                elif self.options.runtime_artifact_cleanup == "all":
                    pipeline.put({"op": "cleanup", "mode": "all", "test": self.instance})

        elif op == "cleanup":
            mode = message.get("mode")
            if mode == "device":
                self.cleanup_device_testing_artifacts()
            elif mode == "passed" or (mode == "all" and self.instance.reason != "Cmake build failure"):
                self.cleanup_artifacts()

    def determine_testcases(self, results):
        yaml_testsuite_name = self.instance.testsuite.id
        logger.debug(f"Determine test cases for test suite: {yaml_testsuite_name}")

        elf_file = self.instance.get_elf_file()
        elf = ELFFile(open(elf_file, "rb"))

        logger.debug(f"Test instance {self.instance.name} already has {len(self.instance.testcases)} cases.")
        new_ztest_unit_test_regex = re.compile(r"z_ztest_unit_test__([^\s]+?)__([^\s]*)")
        detected_cases = []
        for section in elf.iter_sections():
            if isinstance(section, SymbolTableSection):
                for sym in section.iter_symbols():
                    # It is only meant for new ztest fx because only new ztest fx exposes test functions
                    # precisely.

                    # The 1st capture group is new ztest suite name.
                    # The 2nd capture group is new ztest unit test name.
                    matches = new_ztest_unit_test_regex.findall(sym.name)
                    if matches:
                        for m in matches:
                            # new_ztest_suite = m[0] # not used for now
                            test_func_name = m[1].replace("test_", "", 1)
                            testcase_id = f"{yaml_testsuite_name}.{test_func_name}"
                            detected_cases.append(testcase_id)

        if detected_cases:
            logger.debug(f"{', '.join(detected_cases)} in {elf_file}")
            self.instance.testcases.clear()
            self.instance.testsuite.testcases.clear()

            # When the old regex-based test case collection is fully deprecated,
            # this will be the sole place where test cases get added to the test instance.
            # Then we can further include the new_ztest_suite info in the testcase_id.

            for testcase_id in detected_cases:
                self.instance.add_testcase(name=testcase_id)
                self.instance.testsuite.add_testcase(name=testcase_id)


    def cleanup_artifacts(self, additional_keep: List[str] = []):
        logger.debug("Cleaning up {}".format(self.instance.build_dir))
        allow = [
            os.path.join('zephyr', '.config'),
            'handler.log',
            'build.log',
            'device.log',
            'recording.csv',
            # below ones are needed to make --test-only work as well
            'Makefile',
            'CMakeCache.txt',
            'build.ninja',
            os.path.join('CMakeFiles', 'rules.ninja')
            ]

        allow += additional_keep

        if self.options.runtime_artifact_cleanup == 'all':
            allow += [os.path.join('twister', 'testsuite_extra.conf')]

        allow = [os.path.join(self.instance.build_dir, file) for file in allow]

        for dirpath, dirnames, filenames in os.walk(self.instance.build_dir, topdown=False):
            for name in filenames:
                path = os.path.join(dirpath, name)
                if path not in allow:
                    os.remove(path)
            # Remove empty directories and symbolic links to directories
            for dir in dirnames:
                path = os.path.join(dirpath, dir)
                if os.path.islink(path):
                    os.remove(path)
                elif not os.listdir(path):
                    os.rmdir(path)

    def cleanup_device_testing_artifacts(self):
        logger.debug("Cleaning up for Device Testing {}".format(self.instance.build_dir))

        files_to_keep = self._get_binaries()
        files_to_keep.append(os.path.join('zephyr', 'runners.yaml'))

        if self.testsuite.sysbuild:
            files_to_keep.append('domains.yaml')
            for domain in self.instance.domains.get_domains():
                files_to_keep += self._get_artifact_allow_list_for_domain(domain.name)

        self.cleanup_artifacts(files_to_keep)

        self._sanitize_files()

    def _get_artifact_allow_list_for_domain(self, domain: str) -> List[str]:
        """
        Return a list of files needed to test a given domain.
        """
        allow = [
            os.path.join(domain, 'build.ninja'),
            os.path.join(domain, 'CMakeCache.txt'),
            os.path.join(domain, 'CMakeFiles', 'rules.ninja'),
            os.path.join(domain, 'Makefile'),
            os.path.join(domain, 'zephyr', '.config'),
            os.path.join(domain, 'zephyr', 'runners.yaml')
            ]
        return allow

    def _get_binaries(self) -> List[str]:
        """
        Get list of binaries paths (absolute or relative to the
        self.instance.build_dir), basing on information from platform.binaries
        or runners.yaml. If they are not found take default binaries like
        "zephyr/zephyr.hex" etc.
        """
        binaries: List[str] = []

        platform = self.instance.platform
        if platform.binaries:
            for binary in platform.binaries:
                binaries.append(os.path.join('zephyr', binary))

        # Get binaries for a single-domain build
        binaries += self._get_binaries_from_runners()
        # Get binaries in the case of a multiple-domain build
        if self.testsuite.sysbuild:
            for domain in self.instance.domains.get_domains():
                binaries += self._get_binaries_from_runners(domain.name)

        # if binaries was not found in platform.binaries and runners.yaml take default ones
        if len(binaries) == 0:
            binaries = [
                os.path.join('zephyr', 'zephyr.hex'),
                os.path.join('zephyr', 'zephyr.bin'),
                os.path.join('zephyr', 'zephyr.elf'),
                os.path.join('zephyr', 'zephyr.exe'),
            ]
        return binaries

    def _get_binaries_from_runners(self, domain='') -> List[str]:
        """
        Get list of binaries paths (absolute or relative to the
        self.instance.build_dir) from runners.yaml file. May be used for
        multiple-domain builds by passing in one domain at a time.
        """

        runners_file_path: str = os.path.join(self.instance.build_dir,
                                              domain, 'zephyr', 'runners.yaml')
        if not os.path.exists(runners_file_path):
            return []

        with open(runners_file_path, 'r') as file:
            runners_content: dict = yaml.load(file, Loader=SafeLoader)

        if 'config' not in runners_content:
            return []

        runners_config: dict = runners_content['config']
        binary_keys: List[str] = ['elf_file', 'hex_file', 'bin_file']

        binaries: List[str] = []
        for binary_key in binary_keys:
            binary_path = runners_config.get(binary_key)
            if binary_path is None:
                continue
            if os.path.isabs(binary_path):
                binaries.append(binary_path)
            else:
                binaries.append(os.path.join(domain, 'zephyr', binary_path))

        return binaries

    def _sanitize_files(self):
        """
        Sanitize files to make it possible to flash those file on different
        computer/system.
        """
        self._sanitize_runners_file()
        self._sanitize_zephyr_base_from_files()

    def _sanitize_runners_file(self):
        """
        Replace absolute paths of binary files for relative ones. The base
        directory for those files is f"{self.instance.build_dir}/zephyr"
        """
        runners_dir_path: str = os.path.join(self.instance.build_dir, 'zephyr')
        runners_file_path: str = os.path.join(runners_dir_path, 'runners.yaml')
        if not os.path.exists(runners_file_path):
            return

        with open(runners_file_path, 'rt') as file:
            runners_content_text = file.read()
            runners_content_yaml: dict = yaml.load(runners_content_text, Loader=SafeLoader)

        if 'config' not in runners_content_yaml:
            return

        runners_config: dict = runners_content_yaml['config']
        binary_keys: List[str] = ['elf_file', 'hex_file', 'bin_file']

        for binary_key in binary_keys:
            binary_path = runners_config.get(binary_key)
            # sanitize only paths which exist and are absolute
            if binary_path is None or not os.path.isabs(binary_path):
                continue
            binary_path_relative = os.path.relpath(binary_path, start=runners_dir_path)
            runners_content_text = runners_content_text.replace(binary_path, binary_path_relative)

        with open(runners_file_path, 'wt') as file:
            file.write(runners_content_text)

    def _sanitize_zephyr_base_from_files(self):
        """
        Remove Zephyr base paths from selected files.
        """
        files_to_sanitize = [
            'CMakeCache.txt',
            os.path.join('zephyr', 'runners.yaml'),
        ]
        for file_path in files_to_sanitize:
            file_path = os.path.join(self.instance.build_dir, file_path)
            if not os.path.exists(file_path):
                continue

            with open(file_path, "rt") as file:
                data = file.read()

            # add trailing slash at the end of canonical_zephyr_base if it does not exist:
            path_to_remove = os.path.join(canonical_zephyr_base, "")
            data = data.replace(path_to_remove, "")

            with open(file_path, "wt") as file:
                file.write(data)

    def report_out(self, results):
        total_to_do = results.total
        total_tests_width = len(str(total_to_do))
        results.done += 1
        instance = self.instance
        if results.iteration == 1:
            results.cases += len(instance.testcases)

        if instance.status in ["error", "failed"]:
            if instance.status == "error":
                results.error += 1
                txt = " ERROR "
            else:
                results.failed += 1
                txt = " FAILED "
            if self.options.verbose:
                status = Fore.RED + txt + Fore.RESET + instance.reason
            else:
                logger.error(
                    "{:<25} {:<50} {}{}{}: {}".format(
                        instance.platform.name,
                        instance.testsuite.name,
                        Fore.RED,
                        txt,
                        Fore.RESET,
                        instance.reason))
            if not self.options.verbose:
                self.log_info_file(self.options.inline_logs)
        elif instance.status in ["skipped", "filtered"]:
            status = Fore.YELLOW + "SKIPPED" + Fore.RESET
            results.skipped_configs += 1
            # test cases skipped at the test instance level
            results.skipped_cases += len(instance.testsuite.testcases)
        elif instance.status == "passed":
            status = Fore.GREEN + "PASSED" + Fore.RESET
            results.passed += 1
            for case in instance.testcases:
                # test cases skipped at the test case level
                if case.status == 'skipped':
                    results.skipped_cases += 1
        else:
            logger.debug(f"Unknown status = {instance.status}")
            status = Fore.YELLOW + "UNKNOWN" + Fore.RESET

        if self.options.verbose:
            if self.options.cmake_only:
                more_info = "cmake"
            elif instance.status in ["skipped", "filtered"]:
                more_info = instance.reason
            else:
                if instance.handler.ready and instance.run:
                    more_info = instance.handler.type_str
                    htime = instance.execution_time
                    if instance.dut:
                        more_info += f": {instance.dut},"
                    if htime:
                        more_info += " {:.3f}s".format(htime)
                else:
                    more_info = "build"

                if ( instance.status in ["error", "failed", "timeout", "flash_error"]
                     and hasattr(self.instance.handler, 'seed')
                     and self.instance.handler.seed is not None ):
                    more_info += "/seed: " + str(self.options.seed)
            logger.info("{:>{}}/{} {:<25} {:<50} {} ({})".format(
                results.done, total_tests_width, total_to_do , instance.platform.name,
                instance.testsuite.name, status, more_info))

            if instance.status in ["error", "failed", "timeout"]:
                self.log_info_file(self.options.inline_logs)
        else:
            completed_perc = 0
            if total_to_do > 0:
                completed_perc = int((float(results.done) / total_to_do) * 100)

            sys.stdout.write("INFO    - Total complete: %s%4d/%4d%s  %2d%%  skipped: %s%4d%s, failed: %s%4d%s, error: %s%4d%s\r" % (
                Fore.GREEN,
                results.done,
                total_to_do,
                Fore.RESET,
                completed_perc,
                Fore.YELLOW if results.skipped_configs > 0 else Fore.RESET,
                results.skipped_configs,
                Fore.RESET,
                Fore.RED if results.failed > 0 else Fore.RESET,
                results.failed,
                Fore.RESET,
                Fore.RED if results.error > 0 else Fore.RESET,
                results.error,
                Fore.RESET
                )
                )
        sys.stdout.flush()

    @staticmethod
    def cmake_assemble_args(extra_args, handler, extra_conf_files, extra_overlay_confs,
                            extra_dtc_overlay_files, cmake_extra_args,
                            build_dir):
        # Retain quotes around config options
        config_options = [arg for arg in extra_args if arg.startswith("CONFIG_")]
        args = [arg for arg in extra_args if not arg.startswith("CONFIG_")]

        args_expanded = ["-D{}".format(a.replace('"', '\"')) for a in config_options]

        if handler.ready:
            args.extend(handler.args)

        if extra_conf_files:
            args.append(f"CONF_FILE=\"{';'.join(extra_conf_files)}\"")

        if extra_dtc_overlay_files:
            args.append(f"DTC_OVERLAY_FILE=\"{';'.join(extra_dtc_overlay_files)}\"")

        # merge overlay files into one variable
        overlays = extra_overlay_confs.copy()

        additional_overlay_path = os.path.join(
            build_dir, "twister", "testsuite_extra.conf"
        )
        if os.path.exists(additional_overlay_path):
            overlays.append(additional_overlay_path)

        if overlays:
            args.append("OVERLAY_CONFIG=\"%s\"" % (" ".join(overlays)))

        # Build the final argument list
        args_expanded.extend(["-D{}".format(a.replace('"', '\"')) for a in cmake_extra_args])
        args_expanded.extend(["-D{}".format(a.replace('"', '')) for a in args])

        return args_expanded

    def cmake(self, filter_stages=[]):
        args = self.cmake_assemble_args(
            self.testsuite.extra_args.copy(), # extra_args from YAML
            self.instance.handler,
            self.testsuite.extra_conf_files,
            self.testsuite.extra_overlay_confs,
            self.testsuite.extra_dtc_overlay_files,
            self.options.extra_args, # CMake extra args
            self.instance.build_dir,
        )
        return self.run_cmake(args,filter_stages)

    def build(self):
        harness = HarnessImporter.get_harness(self.instance.testsuite.harness.capitalize())
        build_result = self.run_build(['--build', self.build_dir])
        try:
            if harness:
                harness.instance = self.instance
                harness.build()
        except ConfigurationError as error:
            self.instance.status = "error"
            self.instance.reason = str(error)
            logger.error(self.instance.reason)
            return
        return build_result

    def run(self):

        instance = self.instance

        if instance.handler.ready:
            logger.debug(f"Reset instance status from '{instance.status}' to None before run.")
            instance.status = None

            if instance.handler.type_str == "device":
                instance.handler.duts = self.duts

            if(self.options.seed is not None and instance.platform.name.startswith("native_")):
                self.parse_generated()
                if('CONFIG_FAKE_ENTROPY_NATIVE_POSIX' in self.defconfig and
                    self.defconfig['CONFIG_FAKE_ENTROPY_NATIVE_POSIX'] == 'y'):
                    instance.handler.seed = self.options.seed

            if self.options.extra_test_args and instance.platform.arch == "posix":
                instance.handler.extra_test_args = self.options.extra_test_args

            harness = HarnessImporter.get_harness(instance.testsuite.harness.capitalize())
            try:
                harness.configure(instance)
            except ConfigurationError as error:
                instance.status = "error"
                instance.reason = str(error)
                logger.error(instance.reason)
                return
            #
            if isinstance(harness, Pytest):
                harness.pytest_run(instance.handler.get_test_timeout())
            else:
                instance.handler.handle(harness)

        sys.stdout.flush()

    def gather_metrics(self, instance: TestInstance):
        build_result = {"returncode": 0}
        if self.options.create_rom_ram_report:
            build_result = self.run_build(['--build', self.build_dir, "--target", "footprint"])
        if self.options.enable_size_report and not self.options.cmake_only:
            self.calc_size(instance=instance, from_buildlog=self.options.footprint_from_buildlog)
        else:
            instance.metrics["used_ram"] = 0
            instance.metrics["used_rom"] = 0
            instance.metrics["available_rom"] = 0
            instance.metrics["available_ram"] = 0
            instance.metrics["unrecognized"] = []
        return build_result

    @staticmethod
    def calc_size(instance: TestInstance, from_buildlog: bool):
        if instance.status not in ["error", "failed", "skipped"]:
            if not instance.platform.type in ["native", "qemu", "unit"]:
                generate_warning = bool(instance.platform.type == "mcu")
                size_calc = instance.calculate_sizes(from_buildlog=from_buildlog, generate_warning=generate_warning)
                instance.metrics["used_ram"] = size_calc.get_used_ram()
                instance.metrics["used_rom"] = size_calc.get_used_rom()
                instance.metrics["available_rom"] = size_calc.get_available_rom()
                instance.metrics["available_ram"] = size_calc.get_available_ram()
                instance.metrics["unrecognized"] = size_calc.unrecognized_sections()
            else:
                instance.metrics["used_ram"] = 0
                instance.metrics["used_rom"] = 0
                instance.metrics["available_rom"] = 0
                instance.metrics["available_ram"] = 0
                instance.metrics["unrecognized"] = []
            instance.metrics["handler_time"] = instance.execution_time

class TwisterRunner:

    def __init__(self, instances, suites, env=None) -> None:
        self.pipeline = None
        self.options = env.options
        self.env = env
        self.instances = instances
        self.suites = suites
        self.duts = None
        self.jobs = 1
        self.results = None
        self.jobserver = None

    def run(self):

        retries = self.options.retry_failed + 1

        BaseManager.register('LifoQueue', queue.LifoQueue)
        manager = BaseManager()
        manager.start()

        self.results = ExecutionCounter(total=len(self.instances))
        self.iteration = 0
        pipeline = manager.LifoQueue()
        done_queue = manager.LifoQueue()

        # Set number of jobs
        if self.options.jobs:
            self.jobs = self.options.jobs
        elif self.options.build_only:
            self.jobs = multiprocessing.cpu_count() * 2
        else:
            self.jobs = multiprocessing.cpu_count()

        if sys.platform == "linux":
            if os.name == 'posix':
                self.jobserver = GNUMakeJobClient.from_environ(jobs=self.options.jobs)
                if not self.jobserver:
                    self.jobserver = GNUMakeJobServer(self.jobs)
                elif self.jobserver.jobs:
                    self.jobs = self.jobserver.jobs
            # TODO: Implement this on windows/mac also
            else:
                self.jobserver = JobClient()

            logger.info("JOBS: %d", self.jobs)

        self.update_counting_before_pipeline()

        while True:
            self.results.iteration += 1

            if self.results.iteration > 1:
                logger.info("%d Iteration:" % (self.results.iteration))
                time.sleep(self.options.retry_interval)  # waiting for the system to settle down
                self.results.done = self.results.total - self.results.failed - self.results.error
                self.results.failed = 0
                if self.options.retry_build_errors:
                    self.results.error = 0
            else:
                self.results.done = self.results.skipped_filter

            self.execute(pipeline, done_queue)

            while True:
                try:
                    inst = done_queue.get_nowait()
                except queue.Empty:
                    break
                else:
                    inst.metrics.update(self.instances[inst.name].metrics)
                    inst.metrics["handler_time"] = inst.execution_time
                    inst.metrics["unrecognized"] = []
                    self.instances[inst.name] = inst

            print("")

            retry_errors = False
            if self.results.error and self.options.retry_build_errors:
                retry_errors = True

            retries = retries - 1
            if retries == 0 or ( self.results.failed == 0 and not retry_errors):
                break

        self.show_brief()

    def update_counting_before_pipeline(self):
        '''
        Updating counting before pipeline is necessary because statically filterd
        test instance never enter the pipeline. While some pipeline output needs
        the static filter stats. So need to prepare them before pipline starts.
        '''
        for instance in self.instances.values():
            if instance.status == 'filtered' and not instance.reason == 'runtime filter':
                self.results.skipped_filter += 1
                self.results.skipped_configs += 1
                self.results.skipped_cases += len(instance.testsuite.testcases)
                self.results.cases += len(instance.testsuite.testcases)
            elif instance.status == 'error':
                self.results.error += 1

    def show_brief(self):
        logger.info("%d test scenarios (%d test instances) selected, "
                    "%d configurations skipped (%d by static filter, %d at runtime)." %
                    (len(self.suites), len(self.instances),
                    self.results.skipped_configs,
                    self.results.skipped_filter,
                    self.results.skipped_configs - self.results.skipped_filter))

    def add_tasks_to_queue(self, pipeline, build_only=False, test_only=False, retry_build_errors=False):
        for instance in self.instances.values():
            if build_only:
                instance.run = False

            no_retry_statuses = ['passed', 'skipped', 'filtered']
            if not retry_build_errors:
                no_retry_statuses.append("error")

            if instance.status not in no_retry_statuses:
                logger.debug(f"adding {instance.name}")
                if instance.status:
                    instance.retries += 1
                instance.status = None

                # Check if cmake package_helper script can be run in advance.
                instance.filter_stages = []
                if instance.testsuite.filter:
                    instance.filter_stages = self.get_cmake_filter_stages(instance.testsuite.filter, expr_parser.reserved.keys())

                if test_only and instance.run:
                    pipeline.put({"op": "run", "test": instance})
                elif instance.filter_stages and "full" not in instance.filter_stages:
                    pipeline.put({"op": "filter", "test": instance})
                else:
                    cache_file = os.path.join(instance.build_dir, "CMakeCache.txt")
                    if os.path.exists(cache_file) and self.env.options.aggressive_no_clean:
                        pipeline.put({"op": "build", "test": instance})
                    else:
                        pipeline.put({"op": "cmake", "test": instance})


    def pipeline_mgr(self, pipeline, done_queue, lock, results):
        if sys.platform == 'linux':
            with self.jobserver.get_job():
                while True:
                    try:
                        task = pipeline.get_nowait()
                    except queue.Empty:
                        break
                    else:
                        instance = task['test']
                        pb = ProjectBuilder(instance, self.env, self.jobserver)
                        pb.duts = self.duts
                        pb.process(pipeline, done_queue, task, lock, results)

                return True
        else:
            while True:
                try:
                    task = pipeline.get_nowait()
                except queue.Empty:
                    break
                else:
                    instance = task['test']
                    pb = ProjectBuilder(instance, self.env, self.jobserver)
                    pb.duts = self.duts
                    pb.process(pipeline, done_queue, task, lock, results)
            return True

    def execute(self, pipeline, done):
        lock = Lock()
        logger.info("Adding tasks to the queue...")
        self.add_tasks_to_queue(pipeline, self.options.build_only, self.options.test_only,
                                retry_build_errors=self.options.retry_build_errors)
        logger.info("Added initial list of jobs to queue")

        processes = []

        for _ in range(self.jobs):
            p = Process(target=self.pipeline_mgr, args=(pipeline, done, lock, self.results, ))
            processes.append(p)
            p.start()
        logger.debug(f"Launched {self.jobs} jobs")

        try:
            for p in processes:
                p.join()
        except KeyboardInterrupt:
            logger.info("Execution interrupted")
            for p in processes:
                p.terminate()

    @staticmethod
    def get_cmake_filter_stages(filt, logic_keys):
        """ Analyze filter expressions from test yaml and decide if dts and/or kconfig based filtering will be needed."""
        dts_required = False
        kconfig_required = False
        full_required = False
        filter_stages = []

        # Compress args in expressions like "function('x', 'y')" so they are not split when splitting by whitespaces
        filt = filt.replace(", ", ",")
        # Remove logic words
        for k in logic_keys:
            filt = filt.replace(f"{k} ", "")
        # Remove brackets
        filt = filt.replace("(", "")
        filt = filt.replace(")", "")
        # Splite by whitespaces
        filt = filt.split()
        for expression in filt:
            if expression.startswith("dt_"):
                dts_required = True
            elif expression.startswith("CONFIG"):
                kconfig_required = True
            else:
                full_required = True

        if full_required:
            return ["full"]
        if dts_required:
            filter_stages.append("dts")
        if kconfig_required:
            filter_stages.append("kconfig")

        return filter_stages