def __get_custom_sources_to_spec()

in src/core/src/package_managers/AptitudePackageManager.py [0:0]

• 34 lines of code
• 22 McCabe index (conditional complexity)

    def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_classification=str()):
        # type: (str, str) -> (str, str)
        """ Prepares the custom sources list for use in a command. Idempotent. """
        try:
            # basic input validation
            if max_patch_published_date != str() and len(max_patch_published_date) != 16:
                raise Exception("[APM] Invalid max patch published date received. [Value={0}]".format(str(max_patch_published_date)))
            if base_classification != str() and base_classification != Constants.PackageClassification.SECURITY:
                raise Exception("[APM] Invalid classification selection received. [Value={0}]".format(str(base_classification)))

            # utilize caching if data exists and cache is also in the desired state
            formula = "{0}-{1}".format(max_patch_published_date if max_patch_published_date != str() else "any", base_classification.lower() if base_classification != str() else "all")
            if self.current_source_formula == formula:
                return self.current_source_parts_dir, self.current_source_list     # no need to refresh repo as state should match

            # utilize caching with refresh repo if data exists but cache is not in the desired state
            self.current_source_formula = formula
            self.current_source_parts_dir = self.custom_source_parts_dir_template.replace("<FORMULA>", formula)
            self.current_source_list = self.custom_source_list_template.replace("<FORMULA>", formula)
            if os.path.exists(self.current_source_list) and os.path.isdir(self.current_source_parts_dir):
                self.refresh_repo(source_parts_dir=self.current_source_parts_dir, source_list=self.current_source_list)     # refresh repo as requested state has changed
                return self.current_source_parts_dir, self.current_source_list

            # Produce data for combination not seen previously in this execution - source list
            source_list_content = self.__read_one_line_style_list_format(self.APT_SOURCES_LIST_PATH, max_patch_published_date, base_classification) if os.path.exists(self.APT_SOURCES_LIST_PATH) else str()
            self.env_layer.file_system.write_with_retry(self.current_source_list, source_list_content, "w")
            self.composite_logger.log_verbose("[APM] Source list content written. [List={0}][Content={1}]".format(self.current_source_list, source_list_content))

            # Produce data for combination not seen previously in this execution - source parts
            source_parts_deb882_style_content, source_parts_list_content = self.__get_consolidated_source_parts_content(max_patch_published_date, base_classification)
            if len(source_parts_list_content) > 0:  # list(s) in source parts
                self.env_layer.file_system.write_with_retry(self.current_source_list, "\n" + source_parts_list_content, "a")
                self.composite_logger.log_verbose("[APM] Source parts list content appended. [List={0}][Content={1}]".format(self.current_source_list, source_parts_list_content))

            # Source parts debstyle882-only initialization
            current_source_parts_deb882_style_file = os.path.join(self.current_source_parts_dir, self.current_source_parts_file_name)
            if os.path.isdir(self.current_source_parts_dir):
                shutil.rmtree(self.current_source_parts_dir)

            # Create the folder and write to it only if there is debstyle882 content to be written
            if len(source_parts_deb882_style_content) > 0:
                os.makedirs(self.current_source_parts_dir)
                self.env_layer.file_system.write_with_retry(current_source_parts_deb882_style_file, source_parts_deb882_style_content, "w")
                self.composite_logger.log_verbose("[APM] Source parts debstyle882 content written. [Dir={0}][Content={1}]".format(current_source_parts_deb882_style_file, source_parts_deb882_style_content))

        except Exception as error:
            self.composite_logger.log_error("[APM] Error in modifying custom sources list. [Error={0}]".format(repr(error)))
            return str(), str()    # defaults code to safety

        # Refresh repo
        self.refresh_repo(source_parts_dir=self.current_source_parts_dir, source_list=self.current_source_list)

        return self.current_source_parts_dir, self.current_source_list