VT-107: make BuildCacheManager flexible to missing or incorrect OCI credentials

README.md

 # Dedal
-![Coverage Badge](https://gitlab.ebrains.eu/ri/tech-hub/platform/esd/dedal/badges/master/coverage.svg)
 
 This repository provides functionalities to easily ```managed spack environments``` and
 ```helpers for the container image build flow```.

dedal/build_cache/BuildCacheManager.py

+import builtins
 import glob
 import os
 from os.path import join
@@ -14,24 +15,44 @@ class BuildCacheManager(BuildCacheManagerInterface):
         This class aims to manage the push/pull/delete of build cache files
     """
 
-    def __init__(self, registry_host, registry_project, registry_username, registry_password, cache_version='cache',
-                 auth_backend='basic', insecure=False, tls_verify=True):
-        self._logger = get_logger(__name__, BuildCacheManager.__name__)
-        self._registry_project = registry_project
+    def __new__(cls, registry_host, registry_project, registry_username, registry_password, cache_version='cache',
+                auth_backend='basic', insecure=False, tls_verify=True):
+        instance = super().__new__(cls)
+        instance._logger = get_logger(__name__, BuildCacheManager.__name__)
+        instance._registry_project = registry_project
 
-        self._registry_username = registry_username
-        self._registry_password = registry_password
+        instance._registry_username = registry_username
+        instance._registry_password = registry_password
 
-        self._registry_host = registry_host
+        instance._registry_host = registry_host
+
+        # Override input to disable prompts during login.
+        # Define a function that raises an exception when input is called.
+        def disabled_input(prompt=""):
+            raise Exception("Interactive login disabled: credentials are provided via attributes.")
+
+        # Save the original input function.
+        original_input = builtins.input
+        # Override input so that any call to input() during login will raise our exception.
+        builtins.input = disabled_input
         # Initialize an OrasClient instance.
         # This method utilizes the OCI Registry for container image and artifact management.
         # Refer to the official OCI Registry documentation for detailed information on the available authentication methods.
         # Supported authentication types may include basic authentication (username/password), token-based authentication,
-        self._client = oras.client.OrasClient(hostname=self._registry_host, auth_backend=auth_backend,
+        instance._client = oras.client.OrasClient(hostname=instance._registry_host, auth_backend=auth_backend,
                                               insecure=insecure, tls_verify=tls_verify)
-        self._client.login(username=self._registry_username, password=self._registry_password)
-        self.cache_version = cache_version
-        self._oci_registry_path = f'{self._registry_host}/{self._registry_project}/{self.cache_version}'
+
+        try:
+            instance._client.login(username=instance._registry_username, password=instance._registry_password)
+        except Exception:
+            instance._logger.error('Login failed!')
+            return None
+        finally:
+            builtins.input = original_input
+
+        instance.cache_version = cache_version
+        instance._oci_registry_path = f'{instance._registry_host}/{instance._registry_project}/{instance.cache_version}'
+        return instance
 
     def upload(self, upload_dir: Path, override_cache=True):
         """

dedal/spack_factory/SpackOperationCreateCache.py

@@ -50,10 +50,16 @@ class SpackOperationCreateCache(SpackOperation):
             NoSpackEnvironmentException: If the spack environment is not set up.
         """
         super().concretize_spack_env(force=True, test=test)
-        dependency_path = self.spack_config.env.path / self.spack_config.env.name / 'spack.lock'
-        copy_file(dependency_path, self.spack_config.concretization_dir, logger=self.logger)
-        self.cache_dependency.upload(self.spack_config.concretization_dir, override_cache=self.spack_config.override_cache)
-        self.logger.info(f'Created new spack concretization for create cache: {self.spack_config.env.name}')
+        if self.cache_dependency:
+            dependency_path = self.spack_config.env.path / self.spack_config.env.name / 'spack.lock'
+            copy_file(dependency_path, self.spack_config.concretization_dir, logger=self.logger)
+            self.cache_dependency.upload(self.spack_config.concretization_dir,
+                                         override_cache=self.spack_config.override_cache)
+            self.logger.info(
+                f'Finished uploading new spack concretization for create cache: {self.spack_config.env.name}')
+        else:
+            self.logger.info(
+                f'Created new spack concretization for create cache: {self.spack_config.env.name}. No OCI credentials for concretization step were provided!')
 
     @check_spack_env
     def install_packages(self, jobs: int = 2, debug=False, test=None):
@@ -65,4 +71,9 @@ class SpackOperationCreateCache(SpackOperation):
             self.create_gpg_keys()
             self.logger.info('Created gpg keys')
         super().install_packages(jobs=jobs, signed=self.signed, debug=debug, fresh=True, test=test)
-        self.build_cache.upload(self.spack_config.buildcache_dir, override_cache=self.spack_config.override_cache)
+        if self.build_cache:
+            self.build_cache.upload(self.spack_config.buildcache_dir, override_cache=self.spack_config.override_cache)
+            self.logger.info(f'Finished uploading build cache: {self.spack_config.env.name}')
+        else:
+            self.logger.info(
+                f'Created build cache: {self.spack_config.env.name}. No OCI credentials for concretization step were provided!')

dedal/spack_factory/SpackOperationCreator.py

@@ -9,18 +9,21 @@ from dedal.spack_factory.SpackOperationUseCache import SpackOperationUseCache
 class SpackOperationCreator:
     @staticmethod
     def get_spack_operator(spack_config: SpackConfig = None, use_cache: bool = False) -> SpackOperation:
-        env_vars = [os.environ.get('CONCRETIZE_OCI_HOST'),
-                    os.environ.get('CONCRETIZE_OCI_PROJECT'),
-                    os.environ.get('CONCRETIZE_OCI_USERNAME'),
-                    os.environ.get('CONCRETIZE_OCI_PASSWORD'),
-                    os.environ.get('BUILDCACHE_OCI_HOST'),
-                    os.environ.get('BUILDCACHE_OCI_PROJECT'),
-                    os.environ.get('BUILDCACHE_OCI_USERNAME'),
-                    os.environ.get('BUILDCACHE_OCI_PASSWORD')
-                    ]
+        env_vars_concretization_cache = [
+            os.environ.get('CONCRETIZE_OCI_HOST'),
+            os.environ.get('CONCRETIZE_OCI_PROJECT'),
+            os.environ.get('CONCRETIZE_OCI_USERNAME'),
+            os.environ.get('CONCRETIZE_OCI_PASSWORD'),
+        ]
+        env_vars_build_cache = [
+            os.environ.get('BUILDCACHE_OCI_HOST'),
+            os.environ.get('BUILDCACHE_OCI_PROJECT'),
+            os.environ.get('BUILDCACHE_OCI_USERNAME'),
+            os.environ.get('BUILDCACHE_OCI_PASSWORD')
+        ]
         if spack_config is None:
             return SpackOperation()
-        elif None in env_vars:
+        elif None in env_vars_concretization_cache and None in env_vars_build_cache:
             return SpackOperation(spack_config)
         elif spack_config.concretization_dir is None and spack_config.buildcache_dir is None:
             return SpackOperation(spack_config)

dedal/spack_factory/SpackOperationUseCache.py

@@ -39,21 +39,23 @@ class SpackOperationUseCache(SpackOperation):
         """
         super().setup_spack_env()
         # Download concretization cache from OCI Registry
-        self.cache_dependency.download(self.spack_config.concretization_dir)
+        if self.cache_dependency:
+            self.cache_dependency.download(self.spack_config.concretization_dir)
         # 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(str(self.spack_config.buildcache_dir))
-        signed = cached_public_key is not None
-        if signed:
-            self.trust_gpg_key(cached_public_key)
-        # Add build cache mirror
-        self.add_mirror('local_cache',
-                        str(self.spack_config.buildcache_dir),
-                        signed=signed,
-                        autopush=False,
-                        global_mirror=False)
-        self.update_buildcache_index(self.spack_config.buildcache_dir)
-        self.install_gpg_keys()
+        if self.build_cache:
+            self.build_cache.download(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
+            if signed:
+                self.trust_gpg_key(cached_public_key)
+            # Add build cache mirror
+            self.add_mirror('local_cache',
+                            str(self.spack_config.buildcache_dir),
+                            signed=signed,
+                            autopush=False,
+                            global_mirror=False)
+            self.update_buildcache_index(self.spack_config.buildcache_dir)
+            self.install_gpg_keys()
 
     @check_spack_env
     def concretize_spack_env(self, test=None):
@@ -63,7 +65,7 @@ class SpackOperationUseCache(SpackOperation):
             NoSpackEnvironmentException: If the spack environment is not set up.
         """
         concretization_redo = False
-        if file_exists_and_not_empty(self.spack_config.concretization_dir / 'spack.lock'):
+        if self.cache_dependency and file_exists_and_not_empty(self.spack_config.concretization_dir / 'spack.lock'):
             concretization_file_path = self.env_path / 'spack.lock'
             copy_file(self.spack_config.concretization_dir / 'spack.lock', self.env_path)
             # redo the concretization step if spack.lock file was not downloaded from the cache
@@ -90,18 +92,21 @@ class SpackOperationUseCache(SpackOperation):
         signed = '' if signed else '--no-check-signature'
         debug = '--debug' if debug else ''
         test = f'--test {test}' if test else ''
-        install_result = run_command("bash", "-c",
-                                     f'{self.spack_command_on_env} && spack {debug} install -v --cache-only {signed} -j {jobs} {test}',
-                                     stdout=subprocess.PIPE,
-                                     stderr=subprocess.PIPE,
-                                     text=True,
-                                     logger=self.logger,
-                                     info_msg=f"Installing spack packages for {self.spack_config.env.name}",
-                                     exception_msg=f"Error installing spack packages for {self.spack_config.env.name}",
-                                     exception=SpackInstallPackagesException)
-        log_command(install_result, str(Path(os.getcwd()).resolve() / ".generate_cache.log"))
-        if install_result.returncode == 0:
-            self.logger.info(f'Finished installation of spack packages from cache.')
+        if self.build_cache:
+            install_result = run_command("bash", "-c",
+                                         f'{self.spack_command_on_env} && spack {debug} install -v --cache-only {signed} -j {jobs} {test}',
+                                         stdout=subprocess.PIPE,
+                                         stderr=subprocess.PIPE,
+                                         text=True,
+                                         logger=self.logger,
+                                         info_msg=f"Installing spack packages for {self.spack_config.env.name}",
+                                         exception_msg=f"Error installing spack packages for {self.spack_config.env.name}",
+                                         exception=SpackInstallPackagesException)
+            log_command(install_result, str(Path(os.getcwd()).resolve() / ".generate_cache.log"))
+            if install_result.returncode == 0:
+                self.logger.info(f'Finished installation of spack packages from cache.')
+            else:
+                self.logger.error(f'Something went wrong during installation from cache. Please check the logs.')
         else:
-            self.logger.error(f'Something went wrong during installation from cache. Please check the logs.')
+            install_result = super().install_packages(jobs=jobs, signed=signed, debug=debug, test=test)
         return install_result