tests: refactor pytest_plugins for twister tests

Splitted `cli_commands.py` module to multiple modules.

Signed-off-by: Lukasz Fundakowski <lukasz.fundakowski@nordicsemi.no>

Author: fundakol

scripts/pytest_plugins/utils/__init__.py …ipts/pytest_plugins/adapters/__init__.pyscripts/pytest_plugins/utils/__init__.py renamed to scripts/pytest_plugins/adapters/__init__.py scripts/pytest_plugins/utils/__init__.py renamed to scripts/pytest_plugins/adapters/__init__.py

@@ -0,0 +1,79 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+"""Builder pattern implementation for building Zephyr/NCS projects.
+
+This module provides a builder interface and implementation for building
+Nordic Semiconductor projects using the west build system. It includes
+support for parallel builds through the BuildDirector orchestrator.
+"""
+
+import abc
+import concurrent.futures
+import logging
+import multiprocessing
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import Path
+
+from .common import duration
+from .west import west_build
+
+MAX_WORKERS = max(multiprocessing.cpu_count() - 1, 1)
+
+logger = logging.getLogger(__name__)
+
+
+class Builder(abc.ABC):
+    """Interface class for builders."""
+
+    @abc.abstractmethod
+    def build(self) -> tuple[Path, Path]:
+        """Run build and return source and build paths as tuple."""
+
+
+@dataclass
+class WestBuilder(Builder):
+    """Build a sample using west."""
+
+    source_dir: Path
+    build_dir: Path
+    board: str
+    testsuite: str | None = None
+    extra_args: str | None = None
+    timeout: int = 60
+
+    @duration
+    def build(self) -> tuple[Path, Path]:
+        """Run west build for the required build configuration."""
+        return west_build(
+            source_dir=self.source_dir,
+            board=self.board,
+            build_dir=self.build_dir,
+            timeout=self.timeout,
+            testsuite=self.testsuite,
+            extra_args=self.extra_args,
+        )
+
+
+class BuildDirector:
+    """Orchestrator for builders to build in parallel."""
+
+    def __init__(self, builders: Sequence[Builder], max_workers: int = MAX_WORKERS) -> None:
+        self.builders = builders
+        self.max_workers = max_workers
+        self.exceptions: list[Exception] = []
+
+    def run(self) -> None:
+        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
+            tasks = {executor.submit(builder.build) for builder in self.builders}
+            for future in concurrent.futures.as_completed(tasks):
+                try:
+                    _, build_dir = future.result()
+                except Exception as exc:
+                    logger.error('Failed to build: %s', exc)
+                    self.exceptions.append(exc)
+                else:
+                    logger.info("Build success: %s", build_dir)
+        logger.info("Finished building samples")

scripts/pytest_plugins/adapters/builder.py

@@ -0,0 +1,58 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+"""Common utility functions for adapter modules.
+
+This module provides shared utility functions used across adapter modules.
+"""
+
+import logging
+import os
+import shlex
+import subprocess
+import time
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def normalize_path(path: str) -> str:
+    """Normalize a path by expanding user and environment variables."""
+    expanded = os.path.expanduser(os.path.expandvars(path))
+    return str(Path(expanded).resolve())
+
+
+def duration(func: Callable) -> Any:
+    """Decorator to print duration of calling a function."""
+
+    def _inner(*args, **kwargs):
+        start_t = time.monotonic()
+        try:
+            return func(*args, **kwargs)
+        finally:
+            logger.debug(
+                "The `%s` function took %s seconds to execute",
+                func.__name__,
+                round(time.monotonic() - start_t, 3),
+            )
+
+    return _inner
+
+
+def run_command(command: list[str], timeout: int = 30) -> None:
+    """Run command in subprocess."""
+    logger.info(f"CMD: {shlex.join(command)}")
+    ret: subprocess.CompletedProcess = subprocess.run(
+        command,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        timeout=timeout,
+    )
+    if ret.returncode:
+        logger.error(f"Failed command: {shlex.join(command)}")
+        logger.info(ret.stdout)
+        raise subprocess.CalledProcessError(ret.returncode, command)

scripts/pytest_plugins/adapters/common.py

@@ -0,0 +1,31 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+"""Adapter for nrfutil device management commands.
+
+This module provides functions for interacting with Nordic Semiconductor
+devices using the nrfutil command-line tool.
+"""
+
+import logging
+
+from .common import run_command
+
+logger = logging.getLogger(__name__)
+
+
+def reset_board(dev_id: str | None = None) -> None:
+    """Reset device."""
+    command = ["nrfutil", "device", "reset"]
+    if dev_id:
+        command.extend(["--serial-number", dev_id])
+    run_command(command)
+
+
+def erase_board(dev_id: str | None) -> None:
+    """Run nrfutil device erase command."""
+    command = ["nrfutil", "device", "erase"]
+    if dev_id:
+        command.extend(["--serial-number", dev_id])
+    run_command(command)

scripts/pytest_plugins/adapters/nrfutil.py

@@ -0,0 +1,110 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+"""Adapter for west commands."""
+
+import logging
+import shlex
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Literal
+
+from .common import normalize_path, run_command
+
+logger = logging.getLogger(__name__)
+
+
+class WestFlashException(Exception):
+    """Raised when west flash failed."""
+
+
+class WestBuildException(Exception):
+    """Raised when west build failed."""
+
+
+def provision_keys_for_kmu(
+    keys: list[str] | list[Path] | str | Path,
+    *,
+    keyname: Literal["UROT_PUBKEY", "BL_PUBKEY", "APP_PUBKEY"] = "BL_PUBKEY",
+    policy: Literal["revokable", "lock", "lock-last"] | None = None,
+    dev_id: str | None = None,
+):
+    """Upload keys with west provision command."""
+    logger.info("Provision keys using west command.")
+    command = ["west", "ncs-provision", "upload", "--keyname", keyname]
+    if policy:
+        command += ["--policy", policy]
+    if dev_id:
+        command += ["--dev-id", dev_id]
+    if isinstance(keys, list):
+        list_of_keys = keys
+    else:
+        list_of_keys = [keys]
+    for key in list_of_keys:
+        key_path = Path(key) if not isinstance(key, Path) else key
+        assert key_path.exists(), f"Key file does not exist: {key}"
+        command += ["--key", normalize_path(str(key_path))]
+
+    run_command(command)
+    logger.info("Keys provisioned successfully")
+
+
+def west_flash(
+    build_dir: Path, snr: str | None, *, extra_args: str | None = None, timeout: int = 30
+) -> None:
+    """Run west flash."""
+    command = ["west", "flash", "--skip-rebuild", "--build-dir", str(build_dir)]
+    if snr:
+        command += ["--dev-id", snr]
+    if extra_args:
+        command.extend(shlex.split(extra_args, posix=False))
+
+    try:
+        run_command(command, timeout=timeout)
+    except subprocess.CalledProcessError:
+        raise WestFlashException("Failed to flash device") from None
+    except subprocess.TimeoutExpired:
+        logger.error("Timeout flashing device")
+        raise WestFlashException("Timeout flashing device") from None
+
+
+def west_build(
+    source_dir: Path,
+    board: str,
+    build_dir: Path,
+    *,
+    testsuite: str | None = None,
+    extra_args: str | None = None,
+    timeout: int = 60,
+) -> tuple[Path, Path]:
+    """Run west build."""
+    command = [
+        "west",
+        "build",
+        "-p",
+        "-b",
+        board,
+        str(source_dir),
+        "-d",
+        str(build_dir),
+    ]
+    if testsuite:
+        command.extend(["-T", testsuite])
+    if extra_args:
+        command.extend(shlex.split(extra_args, posix=False))
+
+    try:
+        run_command(command, timeout=timeout)
+    except subprocess.CalledProcessError:
+        # create empty file to indicate a build error, will be used
+        # to avoid unnecessary builds in other tests
+        Path(build_dir / "build.error").touch()
+        raise WestBuildException("Failed to build required app") from None
+    except subprocess.TimeoutExpired:
+        logger.error("Timeout building required app")
+        shutil.rmtree(build_dir)
+        raise WestBuildException("Timeout building required app") from None
+
+    return source_dir, build_dir