From 4c31958d95b266c3513533ab0fc14d709baf24e6 Mon Sep 17 00:00:00 2001 From: Mateusz Junkier Date: Wed, 15 May 2024 14:42:56 +0200 Subject: [PATCH] twister: call pre/post scripts from yaml file Add the execution of external scripts at precise moments. These scripts can be strategically deployed in three distinct phases: pre-script, post-flash-script and post-script. This functionality could help configuring the environment optimally before testing. Signed-off-by: Mateusz Junkier --- doc/develop/test/twister.rst | 50 ++++++ .../pylib/twister/twisterlib/environment.py | 14 ++ scripts/pylib/twister/twisterlib/scripting.py | 148 ++++++++++++++++++ scripts/pylib/twister/twisterlib/testplan.py | 58 +++++++ scripts/schemas/twister/scripting-schema.yaml | 67 ++++++++ scripts/tests/twister/test_testplan.py | 1 + 6 files changed, 338 insertions(+) create mode 100644 scripts/pylib/twister/twisterlib/scripting.py create mode 100644 scripts/schemas/twister/scripting-schema.yaml diff --git a/doc/develop/test/twister.rst b/doc/develop/test/twister.rst index a3893e5da6a9..95b5c835cbc2 100644 --- a/doc/develop/test/twister.rst +++ b/doc/develop/test/twister.rst @@ -1307,6 +1307,56 @@ using an external J-Link probe. The ``probe_id`` keyword overrides the runner: jlink serial: null +Additional Scripts +++++++++++++++++++ + +Twister offers users the flexibility to automate the execution of external +scripts at precise moments. These scripts can be strategically deployed in +three distinct phases: pre-script, post-flash-script and post-script. +This functionality could help configuring the environment optimally +before testing. + +To leverage the scripting capability, users must append the argument +``--scripting-list `` to a twister call. +Parameter ``override_script`` added to explicitly confirm the intent +to execute the specified script. When set to true, this flag allow to +override ``--pre_script``, ``--post_flash_script``, ``--post_script`` +commands specified via other sources. + +The scripting YAML should consist of a series of dictionaries, +each containing the keys scenarios, ``scenarios``, ``platforms``, +``pre_script``, ``post_flash_script``, ``post_script``. Each script +is defined by ``path`` representing path to script, ``timeout`` optional +integer specifying the maximum duration allowed for the script execution, +and ``override_script``. These keys define the specific combinations +of scenarios and platforms, as well as the corresponding scripts +to be executed at each stage. Additionally, it is mandatory +to include a comment entry ``comment`` which is used to give more +details about scripts and purpose of use. + +An example of entries in a scripting list yaml: + +.. code-block:: yaml + + - scenarios: + - sample.basic.helloworld + platforms: + - frdm_k64f + pre_script: + path: + timeout: + override_script: + post_flash_script: + path: + timeout: + override_script: + post_script: + path: + timeout: + override_script: + comment: + Testing extra scripts + Using Single Board For Multiple Variants ++++++++++++++++++++++++++++++++++++++++ diff --git a/scripts/pylib/twister/twisterlib/environment.py b/scripts/pylib/twister/twisterlib/environment.py index 7d55c0f8d27f..1791db4f1e08 100644 --- a/scripts/pylib/twister/twisterlib/environment.py +++ b/scripts/pylib/twister/twisterlib/environment.py @@ -658,6 +658,16 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser: help="Use the list of test scenarios under quarantine and run them" "to verify their current status.") + parser.add_argument( + "--scripting-list", + action="append", + metavar="YAML_FILE", + help="YAML configuration file with device handler hooks to run additional " + "pre-/post- flash phase scripts for selected platform and test scenario combinations. " + "The file must comply with `scripting-schema.yaml`. " + "Overrides `--pre-script` and `--hardware-map` settings. " + "Requires `--device-testing`") + parser.add_argument( "--report-name", help="""Create a report with a custom name. @@ -884,6 +894,10 @@ def parse_arguments(parser: argparse.ArgumentParser, args, options = None, on_in logger.error("Use --device-testing with --device-serial, or --device-serial-pty, or --hardware-map.") sys.exit(1) + if options.scripting_list and not options.device_testing: + logger.error("When --scripting_list is used --device-testing is required") + sys.exit(1) + if options.device_testing and (options.device_serial or options.device_serial_pty) and len(options.platform) != 1: logger.error("When --device-testing is used with --device-serial " "or --device-serial-pty, exactly one platform must " diff --git a/scripts/pylib/twister/twisterlib/scripting.py b/scripts/pylib/twister/twisterlib/scripting.py new file mode 100644 index 000000000000..fef6612e262a --- /dev/null +++ b/scripts/pylib/twister/twisterlib/scripting.py @@ -0,0 +1,148 @@ +# Copyright (c) 2024 Intel Corporation. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import logging +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path + +import scl + +logger = logging.getLogger('twister') + + +# Handles test scripting configurations. +class Scripting: + def __init__(self, scripting_files: list[Path | str], scripting_schema: dict) -> None: + self.scripting = ScriptingData() + self.scripting_files = scripting_files or [] + self.scripting_schema = scripting_schema + self.load_and_validate_files() + + # Finds and returns the scripting element that matches the given test name and platform. + def get_matched_scripting(self, testname: str, platform: str) -> ScriptingElement | None: + matched_scripting = self.scripting.find_matching_scripting(testname, platform) + if matched_scripting: + logger.debug( + f"'{testname}' on '{platform}' device handler scripts '{str(matched_scripting)}'" + ) + return matched_scripting + return None + + def load_and_validate_files(self): + for scripting_file in self.scripting_files: + self.scripting.extend( + ScriptingData.load_from_yaml(scripting_file, self.scripting_schema) + ) + + +@dataclass +class Script: + path: str | None = None + timeout: int | None = None + override_script: bool = False + + +@dataclass +# Represents a single scripting element with associated scripts and metadata. +class ScriptingElement: + scenarios: list[str] = field(default_factory=list) + platforms: list[str] = field(default_factory=list) + pre_script: Script | None = None + post_flash_script: Script | None = None + post_script: Script | None = None + comment: str = 'NA' + re_scenarios: list[re.Pattern] = field(init=False, default_factory=list) + re_platforms: list[re.Pattern] = field(init=False, default_factory=list) + + # Compiles regex patterns for scenarios and platforms, and validates the element. + def __post_init__(self): + self.re_scenarios = [re.compile(pat) for pat in self.scenarios] + self.re_platforms = [re.compile(pat) for pat in self.platforms] + if not any([self.pre_script, self.post_flash_script, self.post_script]): + logger.error("At least one of the scripts must be specified") + sys.exit(1) + self.pre_script = self._convert_to_script(self.pre_script) + self.post_flash_script = self._convert_to_script(self.post_flash_script) + self.post_script = self._convert_to_script(self.post_script) + + # Converts a dictionary to a Script instance if necessary. + def _convert_to_script(self, script: dict | Script | None) -> Script | None: + if isinstance(script, dict): + return Script(**script) + return script + + +@dataclass +# Holds a collection of scripting elements. +class ScriptingData: + elements: list[ScriptingElement] = field(default_factory=list) + + # Ensures all elements are ScriptingElement instances. + def __post_init__(self): + self.elements = [ + elem if isinstance(elem, ScriptingElement) else ScriptingElement(**elem) + for elem in self.elements + ] + + @classmethod + # Loads scripting data from a YAML file. + def load_from_yaml(cls, filename: Path | str, schema: dict) -> ScriptingData: + try: + raw_data = scl.yaml_load_verify(filename, schema) or [] + return cls(raw_data) + except scl.EmptyYamlFileException: + logger.error(f'Scripting file {filename} is empty') + sys.exit(1) + except FileNotFoundError: + logger.error(f'Scripting file {filename} not found') + sys.exit(1) + except Exception as e: + logger.error(f'Error loading {filename}: {e}') + sys.exit(1) + + # Extends the current scripting data with another set of scripting data. + def extend(self, other: ScriptingData) -> None: + self.elements.extend(other.elements) + + # Finds a scripting element that matches the given scenario and platform. + def find_matching_scripting(self, scenario: str, platform: str) -> ScriptingElement | None: + matched_elements = [] + + for element in self.elements: + if not isinstance(element, ScriptingElement): + element = ScriptingElement(**element) + if element.scenarios and not _matches_element(scenario, element.re_scenarios): + continue + if element.platforms and not _matches_element(platform, element.re_platforms): + continue + matched_elements.append(element) + + # Check for override_script + override_scripts = [ + elem + for elem in matched_elements + if ( + (elem.pre_script and elem.pre_script.override_script) + or (elem.post_flash_script and elem.post_flash_script.override_script) + or (elem.post_script and elem.post_script.override_script) + ) + ] + + if len(override_scripts) > 1: + logger.error("Multiple override definition for scripts found") + sys.exit(1) + elif len(override_scripts) == 1: + return override_scripts[0] + elif matched_elements: + return matched_elements[0] + + return None + + +# Checks if the given element matches any of the provided regex patterns. +def _matches_element(element: str, patterns: list[re.Pattern]) -> bool: + return any(pattern.match(element) for pattern in patterns) diff --git a/scripts/pylib/twister/twisterlib/testplan.py b/scripts/pylib/twister/twisterlib/testplan.py index 0131e42b8fd1..30c8304ec470 100755 --- a/scripts/pylib/twister/twisterlib/testplan.py +++ b/scripts/pylib/twister/twisterlib/testplan.py @@ -36,6 +36,7 @@ from twisterlib.statuses import TwisterStatus from twisterlib.testinstance import TestInstance from twisterlib.quarantine import Quarantine +from twisterlib.scripting import Scripting import list_boards from zephyr_module import parse_modules @@ -91,6 +92,10 @@ class TestPlan: os.path.join(ZEPHYR_BASE, "scripts", "schemas", "twister", "quarantine-schema.yaml")) + scripting_schema = scl.yaml_load( + os.path.join(ZEPHYR_BASE, + "scripts", "schemas", "twister", "scripting-schema.yaml")) + tc_schema_path = os.path.join(ZEPHYR_BASE, "scripts", "schemas", "twister", "test-config-schema.yaml") SAMPLE_FILENAME = 'sample.yaml' @@ -104,6 +109,7 @@ def __init__(self, env: Namespace): # Keep track of which test cases we've filtered out and why self.testsuites = {} self.quarantine = None + self.scripting = None self.platforms = [] self.platform_names = [] self.selected_platforms = [] @@ -214,6 +220,11 @@ def discover(self): logger.debug(f'Quarantine file {quarantine_file} is empty') self.quarantine = Quarantine(ql) + # handle extra scripts + sl = self.options.scripting_list + if sl: + self.scripting = Scripting(sl, self.scripting_schema) + def load(self): if self.options.report_suffix: @@ -251,6 +262,16 @@ def load(self): else: self.apply_filters() + if self.scripting: + # Check if at least one provided script met the conditions. + # Summarize logs for all calls. + was_script_matched = False + for instance in self.instances.values(): + was_script_matched = was_script_matched or self.handle_additional_scripts(instance.platform.name, instance) + + if not was_script_matched: + logger.info("Scripting list was provided, none of the specified conditions were met") + if self.options.subset: s = self.options.subset try: @@ -1170,6 +1191,43 @@ def _create_build_dir_link(self, links_dir_path, instance): self.link_dir_counter += 1 + def handle_additional_scripts(self, platform_name: str, testsuite: TestInstance) -> bool: + logger.debug(testsuite.testsuite.id) + matched_scripting = self.scripting.get_matched_scripting(testsuite.testsuite.id, platform_name) + if matched_scripting: + # Define a function to validate if the platform is supported by the matched scripting + def validate_boards(platform_scope, platform_from_yaml): + return any(board in platform_scope for board in platform_from_yaml) + + # Define the types of scripts we are interested in as a set + script_types = {'pre_script': 'pre_script_timeout', 'post_flash_script': 'post_flash_timeout', 'post_script': 'post_script_timeout'} + + # Iterate over all DUTs to set the appropriate scripts if they match the platform and are supported + for dut in self.env.hwm.duts: + # Check if the platform matches and if the platform is supported by the matched scripting + if dut.platform in platform_name and validate_boards(platform_name, matched_scripting.platforms): + for script_type, script_timeout in script_types.items(): + # Get the script object from matched_scripting + script_obj = getattr(matched_scripting, script_type, None) + # If a script object is provided, check if the script path is a valid file + if script_obj and script_obj.path: + # Check if there's an existing script and if override is not allowed + if not script_obj.override_script: + logger.info(f"{script_type} will not be overridden on {platform_name}.") + continue + # Check if the script path is a valid file and set it on the DUT + if Path(script_obj.path).is_file(): + setattr(dut, script_type, script_obj.path) + # Check if the script timeout is provided and set it on the DUT + if script_obj.timeout is not None: + setattr(dut, script_timeout, script_obj.timeout) + logger.info(f"{script_type} {script_obj.path} will be executed on {platform_name} with timeout {script_obj.timeout}") + else: + logger.info(f"{script_type} {script_obj.path} will be executed on {platform_name} with no timeout specified") + else: + raise TwisterRuntimeError(f"{script_type} script not found under path: {script_obj.path}") + return True + return False def change_skip_to_error_if_integration(options, instance): ''' All skips on integration_platforms are treated as errors.''' diff --git a/scripts/schemas/twister/scripting-schema.yaml b/scripts/schemas/twister/scripting-schema.yaml new file mode 100644 index 000000000000..576ac1d93dcf --- /dev/null +++ b/scripts/schemas/twister/scripting-schema.yaml @@ -0,0 +1,67 @@ +type: seq +matching: all +sequence: + - type: map + required: true + matching: all + mapping: + "scenarios": + type: seq + required: true + sequence: + - type: str + - unique: true + "platforms": + required: true + type: seq + sequence: + - type: str + - unique: true + "pre_script": + type: map + required: false + mapping: + "path": + type: str + required: true + "timeout": + type: int + default: 30 + required: false + "override_script": + type: bool + default: false + required: false + "post_flash_script": + type: map + required: false + mapping: + "path": + type: str + required: true + "timeout": + type: int + default: 30 + required: false + "override_script": + type: bool + default: false + required: false + "post_script": + type: map + required: false + mapping: + "path": + type: str + required: true + "timeout": + type: int + default: 30 + required: false + "override_script": + type: bool + default: false + required: false + "comment": + type: str + required: true diff --git a/scripts/tests/twister/test_testplan.py b/scripts/tests/twister/test_testplan.py index a885d541f155..564b119075b7 100644 --- a/scripts/tests/twister/test_testplan.py +++ b/scripts/tests/twister/test_testplan.py @@ -567,6 +567,7 @@ def test_testplan_discover( test='ts1', quarantine_list=[tmp_path / qf for qf in ql], quarantine_verify=qv, + scripting_list=[], ) testplan.testsuites = { 'ts1': mock.Mock(id=1),