scripts: twister: Fix Twister Unit tests on Windows

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

@@ -96,7 +96,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

@@ -236,6 +236,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)
@@ -363,6 +364,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

@@ -18,13 +18,13 @@
 from collections.abc import Generator
 from datetime import datetime, timezone
 from importlib import metadata
-from pathlib import Path
 
 import zephyr_module
 from twisterlib.constants import SUPPORTED_SIMS
 from twisterlib.coverage import supported_coverage_formats
 from twisterlib.error import TwisterRuntimeError
 from twisterlib.log_helper import log_command
+from twisterlib.twister_path import TPath
 
 logger = logging.getLogger('twister')
 logger.setLevel(logging.DEBUG)
@@ -133,7 +133,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
     )
 
     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 "
@@ -248,7 +248,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
         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 test suite scenario. These are named by
         'path/relative/to/Zephyr/base/section.subsection_in_testcase_yaml',
         or just 'section.subsection' identifier. With '--testsuite-root' option
@@ -293,16 +293,19 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
 
     # Start of individual args place them in alpha-beta order
 
-    board_root_list = [f"{ZEPHYR_BASE}/boards", f"{ZEPHYR_BASE}/subsys/testsuite/boards"]
+    board_root_list = [
+        TPath(f"{ZEPHYR_BASE}/boards"),
+        TPath(f"{ZEPHYR_BASE}/subsys/testsuite/boards")
+    ]
 
     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>/""")
@@ -349,7 +352,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
         "--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=[],
@@ -374,7 +377,8 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
     parser.add_argument(
         "--test-config",
         action="store",
-        default=os.path.join(ZEPHYR_BASE, "tests", "test_config.yaml"),
+        type=TPath,
+        default=TPath(os.path.join(ZEPHYR_BASE, "tests", "test_config.yaml")),
         help="Path to file with plans and test configurations."
     )
 
@@ -431,7 +435,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
                         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")
 
@@ -513,6 +517,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
         "-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.")
 
@@ -543,7 +548,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
     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(
@@ -604,15 +609,15 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
         )
 
     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
@@ -663,6 +668,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
         "--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 "
@@ -831,7 +837,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
     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, "
@@ -878,8 +884,8 @@ def parse_arguments(
 
         # 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
@@ -1019,34 +1025,38 @@ def __init__(self, options : argparse.Namespace, 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
 
@@ -1143,7 +1153,7 @@ def run_cmake_script(args=None):
         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

@@ -59,13 +59,19 @@ 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:
+        with contextlib.suppress(ProcessLookupError, psutil.NoSuchProcess):
+            p.terminate()
+    _, alive = psutil.wait_procs(to_terminate, timeout=1)
+
+    for p in alive:
         with contextlib.suppress(ProcessLookupError, psutil.NoSuchProcess):
-            os.kill(child.pid, signal.SIGTERM)
-    proc.terminate()
-    # sleep for a while before attempting to kill
-    time.sleep(0.5)
-    proc.kill()
+            p.kill()
+    _, alive = psutil.wait_procs(to_terminate, timeout=1)
 
 
 class Handler:
@@ -203,8 +209,8 @@ def try_kill_process_by_pid(self):
                 pid = int(pid_file.read())
             os.unlink(self.pid_fn)
             self.pid_fn = None  # clear so we don't try to kill the binary twice
-            with contextlib.suppress(ProcessLookupError, psutil.NoSuchProcess):
-                os.kill(pid, signal.SIGKILL)
+            p = psutil.Process(pid)
+            terminate_process(p)
 
     def _output_reader(self, proc):
         self.line = proc.stdout.readline()

scripts/pylib/twister/twisterlib/reports.py

@@ -11,10 +11,11 @@
 import xml.etree.ElementTree as ET
 from datetime import datetime
 from enum import Enum
-from pathlib import Path, PosixPath
+from pathlib import PosixPath, WindowsPath
 
 from colorama import Fore
 from twisterlib.statuses import TwisterStatus
+from twisterlib.twister_path import TPath
 
 logger = logging.getLogger('twister')
 logger.setLevel(logging.DEBUG)
@@ -171,7 +172,7 @@ def xunit_report_suites(self, json_file, filename):
             runnable = suite.get('runnable', 0)
             duration += float(handler_time)
             ts_status = TwisterStatus(suite.get('status'))
-            classname = Path(suite.get("name","")).name
+            classname = TPath(suite.get("name","")).name
             for tc in suite.get("testcases", []):
                 status = TwisterStatus(tc.get('status'))
                 reason = tc.get('reason', suite.get('reason', 'Unknown'))
@@ -253,7 +254,7 @@ def xunit_report(self, json_file, filename, selected_platform=None, full_report=
                 ):
                     continue
                 if full_report:
-                    classname = Path(ts.get("name","")).name
+                    classname = TPath(ts.get("name","")).name
                     for tc in ts.get("testcases", []):
                         status = TwisterStatus(tc.get('status'))
                         reason = tc.get('reason', ts.get('reason', 'Unknown'))
@@ -295,8 +296,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,
@@ -342,7 +349,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

@@ -72,7 +72,7 @@ def __init__(self, testsuite, platform, outdir):
         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,