#!/usr/bin/env python3
# Copyright (c) 2023 Google LLC
#
# SPDX-License-Identifier: Apache-2.0
"""
Tests for runner.py classes
"""

import errno
import mock
import os
import pathlib
import pytest
import queue
import re
import subprocess
import sys
import yaml

from contextlib import nullcontext
from elftools.elf.sections import SymbolTableSection
from typing import List

ZEPHYR_BASE = os.getenv("ZEPHYR_BASE")
sys.path.insert(0, os.path.join(ZEPHYR_BASE, "scripts/pylib/twister"))

from twisterlib.error import BuildError
from twisterlib.harness import Pytest

from twisterlib.runner import (
    CMake,
    ExecutionCounter,
    FilterBuilder,
    ProjectBuilder,
    TwisterRunner
)

@pytest.fixture
def mocked_instance(tmp_path):
    instance = mock.Mock()
    testsuite = mock.Mock()
    testsuite.source_dir: str = ''
    instance.testsuite = testsuite
    platform = mock.Mock()
    platform.binaries: List[str] = []
    instance.platform = platform
    build_dir = tmp_path / 'build_dir'
    os.makedirs(build_dir)
    instance.build_dir: str = str(build_dir)
    return instance


@pytest.fixture
def mocked_env():
    env = mock.Mock()
    options = mock.Mock()
    env.options = options
    return env


@pytest.fixture
def mocked_jobserver():
    jobserver = mock.Mock()
    return jobserver


@pytest.fixture
def project_builder(mocked_instance, mocked_env, mocked_jobserver) -> ProjectBuilder:
    project_builder = ProjectBuilder(mocked_instance, mocked_env, mocked_jobserver)
    return project_builder


@pytest.fixture
def runners(project_builder: ProjectBuilder) -> dict:
    """
    Create runners.yaml file in build_dir/zephyr directory and return file
    content as dict.
    """
    build_dir_zephyr_path = os.path.join(project_builder.instance.build_dir, 'zephyr')
    os.makedirs(build_dir_zephyr_path)
    runners_file_path = os.path.join(build_dir_zephyr_path, 'runners.yaml')
    runners_content: dict = {
        'config': {
            'elf_file': 'zephyr.elf',
            'hex_file': os.path.join(build_dir_zephyr_path, 'zephyr.elf'),
            'bin_file': 'zephyr.bin',
        }
    }
    with open(runners_file_path, 'w') as file:
        yaml.dump(runners_content, file)

    return runners_content


@mock.patch("os.path.exists")
def test_projectbuilder_cmake_assemble_args_single(m):
    # Causes the additional_overlay_path to be appended
    m.return_value = True

    class MockHandler:
        pass

    handler = MockHandler()
    handler.args = ["handler_arg1", "handler_arg2"]
    handler.ready = True

    assert(ProjectBuilder.cmake_assemble_args(
        ["basearg1", "CONFIG_t=\"test\"", "SNIPPET_t=\"test\""],
        handler,
        ["a.conf;b.conf", "c.conf"],
        ["extra_overlay.conf"],
        ["x.overlay;y.overlay", "z.overlay"],
        ["cmake1=foo", "cmake2=bar"],
        "/builddir/",
    ) == [
        "-DCONFIG_t=\"test\"",
        "-Dcmake1=foo", "-Dcmake2=bar",
        "-Dbasearg1", "-DSNIPPET_t=test",
        "-Dhandler_arg1", "-Dhandler_arg2",
        "-DCONF_FILE=a.conf;b.conf;c.conf",
        "-DDTC_OVERLAY_FILE=x.overlay;y.overlay;z.overlay",
        "-DOVERLAY_CONFIG=extra_overlay.conf "
        "/builddir/twister/testsuite_extra.conf",
    ])


def test_if_default_binaries_are_taken_properly(project_builder: ProjectBuilder):
    default_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'),
    ]
    binaries = project_builder._get_binaries()
    assert sorted(binaries) == sorted(default_binaries)


def test_if_binaries_from_platform_are_taken_properly(project_builder: ProjectBuilder):
    platform_binaries = ['spi_image.bin']
    project_builder.platform.binaries = platform_binaries
    platform_binaries_expected = [os.path.join('zephyr', bin) for bin in platform_binaries]
    binaries = project_builder._get_binaries()
    assert sorted(binaries) == sorted(platform_binaries_expected)


def test_if_binaries_from_runners_are_taken_properly(runners, project_builder: ProjectBuilder):
    runners_binaries = list(runners['config'].values())
    runners_binaries_expected = [bin if os.path.isabs(bin) else os.path.join('zephyr', bin) for bin in runners_binaries]
    binaries = project_builder._get_binaries_from_runners()
    assert sorted(binaries) == sorted(runners_binaries_expected)


def test_if_runners_file_is_sanitized_properly(runners, project_builder: ProjectBuilder):
    runners_file_path = os.path.join(project_builder.instance.build_dir, 'zephyr', 'runners.yaml')
    with open(runners_file_path, 'r') as file:
        unsanitized_runners_content = yaml.safe_load(file)
    unsanitized_runners_binaries = list(unsanitized_runners_content['config'].values())
    abs_paths = [bin for bin in unsanitized_runners_binaries if os.path.isabs(bin)]
    assert len(abs_paths) > 0

    project_builder._sanitize_runners_file()

    with open(runners_file_path, 'r') as file:
        sanitized_runners_content = yaml.safe_load(file)
    sanitized_runners_binaries = list(sanitized_runners_content['config'].values())
    abs_paths = [bin for bin in sanitized_runners_binaries if os.path.isabs(bin)]
    assert len(abs_paths) == 0


def test_if_zephyr_base_is_sanitized_properly(project_builder: ProjectBuilder):
    sanitized_path_expected = os.path.join('sanitized', 'path')
    path_to_sanitize = os.path.join(os.path.realpath(ZEPHYR_BASE), sanitized_path_expected)
    cmakecache_file_path = os.path.join(project_builder.instance.build_dir, 'CMakeCache.txt')
    with open(cmakecache_file_path, 'w') as file:
        file.write(path_to_sanitize)

    project_builder._sanitize_zephyr_base_from_files()

    with open(cmakecache_file_path, 'r') as file:
        sanitized_path = file.read()
    assert sanitized_path == sanitized_path_expected


def test_executioncounter(capfd):
    ec = ExecutionCounter(total=12)

    ec.cases = 25
    ec.skipped_cases = 6
    ec.error = 2
    ec.iteration = 2
    ec.done = 9
    ec.passed = 6
    ec.skipped_configs = 3
    ec.skipped_runtime = 1
    ec.skipped_filter = 2
    ec.failed = 1

    ec.summary()

    out, err = capfd.readouterr()
    sys.stdout.write(out)
    sys.stderr.write(err)

    assert (
        f'--------------------------------\n'
        f'Total test suites: 12\n'
        f'Total test cases: 25\n'
        f'Executed test cases: 19\n'
        f'Skipped test cases: 6\n'
        f'Completed test suites: 9\n'
        f'Passing test suites: 6\n'
        f'Failing test suites: 1\n'
        f'Skipped test suites: 3\n'
        f'Skipped test suites (runtime): 1\n'
        f'Skipped test suites (filter): 2\n'
        f'Errors: 2\n'
        f'--------------------------------'
    ) in out

    assert ec.cases == 25
    assert ec.skipped_cases == 6
    assert ec.error == 2
    assert ec.iteration == 2
    assert ec.done == 9
    assert ec.passed == 6
    assert ec.skipped_configs == 3
    assert ec.skipped_runtime == 1
    assert ec.skipped_filter == 2
    assert ec.failed == 1


def test_cmake_parse_generated(mocked_jobserver):
    testsuite_mock = mock.Mock()
    platform_mock = mock.Mock()
    source_dir = os.path.join('source', 'dir')
    build_dir = os.path.join('build', 'dir')

    cmake = CMake(testsuite_mock, platform_mock, source_dir, build_dir,
                  mocked_jobserver)

    result = cmake.parse_generated()

    assert cmake.defconfig == {}
    assert result == {}


TESTDATA_1_1 = [
    ('linux'),
    ('nt')
]
TESTDATA_1_2 = [
    (0, False, 'dummy out',
     True, True, 'passed', None, False, True),
    (0, True, '',
     False, False, 'passed', None, False, False),
    (1, True, 'ERROR: region `FLASH\' overflowed by 123 MB',
     True,  True, 'skipped', 'FLASH overflow', True, False),
    (1, True, 'Error: Image size (99 B) + trailer (1 B) exceeds requested size',
     True, True, 'skipped', 'imgtool overflow', True, False),
    (1, True, 'mock.ANY',
     True, True, 'error', 'Build failure', False, False)
]

@pytest.mark.parametrize(
    'return_code, is_instance_run, p_out, expect_returncode,' \
    ' expect_writes, expected_status, expected_reason,' \
    ' expected_change_skip, expected_add_missing',
    TESTDATA_1_2,
    ids=['no error, no instance run', 'no error, instance run',
         'error - region overflow', 'error - image size exceed', 'error']
)
@pytest.mark.parametrize('sys_platform', TESTDATA_1_1)
def test_cmake_run_build(
    sys_platform,
    return_code,
    is_instance_run,
    p_out,
    expect_returncode,
    expect_writes,
    expected_status,
    expected_reason,
    expected_change_skip,
    expected_add_missing
):
    process_mock = mock.Mock(
        returncode=return_code,
        communicate=mock.Mock(
            return_value=(p_out.encode(sys.getdefaultencoding()), None)
        )
    )

    def mock_popen(*args, **kwargs):
        return process_mock

    testsuite_mock = mock.Mock()
    platform_mock = mock.Mock()
    platform_mock.name = '<platform name>'
    source_dir = os.path.join('source', 'dir')
    build_dir = os.path.join('build', 'dir')
    jobserver_mock = mock.Mock(
        popen=mock.Mock(side_effect=mock_popen)
    )
    instance_mock = mock.Mock(add_missing_case_status=mock.Mock())
    instance_mock.build_time = 0
    instance_mock.run = is_instance_run
    instance_mock.status = None
    instance_mock.reason = None

    cmake = CMake(testsuite_mock, platform_mock, source_dir, build_dir,
                  jobserver_mock)
    cmake.cwd = os.path.join('dummy', 'working', 'dir')
    cmake.instance = instance_mock
    cmake.options = mock.Mock()
    cmake.options.overflow_as_errors = False

    cmake_path = os.path.join('dummy', 'cmake')

    popen_mock = mock.Mock(side_effect=mock_popen)
    change_mock = mock.Mock()

    with mock.patch('sys.platform', sys_platform), \
         mock.patch('shutil.which', return_value=cmake_path), \
         mock.patch('twisterlib.runner.change_skip_to_error_if_integration',
                    change_mock), \
         mock.patch('builtins.open', mock.mock_open()), \
         mock.patch('subprocess.Popen', popen_mock):
        result = cmake.run_build(args=['arg1', 'arg2'])

    expected_results = {}
    if expect_returncode:
        expected_results['returncode'] = return_code
    if expected_results == {}:
        expected_results = None

    assert expected_results == result

    popen_caller = cmake.jobserver.popen if sys_platform == 'linux' else \
                   popen_mock
    popen_caller.assert_called_once_with(
        [os.path.join('dummy', 'cmake'), 'arg1', 'arg2'],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        cwd=os.path.join('dummy', 'working', 'dir')
    )

    assert cmake.instance.status == expected_status
    assert cmake.instance.reason == expected_reason

    if expected_change_skip:
        change_mock.assert_called_once()

    if expected_add_missing:
        cmake.instance.add_missing_case_status.assert_called_once_with(
            'skipped', 'Test was built only'
        )


TESTDATA_2_1 = [
    ('linux'),
    ('nt')
]
TESTDATA_2_2 = [
    (True, ['dummy_stage_1', 'ds2'],
     0, False, '',
     True, True, False,
     None, None,
     [os.path.join('dummy', 'cmake'),
      '-B' + os.path.join('build', 'dir'), '-DTC_RUNID=1',
      '-DCONFIG_COMPILER_WARNINGS_AS_ERRORS=y',
      '-DEXTRA_GEN_DEFINES_ARGS=--edtlib-Werror', '-Gdummy_generator',
      '-S' + os.path.join('source', 'dir'),
      'arg1', 'arg2',
      '-DBOARD=<platform name>',
      '-DSNIPPET=dummy snippet 1;ds2',
      '-DMODULES=dummy_stage_1,ds2',
      '-Pzephyr_base/cmake/package_helper.cmake']),
    (False, [],
     1, True, 'ERROR: region `FLASH\' overflowed by 123 MB',
     True, False, True,
     'error', 'Cmake build failure',
     [os.path.join('dummy', 'cmake'),
      '-B' + os.path.join('build', 'dir'), '-DTC_RUNID=1',
      '-DCONFIG_COMPILER_WARNINGS_AS_ERRORS=n',
      '-DEXTRA_GEN_DEFINES_ARGS=', '-Gdummy_generator',
      '-Szephyr_base/share/sysbuild',
      '-DAPP_DIR=' + os.path.join('source', 'dir'),
      'arg1', 'arg2',
      '-DBOARD=<platform name>',
      '-DSNIPPET=dummy snippet 1;ds2']),
]

@pytest.mark.parametrize(
    'error_warns, f_stages,' \
    ' return_code, is_instance_run, p_out, expect_returncode,' \
    ' expect_filter, expect_writes, expected_status, expected_reason,' \
    ' expected_cmd',
    TESTDATA_2_2,
    ids=['filter_stages with success', 'no stages with error']
)
@pytest.mark.parametrize('sys_platform', TESTDATA_2_1)
def test_cmake_run_cmake(
    sys_platform,
    error_warns,
    f_stages,
    return_code,
    is_instance_run,
    p_out,
    expect_returncode,
    expect_filter,
    expect_writes,
    expected_status,
    expected_reason,
    expected_cmd
):
    process_mock = mock.Mock(
        returncode=return_code,
        communicate=mock.Mock(
            return_value=(p_out.encode(sys.getdefaultencoding()), None)
        )
    )

    def mock_popen(*args, **kwargs):
        return process_mock

    testsuite_mock = mock.Mock()
    testsuite_mock.sysbuild = True
    platform_mock = mock.Mock()
    platform_mock.name = '<platform name>'
    source_dir = os.path.join('source', 'dir')
    build_dir = os.path.join('build', 'dir')
    jobserver_mock = mock.Mock(
        popen=mock.Mock(side_effect=mock_popen)
    )
    instance_mock = mock.Mock(add_missing_case_status=mock.Mock())
    instance_mock.run = is_instance_run
    instance_mock.run_id = 1
    instance_mock.build_time = 0
    instance_mock.status = None
    instance_mock.reason = None
    instance_mock.testsuite = mock.Mock()
    instance_mock.testsuite.required_snippets = ['dummy snippet 1', 'ds2']
    instance_mock.testcases = [mock.Mock(), mock.Mock()]
    instance_mock.testcases[0].status = None
    instance_mock.testcases[1].status = None

    cmake = CMake(testsuite_mock, platform_mock, source_dir, build_dir,
                  jobserver_mock)
    cmake.cwd = os.path.join('dummy', 'working', 'dir')
    cmake.instance = instance_mock
    cmake.options = mock.Mock()
    cmake.options.disable_warnings_as_errors = not error_warns
    cmake.options.overflow_as_errors = False
    cmake.env = mock.Mock()
    cmake.env.generator = 'dummy_generator'

    cmake_path = os.path.join('dummy', 'cmake')

    popen_mock = mock.Mock(side_effect=mock_popen)
    change_mock = mock.Mock()

    with mock.patch('sys.platform', sys_platform), \
         mock.patch('shutil.which', return_value=cmake_path), \
         mock.patch('twisterlib.runner.change_skip_to_error_if_integration',
                    change_mock), \
         mock.patch('twisterlib.runner.canonical_zephyr_base',
                    'zephyr_base'), \
         mock.patch('builtins.open', mock.mock_open()), \
         mock.patch('subprocess.Popen', popen_mock):
        result = cmake.run_cmake(args=['arg1', 'arg2'], filter_stages=f_stages)

    expected_results = {}
    if expect_returncode:
        expected_results['returncode'] = return_code
    if expect_filter:
        expected_results['filter'] = {}
    if expected_results == {}:
        expected_results = None

    assert expected_results == result

    popen_caller = cmake.jobserver.popen if sys_platform == 'linux' else \
                   popen_mock
    popen_caller.assert_called_once_with(
        expected_cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        cwd=os.path.join('dummy', 'working', 'dir')
    )

    assert cmake.instance.status == expected_status
    assert cmake.instance.reason == expected_reason

    for tc in cmake.instance.testcases:
        assert tc.status == cmake.instance.status


TESTDATA_3 = [
    ('unit_testing', [], False, True, None, True, None, True,
     None, None, {}, {}, None, None, [], {}),
    (
        'other', [], True,
        True, ['dummy', 'west', 'options'], True,
        None, True,
        os.path.join('domain', 'build', 'dir', 'zephyr', '.config'),
        os.path.join('domain', 'build', 'dir', 'zephyr', 'edt.pickle'),
        {'CONFIG_FOO': 'no'},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'CONFIG_FOO': 'no', 'dummy cache elem': 1},
        b'dummy edt pickle contents',
        [f'Loaded sysbuild domain data from' \
         f' {os.path.join("build", "dir", "domains.yaml")}'],
        {os.path.join('other', 'dummy.testsuite.name'): True}
    ),
    (
        'other', ['kconfig'], True,
        True, ['dummy', 'west', 'options'], True,
        'Dummy parse results', True,
        os.path.join('build', 'dir', 'zephyr', '.config'),
        os.path.join('build', 'dir', 'zephyr', 'edt.pickle'),
        {'CONFIG_FOO': 'no'},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'CONFIG_FOO': 'no', 'dummy cache elem': 1},
        b'dummy edt pickle contents',
        [],
        {os.path.join('other', 'dummy.testsuite.name'): False}
    ),
    (
        'other', ['other'], False,
        False, None, True,
        'Dummy parse results', True,
        None,
        os.path.join('build', 'dir', 'zephyr', 'edt.pickle'),
        {},
        {},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True},
        b'dummy edt pickle contents',
        [],
        {os.path.join('other', 'dummy.testsuite.name'): False}
    ),
    (
        'other', ['other'], True,
        False, None, True,
        'Dummy parse results', True,
        None,
        None,
        {},
        {},
        {},
        None,
        ['Sysbuild test will be skipped. West must be used for flashing.'],
        {os.path.join('other', 'dummy.testsuite.name'): True}
    ),
    (
        'other', ['other'], True,
        False, ['--erase'], True,
        'Dummy parse results', True,
        None,
        None,
        {},
        {},
        None,
        b'dummy edt pickle contents',
        ['Sysbuild test will be skipped,' \
         ' --erase is not supported with --west-flash'],
        {os.path.join('other', 'dummy.testsuite.name'): True}
    ),
    (
        'other', ['other'], False,
        True, None, False,
        'Dummy parse results', True,
        None,
        None,
        {},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'dummy cache elem': 1},
        None,
        [],
        {os.path.join('other', 'dummy.testsuite.name'): False}
    ),
    (
        'other', ['other'], False,
        True, None, True,
        'Dummy parse results', True,
        None,
        os.path.join('build', 'dir', 'zephyr', 'edt.pickle'),
        {},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'dummy cache elem': 1},
        b'dummy edt pickle contents',
        [],
        {os.path.join('other', 'dummy.testsuite.name'): False}
    ),
    (
        'other', ['other'], False,
        True, None, True,
        None, True,
        None,
        os.path.join('build', 'dir', 'zephyr', 'edt.pickle'),
        {},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'dummy cache elem': 1},
        b'dummy edt pickle contents',
        [],
        {os.path.join('other', 'dummy.testsuite.name'): True}
    ),
    (
        'other', ['other'], False,
        True, None, True,
        'Dummy parse results', False,
        None,
        os.path.join('build', 'dir', 'zephyr', 'edt.pickle'),
        {},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'dummy cache elem': 1},
        b'dummy edt pickle contents',
        [],
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'dummy cache elem': 1}
    ),
    (
        'other', ['other'], False,
        True, None, True,
        SyntaxError, True,
        None,
        os.path.join('build', 'dir', 'zephyr', 'edt.pickle'),
        {},
        {'dummy cache elem': 1},
        {'ARCH': 'dummy arch', 'PLATFORM': 'other', 'env_dummy': True,
         'dummy cache elem': 1},
        b'dummy edt pickle contents',
        ['Failed processing testsuite.yaml'],
        SyntaxError
    ),
]

@pytest.mark.parametrize(
    'platform_name, filter_stages, sysbuild,' \
    ' do_find_cache, west_flash_options, edt_exists,' \
    ' parse_results, testsuite_filter,' \
    ' expected_defconfig_path, expected_edt_pickle_path,' \
    ' expected_defconfig, expected_cmakecache, expected_filter_data,' \
    ' expected_edt,' \
    ' expected_logs, expected_return',
    TESTDATA_3,
    ids=['unit testing', 'domain', 'kconfig', 'no cache',
         'no west options', 'erase west flash option', 'no edt',
         'parse result', 'no parse result', 'no testsuite filter', 'parse err']
)
def test_filterbuilder_parse_generated(
    caplog,
    mocked_jobserver,
    platform_name,
    filter_stages,
    sysbuild,
    do_find_cache,
    west_flash_options,
    edt_exists,
    parse_results,
    testsuite_filter,
    expected_defconfig_path,
    expected_edt_pickle_path,
    expected_defconfig,
    expected_cmakecache,
    expected_filter_data,
    expected_edt,
    expected_logs,
    expected_return
):
    def mock_domains_from_file(*args, **kwargs):
        dom = mock.Mock()
        dom.build_dir = os.path.join('domain', 'build', 'dir')
        res = mock.Mock(get_default_domain=mock.Mock(return_value=dom))
        return res

    def mock_cmakecache_from_file(*args, **kwargs):
        if not do_find_cache:
            raise FileNotFoundError(errno.ENOENT, 'Cache not found')
        cache_elem = mock.Mock()
        cache_elem.name = 'dummy cache elem'
        cache_elem.value = 1
        cache = [cache_elem]
        return cache

    def mock_open(filepath, type, *args, **kwargs):
        if filepath == expected_defconfig_path:
            rd = 'I am not a proper line\n' \
                 'CONFIG_FOO="no"'
        elif filepath == expected_edt_pickle_path:
            rd = b'dummy edt pickle contents'
        else:
            raise FileNotFoundError(errno.ENOENT,
                                    f'File {filepath} not mocked.')
        return mock.mock_open(read_data=rd)()

    def mock_parser(filter, filter_data, edt):
        assert filter_data == expected_filter_data
        if isinstance(parse_results, type) and \
           issubclass(parse_results, Exception):
            raise parse_results
        return parse_results

    def mock_pickle(datafile):
        assert datafile.read() == expected_edt
        return mock.Mock()

    testsuite_mock = mock.Mock()
    testsuite_mock.sysbuild = 'sysbuild' if sysbuild else None
    testsuite_mock.name = 'dummy.testsuite.name'
    testsuite_mock.filter = testsuite_filter
    platform_mock = mock.Mock()
    platform_mock.name = platform_name
    platform_mock.arch = 'dummy arch'
    source_dir = os.path.join('source', 'dir')
    build_dir = os.path.join('build', 'dir')

    fb = FilterBuilder(testsuite_mock, platform_mock, source_dir, build_dir,
                       mocked_jobserver)
    instance_mock = mock.Mock()
    fb.instance = instance_mock
    fb.env = mock.Mock()
    fb.env.options = mock.Mock()
    fb.env.options.west_flash = west_flash_options
    fb.env.options.device_testing = True

    environ_mock = {'env_dummy': True}

    with mock.patch('twisterlib.runner.Domains.from_file',
                    mock_domains_from_file), \
         mock.patch('twisterlib.runner.CMakeCache.from_file',
                    mock_cmakecache_from_file), \
         mock.patch('builtins.open', mock_open), \
         mock.patch('expr_parser.parse', mock_parser), \
         mock.patch('pickle.load', mock_pickle), \
         mock.patch('os.path.exists', return_value=edt_exists), \
         mock.patch('os.environ', environ_mock), \
         pytest.raises(expected_return) if \
             isinstance(parse_results, type) and \
             issubclass(parse_results, Exception) else nullcontext() as err:
        result = fb.parse_generated(filter_stages)

    if err:
        assert True
        return

    assert all([log in caplog.text for log in expected_logs])

    assert fb.defconfig == expected_defconfig

    assert fb.cmake_cache == expected_cmakecache

    assert result == expected_return


TESTDATA_4 = [
    (False, False, [f"see: {os.path.join('dummy', 'path', 'dummy_file.log')}"]),
    (True, False, [os.path.join('dummy', 'path', 'dummy_file.log'),
                    'file contents',
                    os.path.join('dummy', 'path', 'dummy_file.log')]),
    (True, True, [os.path.join('dummy', 'path', 'dummy_file.log'),
                   'Unable to read log data ([Errno 2] ERROR: dummy_file.log)',
                   os.path.join('dummy', 'path', 'dummy_file.log')]),
]

@pytest.mark.parametrize(
    'inline_logs, read_exception, expected_logs',
    TESTDATA_4,
    ids=['basic', 'inline logs', 'inline logs+read_exception']
)
def test_projectbuilder_log_info(
    caplog,
    mocked_jobserver,
    inline_logs,
    read_exception,
    expected_logs
):
    def mock_open(filename, *args, **kwargs):
        if read_exception:
            raise OSError(errno.ENOENT, f'ERROR: {os.path.basename(filename)}')
        return mock.mock_open(read_data='file contents')()

    def mock_realpath(filename, *args, **kwargs):
        return os.path.join('path', filename)

    def mock_abspath(filename, *args, **kwargs):
        return os.path.join('dummy', filename)

    filename = 'dummy_file.log'

    env_mock = mock.Mock()
    instance_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    with mock.patch('builtins.open', mock_open), \
         mock.patch('os.path.realpath', mock_realpath), \
         mock.patch('os.path.abspath', mock_abspath):
        pb.log_info(filename, inline_logs)

    assert all([log in caplog.text for log in expected_logs])


TESTDATA_5 = [
    (True, False, False, "Valgrind error", 0, 0, 'build_dir/valgrind.log'),
    (True, False, False, "Error", 0, 0, 'build_dir/build.log'),
    (False, True, False, None, 1024, 0, 'build_dir/handler.log'),
    (False, True, False, None, 0, 0, 'build_dir/build.log'),
    (False, False, True, None, 0, 1024, 'build_dir/device.log'),
    (False, False, True, None, 0, 0, 'build_dir/build.log'),
    (False, False, False, None, 0, 0, 'build_dir/build.log'),
]

@pytest.mark.parametrize(
    'valgrind_log_exists, handler_log_exists, device_log_exists,' \
    ' instance_reason, handler_log_getsize, device_log_getsize, expected_log',
    TESTDATA_5,
    ids=['valgrind log', 'valgrind log unused',
         'handler log', 'handler log unused',
         'device log', 'device log unused',
         'no logs']
)
def test_projectbuilder_log_info_file(
    caplog,
    mocked_jobserver,
    valgrind_log_exists,
    handler_log_exists,
    device_log_exists,
    instance_reason,
    handler_log_getsize,
    device_log_getsize,
    expected_log
):
    def mock_exists(filename, *args, **kwargs):
        if filename == 'build_dir/handler.log':
            return handler_log_exists
        if filename == 'build_dir/valgrind.log':
            return valgrind_log_exists
        if filename == 'build_dir/device.log':
            return device_log_exists
        return False

    def mock_getsize(filename, *args, **kwargs):
        if filename == 'build_dir/handler.log':
            return handler_log_getsize
        if filename == 'build_dir/device.log':
            return device_log_getsize
        return 0

    env_mock = mock.Mock()
    instance_mock = mock.Mock()
    instance_mock.reason = instance_reason
    instance_mock.build_dir = 'build_dir'

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)

    log_info_mock = mock.Mock()

    with mock.patch('os.path.exists', mock_exists), \
         mock.patch('os.path.getsize', mock_getsize), \
         mock.patch('twisterlib.runner.ProjectBuilder.log_info', log_info_mock):
        pb.log_info_file(None)

    log_info_mock.assert_called_with(expected_log, mock.ANY)


TESTDATA_6 = [
    (
        {'op': 'filter'},
        'failed',
        'Failed',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'report', 'test': mock.ANY},
        'failed',
        'Failed',
        0,
        None
    ),
    (
        {'op': 'filter'},
        'passed',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'filter': { 'dummy instance name': True }},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        ['filtering dummy instance name'],
        {'op': 'report', 'test': mock.ANY},
        'filtered',
        'runtime filter',
        1,
        ('skipped',)
    ),
    (
        {'op': 'filter'},
        'passed',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'filter': { 'another dummy instance name': True }},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'cmake', 'test': mock.ANY},
        'passed',
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cmake'},
        'error',
        'dummy error',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'report', 'test': mock.ANY},
        'error',
        'dummy error',
        0,
        None
    ),
    (
        {'op': 'cmake'},
        None,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        True,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'report', 'test': mock.ANY},
        'passed',
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cmake'},
        'success',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        True,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'report', 'test': mock.ANY},
        'success',
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cmake'},
        'success',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        False,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'filter': {'dummy instance name': True}},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        ['filtering dummy instance name'],
        {'op': 'report', 'test': mock.ANY},
        'filtered',
        'runtime filter',
        1,
        ('skipped',)
    ),
    (
        {'op': 'cmake'},
        'success',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        False,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'filter': {}},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'build', 'test': mock.ANY},
        'success',
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'build'},
        mock.ANY,
        None,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        None,
        mock.ANY,
        mock.ANY,
        ['build test: dummy instance name'],
        {'op': 'report', 'test': mock.ANY},
        'error',
        'Build Failure',
        0,
        None
    ),
    (
        {'op': 'build'},
        'skipped',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'returncode': 0},
        mock.ANY,
        mock.ANY,
        ['build test: dummy instance name',
         'Determine test cases for test instance: dummy instance name'],
        {'op': 'gather_metrics', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        1,
        ('skipped', mock.ANY)
    ),
    (
        {'op': 'build'},
        'passed',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'dummy': 'dummy'},
        mock.ANY,
        mock.ANY,
        ['build test: dummy instance name'],
        {'op': 'report', 'test': mock.ANY},
        'passed',
        mock.ANY,
        0,
        ('blocked', mock.ANY)
    ),
    (
        {'op': 'build'},
        'success',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'returncode': 0},
        mock.ANY,
        mock.ANY,
        ['build test: dummy instance name',
         'Determine test cases for test instance: dummy instance name'],
        {'op': 'gather_metrics', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'build'},
        'success',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        {'returncode': 0},
        mock.ANY,
        BuildError,
        ['build test: dummy instance name',
         'Determine test cases for test instance: dummy instance name'],
        {'op': 'report', 'test': mock.ANY},
        'error',
        'Determine Testcases Error!',
        0,
        None
    ),
    (
        {'op': 'gather_metrics'},
        mock.ANY,
        mock.ANY,
        True,
        True,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'run', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'gather_metrics'},
        mock.ANY,
        mock.ANY,
        False,
        True,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'report', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'run'},
        'success',
        'OK',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        None,
        mock.ANY,
        ['run test: dummy instance name',
         'run status: dummy instance name success'],
        {'op': 'report', 'test': mock.ANY, 'status': 'success', 'reason': 'OK'},
        'success',
        'OK',
        0,
        None
    ),
    (
        {'op': 'run'},
        'failed',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        RuntimeError,
        mock.ANY,
        ['run test: dummy instance name',
         'run status: dummy instance name failed',
         'RuntimeError: Pipeline Error!'],
        None,
        'failed',
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'report'},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        False,
        True,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'cleanup', 'mode': 'device', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'report'},
        'passed',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        False,
        False,
        'pass',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'cleanup', 'mode': 'passed', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'report'},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        False,
        False,
        'all',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        {'op': 'cleanup', 'mode': 'all', 'test': mock.ANY},
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'report'},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        False,
        False,
        'other',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        None,
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cleanup', 'mode': 'device'},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        None,
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cleanup', 'mode': 'passed'},
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        None,
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cleanup', 'mode': 'all'},
        mock.ANY,
        'Valgrind error',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        None,
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
    (
        {'op': 'cleanup', 'mode': 'all'},
        mock.ANY,
        'Cmake build failure',
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        mock.ANY,
        [],
        None,
        mock.ANY,
        mock.ANY,
        0,
        None
    ),
]

@pytest.mark.parametrize(
    'message,' \
    ' instance_status, instance_reason, instance_run, instance_handler_ready,' \
    ' options_cmake_only,' \
    ' options_coverage, options_prep_artifacts, options_runtime_artifacts,' \
    ' cmake_res, build_res,' \
    ' pipeline_runtime_error, determine_testcases_build_error,' \
    ' expected_logs, resulting_message,' \
    ' expected_status, expected_reason, expected_skipped, expected_missing',
    TESTDATA_6,
    ids=[
        'filter, failed', 'filter, cmake res', 'filter, no cmake res',
        'cmake, failed', 'cmake, cmake_only, no status', 'cmake, cmake_only',
        'cmake, no cmake_only, cmake res', 'cmake, no cmake_only, no cmake res',
        'build, no build res', 'build, skipped', 'build, blocked',
        'build, determine testcases', 'build, determine testcases Error',
        'gather metrics, run and ready handler', 'gather metrics',
        'run', 'run, Pipeline Runtime Error',
        'report, prep artifacts for testing',
        'report, runtime artifact cleanup pass, status passed',
        'report, runtime artifact cleanup all', 'report, no message put',
        'cleanup, device', 'cleanup, mode passed', 'cleanup, mode all',
        'cleanup, mode all, cmake build failure'
    ]
)
def test_projectbuilder_process(
    caplog,
    mocked_jobserver,
    message,
    instance_status,
    instance_reason,
    instance_run,
    instance_handler_ready,
    options_cmake_only,
    options_coverage,
    options_prep_artifacts,
    options_runtime_artifacts,
    cmake_res,
    build_res,
    pipeline_runtime_error,
    determine_testcases_build_error,
    expected_logs,
    resulting_message,
    expected_status,
    expected_reason,
    expected_skipped,
    expected_missing
):
    def mock_pipeline_put(msg):
        if isinstance(pipeline_runtime_error, type) and \
           issubclass(pipeline_runtime_error, Exception):
            raise RuntimeError('Pipeline Error!')

    def mock_determine_testcases(res):
        if isinstance(determine_testcases_build_error, type) and \
           issubclass(determine_testcases_build_error, Exception):
            raise BuildError('Determine Testcases Error!')

    instance_mock = mock.Mock()
    instance_mock.name = 'dummy instance name'
    instance_mock.status = instance_status
    instance_mock.reason = instance_reason
    instance_mock.run = instance_run
    instance_mock.handler = mock.Mock()
    instance_mock.handler.ready = instance_handler_ready
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb.options = mock.Mock()
    pb.options.coverage = options_coverage
    pb.options.prep_artifacts_for_testing = options_prep_artifacts
    pb.options.runtime_artifact_cleanup = options_runtime_artifacts
    pb.options.cmake_only = options_cmake_only

    pb.cmake = mock.Mock(return_value=cmake_res)
    pb.build = mock.Mock(return_value=build_res)
    pb.determine_testcases = mock.Mock(side_effect=mock_determine_testcases)

    pb.report_out = mock.Mock()
    pb.cleanup_artifacts = mock.Mock()
    pb.cleanup_device_testing_artifacts = mock.Mock()
    pb.run = mock.Mock()
    pb.gather_metrics = mock.Mock()

    pipeline_mock = mock.Mock(put=mock.Mock(side_effect=mock_pipeline_put))
    done_mock = mock.Mock()
    lock_mock = mock.Mock(
        __enter__=mock.Mock(return_value=(mock.Mock(), mock.Mock())),
        __exit__=mock.Mock(return_value=None)
    )
    results_mock = mock.Mock()
    results_mock.skipped_runtime = 0

    pb.process(pipeline_mock, done_mock, message, lock_mock, results_mock)

    assert all([log in caplog.text for log in expected_logs])

    if resulting_message:
        pipeline_mock.put.assert_called_with(resulting_message)

    assert pb.instance.status == expected_status
    assert pb.instance.reason == expected_reason
    assert results_mock.skipped_runtime == expected_skipped

    if expected_missing:
        pb.instance.add_missing_case_status.assert_called_with(*expected_missing)


TESTDATA_7 = [
    (
        [
            'z_ztest_unit_test__dummy_suite_name__dummy_test_name',
            'z_ztest_unit_test__dummy_suite_name__test_dummy_name',
            'no match'
        ],
        ['dummy_id.dummy_name', 'dummy_id.dummy_name']
    ),
    (
        ['no match'],
        []
    ),
]

@pytest.mark.parametrize(
    'symbols_names, added_tcs',
    TESTDATA_7,
    ids=['two hits, one miss', 'nothing']
)
def test_projectbuilder_determine_testcases(
    mocked_jobserver,
    symbols_names,
    added_tcs
):
    symbols_mock = [mock.Mock(n=name) for name in symbols_names]
    for m in symbols_mock:
        m.configure_mock(name=m.n)

    sections_mock = [mock.Mock(spec=SymbolTableSection)]
    sections_mock[0].iter_symbols = mock.Mock(return_value=symbols_mock)

    elf_mock = mock.Mock()
    elf_mock().iter_sections = mock.Mock(return_value=sections_mock)

    results_mock = mock.Mock()

    instance_mock = mock.Mock()
    instance_mock.testcases = []
    instance_mock.testsuite.id = 'dummy_id'
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)

    with mock.patch('twisterlib.runner.ELFFile', elf_mock), \
         mock.patch('builtins.open', mock.mock_open()):
        pb.determine_testcases(results_mock)

    pb.instance.add_testcase.assert_has_calls(
        [mock.call(name=x) for x in added_tcs]
    )
    pb.instance.testsuite.add_testcase.assert_has_calls(
        [mock.call(name=x) for x in added_tcs]
    )


TESTDATA_8 = [
    (
        ['addition.al'],
        'dummy',
        ['addition.al', '.config', 'zephyr']
    ),
    (
        [],
        'all',
        ['.config', 'zephyr', 'testsuite_extra.conf', 'twister']
    ),
]

@pytest.mark.parametrize(
    'additional_keep, runtime_artifact_cleanup, expected_files',
    TESTDATA_8,
    ids=['additional keep', 'all cleanup']
)
def test_projectbuilder_cleanup_artifacts(
    tmpdir,
    mocked_jobserver,
    additional_keep,
    runtime_artifact_cleanup,
    expected_files
):
    # tmpdir
    # ┣ twister
    # ┃ ┗ testsuite_extra.conf
    # ┣ dummy_dir
    # ┃ ┗ dummy.del
    # ┣ dummy_link_dir -> zephyr
    # ┣ zephyr
    # ┃ ┗ .config
    # ┗ addition.al
    twister_dir = tmpdir.mkdir('twister')
    testsuite_extra_conf = twister_dir.join('testsuite_extra.conf')
    testsuite_extra_conf.write_text('dummy', 'utf-8')

    dummy_dir = tmpdir.mkdir('dummy_dir')
    dummy_del = dummy_dir.join('dummy.del')
    dummy_del.write_text('dummy', 'utf-8')

    zephyr = tmpdir.mkdir('zephyr')
    config = zephyr.join('.config')
    config.write_text('dummy', 'utf-8')

    dummy_link_dir = tmpdir.join('dummy_link_dir')
    os.symlink(zephyr, dummy_link_dir)

    addition_al = tmpdir.join('addition.al')
    addition_al.write_text('dummy', 'utf-8')

    instance_mock = mock.Mock()
    instance_mock.build_dir = tmpdir
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb.options = mock.Mock(runtime_artifact_cleanup=runtime_artifact_cleanup)

    pb.cleanup_artifacts(additional_keep)

    files_left = [p.name for p in list(pathlib.Path(tmpdir).glob('**/*'))]

    assert sorted(files_left) == sorted(expected_files)


def test_projectbuilder_cleanup_device_testing_artifacts(
    caplog,
    mocked_jobserver
):
    bins = [os.path.join('zephyr', 'file.bin')]

    instance_mock = mock.Mock()
    build_dir = os.path.join('build', 'dir')
    instance_mock.build_dir = build_dir
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb._get_binaries = mock.Mock(return_value=bins)
    pb.cleanup_artifacts = mock.Mock()
    pb._sanitize_files = mock.Mock()

    pb.cleanup_device_testing_artifacts()

    assert f'Cleaning up for Device Testing {build_dir}' in caplog.text

    pb.cleanup_artifacts.assert_called_once_with(
        [os.path.join('zephyr', 'file.bin'),
         os.path.join('zephyr', 'runners.yaml')]
    )
    pb._sanitize_files.assert_called_once()


TESTDATA_9 = [
    (
        None,
        [],
        [os.path.join('zephyr', 'zephyr.hex'),
         os.path.join('zephyr', 'zephyr.bin'),
         os.path.join('zephyr', 'zephyr.elf'),
         os.path.join('zephyr', 'zephyr.exe')]
    ),
    (
        [os.path.join('dummy.bin'), os.path.join('dummy.hex')],
        [os.path.join('dir2', 'dummy.elf')],
        [os.path.join('zephyr', 'dummy.bin'),
         os.path.join('zephyr', 'dummy.hex'),
         os.path.join('dir2', 'dummy.elf')]
    ),
]

@pytest.mark.parametrize(
    'platform_binaries, runner_binaries, expected_binaries',
    TESTDATA_9,
    ids=['default', 'valid']
)
def test_projectbuilder_get_binaries(
    mocked_jobserver,
    platform_binaries,
    runner_binaries,
    expected_binaries
):
    instance_mock = mock.Mock()
    instance_mock.platform = mock.Mock()
    instance_mock.platform.binaries = platform_binaries
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb._get_binaries_from_runners = mock.Mock(return_value=runner_binaries)

    bins = pb._get_binaries()

    assert all(bin in expected_binaries for bin in bins)
    assert all(bin in bins for bin in expected_binaries)


TESTDATA_10 = [
    (None, []),
    ({'dummy': 'dummy'}, []),
    (
        {
            'config': {
                'elf_file': '/absolute/path/dummy.elf',
                'bin_file': 'path/dummy.bin'
            }
        },
        ['/absolute/path/dummy.elf', os.path.join('zephyr', 'path/dummy.bin')]
    ),
]

@pytest.mark.parametrize(
    'runners_content, expected_binaries',
    TESTDATA_10,
    ids=['no file', 'no config', 'valid']
)
def test_projectbuilder_get_binaries_from_runners(
    mocked_jobserver,
    runners_content,
    expected_binaries
):
    def mock_exists(fname):
        assert fname == os.path.join('build', 'dir', 'zephyr', 'runners.yaml')
        return runners_content is not None

    instance_mock = mock.Mock()
    instance_mock.build_dir = os.path.join('build', 'dir')
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)

    with mock.patch('os.path.exists', mock_exists), \
         mock.patch('builtins.open', mock.mock_open()), \
         mock.patch('yaml.safe_load', return_value=runners_content):
        bins = pb._get_binaries_from_runners()

    assert all(bin in expected_binaries for bin in bins)
    assert all(bin in bins for bin in expected_binaries)


def test_projectbuilder_sanitize_files(mocked_jobserver):
    instance_mock = mock.Mock()
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb._sanitize_runners_file = mock.Mock()
    pb._sanitize_zephyr_base_from_files = mock.Mock()

    pb._sanitize_files()

    pb._sanitize_runners_file.assert_called_once()
    pb._sanitize_zephyr_base_from_files.assert_called_once()


TESTDATA_11 = [
    (None, None),
    ('dummy: []', None),
    (
"""
config:
  elf_file: relative/path/dummy.elf
  hex_file: /absolute/path/build_dir/zephyr/dummy.hex
""",
"""
config:
  elf_file: relative/path/dummy.elf
  hex_file: dummy.hex
"""
    ),
]

@pytest.mark.parametrize(
    'runners_text, expected_write_text',
    TESTDATA_11,
    ids=['no file', 'no config', 'valid']
)
def test_projectbuilder_sanitize_runners_file(
    mocked_jobserver,
    runners_text,
    expected_write_text
):
    def mock_exists(fname):
        return runners_text is not None

    instance_mock = mock.Mock()
    instance_mock.build_dir = '/absolute/path/build_dir'
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)

    with mock.patch('os.path.exists', mock_exists), \
         mock.patch('builtins.open',
                    mock.mock_open(read_data=runners_text)) as f:
        pb._sanitize_runners_file()

    if expected_write_text is not None:
        f().write.assert_called_with(expected_write_text)
    else:
        f().write.assert_not_called()


TESTDATA_12 = [
    (
        {
            'CMakeCache.txt': mock.mock_open(
                read_data='canonical/zephyr/base/dummy.file: ERROR'
            )
        },
        {
            'CMakeCache.txt': 'dummy.file: ERROR'
        }
    ),
    (
        {
            os.path.join('zephyr', 'runners.yaml'): mock.mock_open(
                read_data='There was canonical/zephyr/base/dummy.file here'
            )
        },
        {
            os.path.join('zephyr', 'runners.yaml'): 'There was dummy.file here'
        }
    ),
]

@pytest.mark.parametrize(
    'text_mocks, expected_write_texts',
    TESTDATA_12,
    ids=['CMakeCache file', 'runners.yaml file']
)
def test_projectbuilder_sanitize_zephyr_base_from_files(
    mocked_jobserver,
    text_mocks,
    expected_write_texts
):
    build_dir_path = 'canonical/zephyr/base/build_dir/'

    def mock_exists(fname):
        if not fname.startswith(build_dir_path):
            return False
        return fname[len(build_dir_path):] in text_mocks

    def mock_open(fname, *args, **kwargs):
        if not fname.startswith(build_dir_path):
            raise FileNotFoundError(errno.ENOENT, f'File {fname} not found.')
        return text_mocks[fname[len(build_dir_path):]]()

    instance_mock = mock.Mock()
    instance_mock.build_dir = build_dir_path
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)

    with mock.patch('os.path.exists', mock_exists), \
         mock.patch('builtins.open', mock_open), \
         mock.patch('twisterlib.runner.canonical_zephyr_base',
                    'canonical/zephyr/base'):
        pb._sanitize_zephyr_base_from_files()

    for fname, fhandler in text_mocks.items():
        fhandler().write.assert_called_with(expected_write_texts[fname])


TESTDATA_13 = [
    (
        'error', True, True, False,
        ['INFO      20/25 dummy platform' \
         '            dummy.testsuite.name' \
         '                                ERROR dummy reason (cmake)'],
        None
    ),
    (
        'failed', False, False, False,
        ['ERROR     dummy platform' \
         '            dummy.testsuite.name' \
         '                                FAILED : dummy reason'],
        'INFO    - Total complete:   20/  25  80%  skipped:    3,' \
        ' failed:    3, error:    1'
    ),
    (
        'skipped', True, False, False,
        ['INFO      20/25 dummy platform' \
         '            dummy.testsuite.name' \
         '                               SKIPPED (dummy reason)'],
        None
    ),
    (
        'filtered', False, False, False,
        [],
        'INFO    - Total complete:   20/  25  80%  skipped:    4,' \
        ' failed:    2, error:    1'
    ),
    (
        'passed', True, False, True,
        ['INFO      20/25 dummy platform' \
         '            dummy.testsuite.name' \
         '                               PASSED' \
         ' (dummy handler type: dummy dut, 60.000s)'],
        None
    ),
    (
        'passed', True, False, False,
        ['INFO      20/25 dummy platform' \
         '            dummy.testsuite.name' \
         '                               PASSED (build)'],
        None
    ),
    (
        'unknown status', False, False, False,
        ['Unknown status = unknown status'],
        'INFO    - Total complete:   20/  25  80%  skipped:    3,' \
        ' failed:    2, error:    1\r'
    ),
    (
        'timeout', True, False, True,
        ['INFO      20/25 dummy platform' \
         '            dummy.testsuite.name' \
         '                               UNKNOWN' \
         ' (dummy handler type: dummy dut, 60.000s/seed: 123)'],
        None
    ),
]

@pytest.mark.parametrize(
    'status, verbose, cmake_only, ready_run, expected_logs, expected_out',
    TESTDATA_13,
    ids=['verbose error cmake only', 'failed', 'verbose skipped', 'filtered',
         'verbose passed ready run', 'verbose passed', 'unknown status',
         'timeout']
)
def test_projectbuilder_report_out(
    capfd,
    caplog,
    mocked_jobserver,
    status,
    verbose,
    cmake_only,
    ready_run,
    expected_logs,
    expected_out
):
    instance_mock = mock.Mock()
    instance_mock.handler.type_str = 'dummy handler type'
    instance_mock.handler.seed = 123
    instance_mock.handler.ready = ready_run
    instance_mock.run = ready_run
    instance_mock.dut = 'dummy dut'
    instance_mock.execution_time = 60
    instance_mock.platform.name = 'dummy platform'
    instance_mock.status = status
    instance_mock.reason = 'dummy reason'
    instance_mock.testsuite.name = 'dummy.testsuite.name'
    instance_mock.testsuite.testcases = [mock.Mock() for _ in range(25)]
    instance_mock.testcases = [mock.Mock() for _ in range(24)] + \
                              [mock.Mock(status='skipped')]
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb.options.verbose = verbose
    pb.options.cmake_only = cmake_only
    pb.options.seed = 123
    pb.log_info_file = mock.Mock()

    results_mock = mock.Mock()
    results_mock.iteration = 1
    results_mock.total = 25
    results_mock.done = 19
    results_mock.passed = 17
    results_mock.skipped_configs = 3
    results_mock.skipped_cases = 4
    results_mock.failed = 2
    results_mock.error = 1
    results_mock.cases = 0

    pb.report_out(results_mock)

    assert results_mock.cases == 25

    trim_actual_log = re.sub(
        r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])',
        '',
        caplog.text
    )
    trim_actual_log = re.sub(r'twister:runner.py:\d+', '', trim_actual_log)
    assert all([log in trim_actual_log for log in expected_logs])

    if expected_out:
        out, err = capfd.readouterr()
        sys.stdout.write(out)
        sys.stderr.write(err)

        # Remove 7b ANSI C1 escape sequences (colours)
        out = re.sub(
            r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])',
            '',
            out
        )

        assert expected_out in out


def test_projectbuilder_cmake_assemble_args():
    extra_args = ['CONFIG_FOO=y', 'DUMMY_EXTRA="yes"']
    handler = mock.Mock(ready=True, args=['dummy_handler'])
    extra_conf_files = ['extrafile1.conf', 'extrafile2.conf']
    extra_overlay_confs = ['extra_overlay_conf']
    extra_dtc_overlay_files = ['overlay1.dtc', 'overlay2.dtc']
    cmake_extra_args = ['CMAKE1="yes"', 'CMAKE2=n']
    build_dir = os.path.join('build', 'dir')

    with mock.patch('os.path.exists', return_value=True):
        results = ProjectBuilder.cmake_assemble_args(extra_args, handler,
                                                     extra_conf_files,
                                                     extra_overlay_confs,
                                                     extra_dtc_overlay_files,
                                                     cmake_extra_args,
                                                     build_dir)

    expected_results = [
        '-DCONFIG_FOO=y',
        '-DCMAKE1=\"yes\"',
        '-DCMAKE2=n',
        '-DDUMMY_EXTRA=yes',
        '-Ddummy_handler',
        '-DCONF_FILE=extrafile1.conf;extrafile2.conf',
        '-DDTC_OVERLAY_FILE=overlay1.dtc;overlay2.dtc',
        f'-DOVERLAY_CONFIG=extra_overlay_conf ' \
        f'{os.path.join("build", "dir", "twister", "testsuite_extra.conf")}'
    ]

    assert results == expected_results


def test_projectbuilder_cmake():
    instance_mock = mock.Mock()
    instance_mock.handler = 'dummy handler'
    instance_mock.build_dir = os.path.join('build', 'dir')
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb.build_dir = 'build_dir'
    pb.testsuite.extra_args = ['some', 'args']
    pb.testsuite.extra_conf_files = ['some', 'files1']
    pb.testsuite.extra_overlay_confs = ['some', 'files2']
    pb.testsuite.extra_dtc_overlay_files = ['some', 'files3']
    pb.options.extra_args = ['other', 'args']
    pb.cmake_assemble_args = mock.Mock(return_value=['dummy'])
    cmake_res_mock = mock.Mock()
    pb.run_cmake = mock.Mock(return_value=cmake_res_mock)

    res = pb.cmake(['dummy filter'])

    assert res == cmake_res_mock
    pb.cmake_assemble_args.assert_called_once_with(
        pb.testsuite.extra_args,
        pb.instance.handler,
        pb.testsuite.extra_conf_files,
        pb.testsuite.extra_overlay_confs,
        pb.testsuite.extra_dtc_overlay_files,
        pb.options.extra_args,
        pb.instance.build_dir
    )
    pb.run_cmake.assert_called_once_with(['dummy'], ['dummy filter'])


def test_projectbuilder_build(mocked_jobserver):
    instance_mock = mock.Mock()
    instance_mock.testsuite.harness = 'test'
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)

    pb.build_dir = 'build_dir'
    pb.run_build = mock.Mock(return_value={'dummy': 'dummy'})

    res = pb.build()

    pb.run_build.assert_called_once_with(['--build', 'build_dir'])
    assert res == {'dummy': 'dummy'}


TESTDATA_14 = [
    (
        True,
        'device',
        234,
        'native_sim',
        'posix',
        {'CONFIG_FAKE_ENTROPY_NATIVE_POSIX': 'y'},
        'pytest',
        True,
        True,
        True,
        True,
        True,
        False
    ),
    (
        True,
        'not device',
        None,
        'native_sim',
        'not posix',
        {'CONFIG_FAKE_ENTROPY_NATIVE_POSIX': 'y'},
        'not pytest',
        False,
        False,
        False,
        False,
        False,
        True
    ),
    (
        False,
        'device',
        234,
        'native_sim',
        'posix',
        {'CONFIG_FAKE_ENTROPY_NATIVE_POSIX': 'y'},
        'pytest',
        False,
        False,
        False,
        False,
        False,
        False
    ),
]

@pytest.mark.parametrize(
    'ready, type_str, seed, platform_name, platform_arch, defconfig, harness,' \
    ' expect_duts, expect_parse_generated, expect_seed,' \
    ' expect_extra_test_args, expect_pytest, expect_handle',
    TESTDATA_14,
    ids=['pytest full', 'not pytest minimal', 'not ready']
)
def test_projectbuilder_run(
    mocked_jobserver,
    ready,
    type_str,
    seed,
    platform_name,
    platform_arch,
    defconfig,
    harness,
    expect_duts,
    expect_parse_generated,
    expect_seed,
    expect_extra_test_args,
    expect_pytest,
    expect_handle
):
    pytest_mock = mock.Mock(spec=Pytest)
    harness_mock = mock.Mock()

    def mock_harness(name):
        if name == 'Pytest':
            return pytest_mock
        else:
            return harness_mock

    instance_mock = mock.Mock()
    instance_mock.handler.get_test_timeout = mock.Mock(return_value=60)
    instance_mock.handler.seed = 123
    instance_mock.handler.ready = ready
    instance_mock.handler.type_str = type_str
    instance_mock.handler.duts = [mock.Mock(name='dummy dut')]
    instance_mock.platform.name = platform_name
    instance_mock.platform.arch = platform_arch
    instance_mock.testsuite.harness = harness
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb.options.extra_test_args = ['dummy_arg1', 'dummy_arg2']
    pb.duts = ['another dut']
    pb.options.seed = seed
    pb.defconfig = defconfig
    pb.parse_generated = mock.Mock()

    with mock.patch('twisterlib.runner.HarnessImporter.get_harness',
                    mock_harness):
        pb.run()

    if expect_duts:
        assert pb.instance.handler.duts == ['another dut']

    if expect_parse_generated:
        pb.parse_generated.assert_called_once()

    if expect_seed:
        assert pb.instance.handler.seed == seed

    if expect_extra_test_args:
        assert pb.instance.handler.extra_test_args == ['dummy_arg1',
                                                       'dummy_arg2']

    if expect_pytest:
        pytest_mock.pytest_run.assert_called_once_with(60)

    if expect_handle:
        pb.instance.handler.handle.assert_called_once_with(harness_mock)


TESTDATA_15 = [
    (False, False, False, True),
    (True, False, True, False),
    (False, True, False, True),
    (True, True, False, True),
]

@pytest.mark.parametrize(
    'enable_size_report, cmake_only, expect_calc_size, expect_zeroes',
    TESTDATA_15,
    ids=['none', 'size_report', 'cmake', 'size_report+cmake']
)
def test_projectbuilder_gather_metrics(
    mocked_jobserver,
    enable_size_report,
    cmake_only,
    expect_calc_size,
    expect_zeroes
):
    instance_mock = mock.Mock()
    instance_mock.metrics = {}
    env_mock = mock.Mock()

    pb = ProjectBuilder(instance_mock, env_mock, mocked_jobserver)
    pb.options.enable_size_report = enable_size_report
    pb.options.create_rom_ram_report = False
    pb.options.cmake_only = cmake_only
    pb.calc_size = mock.Mock()

    pb.gather_metrics(instance_mock)

    if expect_calc_size:
        pb.calc_size.assert_called_once()

    if expect_zeroes:
        assert instance_mock.metrics['used_ram'] == 0
        assert instance_mock.metrics['used_rom'] == 0
        assert instance_mock.metrics['available_rom'] == 0
        assert instance_mock.metrics['available_ram'] == 0
        assert instance_mock.metrics['unrecognized'] == []


TESTDATA_16 = [
    ('error', mock.ANY, False, False, False),
    ('failed', mock.ANY, False, False, False),
    ('skipped', mock.ANY, False, False, False),
    ('filtered', 'native', False, False, True),
    ('passed', 'qemu', False, False, True),
    ('filtered', 'unit', False, False, True),
    ('filtered', 'mcu', True, True, False),
    ('passed', 'frdm_k64f', False, True, False),
]

@pytest.mark.parametrize(
    'status, platform_type, expect_warnings, expect_calcs, expect_zeroes',
    TESTDATA_16,
    ids=[x[0] + (', ' + x[1]) if x[1] != mock.ANY else '' for x in TESTDATA_16]
)
def test_projectbuilder_calc_size(
    status,
    platform_type,
    expect_warnings,
    expect_calcs,
    expect_zeroes
):
    size_calc_mock = mock.Mock()

    instance_mock = mock.Mock()
    instance_mock.status = status
    instance_mock.platform.type = platform_type
    instance_mock.metrics = {}
    instance_mock.calculate_sizes = mock.Mock(return_value=size_calc_mock)

    from_buildlog = True

    ProjectBuilder.calc_size(instance_mock, from_buildlog)

    if expect_calcs:
        instance_mock.calculate_sizes.assert_called_once_with(
            from_buildlog=from_buildlog,
            generate_warning=expect_warnings
        )

        assert instance_mock.metrics['used_ram'] == \
               size_calc_mock.get_used_ram()
        assert instance_mock.metrics['used_rom'] == \
               size_calc_mock.get_used_rom()
        assert instance_mock.metrics['available_rom'] == \
               size_calc_mock.get_available_rom()
        assert instance_mock.metrics['available_ram'] == \
               size_calc_mock.get_available_ram()
        assert instance_mock.metrics['unrecognized'] == \
               size_calc_mock.unrecognized_sections()

    if expect_zeroes:
        assert instance_mock.metrics['used_ram'] == 0
        assert instance_mock.metrics['used_rom'] == 0
        assert instance_mock.metrics['available_rom'] == 0
        assert instance_mock.metrics['available_ram'] == 0
        assert instance_mock.metrics['unrecognized'] == []

    if expect_calcs or expect_zeroes:
        assert instance_mock.metrics['handler_time'] == \
               instance_mock.execution_time
    else:
        assert instance_mock.metrics == {}


TESTDATA_17 = [
    ('linux', 'posix', {'jobs': 4}, True, 32, 'GNUMakeJobClient'),
    ('linux', 'posix', {'build_only': True}, False, 16, 'GNUMakeJobServer'),
    ('linux', '???', {}, False, 8, 'JobClient'),
    ('linux', '???', {'jobs': 4}, False, 4, 'JobClient'),
]

@pytest.mark.parametrize(
    'platform, os_name, options, jobclient_from_environ, expected_jobs,' \
    ' expected_jobserver',
    TESTDATA_17,
    ids=['GNUMakeJobClient', 'GNUMakeJobServer',
         'JobClient', 'Jobclient+options']
)
def test_twisterrunner_run(
    caplog,
    platform,
    os_name,
    options,
    jobclient_from_environ,
    expected_jobs,
    expected_jobserver
):
    def mock_client_from_environ(jobs):
        if jobclient_from_environ:
            jobclient_mock = mock.Mock(jobs=32)
            jobclient_mock.name = 'GNUMakeJobClient'
            return jobclient_mock
        return None

    instances = {'dummy instance': mock.Mock(metrics={'k': 'v'})}
    suites = [mock.Mock()]
    env_mock = mock.Mock()

    tr = TwisterRunner(instances, suites, env=env_mock)
    tr.options.retry_failed = 2
    tr.options.retry_interval = 10
    tr.options.retry_build_errors = True
    tr.options.jobs = None
    tr.options.build_only = None
    for k, v in options.items():
        setattr(tr.options, k, v)
    tr.update_counting_before_pipeline = mock.Mock()
    tr.execute = mock.Mock()
    tr.show_brief = mock.Mock()

    gnumakejobserver_mock = mock.Mock()
    gnumakejobserver_mock().name='GNUMakeJobServer'
    jobclient_mock = mock.Mock()
    jobclient_mock().name='JobClient'

    pipeline_q = queue.LifoQueue()
    done_q = queue.LifoQueue()
    done_instance = mock.Mock(
        metrics={'k2': 'v2'},
        execution_time=30
    )
    done_instance.name='dummy instance'
    done_q.put(done_instance)
    manager_mock = mock.Mock()
    manager_mock().LifoQueue = mock.Mock(
        side_effect=iter([pipeline_q, done_q])
    )

    results_mock = mock.Mock()
    results_mock().error = 1
    results_mock().iteration = 0
    results_mock().failed = 2
    results_mock().total = 9

    with mock.patch('twisterlib.runner.ExecutionCounter', results_mock), \
         mock.patch('twisterlib.runner.BaseManager', manager_mock), \
         mock.patch('twisterlib.runner.GNUMakeJobClient.from_environ',
                    mock_client_from_environ), \
         mock.patch('twisterlib.runner.GNUMakeJobServer',
                    gnumakejobserver_mock), \
         mock.patch('twisterlib.runner.JobClient', jobclient_mock), \
         mock.patch('multiprocessing.cpu_count', return_value=8), \
         mock.patch('sys.platform', platform), \
         mock.patch('time.sleep', mock.Mock()), \
         mock.patch('os.name', os_name):
        tr.run()

    assert f'JOBS: {expected_jobs}' in caplog.text

    assert tr.jobserver.name == expected_jobserver

    assert tr.instances['dummy instance'].metrics == {
        'k': 'v',
        'k2': 'v2',
        'handler_time': 30,
        'unrecognized': []
    }

    assert results_mock().error == 0


def test_twisterrunner_update_counting_before_pipeline():
    instances = {
        'dummy1': mock.Mock(
            status='filtered',
            reason='runtime filter',
            testsuite=mock.Mock(
                testcases=[mock.Mock()]
            )
        ),
        'dummy2': mock.Mock(
            status='filtered',
            reason='static filter',
            testsuite=mock.Mock(
                testcases=[mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock()]
            )
        ),
        'dummy3': mock.Mock(
            status='error',
            reason='error',
            testsuite=mock.Mock(
                testcases=[mock.Mock()]
            )
        ),
        'dummy4': mock.Mock(
            status='passed',
            reason='OK',
            testsuite=mock.Mock(
                testcases=[mock.Mock()]
            )
        ),
        'dummy5': mock.Mock(
            status='skipped',
            reason=None,
            testsuite=mock.Mock(
                testcases=[mock.Mock()]
            )
        )
    }
    suites = [mock.Mock()]
    env_mock = mock.Mock()

    tr = TwisterRunner(instances, suites, env=env_mock)
    tr.results = mock.Mock(
        skipped_filter = 0,
        skipped_configs = 0,
        skipped_cases = 0,
        cases = 0,
        error = 0
    )

    tr.update_counting_before_pipeline()

    assert tr.results.skipped_filter == 1
    assert tr.results.skipped_configs == 1
    assert tr.results.skipped_cases == 4
    assert tr.results.cases == 4
    assert tr.results.error == 1


def test_twisterrunner_show_brief(caplog):
    instances = {
        'dummy1': mock.Mock(),
        'dummy2': mock.Mock(),
        'dummy3': mock.Mock(),
        'dummy4': mock.Mock(),
        'dummy5': mock.Mock()
    }
    suites = [mock.Mock(), mock.Mock()]
    env_mock = mock.Mock()

    tr = TwisterRunner(instances, suites, env=env_mock)
    tr.results = mock.Mock(
        skipped_filter = 3,
        skipped_configs = 4,
        skipped_cases = 0,
        cases = 0,
        error = 0
    )

    tr.show_brief()

    log = '2 test scenarios (5 test instances) selected,' \
          ' 4 configurations skipped (3 by static filter, 1 at runtime).'

    assert log in caplog.text


TESTDATA_18 = [
    (False, False, False, [{'op': 'cmake', 'test': mock.ANY}]),
    (False, False, True, [{'op': 'filter', 'test': mock.ANY},
                          {'op': 'cmake', 'test': mock.ANY}]),
    (False, True, True, [{'op': 'run', 'test': mock.ANY},
                         {'op': 'run', 'test': mock.ANY}]),
    (False, True, False, [{'op': 'run', 'test': mock.ANY}]),
    (True, True, False, [{'op': 'cmake', 'test': mock.ANY}]),
    (True, True, True, [{'op': 'filter', 'test': mock.ANY},
                        {'op': 'cmake', 'test': mock.ANY}]),
    (True, False, True, [{'op': 'filter', 'test': mock.ANY},
                         {'op': 'cmake', 'test': mock.ANY}]),
    (True, False, False, [{'op': 'cmake', 'test': mock.ANY}]),
]

@pytest.mark.parametrize(
    'build_only, test_only, retry_build_errors, expected_pipeline_elements',
    TESTDATA_18,
    ids=['none', 'retry', 'test+retry', 'test', 'build+test',
         'build+test+retry', 'build+retry', 'build']
)
def test_twisterrunner_add_tasks_to_queue(
    build_only,
    test_only,
    retry_build_errors,
    expected_pipeline_elements
):
    def mock_get_cmake_filter_stages(filter, keys):
        return [filter]

    instances = {
        'dummy1': mock.Mock(run=True, retries=0, status='passed', build_dir="/tmp"),
        'dummy2': mock.Mock(run=True, retries=0, status='skipped', build_dir="/tmp"),
        'dummy3': mock.Mock(run=True, retries=0, status='filtered', build_dir="/tmp"),
        'dummy4': mock.Mock(run=True, retries=0, status='error', build_dir="/tmp"),
        'dummy5': mock.Mock(run=True, retries=0, status='failed', build_dir="/tmp")
    }
    instances['dummy4'].testsuite.filter = 'some'
    instances['dummy5'].testsuite.filter = 'full'
    suites = [mock.Mock(), mock.Mock()]
    env_mock = mock.Mock()

    tr = TwisterRunner(instances, suites, env=env_mock)
    tr.get_cmake_filter_stages = mock.Mock(
        side_effect=mock_get_cmake_filter_stages
    )

    pipeline_mock = mock.Mock()

    tr.add_tasks_to_queue(
        pipeline_mock,
        build_only,
        test_only,
        retry_build_errors
    )

    assert all(
        [build_only != instance.run for instance in instances.values()]
    )

    tr.get_cmake_filter_stages.assert_any_call('full', mock.ANY)
    if retry_build_errors:
        tr.get_cmake_filter_stages.assert_any_call('some', mock.ANY)

    print(pipeline_mock.put.call_args_list)
    print([mock.call(el) for el in expected_pipeline_elements])

    assert pipeline_mock.put.call_args_list == \
           [mock.call(el) for el in expected_pipeline_elements]


TESTDATA_19 = [
    ('linux'),
    ('nt')
]

@pytest.mark.parametrize(
    'platform',
    TESTDATA_19,
)
def test_twisterrunner_pipeline_mgr(mocked_jobserver, platform):
    counter = 0
    def mock_get_nowait():
        nonlocal counter
        counter += 1
        if counter > 5:
            raise queue.Empty()
        return {'test': 'dummy'}

    instances = {}
    suites = []
    env_mock = mock.Mock()

    tr = TwisterRunner(instances, suites, env=env_mock)
    tr.jobserver = mock.Mock(
        get_job=mock.Mock(
            return_value=nullcontext()
        )
    )

    pipeline_mock = mock.Mock()
    pipeline_mock.get_nowait = mock.Mock(side_effect=mock_get_nowait)
    done_queue_mock = mock.Mock()
    lock_mock = mock.Mock()
    results_mock = mock.Mock()

    with mock.patch('sys.platform', platform), \
         mock.patch('twisterlib.runner.ProjectBuilder',\
                    return_value=mock.Mock()) as pb:
        tr.pipeline_mgr(pipeline_mock, done_queue_mock, lock_mock, results_mock)

    assert len(pb().process.call_args_list) == 5

    if platform == 'linux':
        tr.jobserver.get_job.assert_called_once()


def test_twisterrunner_execute(caplog):
    counter = 0
    def mock_join():
        nonlocal counter
        counter += 1
        if counter > 3:
            raise KeyboardInterrupt()

    instances = {}
    suites = []
    env_mock = mock.Mock()

    tr = TwisterRunner(instances, suites, env=env_mock)
    tr.add_tasks_to_queue = mock.Mock()
    tr.jobs = 5

    process_mock = mock.Mock()
    process_mock().join = mock.Mock(side_effect=mock_join)
    pipeline_mock = mock.Mock()
    done_mock = mock.Mock()

    with mock.patch('twisterlib.runner.Process', process_mock):
        tr.execute(pipeline_mock, done_mock)

    assert 'Execution interrupted' in caplog.text

    assert len(process_mock().start.call_args_list) == 5
    assert len(process_mock().join.call_args_list) == 4
    assert len(process_mock().terminate.call_args_list) == 5


TESTDATA_20 = [
    ('', []),
    ('not ARCH in ["x86", "arc"]', ['full']),
    ('dt_dummy(x, y)', ['dts']),
    ('not CONFIG_FOO', ['kconfig']),
    ('dt_dummy and CONFIG_FOO', ['dts', 'kconfig']),
]

@pytest.mark.parametrize(
    'filter, expected_result',
    TESTDATA_20,
    ids=['none', 'full', 'dts', 'kconfig', 'dts+kconfig']
)
def test_twisterrunner_get_cmake_filter_stages(filter, expected_result):
    result = TwisterRunner.get_cmake_filter_stages(filter, ['not', 'and'])

    assert sorted(result) == sorted(expected_result)