Twister: additional scripts

doc/develop/test/twister.rst

@@ -1360,6 +1360,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 <PATH_TO_SCRIPTING_LIST_YAML>`` 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: <PATH_TO_PRE_SCRIPT>
+            timeout: <TIMEOUT>
+            override_script: <BOOLEAN>
+        post_flash_script:
+            path: <PATH_TO_POST_FLASH_SCRIPT>
+            timeout: <TIMEOUT>
+            override_script: <BOOLEAN>
+        post_script:
+            path: <PATH_TO_POST_SCRIPT>
+            timeout: <TIMEOUT>
+            override_script: <BOOLEAN>
+        comment:
+            Testing extra scripts
+
 Using Single Board For Multiple Variants
 ++++++++++++++++++++++++++++++++++++++++
 

scripts/pylib/twister/twisterlib/environment.py

@@ -686,6 +686,16 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
         help="""quit twister once there is build / run failure
         """)
 
+    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.
@@ -922,6 +932,10 @@ def parse_arguments(
         )
         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

scripts/pylib/twister/twisterlib/scripting.py

@@ -0,0 +1,141 @@
+# Copyright (c) 2024 Intel Corporation.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import logging
+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'
+
+    # Ensures all required scripts are present and validates the element.
+    def __post_init__(self):
+        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 element.scenarios and not _matches_element(scenario, element.scenarios):
+                continue
+            if element.platforms and not _matches_element(platform, element.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 patterns.
+def _matches_element(element: str, patterns: list[str]) -> bool:
+    return any(pattern in element for pattern in patterns)

scripts/pylib/twister/twisterlib/testplan.py

@@ -34,6 +34,7 @@
 from twisterlib.error import TwisterRuntimeError
 from twisterlib.platform import Platform
 from twisterlib.quarantine import Quarantine
+from twisterlib.scripting import Scripting
 from twisterlib.statuses import TwisterStatus
 from twisterlib.testinstance import TestInstance
 from twisterlib.testsuite import TestSuite, scan_testsuite_path
@@ -94,6 +95,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",
@@ -113,6 +118,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 = []
@@ -225,6 +231,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:
@@ -265,6 +276,21 @@ 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:
@@ -1307,6 +1333,67 @@ 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 "
+                                        f"{platform_name} with timeout {script_obj.timeout}"
+                                    )
+                                else:
+                                    logger.info(
+                                        f"{script_type} {script_obj.path} will be executed on "
+                                        f"{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.'''