diff --git a/src/core/src/core_logic/ConfigurePatchingProcessor.py b/src/core/src/core_logic/ConfigurePatchingProcessor.py index 2db851f0..e7ed36f2 100644 --- a/src/core/src/core_logic/ConfigurePatchingProcessor.py +++ b/src/core/src/core_logic/ConfigurePatchingProcessor.py @@ -80,6 +80,8 @@ def __try_set_patch_mode(self): # NOTE: this condition will be false for Assessment operations, since patchMode is not sent in the API request if self.current_auto_os_patch_state != Constants.AutomaticOSPatchStates.DISABLED and self.execution_config.patch_mode == Constants.PatchModes.AUTOMATIC_BY_PLATFORM: self.package_manager.disable_auto_os_update() + elif self.current_auto_os_patch_state == Constants.AutomaticOSPatchStates.DISABLED and self.execution_config.patch_mode == Constants.PatchModes.IMAGE_DEFAULT: + self.package_manager.revert_auto_os_update_to_system_default() self.current_auto_os_patch_state = self.package_manager.get_current_auto_os_patch_state() diff --git a/src/core/src/package_managers/AptitudePackageManager.py b/src/core/src/package_managers/AptitudePackageManager.py index 1229117a..a16852a1 100644 --- a/src/core/src/package_managers/AptitudePackageManager.py +++ b/src/core/src/package_managers/AptitudePackageManager.py @@ -675,7 +675,7 @@ def get_package_size(self, output): # region auto OS updates def get_current_auto_os_patch_state(self): """ Gets the current auto OS update patch state on the machine """ - self.composite_logger.log("Fetching the current automatic OS patch state on the machine...") + self.composite_logger.log("[APM] Fetching the current automatic OS patch state on the machine...") if os.path.exists(self.os_patch_configuration_settings_file_path): self.__get_current_auto_os_updates_setting_on_machine() if not os.path.exists(self.os_patch_configuration_settings_file_path) or int(self.unattended_upgrade_value) == 0: @@ -685,7 +685,7 @@ def get_current_auto_os_patch_state(self): else: current_auto_os_patch_state = Constants.AutomaticOSPatchStates.UNKNOWN - self.composite_logger.log_debug("Current Auto OS Patch State is [State={0}]".format(str(current_auto_os_patch_state))) + self.composite_logger.log_debug("[APM] Current Auto OS Patch State is [State={0}]".format(str(current_auto_os_patch_state))) return current_auto_os_patch_state def __get_current_auto_os_updates_setting_on_machine(self): @@ -700,47 +700,39 @@ def __get_current_auto_os_updates_setting_on_machine(self): self.unattended_upgrade_value = re.search(self.unattended_upgrade + ' *"(.*?)".', str(setting)).group(1) if self.update_package_list_value == "": - self.composite_logger.log_debug("Machine did not have any value set for [Setting={0}]".format(str(self.update_package_list))) + self.composite_logger.log_debug("[APM] Machine did not have any value set for [Setting={0}]".format(str(self.update_package_list))) if self.unattended_upgrade_value == "": - self.composite_logger.log_debug("Machine did not have any value set for [Setting={0}]".format(str(self.unattended_upgrade))) + self.composite_logger.log_debug("[APM] Machine did not have any value set for [Setting={0}]".format(str(self.unattended_upgrade))) except Exception as error: - raise Exception("Error occurred in fetching default auto OS updates from the machine. [Exception={0}]".format(repr(error))) + raise Exception("[APM] Error occurred in fetching default auto OS updates from the machine. [Exception={0}]".format(repr(error))) def disable_auto_os_update(self): """ Disables auto OS updates on the machine only if they are enabled and logs the default settings the machine comes with """ try: - self.composite_logger.log_debug("Disabling auto OS updates if they are enabled") + self.composite_logger.log_debug("[APM] Disabling auto OS updates if they are enabled") self.backup_image_default_patch_configuration_if_not_exists() self.update_os_patch_configuration_sub_setting(self.update_package_list, "0") self.update_os_patch_configuration_sub_setting(self.unattended_upgrade, "0") - self.composite_logger.log("Successfully disabled auto OS updates") + self.composite_logger.log("[APM] Successfully disabled auto OS updates") except Exception as error: - self.composite_logger.log_error("Could not disable auto OS updates. [Error={0}]".format(repr(error))) + self.composite_logger.log_error("[APM] Could not disable auto OS updates. [Error={0}]".format(repr(error))) raise def backup_image_default_patch_configuration_if_not_exists(self): """ Records the default system settings for auto OS updates within patch extension artifacts for future reference. We only log the default system settings a VM comes with, any subsequent updates will not be recorded""" try: - image_default_patch_configuration_backup = {} - image_default_patch_configuration_backup_exists = self.image_default_patch_configuration_backup_exists() - - # read existing backup since it also contains backup from other update services. We need to preserve any existing data with backup file - if image_default_patch_configuration_backup_exists: - try: - image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) - except Exception as error: - self.composite_logger.log_error("Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error))) + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() # verify if existing backup is valid if not, write to backup - is_backup_valid = image_default_patch_configuration_backup_exists and self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) if is_backup_valid: - self.composite_logger.log_debug("Since extension has a valid backup, no need to log the current settings again. [Default Auto OS update settings={0}] [File path={1}]" + self.composite_logger.log_debug("[APM] Since extension has a valid backup, no need to log the current settings again. [Default Auto OS update settings={0}] [File path={1}]" .format(str(image_default_patch_configuration_backup), self.image_default_patch_configuration_backup_path)) else: - self.composite_logger.log_debug("Since the backup is invalid or does not exist, will add a new backup with the current auto OS update settings") + self.composite_logger.log_debug("[APM] Since the backup is invalid or does not exist, will add a new backup with the current auto OS update settings") self.__get_current_auto_os_updates_setting_on_machine() backup_image_default_patch_configuration_json = { @@ -748,28 +740,34 @@ def backup_image_default_patch_configuration_if_not_exists(self): self.unattended_upgrade: self.unattended_upgrade_value } - self.composite_logger.log_debug("Logging default system configuration settings for auto OS updates. [Settings={0}] [Log file path={1}]" + self.composite_logger.log_debug("[APM] Logging default system configuration settings for auto OS updates. [Settings={0}] [Log file path={1}]" .format(str(backup_image_default_patch_configuration_json), self.image_default_patch_configuration_backup_path)) self.env_layer.file_system.write_with_retry(self.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(backup_image_default_patch_configuration_json)), mode='w+') except Exception as error: - error_message = "Exception during fetching and logging default auto update settings on the machine. [Exception={0}]".format(repr(error)) + error_message = "[APM] Exception during fetching and logging default auto update settings on the machine. [Exception={0}]".format(repr(error)) self.composite_logger.log_error(error_message) self.status_handler.add_error_to_status(error_message, Constants.PatchOperationErrorCodes.DEFAULT_ERROR) raise def is_image_default_patch_configuration_backup_valid(self, image_default_patch_configuration_backup): if self.update_package_list in image_default_patch_configuration_backup and self.unattended_upgrade in image_default_patch_configuration_backup: - self.composite_logger.log_debug("Extension already has a valid backup of the default system configuration settings for auto OS updates.") + self.composite_logger.log_debug("[APM] Extension already has a valid backup of the default system configuration settings for auto OS updates.") return True else: - self.composite_logger.log_error("Extension does not have a valid backup of the default system configuration settings for auto OS updates.") + self.composite_logger.log_error("[APM] Extension does not have a valid backup of the default system configuration settings for auto OS updates.") return False def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_setting, value="0", patch_configuration_sub_setting_pattern_to_match=""): """ Updates (or adds if it doesn't exist) the given patch_configuration_sub_setting with the given value in os_patch_configuration_settings_file """ try: # note: adding space between the patch_configuration_sub_setting and value since, we will have to do that if we have to add a patch_configuration_sub_setting that did not exist before - self.composite_logger.log("Updating system configuration settings for auto OS updates. [Patch Configuration Sub Setting={0}] [Value={1}]".format(str(patch_configuration_sub_setting), value)) + self.composite_logger.log("[APM] Updating system configuration settings for auto OS updates. [PatchConfigurationSubSetting={0}][Value={1}]".format(str(patch_configuration_sub_setting), value)) + + if value == '': + self.composite_logger.log_debug("[APM] We won't update the system configuration settings since new configuration value to update does not an acceptable value. [PatchConfigurationSubSetting={0}][ValueToUpdate={1}]" + .format(str(patch_configuration_sub_setting), value)) + return + os_patch_configuration_settings = self.env_layer.file_system.read_with_retry(self.os_patch_configuration_settings_file_path) patch_configuration_sub_setting_to_update = patch_configuration_sub_setting + ' "' + value + '";' patch_configuration_sub_setting_found_in_file = False @@ -789,10 +787,46 @@ def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_sett self.env_layer.file_system.write_with_retry(self.os_patch_configuration_settings_file_path, '{0}'.format(updated_patch_configuration_sub_setting.lstrip()), mode='w+') except Exception as error: - error_msg = "Error occurred while updating system configuration settings for auto OS updates. [Patch Configuration={0}] [Error={1}]".format(str(patch_configuration_sub_setting), repr(error)) + error_msg = "[APM] Error occurred while updating system configuration settings for auto OS updates. [Patch Configuration={0}] [Error={1}]".format(str(patch_configuration_sub_setting), repr(error)) self.composite_logger.log_error(error_msg) self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR) raise + + def revert_auto_os_update_to_system_default(self): + """ Reverts the auto OS update patch state on the machine to its system default value, if one exists in our backup file """ + # type () -> None + self.composite_logger.log("[APM] Reverting the current automatic OS patch state on the machine to its system default value before patchmode was set to 'AutomaticByPlatform'") + + if not os.path.exists(self.os_patch_configuration_settings_file_path): + self.composite_logger.log_debug("[APM] Automatic OS patch config file not found on the VM. We won't be able to revert auto OS patch settings to their system default values") + return + + self.__get_current_auto_os_updates_setting_on_machine() + self.composite_logger.log_debug("[APM] Logging current configuration settings for auto OS updates. [{0}={1}][{2}={3}]" + .format(str(self.update_package_list), str(self.update_package_list_value), str(self.unattended_upgrade), str(self.unattended_upgrade_value))) + + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() + self.composite_logger.log_debug("[APM] Logging system default configuration settings for auto OS updates. [Settings={0}]".format(str(image_default_patch_configuration_backup))) + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + + if is_backup_valid: + self.update_os_patch_configuration_sub_setting(self.update_package_list, image_default_patch_configuration_backup[self.update_package_list]) + self.update_os_patch_configuration_sub_setting(self.unattended_upgrade, image_default_patch_configuration_backup[self.unattended_upgrade]) + self.composite_logger.log_debug("[APM] Successfully reverted auto OS updates to system default config") + else: + self.composite_logger.log_debug("[APM] Since the backup is invalid or does not exist, we won't be able to revert auto OS patch settings to their system default value") + + def __get_image_default_patch_configuration_backup(self): + """ Get image_default_patch_configuration_backup file""" + image_default_patch_configuration_backup = {} + + # read existing backup since it also contains backup from other update services. We need to preserve any existing data with backup file + if self.image_default_patch_configuration_backup_exists(): + try: + image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) + except Exception as error: + self.composite_logger.log_error("[APM] Unable to read backup for default patch state. [Exception={0}]".format(repr(error))) + return image_default_patch_configuration_backup # endregion # region Reboot Management diff --git a/src/core/src/package_managers/PackageManager.py b/src/core/src/package_managers/PackageManager.py index 62ee261e..b0fa6efe 100644 --- a/src/core/src/package_managers/PackageManager.py +++ b/src/core/src/package_managers/PackageManager.py @@ -430,6 +430,9 @@ def is_image_default_patch_configuration_backup_valid(self, image_default_patch_ @abstractmethod def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_setting, value, patch_configuration_sub_setting_pattern_to_match): pass + + def revert_auto_os_update_to_system_default(self): + pass # endregion # region Handling known errors diff --git a/src/core/src/package_managers/TdnfPackageManager.py b/src/core/src/package_managers/TdnfPackageManager.py index 0be3cfa9..f5b699ca 100644 --- a/src/core/src/package_managers/TdnfPackageManager.py +++ b/src/core/src/package_managers/TdnfPackageManager.py @@ -60,6 +60,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ self.apply_updates_identifier_text = "" self.enable_on_reboot_identifier_text = "" self.enable_on_reboot_check_cmd = '' + self.enable_on_reboot_cmd = '' self.installation_state_identifier_text = "" self.install_check_cmd = "" self.apply_updates_enabled = "Enabled" @@ -375,6 +376,7 @@ def __init_constants_for_dnf_automatic(self): self.dnf_automatic_install_check_cmd = 'systemctl list-unit-files --type=service | grep dnf-automatic.service' # list-unit-files returns installed services, ref: https://www.freedesktop.org/software/systemd/man/systemctl.html#Unit%20File%20Commands self.dnf_automatic_enable_on_reboot_check_cmd = 'systemctl is-enabled dnf-automatic.timer' self.dnf_automatic_disable_on_reboot_cmd = 'systemctl disable dnf-automatic.timer' + self.dnf_automatic_enable_on_reboot_cmd = 'systemctl enable dnf-automatic.timer' self.dnf_automatic_config_pattern_match_text = ' = (no|yes)' self.dnf_automatic_download_updates_identifier_text = 'download_updates' self.dnf_automatic_apply_updates_identifier_text = 'apply_updates' @@ -384,11 +386,11 @@ def __init_constants_for_dnf_automatic(self): def get_current_auto_os_patch_state(self): """ Gets the current auto OS update patch state on the machine """ - self.composite_logger.log("Fetching the current automatic OS patch state on the machine...") + self.composite_logger.log("[TDNF] Fetching the current automatic OS patch state on the machine...") current_auto_os_patch_state_for_dnf_automatic = self.__get_current_auto_os_patch_state_for_dnf_automatic() - self.composite_logger.log("OS patch state per auto OS update service: [dnf-automatic={0}]".format(str(current_auto_os_patch_state_for_dnf_automatic))) + self.composite_logger.log("[TDNF] OS patch state per auto OS update service: [dnf-automatic={0}]".format(str(current_auto_os_patch_state_for_dnf_automatic))) if current_auto_os_patch_state_for_dnf_automatic == Constants.AutomaticOSPatchStates.ENABLED: current_auto_os_patch_state = Constants.AutomaticOSPatchStates.ENABLED @@ -397,12 +399,12 @@ def get_current_auto_os_patch_state(self): else: current_auto_os_patch_state = Constants.AutomaticOSPatchStates.UNKNOWN - self.composite_logger.log_debug("Overall Auto OS Patch State based on all auto OS update service states [OverallAutoOSPatchState={0}]".format(str(current_auto_os_patch_state))) + self.composite_logger.log_debug("[TDNF] Overall Auto OS Patch State based on all auto OS update service states [OverallAutoOSPatchState={0}]".format(str(current_auto_os_patch_state))) return current_auto_os_patch_state def __get_current_auto_os_patch_state_for_dnf_automatic(self): """ Gets current auto OS update patch state for dnf-automatic """ - self.composite_logger.log_debug("Fetching current automatic OS patch state in dnf-automatic service. This includes checks on whether the service is installed, current auto patch enable state and whether it is set to enable on reboot") + self.composite_logger.log_debug("[TDNF] Fetching current automatic OS patch state in dnf-automatic service. This includes checks on whether the service is installed, current auto patch enable state and whether it is set to enable on reboot") self.__init_auto_update_for_dnf_automatic() is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() @@ -425,6 +427,7 @@ def __init_auto_update_for_dnf_automatic(self): self.installation_state_identifier_text = self.dnf_automatic_installation_state_identifier_text self.auto_update_config_pattern_match_text = self.dnf_automatic_config_pattern_match_text self.enable_on_reboot_check_cmd = self.dnf_automatic_enable_on_reboot_check_cmd + self.enable_on_reboot_cmd = self.dnf_automatic_enable_on_reboot_cmd self.install_check_cmd = self.dnf_automatic_install_check_cmd self.current_auto_os_update_service = self.dnf_auto_os_update_service @@ -550,19 +553,12 @@ def backup_image_default_patch_configuration_if_not_exists(self): "apply_updates": "yes/no/empty string", "download_updates": "yes/no/empty string", "enable_on_reboot": true/false, - "install_state": true/false + "installation_state": true/false } } """ try: self.composite_logger.log_debug("[TDNF] Ensuring there is a backup of the default patch state for [AutoOSUpdateService={0}]".format(str(self.current_auto_os_update_service))) - image_default_patch_configuration_backup = {} - - # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file - if self.image_default_patch_configuration_backup_exists(): - try: - image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) - except Exception as error: - self.composite_logger.log_error("Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error))) + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() # verify if existing backup is valid if not, write to backup is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) @@ -640,6 +636,69 @@ def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_sett self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR) raise + def revert_auto_os_update_to_system_default(self): + """ Reverts the auto OS update patch state on the machine to its system default value, if one exists in our backup file """ + # type () -> None + self.composite_logger.log("[TDNF] Reverting the current automatic OS patch state on the machine to its system default value before patchmode was set to 'AutomaticByPlatform'") + self.revert_auto_os_update_to_system_default_for_dnf_automatic() + self.composite_logger.log_debug("[TDNF] Successfully reverted auto OS updates to system default config") + + def revert_auto_os_update_to_system_default_for_dnf_automatic(self): + """ Reverts the auto OS update patch state on the machine to its system default value for given service, if applicable """ + # type () -> None + self.__init_auto_update_for_dnf_automatic() + self.composite_logger.log("[TDNF] Reverting the current automatic OS patch state on the machine to its system default value for [Service={0}]".format(str(self.current_auto_os_update_service))) + is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() + + if not is_service_installed: + self.composite_logger.log_debug("[TDNF] Machine default auto OS update service is not installed on the VM and hence no config to revert. [Service={0}]".format(str(self.current_auto_os_update_service))) + return + + self.composite_logger.log_debug("[TDNF] Logging current configuration settings for auto OS updates [Service={0}][Is_Service_Installed={1}][Machine_default_update_enable_on_reboot={2}][{3}={4}]][{5}={6}]" + .format(str(self.current_auto_os_update_service), str(is_service_installed), str(enable_on_reboot_value), str(self.download_updates_identifier_text), str(download_updates_value), str(self.apply_updates_identifier_text), str(apply_updates_value))) + + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() + self.composite_logger.log_debug("[TDNF] Logging system default configuration settings for auto OS updates. [Settings={0}]".format(str(image_default_patch_configuration_backup))) + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + + if is_backup_valid: + download_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.download_updates_identifier_text] + apply_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.apply_updates_identifier_text] + enable_on_reboot_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.enable_on_reboot_identifier_text] + + self.update_os_patch_configuration_sub_setting(self.download_updates_identifier_text, download_updates_value_from_backup, self.auto_update_config_pattern_match_text) + self.update_os_patch_configuration_sub_setting(self.apply_updates_identifier_text, apply_updates_value_from_backup, self.auto_update_config_pattern_match_text) + if str(enable_on_reboot_value_from_backup).lower() == 'true': + self.enable_auto_update_on_reboot() + else: + self.composite_logger.log_debug("[TDNF] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service={0}]".format(str(self.current_auto_os_update_service))) + + def enable_auto_update_on_reboot(self): + """Enables machine default auto update on reboot""" + # type () -> None + command = self.enable_on_reboot_cmd + self.composite_logger.log_verbose("[TDNF] Enabling auto update on reboot. [Command={0}] ".format(command)) + code, out = self.env_layer.run_command_output(command, False, False) + + if code != 0: + self.composite_logger.log_error("[TDNF][ERROR] Error enabling auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + error_msg = 'Unexpected return code (' + str(code) + ') on command: ' + command + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.OPERATION_FAILED) + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + else: + self.composite_logger.log_debug("[TDNF] Enabled auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + + def __get_image_default_patch_configuration_backup(self): + """ Get image_default_patch_configuration_backup file""" + image_default_patch_configuration_backup = {} + + # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file + if self.image_default_patch_configuration_backup_exists(): + try: + image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) + except Exception as error: + self.composite_logger.log_error("[TDNF] Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error))) + return image_default_patch_configuration_backup # endregion # region Reboot Management diff --git a/src/core/src/package_managers/YumPackageManager.py b/src/core/src/package_managers/YumPackageManager.py index 1773fb75..dc888dd8 100644 --- a/src/core/src/package_managers/YumPackageManager.py +++ b/src/core/src/package_managers/YumPackageManager.py @@ -63,6 +63,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ self.apply_updates_identifier_text = "" self.enable_on_reboot_identifier_text = "" self.enable_on_reboot_check_cmd = '' + self.enable_on_reboot_cmd = '' self.installation_state_identifier_text = "" self.install_check_cmd = "" self.apply_updates_enabled = "Enabled" @@ -415,6 +416,7 @@ def __init_constants_for_yum_cron(self): self.yum_cron_install_check_cmd = 'systemctl list-unit-files --type=service | grep yum-cron.service' # list-unit-files returns installed services, ref: https://www.freedesktop.org/software/systemd/man/systemctl.html#Unit%20File%20Commands self.yum_cron_enable_on_reboot_check_cmd = 'systemctl is-enabled yum-cron' self.yum_cron_disable_on_reboot_cmd = 'systemctl disable yum-cron' + self.yum_cron_enable_on_reboot_cmd = 'systemctl enable yum-cron' self.yum_cron_config_pattern_match_text = ' = (no|yes)' self.yum_cron_download_updates_identifier_text = 'download_updates' self.yum_cron_apply_updates_identifier_text = 'apply_updates' @@ -426,6 +428,7 @@ def __init_constants_for_dnf_automatic(self): self.dnf_automatic_install_check_cmd = 'systemctl list-unit-files --type=service | grep dnf-automatic.service' # list-unit-files returns installed services, ref: https://www.freedesktop.org/software/systemd/man/systemctl.html#Unit%20File%20Commands self.dnf_automatic_enable_on_reboot_check_cmd = 'systemctl is-enabled dnf-automatic.timer' self.dnf_automatic_disable_on_reboot_cmd = 'systemctl disable dnf-automatic.timer' + self.dnf_automatic_enable_on_reboot_cmd = 'systemctl enable dnf-automatic.timer' self.dnf_automatic_config_pattern_match_text = ' = (no|yes)' self.dnf_automatic_download_updates_identifier_text = 'download_updates' self.dnf_automatic_apply_updates_identifier_text = 'apply_updates' @@ -437,6 +440,7 @@ def __init_constants_for_packagekit(self): self.packagekit_install_check_cmd = 'systemctl list-unit-files --type=service | grep packagekit.service' # list-unit-files returns installed services, ref: https://www.freedesktop.org/software/systemd/man/systemctl.html#Unit%20File%20Commands self.packagekit_enable_on_reboot_check_cmd = 'systemctl is-enabled packagekit' self.packagekit_disable_on_reboot_cmd = 'systemctl disable packagekit' + self.packagekit_enable_on_reboot_cmd = 'systemctl enable packagekit' self.packagekit_config_pattern_match_text = ' = (false|true)' self.packagekit_download_updates_identifier_text = 'GetPreparedUpdates' # todo: dummy value, get real value or add telemetry to gather value self.packagekit_apply_updates_identifier_text = 'WritePreparedUpdates' @@ -533,6 +537,7 @@ def __init_auto_update_for_yum_cron(self): self.installation_state_identifier_text = self.yum_cron_installation_state_identifier_text self.auto_update_config_pattern_match_text = self.yum_cron_config_pattern_match_text self.enable_on_reboot_check_cmd = self.yum_cron_enable_on_reboot_check_cmd + self.enable_on_reboot_cmd = self.yum_cron_enable_on_reboot_cmd self.install_check_cmd = self.yum_cron_install_check_cmd self.current_auto_os_update_service = Constants.YumAutoOSUpdateServices.YUM_CRON @@ -545,6 +550,7 @@ def __init_auto_update_for_dnf_automatic(self): self.installation_state_identifier_text = self.dnf_automatic_installation_state_identifier_text self.auto_update_config_pattern_match_text = self.dnf_automatic_config_pattern_match_text self.enable_on_reboot_check_cmd = self.dnf_automatic_enable_on_reboot_check_cmd + self.enable_on_reboot_cmd = self.dnf_automatic_enable_on_reboot_cmd self.install_check_cmd = self.dnf_automatic_install_check_cmd self.current_auto_os_update_service = Constants.YumAutoOSUpdateServices.DNF_AUTOMATIC @@ -557,6 +563,7 @@ def __init_auto_update_for_packagekit(self): self.installation_state_identifier_text = self.packagekit_installation_state_identifier_text self.auto_update_config_pattern_match_text = self.packagekit_config_pattern_match_text self.enable_on_reboot_check_cmd = self.packagekit_enable_on_reboot_check_cmd + self.enable_on_reboot_cmd = self.packagekit_enable_on_reboot_cmd self.install_check_cmd = self.packagekit_install_check_cmd self.current_auto_os_update_service = Constants.YumAutoOSUpdateServices.PACKAGEKIT @@ -646,31 +653,24 @@ def backup_image_default_patch_configuration_if_not_exists(self): "apply_updates": "yes/no/empty string", "download_updates": "yes/no/empty string", "enable_on_reboot": true/false, - "install_state": true/false + "installation_state": true/false }, "dnf-automatic": { "apply_updates": "yes/no/empty string", "download_updates": "yes/no/empty string", "enable_on_reboot": true/false, - "install_state": true/false + "installation_state": true/false }, "packagekit": { "WritePreparedUpdates": "true/false/empty string", "GetPreparedUpdates": "true/false/empty string", //NOTE: This property name is pending validation as noted in another comment where the name is initialized "enable_on_reboot": true/false, - "install_state": true/false + "installation_state": true/false } } """ try: self.composite_logger.log_debug("[YPM] Ensuring there is a backup of the default patch state for [AutoOSUpdateService={0}]".format(str(self.current_auto_os_update_service))) - image_default_patch_configuration_backup = {} - - # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file - if self.image_default_patch_configuration_backup_exists(): - try: - image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) - except Exception as error: - self.composite_logger.log_error("Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error))) + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() # verify if existing backup is valid if not, write to backup is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) @@ -798,6 +798,12 @@ def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_sett try: # note: adding space between the patch_configuration_sub_setting and value since, we will have to do that if we have to add a patch_configuration_sub_setting that did not exist before self.composite_logger.log_debug("[YPM] Updating system configuration settings for auto OS updates. [Patch Configuration Sub Setting={0}] [Value={1}]".format(str(patch_configuration_sub_setting), value)) + + if value == '': + self.composite_logger.log_debug("[YPM] We won't update the system configuration settings since new configuration value to update does not match any of it's acceptable values. [Patch Configuration Sub Setting={0}] [Value To Update={1}][Acceptable Values={2}]" + .format(str(patch_configuration_sub_setting), value, config_pattern_match_text)) + return + os_patch_configuration_settings = self.env_layer.file_system.read_with_retry(self.os_patch_configuration_settings_file_path) patch_configuration_sub_setting_to_update = patch_configuration_sub_setting + ' = ' + value patch_configuration_sub_setting_found_in_file = False @@ -835,6 +841,21 @@ def disable_auto_update_on_reboot(self, command): else: self.composite_logger.log_debug("[YPM] Disabled auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + def enable_auto_update_on_reboot(self): + """Enables machine default auto update on reboot""" + # type () -> None + command = self.enable_on_reboot_cmd + self.composite_logger.log_verbose("[YPM] Enabling auto update on reboot. [Command={0}] ".format(command)) + code, out = self.env_layer.run_command_output(command, False, False) + + if code != 0: + self.composite_logger.log_error("[YPM][ERROR] Error enabling auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + error_msg = 'Unexpected return code (' + str(code) + ') on command: ' + command + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.OPERATION_FAILED) + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + else: + self.composite_logger.log_debug("[YPM] Enabled auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + def is_auto_update_service_installed(self, install_check_cmd): """ Checks if the auto update service is enable_on_reboot on the VM """ code, out = self.env_layer.run_command_output(install_check_cmd, False, False) @@ -845,6 +866,66 @@ def is_auto_update_service_installed(self, install_check_cmd): else: self.composite_logger.log_debug("[YPM] > Auto OS update service is NOT installed on the machine") return False + + def revert_auto_os_update_to_system_default(self): + """ Reverts the auto OS update patch state on the machine to its system default value, if one exists in our backup file """ + # type () -> None + self.composite_logger.log("[YPM] Reverting the current automatic OS patch state on the machine to its system default value before patchmode was set to 'AutomaticByPlatform'") + self.revert_auto_os_update_to_system_default_for_service(Constants.YumAutoOSUpdateServices.YUM_CRON) + self.revert_auto_os_update_to_system_default_for_service(Constants.YumAutoOSUpdateServices.DNF_AUTOMATIC) + self.revert_auto_os_update_to_system_default_for_service(Constants.YumAutoOSUpdateServices.PACKAGEKIT) + self.composite_logger.log_debug("[YPM] Successfully reverted auto OS updates to system default config") + + def revert_auto_os_update_to_system_default_for_service(self, service): + """ Reverts the auto OS update patch state on the machine to its system default value for given service, if applicable """ + # type () -> None + self.composite_logger.log("[YPM] Reverting the current automatic OS patch state on the machine to its system default value for [Service={0}]]".format(str(service))) + self.__init_auto_update_for_service(service) + is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() + + if not is_service_installed: + self.composite_logger.log_debug("[YPM] Machine default auto OS update service is not installed on the VM and hence no config to revert. [Service={0}]".format(str(service))) + return + + self.composite_logger.log_debug("[YPM] Logging current configuration settings for auto OS updates [Service={0}][Is_Service_Installed={1}][Machine_default_update_enable_on_reboot={2}][{3}={4}]][{5}={6}]" + .format(str(self.current_auto_os_update_service), str(is_service_installed), str(enable_on_reboot_value), str(self.download_updates_identifier_text), str(download_updates_value), str(self.apply_updates_identifier_text), str(apply_updates_value))) + + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() + self.composite_logger.log_debug("[YPM] Logging system default configuration settings for auto OS updates. [Settings={0}]".format(str(image_default_patch_configuration_backup))) + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + + if is_backup_valid: + download_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.download_updates_identifier_text] + apply_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.apply_updates_identifier_text] + enable_on_reboot_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.enable_on_reboot_identifier_text] + + self.update_os_patch_configuration_sub_setting(self.download_updates_identifier_text, download_updates_value_from_backup, self.auto_update_config_pattern_match_text) + self.update_os_patch_configuration_sub_setting(self.apply_updates_identifier_text, apply_updates_value_from_backup, self.auto_update_config_pattern_match_text) + if str(enable_on_reboot_value_from_backup).lower() == 'true': + self.enable_auto_update_on_reboot() + else: + self.composite_logger.log_debug("[YPM] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service={0}]".format(str(service))) + + def __init_auto_update_for_service(self, service): + """ Verifies if default auto update configurations, for a service under consideration, are saved in backup """ + switcher = { + Constants.YumAutoOSUpdateServices.YUM_CRON: self.__init_auto_update_for_yum_cron, + Constants.YumAutoOSUpdateServices.DNF_AUTOMATIC: self.__init_auto_update_for_dnf_automatic, + Constants.YumAutoOSUpdateServices.PACKAGEKIT: self.__init_auto_update_for_packagekit + } + return switcher[service]() + + def __get_image_default_patch_configuration_backup(self): + """ Get image_default_patch_configuration_backup file""" + image_default_patch_configuration_backup = {} + + # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file + if self.image_default_patch_configuration_backup_exists(): + try: + image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) + except Exception as error: + self.composite_logger.log_error("[YPM] Unable to read backup for default patch state. [Exception={0}]".format(repr(error))) + return image_default_patch_configuration_backup # endregion # region Handling known errors diff --git a/src/core/src/package_managers/ZypperPackageManager.py b/src/core/src/package_managers/ZypperPackageManager.py index 574d3fd3..6fe5c4d0 100644 --- a/src/core/src/package_managers/ZypperPackageManager.py +++ b/src/core/src/package_managers/ZypperPackageManager.py @@ -570,12 +570,6 @@ def get_package_size(self, output): # endregion # region auto OS updates - # def __init_constants_for_yast2_online_update_configuration(self): - # self.yast2_online_update_configuration_os_patch_configuration_settings_file_path = '/etc/sysconfig/automatic_online_update' - # self.yast2_online_update_configuration_apply_updates_identifier_text = 'AOU_ENABLE_CRONJOB' - # self.yast2_online_update_configuration_auto_update_config_pattern_match_text = '="(true|false)"' - # self.yast2_online_update_configuration_installation_state_identifier_text = "installation_state" - def get_current_auto_os_patch_state(self): """ Gets the current auto OS update patch state on the machine """ self.composite_logger.log_debug("[ZPM] Fetching the current automatic OS patch state on the machine...") @@ -670,10 +664,10 @@ def disable_auto_os_update_for_yast_online_update_configuration(self): self.composite_logger.log_debug("[ZPM] Cannot disable auto updates using yast2-online-update-configuration because the configuration file does not exist, indicating the service is not installed") return - self.composite_logger.log_debug("[ZPM] Preemptively disabling auto OS updates using yum-cron") + self.composite_logger.log_debug("[ZPM] Preemptively disabling auto OS updates using yast2-online-update-configuration") self.update_os_patch_configuration_sub_setting(self.apply_updates_identifier_text, "false", self.auto_update_config_pattern_match_text) - self.composite_logger.log_debug("[ZPM] [ZPM] Successfully disabled auto OS updates using yast2-online-update-configuration") + self.composite_logger.log_debug("[ZPM] Successfully disabled auto OS updates using yast2-online-update-configuration") def backup_image_default_patch_configuration_if_not_exists(self): """ Records the default system settings for auto OS updates within patch extension artifacts for future reference. @@ -681,20 +675,13 @@ def backup_image_default_patch_configuration_if_not_exists(self): """ JSON format for backup file: { "yast2-online-update-configuration": { - "apply_updates": "true/false/empty string" - "install_state": true/false + "AOU_ENABLE_CRONJOB": "true/false/empty string" + "installation_state": true/false } } """ try: self.composite_logger.log_debug("Ensuring there is a backup of the default patch state for [AutoOSUpdateService={0}]".format(str(self.current_auto_os_update_service))) - image_default_patch_configuration_backup = {} - - # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file - if self.image_default_patch_configuration_backup_exists(): - try: - image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) - except Exception as error: - self.composite_logger.log_error("Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error))) + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() # verify if existing backup is valid if not, write to backup is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) @@ -733,16 +720,22 @@ def is_image_default_patch_configuration_backup_valid(self, image_default_patch_ def is_backup_valid_for_yast_online_update_configuration(self, image_default_patch_configuration_backup): if self.ZypperAutoOSUpdateServices.YAST2_ONLINE_UPDATE_CONFIGURATION in image_default_patch_configuration_backup \ and self.apply_updates_identifier_text in image_default_patch_configuration_backup[self.ZypperAutoOSUpdateServices.YAST2_ONLINE_UPDATE_CONFIGURATION]: - self.composite_logger.log_debug("[ZPM] Extension has a valid backup for default yum-cron configuration settings") + self.composite_logger.log_debug("[ZPM] Extension has a valid backup for default yast2-online-update-configuration settings") return True else: - self.composite_logger.log_debug("[ZPM] Extension does not have a valid backup for default yum-cron configuration settings") + self.composite_logger.log_debug("[ZPM] Extension does not have a valid backup for default yast2-online-update-configuration settings") return False def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_setting, value="false", config_pattern_match_text=""): """ Updates (or adds if it doesn't exist) the given patch_configuration_sub_setting with the given value in os_patch_configuration_settings_file """ try: self.composite_logger.log_debug("[ZPM] Updating system configuration settings for auto OS updates. [Patch Configuration Sub Setting={0}] [Value={1}]".format(str(patch_configuration_sub_setting), value)) + + if value == '': + self.composite_logger.log_debug("[ZPM] We won't update the system configuration settings since new configuration value to update does not match any of it's acceptable values. [Patch Configuration Sub Setting={0}] [Value To Update={1}][Acceptable Values={2}]" + .format(str(patch_configuration_sub_setting), value, config_pattern_match_text)) + return + os_patch_configuration_settings = self.env_layer.file_system.read_with_retry(self.os_patch_configuration_settings_file_path) patch_configuration_sub_setting_to_update = patch_configuration_sub_setting + '="' + value + '"' patch_configuration_sub_setting_found_in_file = False @@ -767,6 +760,43 @@ def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_sett self.composite_logger.log_error("[ZPM] " + error_msg) self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR) raise + + def revert_auto_os_update_to_system_default(self): + """ Reverts the auto OS update patch state on the machine to its system default value, if one exists in our backup file """ + # type () -> None + self.composite_logger.log("[ZPM] Reverting the current automatic OS patch state on the machine to its system default value before patchmode was set to 'AutomaticByPlatform'") + self.__init_auto_update_for_yast_online_update_configuration() + is_service_installed, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() + + if not is_service_installed: + self.composite_logger.log_debug("[ZPM] Machine default auto OS update service is not installed on the VM and hence no config to revert. [Service={0}]".format(str(self.current_auto_os_update_service))) + return + + self.composite_logger.log_debug("[ZPM] Logging current configuration settings for auto OS updates [Service={0}][Is_Service_Installed={1}][{2}={3}]" + .format(str(self.current_auto_os_update_service), str(is_service_installed), str(self.apply_updates_identifier_text), str(apply_updates_value))) + + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() + self.composite_logger.log_debug("[ZPM] Logging system default configuration settings for auto OS updates. [Settings={0}]".format(str(image_default_patch_configuration_backup))) + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + + if is_backup_valid: + apply_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.apply_updates_identifier_text] + self.update_os_patch_configuration_sub_setting(self.apply_updates_identifier_text, apply_updates_value_from_backup, self.auto_update_config_pattern_match_text) + self.composite_logger.log_debug("[ZPM] Successfully reverted auto OS updates to system default config") + else: + self.composite_logger.log_debug("[ZPM] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service={0}]".format(str(self.current_auto_os_update_service))) + + def __get_image_default_patch_configuration_backup(self): + """ Get image_default_patch_configuration_backup file""" + image_default_patch_configuration_backup = {} + + # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file + if self.image_default_patch_configuration_backup_exists(): + try: + image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) + except Exception as error: + self.composite_logger.log_error("[ZPM] Unable to read backup for default patch state. [Exception={0}]".format(repr(error))) + return image_default_patch_configuration_backup # endregion # region Reboot Management diff --git a/src/core/tests/Test_AptitudePackageManager.py b/src/core/tests/Test_AptitudePackageManager.py index c014a1e0..afc48095 100644 --- a/src/core/tests/Test_AptitudePackageManager.py +++ b/src/core/tests/Test_AptitudePackageManager.py @@ -16,6 +16,13 @@ import json import os import unittest +import sys +# Conditional import for StringIO +try: + from StringIO import StringIO # Python 2 +except ImportError: + from io import StringIO # Python 3 + from core.src.bootstrap.Constants import Constants from core.src.core_logic.ExecutionConfig import ExecutionConfig from core.tests.Test_UbuntuProClient import MockVersionResult, MockRebootRequiredResult, MockUpdatesResult @@ -67,6 +74,56 @@ def mock_get_security_updates_return_empty_list(self): # endregion Mocks + # region Utility Functions + def __setup_config_and_invoke_revert_auto_os_to_system_default(self, package_manager, create_current_auto_os_config=True, create_backup_for_system_default_config=True, current_auto_os_update_config_value='', update_package_list_value="", set_unattended_upgrade=True, unattended_upgrade_value=""): + """ Sets up current auto OS update config, backup for system default config (if requested) and invoke revert to system default """ + # setup current auto OS update config + if create_current_auto_os_config: + self.__setup_current_auto_os_update_config(package_manager, current_auto_os_update_config_value) + + # setup backup for system default auto OS update config + if create_backup_for_system_default_config: + self.__setup_backup_for_system_default_OS_update_config(package_manager, update_package_list_value=update_package_list_value, set_unattended_upgrade=set_unattended_upgrade, unattended_upgrade_value=unattended_upgrade_value) + + package_manager.revert_auto_os_update_to_system_default() + + def __setup_current_auto_os_update_config(self, package_manager, config_value='', config_file_name="20auto-upgrades"): + # setup current auto OS update config + package_manager.os_patch_configuration_settings_file_path = os.path.join(self.runtime.execution_config.config_folder, config_file_name) + self.runtime.write_to_file(package_manager.os_patch_configuration_settings_file_path, config_value) + + def __setup_backup_for_system_default_OS_update_config(self, package_manager, update_package_list_value="", set_unattended_upgrade=True, unattended_upgrade_value=""): + # setup backup for system default auto OS update config + package_manager.image_default_patch_configuration_backup_path = os.path.join(self.runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + backup_image_default_patch_configuration_json = { + package_manager.update_package_list: update_package_list_value + } + if set_unattended_upgrade: + backup_image_default_patch_configuration_json[package_manager.unattended_upgrade] = unattended_upgrade_value + self.runtime.write_to_file(package_manager.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(backup_image_default_patch_configuration_json))) + + @staticmethod + def __capture_std_io(): + # arrange capture std IO + captured_output = StringIO() + original_stdout = sys.stdout + sys.stdout = captured_output + return captured_output, original_stdout + + def __assert_std_io(self, captured_output, expected_output=''): + output = captured_output.getvalue() + self.assertTrue(expected_output in output) + + def __assert_reverted_automatic_patch_configuration_settings(self, package_manager, config_exists=True, update_package_list_value_expected='', unattended_upgrade_value_expected=''): + if config_exists: + reverted_os_patch_configuration_settings = self.runtime.env_layer.file_system.read_with_retry(package_manager.os_patch_configuration_settings_file_path) + self.assertTrue(reverted_os_patch_configuration_settings is not None) + self.assertTrue(update_package_list_value_expected in reverted_os_patch_configuration_settings) + self.assertTrue(unattended_upgrade_value_expected in reverted_os_patch_configuration_settings) + else: + self.assertFalse(os.path.exists(package_manager.os_patch_configuration_settings_file_path)) + # endregion + def test_package_manager_no_updates(self): """Unit test for aptitude package manager with no updates""" self.runtime.set_legacy_test_type('SadPath') @@ -407,6 +464,157 @@ def test_update_image_default_patch_mode_raises_exception(self): self.runtime.env_layer.file_system.write_with_retry = self.mock_write_with_retry_raise_exception self.assertRaises(Exception, package_manager.update_os_patch_configuration_sub_setting) + def test_revert_auto_os_update_to_system_default(self): + revert_success_testcase = { + "stdio": { + "capture_output": False, + "expected_output": '' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'APT::Periodic::Update-Package-Lists "0";\nAPT::Periodic::Unattended-Upgrade "0";\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "update_package_list_value": "1", + "unattended_upgrade_value": "1", + "set_unattended_upgrade": True + } + }, + "assertions": { + "update_package_list_value_expected": 'APT::Periodic::Update-Package-Lists "1"', + "unattended_upgrade_value_expected": 'APT::Periodic::Unattended-Upgrade "1"', + "config_exists": True + } + } + + revert_success_auto_os_update_config_does_not_exist = { + "stdio": { + "capture_output": True, + "expected_output": 'Automatic OS patch config file not found on the VM. We won\'t be able to revert auto OS patch settings to their system default values' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": False, + "current_auto_os_update_config_value": '' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "update_package_list_value": "1", + "unattended_upgrade_value": "1", + "set_unattended_upgrade": True + } + }, + "assertions": { + "update_package_list_value_expected": '', + "unattended_upgrade_value_expected": '', + "config_exists": False + } + } + + revert_success_backup_config_does_not_exist = { + "stdio": { + "capture_output": True, + "expected_output": 'Since the backup is invalid or does not exist, we won\'t be able to revert auto OS patch settings to their system default value' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'APT::Periodic::Update-Package-Lists "0";\nAPT::Periodic::Unattended-Upgrade "0";\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": False, + "update_package_list_value": "", + "unattended_upgrade_value": "", + "set_unattended_upgrade": False + } + }, + "assertions": { + "update_package_list_value_expected": 'APT::Periodic::Update-Package-Lists "0"', + "unattended_upgrade_value_expected": 'APT::Periodic::Unattended-Upgrade "0"', + "config_exists": True + } + } + + revert_success_backup_config_invalid = { + "stdio": { + "capture_output": True, + "expected_output": 'Since the backup is invalid or does not exist, we won\'t be able to revert auto OS patch settings to their system default value' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'APT::Periodic::Update-Package-Lists "0";\nAPT::Periodic::Unattended-Upgrade "0";\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "update_package_list_value": "1", + "unattended_upgrade_value": "", + "set_unattended_upgrade": False + } + }, + "assertions": { + "update_package_list_value_expected": 'APT::Periodic::Update-Package-Lists "0"', + "unattended_upgrade_value_expected": 'APT::Periodic::Unattended-Upgrade "0"', + "config_exists": True + } + } + + revert_success_backup_config_contains_empty_values = { + "stdio": { + "capture_output": False, + "expected_output": '' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'APT::Periodic::Update-Package-Lists "0";\nAPT::Periodic::Unattended-Upgrade "0";\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "update_package_list_value": "1", + "unattended_upgrade_value": "", + "set_unattended_upgrade": True + } + }, + "assertions": { + "update_package_list_value_expected": 'APT::Periodic::Update-Package-Lists "1"', + "unattended_upgrade_value_expected": 'APT::Periodic::Unattended-Upgrade "0"', + "config_exists": True + } + } + + all_testcases = [revert_success_testcase, revert_success_auto_os_update_config_does_not_exist, revert_success_backup_config_does_not_exist, revert_success_backup_config_invalid, revert_success_backup_config_contains_empty_values] + + for testcase in all_testcases: + self.tearDown() + self.setUp() + captured_output, original_stdout = None, None + if testcase["stdio"]["capture_output"]: + # arrange capture std IO + captured_output, original_stdout = self.__capture_std_io() + + package_manager = self.container.get('package_manager') + + # setup current auto OS update config, backup for system default config and invoke revert to system default + self.__setup_config_and_invoke_revert_auto_os_to_system_default(package_manager, + create_current_auto_os_config=bool(testcase["config"]["current_auto_update_config"]["create_current_auto_os_config"]), + current_auto_os_update_config_value=testcase["config"]["current_auto_update_config"]["current_auto_os_update_config_value"], + create_backup_for_system_default_config=bool(testcase["config"]["backup_system_default_config"]["create_backup_for_system_default_config"]), + update_package_list_value=testcase["config"]["backup_system_default_config"]["update_package_list_value"], + unattended_upgrade_value=testcase["config"]["backup_system_default_config"]["unattended_upgrade_value"], + set_unattended_upgrade=bool(testcase["config"]["backup_system_default_config"]["set_unattended_upgrade"])) + + # assert + if testcase["stdio"]["capture_output"]: + # restore sys.stdout output + sys.stdout = original_stdout + self.__assert_std_io(captured_output=captured_output, expected_output=testcase["stdio"]["expected_output"]) + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_exists=bool(testcase["assertions"]["config_exists"]), + update_package_list_value_expected=testcase["assertions"]["update_package_list_value_expected"], + unattended_upgrade_value_expected=testcase["assertions"]["unattended_upgrade_value_expected"]) + def test_is_reboot_pending_prerequisite_not_met_should_return_false(self): package_manager = self.container.get('package_manager') package_manager._AptitudePackageManager__pro_client_prereq_met = False diff --git a/src/core/tests/Test_ConfigurePatchingProcessor.py b/src/core/tests/Test_ConfigurePatchingProcessor.py index 11f1d38c..04242e0d 100644 --- a/src/core/tests/Test_ConfigurePatchingProcessor.py +++ b/src/core/tests/Test_ConfigurePatchingProcessor.py @@ -376,6 +376,58 @@ def test_configure_patching_raise_exception_auto_assessment_systemd(self): runtime.stop() + def test_configure_patching_with_patch_mode_set_to_image_default(self): + # create and adjust arguments + argument_composer = ArgumentComposer() + argument_composer.operation = Constants.CONFIGURE_PATCHING + argument_composer.patch_mode = Constants.PatchModes.IMAGE_DEFAULT + argument_composer.assessment_mode = Constants.AssessmentModes.AUTOMATIC_BY_PLATFORM + + # create and patch runtime + runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT) + runtime.package_manager.get_current_auto_os_patch_state = runtime.backup_get_current_auto_os_patch_state + runtime.package_manager.os_patch_configuration_settings_file_path = os.path.join(runtime.execution_config.config_folder, "20auto-upgrades") + runtime.package_manager.image_default_patch_configuration_backup_path = os.path.join(runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + runtime.set_legacy_test_type('HappyPath') + + # mock os patch configuration + os_patch_configuration_settings = 'APT::Periodic::Update-Package-Lists "0";\nAPT::Periodic::Unattended-Upgrade "0";\n' + runtime.write_to_file(runtime.package_manager.os_patch_configuration_settings_file_path, os_patch_configuration_settings) + + # mock backup for system default auto OS update config + backup_image_default_patch_configuration_json = { + runtime.package_manager.update_package_list: "1", + runtime.package_manager.unattended_upgrade: "1" + } + runtime.write_to_file(runtime.package_manager.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(backup_image_default_patch_configuration_json))) + + # execute Core + CoreMain(argument_composer.get_composed_arguments()) + + # check telemetry events + self.__check_telemetry_events(runtime) + + # check status file for configure patching patch mode + with runtime.env_layer.file_system.open(runtime.execution_config.status_file_path, 'r') as file_handle: + substatus_file_data = json.load(file_handle)[0]["status"]["substatus"] + + # check status file for configure patching patch state + self.assertTrue(runtime.package_manager.image_default_patch_configuration_backup_exists()) + self.assertEqual(len(substatus_file_data), 2) + self.assertTrue(substatus_file_data[1]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) + self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_SUCCESS.lower()) + message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) + self.assertEqual(message["automaticOSPatchState"], Constants.AutomaticOSPatchStates.ENABLED) # auto OS updates are disabled on patch mode 'AutomaticByPlatform' + self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) + self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) + + # check status file for configure patching assessment state + message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) + self.assertEqual(message["autoAssessmentStatus"]["autoAssessmentState"], Constants.AutoAssessmentStates.ENABLED) # auto assessment is enabled + + # stop test runtime + runtime.stop() + def __check_telemetry_events(self, runtime): all_events = os.listdir(runtime.telemetry_writer.events_folder_path) self.assertTrue(len(all_events) > 0) diff --git a/src/core/tests/Test_TdnfPackageManager.py b/src/core/tests/Test_TdnfPackageManager.py index ddaa9852..f7f53922 100644 --- a/src/core/tests/Test_TdnfPackageManager.py +++ b/src/core/tests/Test_TdnfPackageManager.py @@ -16,6 +16,13 @@ import json import os import unittest +import sys +# Conditional import for StringIO +try: + from StringIO import StringIO # Python 2 +except ImportError: + from io import StringIO # Python 3 + from core.src.bootstrap.Constants import Constants from core.tests.library.LegacyEnvLayerExtensions import LegacyEnvLayerExtensions from core.tests.library.ArgumentComposer import ArgumentComposer @@ -41,6 +48,61 @@ def mock_write_with_retry_raise_exception(self, file_path_or_handle, data, mode= raise Exception # endregion + # region Utility Functions + def __setup_config_and_invoke_revert_auto_os_to_system_default(self, package_manager, create_current_auto_os_config=True, create_backup_for_system_default_config=True, current_auto_os_update_config_value='', apply_updates_value="", + download_updates_value="", enable_on_reboot_value=False, installation_state_value=False, set_installation_state=True): + """ Sets up current auto OS update config, backup for system default config (if requested) and invoke revert to system default """ + # setup current auto OS update config + if create_current_auto_os_config: + self.__setup_current_auto_os_update_config(package_manager, current_auto_os_update_config_value) + + # setup backup for system default auto OS update config + if create_backup_for_system_default_config: + self.__setup_backup_for_system_default_OS_update_config(package_manager, apply_updates_value=apply_updates_value, download_updates_value=download_updates_value, enable_on_reboot_value=enable_on_reboot_value, + installation_state_value=installation_state_value, set_installation_state=set_installation_state) + + package_manager.revert_auto_os_update_to_system_default() + + def __setup_current_auto_os_update_config(self, package_manager, config_value='', config_file_name="automatic.conf"): + # setup current auto OS update config + package_manager.dnf_automatic_configuration_file_path = os.path.join(self.runtime.execution_config.config_folder, config_file_name) + self.runtime.write_to_file(package_manager.dnf_automatic_configuration_file_path, config_value) + + def __setup_backup_for_system_default_OS_update_config(self, package_manager, apply_updates_value="", download_updates_value="", enable_on_reboot_value=False, installation_state_value=False, set_installation_state=True): + # setup backup for system default auto OS update config + package_manager.image_default_patch_configuration_backup_path = os.path.join(self.runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + backup_image_default_patch_configuration_json = { + "dnf-automatic": { + "apply_updates": apply_updates_value, + "download_updates": download_updates_value, + "enable_on_reboot": enable_on_reboot_value + } + } + if set_installation_state: + backup_image_default_patch_configuration_json["dnf-automatic"]["installation_state"] = installation_state_value + self.runtime.write_to_file(package_manager.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(backup_image_default_patch_configuration_json))) + + @staticmethod + def __capture_std_io(): + # arrange capture std IO + captured_output = StringIO() + original_stdout = sys.stdout + sys.stdout = captured_output + return captured_output, original_stdout + + def __assert_std_io(self, captured_output, expected_output=''): + output = captured_output.getvalue() + self.assertTrue(expected_output in output) + + def __assert_reverted_automatic_patch_configuration_settings(self, package_manager, config_exists=True, config_value_expected=''): + if config_exists: + reverted_dnf_automatic_patch_configuration_settings = self.runtime.env_layer.file_system.read_with_retry(package_manager.dnf_automatic_configuration_file_path) + self.assertTrue(reverted_dnf_automatic_patch_configuration_settings is not None) + self.assertTrue(config_value_expected in reverted_dnf_automatic_patch_configuration_settings) + else: + self.assertFalse(os.path.exists(package_manager.dnf_automatic_configuration_file_path)) + # endregion + def test_do_processes_require_restart(self): """Unit test for tdnf package manager""" # Restart required @@ -647,6 +709,168 @@ def test_get_current_auto_os_patch_state_with_installed_services_and_state_unkno self.assertFalse(package_manager.image_default_patch_configuration_backup_exists()) self.assertEqual(current_auto_os_patch_state, Constants.AutomaticOSPatchStates.UNKNOWN) + def test_revert_auto_os_update_to_system_default(self): + revert_success_testcase = { + "legacy_type": 'HappyPath', + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'apply_updates = no\ndownload_updates = no\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "yes", + "download_updates_value": "yes", + "enable_on_reboot_value": True, + "installation_state_value": True, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": 'apply_updates = yes\ndownload_updates = yes\n', + "config_exists": True + } + } + + revert_success_with_dnf_not_installed_testcase = { + "legacy_type": 'SadPath', + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": False, + "current_auto_os_update_config_value": '' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "", + "download_updates_value": "", + "enable_on_reboot_value": False, + "installation_state_value": False, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": "", + "config_exists": False + } + } + + revert_success_with_dnf_installed_but_no_config_value_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'test_value = yes\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "", + "download_updates_value": "", + "enable_on_reboot_value": False, + "installation_state_value": False, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": 'download_updates =\napply_updates = \n', + "config_exists": True + } + } + + revert_success_backup_config_does_not_exist_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": True, + "expected_output": "[TDNF] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service=dnf-automatic]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'apply_updates = no\ndownload_updates = no\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": False, + "apply_updates_value": "", + "download_updates_value": "", + "enable_on_reboot_value": False, + "installation_state_value": False, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": 'apply_updates = no\ndownload_updates = no\n', + "config_exists": True + } + } + + revert_success_default_backup_config_invalid_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": True, + "expected_output": "[TDNF] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service=dnf-automatic]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'apply_updates = no\ndownload_updates = no\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "yes", + "download_updates_value": "yes", + "enable_on_reboot_value": True, + "installation_state_value": False, + "set_installation_state": False + } + }, + "assertions": { + "config_value_expected": 'apply_updates = no\ndownload_updates = no\n', + "config_exists": True + } + } + + all_testcases = [revert_success_testcase, revert_success_with_dnf_not_installed_testcase, revert_success_with_dnf_installed_but_no_config_value_testcase, revert_success_backup_config_does_not_exist_testcase, revert_success_default_backup_config_invalid_testcase] + + for testcase in all_testcases: + self.tearDown() + self.setUp() + captured_output, original_stdout = None, None + if testcase["stdio"]["capture_output"]: + # arrange capture std IO + captured_output, original_stdout = self.__capture_std_io() + + self.runtime.set_legacy_test_type(testcase["legacy_type"]) + package_manager = self.container.get('package_manager') + + # setup current auto OS update config, backup for system default config and invoke revert to system default + self.__setup_config_and_invoke_revert_auto_os_to_system_default(package_manager, + create_current_auto_os_config=bool(testcase["config"]["current_auto_update_config"]["create_current_auto_os_config"]), + current_auto_os_update_config_value=testcase["config"]["current_auto_update_config"]["current_auto_os_update_config_value"], + create_backup_for_system_default_config=bool(testcase["config"]["backup_system_default_config"]["create_backup_for_system_default_config"]), + apply_updates_value=testcase["config"]["backup_system_default_config"]["apply_updates_value"], + download_updates_value=testcase["config"]["backup_system_default_config"]["download_updates_value"], + enable_on_reboot_value=bool(testcase["config"]["backup_system_default_config"]["enable_on_reboot_value"]), + installation_state_value=bool(testcase["config"]["backup_system_default_config"]["installation_state_value"]), + set_installation_state=bool(testcase["config"]["backup_system_default_config"]["set_installation_state"])) + + # assert + if testcase["stdio"]["capture_output"]: + # restore sys.stdout output + sys.stdout = original_stdout + self.__assert_std_io(captured_output=captured_output, expected_output=testcase["stdio"]["expected_output"]) + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_exists=bool(testcase["assertions"]["config_exists"]), config_value_expected=testcase["assertions"]["config_value_expected"]) + if __name__ == '__main__': unittest.main() diff --git a/src/core/tests/Test_YumPackageManager.py b/src/core/tests/Test_YumPackageManager.py index 08d40414..44d0027a 100644 --- a/src/core/tests/Test_YumPackageManager.py +++ b/src/core/tests/Test_YumPackageManager.py @@ -16,6 +16,13 @@ import json import os import unittest +import sys +# Conditional import for StringIO +try: + from StringIO import StringIO # Python 2 +except ImportError: + from io import StringIO # Python 3 + from core.src.bootstrap.Constants import Constants from core.tests.library.ArgumentComposer import ArgumentComposer from core.tests.library.LegacyEnvLayerExtensions import LegacyEnvLayerExtensions @@ -44,6 +51,109 @@ def mock_linux8_distribution_to_return_redhat(self): return ['Red Hat Enterprise Linux Server', '8', 'Ootpa'] #endregion Mocks + # region Utility Functions + def __setup_config_and_invoke_revert_auto_os_to_system_default(self, package_manager, create_current_auto_os_config=True, create_backup_for_system_default_config=True, + set_yum_cron=True, yum_cron_config_value='', set_dnf_automatic=True, dnf_automatic_config_value='', + set_packagekit=True, packagekit_config_value='', + yum_cron_apply_updates_value="", yum_cron_download_updates_value="", yum_cron_enable_on_reboot_value=False, yum_cron_installation_state_value=False, yum_cron_set_installation_state=True, + dnf_automatic_apply_updates_value="", dnf_automatic_download_updates_value="", dnf_automatic_enable_on_reboot_value=False, dnf_automatic_installation_state_value=False, dnf_automatic_set_installation_state=True, + packagekit_apply_updates_value="", packagekit_download_updates_value="", packagekit_enable_on_reboot_value=False, packagekit_installation_state_value=False, packagekit_set_installation_state=True): + """ Sets up current auto OS update config, backup for system default config (if requested) and invoke revert to system default """ + # setup current auto OS update config + if create_current_auto_os_config: + self.__setup_all_current_auto_os_update_config(package_manager, set_yum_cron=set_yum_cron, yum_cron_config_value=yum_cron_config_value, + set_dnf_automatic=set_dnf_automatic, dnf_automatic_config_value=dnf_automatic_config_value, + set_packagekit=set_packagekit, packagekit_config_value=packagekit_config_value) + + # setup backup for system default auto OS update config + if create_backup_for_system_default_config: + self.__setup_backup_for_system_default_OS_update_config(package_manager, yum_cron_apply_updates_value=yum_cron_apply_updates_value, yum_cron_download_updates_value=yum_cron_download_updates_value, yum_cron_enable_on_reboot_value=yum_cron_enable_on_reboot_value, + yum_cron_installation_state_value=yum_cron_installation_state_value, yum_cron_set_installation_state=yum_cron_set_installation_state, + dnf_automatic_apply_updates_value=dnf_automatic_apply_updates_value, dnf_automatic_download_updates_value=dnf_automatic_download_updates_value, dnf_automatic_enable_on_reboot_value=dnf_automatic_enable_on_reboot_value, + dnf_automatic_installation_state_value=dnf_automatic_installation_state_value, dnf_automatic_set_installation_state=dnf_automatic_set_installation_state, + packagekit_apply_updates_value=packagekit_apply_updates_value, packagekit_download_updates_value=packagekit_download_updates_value, packagekit_enable_on_reboot_value=packagekit_enable_on_reboot_value, + packagekit_installation_state_value=packagekit_installation_state_value, packagekit_set_installation_state=packagekit_set_installation_state) + + package_manager.revert_auto_os_update_to_system_default() + + def __setup_auto_os_update_config_and_return_file_path(self, config_value='', config_file_name=''): + config_file_path = os.path.join(self.runtime.execution_config.config_folder, config_file_name) + self.runtime.write_to_file(config_file_path, config_value) + return config_file_path + + def __setup_all_current_auto_os_update_config(self, package_manager, set_yum_cron=True, yum_cron_config_value='', set_dnf_automatic=True, dnf_automatic_config_value='', set_packagekit=True, packagekit_config_value=''): + # setup current auto OS update config + if set_yum_cron: + package_manager.yum_cron_configuration_settings_file_path = self.__setup_auto_os_update_config_and_return_file_path(config_value=yum_cron_config_value, config_file_name="yum-cron.conf") + if set_dnf_automatic: + package_manager.dnf_automatic_configuration_file_path = self.__setup_auto_os_update_config_and_return_file_path(config_value=dnf_automatic_config_value, config_file_name="automatic.conf") + if set_packagekit: + package_manager.packagekit_configuration_file_path = self.__setup_auto_os_update_config_and_return_file_path(config_value=packagekit_config_value, config_file_name="PackageKit.conf") + + def __setup_backup_for_system_default_OS_update_config(self, package_manager, + yum_cron_apply_updates_value="", yum_cron_download_updates_value="", yum_cron_enable_on_reboot_value=False, yum_cron_installation_state_value=False, yum_cron_set_installation_state=True, + dnf_automatic_apply_updates_value="", dnf_automatic_download_updates_value="", dnf_automatic_enable_on_reboot_value=False, dnf_automatic_installation_state_value=False, dnf_automatic_set_installation_state=True, + packagekit_apply_updates_value="", packagekit_download_updates_value="", packagekit_enable_on_reboot_value=False, packagekit_installation_state_value=False, packagekit_set_installation_state=True): + # setup backup for system default auto OS update config + package_manager.image_default_patch_configuration_backup_path = os.path.join(self.runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + backup_image_default_patch_configuration_json = { + "yum-cron": self.__set_config_json(apply_updates_value=yum_cron_apply_updates_value, download_updates_value=yum_cron_download_updates_value, + enable_on_reboot_value=yum_cron_enable_on_reboot_value, installation_state_value=yum_cron_installation_state_value, set_installation_state=yum_cron_set_installation_state), + "dnf-automatic": self.__set_config_json(apply_updates_value=dnf_automatic_apply_updates_value, download_updates_value=dnf_automatic_download_updates_value, + enable_on_reboot_value=dnf_automatic_enable_on_reboot_value, installation_state_value=dnf_automatic_installation_state_value, set_installation_state=dnf_automatic_set_installation_state), + "packagekit": self.__set_config_json(apply_updates_value=packagekit_apply_updates_value, download_updates_value=packagekit_download_updates_value, + enable_on_reboot_value=packagekit_enable_on_reboot_value, installation_state_value=packagekit_installation_state_value, set_installation_state=packagekit_set_installation_state, is_packagekit=True) + } + + self.runtime.write_to_file(package_manager.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(backup_image_default_patch_configuration_json))) + + @staticmethod + def __set_config_json(apply_updates_value="", download_updates_value="", enable_on_reboot_value=False, installation_state_value=False, set_installation_state=True, is_packagekit=False): + image_default_patch_configuration_json = { + "enable_on_reboot": enable_on_reboot_value + } + if is_packagekit: + image_default_patch_configuration_json["WritePreparedUpdates"] = apply_updates_value + image_default_patch_configuration_json["GetPreparedUpdates"] = download_updates_value + else: + image_default_patch_configuration_json["apply_updates"] = apply_updates_value + image_default_patch_configuration_json["download_updates"] = download_updates_value + if set_installation_state: + image_default_patch_configuration_json["installation_state"] = installation_state_value + return image_default_patch_configuration_json + + @staticmethod + def __capture_std_io(): + # arrange capture std IO + captured_output = StringIO() + original_stdout = sys.stdout + sys.stdout = captured_output + return captured_output, original_stdout + + def __assert_std_io(self, captured_output, expected_output=''): + output = captured_output.getvalue() + self.assertTrue(expected_output in output) + + def __assert_all_reverted_automatic_patch_configuration_settings(self, package_manager, yum_cron_config_exists=True, yum_cron_apply_updates_expected='', yum_cron_download_updates_expected='', + dnf_automatic_config_exists=True, dnf_automatic_apply_updates_expected='', dnf_automatic_download_updates_expected='', + packagekit_config_exists=True, packagekit_apply_updates_expected='', packagekit_download_updates_expected=''): + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_file_path=package_manager.yum_cron_configuration_settings_file_path, config_exists=yum_cron_config_exists, + apply_updates_value_expected=yum_cron_apply_updates_expected, download_updates_value_expected=yum_cron_download_updates_expected) + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_file_path=package_manager.dnf_automatic_configuration_file_path, config_exists=dnf_automatic_config_exists, + apply_updates_value_expected=dnf_automatic_apply_updates_expected, download_updates_value_expected=dnf_automatic_download_updates_expected) + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_file_path=package_manager.packagekit_configuration_file_path, config_exists=packagekit_config_exists, + apply_updates_value_expected=packagekit_apply_updates_expected, download_updates_value_expected=packagekit_download_updates_expected) + + def __assert_reverted_automatic_patch_configuration_settings(self, package_manager, config_file_path, config_exists=True, apply_updates_value_expected='', download_updates_value_expected=''): + if config_exists: + reverted_patch_configuration_settings = self.runtime.env_layer.file_system.read_with_retry(config_file_path) + self.assertTrue(reverted_patch_configuration_settings is not None) + self.assertTrue(apply_updates_value_expected in reverted_patch_configuration_settings) + self.assertTrue(download_updates_value_expected in reverted_patch_configuration_settings) + else: + self.assertFalse(os.path.exists(package_manager.dnf_automatic_configuration_file_path)) + # endregion + def mock_do_processes_require_restart_raise_exception(self): raise Exception @@ -648,6 +758,392 @@ def test_update_image_default_patch_mode_raises_exception(self): self.runtime.env_layer.file_system.write_with_retry = self.mock_write_with_retry_raise_exception self.assertRaises(Exception, package_manager.update_os_patch_configuration_sub_setting) + def test_revert_auto_os_update_to_system_default(self): + revert_success_testcase = { + "legacy_type": 'HappyPath', + "stdio": { + "capture_output": False, + "expected_output": '' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "yum_cron": { + "set_yum_cron": True, + "yum_cron_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "dnf_automatic": { + "set_dnf_automatic": True, + "dnf_automatic_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "packagekit": { + "set_packagekit": True, + "packagekit_config_value": "WritePreparedUpdates = false\nGetPreparedUpdates = false\n" + } + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "yum_cron": { + "yum_cron_apply_updates_value": "yes", + "yum_cron_download_updates_value": "yes", + "yum_cron_enable_on_reboot_value": True, + "yum_cron_installation_state_value": True, + "yum_cron_set_installation_state": True + }, + "dnf_automatic": { + "dnf_automatic_apply_updates_value": "yes", + "dnf_automatic_download_updates_value": "yes", + "dnf_automatic_enable_on_reboot_value": True, + "dnf_automatic_installation_state_value": True, + "dnf_automatic_set_installation_state": True + }, + "packagekit": { + "packagekit_apply_updates_value": "true", + "packagekit_download_updates_value": "true", + "packagekit_enable_on_reboot_value": True, + "packagekit_installation_state_value": True, + "packagekit_set_installation_state": True + } + } + }, + "assertions": { + "yum_cron": { + "yum_cron_config_exists": True, + "yum_cron_apply_updates_expected": 'apply_updates = yes', + "yum_cron_download_updates_expected": 'download_updates = yes', + }, + "dnf_automatic": { + "dnf_automatic_config_exists": True, + "dnf_automatic_apply_updates_expected": 'apply_updates = yes', + "dnf_automatic_download_updates_expected": 'download_updates = yes', + }, + "packagekit": { + "packagekit_config_exists": True, + "packagekit_apply_updates_expected": 'WritePreparedUpdates = true', + "packagekit_download_updates_expected": 'GetPreparedUpdates = true', + } + } + } + + revert_success_with_only_yum_cron_installed_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": False, + "expected_output": '' + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "yum_cron": { + "set_yum_cron": True, + "yum_cron_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "dnf_automatic": { + "set_dnf_automatic": False, + "dnf_automatic_config_value": "" + }, + "packagekit": { + "set_packagekit": False, + "packagekit_config_value": "" + } + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "yum_cron": { + "yum_cron_apply_updates_value": "yes", + "yum_cron_download_updates_value": "yes", + "yum_cron_enable_on_reboot_value": True, + "yum_cron_installation_state_value": True, + "yum_cron_set_installation_state": True + }, + "dnf_automatic": { + "dnf_automatic_apply_updates_value": "", + "dnf_automatic_download_updates_value": "", + "dnf_automatic_enable_on_reboot_value": False, + "dnf_automatic_installation_state_value": False, + "dnf_automatic_set_installation_state": True + }, + "packagekit": { + "packagekit_apply_updates_value": "", + "packagekit_download_updates_value": "", + "packagekit_enable_on_reboot_value": False, + "packagekit_installation_state_value": False, + "packagekit_set_installation_state": True + } + } + }, + "assertions": { + "yum_cron": { + "yum_cron_config_exists": True, + "yum_cron_apply_updates_expected": 'apply_updates = yes', + "yum_cron_download_updates_expected": 'download_updates = yes', + }, + "dnf_automatic": { + "dnf_automatic_config_exists": False, + "dnf_automatic_apply_updates_expected": '', + "dnf_automatic_download_updates_expected": '', + }, + "packagekit": { + "packagekit_config_exists": False, + "packagekit_apply_updates_expected": '', + "packagekit_download_updates_expected": '', + } + } + } + + revert_success_backup_config_does_not_exist_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": True, + "expected_output": "[YPM] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "yum_cron": { + "set_yum_cron": True, + "yum_cron_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "dnf_automatic": { + "set_dnf_automatic": False, + "dnf_automatic_config_value": "" + }, + "packagekit": { + "set_packagekit": False, + "packagekit_config_value": "" + } + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": False, + "yum_cron": { + "yum_cron_apply_updates_value": "", + "yum_cron_download_updates_value": "", + "yum_cron_enable_on_reboot_value": False, + "yum_cron_installation_state_value": False, + "yum_cron_set_installation_state": True + }, + "dnf_automatic": { + "dnf_automatic_apply_updates_value": "", + "dnf_automatic_download_updates_value": "", + "dnf_automatic_enable_on_reboot_value": False, + "dnf_automatic_installation_state_value": False, + "dnf_automatic_set_installation_state": True + }, + "packagekit": { + "packagekit_apply_updates_value": "", + "packagekit_download_updates_value": "", + "packagekit_enable_on_reboot_value": False, + "packagekit_installation_state_value": False, + "packagekit_set_installation_state": True + } + } + }, + "assertions": { + "yum_cron": { + "yum_cron_config_exists": True, + "yum_cron_apply_updates_expected": 'apply_updates = no', + "yum_cron_download_updates_expected": 'download_updates = no', + }, + "dnf_automatic": { + "dnf_automatic_config_exists": False, + "dnf_automatic_apply_updates_expected": '', + "dnf_automatic_download_updates_expected": '', + }, + "packagekit": { + "packagekit_config_exists": False, + "packagekit_apply_updates_expected": '', + "packagekit_download_updates_expected": '', + } + } + } + + revert_success_backup_config_invalid_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": True, + "expected_output": "[YPM] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "yum_cron": { + "set_yum_cron": True, + "yum_cron_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "dnf_automatic": { + "set_dnf_automatic": False, + "dnf_automatic_config_value": "" + }, + "packagekit": { + "set_packagekit": False, + "packagekit_config_value": "" + } + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "yum_cron": { + "yum_cron_apply_updates_value": "yes", + "yum_cron_download_updates_value": "yes", + "yum_cron_enable_on_reboot_value": True, + "yum_cron_installation_state_value": False, + "yum_cron_set_installation_state": False + }, + "dnf_automatic": { + "dnf_automatic_apply_updates_value": "", + "dnf_automatic_download_updates_value": "", + "dnf_automatic_enable_on_reboot_value": False, + "dnf_automatic_installation_state_value": False, + "dnf_automatic_set_installation_state": True + }, + "packagekit": { + "packagekit_apply_updates_value": "", + "packagekit_download_updates_value": "", + "packagekit_enable_on_reboot_value": False, + "packagekit_installation_state_value": False, + "packagekit_set_installation_state": True + } + } + }, + "assertions": { + "yum_cron": { + "yum_cron_config_exists": True, + "yum_cron_apply_updates_expected": 'apply_updates = no', + "yum_cron_download_updates_expected": 'download_updates = no', + }, + "dnf_automatic": { + "dnf_automatic_config_exists": False, + "dnf_automatic_apply_updates_expected": '', + "dnf_automatic_download_updates_expected": '', + }, + "packagekit": { + "packagekit_config_exists": False, + "packagekit_apply_updates_expected": '', + "packagekit_download_updates_expected": '', + } + } + } + + revert_success_backup_config_contains_empty_values_testcase = { + "legacy_type": 'HappyPath', + "stdio": { + "capture_output": False, + "expected_output": "" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "yum_cron": { + "set_yum_cron": True, + "yum_cron_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "dnf_automatic": { + "set_dnf_automatic": True, + "dnf_automatic_config_value": "apply_updates = no\ndownload_updates = no\n" + }, + "packagekit": { + "set_packagekit": True, + "packagekit_config_value": "WritePreparedUpdates = false\n" + } + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "yum_cron": { + "yum_cron_apply_updates_value": "no", + "yum_cron_download_updates_value": "yes", + "yum_cron_enable_on_reboot_value": True, + "yum_cron_installation_state_value": True, + "yum_cron_set_installation_state": True + }, + "dnf_automatic": { + "dnf_automatic_apply_updates_value": "yes", + "dnf_automatic_download_updates_value": "", + "dnf_automatic_enable_on_reboot_value": True, + "dnf_automatic_installation_state_value": True, + "dnf_automatic_set_installation_state": True + }, + "packagekit": { + "packagekit_apply_updates_value": "", + "packagekit_download_updates_value": "", + "packagekit_enable_on_reboot_value": True, + "packagekit_installation_state_value": True, + "packagekit_set_installation_state": True + } + } + }, + "assertions": { + "yum_cron": { + "yum_cron_config_exists": True, + "yum_cron_apply_updates_expected": 'apply_updates = no', + "yum_cron_download_updates_expected": 'download_updates = yes', + }, + "dnf_automatic": { + "dnf_automatic_config_exists": True, + "dnf_automatic_apply_updates_expected": 'apply_updates = yes', + "dnf_automatic_download_updates_expected": 'download_updates = no', + }, + "packagekit": { + "packagekit_config_exists": True, + "packagekit_apply_updates_expected": 'WritePreparedUpdates = false', + "packagekit_download_updates_expected": '', + } + } + } + + all_testcases = [revert_success_testcase, revert_success_with_only_yum_cron_installed_testcase, revert_success_backup_config_does_not_exist_testcase, revert_success_backup_config_invalid_testcase, revert_success_backup_config_contains_empty_values_testcase] + + for testcase in all_testcases: + self.tearDown() + self.setUp() + captured_output, original_stdout = None, None + if testcase["stdio"]["capture_output"]: + # arrange capture std IO + captured_output, original_stdout = self.__capture_std_io() + + self.runtime.set_legacy_test_type(testcase["legacy_type"]) + package_manager = self.container.get('package_manager') + + # setup current auto OS update config, backup for system default config and invoke revert to system default + self.__setup_config_and_invoke_revert_auto_os_to_system_default(package_manager, + create_current_auto_os_config=bool(testcase["config"]["current_auto_update_config"]["create_current_auto_os_config"]), + set_yum_cron=bool(testcase["config"]["current_auto_update_config"]["yum_cron"]["set_yum_cron"]), + yum_cron_config_value=testcase["config"]["current_auto_update_config"]["yum_cron"]["yum_cron_config_value"], + set_dnf_automatic=bool(testcase["config"]["current_auto_update_config"]["dnf_automatic"]["set_dnf_automatic"]), + dnf_automatic_config_value=testcase["config"]["current_auto_update_config"]["dnf_automatic"]["dnf_automatic_config_value"], + set_packagekit=bool(testcase["config"]["current_auto_update_config"]["packagekit"]["set_packagekit"]), + packagekit_config_value=testcase["config"]["current_auto_update_config"]["packagekit"]["packagekit_config_value"], + create_backup_for_system_default_config=bool(testcase["config"]["backup_system_default_config"]["create_backup_for_system_default_config"]), + yum_cron_apply_updates_value=testcase["config"]["backup_system_default_config"]["yum_cron"]["yum_cron_apply_updates_value"], + yum_cron_download_updates_value=testcase["config"]["backup_system_default_config"]["yum_cron"]["yum_cron_download_updates_value"], + yum_cron_enable_on_reboot_value=bool(testcase["config"]["backup_system_default_config"]["yum_cron"]["yum_cron_enable_on_reboot_value"]), + yum_cron_installation_state_value=bool(testcase["config"]["backup_system_default_config"]["yum_cron"]["yum_cron_installation_state_value"]), + yum_cron_set_installation_state=bool(testcase["config"]["backup_system_default_config"]["yum_cron"]["yum_cron_set_installation_state"]), + dnf_automatic_apply_updates_value=testcase["config"]["backup_system_default_config"]["dnf_automatic"]["dnf_automatic_apply_updates_value"], + dnf_automatic_download_updates_value=testcase["config"]["backup_system_default_config"]["dnf_automatic"]["dnf_automatic_download_updates_value"], + dnf_automatic_enable_on_reboot_value=bool(testcase["config"]["backup_system_default_config"]["dnf_automatic"]["dnf_automatic_enable_on_reboot_value"]), + dnf_automatic_installation_state_value=bool(testcase["config"]["backup_system_default_config"]["dnf_automatic"]["dnf_automatic_installation_state_value"]), + dnf_automatic_set_installation_state=bool(testcase["config"]["backup_system_default_config"]["dnf_automatic"]["dnf_automatic_set_installation_state"]), + packagekit_apply_updates_value=testcase["config"]["backup_system_default_config"]["packagekit"]["packagekit_apply_updates_value"], + packagekit_download_updates_value=testcase["config"]["backup_system_default_config"]["packagekit"]["packagekit_download_updates_value"], + packagekit_enable_on_reboot_value=bool(testcase["config"]["backup_system_default_config"]["packagekit"]["packagekit_enable_on_reboot_value"]), + packagekit_installation_state_value=bool(testcase["config"]["backup_system_default_config"]["packagekit"]["packagekit_installation_state_value"]), + packagekit_set_installation_state=bool(testcase["config"]["backup_system_default_config"]["packagekit"]["packagekit_set_installation_state"])) + + # assert + if testcase["stdio"]["capture_output"]: + # restore sys.stdout output + sys.stdout = original_stdout + self.__assert_std_io(captured_output=captured_output, expected_output=testcase["stdio"]["expected_output"]) + self.__assert_all_reverted_automatic_patch_configuration_settings(package_manager, + yum_cron_config_exists=bool(testcase["assertions"]["yum_cron"]["yum_cron_config_exists"]), + yum_cron_apply_updates_expected=testcase["assertions"]["yum_cron"]["yum_cron_apply_updates_expected"], + yum_cron_download_updates_expected=testcase["assertions"]["yum_cron"]["yum_cron_download_updates_expected"], + dnf_automatic_config_exists=bool(testcase["assertions"]["dnf_automatic"]["dnf_automatic_config_exists"]), + dnf_automatic_apply_updates_expected=testcase["assertions"]["dnf_automatic"]["dnf_automatic_apply_updates_expected"], + dnf_automatic_download_updates_expected=testcase["assertions"]["dnf_automatic"]["dnf_automatic_download_updates_expected"], + packagekit_config_exists=bool(testcase["assertions"]["packagekit"]["packagekit_config_exists"]), + packagekit_apply_updates_expected=testcase["assertions"]["packagekit"]["packagekit_apply_updates_expected"], + packagekit_download_updates_expected=testcase["assertions"]["packagekit"]["packagekit_download_updates_expected"]) + def test_is_reboot_pending_return_true_when_exception_raised(self): package_manager = self.container.get('package_manager') backup_do_process_require_restart = package_manager.do_processes_require_restart @@ -757,5 +1253,6 @@ def test_get_dependent_list_yum_version_4_update_in_two_lines_with_unexpected_ou dependent_list = package_manager.get_dependent_list(["polkit.x86_64"]) self.assertEqual(len(dependent_list), 0) + if __name__ == '__main__': unittest.main() diff --git a/src/core/tests/Test_ZypperPackageManager.py b/src/core/tests/Test_ZypperPackageManager.py index ffadb910..eb0b9375 100644 --- a/src/core/tests/Test_ZypperPackageManager.py +++ b/src/core/tests/Test_ZypperPackageManager.py @@ -18,6 +18,13 @@ import unittest import tempfile import shutil +import sys +# Conditional import for StringIO +try: + from StringIO import StringIO # Python 2 +except ImportError: + from io import StringIO # Python 3 + from core.src.bootstrap.Constants import Constants from core.tests.library.ArgumentComposer import ArgumentComposer from core.tests.library.RuntimeCompositor import RuntimeCompositor @@ -75,6 +82,57 @@ def mock_do_processes_require_restart(self): raise Exception #endregion Mocks + # region Utility Functions + def __setup_config_and_invoke_revert_auto_os_to_system_default(self, package_manager, create_current_auto_os_config=True, create_backup_for_system_default_config=True, current_auto_os_update_config_value='', setup_enable_config=True, enable_cron_value="", installation_state_value=False): + """ Sets up current auto OS update config, backup for system default config (if requested) and invoke revert to system default """ + # setup current auto OS update config + if create_current_auto_os_config: + self.__setup_current_auto_os_update_config(package_manager, current_auto_os_update_config_value) + + # setup backup for system default auto OS update config + if create_backup_for_system_default_config: + self.__setup_backup_for_system_default_OS_update_config(package_manager, setup_enable_config=setup_enable_config, enable_cron_value=enable_cron_value, installation_state_value=installation_state_value) + + package_manager.revert_auto_os_update_to_system_default() + + def __setup_current_auto_os_update_config(self, package_manager, config_value='', config_file_name="automatic_online_update"): + # setup current auto OS update config + package_manager.YastOnlineUpdateConfigurationConstants.OS_PATCH_CONFIGURATION_SETTINGS_FILE_PATH = os.path.join(self.runtime.execution_config.config_folder, config_file_name) + self.runtime.write_to_file(package_manager.YastOnlineUpdateConfigurationConstants.OS_PATCH_CONFIGURATION_SETTINGS_FILE_PATH, config_value) + + def __setup_backup_for_system_default_OS_update_config(self, package_manager, setup_enable_config=True, enable_cron_value="", installation_state_value=False): + # setup backup for system default auto OS update config + package_manager.image_default_patch_configuration_backup_path = os.path.join(self.runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + backup_image_default_patch_configuration_json = { + "yast2-online-update-configuration": { + "installation_state": installation_state_value + } + } + if setup_enable_config: + backup_image_default_patch_configuration_json["yast2-online-update-configuration"]["AOU_ENABLE_CRONJOB"] = enable_cron_value + self.runtime.write_to_file(package_manager.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(backup_image_default_patch_configuration_json))) + + @staticmethod + def __capture_std_io(): + # arrange capture std IO + captured_output = StringIO() + original_stdout = sys.stdout + sys.stdout = captured_output + return captured_output, original_stdout + + def __assert_std_io(self, captured_output, expected_output): + output = captured_output.getvalue() + self.assertTrue(expected_output in output) + + def __assert_reverted_automatic_patch_configuration_settings(self, package_manager, config_exists=True, config_value_expected=''): + if config_exists: + reverted_os_patch_configuration_settings = self.runtime.env_layer.file_system.read_with_retry(package_manager.os_patch_configuration_settings_file_path) + self.assertTrue(reverted_os_patch_configuration_settings is not None) + self.assertTrue(config_value_expected in reverted_os_patch_configuration_settings) + else: + self.assertFalse(os.path.exists(package_manager.os_patch_configuration_settings_file_path)) + # endregion + def test_package_manager_no_updates(self): """Unit test for zypper package manager with no updates""" # Path change @@ -427,6 +485,150 @@ def test_update_image_default_patch_mode(self): self.assertTrue(yast2_online_update_configuration_os_patch_configuration_settings_file_path_read is not None) self.assertTrue('AOU_ENABLE_CRONJOB="false"' in yast2_online_update_configuration_os_patch_configuration_settings_file_path_read) + def test_revert_auto_os_update_to_system_default(self): + revert_success_testcase = { + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'AOU_ENABLE_CRONJOB="false"' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "setup_enable_config": True, + "enable_cron_value": "true", + "installation_state_value": True + } + }, + "assertions": { + "config_value_expected": 'AOU_ENABLE_CRONJOB="true"', + "config_exists": True + } + } + + revert_success_auto_os_update_config_does_not_exist = { + "stdio": { + "capture_output": True, + "expected_output": "[ZPM] Machine default auto OS update service is not installed on the VM and hence no config to revert. [Service=yast2-online-update-configuration]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": False, + "current_auto_os_update_config_value": '' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "setup_enable_config": True, + "enable_cron_value": "true", + "installation_state_value": True + } + }, + "assertions": { + "config_value_expected": '', + "config_exists": False + } + } + + revert_success_backup_config_does_not_exist = { + "stdio": { + "capture_output": True, + "expected_output": "[ZPM] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service=yast2-online-update-configuration]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'AOU_ENABLE_CRONJOB="false"' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": False, + "setup_enable_config": True, + "enable_cron_value": "", + "installation_state_value": False + } + }, + "assertions": { + "config_value_expected": 'AOU_ENABLE_CRONJOB="false"', + "config_exists": True + } + } + + revert_success_backup_config_invalid = { + "stdio": { + "capture_output": True, + "expected_output": "[ZPM] Since the backup is invalid or does not exist for current service, we won't be able to revert auto OS patch settings to their system default value. [Service=yast2-online-update-configuration]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'AOU_ENABLE_CRONJOB="false"' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "setup_enable_config": False, + "enable_cron_value": "", + "installation_state_value": True + } + }, + "assertions": { + "config_value_expected": 'AOU_ENABLE_CRONJOB="false"', + "config_exists": True + } + } + + revert_success_backup_config_contains_empty_values = { + "stdio": { + "capture_output": False, + "expected_output": "" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'AOU_ENABLE_CRONJOB="false"' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "setup_enable_config": True, + "enable_cron_value": "", + "installation_state_value": True + } + }, + "assertions": { + "config_value_expected": 'AOU_ENABLE_CRONJOB="false"', + "config_exists": True + } + } + + all_testcases = [revert_success_testcase, revert_success_auto_os_update_config_does_not_exist, revert_success_backup_config_does_not_exist, revert_success_backup_config_invalid, revert_success_backup_config_contains_empty_values] + + for testcase in all_testcases: + self.tearDown() + self.setUp() + captured_output, original_stdout = None, None + if testcase["stdio"]["capture_output"]: + # arrange capture std IO + captured_output, original_stdout = self.__capture_std_io() + + package_manager = self.container.get('package_manager') + + # setup current auto OS update config, backup for system default config and invoke revert to system default + self.__setup_config_and_invoke_revert_auto_os_to_system_default(package_manager, + create_current_auto_os_config=bool(testcase["config"]["current_auto_update_config"]["create_current_auto_os_config"]), + current_auto_os_update_config_value=testcase["config"]["current_auto_update_config"]["current_auto_os_update_config_value"], + create_backup_for_system_default_config=bool(testcase["config"]["backup_system_default_config"]["create_backup_for_system_default_config"]), + setup_enable_config=bool(testcase["config"]["backup_system_default_config"]["setup_enable_config"]), + enable_cron_value=testcase["config"]["backup_system_default_config"]["enable_cron_value"], + installation_state_value=bool(testcase["config"]["backup_system_default_config"]["installation_state_value"])) + + # assert + if testcase["stdio"]["capture_output"]: + # restore sys.stdout output + sys.stdout = original_stdout + self.__assert_std_io(captured_output=captured_output, expected_output=testcase["stdio"]["expected_output"]) + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_exists=bool(testcase["assertions"]["config_exists"]), config_value_expected=str(testcase["assertions"]["config_value_expected"])) + def is_string_in_status_file(self, str_to_find): with open(self.runtime.execution_config.status_file_path, 'r') as file_handle: file_contents = json.loads(file_handle.read()) diff --git a/src/core/tests/library/LegacyEnvLayerExtensions.py b/src/core/tests/library/LegacyEnvLayerExtensions.py index e2d75db3..ede2bf81 100644 --- a/src/core/tests/library/LegacyEnvLayerExtensions.py +++ b/src/core/tests/library/LegacyEnvLayerExtensions.py @@ -667,8 +667,12 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): code = 103 output = '' elif self.legacy_package_manager_name is Constants.TDNF: - code = 0 - output = '' + if cmd.find("systemctl list-unit-files --type=service | grep dnf-automatic.service") > -1: + code = 1 + output = 'Auto update service is not installed' + else: + code = 0 + output = '' elif cmd.find("systemctl") > -1: code = 1 output = '' @@ -1448,6 +1452,18 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): "\n" + \ "Total download size: 231 k\n" + \ "Operation aborted.\n" + elif self.legacy_test_type == 'RevertToImageDefault': + if self.legacy_package_manager_name is Constants.YUM: + if cmd.find("systemctl list-unit-files --type=service | grep yum-cron.service") > -1: + code = 0 + output = 'Auto update service installed' + elif cmd.find("systemctl list-unit-files --type=service ") > -1: + code = 1 + output = 'Auto update service not installed' + elif self.legacy_package_manager_name is Constants.TDNF: + if cmd.find("systemctl list-unit-files --type=service | grep dnf-automatic.service") > -1: + code = 0 + output = 'Auto update service installed' major_version = self.get_python_major_version() if major_version == 2: return code, output.decode('utf8', 'ignore').encode('ascii', 'ignore')