WIP tests

Author: kinow

CHANGELOG.md

@@ -3,7 +3,6 @@
 **Bug fixes:**
 
 - Fixed issue with the verification of dirty Git local repositories in operational experiments #2446
-- Fixed issue where Autosubmit did not retry when there were networking issues in platforms #1369
 
 **Enhancements:**
 

autosubmit/database/db_manager.py

@@ -35,9 +35,9 @@ def __init__(self, root_path: str, db_name: str, db_version: int):
         if is_new:
             self._initialize_database()
     def backup(self):
-        pass
+        pass  # pragma: no cover
     def restore(self):
-        pass
+        pass  # pragma: no cover
 
     def disconnect(self):
         """

autosubmit/history/platform_monitor/platform_utils.py

@@ -49,7 +49,6 @@ def parse_output_number(string_number):
             number = float(number) * multiplier
         except Exception:
             number = 0.0
-            pass        
     return number
 
 def try_parse_time_to_timestamp(input_):

autosubmit/history/strategies.py

@@ -40,14 +40,14 @@ def execute_distribution(self, job_data_dc, job_data_dcs_in_wrapper, slurm_monit
 
 
 class Strategy(metaclass=ABCMeta):
-  """ Strategy Interface """
+  """Strategy Interface"""
 
   def __init__(self, historiclog_dir_path=DEFAULT_HISTORICAL_LOGS_DIR):
     self.historiclog_dir_path = historiclog_dir_path
 
   @abstractmethod
   def apply_distribution(self, job_data_dc, job_data_dcs_in_wrapper, slurm_monitor):
-    pass
+    pass  # pragma: no cover
 
   def set_job_data_dc_as_processed(self, job_data_dc, original_ssh_output):
     job_data_dc.platform_output = original_ssh_output

autosubmit/job/job_list.py

@@ -1763,7 +1763,7 @@ def get_date_format(self):
         return date_format
 
     def copy_ordered_jobs_by_date_member(self):
-        pass
+        pass  # pragma: no cover
 
     def get_ordered_jobs_by_date_member(self, section):
         """

autosubmit/job/job_list_persistence.py

@@ -42,7 +42,7 @@ def save(self, persistence_path, persistence_file, job_list , graph):
         :param persistence_path: str
 
         """
-        raise NotImplementedError
+        raise NotImplementedError  # pragma: no cover
 
     def load(self, persistence_path, persistence_file):
         """
@@ -51,7 +51,7 @@ def load(self, persistence_path, persistence_file):
         :param persistence_path: str
 
         """
-        raise NotImplementedError
+        raise NotImplementedError  # pragma: no cover
 
 
 class JobListPersistencePkl(JobListPersistence):

autosubmit/job/job_packager.py

@@ -20,7 +20,7 @@
 from contextlib import suppress
 from math import ceil
 from operator import attrgetter
-from typing import List
+from typing import List, TYPE_CHECKING
 
 from bscearth.utils.date import sum_str_hours
 
@@ -31,6 +31,11 @@
 from autosubmit.job.template import Language
 from autosubmit.log.log import Log, AutosubmitCritical
 
+if TYPE_CHECKING:
+    from autosubmit.config.configcommon import AutosubmitConfig
+    from autosubmit.job.job_list import JobList
+    from autosubmit.platforms.paramiko_platform import ParamikoPlatform
+
 
 class JobPackager(object):
     """
@@ -44,8 +49,7 @@ class JobPackager(object):
     :type jobs_list: JobList object.
     """
 
-
-    def __init__(self, as_config, platform, jobs_list, hold=False):
+    def __init__(self, as_config: 'AutosubmitConfig', platform: 'ParamikoPlatform', jobs_list: 'JobList', hold=False):
         self.current_wrapper_section = "WRAPPERS"
         self._as_config = as_config
         self._platform = platform
@@ -65,9 +69,8 @@ def __init__(self, as_config, platform, jobs_list, hold=False):
         self.special_variables = dict()
         self.wrappers_with_error = {}
 
-
-        #todo add default values
-        #Wrapper building starts here
+        # TODO: Add default values
+        # Wrapper building starts here
         for wrapper_section,wrapper_data in self._as_config.experiment_data.get("WRAPPERS",{}).items():
             if isinstance(wrapper_data,collections.abc.Mapping ):
                 self.wrapper_type[wrapper_section] = self._as_config.get_wrapper_type(wrapper_data)
@@ -127,7 +130,6 @@ def compute_weight(self, job_list):
                 job.distance_weight = job.distance_weight - 1
 
     def calculate_wrapper_bounds(self, section_list):
-
         """
         Returns the minimum and maximum number of jobs that can be wrapped
 
@@ -433,56 +435,62 @@ def error_message_policy(self, min_h: int, min_v: int, wrapper_limits: dict, bal
         message += "\nThis message is activated when only jobs_in_wrappers are in active(Ready+) status.\n"
         return message
 
-    def check_if_packages_are_ready_to_build(self):
-        """
-        Check if the packages are ready to be built
-        :return: List of jobs ready to be built, boolean indicating if packages can't be built for other reasons ( max_total_jobs...)
+    def check_if_packages_are_ready_to_build(self) -> tuple[list[Job], bool]:
+        """Check if the packages are ready to be built.
+
+        Returns a tuple with two elements. First contains the list of jobs ready to be built.
+        The second element in the tuple is a boolean indicating if it can be built or not.
+
+        :return: list of jobs ready to be built, boolean indicating if there are underlying blocking errors.
         """
         Log.info("Calculating possible ready jobs for {0}".format(self._platform.name))
-        jobs_ready = list()
+        jobs_ready = []
         if len(self._jobs_list.jobs_to_run_first) > 0:
-            jobs_ready = [job for job in self._jobs_list.jobs_to_run_first if
-                     ( self._platform is None or job.platform.name.upper() == self._platform.name.upper()) and
-                     job.status == Status.READY]
-        if len(jobs_ready) == 0:
+            jobs_ready = [
+                job
+                for job in self._jobs_list.jobs_to_run_first
+                if (self._platform is None or job.platform.name.upper() == self._platform.name.upper())
+                and job.status == Status.READY
+            ]
+        if not jobs_ready:
             if self.hold:
                 jobs_ready = self._jobs_list.get_prepared(self._platform)
             else:
                 jobs_ready = self._jobs_list.get_ready(self._platform)
 
-        if self.hold and len(jobs_ready) > 0:
+        if self.hold and jobs_ready:
             self.compute_weight(jobs_ready)
-            sorted_jobs = sorted(
-                jobs_ready, key=operator.attrgetter('distance_weight'))
-            jobs_in_held_status = self._jobs_list.get_held_jobs() + self._jobs_list.get_submitted(self._platform, hold=self.hold)
+            sorted_jobs = sorted(jobs_ready, key=operator.attrgetter('distance_weight'))
+            jobs_in_held_status = self._jobs_list.get_held_jobs() + self._jobs_list.get_submitted(
+                self._platform, hold=self.hold)
             held_by_id = dict()
             for held_job in jobs_in_held_status:
                 if held_job.id not in held_by_id:
                     held_by_id[held_job.id] = []
                 held_by_id[held_job.id].append(held_job)
             current_held_jobs = len(list(held_by_id.keys()))
             remaining_held_slots = 5 - current_held_jobs
-            Log.debug("there are currently {0} held jobs".format(remaining_held_slots))
-            try:
+            Log.debug(f"There are currently {remaining_held_slots} held jobs")
+            with suppress(IndexError):
                 while len(sorted_jobs) > remaining_held_slots:
                     del sorted_jobs[-1]
                 for job in sorted_jobs:
                     if job.distance_weight > 3:
                         sorted_jobs.remove(job)
                 jobs_ready = sorted_jobs
-                pass
-            except IndexError:
-                pass
-        if len(jobs_ready) == 0:
-            # If there are no jobs ready, result is tuple of empty
-            return jobs_ready,False
-        #check if there are jobs listed on calculate_job_limits
+
+        # If there are no jobs ready, result is tuple of empty
+        if not jobs_ready:
+            return jobs_ready, False
+
+        # Check if there are jobs listed on calculate_job_limits
         self.calculate_job_limits(self._platform)
+        # If there is no more space in platform, result is tuple of empty
         if not (self._max_wait_jobs_to_submit > 0 and self._max_jobs_to_submit > 0):
-            # If there is no more space in platform, result is tuple of empty
-            Log.debug('Max jobs to submit reached, waiting for more space in platform {0}'.format(self._platform.name))
-            return jobs_ready,False
-        return jobs_ready,True
+            Log.debug(f'Max jobs to submit reached, waiting for more space in platform {self._platform.name}')
+            return jobs_ready, False
+
+        return jobs_ready, True
 
     def calculate_job_limits(self,platform,job=None):
         jobs_list = self._jobs_list
@@ -683,7 +691,6 @@ def _divide_list_by_section(self, jobs_list):
                 del jobs_by_section[wrappers]
         return jobs_by_section
 
-
     def _build_horizontal_packages(self, section_list, wrapper_limits, section, wrapper_info={}):
         packages = []
         horizontal_packager = JobPackagerHorizontal(section_list, self._platform.max_processors, wrapper_limits,
@@ -731,7 +738,7 @@ def _build_vertical_packages(self, section_list, wrapper_limits,wrapper_info={})
         return packages
 
     def _build_hybrid_package(self, jobs_list, wrapper_limits, section,wrapper_info={}):
-        #self.wrapper_info = wrapper_info
+        # self.wrapper_info = wrapper_info
         jobs_resources = dict()
         jobs_resources['MACHINEFILES'] = self._as_config.get_wrapper_machinefiles()
 
@@ -802,7 +809,9 @@ def _build_vertical_horizontal_package(self, horizontal_packager, jobs_resources
         return JobPackageVerticalHorizontal(current_package, total_processors, total_wallclock,
                                             jobs_resources=jobs_resources, method=self.wrapper_method[self.current_wrapper_section], configuration=self._as_config, wrapper_section=self.current_wrapper_section )
 
-#TODO rename and unite JobPackerVerticalMixed to JobPackerVertical since the difference between the two is not needed anymore
+
+# TODO: Rename and unite JobPackerVerticalMixed to JobPackerVertical since
+#       the difference between the two is not needed anymore
 class JobPackagerVertical(object):
     """
     Vertical Packager Parent Class
@@ -867,7 +876,7 @@ def build_vertical_package(self, job, wrapper_info):
         return self.jobs_list
 
     def get_wrappable_child(self, job):
-        pass
+        pass  # pragma: no cover
 
     def _is_wrappable(self, job):
         """
@@ -888,6 +897,7 @@ def _is_wrappable(self, job):
             return True
         return False
 
+
 class JobPackagerVerticalMixed(JobPackagerVertical):
     """
     Vertical Mixed Class. First statement of the constructor builds JobPackagerVertical.
@@ -926,7 +936,6 @@ def __init__(self, dict_jobs, ready_job, jobs_list, total_wallclock, max_jobs, w
         self.sorted_jobs = dict_jobs[date][member]
         self.index = 0
 
-
     def get_wrappable_child(self, job: Job) -> Job:
         """
         Goes through the jobs with the same date and member as the input job, and returns the first that satisfies self._is_wrappable().
@@ -1113,7 +1122,8 @@ def components_dict(self):
     def create_components_dict(self):
         self._sectionList = []
         # it was job.parameters
-        parameters = {}  # TODO machinefiles, can wait nobody is using it and I really think this was not working before anyway
+        # TODO machinefiles, can wait nobody is using it and I really think this was not working before anyway
+        parameters = {}
         for job in self.job_list:
             if job.section not in self._sectionList:
                 self._sectionList.append(job.section)

autosubmit/job/job_packages.py

@@ -134,7 +134,7 @@ def _create_scripts_threaded(self, jobs: list[Job], configuration: 'AutosubmitCo
             self._job_scripts[jobs[i].name] = jobs[i].create_script(configuration)
 
     def _create_common_script(self, filename: str = ""):
-        pass
+        pass  # pragma: no cover
 
     def submit_unthreaded(self, configuration: 'AutosubmitConfig', only_generate: bool = False):
         """
@@ -225,7 +225,7 @@ def _send_files(self):
 
     def _do_submission(self, job_scripts=None, hold: bool = False):
         """ Submit package to the platform. """
-        pass
+        pass  # pragma: no cover
 
     def process_jobs_to_submit(self, job_id: str, hold: bool = False) -> None:
         for i, job in enumerate(self.jobs):
@@ -664,7 +664,7 @@ def _do_submission(self, job_scripts: dict[str, str] = None, hold: bool = False)
             self.jobs[i].wrapper_name = self.name
 
     def _common_script_content(self) -> str:
-        pass
+        pass  # pragma: no cover
 
 
 class JobPackageThreadWrapped(JobPackageThread):

autosubmit/log/log.py

@@ -177,7 +177,7 @@ class Log:
     log.addHandler(console_handler)
 
     def __init__(self):
-        pass
+        pass  # pragma: no cover
 
     def init_variables(self, file_path="") -> None:
         self.file_path = file_path

autosubmit/migrate/migrate.py

@@ -85,7 +85,7 @@ def migrate_pickup(self):
                             finished = False
                             pipeline_broke = False
                             Log.info(
-                                "Rsync launched {0} times. Can take up to 150 retries or until all data is transferred".format(
+                                "Rsync launched {0} times. Can take up to 150 retrials or until all data is transferred".format(
                                     rsync_retries + 1))
                             try:
                                 p.send_command(