diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 65062ec73e35ad1daa1f5e88cae077a86e7c9cc0..98a371df8b3604b478d5bf783571dc22e0fe1424 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 0000000000000000000000000000000000000000..d6809e0967f4d931510b20c673a811a27c9825a2
--- /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 cccb584695b9c59f0b4ca1ab2eb981f5c9ad9242..8641a68e98c916d85321d8853c81abf55ac4a560 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 0000000000000000000000000000000000000000..ea9c384e46ebd5602957ff010a7c2b7c2d7d52e6
--- /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 0000000000000000000000000000000000000000..1811b757b1fe1fd14702d56a5b1e63a924ca0d6b
--- /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 0000000000000000000000000000000000000000..07e9837c142122ff43a5fcd47959a12d7ab0f28f
--- /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 0000000000000000000000000000000000000000..7ef184cbfab3e193c292f7fcbdbedcb7a5783211
--- /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 0000000000000000000000000000000000000000..ea2ab6466eaccc7d3df11b41f1ff7286a5a5ec60
--- /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 0000000000000000000000000000000000000000..a9bd89a02b12adbc44ec33a92e9a7647f048c680
--- /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 0000000000000000000000000000000000000000..d0a3b469dc7941ee26735a2f0993ef8bbdf35af3
--- /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 0000000000000000000000000000000000000000..5f1d2c3f440573920e6957201019bbf1737aaa39
--- /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 0000000000000000000000000000000000000000..96804660f2f18b3ef42ac479ecad51885d0f57e3
--- /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 0000000000000000000000000000000000000000..ea40079d2d810ef2475390b54ae43f94c222b0f5
--- /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 0000000000000000000000000000000000000000..14b747ada43cf6e5eb31c0ddcbf1753fa2eac485
--- /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 0000000000000000000000000000000000000000..99baec39b378a9f30eb4b4b31fb4b6813fc9233d
--- /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 0000000000000000000000000000000000000000..e8d7e52229ae7aea84b3a8dee0e1ddf6c2549e01
--- /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 93a2e87423b871b0b9b4f583efec79e328739edf..26c6617fafad7c05c9374b2b84a8b3aff447a325 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 904582f45b6c6360edadc588c5851e2911e2f4b2..5a026283a36da8707bd1b33d524bcb7fd1961979 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 313522d235515955124ba44536976d995e17ff77..b6e7846cbd5791db5dda84ee3ba1e331b908ea2a 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 0000000000000000000000000000000000000000..70633fa2121e3bd3bddaed9055bd1afdd56aaa67
--- /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 0000000000000000000000000000000000000000..687570eb40ef8d088b999d45ab37c07724b2c274
--- /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 0000000000000000000000000000000000000000..8f957a35adef61b48b3cf91e6280499bdc09d9f3
--- /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 0000000000000000000000000000000000000000..57c4ec3427836247dc97f693fa00188e0afcfb7c
--- /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 0000000000000000000000000000000000000000..24ec4a93a0e000c4098ac9450ded0f576915c0f1
--- /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 0000000000000000000000000000000000000000..8b68e10747507b98106e22540eaa8ca96caf8f2c
--- /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 0000000000000000000000000000000000000000..6c4062a4d15dd6bc5ebd5288abfb092fc5c50e2c
--- /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 0000000000000000000000000000000000000000..7515f14af5cc65d27779e9b8cd1e886aee6646ed
--- /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 0000000000000000000000000000000000000000..e9287e0494bb60645cf32b9acf305f12c36b22b6
--- /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 0000000000000000000000000000000000000000..02539d5529851b8a7beb223da518801129833e2d
--- /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 0000000000000000000000000000000000000000..f2f95b908839d868ed73b8403ad057cd292dddb1
--- /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 0000000000000000000000000000000000000000..9b6bc572b93ff8a36dd96848a6f77fcaf4c8ac84
--- /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 0000000000000000000000000000000000000000..4e2c520ee4105b9613dcee4b8b1e56963e2d8ec4
--- /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 0000000000000000000000000000000000000000..fd4e277823d95fbcfc4b45ef426ee2721e891d09
--- /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 abcbe05d6637b2832ab0c7b9878516380e0087b2..c8b8f7b69e8c777453bbe3153265ef8d268a333e 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