scripts: twister: Fix Unit Tests on Windows

Unit tests were failing on Windows, indicating that current Twister
code is not truly multiplatform. Paths were deemed the main culprit,
so this commit introduces a new TPath, that aims to fix the Windows
limitation of supporting only up to 260 char paths by default.

Signed-off-by: Lukasz Mrugala <lukaszx.mrugala@intel.com>

scripts/pylib/twister/twisterlib/config_parser.py

@@ -93,7 +93,7 @@ def __init__(self, filename, schema):
         self.common = {}
 
     def load(self):
-        self.data = scl.yaml_load_verify(self.filename, self.schema)
+        self.data = scl.yaml_load_verify(str(self.filename), self.schema)
 
         if 'tests' in self.data:
             self.scenarios = self.data['tests']

scripts/pylib/twister/twisterlib/coverage.py

@@ -221,6 +221,7 @@ def run_command(self, cmd, coveragelog):
                 "--ignore-errors", "mismatch,mismatch",
             ]
 
+        cmd = [str(c) for c in cmd]
         cmd_str = " ".join(cmd)
         logger.debug(f"Running {cmd_str}...")
         return subprocess.call(cmd, stdout=coveragelog)
@@ -346,6 +347,7 @@ def _generate(self, outdir, coveragelog):
                "--gcov-executable", self.gcov_tool,
                "-e", "tests/*"]
         cmd += excludes + mode_options + ["--json", "-o", coveragefile, outdir]
+        cmd = [str(c) for c in cmd]
         cmd_str = " ".join(cmd)
         logger.debug(f"Running {cmd_str}...")
         subprocess.call(cmd, stdout=coveragelog)

scripts/pylib/twister/twisterlib/environment.py

@@ -17,7 +17,6 @@
 import sys
 from datetime import datetime, timezone
 from importlib import metadata
-from pathlib import Path
 from typing import Generator, List
 
 from twisterlib.coverage import supported_coverage_formats
@@ -27,6 +26,7 @@
 
 from twisterlib.error import TwisterRuntimeError
 from twisterlib.log_helper import log_command
+from twisterlib.twister_path import TPath
 
 ZEPHYR_BASE = os.getenv("ZEPHYR_BASE")
 if not ZEPHYR_BASE:
@@ -122,7 +122,7 @@ def add_parse_arguments(parser = None):
         help="Load a list of tests and platforms to be run from a JSON file ('testplan.json' schema).")
 
     case_select.add_argument(
-        "-T", "--testsuite-root", action="append", default=[], type = norm_path,
+        "-T", "--testsuite-root", action="append", default=[], type = TPath,
         help="Base directory to recursively search for test cases. All "
              "testcase.yaml files under here will be processed. May be "
              "called multiple times. Defaults to the 'samples/' and "
@@ -227,7 +227,7 @@ def add_parse_arguments(parser = None):
         and global timeout multiplier (this parameter)""")
 
     test_xor_subtest.add_argument(
-        "-s", "--test", "--scenario", action="append", type = norm_path,
+        "-s", "--test", "--scenario", action="append", type = TPath,
         help="Run only the specified testsuite scenario. These are named by "
              "<path/relative/to/Zephyr/base/section.name.in.testcase.yaml>")
 
@@ -265,17 +265,17 @@ def add_parse_arguments(parser = None):
 
     # Start of individual args place them in alpha-beta order
 
-    board_root_list = ["%s/boards" % ZEPHYR_BASE,
-                       "%s/subsys/testsuite/boards" % ZEPHYR_BASE]
+    board_root_list = [TPath("%s/boards" % ZEPHYR_BASE),
+                       TPath("%s/subsys/testsuite/boards" % ZEPHYR_BASE)]
 
     modules = zephyr_module.parse_modules(ZEPHYR_BASE)
     for module in modules:
         board_root = module.meta.get("build", {}).get("settings", {}).get("board_root")
         if board_root:
-            board_root_list.append(os.path.join(module.project, board_root, "boards"))
+            board_root_list.append(TPath(os.path.join(module.project, board_root, "boards")))
 
     parser.add_argument(
-        "-A", "--board-root", action="append", default=board_root_list,
+        "-A", "--board-root", action="append", default=board_root_list, type=TPath,
         help="""Directory to search for board configuration files. All .yaml
 files in the directory will be processed. The directory should have the same
 structure in the main Zephyr tree: boards/<vendor>/<board_name>/""")
@@ -322,7 +322,7 @@ def add_parse_arguments(parser = None):
         "--cmake-only", action="store_true",
         help="Only run cmake, do not build or run.")
 
-    parser.add_argument("--coverage-basedir", default=ZEPHYR_BASE,
+    parser.add_argument("--coverage-basedir", default=ZEPHYR_BASE, type=TPath,
                         help="Base source directory for coverage report.")
 
     parser.add_argument("--coverage-platform", action="append", default=[],
@@ -340,7 +340,8 @@ def add_parse_arguments(parser = None):
                              " Valid options for 'lcov' tool are: " +
                              ','.join(supported_coverage_formats['lcov']) + " (html,lcov - default).")
 
-    parser.add_argument("--test-config", action="store", default=os.path.join(ZEPHYR_BASE, "tests", "test_config.yaml"),
+    parser.add_argument("--test-config", action="store", type=TPath,
+        default=TPath(os.path.join(ZEPHYR_BASE, "tests", "test_config.yaml")),
         help="Path to file with plans and test configurations.")
 
     parser.add_argument("--level", action="store",
@@ -396,7 +397,7 @@ def add_parse_arguments(parser = None):
                         help="Do not filter based on toolchain, use the set "
                              " toolchain unconditionally")
 
-    parser.add_argument("--gcov-tool", type=Path, default=None,
+    parser.add_argument("--gcov-tool", type=TPath, default=None,
                         help="Path to the gcov tool to use for code coverage "
                              "reports")
 
@@ -478,6 +479,7 @@ def add_parse_arguments(parser = None):
         "-z", "--size",
         action="append",
         metavar='FILENAME',
+        type=TPath,
         help="Ignore all other command line options and just produce a report to "
              "stdout with ROM/RAM section sizes on the specified binary images.")
 
@@ -508,7 +510,7 @@ def add_parse_arguments(parser = None):
     test_plan_report_xor.add_argument("--list-tags", action="store_true",
                         help="List all tags occurring in selected tests.")
 
-    parser.add_argument("--log-file", metavar="FILENAME", action="store",
+    parser.add_argument("--log-file", metavar="FILENAME", action="store", type=TPath,
                         help="Specify a file where to save logs.")
 
     parser.add_argument(
@@ -563,15 +565,15 @@ def add_parse_arguments(parser = None):
         )
 
     parser.add_argument(
-        "-O", "--outdir",
-        default=os.path.join(os.getcwd(), "twister-out"),
+        "-O", "--outdir", type=TPath,
+        default=TPath(os.path.join(os.getcwd(), "twister-out")),
         help="Output directory for logs and binaries. "
              "Default is 'twister-out' in the current directory. "
              "This directory will be cleaned unless '--no-clean' is set. "
              "The '--clobber-output' option controls what cleaning does.")
 
     parser.add_argument(
-        "-o", "--report-dir",
+        "-o", "--report-dir", type=TPath,
         help="""Output reports containing results of the test run into the
         specified directory.
         The output will be both in JSON and JUNIT format
@@ -622,6 +624,7 @@ def add_parse_arguments(parser = None):
         "--quarantine-list",
         action="append",
         metavar="FILENAME",
+        type=TPath,
         help="Load list of test scenarios under quarantine. The entries in "
              "the file need to correspond to the test scenarios names as in "
              "corresponding tests .yaml files. These scenarios "
@@ -784,7 +787,7 @@ def add_parse_arguments(parser = None):
     parser.add_argument("extra_test_args", nargs=argparse.REMAINDER,
         help="Additional args following a '--' are passed to the test binary")
 
-    parser.add_argument("--alt-config-root", action="append", default=[],
+    parser.add_argument("--alt-config-root", action="append", default=[], type=TPath,
         help="Alternative test configuration root/s. When a test is found, "
              "Twister will check if a test configuration file exist in any of "
              "the alternative test configuration root folders. For example, "
@@ -826,8 +829,8 @@ def parse_arguments(parser, args, options = None, on_init=True):
 
         # check again and make sure we have something set
         if not options.testsuite_root:
-            options.testsuite_root = [os.path.join(ZEPHYR_BASE, "tests"),
-                                     os.path.join(ZEPHYR_BASE, "samples")]
+            options.testsuite_root = [TPath(os.path.join(ZEPHYR_BASE, "tests")),
+                                     TPath(os.path.join(ZEPHYR_BASE, "samples"))]
 
     if options.last_metrics or options.compare_report:
         options.enable_size_report = True
@@ -959,23 +962,27 @@ def __init__(self, options, default_options=None) -> None:
 
         self.test_roots = options.testsuite_root
 
-        if not isinstance(options.board_root, list):
-            self.board_roots = [options.board_root]
+        if options:
+            if not isinstance(options.board_root, list):
+                self.board_roots = [self.options.board_root]
+            else:
+                self.board_roots = self.options.board_root
+            self.outdir = TPath(os.path.abspath(options.outdir))
         else:
             self.board_roots = options.board_root
         self.outdir = os.path.abspath(options.outdir)
 
-        self.snippet_roots = [Path(ZEPHYR_BASE)]
+        self.snippet_roots = [TPath(ZEPHYR_BASE)]
         modules = zephyr_module.parse_modules(ZEPHYR_BASE)
         for module in modules:
             snippet_root = module.meta.get("build", {}).get("settings", {}).get("snippet_root")
             if snippet_root:
-                self.snippet_roots.append(Path(module.project) / snippet_root)
+                self.snippet_roots.append(TPath(module.project) / snippet_root)
 
 
-        self.soc_roots = [Path(ZEPHYR_BASE), Path(ZEPHYR_BASE) / 'subsys' / 'testsuite']
-        self.dts_roots = [Path(ZEPHYR_BASE)]
-        self.arch_roots = [Path(ZEPHYR_BASE)]
+        self.soc_roots = [TPath(ZEPHYR_BASE), TPath(ZEPHYR_BASE) / 'subsys' / 'testsuite']
+        self.dts_roots = [TPath(ZEPHYR_BASE)]
+        self.arch_roots = [TPath(ZEPHYR_BASE)]
 
         for module in modules:
             soc_root = module.meta.get("build", {}).get("settings", {}).get("soc_root")
@@ -1081,7 +1088,7 @@ def run_cmake_script(args=[]):
         return results
 
     def get_toolchain(self):
-        toolchain_script = Path(ZEPHYR_BASE) / Path('cmake/verify-toolchain.cmake')
+        toolchain_script = TPath(ZEPHYR_BASE) / TPath('cmake/verify-toolchain.cmake')
         result = self.run_cmake_script([toolchain_script, "FORMAT=json"])
 
         try:

scripts/pylib/twister/twisterlib/handlers.py

@@ -61,15 +61,23 @@ def terminate_process(proc):
     so we need to use try_kill_process_by_pid.
     """
 
-    for child in psutil.Process(proc.pid).children(recursive=True):
+    parent = psutil.Process(proc.pid)
+    to_terminate = parent.children(recursive=True)
+    to_terminate.append(parent)
+
+    for p in to_terminate:
+        try:
+            p.terminate()
+        except (ProcessLookupError, psutil.NoSuchProcess):
+            pass
+    _, alive = psutil.wait_procs(to_terminate, timeout=1)
+
+    for p in alive:
         try:
-            os.kill(child.pid, signal.SIGTERM)
+            p.kill()
         except (ProcessLookupError, psutil.NoSuchProcess):
             pass
-    proc.terminate()
-    # sleep for a while before attempting to kill
-    time.sleep(0.5)
-    proc.kill()
+    _, alive = psutil.wait_procs(to_terminate, timeout=1)
 
 
 class Handler:
@@ -193,10 +201,8 @@ def try_kill_process_by_pid(self):
             pid = int(open(self.pid_fn).read())
             os.unlink(self.pid_fn)
             self.pid_fn = None  # clear so we don't try to kill the binary twice
-            try:
-                os.kill(pid, signal.SIGKILL)
-            except (ProcessLookupError, psutil.NoSuchProcess):
-                pass
+            p = psutil.Process(pid)
+            terminate_process(p)
 
     def _output_reader(self, proc):
         self.line = proc.stdout.readline()

scripts/pylib/twister/twisterlib/package.py

@@ -30,12 +30,21 @@ def package(self):
                 if t['status'] != TwisterStatus.FILTER:
                     p = t['platform']
                     normalized  = p.replace("/", "_")
-                    dirs.append(os.path.join(self.options.outdir, normalized, t['name']))
+                    print(os.getcwd())
+                    print(self.options.outdir)
+                    print(normalized)
+                    print(t['name'])
+                    dir = os.path.join(self.options.outdir, normalized, t['name'])
+                    print(dir)
+                    dirs.append(dir)
 
         dirs.extend(
             [
                 os.path.join(self.options.outdir, "twister.json"),
                 os.path.join(self.options.outdir, "testplan.json")
                 ]
             )
+
+        print(dirs)
+
         self.make_tarfile(self.options.package_artifacts, dirs)

scripts/pylib/twister/twisterlib/reports.py

@@ -12,7 +12,9 @@
 import xml.etree.ElementTree as ET
 import string
 from datetime import datetime
-from pathlib import PosixPath
+from pathlib import PosixPath, WindowsPath
+
+from twisterlib.twister_path import TPath
 
 from twisterlib.statuses import TwisterStatus
 
@@ -267,8 +269,14 @@ def json_report(self, filename, version="NA", platform=None, filters=None):
             report_options = self.env.non_default_options()
 
         # Resolve known JSON serialization problems.
-        for k,v in report_options.items():
-            report_options[k] = str(v) if type(v) in [PosixPath] else v
+        for k, v in report_options.items():
+            pathlikes = [PosixPath, WindowsPath, TPath]
+            value = v
+            if type(v) in pathlikes:
+                value = os.fspath(v)
+            if type(v) in [list]:
+                value = [os.fspath(x) if type(x) in pathlikes else x for x in v]
+            report_options[k] = value
 
         report = {}
         report["environment"] = {"os": os.name,
@@ -310,7 +318,7 @@ def json_report(self, filename, version="NA", platform=None, filters=None):
                 "name": instance.testsuite.name,
                 "arch": instance.platform.arch,
                 "platform": instance.platform.name,
-                "path": instance.testsuite.source_dir_rel
+                "path": os.fspath(instance.testsuite.source_dir_rel)
             }
             if instance.run_id:
                 suite['run_id'] = instance.run_id

scripts/pylib/twister/twisterlib/size_calc.py

@@ -205,7 +205,7 @@ def _check_is_xip(self) -> None:
         # Search for CONFIG_XIP in the ELF's list of symbols using NM and AWK.
         # GREP can not be used as it returns an error if the symbol is not
         # found.
-        is_xip_command = "nm " + self.elf_filename + \
+        is_xip_command = "nm " + str(self.elf_filename) + \
                          " | awk '/CONFIG_XIP/ { print $3 }'"
         is_xip_output = subprocess.check_output(
             is_xip_command, shell=True, stderr=subprocess.STDOUT).decode(
@@ -221,7 +221,7 @@ def _check_is_xip(self) -> None:
 
     def _get_info_elf_sections(self) -> None:
         """Calculate RAM and ROM usage and information about issues by section"""
-        objdump_command = "objdump -h " + self.elf_filename
+        objdump_command = "objdump -h " + str(self.elf_filename)
         objdump_output = subprocess.check_output(
             objdump_command, shell=True).decode("utf-8").splitlines()
 

scripts/pylib/twister/twisterlib/testinstance.py

@@ -68,7 +68,7 @@ def __init__(self, testsuite, platform, outdir):
             self.build_dir = os.path.join(outdir, platform.normalized_name, testsuite.name)
         else:
             # if suite is not in zephyr, keep only the part after ".." in reconstructed dir structure
-            source_dir_rel = testsuite.source_dir_rel.rsplit(os.pardir+os.path.sep, 1)[-1]
+            source_dir_rel = testsuite.source_dir_rel.get_rel_after_dots()
             self.build_dir = os.path.join(outdir, platform.normalized_name, source_dir_rel, testsuite.name)
         self.run_id = self._get_run_id()
         self.domains = None

scripts/pylib/twister/twisterlib/testplan.py

@@ -34,6 +34,7 @@
 from twisterlib.config_parser import TwisterConfigParser
 from twisterlib.statuses import TwisterStatus
 from twisterlib.testinstance import TestInstance
+from twisterlib.twister_path import TPath
 from twisterlib.quarantine import Quarantine
 
 import list_boards
@@ -404,7 +405,9 @@ def add_configurations(self):
         # Note, internally in twister a board root includes the `boards` folder
         # but in Zephyr build system, the board root is without the `boards` in folder path.
         board_roots = [Path(os.path.dirname(root)) for root in self.env.board_roots]
-        lb_args = Namespace(arch_roots=self.env.arch_roots, soc_roots=self.env.soc_roots,
+        arch_roots = [Path(os.path.dirname(root)) for root in self.env.arch_roots]
+        soc_roots = [Path(os.path.dirname(root)) for root in self.env.soc_roots]
+        lb_args = Namespace(arch_roots=arch_roots, soc_roots=soc_roots,
                             board_roots=board_roots, board=None, board_dir=None)
 
         v1_boards = list_boards.find_boards(lb_args)
@@ -528,13 +531,13 @@ def add_testsuites(self, testsuite_filter=[]):
 
                 logger.debug("Found possible testsuite in " + dirpath)
 
-                suite_yaml_path = os.path.join(dirpath, filename)
+                suite_yaml_path = TPath(os.path.join(dirpath, filename))
                 suite_path = os.path.dirname(suite_yaml_path)
 
                 for alt_config_root in self.env.alt_config_root:
-                    alt_config = os.path.join(os.path.abspath(alt_config_root),
+                    alt_config = TPath(os.path.join(os.path.abspath(alt_config_root),
                                               os.path.relpath(suite_path, root),
-                                              filename)
+                                              filename))
                     if os.path.exists(alt_config):
                         logger.info("Using alternative configuration from %s" %
                                     os.path.normpath(alt_config))