scripts: twister: Fix Unit Tests on Windows

Unit tests were failing on Windows, indicating that current Twister
code is not truly multiplatform. Linux-specific code parts were
changed into multiplatform ones.

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/pytest-twister-harness/tests/device/hardware_adapter_test.py

@@ -205,6 +205,7 @@ def test_device_log_correct_error_handle(patched_popen, device: HardwareAdapter,
         assert 'flashing error' in file.readlines()
 
 
+@pytest.mark.skipif(os.name == 'nt', reason='PTY is not used on Windows.')
 @mock.patch('twister_harness.device.hardware_adapter.subprocess.Popen')
 @mock.patch('twister_harness.device.hardware_adapter.serial.Serial')
 def test_if_hardware_adapter_uses_serial_pty(
@@ -238,6 +239,7 @@ def test_if_hardware_adapter_uses_serial_pty(
     assert not device._serial_pty_proc
 
 
+@pytest.mark.skipif(os.name == 'nt', reason='PTY is not used on Windows.')
 def test_if_hardware_adapter_properly_send_data_to_subprocess(
     device: HardwareAdapter, shell_simulator_path: str
 ) -> None:

scripts/pylib/pytest-twister-harness/tests/device/qemu_adapter_test.py

@@ -32,6 +32,7 @@ def test_if_generate_command_creates_proper_command(patched_which, device: QemuA
     assert device.command == ['west', 'build', '-d', 'build_dir', '-t', 'run']
 
 
+@pytest.mark.skipif(os.name == 'nt', reason='QEMU FIFO fails on Windows.')
 def test_if_qemu_adapter_runs_without_errors(resources, device: QemuAdapter) -> None:
     fifo_file_path = str(device.device_config.build_dir / 'qemu-fifo')
     script_path = resources.joinpath('fifo_mock.py')
@@ -46,6 +47,7 @@ def test_if_qemu_adapter_runs_without_errors(resources, device: QemuAdapter) ->
     assert file_lines[-2:] == lines[-2:]
 
 
+@pytest.mark.skipif(os.name == 'nt', reason='QEMU FIFO fails on Windows.')
 def test_if_qemu_adapter_raise_exception_due_to_no_fifo_connection(device: QemuAdapter) -> None:
     device.base_timeout = 0.3
     device.command = ['sleep', '1']

scripts/pylib/twister/twisterlib/config_parser.py

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

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:
@@ -133,7 +133,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 "
@@ -238,7 +238,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>")
 
@@ -276,17 +276,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>/""")
@@ -333,7 +333,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=[],
@@ -351,7 +351,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",
@@ -407,7 +408,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")
 
@@ -489,6 +490,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.")
 
@@ -519,7 +521,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(
@@ -574,15 +576,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
@@ -633,6 +635,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 "
@@ -795,7 +798,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, "
@@ -837,8 +840,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
@@ -970,34 +973,38 @@ 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")
             if soc_root:
-                self.soc_roots.append(Path(module.project) / Path(soc_root))
+                self.soc_roots.append(TPath(module.project) / TPath(soc_root))
             dts_root = module.meta.get("build", {}).get("settings", {}).get("dts_root")
             if dts_root:
-                self.dts_roots.append(Path(module.project) / Path(dts_root))
+                self.dts_roots.append(TPath(module.project) / TPath(dts_root))
             arch_root = module.meta.get("build", {}).get("settings", {}).get("arch_root")
             if arch_root:
-                self.arch_roots.append(Path(module.project) / Path(arch_root))
+                self.arch_roots.append(TPath(module.project) / TPath(arch_root))
 
         self.hwm = None
 
@@ -1092,7 +1099,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/reports.py

@@ -12,9 +12,10 @@
 import xml.etree.ElementTree as ET
 import string
 from datetime import datetime
-from pathlib import PosixPath
+from pathlib import PosixPath, WindowsPath
 
 from twisterlib.statuses import TwisterStatus
+from twisterlib.twister_path import TPath
 
 logger = logging.getLogger('twister')
 logger.setLevel(logging.DEBUG)
@@ -269,8 +270,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,
@@ -312,7 +319,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