Skip to content
Snippets Groups Projects
Commit 5e4b78f9 authored by Jithu Murugan's avatar Jithu Murugan
Browse files

- Corrected the failing unit tests and minor refactorings.

parent ddddaa44
No related branches found
No related tags found
1 merge request!4feat(spack_operation): implement setup_spack_env functionality
Pipeline #59533 canceled with stages
in 46 minutes and 53 seconds
......@@ -7,9 +7,9 @@
# 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 os import name as os_name
from dedal.commands.command import Command
from dedal.commands.command_sequence import CommandSequence
......@@ -29,7 +29,7 @@ class BashCommandExecutor:
"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"])
self.bash_command: list[str] = command_map.get(os_name, ["undefined"])
def add_command(self, command: Command) -> None:
"""Adds a command to the sequence.
......
......@@ -33,14 +33,14 @@ class SpackOperationUseCache(SpackOperation):
and adds the build cache mirror.
Raises:
ValueError: If there is an issue with the build cache setup.
ValueError: If there is an issue with the build cache setup (mirror_name/mirror_path are empty).
NoSpackEnvironmentException: If the spack environment is not set up.
"""
super().setup_spack_env()
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)
cached_public_key = self.build_cache.get_public_key_from_cache(str(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!")
......
......@@ -33,14 +33,11 @@ class TestBashCommandExecutor:
)
def test_init_success_path(self, mocker, test_id, os_name, expected_bash_command):
# Arrange
original_os = os.name
mock_get_logger = mocker.patch("dedal.commands.bash_command_executor.logging.getLogger")
mocker.patch("dedal.commands.bash_command_executor.os.name", os_name)
# mock_os_name.return_value = os_name
mocker.patch("dedal.commands.bash_command_executor.os_name", 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
......@@ -76,7 +73,7 @@ class TestBashCommandExecutor:
# Assert
assert str(except_info.value) == "Invalid command type. Use Command."
@patch("dedal.commands.bash_command_executor.os.name", "unknown")
@patch("dedal.commands.bash_command_executor.os_name", "unknown")
def test_init_unknown_os(self):
# Act
......@@ -135,8 +132,7 @@ class TestBashCommandExecutor:
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("dedal.commands.bash_command_executor.os.name", "nt")
mocker.patch("dedal.commands.bash_command_executor.os_name", "nt")
executor = BashCommandExecutor()
executor.add_command(MockCommand("failing_command"))
......@@ -149,9 +145,6 @@ class TestBashCommandExecutor:
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",
[
......@@ -169,8 +162,7 @@ class TestBashCommandExecutor:
)
def test_execute_other_errors(self, test_id, exception, expected_error_message, mocker):
# Arrange
original_os = os.name
mocker.patch("dedal.commands.bash_command_executor.os.name", "nt")
mocker.patch("dedal.commands.bash_command_executor.os_name", "nt")
with patch("dedal.commands.bash_command_executor.subprocess.run", side_effect=exception) as mock_subprocess_run:
executor = BashCommandExecutor()
executor.add_command(MockCommand("some_command"))
......@@ -184,9 +176,6 @@ class TestBashCommandExecutor:
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()
......@@ -197,12 +186,11 @@ class TestBashCommandExecutor:
# Assert
assert str(except_info.value) == "No commands to execute."
@patch("dedal.commands.bash_command_executor.subprocess.run")
def test_execute_happy_path_nt(self, mock_subprocess_run, mocker):
# Arrange
original_os = os.name
mocker.patch("dedal.commands.bash_command_executor.os.name", "nt")
mocker.patch("dedal.commands.bash_command_executor.os_name", "nt")
executor = BashCommandExecutor()
executor.add_command(MockCommand("echo hello"))
mock_subprocess_run.return_value.stdout = "hello\n"
......@@ -217,19 +205,21 @@ class TestBashCommandExecutor:
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("dedal.commands.bash_command_executor.os.name", "unknown")
def test_execute_unknown_os(self):
def test_execute_unknown_os(self, mocker):
# Arrange
errors = {
"posix": "Error: Bash Command: ['undefined'] not found: [Errno 2] No such file or directory",
"nt": "Error: Bash Command: ['undefined'] not found: [WinError 2] The system cannot find the file "
'specified'
}
original_os = os.name
expected_error = errors.get(original_os, "Error: Unknown OS")
mocker.patch("dedal.commands.bash_command_executor.os_name")
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')
assert executor.execute() == (None, expected_error)
def test_reset(self, mocker):
# Test ID: reset
......
......@@ -60,7 +60,9 @@ class TestBuildCacheManager:
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")
assert result in [str(build_cache_dir / "project0" / "_pgp" / "key0.pub"),
str(build_cache_dir / "project0" / "_pgp" / "key1.pub"),
str(build_cache_dir / "project1" / "_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)
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment