From 7cc4f7585c2925838fc841aa720e280caa1e5266 Mon Sep 17 00:00:00 2001
From: Jithu Murugan <j.murugan@fz-juelich.de>
Date: Thu, 20 Feb 2025 18:07:54 +0100
Subject: [PATCH] Following changes have been incorporated: -
SpackOperation.py: The handle_result and trust_gpg_key methods were
specifically updated with tests and docstrings. Also modified the add_mirror
method as per the changes. - SpackOperationUseCache.py: Similar to
SpackOperation.py, this file also got enhanced tests and docstrings,
specifically for the setup_spack_env method. The focus is on setting up the
environment for a user spack cache setup along with testing of the caching
behavior and improving documentation. - BuildCacheManager.py: This file saw
improvements in its test suite and the addition of docstrings. The tests
likely cover various cache operations, and the docstrings explain the purpose
and usage of the cache management methods. The get_public_key_from_cache,
download, _log_warning_if_needed, and other methods were updated with tests
and docstrings. - .gitlab-ci.yml: Changes likely include adding or modifying
stages for running the new pytest tests and generating coverage reports. -
pyproject.toml: This file was modified to add testing dependencies required
for the enhanced test suite. Specifically, pytest, pytest-mock,
pytest-ordering, and coverage were added under the test extra. This change
enables running the new tests and generating coverage reports.
The commands folder implements a command execution and management system, especially for Spack, using these key components and patterns:
- Command Definitions (Enum/Constants): command_enum.py likely defines an enumeration (or constants) representing different commands. This promotes type safety and avoids stringly-typed code.
- Command Sequence Builder: command_sequence_builder.py implements the Builder Pattern to construct CommandSequence objects (from command_sequence.py). This allows for flexible and readable creation of complex command sequences.
- Command Sequence: command_sequence.py represents a sequence of commands to be executed. It encapsulates the order and configuration of commands.
- Spack Command Factory: spack_command_sequence_factory.py uses the Factory Pattern to create Spack-specific CommandSequence objects. This centralizes the creation logic for different Spack commands and simplifies their usage. It likely uses command_sequence_builder.py internally.
- Generic Command Runner: A command_runner.py (implied, not explicitly mentioned in the diff) likely handles the actual execution of commands, possibly using subprocesses. This abstracts the execution details from the command sequence logic.
- Focus on Spack: The structure strongly suggests a focus on managing Spack environments and packages. The factory and command definitions likely reflect common Spack operations.
Patterns used:
- Enum/Constants Pattern: Used for defining distinct command types.
- Builder Pattern: Used for constructing command sequences.
- Factory Pattern: Used for creating Spack-specific command sequences.
- Command Pattern: A close variant of the Command Pattern for encapsulating command execution. Abstraction of shell commands along with an invoker (command_runner) been implemented
---
.gitlab-ci.yml | 2 +-
MANIFEST.ini | 3 +
esd/build_cache/BuildCacheManager.py | 51 +++-
esd/commands/__init__.py | 8 +
esd/commands/bash_command_executor.py | 93 +++++++
esd/commands/command.py | 29 +++
esd/commands/command_enum.py | 37 +++
esd/commands/command_registry.py | 59 +++++
esd/commands/command_runner.py | 150 +++++++++++
esd/commands/command_sequence.py | 47 ++++
esd/commands/command_sequence_builder.py | 71 ++++++
esd/commands/command_sequence_factory.py | 45 ++++
esd/commands/generic_shell_command.py | 46 ++++
esd/commands/preconfigured_command_enum.py | 28 +++
esd/commands/shell_command_factory.py | 33 +++
.../spack_command_sequence_factory.py | 211 ++++++++++++++++
esd/configuration/SpackConfig.py | 5 +-
esd/spack_factory/SpackOperation.py | 134 ++++++++--
esd/spack_factory/SpackOperationUseCache.py | 29 ++-
.../unit_tests/test_bash_command_executor.py | 232 ++++++++++++++++++
.../unit_tests/test_build_cache_manager.py | 155 ++++++++++++
esd/tests/unit_tests/test_command.py | 45 ++++
esd/tests/unit_tests/test_command_enum.py | 38 +++
esd/tests/unit_tests/test_command_runner.py | 123 ++++++++++
esd/tests/unit_tests/test_command_sequence.py | 91 +++++++
.../test_command_sequence_builder.py | 95 +++++++
.../test_command_sequence_factory.py | 49 ++++
.../unit_tests/test_generic_shell_command.py | 63 +++++
.../test_preconfigured_command_enum.py | 37 +++
.../unit_tests/test_shell_command_factory.py | 61 +++++
.../test_spack_command_sequence_factory.py | 159 ++++++++++++
esd/tests/unit_tests/test_spack_operation.py | 209 ++++++++++++++++
.../test_spack_operation_use_cache.py | 90 +++++++
pyproject.toml | 20 +-
34 files changed, 2504 insertions(+), 44 deletions(-)
create mode 100644 MANIFEST.ini
create mode 100644 esd/commands/__init__.py
create mode 100644 esd/commands/bash_command_executor.py
create mode 100644 esd/commands/command.py
create mode 100644 esd/commands/command_enum.py
create mode 100644 esd/commands/command_registry.py
create mode 100644 esd/commands/command_runner.py
create mode 100644 esd/commands/command_sequence.py
create mode 100644 esd/commands/command_sequence_builder.py
create mode 100644 esd/commands/command_sequence_factory.py
create mode 100644 esd/commands/generic_shell_command.py
create mode 100644 esd/commands/preconfigured_command_enum.py
create mode 100644 esd/commands/shell_command_factory.py
create mode 100644 esd/commands/spack_command_sequence_factory.py
create mode 100644 esd/tests/unit_tests/test_bash_command_executor.py
create mode 100644 esd/tests/unit_tests/test_build_cache_manager.py
create mode 100644 esd/tests/unit_tests/test_command.py
create mode 100644 esd/tests/unit_tests/test_command_enum.py
create mode 100644 esd/tests/unit_tests/test_command_runner.py
create mode 100644 esd/tests/unit_tests/test_command_sequence.py
create mode 100644 esd/tests/unit_tests/test_command_sequence_builder.py
create mode 100644 esd/tests/unit_tests/test_command_sequence_factory.py
create mode 100644 esd/tests/unit_tests/test_generic_shell_command.py
create mode 100644 esd/tests/unit_tests/test_preconfigured_command_enum.py
create mode 100644 esd/tests/unit_tests/test_shell_command_factory.py
create mode 100644 esd/tests/unit_tests/test_spack_command_sequence_factory.py
create mode 100644 esd/tests/unit_tests/test_spack_operation.py
create mode 100644 esd/tests/unit_tests/test_spack_operation_use_cache.py
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 65062ec7..98a371df 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -31,7 +31,7 @@ testing-pytest:
- chmod +x esd/utils/bootstrap.sh
- ./esd/utils/bootstrap.sh
- echo "$SPACK_ENV_ACCESS_TOKEN"
- - pip install .
+ - pip install -e .[tests,dev]
- pytest ./esd/tests/ -s --junitxml=test-results.xml
artifacts:
when: always
diff --git a/MANIFEST.ini b/MANIFEST.ini
new file mode 100644
index 00000000..d6809e09
--- /dev/null
+++ b/MANIFEST.ini
@@ -0,0 +1,3 @@
+
+include README.md
+recursive-include yashchiki/esd *.*
\ No newline at end of file
diff --git a/esd/build_cache/BuildCacheManager.py b/esd/build_cache/BuildCacheManager.py
index cccb5846..8641a68e 100644
--- a/esd/build_cache/BuildCacheManager.py
+++ b/esd/build_cache/BuildCacheManager.py
@@ -1,10 +1,12 @@
+import glob
import os
-import oras.client
+from os.path import join
from pathlib import Path
+import oras.client
+
from esd.build_cache.BuildCacheManagerInterface import BuildCacheManagerInterface
from esd.logger.logger_builder import get_logger
-from esd.utils.utils import clean_up
class BuildCacheManager(BuildCacheManagerInterface):
@@ -111,7 +113,50 @@ class BuildCacheManager(BuildCacheManagerInterface):
if tags is not None:
try:
self._client.delete_tags(self._oci_registry_path, tags)
- self._logger.info(f"Successfully deleted all artifacts form OCI registry.")
+ self._logger.info("Successfully deleted all artifacts form OCI registry.")
except RuntimeError as e:
self._logger.error(
f"Failed to delete artifacts: {e}")
+
+ def _log_warning_if_needed(self, warn_message: str, items: list[str]) -> None:
+ """Logs a warning message if the number of items is greater than 1. (Private function)
+
+ This method logs a warning message using the provided message and items if the list of items has more than one element.
+
+ Args:
+ warn_message (str): The warning message to log.
+ items (list[str]): The list of items to include in the log message.
+ """
+ if len(items) > 1:
+ self._logger.warning(warn_message, items, items[0])
+
+ def get_public_key_from_cache(self, build_cache_dir: str | None) -> str | None:
+ """Retrieves the public key from the build cache.
+
+ This method searches for the public key within the specified build cache directory.
+
+ Args:
+ build_cache_dir (str | None): The path to the build cache directory.
+
+ Returns:
+ str | None: The path to the public key file if found, otherwise None.
+ """
+
+ if not build_cache_dir or not os.path.exists(build_cache_dir):
+ self._logger.warning("Build cache directory does not exist!")
+ return None
+ pgp_folders = glob.glob(f"{build_cache_dir}/**/_pgp", recursive=True)
+ if not pgp_folders:
+ self._logger.warning("No _pgp folder found in the build cache!")
+ return None
+ self._log_warning_if_needed(
+ "More than one PGP folders found in the build cache: %s, using the first one in the list: %s", pgp_folders)
+ pgp_folder = pgp_folders[0]
+ key_files = glob.glob(join(pgp_folder, "**"))
+ if not key_files:
+ self._logger.warning("No PGP key files found in the build cache!")
+ return None
+ self._log_warning_if_needed(
+ "More than one PGP key files found in the build cache: %s, using the first one in the list: %s", key_files)
+ return key_files[0]
+
diff --git a/esd/commands/__init__.py b/esd/commands/__init__.py
new file mode 100644
index 00000000..ea9c384e
--- /dev/null
+++ b/esd/commands/__init__.py
@@ -0,0 +1,8 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: __init__.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
diff --git a/esd/commands/bash_command_executor.py b/esd/commands/bash_command_executor.py
new file mode 100644
index 00000000..1811b757
--- /dev/null
+++ b/esd/commands/bash_command_executor.py
@@ -0,0 +1,93 @@
+""" Bash Command Executor module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: bash_command_executor.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+import logging
+import os
+import subprocess
+from logging import Logger
+
+from esd.commands.command import Command
+from esd.commands.command_sequence import CommandSequence
+
+
+class BashCommandExecutor:
+ """Executes commands in a Bash shell.
+
+ Manages a sequence of commands and executes them in a single Bash session,
+ handling potential errors during execution.
+ """
+
+ def __init__(self) -> None:
+ self.logger: Logger = logging.getLogger(__name__)
+ self.sequence: CommandSequence = CommandSequence()
+ command_map: dict[str, list[str]] = {
+ "nt": ["wsl", "bash", "-c"], # Works only if wsl is installed on windows
+ "posix": ["bash", "-c"],
+ }
+ self.bash_command: list[str] = command_map.get(os.name, ["undefined"])
+
+ def add_command(self, command: Command) -> None:
+ """Adds a command to the sequence.
+
+ Appends the given command to the internal sequence of commands to be executed.
+
+ Args:
+ command (Command): The command to add.
+
+ Raises:
+ ValueError: If the command is not an instance of the `Command` class.
+ """
+ if not isinstance(command, Command):
+ raise ValueError("Invalid command type. Use Command.")
+ self.logger.info("Adding command to the sequence: %s", command)
+ self.sequence.add_command(command)
+
+ def execute(self) -> tuple[str | None, str | None]:
+ """Executes all commands in a single Bash session.
+
+ Runs the accumulated commands in a Bash shell and returns the output.
+ Handles various potential errors during execution. The execution is time
+
+ Returns (tuple[str | None, str | None]):
+ A tuple containing the output and error message (if any).
+ output will be None if an error occurred, and the error message will
+ contain details about the error.
+
+ Raises:
+ ValueError: If no commands have been added to the sequence.
+ """
+ if not self.sequence.commands:
+ raise ValueError("No commands to execute.")
+ try:
+ result = subprocess.run(
+ [*self.bash_command, self.sequence.execute()],
+ capture_output=True,
+ text=True,
+ check=True,
+ timeout=172800 # Given a default timeout of 48 hours
+ )
+ self.logger.info("Successfully executed command sequence, output: %s", result.stdout)
+ return result.stdout, None
+ except FileNotFoundError as e:
+ error = f"Error: Bash Command: {self.bash_command} not found: {e}"
+ except subprocess.CalledProcessError as e:
+ error = (f"Error: Command failed with exit code "
+ f"{e.returncode}, Error Output: {e.stderr}")
+ except PermissionError as e:
+ error = f"Error: Permission denied: {e}"
+ except OSError as e:
+ error = f"Error: OS error occurred: {e}"
+ except ValueError as e:
+ error = f"Error: Invalid argument passed: {e}"
+ except TypeError as e:
+ error = f"Error: Invalid type for arguments: {e}"
+ except subprocess.TimeoutExpired as e:
+ error = f"Error: Command timed out after {e.timeout} seconds"
+ except subprocess.SubprocessError as e:
+ error = f"Subprocess error occurred: {e}"
+ return None, error
diff --git a/esd/commands/command.py b/esd/commands/command.py
new file mode 100644
index 00000000..07e9837c
--- /dev/null
+++ b/esd/commands/command.py
@@ -0,0 +1,29 @@
+""" Command module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command.py
+# Description: Abstract base class for executable commands
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from abc import ABC, abstractmethod
+
+
+class Command(ABC):
+ """Abstract base class for executable commands.
+
+ Provides a common interface for defining and executing commands.
+ Subclasses must implement the `execute` method.
+ """
+
+ @abstractmethod
+ def execute(self) -> str:
+ """Executes the command.
+
+ This method must be implemented by subclasses to define the specific
+ behavior of the command.
+
+ Returns:
+ str: The result of the command execution.
+ """
diff --git a/esd/commands/command_enum.py b/esd/commands/command_enum.py
new file mode 100644
index 00000000..7ef184cb
--- /dev/null
+++ b/esd/commands/command_enum.py
@@ -0,0 +1,37 @@
+""" Command Enum module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_enum.py
+# Description: Enumeration of supported commands in command registry
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from enum import Enum
+
+
+class CommandEnum(Enum):
+ """Enumeration of supported commands.
+
+ Provides a standardized way to refer to different command types,
+ including Linux commands and Spack commands.
+ """
+ # Linux commands_
+
+ SOURCE = "source"
+ LIST_FILES = "list_files"
+ SHOW_DIRECTORY = "show_directory"
+ FIND_IN_FILE = "find_in_file"
+ ECHO_MESSAGE = "echo_message"
+ CHANGE_DIRECTORY = "change_directory"
+
+ # Spack commands_
+ SPACK_COMPILER_FIND = "spack_compiler_find"
+ SPACK_COMPILERS = "spack_compilers"
+ SPACK_COMPILER_INFO = "spack_compiler_info"
+ SPACK_COMPILER_LIST = "spack_compiler_list"
+ SPACK_ENVIRONMENT_ACTIVATE = "spack_environment_activate"
+ SPACK_FIND = "spack_find"
+ SPACK_INSTALL = "spack_install"
+ SPACK_MIRROR_ADD = "spack_mirror_add"
+ SPACK_GPG_TRUST = "spack_gpg_trust"
diff --git a/esd/commands/command_registry.py b/esd/commands/command_registry.py
new file mode 100644
index 00000000..ea2ab646
--- /dev/null
+++ b/esd/commands/command_registry.py
@@ -0,0 +1,59 @@
+""" Command Registry module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_registry.py
+# Description: Registry for storing and retrieving shell commands with placeholders
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from esd.commands.command_enum import CommandEnum
+
+
+class CommandRegistry:
+ """Registry for storing and retrieving commands.
+
+ Holds a dictionary of commands, keyed by `CommandEnum` members,
+ allowing easy access to command definitions. The generic shell command factory uses this registry for constructing the shell commands.
+
+ The commands are stored in a dictionary where the keys or command names are `CommandEnum` members, and the values corresponding shell commands stored in a list.
+ The placeholders `{}` in the commands are replaced with the actual values when the command is executed.
+
+ Extend this registry if you want to add more commands and do the necessary mapping in command_runner module.
+ """
+
+ COMMANDS: dict[CommandEnum, list[str]] = {
+ # Linux commands_
+ CommandEnum.LIST_FILES: ["ls", "-l", "{folder_location}"],
+ CommandEnum.SHOW_DIRECTORY: ["pwd"],
+ CommandEnum.FIND_IN_FILE: ["grep", "-i", "{search_term}", "{file_path}"],
+ CommandEnum.ECHO_MESSAGE: ["echo", "{message}"],
+ CommandEnum.SOURCE: ["source", "{file_path}"],
+ CommandEnum.CHANGE_DIRECTORY: ["cd", "{folder_location}"],
+
+ # Spack commands_
+ CommandEnum.SPACK_COMPILER_FIND: ["spack", "compiler", "find", "{compiler_name}"],
+ CommandEnum.SPACK_COMPILERS: ["spack", "compilers"],
+ CommandEnum.SPACK_COMPILER_LIST: ["spack", "compiler", "list"],
+ CommandEnum.SPACK_COMPILER_INFO: ["spack", "compiler", "info", "{compiler_name_with_version}"],
+ CommandEnum.SPACK_ENVIRONMENT_ACTIVATE: ["spack", "env", "activate", "-p", "{env_path}"],
+ CommandEnum.SPACK_FIND: ["spack", "find", "{package_name_with_version}"],
+ CommandEnum.SPACK_INSTALL: ["spack", "install", "{package_name_with_version}"],
+ CommandEnum.SPACK_MIRROR_ADD: ["spack", "mirror", "add", "{autopush}", "{signed}", "{mirror_name}",
+ "{mirror_path}"],
+ CommandEnum.SPACK_GPG_TRUST: ["spack", "gpg", "trust", "{public_key_path}"]
+ }
+
+ @classmethod
+ def get_command(cls, command_name: CommandEnum) -> list[str] | None:
+ """Retrieve a command from the registry.
+
+ Returns the command definition associated with the given `CommandEnum` member.
+
+ Args:
+ command_name (CommandEnum): The name of the command to retrieve.
+
+ Returns (list[str]):
+ The shell command in a list format, or None if the command is not found.
+ """
+ return cls.COMMANDS.get(command_name)
diff --git a/esd/commands/command_runner.py b/esd/commands/command_runner.py
new file mode 100644
index 00000000..a9bd89a0
--- /dev/null
+++ b/esd/commands/command_runner.py
@@ -0,0 +1,150 @@
+""" Command Runner module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_runner.py
+# Description: Manages creation, execution, and result handling of command sequences.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from __future__ import annotations
+
+import logging
+from logging import Logger
+from typing import Callable, Any
+
+from esd.commands.bash_command_executor import BashCommandExecutor
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.command_sequence_factory import CommandSequenceFactory
+from esd.commands.preconfigured_command_enum import PreconfiguredCommandEnum
+from esd.commands.spack_command_sequence_factory import SpackCommandSequenceFactory
+
+
+class CommandRunner:
+ """This module provides a unified interface for executing both predefined and custom command sequences.
+
+ Functionality
+
+ The module offers two primary methods for running command sequences:
+
+ 1. Run Preconfigured Command Sequence: This method allows users to execute preconfigured command sequences using the 'run_preconfigured_command_sequence' method.
+ 2. Run Custom Command Sequence: This method enables users to execute dynamically generated custom command sequences using the 'run_custom_command_sequence' method.
+
+ Preconfigured Command Sequences
+ To run a preconfigured command sequence, users can simply call the run_preconfigured_command_sequence method. The module will execute the corresponding command sequence.
+ To create and run a preconfigured command sequence, users must:
+ 1. Define a new preconfigured command in the PreconfiguredCommandEnum.
+ 2. Add a new entry in CommandRunner.commands_map that maps the preconfigured command to its corresponding function defined in the SpackCommandSequenceFactory class.
+ 4. Call the run_preconfigured_command_sequence method to execute the preconfigured command sequence.
+
+ Custom Command Sequences
+ To create and execute a custom command sequence, users must:
+ 1. Define the commands as a dictionary, where the key represents the command already registered in the command_registry, and the value contains the corresponding placeholder values expected by the command.
+ 2. Invoke the run_custom_command_sequence using generated dictionary to execute the custom command sequence.
+
+ Benefits
+ This module provides a flexible and unified interface for executing both predefined and custom command sequences, allowing users to easily manage and execute complex command workflows.
+ """
+
+ def __init__(self) -> None:
+ """Initializes the CommandRunner."""
+ self.logger: Logger = logging.getLogger(__name__)
+ self.executor: BashCommandExecutor = BashCommandExecutor()
+ self.commands_map: dict[PreconfiguredCommandEnum, Callable[..., CommandSequence]] = {
+ PreconfiguredCommandEnum.SPACK_COMPILER_FIND:
+ SpackCommandSequenceFactory.create_spack_compilers_command_sequence,
+ PreconfiguredCommandEnum.SPACK_COMPILER_LIST:
+ SpackCommandSequenceFactory.create_spack_compiler_list_command_sequence,
+ PreconfiguredCommandEnum.SPACK_COMPILERS:
+ SpackCommandSequenceFactory.create_spack_compiler_list_command_sequence,
+ PreconfiguredCommandEnum.SPACK_COMPILER_INFO:
+ SpackCommandSequenceFactory.create_spack_compiler_info_command_sequence,
+ PreconfiguredCommandEnum.SPACK_FIND:
+ SpackCommandSequenceFactory.create_spack_post_install_find_command_sequence,
+ PreconfiguredCommandEnum.SPACK_INSTALL:
+ SpackCommandSequenceFactory.create_spack_install_package_command_sequence,
+ PreconfiguredCommandEnum.SPACK_MIRROR_ADD:
+ SpackCommandSequenceFactory.create_spack_mirror_add_command_sequence,
+ PreconfiguredCommandEnum.SPACK_GPG_TRUST:
+ SpackCommandSequenceFactory.create_spack_gpg_trust_command_sequence,
+ }
+
+ def execute_command(self,
+ command_sequence: CommandSequence) -> dict[str, str | bool | None]:
+ """Executes a given command sequence.
+
+ Adds the command sequence to the executor and runs it, returning the result as a dictionary.
+
+ Args:
+ command_sequence (CommandSequence): The command sequence to execute.
+
+ Returns (dict[str, str | bool | None]):
+ A dictionary containing the execution result.
+ The dictionary has the following keys:
+ - success (bool): True if the command executed successfully, False otherwise.
+ - output (str | None): The output of the command if successful, None otherwise.
+ - error (str | None): The error message if the command failed, None otherwise.
+ """
+ self.executor.add_command(command_sequence)
+ output, error = self.executor.execute()
+ return {
+ "success": error is None,
+ "output": output.strip() if output else None,
+ "error": error
+ }
+
+ def run_preconfigured_command_sequence(self,
+ command_name: PreconfiguredCommandEnum,
+ *args: Any) -> dict[str, str | bool | None]:
+ """Runs a predefined command sequence.
+
+ Creates and executes a predefined command based on the given name and arguments.
+
+ For example `run_preconfigured_command_sequence(PreconfiguredCommandEnum.SPACK_COMPILER_FIND, 'gcc', '11')`
+ will execute the command `source spack_setup_path && spack find gcc@11`
+
+ Args:
+ command_name (PreconfiguredCommandEnum): The name of the predefined command sequence.
+ args (tuple(Any, ...)): Arguments to pass to the command constructor. The arguments should correspond to the flags, options, and placeholders of the command.
+
+ Returns:
+ A dictionary containing the execution result.
+ The dictionary has the following keys:
+ - success (bool): True if the command executed successfully, False otherwise.
+ - output (str | None): The output of the command if successful, None otherwise.
+ - error (str | None): The error message if the command failed, None otherwise.
+ If command_type is invalid, returns a dictionary with "success" as False and an error message.
+ """
+ if command_name not in self.commands_map:
+ return {"success": False, "error": f"Invalid command name: {command_name}"}
+
+ command_sequence = self.commands_map[command_name](*args)
+ return self.execute_command(command_sequence)
+
+ def run_custom_command_sequence(self,
+ command_placeholders_map: dict[CommandEnum, dict[str, str]]) \
+ -> dict[str, str | bool | None]:
+ """Runs a custom command sequence.
+
+ Creates and executes a custom command sequence from a map of command names to placeholder values.
+
+ Args:
+ command_placeholders_map (dict[CommandEnum, dict[str, str]]): A dictionary mapping command name enums to
+ placeholder values. The key is the command name in `CommandEnum` and the value is a dictionary
+ mapping placeholder names to their actual values.
+
+ For example, if the key command is `CommandEnum.FIND_IN_FILE`,
+ this corresponds to the command `grep -i {search_term} {file_path}` in CommandRegistry.COMMANDS.
+ Hence the placeholder_map_value should be `{search_term: 'python', file_path: '/path/to/file.txt'}`.
+
+ Returns (dict[str, str | bool | None]):
+ A dictionary containing the execution result.
+ The dictionary has the following keys:
+ - success (bool): True if the command executed successfully, False otherwise.
+ - output (str | None): The output of the command if successful, None otherwise.
+ - error (str | None): The error message if the command failed, None otherwise.
+ """
+ command_sequence = (CommandSequenceFactory
+ .create_custom_command_sequence(command_placeholders_map))
+ return self.execute_command(command_sequence)
diff --git a/esd/commands/command_sequence.py b/esd/commands/command_sequence.py
new file mode 100644
index 00000000..d0a3b469
--- /dev/null
+++ b/esd/commands/command_sequence.py
@@ -0,0 +1,47 @@
+""" Command Sequence module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_sequence.py
+# Description: Command Sequence abstraction
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from esd.commands.command import Command
+
+
+class CommandSequence(Command):
+ """Represents a sequence of executable commands.
+
+ Allows adding commands to a sequence and executing them in order,
+ combining/chaining their execution strings with "&&".
+ """
+
+ def __init__(self) -> None:
+ """Initializes an empty command sequence."""
+ self.commands: list[Command] = []
+
+ def add_command(self, command: Command) -> None:
+ """Adds a command to the sequence.
+
+ Appends the given command to the list of commands.
+
+ Args:
+ command (Command): The command to add.
+
+ Raises:
+ ValueError: If the provided command is not an instance of `Command`.
+ """
+ if not isinstance(command, Command):
+ raise ValueError("Command must be an instance of Command")
+ self.commands.append(command)
+
+ def execute(self) -> str:
+ """Executes the command sequence.
+
+ Executes each command in the sequence and joins their results with "&&".
+
+ Returns:
+ The combined execution string of all commands in the sequence.
+ """
+ return " && ".join(cmd.execute().strip() for cmd in self.commands).strip()
diff --git a/esd/commands/command_sequence_builder.py b/esd/commands/command_sequence_builder.py
new file mode 100644
index 00000000..5f1d2c3f
--- /dev/null
+++ b/esd/commands/command_sequence_builder.py
@@ -0,0 +1,71 @@
+""" Command Sequence Builder module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_sequence_builder.py
+# Description: Command sequence builder module for creating command sequences
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+from __future__ import annotations
+
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_registry import CommandRegistry
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.shell_command_factory import ShellCommandFactory
+
+
+class CommandSequenceBuilder:
+ """Builds a sequence of commands using the builder pattern.
+
+ Facilitates the creation of CommandSequence objects by adding commands incrementally
+ and then building the final sequence.
+ """
+
+ def __init__(self) -> None:
+ """Initializes a new CommandSequenceBuilder with an empty sequence."""
+ self.sequence: CommandSequence = CommandSequence()
+
+ def add_generic_command(self,
+ command_name: CommandEnum,
+ placeholders: dict[str, str]) -> CommandSequenceBuilder:
+ """Adds a generic command to the sequence.
+
+ Retrieves the command definition from the CommandRegistry, replaces placeholders with
+ provided values, creates a ShellCommand, and adds it to the sequence.
+
+ Args:
+ command_name (CommandEnum): The enum representing the command name to add.
+ placeholders (dict[str, str]): A dictionary of placeholder values to substitute in the command.
+
+ Returns:
+ The CommandSequenceBuilder instance (self) for method chaining.
+
+ Raises:
+ ValueError: If the command type is invalid or if the command is unknown.
+ """
+ if not isinstance(command_name, CommandEnum):
+ raise ValueError("Invalid command type. Use CommandEnum.")
+ command = CommandRegistry.get_command(command_name)
+
+ if command is None:
+ raise ValueError(f"Unknown command: {command_name}")
+ full_command = command[:] # Copy the command list to avoid mutation
+ # Replace placeholders with actual values
+ if placeholders:
+ full_command = [placeholders.get(arg.strip("{}"), arg) for arg in full_command]
+ full_command = list(filter(None, full_command))
+ shell_command = ShellCommandFactory.create_command(*full_command)
+ self.sequence.add_command(shell_command)
+ return self
+
+ def build(self) -> CommandSequence:
+ """Builds and returns the CommandSequence.
+
+ Returns the constructed CommandSequence and resets the builder for creating new sequences.
+
+ Returns:
+ The built CommandSequence.
+ """
+ sequence = self.sequence
+ self.sequence = CommandSequence() # Reset for next build
+ return sequence
diff --git a/esd/commands/command_sequence_factory.py b/esd/commands/command_sequence_factory.py
new file mode 100644
index 00000000..96804660
--- /dev/null
+++ b/esd/commands/command_sequence_factory.py
@@ -0,0 +1,45 @@
+""" Command Sequence Factory module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_sequence_factory.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.command_sequence_builder import CommandSequenceBuilder
+
+
+class CommandSequenceFactory:
+ """Factory for creating CommandSequence objects."""
+ @staticmethod
+ def create_custom_command_sequence(
+ command_placeholders_map: dict[CommandEnum, dict[str, str]]) -> CommandSequence:
+ """Creates a custom CommandSequence.
+
+ Builds a CommandSequence from a dictionary mapping CommandEnums to placeholder values.
+
+ For example if the key command is `CommandEnum.FIND_IN_FILE` and the value is `{search_term: 'python', file_path: '/path/to/file.txt'}`,
+ this corresponds to the command `grep -i {search_term} {file_path}` in CommandRegistry.COMMANDS. So the user can create a sequence of such commands.
+
+ e.g. command_placeholders_map = {
+ CommandEnum.FIND_IN_FILE: {
+ "search_term": "python",
+ "file_path": "/path/to/file.txt"
+ },
+ CommandEnum.SHOW_DIRECTORY: {},
+ CommandEnum.LIST_FILES: {"folder_location": "/tmp"},
+ CommandEnum.ECHO_MESSAGE: {"message": "Hello, world!"}
+ }
+
+ Args:
+ command_placeholders_map: A dictionary mapping CommandEnum members to dictionaries of placeholder values.
+ Returns:
+ A CommandSequence object representing the custom command sequence.
+ """
+ builder = CommandSequenceBuilder()
+ for command_type, placeholders in command_placeholders_map.items():
+ builder.add_generic_command(command_type, placeholders)
+ return builder.build()
diff --git a/esd/commands/generic_shell_command.py b/esd/commands/generic_shell_command.py
new file mode 100644
index 00000000..ea40079d
--- /dev/null
+++ b/esd/commands/generic_shell_command.py
@@ -0,0 +1,46 @@
+""" Generic Shell Command module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: generic_shell_command.py
+# Description: Generic shell command implementation
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from esd.commands.command import Command
+
+
+class GenericShellCommand(Command):
+ """Represents a generic shell command.
+
+ Encapsulates a shell command with its name and arguments, providing
+ a way to execute it.
+ """
+ def __init__(self, command_name: str, *args: str) -> None:
+ """Initializes a new GenericShellCommand.
+
+ Args:
+ command_name (str): The name of the command.
+ *args (str): The arguments for the command.
+
+ Raises:
+ ValueError: If the command name is empty, not a string, or if any of the arguments are not strings.
+ """
+ if not command_name:
+ raise ValueError("Command name is required!")
+ if not isinstance(command_name, str):
+ raise ValueError("Command name must be a string!")
+ if not all(isinstance(arg, str) for arg in args):
+ raise ValueError("All arguments must be strings!")
+ self.args: tuple[str, ...] = tuple(map(str.strip, args))
+ self.command_name: str = command_name.strip()
+
+ def execute(self) -> str:
+ """Executes the command.
+
+ Constructs and returns the full command string, including the command name and arguments.
+
+ Returns:
+ The full command string.
+ """
+ return f"{self.command_name} {' '.join(self.args)}" if self.args else self.command_name
diff --git a/esd/commands/preconfigured_command_enum.py b/esd/commands/preconfigured_command_enum.py
new file mode 100644
index 00000000..14b747ad
--- /dev/null
+++ b/esd/commands/preconfigured_command_enum.py
@@ -0,0 +1,28 @@
+""" Preconfigured Command Enum module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: command_enum.py
+# Description: Enumeration of supported predefined commands used in command_runner module:
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from enum import Enum
+
+
+class PreconfiguredCommandEnum(Enum):
+ """Enumeration of preconfigured composite commands.
+
+ Provides a predefined set of command identifiers for commonly used operations.
+
+ The command_runner module uses these identifiers to construct command sequences and execute them via 'run_preconfigured_command_sequence' method.
+ """
+ SPACK_COMPILER_FIND = "spack_compiler_find"
+ SPACK_COMPILERS = "spack_compilers"
+ SPACK_COMPILER_INFO = "spack_compiler_info"
+ SPACK_COMPILER_LIST = "spack_compiler_list"
+ SPACK_ENVIRONMENT_ACTIVATE = "spack_environment_activate"
+ SPACK_FIND = "spack_find"
+ SPACK_INSTALL = "spack_install"
+ SPACK_MIRROR_ADD = "spack_mirror_add"
+ SPACK_GPG_TRUST = "spack_gpg_trust"
diff --git a/esd/commands/shell_command_factory.py b/esd/commands/shell_command_factory.py
new file mode 100644
index 00000000..99baec39
--- /dev/null
+++ b/esd/commands/shell_command_factory.py
@@ -0,0 +1,33 @@
+""" Shell Command Factory module. """
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: shell_command_factory.py
+# Description: Shell command factory to create shell command instances
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from esd.commands.command import Command
+from esd.commands.generic_shell_command import GenericShellCommand
+
+
+class ShellCommandFactory:
+ """Factory for creating shell command instances.
+
+ Provides a static method for creating GenericShellCommand objects.
+ """
+
+ @staticmethod
+ def create_command(command_name: str, *args: str) -> Command:
+ """Creates a generic shell command.
+
+ Instantiates and returns a GenericShellCommand object with the given command name and arguments.
+
+ Args:
+ command_name (str): The name of the command.
+ *args (str): The arguments for the command.
+
+ Returns:
+ A GenericShellCommand object.
+ """
+ return GenericShellCommand(command_name, *args)
diff --git a/esd/commands/spack_command_sequence_factory.py b/esd/commands/spack_command_sequence_factory.py
new file mode 100644
index 00000000..e8d7e522
--- /dev/null
+++ b/esd/commands/spack_command_sequence_factory.py
@@ -0,0 +1,211 @@
+"""Factory for generating predefined spack related command sequences."""
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: spack_command_sequence_factory.py
+# Description: Factory for generating predefined spack related command sequences
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-17
+
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.command_sequence_builder import CommandSequenceBuilder
+
+
+def get_name_version(name: str | None, version: str | None) -> str:
+ """Formats a name and version string.
+
+ Returns a string combining the name and version with "@" if both are provided,
+ otherwise returns the name or an empty string.
+
+ Args:
+ name (str): The name.
+ version (str): The version.
+
+ Returns:
+ The formatted name and version string.
+ """
+ return f"{name}@{version}" if name and version else name or ""
+
+
+class SpackCommandSequenceFactory:
+ """Factory for creating Spack command sequences.
+
+ Provides methods for building CommandSequence objects for various Spack operations.
+ """
+
+ @staticmethod
+ def create_spack_enabled_command_sequence_builder(spack_setup_script: str) -> CommandSequenceBuilder:
+ """Creates a CommandSequenceBuilder with Spack setup.
+
+ Initializes a builder with the 'source' command for the given Spack setup script.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+
+ Returns:
+ A CommandSequenceBuilder pre-configured with the Spack setup command.
+ """
+ return (CommandSequenceBuilder()
+ .add_generic_command(CommandEnum.SOURCE, {"file_path": spack_setup_script}))
+
+ @staticmethod
+ def create_spack_compilers_command_sequence(spack_setup_script: str) -> CommandSequence:
+ """Creates a command sequence for listing Spack compilers.
+
+ Builds a sequence that sources the Spack setup script and then lists available compilers.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+
+ Returns:
+ A CommandSequence for listing Spack compilers.
+ """
+ return (SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ .add_generic_command(CommandEnum.SPACK_COMPILERS, {})
+ .build())
+
+ @staticmethod
+ def create_spack_compiler_list_command_sequence(spack_setup_script: str) -> CommandSequence:
+ """Creates a command sequence for listing Spack compilers.
+
+ Builds a sequence that sources the Spack setup script and then lists available compilers.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+
+ Returns:
+ A CommandSequence for listing Spack compilers.
+ """
+ return (SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ .add_generic_command(CommandEnum.SPACK_COMPILER_LIST, {})
+ .build())
+
+ @staticmethod
+ def create_spack_install_package_command_sequence(spack_setup_script: str,
+ package_name: str | None,
+ version: str | None) -> CommandSequence:
+ """Creates a command sequence for installing a Spack package.
+
+ Builds a sequence that sources the Spack setup script and then installs the specified package.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+ package_name (str | None): The name of the package to install.
+ version (str | None): The version of the package to install.
+
+ Returns:
+ A CommandSequence for installing the Spack package.
+ """
+ return (SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ .add_generic_command(CommandEnum.SPACK_INSTALL, {
+ "package_name_with_version": get_name_version(package_name, version)
+ }).build())
+
+ @staticmethod
+ def create_spack_mirror_add_command_sequence(spack_setup_script: str,
+ env_name: str,
+ mirror_name: str,
+ mirror_path: str,
+ autopush: bool = False,
+ signed: bool = False) -> CommandSequence:
+ """Creates a command sequence for adding a Spack mirror.
+
+ Builds a sequence that sources the Spack setup script, activates an environment (if specified),
+ and adds the given mirror.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+ env_name (str): The name of the environment to activate.
+ mirror_name (str): The name of the mirror.
+ mirror_path (str): The URL or path of the mirror.
+ autopush (bool): Whether to enable autopush for the mirror.
+ signed (bool): Whether to require signed packages from the mirror.
+
+ Returns:
+ A CommandSequence for adding the Spack mirror.
+ """
+ builder = SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ if env_name:
+ builder = builder.add_generic_command(CommandEnum.SPACK_ENVIRONMENT_ACTIVATE,
+ {"env_path": env_name})
+ place_holders = {
+ "mirror_name": mirror_name,
+ "mirror_path": mirror_path,
+ "autopush": "--autopush" if autopush else "",
+ "signed": "--signed" if signed else ""
+ }
+ builder = builder.add_generic_command(CommandEnum.SPACK_MIRROR_ADD, place_holders)
+ return builder.build()
+
+ @staticmethod
+ def create_spack_compiler_info_command_sequence(spack_setup_script: str,
+ compiler_name: str,
+ compiler_version: str) -> CommandSequence:
+ """Creates a command sequence for getting Spack compiler information.
+
+ Builds a sequence that sources the Spack setup script and retrieves information about the specified compiler.
+
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+ compiler_name (str): The name of the compiler.
+ compiler_version (str): The version of the compiler.
+
+ Returns:
+ A CommandSequence for getting compiler information.
+ """
+ return (SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ .add_generic_command(CommandEnum.SPACK_COMPILER_INFO,
+ {
+ "compiler_name_with_version":
+ get_name_version(compiler_name, compiler_version)
+ })
+ .build())
+
+ @staticmethod
+ def create_spack_post_install_find_command_sequence(spack_setup_script: str,
+ env_path: str,
+ package_name: str | None,
+ version: str | None) -> CommandSequence:
+ """Creates a command sequence for finding installed Spack packages after installation.
+
+ Builds a sequence that sources the Spack setup script, activates the specified environment,
+ and then searches for the given package.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+ env_path (str): The path to the Spack environment.
+ package_name (str | None): The name of the package to find.
+ version (str | None): The version of the package to find.
+
+ Returns:
+ A CommandSequence for finding installed Spack packages.
+ """
+ return (SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ .add_generic_command(CommandEnum.SPACK_ENVIRONMENT_ACTIVATE, {"env_path": env_path})
+ .add_generic_command(CommandEnum.SPACK_FIND,
+ {
+ "package_name_with_version":
+ get_name_version(package_name, version)
+ }).build())
+
+ @staticmethod
+ def create_spack_gpg_trust_command_sequence(spack_setup_script: str,
+ public_key_path: str) -> CommandSequence:
+ """Creates a command sequence for trusting a GPG key in Spack.
+
+ Builds a sequence that sources the Spack setup script and then trusts the given GPG key.
+
+ Args:
+ spack_setup_script (str): Path to the Spack setup script.
+ public_key_path (str): Path to the public key file.
+
+ Returns:
+ A CommandSequence for trusting the GPG key.
+ """
+ return (SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+ .add_generic_command(CommandEnum.SPACK_GPG_TRUST,
+ {
+ "public_key_path": public_key_path
+ }).build())
diff --git a/esd/configuration/SpackConfig.py b/esd/configuration/SpackConfig.py
index 93a2e874..26c6617f 100644
--- a/esd/configuration/SpackConfig.py
+++ b/esd/configuration/SpackConfig.py
@@ -8,10 +8,7 @@ class SpackConfig:
install_dir=Path(os.getcwd()).resolve(), upstream_instance=None, system_name=None,
concretization_dir: Path = None, buildcache_dir: Path = None):
self.env = env
- if repos is None:
- self.repos = []
- else:
- self.repos = repos
+ self.repos = [] if repos is None else repos
self.install_dir = install_dir
self.upstream_instance = upstream_instance
self.system_name = system_name
diff --git a/esd/spack_factory/SpackOperation.py b/esd/spack_factory/SpackOperation.py
index 904582f4..5a026283 100644
--- a/esd/spack_factory/SpackOperation.py
+++ b/esd/spack_factory/SpackOperation.py
@@ -2,13 +2,16 @@ import os
import re
import subprocess
from pathlib import Path
+
+from esd.commands.command_runner import CommandRunner
+from esd.commands.preconfigured_command_enum import PreconfiguredCommandEnum
+from esd.configuration.SpackConfig import SpackConfig
from esd.error_handling.exceptions import BashCommandException, NoSpackEnvironmentException, \
SpackInstallPackagesException, SpackConcertizeException, SpackMirrorException, SpackGpgException
from esd.logger.logger_builder import get_logger
-from esd.configuration.SpackConfig import SpackConfig
from esd.tests.testing_variables import SPACK_VERSION
-from esd.wrapper.spack_wrapper import check_spack_env
from esd.utils.utils import run_command, git_clone_repo, log_command, set_bashrc_variable
+from esd.wrapper.spack_wrapper import check_spack_env
class SpackOperation:
@@ -36,6 +39,7 @@ class SpackOperation:
self.spack_config.env.path.mkdir(parents=True, exist_ok=True)
self.env_path = spack_config.env.path / spack_config.env.env_name
self.spack_command_on_env = f'source {self.spack_setup_script} && spack env activate -p {self.env_path}'
+ self.command_runner = CommandRunner()
def create_fetch_spack_environment(self):
if self.spack_config.env.git_path:
@@ -109,9 +113,7 @@ class SpackOperation:
check=True,
capture_output=True, text=True, logger=self.logger,
info_msg=f'Checking if environment {self.spack_config.env.env_name} exists')
- if result is None:
- return False
- return True
+ return result is not None
@check_spack_env
def add_spack_repo(self, repo_path: Path, repo_name: str):
@@ -159,7 +161,7 @@ class SpackOperation:
run_command("bash", "-c",
f'{self.spack_command_on_env} && spack concretize {force}',
check=True,
- logger=self.logger,
+ logger=self.logger,
info_msg=f'Concertization step for {self.spack_config.env.env_name}',
exception_msg=f'Failed the concertization step for {self.spack_config.env.env_name}',
exception=SpackConcertizeException)
@@ -173,26 +175,104 @@ class SpackOperation:
exception_msg=f'Failed to create pgp keys mirror {self.spack_config.env.env_name}',
exception=SpackGpgException)
- def add_mirror(self, mirror_name: str, mirror_path: Path, signed=False, autopush=False, global_mirror=False):
- autopush = '--autopush' if autopush else ''
- signed = '--signed' if signed else ''
- if global_mirror:
- run_command("bash", "-c",
- f'source {self.spack_setup_script} && spack mirror add {autopush} {signed} {mirror_name} {mirror_path}',
- check=True,
- logger=self.logger,
- info_msg=f'Added mirror {mirror_name}',
- exception_msg=f'Failed to add mirror {mirror_name}',
- exception=SpackMirrorException)
- else:
- check_spack_env(
- run_command("bash", "-c",
- f'{self.spack_command_on_env} && spack mirror add {autopush} {signed} {mirror_name} {mirror_path}',
- check=True,
- logger=self.logger,
- info_msg=f'Added mirror {mirror_name}',
- exception_msg=f'Failed to add mirror {mirror_name}',
- exception=SpackMirrorException))
+ def add_mirror(self,
+ mirror_name: str,
+ mirror_path: str,
+ signed=False,
+ autopush=False,
+ global_mirror=False) -> bool:
+ """Adds a Spack mirror.
+
+ Adds a new mirror to the Spack configuration, either globally or to a specific environment.
+
+ Args:
+ mirror_name (str): The name of the mirror.
+ mirror_path (str): The path or URL of the mirror.
+ signed (bool): Whether to require signed packages from the mirror.
+ autopush (bool): Whether to enable autopush for the mirror.
+ global_mirror (bool): Whether to add the mirror globally (True) or to the current environment (False).
+
+ Returns:
+ True if the mirror was added successfully, False otherwise.
+
+ Raises:
+ ValueError: If mirror_name or mirror_path are empty.
+ NoSpackEnvironmentException: If global_mirror is False and no environment is defined.
+ """
+ if not mirror_name or not mirror_path:
+ raise ValueError("mirror_name and mirror_path are required")
+ if not global_mirror and not self.env_path:
+ raise NoSpackEnvironmentException('No spack environment defined')
+ result = self.command_runner.run_preconfigured_command_sequence(
+ PreconfiguredCommandEnum.SPACK_MIRROR_ADD,
+ self.spack_setup_script,
+ "" if global_mirror else self.env_path,
+ mirror_name,
+ mirror_path,
+ autopush,
+ signed)
+ return self.handle_result(
+ result,
+ "Failed to add mirror %s, reason: %s, output: %s",
+ (mirror_name,),
+ "Added mirror %s",
+ (mirror_name,)
+ )
+
+ def trust_gpg_key(self, public_key_path: str):
+ """Adds a GPG public key to the trusted keyring.
+
+ This method attempts to add the provided GPG public key to the
+ Spack trusted keyring.
+
+ Args:
+ public_key_path (str): Path to the GPG public key file.
+
+ Returns:
+ bool: True if the key was added successfully, False otherwise.
+
+ Raises:
+ ValueError: If public_key_path is empty.
+ """
+ if not public_key_path:
+ raise ValueError("public_key_path is required")
+ result = self.command_runner.run_preconfigured_command_sequence(
+ PreconfiguredCommandEnum.SPACK_GPG_TRUST,
+ self.spack_setup_script,
+ public_key_path)
+ return self.handle_result(
+ result,
+ "Failed to add public key %s as trusted, reason: %s, output: %s",
+ (public_key_path,),
+ "Added public key %s as trusted",
+ (public_key_path,),
+ )
+
+ def handle_result(self,
+ result: dict[str, str | bool | None],
+ error_msg: str,
+ error_msg_args: tuple[str, ...],
+ info_msg: str,
+ info_msg_args: tuple[str, ...]):
+ """Handles the result of a command execution.
+
+ Checks the success status of the result and logs either an error or an info message accordingly.
+
+ Args:
+ result (dict[str, str | bool | None]): A dictionary containing the result of the command execution.
+ error_msg (str): The error message to log if the command failed.
+ error_msg_args (tuple[str, ...]): Arguments to format the error message.
+ info_msg (str): The info message to log if the command succeeded.
+ info_msg_args (tuple[str, ...]): Arguments to format the info message.
+
+ Returns:
+ bool: True if the command succeeded, False otherwise.
+ """
+ if not result["success"]:
+ self.logger.error(error_msg, *error_msg_args, result['error'], result['output'])
+ return False
+ self.logger.info(info_msg, *info_msg_args)
+ return True
def remove_mirror(self, mirror_name: str):
run_command("bash", "-c",
@@ -242,7 +322,7 @@ class SpackOperation:
# ensure the file exists before opening it
if not os.path.exists(bashrc_path):
open(bashrc_path, "w").close()
- # add spack setup commands to .bashrc
+ # add spack setup commands_ to .bashrc
with open(bashrc_path, "a") as bashrc:
bashrc.write(f'export PATH="{self.spack_dir}/bin:$PATH"\n')
bashrc.write(f"source {self.spack_setup_script}\n")
diff --git a/esd/spack_factory/SpackOperationUseCache.py b/esd/spack_factory/SpackOperationUseCache.py
index 313522d2..b6e7846c 100644
--- a/esd/spack_factory/SpackOperationUseCache.py
+++ b/esd/spack_factory/SpackOperationUseCache.py
@@ -1,8 +1,10 @@
import os
+
from esd.build_cache.BuildCacheManager import BuildCacheManager
+from esd.configuration.SpackConfig import SpackConfig
+from esd.error_handling.exceptions import NoSpackEnvironmentException
from esd.logger.logger_builder import get_logger
from esd.spack_factory.SpackOperation import SpackOperation
-from esd.configuration.SpackConfig import SpackConfig
class SpackOperationUseCache(SpackOperation):
@@ -25,8 +27,31 @@ class SpackOperationUseCache(SpackOperation):
cache_version=cache_version_build)
def setup_spack_env(self):
+ """Sets up the Spack environment using cached data.
+
+ Downloads the build cache, trusts the cached public key (if available),
+ and adds the build cache as a local mirror.
+ """
super().setup_spack_env()
- # todo add buildcache to the spack environment
+ try:
+ # Download build cache from OCI Registry and add public key to trusted keys
+ self.build_cache.download(self.spack_config.buildcache_dir)
+ cached_public_key = self.build_cache.get_public_key_from_cache(self.spack_config.buildcache_dir)
+ signed = cached_public_key is not None and self.trust_gpg_key(cached_public_key)
+ if not signed:
+ self.logger.warning("Public key not found in cache or failed to trust pgp keys!")
+ # Add build cache mirror
+ self.add_mirror('local_cache',
+ str(self.spack_config.buildcache_dir),
+ signed=signed,
+ autopush=True,
+ global_mirror=False)
+ except (ValueError, NoSpackEnvironmentException) as e:
+ self.logger.error("Error adding buildcache mirror: %s", e)
+ raise
+
+
+
def concretize_spack_env(self, force=True):
pass
diff --git a/esd/tests/unit_tests/test_bash_command_executor.py b/esd/tests/unit_tests/test_bash_command_executor.py
new file mode 100644
index 00000000..70633fa2
--- /dev/null
+++ b/esd/tests/unit_tests/test_bash_command_executor.py
@@ -0,0 +1,232 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_bash_command_executor.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-18
+import os
+import subprocess
+from unittest.mock import patch
+
+import pytest
+
+from esd.commands.bash_command_executor import BashCommandExecutor
+from esd.commands.command import Command
+
+
+class MockCommand(Command):
+ def __init__(self, cmd: str):
+ self._cmd = cmd
+
+ def execute(self) -> str:
+ return self._cmd
+
+
+class TestBashCommandExecutor:
+ @pytest.mark.parametrize(
+ "test_id, os_name, expected_bash_command",
+ [
+ ("posix_system", "posix", ["bash", "-c"]),
+ ("nt_system", "nt", ["wsl", "bash", "-c"]),
+ ],
+ )
+ def test_init_success_path(self, mocker, test_id, os_name, expected_bash_command):
+ # Arrange
+ original_os = os.name
+ mock_get_logger = mocker.patch("esd.commands.bash_command_executor.logging.getLogger")
+ mocker.patch("esd.commands.bash_command_executor.os.name", os_name)
+ # mock_os_name.return_value = os_name
+
+ # Act
+ executor = BashCommandExecutor()
+ os.name = original_os # Reset the os.name to the original value
+
+ # Assert
+ assert executor.bash_command == expected_bash_command
+ mock_get_logger.assert_called_once_with('esd.commands.bash_command_executor')
+
+ @pytest.mark.parametrize(
+ "test_id, num_commands", [(1, 1), (5, 5), (0, 0)]
+ )
+ def test_add_command_success_path(self, mocker, test_id: int, num_commands: int):
+ # Arrange
+ executor = BashCommandExecutor()
+ executor.logger = mocker.MagicMock()
+ commands = [MockCommand("some_command") for _ in range(num_commands)]
+
+ # Act
+ for command in commands:
+ executor.add_command(command)
+
+ # Assert
+ assert len(executor.sequence.commands) == num_commands
+ executor.logger.info.assert_has_calls(
+ [mocker.call("Adding command to the sequence: %s", command) for command in commands])
+
+ def test_add_command_invalid_type(self):
+ # Arrange
+ executor = BashCommandExecutor()
+ invalid_command = "not a command object" # type: ignore
+
+ # Act
+ with pytest.raises(ValueError) as except_info:
+ executor.add_command(invalid_command)
+
+ # Assert
+ assert str(except_info.value) == "Invalid command type. Use Command."
+
+ @patch("esd.commands.bash_command_executor.os.name", "unknown")
+ def test_init_unknown_os(self):
+
+ # Act
+ executor = BashCommandExecutor()
+
+ # Assert
+ assert executor.bash_command == ["undefined"]
+
+ @pytest.mark.parametrize(
+ "test_id, commands, expected_output",
+ [
+ ("single_command", [MockCommand("echo hello")], "hello\n"),
+ ("multiple_commands", [MockCommand("echo hello"), MockCommand("echo world")], "hello\nworld\n"),
+ ("command_with_pipe", [MockCommand("echo hello | grep hello")], "hello\n"),
+
+ ],
+ )
+ @patch("esd.commands.bash_command_executor.subprocess.run")
+ def test_execute_success_path_posix(self, mock_subprocess_run, test_id, commands, expected_output, mocker):
+ # Arrange
+ executor = BashCommandExecutor()
+ executor.logger = mocker.MagicMock()
+ for cmd in commands:
+ executor.add_command(cmd)
+ mock_subprocess_run.return_value.stdout = expected_output
+
+ # Act
+ stdout, err = executor.execute()
+
+ # Assert
+ assert stdout == expected_output
+ assert err is None
+ executor.logger.info.assert_has_calls(
+ [mocker.call('Adding command to the sequence: %s', cmd) for cmd in commands] +
+ [mocker.call("Successfully executed command sequence, output: %s",
+ mock_subprocess_run.return_value.stdout)])
+
+ @patch("esd.commands.bash_command_executor.subprocess.run",
+ side_effect=FileNotFoundError("Mock file not found error"))
+ def test_execute_file_not_found_error(self, mock_subprocess_run):
+ # Arrange
+ executor = BashCommandExecutor()
+ executor.bash_command = ["bash", "-c"]
+ executor.add_command(MockCommand("some_command"))
+
+ # Act
+ stdout, err = executor.execute()
+
+ # Assert
+ assert stdout is None
+ assert err == "Error: Bash Command: ['bash', '-c'] not found: Mock file not found error"
+ mock_subprocess_run.assert_called_once_with(['bash', '-c', 'some_command'], capture_output=True, text=True,
+ check=True, timeout=172800)
+
+ @patch("esd.commands.bash_command_executor.subprocess.run",
+ side_effect=subprocess.CalledProcessError(1, "some_command", stderr="Mock stderr"))
+ def test_execute_called_process_error(self, mock_subprocess_run, mocker):
+ # Arrange
+ original_os = os.name
+ mocker.patch("esd.commands.bash_command_executor.os.name", "nt")
+ executor = BashCommandExecutor()
+ executor.add_command(MockCommand("failing_command"))
+
+ # Act
+ stdout, err = executor.execute()
+
+ # Assert
+ assert stdout is None
+ assert err == "Error: Command failed with exit code 1, Error Output: Mock stderr"
+ mock_subprocess_run.assert_called_once_with(['wsl', 'bash', '-c', 'failing_command'], capture_output=True,
+ text=True, check=True, timeout=172800)
+
+ # Cleanup
+ os.name = original_os
+
+ @pytest.mark.parametrize(
+ "test_id, exception, expected_error_message",
+ [
+ ("permission_error", PermissionError("Mock permission denied"),
+ "Error: Permission denied: Mock permission denied"),
+ ("os_error", OSError("Mock OS error"), "Error: OS error occurred: Mock OS error"),
+ ("value_error", ValueError("Mock invalid argument"),
+ "Error: Invalid argument passed: Mock invalid argument"),
+ ("type_error", TypeError("Mock invalid type"), "Error: Invalid type for arguments: Mock invalid type"),
+ ("timeout_expired", subprocess.TimeoutExpired("some_command", 10),
+ "Error: Command timed out after 10 seconds"),
+ ("subprocess_error", subprocess.SubprocessError("Mock subprocess error"),
+ "Subprocess error occurred: Mock subprocess error"),
+ ],
+ )
+ def test_execute_other_errors(self, test_id, exception, expected_error_message, mocker):
+ # Arrange
+ original_os = os.name
+ mocker.patch("esd.commands.bash_command_executor.os.name", "nt")
+ with patch("esd.commands.bash_command_executor.subprocess.run", side_effect=exception) as mock_subprocess_run:
+ executor = BashCommandExecutor()
+ executor.add_command(MockCommand("some_command"))
+
+ # Act
+ stdout, err = executor.execute()
+
+ # Assert
+ assert stdout is None
+ assert err == expected_error_message
+ mock_subprocess_run.assert_called_once_with(['wsl', 'bash', '-c', 'some_command'], capture_output=True,
+ text=True, check=True, timeout=172800)
+
+ # Cleanup
+ os.name = original_os
+
+ def test_execute_no_commands(self):
+ # Arrange
+ executor = BashCommandExecutor()
+
+ # Act
+ with pytest.raises(ValueError) as except_info:
+ executor.execute()
+
+ # Assert
+ assert str(except_info.value) == "No commands to execute."
+ @patch("esd.commands.bash_command_executor.subprocess.run")
+ def test_execute_happy_path_nt(self, mock_subprocess_run, mocker):
+ # Arrange
+ original_os = os.name
+
+ mocker.patch("esd.commands.bash_command_executor.os.name", "nt")
+ executor = BashCommandExecutor()
+ executor.add_command(MockCommand("echo hello"))
+ mock_subprocess_run.return_value.stdout = "hello\n"
+
+ # Act
+ stdout, err = executor.execute()
+
+ # Assert
+ assert stdout == "hello\n"
+ assert err is None
+ assert executor.bash_command == ['wsl', 'bash', '-c']
+ mock_subprocess_run.assert_called_once_with(['wsl', 'bash', '-c', 'echo hello'], capture_output=True, text=True,
+ check=True, timeout=172800)
+
+ # Cleanup
+ os.name = original_os
+
+ @patch("esd.commands.bash_command_executor.os.name", "unknown")
+ def test_execute_unknown_os(self):
+ # Arrange
+ executor = BashCommandExecutor()
+ executor.add_command(MockCommand("echo hello"))
+
+ # Act
+ assert executor.execute() == (None,
+ "Error: Bash Command: ['undefined'] not found: [WinError 2] The system cannot "
+ 'find the file specified')
diff --git a/esd/tests/unit_tests/test_build_cache_manager.py b/esd/tests/unit_tests/test_build_cache_manager.py
new file mode 100644
index 00000000..687570eb
--- /dev/null
+++ b/esd/tests/unit_tests/test_build_cache_manager.py
@@ -0,0 +1,155 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_build_cache_manager.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-20
+import pytest
+from _pytest.fixtures import fixture
+
+from esd.build_cache.BuildCacheManager import BuildCacheManager
+
+
+class TestBuildCacheManager:
+
+ @fixture(scope="function")
+ def mock_build_cache_manager(self, mocker):
+ mocker.patch("esd.build_cache.BuildCacheManager.get_logger")
+ return BuildCacheManager("TEST_HOST", "TEST_PROJECT", "TEST_USERNAME", "TEST_PASSWORD", "TEST_VERSION")
+
+ def test_get_public_key_from_cache_success_path(self, mock_build_cache_manager, tmp_path):
+
+ # Arrange
+ build_cache_dir = tmp_path / "build_cache"
+ pgp_folder = build_cache_dir / "project" / "_pgp"
+ pgp_folder.mkdir(parents=True)
+ key_file = pgp_folder / "key.pub"
+ key_file.write_text("public key content")
+
+ # Act
+ result = mock_build_cache_manager.get_public_key_from_cache(str(build_cache_dir))
+
+ # Assert
+ assert result == str(key_file)
+
+ @pytest.mark.parametrize("test_id, num_pgp_folders, num_key_files, expected_log_message", [
+ ("more_than_one_gpg_folder", 2, 1,
+ "More than one PGP folders found in the build cache: %s, using the first one in the list: %s"),
+ ("more_than_one_key_file", 1, 2,
+ "More than one PGP key files found in the build cache: %s, using the first one in the list: %s"),
+ ])
+ def test_get_public_key_from_cache_multiple_files_or_folders(self, mock_build_cache_manager, test_id,
+ tmp_path, num_pgp_folders,
+ num_key_files, expected_log_message):
+
+ # Arrange
+ pgp_folders = []
+ key_files = []
+ build_cache_dir = tmp_path / "build_cache"
+ for i in range(num_pgp_folders):
+ pgp_folder = build_cache_dir / f"project{i}" / "_pgp"
+ pgp_folders.append(str(pgp_folder))
+ pgp_folder.mkdir(parents=True)
+ for j in range(num_key_files):
+ key_file = pgp_folder / f"key{j}.pub"
+ key_files.append(str(key_file))
+ key_file.write_text(f"public key {j} content")
+
+ # Act
+ result = mock_build_cache_manager.get_public_key_from_cache(str(build_cache_dir))
+
+ # Assert
+ assert result == str(build_cache_dir / "project0" / "_pgp" / "key0.pub")
+ log = (expected_log_message, *pgp_folders, pgp_folders[0]) if test_id == "more_than_one_gpg_folder" else (
+ expected_log_message, *key_files, key_files[0])
+ mock_build_cache_manager._logger.warning.assert_called_once_with(*log)
+
+ @pytest.mark.parametrize("build_cache_dir, expected_log_message", [
+ (None, 'Build cache directory does not exist!'),
+ ("non_existent_dir", 'Build cache directory does not exist!'),
+ ])
+ def test_get_public_key_from_cache_no_build_cache(self, mock_build_cache_manager, build_cache_dir,
+ expected_log_message, tmp_path):
+
+ # Arrange
+ build_cache_dir = str(tmp_path / build_cache_dir) if build_cache_dir else None
+
+ # Act
+ result = mock_build_cache_manager.get_public_key_from_cache(build_cache_dir)
+
+ # Assert
+ assert result is None
+ mock_build_cache_manager._logger.warning.assert_called_once_with(expected_log_message)
+
+ # Assert
+ assert result is None
+ mock_build_cache_manager._logger.warning.assert_called_once_with(expected_log_message)
+
+ @pytest.mark.parametrize("build_cache_dir, expected_log_message", [
+ ("non_existent_dir", "No _pgp folder found in the build cache!"),
+ ])
+ def test_get_public_key_from_cache_no_pgp_folder(self, mock_build_cache_manager, build_cache_dir,
+ expected_log_message, tmp_path):
+
+ # Arrange
+ if build_cache_dir == "non_existent_dir":
+ build_cache_dir = tmp_path / build_cache_dir
+ build_cache_dir.mkdir(parents=True)
+
+ # Act
+ result = mock_build_cache_manager.get_public_key_from_cache(build_cache_dir)
+
+ # Assert
+ assert result is None
+ mock_build_cache_manager._logger.warning.assert_called_once_with(expected_log_message)
+
+ # Assert
+ assert result is None
+ mock_build_cache_manager._logger.warning.assert_called_once_with(expected_log_message)
+
+ def test_get_public_key_from_cache_empty_pgp_folder(self, mock_build_cache_manager, tmp_path):
+
+ # Arrange
+ build_cache_dir = tmp_path / "build_cache"
+ pgp_folder = build_cache_dir / "project" / "_pgp"
+ pgp_folder.mkdir(parents=True)
+
+ # Act
+ result = mock_build_cache_manager.get_public_key_from_cache(str(build_cache_dir))
+
+ # Assert
+ assert result is None
+ mock_build_cache_manager._logger.warning.assert_called_once_with("No PGP key files found in the build cache!")
+
+ @pytest.mark.parametrize("items, expected_log_message", [
+ (["item1", "item2"], "test message item1 item2 item1"),
+ (["item1", "item2", "item3"], "test message item1 item2 item3 item1"),
+ ])
+ def test_log_warning_if_needed_multiple_items(self, mock_build_cache_manager, items, expected_log_message):
+ # Test ID: multiple_items
+
+ # Arrange
+ warn_message = "test message"
+
+ # Act
+ mock_build_cache_manager.log_warning_if_needed(warn_message, items)
+
+ # Assert
+ mock_build_cache_manager._logger.warning.assert_called_once_with(warn_message, *items, items[0])
+
+ @pytest.mark.parametrize("items", [
+ [],
+ ["item1"],
+ ])
+ def test_log_warning_if_needed_no_warning(self, mock_build_cache_manager, items):
+ # Test ID: no_warning
+
+ # Arrange
+ warn_message = "test message"
+
+ # Act
+ mock_build_cache_manager.log_warning_if_needed(warn_message, items)
+
+ # Assert
+ mock_build_cache_manager._logger.warning.assert_not_called()
diff --git a/esd/tests/unit_tests/test_command.py b/esd/tests/unit_tests/test_command.py
new file mode 100644
index 00000000..8f957a35
--- /dev/null
+++ b/esd/tests/unit_tests/test_command.py
@@ -0,0 +1,45 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_command.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-18
+
+import pytest
+
+from esd.commands.command import Command
+
+class ConcreteCommand(Command):
+ def __init__(self, return_value: str):
+ self._return_value = return_value
+
+ def execute(self) -> str:
+ return self._return_value
+
+
+class TestCommand:
+ def test_execute_abstract_method(self):
+ # Act
+ with pytest.raises(TypeError):
+ Command() # type: ignore
+
+
+ @pytest.mark.parametrize(
+ "test_id, return_value",
+ [
+ ("empty_string", ""),
+ ("non_empty_string", "some_command"),
+ ("string_with_spaces", "command with spaces"),
+ ("non_ascii_chars", "αβγδ"),
+ ],
+ )
+ def test_execute_concrete_command(self, test_id, return_value):
+ # Arrange
+ command = ConcreteCommand(return_value)
+
+ # Act
+ result = command.execute()
+
+ # Assert
+ assert result == return_value
diff --git a/esd/tests/unit_tests/test_command_enum.py b/esd/tests/unit_tests/test_command_enum.py
new file mode 100644
index 00000000..57c4ec34
--- /dev/null
+++ b/esd/tests/unit_tests/test_command_enum.py
@@ -0,0 +1,38 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_command_enum.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-18
+
+import pytest
+
+from esd.commands.command_enum import CommandEnum
+
+class TestCommandEnum:
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, expected_value",
+ [
+ ("source", "SOURCE", "source"),
+ ("list_files", "LIST_FILES", "list_files"),
+ ("spack_install", "SPACK_INSTALL", "spack_install"),
+ ],
+ )
+ def test_command_enum_values(self, test_id, command_name, expected_value):
+
+ # Act
+ command = CommandEnum[command_name]
+
+ # Assert
+ assert command.value == expected_value
+ assert command.name == command_name
+
+
+ def test_command_enum_invalid_name(self):
+
+ # Act
+ with pytest.raises(KeyError):
+ CommandEnum["INVALID_COMMAND"] # type: ignore
+
diff --git a/esd/tests/unit_tests/test_command_runner.py b/esd/tests/unit_tests/test_command_runner.py
new file mode 100644
index 00000000..24ec4a93
--- /dev/null
+++ b/esd/tests/unit_tests/test_command_runner.py
@@ -0,0 +1,123 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_command_runner.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-18
+
+import pytest
+
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_runner import CommandRunner
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.preconfigured_command_enum import PreconfiguredCommandEnum
+
+
+class MockCommandSequence(CommandSequence):
+ def __init__(self, cmd_str: str):
+ self._cmd_str = cmd_str
+
+ def execute(self) -> str:
+ return self._cmd_str
+
+
+class TestCommandRunner:
+ @pytest.fixture(scope="function")
+ def mock_command_runner(self, mocker):
+ mocker.patch("esd.commands.command_runner.BashCommandExecutor")
+ mocker.patch("esd.commands.command_runner.logging.getLogger")
+ return CommandRunner()
+
+ @pytest.mark.parametrize(
+ "test_id, command_str, mock_output, expected_output",
+ [
+ ("simple_command", "echo hello", "hello\n", {"success": True, "output": "hello", "error": None}),
+ ("complex_command", "ls -l /tmp", "mock_output\n",
+ {"success": True, "output": "mock_output", "error": None}),
+ ("empty_output", "ls -l /tmp", "", {"success": True, "output": None, "error": None}),
+ ("command_with_error", "failing_command", "", {"success": False, "output": None, "error": "mock_error"}),
+
+ ],
+ )
+ def test_execute_command(self, mock_command_runner, test_id, command_str, mock_output, expected_output):
+ # Arrange
+ mock_command_runner.executor.execute.return_value = (
+ mock_output, "mock_error" if "failing" in command_str else None)
+ command_sequence = MockCommandSequence(command_str)
+
+ # Act
+ result = mock_command_runner.execute_command(command_sequence)
+
+ # Assert
+ assert result == expected_output
+
+ @pytest.mark.parametrize(
+ "test_id, command_type, args, expected_result",
+ [
+ (
+ "valid_command",
+ PreconfiguredCommandEnum.SPACK_COMPILER_FIND,
+ ["gcc"],
+ {"success": True, "output": "mock_output", "error": None},
+ ),
+ ],
+ )
+ def test_run_predefined_command_success_path(self, mock_command_runner, test_id, command_type, args,
+ expected_result):
+ # Arrange
+ mock_command_runner.executor.execute.return_value = ("mock_output\n", None)
+
+ # Act
+ result = mock_command_runner.run_preconfigured_command_sequence(command_type, *args)
+
+ # Assert
+ assert result == expected_result
+
+ def test_run_predefined_command_invalid_type(self, mock_command_runner):
+ # Arrange
+ invalid_command_type = "INVALID_COMMAND" # type: ignore
+
+ # Act
+ result = mock_command_runner.run_preconfigured_command_sequence(invalid_command_type)
+
+ # Assert
+ assert result == {"success": False, "error": "Invalid command name: INVALID_COMMAND"}
+
+ @pytest.mark.parametrize(
+ "test_id, command_placeholders_map, expected_result",
+ [
+ (
+ "single_command",
+ {CommandEnum.LIST_FILES: {"folder_location": "/tmp"}},
+ {"success": True, "output": "mock_output", "error": None},
+ ),
+ (
+ "multiple_commands",
+ {
+ CommandEnum.LIST_FILES: {"folder_location": "/tmp"},
+ CommandEnum.SHOW_DIRECTORY: {},
+ },
+ {"success": True, "output": "mock_output", "error": None},
+ ),
+ (
+ "empty_placeholders",
+ {CommandEnum.SHOW_DIRECTORY: {}},
+ {"success": True, "output": "mock_output", "error": None},
+ ),
+ ],
+ )
+ def test_run_custom_command(self, mocker, mock_command_runner, test_id, command_placeholders_map, expected_result):
+ # Arrange
+ mock_command_runner.execute_command = mocker.MagicMock(return_value=expected_result)
+ mock_create_custom_command_sequence = mocker.patch(
+ "esd.commands.command_runner.CommandSequenceFactory.create_custom_command_sequence")
+ mock_create_custom_command_sequence.return_value = MockCommandSequence("mock_command")
+
+ # Act
+ result = mock_command_runner.run_custom_command_sequence(command_placeholders_map)
+
+ # Assert
+ assert result == expected_result
+ mock_create_custom_command_sequence.assert_called_once_with(command_placeholders_map)
+ mock_command_runner.execute_command.assert_called_once_with(mock_create_custom_command_sequence.return_value)
diff --git a/esd/tests/unit_tests/test_command_sequence.py b/esd/tests/unit_tests/test_command_sequence.py
new file mode 100644
index 00000000..8b68e107
--- /dev/null
+++ b/esd/tests/unit_tests/test_command_sequence.py
@@ -0,0 +1,91 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_command_sequence.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+import pytest
+
+from esd.commands.command import Command
+from esd.commands.command_sequence import CommandSequence
+
+
+class MockCommand(Command):
+ def __init__(self, cmd_str: str):
+ self._cmd_str = cmd_str
+
+ def execute(self) -> str:
+ return self._cmd_str
+
+
+class TestCommandSequence:
+
+ @pytest.mark.parametrize(
+ "test_id, commands, expected_output",
+ [
+ ("single_command", ["echo hello"], "echo hello"),
+ ("single_command_with_spaces at_beginning", [" pwd"], "pwd"),
+ ("single_command_with_spaces at end", [" pwd "], "pwd"),
+ ("multiple_commands_with_spaces", [" echo hello", " ls -l /tmp "], "echo hello && ls -l /tmp"),
+ ("multiple_commands", ["echo hello", "ls -l /tmp"], "echo hello && ls -l /tmp"),
+ ("multiple_commands_with_spaces_in_between_and_end", ["echo hello ", "ls -l /tmp "], "echo hello && ls -l /tmp"),
+ ("empty_command", [""], ""),
+ ("commands_with_spaces", ["command with spaces", "another command"],
+ "command with spaces && another command"),
+ ],
+ )
+ def test_execute_success_path(self, test_id, commands, expected_output):
+ # Arrange
+ sequence = CommandSequence()
+ for cmd in commands:
+ sequence.add_command(MockCommand(cmd))
+
+ # Act
+ result = sequence.execute()
+
+ # Assert
+ assert result == expected_output
+
+ def test_execute_no_commands(self):
+ # Arrange
+ sequence = CommandSequence()
+
+ # Act
+ result = sequence.execute()
+
+ # Assert
+ assert result == ""
+
+ @pytest.mark.parametrize(
+ "test_id, num_commands",
+ [
+ ("one_command", 1),
+ ("multiple_commands", 5),
+ ("no_commands", 0),
+ ],
+ )
+ def test_add_command_success_path(self, test_id, num_commands):
+ # Arrange
+ sequence = CommandSequence()
+ commands = [MockCommand(f"command_{i}") for i in range(num_commands)]
+
+ # Act
+ for command in commands:
+ sequence.add_command(command)
+
+ # Assert
+ assert len(sequence.commands) == num_commands
+
+ def test_add_command_invalid_type(self):
+ # Arrange
+ sequence = CommandSequence()
+ invalid_command = "not a command object" # type: ignore
+
+ # Act
+ with pytest.raises(ValueError) as except_info:
+ sequence.add_command(invalid_command)
+
+ # Assert
+ assert str(except_info.value) == "Command must be an instance of Command"
diff --git a/esd/tests/unit_tests/test_command_sequence_builder.py b/esd/tests/unit_tests/test_command_sequence_builder.py
new file mode 100644
index 00000000..6c4062a4
--- /dev/null
+++ b/esd/tests/unit_tests/test_command_sequence_builder.py
@@ -0,0 +1,95 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_command_sequence_builder.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esd.commands.command import Command
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_registry import CommandRegistry
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.command_sequence_builder import CommandSequenceBuilder
+from esd.commands.shell_command_factory import ShellCommandFactory
+
+
+class TestCommandSequenceBuilder:
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, placeholders, expected_command",
+ [
+ ("no_placeholders", CommandEnum.SHOW_DIRECTORY, {}, ["pwd"]),
+ ("with_placeholders", CommandEnum.LIST_FILES, {"folder_location": "/tmp "}, ["ls", "-l", "/tmp"]),
+ ("missing_placeholders", CommandEnum.FIND_IN_FILE, {"search_term": "error"},
+ ["grep", "-i", "error", "{file_path}"]),
+ ("missing_placeholders_values_1", CommandEnum.SPACK_MIRROR_ADD, {"autopush": ""},
+ ["spack", "mirror", "add", "{signed}", "{mirror_name}","{mirror_path}"]),
+ ("missing_placeholders_values_2", CommandEnum.SPACK_MIRROR_ADD, {"autopush": "", "signed": ""},
+ ["spack", "mirror", "add", "{mirror_name}", "{mirror_path}"]),
+ ("missing_placeholders_values_3", CommandEnum.SPACK_MIRROR_ADD, {"autopush": "", "signed": "", "mirror_name": "test", "mirror_path": "test_path"},
+ ["spack", "mirror", "add", "test", "test_path"]),
+ ("extra_placeholders", CommandEnum.ECHO_MESSAGE, {"message": "hello", "extra": "world"}, ["echo", "hello"]),
+ ],
+ )
+ def test_add_generic_command_success_path(self, mocker, test_id, command_name,
+ placeholders, expected_command):
+ # Arrange
+ mock_get_command = mocker.patch.object(CommandRegistry, "get_command")
+ mock_create_command = mocker.patch.object(ShellCommandFactory, "create_command")
+ builder = CommandSequenceBuilder()
+ mock_get_command.return_value = expected_command
+ mock_create_command.return_value = Mock(spec=Command,
+ execute=lambda: " ".join(expected_command) if expected_command else "")
+
+ # Act
+ builder.add_generic_command(command_name, placeholders)
+
+ # Assert
+ mock_get_command.assert_called_once_with(command_name)
+ mock_create_command.assert_called_once_with(*expected_command)
+ assert len(builder.sequence.commands) == 1
+ assert builder.sequence.commands[0].execute() == " ".join(expected_command) if expected_command else ""
+
+ def test_add_generic_command_invalid_command_type(self):
+ # Arrange
+ builder = CommandSequenceBuilder()
+ invalid_command_name = "invalid" # type: ignore
+
+ # Act
+ with pytest.raises(ValueError) as except_info:
+ builder.add_generic_command(invalid_command_name, {})
+
+ # Assert
+ assert str(except_info.value) == "Invalid command type. Use CommandEnum."
+
+ def test_add_generic_command_unknown_command(self, mocker):
+ # Arrange
+ mock_get_command = mocker.patch.object(CommandRegistry, "get_command")
+ builder = CommandSequenceBuilder()
+ mock_get_command.return_value = None
+
+ # Act
+ with pytest.raises(ValueError) as except_info:
+ builder.add_generic_command(CommandEnum.LIST_FILES, {})
+
+ # Assert
+ assert str(except_info.value) == "Unknown command: CommandEnum.LIST_FILES"
+
+ def test_build(self):
+ # Arrange
+ builder = CommandSequenceBuilder()
+ builder.add_generic_command(CommandEnum.SHOW_DIRECTORY, {})
+
+ # Act
+ sequence = builder.build()
+
+ # Assert
+ assert isinstance(sequence, CommandSequence)
+ assert sequence.execute() == "pwd"
+ assert len(sequence.commands) == 1
+ assert len(builder.sequence.commands) == 0 # Check if the builder's sequence is reset
diff --git a/esd/tests/unit_tests/test_command_sequence_factory.py b/esd/tests/unit_tests/test_command_sequence_factory.py
new file mode 100644
index 00000000..7515f14a
--- /dev/null
+++ b/esd/tests/unit_tests/test_command_sequence_factory.py
@@ -0,0 +1,49 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_command_sequence_factory.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.command_sequence_builder import CommandSequenceBuilder
+from esd.commands.command_sequence_factory import CommandSequenceFactory
+
+
+class TestCommandSequenceFactory:
+
+ @pytest.mark.parametrize(
+ "test_id, command_placeholders_map, expected_calls",
+ [
+ ("single_command", {CommandEnum.SHOW_DIRECTORY: {}}, [(CommandEnum.SHOW_DIRECTORY, {})]),
+ (
+ "multiple_commands",
+ {CommandEnum.LIST_FILES: {"folder_location": "/tmp"}, CommandEnum.SHOW_DIRECTORY: {}},
+ [(CommandEnum.LIST_FILES, {"folder_location": "/tmp"}), (CommandEnum.SHOW_DIRECTORY, {})],
+ ),
+ ("no_commands", {}, []),
+ ],
+ )
+ def test_create_custom_command_sequence(self, mocker, test_id, command_placeholders_map, expected_calls):
+ # Arrange
+ mock_add_generic_command = mocker.patch.object(CommandSequenceBuilder, "add_generic_command")
+ mock_builder = Mock(spec=CommandSequenceBuilder, build=Mock(return_value=CommandSequence()))
+ mock_builder.add_generic_command = mock_add_generic_command
+ mock_add_generic_command.return_value = mock_builder
+
+ with patch("esd.commands.command_sequence_factory.CommandSequenceBuilder", return_value=mock_builder):
+ # Act
+ CommandSequenceFactory.create_custom_command_sequence(command_placeholders_map)
+
+ # Assert
+ assert mock_add_generic_command.call_count == len(expected_calls)
+ mock_add_generic_command.assert_has_calls(
+ [mocker.call(args, kwargs) for args, kwargs in expected_calls], any_order=True
+ )
+ mock_builder.build.assert_called_once()
diff --git a/esd/tests/unit_tests/test_generic_shell_command.py b/esd/tests/unit_tests/test_generic_shell_command.py
new file mode 100644
index 00000000..e9287e04
--- /dev/null
+++ b/esd/tests/unit_tests/test_generic_shell_command.py
@@ -0,0 +1,63 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_generic_shell_command.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+import pytest
+
+from esd.commands.command import Command
+from esd.commands.generic_shell_command import GenericShellCommand
+
+class TestGenericShellCommand:
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, args, expected_output",
+ [
+ ("no_args", "ls", [], "ls"),
+ ("single_arg", "echo", ["hello"], "echo hello"),
+ ("multiple_args", "grep", ["-i", "error", "file.txt"], "grep -i error file.txt"),
+ ("command_with_spaces", " ls ", ["-l", "/tmp"], "ls -l /tmp"),
+ ("args_with_spaces", "ls", [" -l ", " /tmp "], "ls -l /tmp"),
+ ("command_and_args_with_spaces", " ls ", [" -l ", " /tmp "], "ls -l /tmp"),
+ ("empty_args", "ls", [""], "ls "), # Empty arguments are preserved, but stripped
+ ],
+ )
+ def test_execute_success_path(self, test_id, command_name, args, expected_output):
+
+ # Act
+ command = GenericShellCommand(command_name, *args)
+ result = command.execute()
+
+ # Assert
+ assert result == expected_output
+
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, args, expected_error",
+ [
+ ("empty_command_name", "", [], ValueError),
+ ("none_command_name", None, [], ValueError), # type: ignore
+ ("int_command_name", 123, [], ValueError), # type: ignore
+ ("invalid_arg_type", "ls", [123], ValueError), # type: ignore
+ ("mixed_arg_types", "ls", ["valid", 456], ValueError), # type: ignore
+ ],
+ )
+ def test_init_error_cases(self, test_id, command_name, args, expected_error):
+
+ # Act
+ with pytest.raises(expected_error) as except_info:
+ GenericShellCommand(command_name, *args)
+
+ # Assert
+ if expected_error is ValueError:
+ if not command_name:
+ assert str(except_info.value) == "Command name is required!"
+ elif not isinstance(command_name, str):
+ assert str(except_info.value) == "Command name must be a string!"
+ else:
+ assert str(except_info.value) == "All arguments must be strings!"
+
+
diff --git a/esd/tests/unit_tests/test_preconfigured_command_enum.py b/esd/tests/unit_tests/test_preconfigured_command_enum.py
new file mode 100644
index 00000000..02539d55
--- /dev/null
+++ b/esd/tests/unit_tests/test_preconfigured_command_enum.py
@@ -0,0 +1,37 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_preconfigured_command_enum.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+import pytest
+
+from esd.commands.preconfigured_command_enum import PreconfiguredCommandEnum
+
+class TestPreconfiguredCommandEnum:
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, expected_value",
+ [
+ ("spack_compiler_find", "SPACK_COMPILER_FIND", "spack_compiler_find"),
+ ("spack_compilers", "SPACK_COMPILERS", "spack_compilers"),
+ ("spack_install", "SPACK_INSTALL", "spack_install"),
+ ],
+ )
+ def test_preconfigured_command_enum_values(self, test_id, command_name, expected_value):
+
+ # Act
+ command = PreconfiguredCommandEnum[command_name]
+
+ # Assert
+ assert command.value == expected_value
+ assert command.name == command_name
+
+
+ def test_preconfigured_command_enum_invalid_name(self):
+
+ # Act
+ with pytest.raises(KeyError):
+ PreconfiguredCommandEnum["INVALID_COMMAND"] # type: ignore
diff --git a/esd/tests/unit_tests/test_shell_command_factory.py b/esd/tests/unit_tests/test_shell_command_factory.py
new file mode 100644
index 00000000..f2f95b90
--- /dev/null
+++ b/esd/tests/unit_tests/test_shell_command_factory.py
@@ -0,0 +1,61 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_shell_command_factory.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+import pytest
+
+from esd.commands.generic_shell_command import GenericShellCommand
+from esd.commands.shell_command_factory import ShellCommandFactory
+
+
+class TestShellCommandFactory:
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, args",
+ [
+ ("no_args", "ls", []),
+ ("single_arg", "echo", ["hello"]),
+ ("multiple_args", "grep", ["-i", "error", "file.txt"]),
+ ("command_with_spaces", " ls ", ["-l", "/tmp"]),
+ ("args_with_spaces", "ls", [" -l ", " /tmp "]),
+ ("command_and_args_with_spaces", " ls ", [" -l ", " /tmp "]),
+ ],
+ )
+ def test_create_command_happy_path(self, test_id, command_name, args):
+
+ # Act
+ command = ShellCommandFactory.create_command(command_name, *args)
+
+ # Assert
+ assert isinstance(command, GenericShellCommand)
+ assert command.command_name == command_name.strip()
+ assert command.args == tuple(map(str.strip, args))
+
+ @pytest.mark.parametrize(
+ "test_id, command_name, args, expected_error",
+ [
+ ("empty_command_name", "", [], ValueError),
+ ("none_command_name", None, [], ValueError), # type: ignore
+ ("int_command_name", 123, [], ValueError), # type: ignore
+ ("invalid_arg_type", "ls", [123], ValueError), # type: ignore
+ ("mixed_arg_types", "ls", ["valid", 456], ValueError), # type: ignore
+ ],
+ )
+ def test_create_command_error_cases(self, test_id, command_name, args, expected_error):
+
+ # Act
+ with pytest.raises(expected_error) as except_info:
+ ShellCommandFactory.create_command(command_name, *args)
+
+ # Assert
+ if expected_error is ValueError:
+ if not command_name:
+ assert str(except_info.value) == "Command name is required!"
+ elif not isinstance(command_name, str):
+ assert str(except_info.value) == "Command name must be a string!"
+ else:
+ assert str(except_info.value) == "All arguments must be strings!"
diff --git a/esd/tests/unit_tests/test_spack_command_sequence_factory.py b/esd/tests/unit_tests/test_spack_command_sequence_factory.py
new file mode 100644
index 00000000..9b6bc572
--- /dev/null
+++ b/esd/tests/unit_tests/test_spack_command_sequence_factory.py
@@ -0,0 +1,159 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_spack_command_sequence_factory.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+import pytest
+
+from esd.commands.command_enum import CommandEnum
+from esd.commands.command_sequence import CommandSequence
+from esd.commands.command_sequence_builder import CommandSequenceBuilder
+from esd.commands.spack_command_sequence_factory import SpackCommandSequenceFactory
+
+
+class TestSpackCommandSequenceFactory:
+
+ @pytest.mark.parametrize(
+ "test_id, spack_setup_script",
+ [
+ ("with_setup_script", "/path/to/setup.sh"),
+ ("empty_setup_script", "")],
+ )
+ def test_create_spack_enabled_command_sequence_builder(self, test_id, spack_setup_script):
+ # Act
+ builder = SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder(spack_setup_script)
+
+ # Assert
+ assert isinstance(builder, CommandSequenceBuilder)
+ assert len(builder.sequence.commands) == 1
+ assert builder.sequence.commands[0].command_name == "source"
+ if spack_setup_script:
+ assert builder.sequence.commands[0].args == ("/path/to/setup.sh",)
+ else:
+ assert len(builder.sequence.commands[0].args) == 0
+
+ @pytest.mark.parametrize(
+ "test_id, method_name, args, kwargs, expected_command, expected_args",
+ [
+ (
+ "spack_compilers", "create_spack_compilers_command_sequence", ["/path/to/setup.sh"], {}, "spack",
+ ["compilers"]
+ ),
+ (
+ "spack_compiler_list", "create_spack_compiler_list_command_sequence", ["/path/to/setup.sh"], {},
+ "spack", ["compiler", "list"]
+ ),
+ (
+ "spack_install_with_version", "create_spack_install_package_command_sequence",
+ ["/path/to/setup.sh", "package", "1.0"], {}, "spack", ["install", "package@1.0"]
+ ),
+ (
+ "spack_install_no_version", "create_spack_install_package_command_sequence",
+ ["/path/to/setup.sh", "package", None], {}, "spack", ["install", "package"]
+ ),
+ (
+ "spack_install_no_package", "create_spack_install_package_command_sequence",
+ ["/path/to/setup.sh", None, "1.0"], {}, "spack", ["install"]
+ ),
+ (
+ "spack_compiler_info", "create_spack_compiler_info_command_sequence",
+ ["/path/to/setup.sh", "gcc", "10.2"], {}, "spack", ["compiler", "info", "gcc@10.2"]
+ ),
+ ],
+ )
+ def test_create_command_sequence(self, test_id, method_name, args, kwargs, expected_command, expected_args):
+ # Arrange
+ method = getattr(SpackCommandSequenceFactory, method_name)
+
+ # Act
+ sequence = method(*args, **kwargs)
+
+ # Assert
+ assert isinstance(sequence, CommandSequence)
+ assert len(sequence.commands) == 2
+ assert sequence.commands[1].command_name == expected_command
+ assert sequence.commands[1].args == tuple(expected_args)
+
+ @pytest.mark.parametrize(
+ "test_id, env_name, mirror_name, mirror_path, autopush, signed, expected_length, expected_autopush, expected_signed, expected_output",
+ [
+ ("no_env", "", "mymirror", "/path/to/mirror", False, False, 2, "", "",
+ ("mirror", "add", "mymirror", "/path/to/mirror")),
+ ("with_env", "myenv", "mymirror", "/path/to/mirror", True, True, 3, "--autopush", "--signed",
+ ("mirror", "add", "--autopush", "--signed", "mymirror", "/path/to/mirror")),
+ ],
+ )
+ def test_create_spack_mirror_add_command_sequence(self, test_id, env_name, mirror_name, mirror_path, autopush,
+ signed, expected_length, expected_autopush, expected_signed,
+ expected_output):
+ # Arrange
+ spack_setup_script = "/path/to/setup.sh"
+
+ # Act
+ sequence = SpackCommandSequenceFactory.create_spack_mirror_add_command_sequence(
+ spack_setup_script, env_name, mirror_name, mirror_path, autopush, signed
+ )
+
+ # Assert
+ assert isinstance(sequence, CommandSequence)
+ assert len(sequence.commands) == expected_length
+ assert sequence.commands[-1].command_name == "spack"
+ assert sequence.commands[-1].args == expected_output
+
+ @pytest.mark.parametrize(
+ "test_id, package_name, version, expected_package_arg",
+ [
+ ("with_package_and_version", "mypackage", "1.2.3", "mypackage@1.2.3"),
+ ("only_package_name", "mypackage", None, "mypackage"),
+ ("no_package", None, "1.2.3", ""),
+ ],
+ )
+ def test_create_spack_post_install_find_command_sequence(self, test_id, package_name, version,
+ expected_package_arg):
+ # Arrange
+ spack_setup_script = "/path/to/setup.sh"
+ env_path = "/path/to/env"
+
+ # Act
+ sequence = SpackCommandSequenceFactory.create_spack_post_install_find_command_sequence(
+ spack_setup_script, env_path, package_name, version
+ )
+
+ # Assert
+ assert isinstance(sequence, CommandSequence)
+ assert len(sequence.commands) == 3
+ assert sequence.commands[1].command_name == "spack"
+ assert sequence.commands[1].args == ("env", "activate", "-p", env_path)
+ assert sequence.commands[2].command_name == "spack"
+ assert sequence.commands[2].args == ("find", expected_package_arg) if expected_package_arg else ("find",)
+
+ @pytest.mark.parametrize("spack_setup_script, public_key_path", [
+ ("/path/to/setup-env.sh", "key.gpg"),
+ ("./setup-env.sh", "path/to/key.gpg"),
+ ])
+ def test_create_spack_gpg_trust_command_sequence(self, mocker, spack_setup_script, public_key_path):
+ # Test ID: create_spack_gpg_trust_command_sequence
+
+ # Arrange
+ mock_command_sequence_builder = mocker.MagicMock()
+ mocker.patch.object(SpackCommandSequenceFactory, "create_spack_enabled_command_sequence_builder",
+ return_value=mock_command_sequence_builder)
+
+ mock_command_sequence = mocker.MagicMock()
+ mock_command_sequence_builder.add_generic_command.return_value = mock_command_sequence_builder
+ mock_command_sequence_builder.build.return_value = mock_command_sequence
+
+ # Act
+ result = SpackCommandSequenceFactory.create_spack_gpg_trust_command_sequence(spack_setup_script,
+ public_key_path)
+
+ # Assert
+ assert result == mock_command_sequence
+ SpackCommandSequenceFactory.create_spack_enabled_command_sequence_builder.assert_called_once_with(
+ spack_setup_script)
+ mock_command_sequence_builder.add_generic_command.assert_called_once_with(CommandEnum.SPACK_GPG_TRUST,
+ {"public_key_path": public_key_path})
+ mock_command_sequence_builder.build.assert_called_once()
diff --git a/esd/tests/unit_tests/test_spack_operation.py b/esd/tests/unit_tests/test_spack_operation.py
new file mode 100644
index 00000000..4e2c520e
--- /dev/null
+++ b/esd/tests/unit_tests/test_spack_operation.py
@@ -0,0 +1,209 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_spack_operation.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-19
+
+import pytest
+from _pytest.fixtures import fixture
+
+from esd.build_cache.BuildCacheManager import BuildCacheManager
+from esd.commands.preconfigured_command_enum import PreconfiguredCommandEnum
+from esd.error_handling.exceptions import NoSpackEnvironmentException
+from esd.spack_factory.SpackOperation import SpackOperation
+
+
+class TestSpackOperationAddMirrorWithComposite:
+
+ @fixture
+ def mock_spack_operation(self, mocker):
+ mocker.resetall()
+ mocker.patch("esd.spack_factory.SpackOperation.CommandRunner")
+ mocker.patch("esd.spack_factory.SpackOperation.get_logger")
+ mock_object = SpackOperation()
+ mock_object.logger = mocker.MagicMock()
+ return mock_object
+
+ @pytest.mark.parametrize(
+ "test_id, global_mirror, env_path, mirror_name, mirror_path, autopush, signed, expected_result, expected_env_arg",
+ [
+ ("global_mirror", True, None, "global_mirror", "/path/to/global/mirror", False, False, True, ""),
+ ("local_mirror", False, "/path/to/env", "local_mirror", "/path/to/local/mirror", True, True, True,
+ "/path/to/env"),
+ ],
+ )
+ def test_add_mirror_with_composite_success_path(self, mock_spack_operation, test_id, global_mirror, env_path,
+ mirror_name, mirror_path, autopush, signed, expected_result,
+ expected_env_arg):
+ # Arrange
+ mock_spack_operation.spack_setup_script = "setup.sh"
+ mock_spack_operation.env_path = env_path
+ mock_spack_operation.command_runner.run_preconfigured_command_sequence.return_value = {"success": True,
+ "error": None,
+ "output": None}
+
+ # Act
+ result = mock_spack_operation.add_mirror(mirror_name, mirror_path, signed, autopush,
+ global_mirror)
+
+ # Assert
+ assert result == expected_result
+ mock_spack_operation.command_runner.run_preconfigured_command_sequence.assert_called_once_with(
+ PreconfiguredCommandEnum.SPACK_MIRROR_ADD,
+ "setup.sh",
+ expected_env_arg,
+ mirror_name,
+ mirror_path,
+ autopush,
+ signed,
+ )
+ mock_spack_operation.logger.info.assert_called_once_with("Added mirror %s", mirror_name)
+
+ @pytest.mark.parametrize(
+ "test_id, mirror_name, mirror_path, expected_error",
+ [
+ ("missing_mirror_name", "", "/path/to/mirror", ValueError),
+ ("missing_mirror_path", "mirror_name", "", ValueError),
+ ("missing_both", "", "", ValueError),
+ ],
+ )
+ def test_add_mirror_with_composite_missing_args(self, mock_spack_operation, test_id, mirror_name, mirror_path,
+ expected_error):
+ # Arrange
+ mock_spack_operation.spack_setup_script = "setup.sh"
+
+ # Act
+ with pytest.raises(expected_error) as except_info:
+ mock_spack_operation.add_mirror(mirror_name, mirror_path)
+
+ # Assert
+ assert str(except_info.value) == "mirror_name and mirror_path are required"
+
+ def test_add_mirror_with_composite_no_env(self, mock_spack_operation):
+ # Arrange
+ mock_spack_operation.spack_setup_script = "setup.sh"
+ mock_spack_operation.env_path = None
+
+ # Act
+ with pytest.raises(NoSpackEnvironmentException) as except_info:
+ mock_spack_operation.add_mirror("mirror_name", "/path/to/mirror")
+
+ # Assert
+ assert str(except_info.value) == "No spack environment defined"
+
+ def test_add_mirror_with_composite_command_failure(self, mock_spack_operation):
+ # Arrange
+ mock_spack_operation.spack_setup_script = "setup.sh"
+ mock_spack_operation.env_path = "/path/to/env"
+ mock_spack_operation.command_runner.run_preconfigured_command_sequence.return_value = {"success": False,
+ "error": "Error: Command failed with exit code 1, Error Output: ==> Error: Mirror with name mirror_name already exists.",
+ "output": None}
+
+ # Act
+ result = mock_spack_operation.add_mirror("mirror_name", "/path/to/mirror")
+
+ # Assert
+ assert result is False
+ mock_spack_operation.command_runner.run_preconfigured_command_sequence.assert_called_once_with(
+ PreconfiguredCommandEnum.SPACK_MIRROR_ADD,
+ "setup.sh",
+ "/path/to/env",
+ "mirror_name",
+ "/path/to/mirror",
+ False,
+ False
+ )
+ mock_spack_operation.logger.error.assert_called_once_with('Failed to add mirror %s, reason: %s, output: %s',
+ 'mirror_name',
+ 'Error: Command failed with exit code 1, Error Output: ==> Error: Mirror with '
+ 'name mirror_name already exists.',
+ None)
+
+ @pytest.mark.parametrize("result, expected_log_message, expected_return_value", [
+ ({"success": True, "output": "test output"}, "test info message", True),
+ ({"success": True, "error": "test error", "output": "test output"}, "test info message", True),
+ ])
+ def test_handle_result_success(self, mock_spack_operation, result, expected_log_message, expected_return_value):
+ # Test ID: success
+
+ # Arrange
+ error_msg = "test error message"
+ error_msg_args = ("arg1", "arg2")
+ info_msg = "test info message"
+ info_msg_args = ("arg3", "arg4")
+
+ # Act
+ return_value = mock_spack_operation.handle_result(result, error_msg, error_msg_args, info_msg, info_msg_args)
+
+ # Assert
+ assert return_value == expected_return_value
+ mock_spack_operation.logger.info.assert_called_once_with(info_msg, *info_msg_args)
+ mock_spack_operation.logger.error.assert_not_called()
+
+ @pytest.mark.parametrize("result, expected_log_message", [
+ ({"success": False, "error": "test error", "output": "test output"},
+ "test error message arg1 arg2 test error test output"),
+ ({"success": False, "error": None, "output": None}, "test error message arg1 arg2 None None"),
+ ])
+ def test_handle_result_failure(self, mock_spack_operation, result, expected_log_message):
+ # Test ID: failure
+
+ # Arrange
+ error_msg = "test error message"
+ error_msg_args = ("arg1", "arg2")
+ info_msg = "test info message"
+ info_msg_args = ("arg3", "arg4")
+
+ # Act
+ return_value = mock_spack_operation.handle_result(result, error_msg, error_msg_args, info_msg, info_msg_args)
+
+ # Assert
+ assert return_value is False
+ mock_spack_operation.logger.error.assert_called_once_with(error_msg, *error_msg_args, result["error"],
+ result["output"])
+ mock_spack_operation.logger.info.assert_not_called()
+
+ @pytest.mark.parametrize("public_key_path, result_success, expected_log_message", [
+ ("test_key.gpg", True, ('Added public key %s as trusted', 'test_key.gpg')),
+ ("test_key.gpg", False,
+ ('Failed to add public key %s as trusted, reason: %s, output: %s',
+ 'test_key.gpg',
+ 'test_error',
+ 'test_output')),
+ ])
+ def test_trust_gpg_key(self, mock_spack_operation, public_key_path, result_success, expected_log_message):
+ # Test ID: trust_gpg_key
+
+ # Arrange
+ mock_result = {"success": result_success, "error": "test_error", "output": "test_output"}
+ mock_spack_operation.command_runner.run_preconfigured_command_sequence.return_value = mock_result
+
+ # Act
+ if result_success:
+ result = mock_spack_operation.trust_gpg_key(public_key_path)
+ else:
+ result = mock_spack_operation.trust_gpg_key(public_key_path)
+
+ # Assert
+ if result_success:
+ assert result is True
+ mock_spack_operation.logger.info.assert_called_once_with(*expected_log_message)
+ else:
+ assert result is False
+ mock_spack_operation.logger.error.assert_called_once_with(*expected_log_message)
+
+ mock_spack_operation.command_runner.run_preconfigured_command_sequence.assert_called_once_with(
+ PreconfiguredCommandEnum.SPACK_GPG_TRUST,
+ mock_spack_operation.spack_setup_script,
+ public_key_path
+ )
+
+ def test_trust_gpg_key_empty_path(self, mock_spack_operation):
+ # Test ID: empty_path
+
+ # Act & Assert
+ with pytest.raises(ValueError) as except_info:
+ mock_spack_operation.trust_gpg_key("")
+ assert str(except_info.value) == "public_key_path is required"
diff --git a/esd/tests/unit_tests/test_spack_operation_use_cache.py b/esd/tests/unit_tests/test_spack_operation_use_cache.py
new file mode 100644
index 00000000..fd4e2778
--- /dev/null
+++ b/esd/tests/unit_tests/test_spack_operation_use_cache.py
@@ -0,0 +1,90 @@
+# Copyright (c) 2025
+# License Information: To be decided!
+#
+# File: test_spack_operation_use_cache.py
+# Description: Brief description of the file.
+# Created by: Murugan, Jithu <j.murugan@fz-juelich.de>
+# Created on: 2025-02-20
+
+from pathlib import Path
+
+import pytest
+
+from esd.error_handling.exceptions import NoSpackEnvironmentException
+from esd.spack_factory.SpackOperationUseCache import SpackOperationUseCache
+
+
+@pytest.fixture
+def spack_operation_use_cache_mock(mocker):
+ super_mock = mocker.patch("esd.spack_factory.SpackOperationUseCache.super")
+ super_mock.return_value.setup_spack_env = mocker.MagicMock()
+ mocker.patch("esd.spack_factory.SpackOperationUseCache.BuildCacheManager")
+ mock_spack_operation_use_cache = SpackOperationUseCache()
+ mock_spack_operation_use_cache.build_cache = mocker.MagicMock()
+ mock_spack_operation_use_cache.spack_config = mocker.MagicMock()
+ mock_spack_operation_use_cache.spack_config.buildcache_dir = Path("path/to/buildcache")
+ mock_spack_operation_use_cache.logger = mocker.MagicMock()
+ return mock_spack_operation_use_cache
+
+
+class TestSpackOperationUseCache:
+
+ @pytest.mark.parametrize("test_id, signed, key_path", [
+ ("key_path_exists", True, "path/to/key.gpg"),
+ ("key_path_does_not_exist", False, None)])
+ def test_setup_spack_env(self, mocker, spack_operation_use_cache_mock, test_id, signed, key_path):
+ # Test ID: setup_spack_env_success
+ super_mock = mocker.patch("esd.spack_factory.SpackOperationUseCache.super")
+ spack_operation_use_cache_mock.trust_gpg_key = mocker.MagicMock()
+ spack_operation_use_cache_mock.add_mirror = mocker.MagicMock()
+
+ # Arrange
+ spack_operation_use_cache_mock.build_cache.get_public_key_from_cache.return_value = key_path
+ spack_operation_use_cache_mock.trust_gpg_key.return_value = signed
+ spack_operation_use_cache_mock.add_mirror.return_value = None
+
+ # Act
+ spack_operation_use_cache_mock.setup_spack_env()
+
+ # Assert
+ spack_operation_use_cache_mock.build_cache.download.assert_called_once_with(
+ spack_operation_use_cache_mock.spack_config.buildcache_dir)
+ spack_operation_use_cache_mock.build_cache.get_public_key_from_cache.assert_called_once_with(
+ spack_operation_use_cache_mock.spack_config.buildcache_dir)
+
+ if key_path:
+ spack_operation_use_cache_mock.trust_gpg_key.assert_called_once_with(key_path)
+ else:
+ spack_operation_use_cache_mock.trust_gpg_key.assert_not_called()
+
+ if not signed:
+ spack_operation_use_cache_mock.logger.warning.assert_called_once_with(
+ "Public key not found in cache or failed to trust pgp keys!")
+
+ spack_operation_use_cache_mock.add_mirror.assert_called_once_with(
+ 'local_cache',
+ str(spack_operation_use_cache_mock.spack_config.buildcache_dir),
+ signed=signed,
+ autopush=True,
+ global_mirror=False
+ )
+ super_mock.return_value.setup_spack_env.assert_called_once() # call original method
+
+ @pytest.mark.parametrize("exception_type", [ValueError, NoSpackEnvironmentException])
+ def test_setup_spack_env_exceptions(self, mocker, spack_operation_use_cache_mock, exception_type):
+ # Test ID: setup_spack_env_exceptions
+ spack_operation_use_cache_mock.trust_gpg_key = mocker.MagicMock()
+ spack_operation_use_cache_mock.add_mirror = mocker.MagicMock()
+
+ # Arrange
+ spack_operation_use_cache_mock.build_cache.get_public_key_from_cache.return_value = "path/to/key.gpg"
+ spack_operation_use_cache_mock.trust_gpg_key.return_value = True
+ exception = exception_type("test exception")
+ spack_operation_use_cache_mock.add_mirror.side_effect = exception
+
+ # Act & Assert
+ with pytest.raises(exception_type):
+ spack_operation_use_cache_mock.setup_spack_env()
+
+ spack_operation_use_cache_mock.logger.error.assert_called_once_with("Error adding buildcache mirror: %s",
+ exception)
diff --git a/pyproject.toml b/pyproject.toml
index abcbe05d..c8b8f7b6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -6,20 +6,26 @@ build-backend = "setuptools.build_meta"
name = "esd-tools"
version = "0.1.0"
authors = [
- {name = "Eric Müller", email = "mueller@kip.uni-heidelberg.de"},
- {name = "Adrian Ciu", email = "adrian.ciu@codemart.ro"},
+ { name = "Eric Müller", email = "mueller@kip.uni-heidelberg.de" },
+ { name = "Adrian Ciu", email = "adrian.ciu@codemart.ro" },
+ { name = "Jithu Murugan", email = "j.murugan@fz-juelich.de" }
]
-description = "This package provides all the necessary tools to create an Ebrains Software Distribution environment"
+description = "This package includes all the essential tools required to set up an EBRAINS Software Distribution environment."
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"oras",
"spack",
"ruamel.yaml",
- "pytest",
- "pytest-mock",
- "pytest-ordering",
]
[tool.setuptools.data-files]
-"esd-tools" = ["esd/logger/logging.conf"]
\ No newline at end of file
+"esd-tools" = ["esd/logger/logging.conf"]
+
+[options.extras_require]
+test = [
+ "pytest",
+ "pytest-mock",
+ "pytest-ordering",
+ "coverage"
+]
\ No newline at end of file
--
GitLab