#!/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'), ] project_builder.testsuite.sysbuild = False 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 project_builder.testsuite.sysbuild = False 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 = '' 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', '-DSB_CONFIG_COMPILER_WARNINGS_AS_ERRORS=y', '-DEXTRA_GEN_DEFINES_ARGS=--edtlib-Werror', '-Gdummy_generator', '-S' + os.path.join('source', 'dir'), 'arg1', 'arg2', '-DBOARD=', '-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', '-DSB_CONFIG_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=', '-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 = '' 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() instance_mock.testsuite.sysbuild = False 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 ): def mock_get_domains(*args, **kwargs): return [] instance_mock = mock.Mock() instance_mock.build_dir = os.path.join('build', 'dir') instance_mock.domains.get_domains.side_effect = mock_get_domains 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, None, []), (None, {'dummy': 'dummy'}, []), ( None, { 'config': { 'elf_file': '/absolute/path/dummy.elf', 'bin_file': 'path/dummy.bin' } }, ['/absolute/path/dummy.elf', os.path.join('zephyr', 'path/dummy.bin')] ), ( 'test_domain', { 'config': { 'elf_file': '/absolute/path/dummy.elf', 'bin_file': 'path/dummy.bin' } }, ['/absolute/path/dummy.elf', os.path.join('test_domain', 'zephyr', 'path/dummy.bin')] ), ] @pytest.mark.parametrize( 'domain, runners_content, expected_binaries', TESTDATA_10, ids=['no file', 'no config', 'valid', 'with domain'] ) def test_projectbuilder_get_binaries_from_runners( mocked_jobserver, domain, runners_content, expected_binaries ): def mock_exists(fname): assert fname == os.path.join('build', 'dir', domain if domain else '', '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): if domain: bins = pb._get_binaries_from_runners(domain) else: 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)