diff --git a/src/core/src/core_logic/PatchInstaller.py b/src/core/src/core_logic/PatchInstaller.py index 81bb0976..e697bfae 100644 --- a/src/core/src/core_logic/PatchInstaller.py +++ b/src/core/src/core_logic/PatchInstaller.py @@ -78,7 +78,7 @@ def start_installation(self, simulate=False): reboot_manager.start_reboot_if_required_and_time_available(maintenance_window.get_remaining_time_in_minutes(None, False)) # Update certificates if feature flag to update certs is set - if self.execution_config.enable_uefi_cert_update: + if self.execution_config.enable_uefi_cert_update and self.package_manager.is_certificate_update_enabled_for_package_manager: self.try_update_certificates_for_default_patching() if self.execution_config.max_patch_publish_date != str(): @@ -808,24 +808,138 @@ def try_update_certificates_for_default_patching(self): self.composite_logger.log_debug("Not updating certificates since this is not a default patching operation.") return + if self.__are_prerequisites_for_updating_certs_met(): + try: + self.composite_logger.log_verbose("Updating current certificates if needed...") + certs_updated = self.package_manager.try_update_certs() + if not certs_updated: + error_msg = "UEFI certificate update did not complete successfully. Certificates may not have been updated." + self.composite_logger.log_error(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + except Exception as e: + error_msg = "An error was encountered while attempting to update certificates. Continuing with patch installation... [Error: {0}]".format(str(e)) + self.composite_logger.log_error(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + + def __is_default_patching(self): + # type: () -> bool + """ Returns true if the patching run is a default patching run""" + return (self.execution_config.health_store_id is not None and + self.execution_config.health_store_id != "" and + self.execution_config.operation.lower() == Constants.INSTALLATION.lower()) + + def __are_prerequisites_for_updating_certs_met(self): + # type: () -> bool + """ Returns true if all the pre-requisites for updating certs are met. These pre-reqs are: + 1. Should not be a CVM + 2. Hibernation should be turned off + 3. Should not already contain latest certificates + 4. System uptime should not be more than 7 days, if it is, we issue a reboot if permitted""" + + self.composite_logger.log_verbose("Verifying all pre-requisites for updating certificates are met...") + if not self.__is_definitely_not_cvm(): + return False + + if not self.can_continue_cert_update_after_hibernation_check(): + return False + + if not self.can_continue_cert_update_after_latest_cert_check(): + return False + + if not self.ensure_pre_cert_update_reboot_completed(): + return False + + return True + + def __is_definitely_not_cvm(self): + # type: () -> bool + """ Returns true if this is not a Confidential VM (CVM), false otherwise (i.e. CVM or undeterministic) """ try: + self.composite_logger.log_verbose("Verifying if this is a Confidential VM...") is_confidential_vm, detection_details = self.env_layer.detect_confidential_vm() except Exception as e: self.composite_logger.log_warning("Unable to determine whether the VM is a Confidential VM before attempting the UEFI certificate update. Continuing with patch installation... [Error: {0}]".format(str(e))) - return + return False if is_confidential_vm: self.composite_logger.log("Skipping UEFI certificate update because this VM was detected as a Confidential VM. [Detection={0}]".format(detection_details)) - return + return not is_confidential_vm + + def ensure_pre_cert_update_reboot_completed(self): + # type: () -> bool + """Attempts pre-cert reboot when required. + + Returns True only when cert update can continue in the current cycle. + Returns False when reboot is required-but-blocked/failed, or when reboot was initiated + and cert update must wait for the post-reboot cycle. + """ + self.composite_logger.log_verbose("Ensuring pre certificate update reboot is completed...") + is_reboot_required_before_cert_update = self.package_manager.is_reboot_required_before_cert_update() + if is_reboot_required_before_cert_update: + if self.reboot_manager.is_setting(Constants.REBOOT_NEVER): + error_msg = "UEFI certificate update requires a reboot first (VM uptime exceeded 7 days), but reboot setting is 'Never'. Skipping certificate update." + self.composite_logger.log_error(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + return False + + reboot_started = False + original_force_reboot = self.package_manager.force_reboot + self.package_manager.force_reboot = True + try: + remaining_time = self.maintenance_window.get_remaining_time_in_minutes(None, False) + reboot_started = self.reboot_manager.start_reboot_if_required_and_time_available(remaining_time) + except Exception as e: + error_msg = "Unable to reboot before UEFI certificate update. Skipping certificate update. [Error: {0}]".format(str(e)) + self.composite_logger.log_warning(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + return False + finally: + self.package_manager.force_reboot = original_force_reboot + + if not reboot_started: + error_msg = "UEFI certificate update requires a reboot first (VM uptime exceeded 7 days), but reboot could not be initiated. Skipping certificate update." + self.composite_logger.log_warning(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + return False + + # If reboot was started, this process is expected to exit; avoid running cert update in the same cycle. + return False + + return True + + def can_continue_cert_update_after_hibernation_check(self): + # type: () -> bool + """Attempts hibernation pre-cert check. + + Returns True when cert update can continue in the current cycle. + Returns False when hibernation is enabled and cert update must be skipped or hibernate state cannot be determined. + """ try: - self.package_manager.update_certs() + is_hibernation_enabled = self.package_manager.is_hibernation_enabled_for_cert_update() except Exception as e: - self.composite_logger.log_warning("An error was encountered while attempting to update certificates. Continuing with patch installation... [Error: {0}]".format(str(e))) + error_msg = "Unable to determine hibernation state before UEFI certificate update. Not continuing with certificate update. [Error: {0}]".format(str(e)) + self.composite_logger.log_warning(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + return False - def __is_default_patching(self): + if is_hibernation_enabled: + error_msg = "UEFI certificate update requires hibernation to be turned off for the duration of the update. Turn off hibernation and re-run patching. Skipping certificate update." + self.composite_logger.log_warning(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + return False + + return True + + def can_continue_cert_update_after_latest_cert_check(self): # type: () -> bool - """ Returns true if the patching run is a default patching run""" - return (self.execution_config.health_store_id is not None and - self.execution_config.health_store_id != "" and - self.execution_config.operation.lower() == Constants.INSTALLATION.lower()) + """Returns False when latest certs are already installed and cert update can be skipped.""" + self.composite_logger.log_verbose("Checking if latest UEFI certificates are already present...") + latest_certs_present = self.package_manager.are_latest_certs_present_with_mokutil_check() + if latest_certs_present: + self.composite_logger.log("Latest UEFI certificates are already present. Skipping certificate update.") + return False + + return True + + diff --git a/src/core/src/package_managers/AptitudePackageManager.py b/src/core/src/package_managers/AptitudePackageManager.py index 860ef9a7..2730324c 100644 --- a/src/core/src/package_managers/AptitudePackageManager.py +++ b/src/core/src/package_managers/AptitudePackageManager.py @@ -93,7 +93,8 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ self.package_install_expected_avg_time_in_seconds = 90 # As per telemetry data, the average time to install package is around 81 seconds for apt. - # Update certificates in factory defaults using default Ubuntu repos. + # Update certificates in factory defaults. + self.is_certificate_update_enabled_for_package_manager = True self.install_mokutil_cmd = "sudo apt-get install -y -qq mokutil" self.apt_update_cmd = "sudo apt-get -q update" self.min_fwupd_version = "2.0.8" # Refer public docs: https://github.com/fwupd/fwupd/releases/tag/2.0.8 and https://discourse.ubuntu.com/t/microsoft-uefi-ca-rotation-what-it-means-for-ubuntu-users-and-vendors/82652 @@ -102,6 +103,9 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ self.install_fwupd_cmd = "sudo apt-get install -y fwupd" self.fwupd_refresh_cmd = "sudo fwupdmgr refresh" # NOTE: This could be made generic in package manager, depending on what solution type is adopted for other distros self.fwupd_update_cmd = "sudo fwupdmgr update -y" + self.get_uptime_seconds_cmd = "cat /proc/uptime" + self.get_hibernation_state_cmd = "cat /sys/power/disk" + self.pre_cert_reboot_uptime_threshold_seconds = 7 * 24 * 60 * 60 # NOTE: This needs to be reviewed again later as these aren't specific to this class # region Sources Management def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_classification=str()): @@ -957,6 +961,59 @@ def get_package_install_expected_avg_time_in_seconds(self): return self.package_install_expected_avg_time_in_seconds # region Update certificates in factory defaults + def is_reboot_required_before_cert_update(self): + # type: () -> bool + """ Long-running VMs may not have the minimum firmware required for certificate updates. + We will reboot VMs with uptime > threshold to load the required firmware version. """ + """ Return True when certificate update should be preceded by a reboot.""" + uptime_seconds = self.__get_vm_uptime_seconds() + if uptime_seconds is None: + # Best-effort check: do not block cert flow if uptime cannot be determined. + return False + + requires_reboot = uptime_seconds > self.pre_cert_reboot_uptime_threshold_seconds + self.composite_logger.log_debug("[APM][Certs] Pre-cert reboot check. [UptimeInSeconds={0}][ThresholdInSeconds={1}][RequiresReboot={2}]" + .format(str(uptime_seconds), str(self.pre_cert_reboot_uptime_threshold_seconds), str(requires_reboot))) + return requires_reboot + + def is_hibernation_enabled_for_cert_update(self): + # type: () -> bool + """Returns True when hibernation appears enabled based on /sys/power/disk state.""" + cmd = self.get_hibernation_state_cmd + self.composite_logger.log_verbose('[APM][Certs] Checking hibernation state before cert update. [Command={0}]'.format(cmd)) + code, out = self.env_layer.run_command_output(cmd, False, False) + out = str(out) if out is not None else str() + + if code != self.apt_exitcode_ok or out.strip() == str(): + self.composite_logger.log_debug('[APM][Certs] Unable to determine hibernation state for cert update. Assuming enabled. [Code={0}][Output={1}]'.format(str(code), out)) + return True + + # Linux shows current mode in brackets; [disabled] means hibernation is turned off. + is_hibernation_enabled = '[disabled]' not in out + self.composite_logger.log_debug('[APM][Certs] Hibernation state for cert update. [Enabled={0}][RawState={1}]'.format(str(is_hibernation_enabled), out.strip())) + return is_hibernation_enabled + + def __get_vm_uptime_seconds(self): + # type: () -> any + """Return VM uptime in seconds, or None when unavailable.""" + cmd = self.get_uptime_seconds_cmd + code, out = self.env_layer.run_command_output(cmd, False, False) + self.composite_logger.log_debug("[APM][Certs] VM uptime command executed. [Command={0}][Code={1}][Output={2}]".format(str(cmd), str(code), str(out))) + if code != self.apt_exitcode_ok: + self.composite_logger.log_warning("[APM][Certs] Unable to determine VM uptime. [Command={0}][Code={1}][Output={2}]".format(str(cmd), str(code), str(out))) + return None + + output = str(out).strip() + if output == str() or len(output.split()) == 0: + self.composite_logger.log_warning("[APM][Certs] VM uptime command returned an empty result.") + return None + + try: + return int(float(output.split()[0])) + except Exception as error: + self.composite_logger.log_warning("[APM][Certs] Failed to parse VM uptime output. [Output={0}][Error={1}]".format(str(output), repr(error))) + return None + def try_install_mokutil(self): # type: () -> bool """ Attempts to install mokutil """ @@ -969,9 +1026,6 @@ def try_install_mokutil(self): def try_update_certs(self): """ Attempts to update certificate status """ self.composite_logger.log("[APM][Certs] Starting cert update flow.") - if self.are_latest_certs_present(): - self.composite_logger.log("[APM][Certs] Latest certs already present. Skipping.") - return True success = False try: @@ -985,7 +1039,7 @@ def try_update_certs(self): """ NOTE: They tooling used to update here is fwupd (firmware update manager). In this method of updating certs, the exact version of current certs is never pinned or set/referred while installing. fwupd fetches and installs latest available certs. This is beneficial because our code doesn't become dated in the future (if it was pinned to an old version) but adds the need to verify if the certs we intend to update were applied. Hence the validation on latest certs. """ - if self.are_latest_certs_present(): + if self.are_latest_certs_present_with_mokutil_check(): self.composite_logger.log("[APM][Certs] Cert update completed and verified.") success = True else: diff --git a/src/core/src/package_managers/Dnf5PackageManager.py b/src/core/src/package_managers/Dnf5PackageManager.py index d2942a1c..c077714e 100644 --- a/src/core/src/package_managers/Dnf5PackageManager.py +++ b/src/core/src/package_managers/Dnf5PackageManager.py @@ -718,4 +718,12 @@ def try_install_mokutil(self): def try_update_certs(self): """ Attempts to update certificate status """ pass + + def is_hibernation_enabled_for_cert_update(self): + """Checks whether hibernation is enabled""" + return False + + def is_reboot_required_before_cert_update(self): + """ Checks if a reboot is required before updating certificates """ + return False # endregion diff --git a/src/core/src/package_managers/PackageManager.py b/src/core/src/package_managers/PackageManager.py index 5154aafa..30c08ee3 100644 --- a/src/core/src/package_managers/PackageManager.py +++ b/src/core/src/package_managers/PackageManager.py @@ -15,7 +15,6 @@ # Requires Python 2.7+ """The is base package manager, which defines the package management relevant operations""" -import json import os from abc import ABCMeta, abstractmethod from core.src.bootstrap.Constants import Constants @@ -55,6 +54,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ self.REBOOT_PENDING_FILE_PATH = '/var/run/reboot-required' # Update certificates in factory defaults + self.is_certificate_update_enabled_for_package_manager = False # NOTE: This should be removed once certificate update is enabled for all package managers self.check_mokutil_exists_cmd = "command -v mokutil" self.get_kek_cert_status_cmd = "mokutil --kek | grep 'CN='" self.get_db_cert_status_cmd = "mokutil --db | grep 'CN='" @@ -498,31 +498,21 @@ def get_package_install_expected_avg_time_in_seconds(self): pass # region Update certificates in factory defaults - # TODO (Before removing in-VM feature flag): This update should not be done in CVMs in current form - def update_certs(self): - # 1. Fetch and Log current certs on the VM NOTE: Common across distros - # ensure mokutil exists - # if not, install mokutil - # If success, fetch cert detail - # If not, throw exception and stop - # log output - # 2. If 2023 certs already exists, do nothing - # 3. If not, perform all the steps to update certs - # 4. Validate and log new certs were installed - - self.composite_logger.log_verbose("[PM] Updating current certificates if needed...") - is_mokutil_installed = self.is_mokutil_installed() - try_install_mokutil_status = False - if not is_mokutil_installed: - try_install_mokutil_status = self.try_install_mokutil() - - if is_mokutil_installed or try_install_mokutil_status: - self.try_update_certs() - else: - error_msg = "[PM][UpdateCerts] Mokutil is not installed and could not be installed. Cannot fetch current certs." - self.composite_logger.log_error(error_msg) - self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) - raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + def ensure_mokutil_available_for_cert_checks(self): + # type: () -> bool + """Ensures mokutil is available before certificate status checks.""" + if self.is_mokutil_installed(): + return True + + self.composite_logger.log_verbose("[PM][Certs] Mokutil is not installed. Attempting installation before cert check.") + try_install_mokutil_status = self.try_install_mokutil() + if try_install_mokutil_status: + return True + + error_msg = "[PM][UpdateCerts] Mokutil is not installed and could not be installed. Cannot fetch current certs." + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + self.composite_logger.log_error(error_msg) + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) def is_mokutil_installed(self): # type: () -> bool @@ -538,6 +528,17 @@ def are_latest_certs_present(self): db_code, db_out = self.fetch_current_certs(Constants.Certificates.DB, self.get_db_cert_status_cmd, raise_on_exception=False) return self.is_latest_cert_installed(kek_code, kek_out) and self.is_latest_cert_installed(db_code, db_out) + def are_latest_certs_present_with_mokutil_check(self): + # type: () -> bool + """ Checks if the latest certs (i.e. 2023 certs) are already installed on the machine based on mokutil output, ensuring mokutil is available for the check.""" + are_latest_certs_present = False + try: + if self.ensure_mokutil_available_for_cert_checks(): + are_latest_certs_present = self.are_latest_certs_present() + except Exception as e: + self.composite_logger.log_warning("Unable to determine whether latest certs are already installed before UEFI certificate update.[Error: {0}]".format( str(e))) + return are_latest_certs_present + def fetch_current_certs(self, cert_type, get_cert_status_cmd, raise_on_exception=False): # type: (str, str, bool) -> (int, str) """ Fetches the status of the certificates on the machine """ @@ -577,5 +578,16 @@ def try_install_mokutil(self): def try_update_certs(self): """ Attempts to update certificate status """ pass + + @abstractmethod + def is_hibernation_enabled_for_cert_update(self): + # type: () -> bool + """Checks whether hibernation is enabled.""" + pass + + @abstractmethod + def is_reboot_required_before_cert_update(self): + """ Checks if a reboot is required before updating certificates """ + pass # endregion diff --git a/src/core/src/package_managers/TdnfPackageManager.py b/src/core/src/package_managers/TdnfPackageManager.py index a23d79e5..59726cc1 100644 --- a/src/core/src/package_managers/TdnfPackageManager.py +++ b/src/core/src/package_managers/TdnfPackageManager.py @@ -773,5 +773,13 @@ def try_install_mokutil(self): def try_update_certs(self): """ Attempts to update certificate status """ pass + + def is_hibernation_enabled_for_cert_update(self): + """Checks whether hibernation is enabled.""" + return False + + def is_reboot_required_before_cert_update(self): + """ Checks if a reboot is required before updating certificates """ + return False # endregion diff --git a/src/core/src/package_managers/YumPackageManager.py b/src/core/src/package_managers/YumPackageManager.py index d0512bf2..27cbf846 100644 --- a/src/core/src/package_managers/YumPackageManager.py +++ b/src/core/src/package_managers/YumPackageManager.py @@ -1130,5 +1130,13 @@ def try_install_mokutil(self): def try_update_certs(self): """ Attempts to update certificate status """ pass + + def is_hibernation_enabled_for_cert_update(self): + """Checks whether hibernation is enabled.""" + return False + + def is_reboot_required_before_cert_update(self): + """ Checks if a reboot is required before updating certificates """ + return False # endregion diff --git a/src/core/src/package_managers/ZypperPackageManager.py b/src/core/src/package_managers/ZypperPackageManager.py index bc5b0211..c341bf36 100644 --- a/src/core/src/package_managers/ZypperPackageManager.py +++ b/src/core/src/package_managers/ZypperPackageManager.py @@ -882,5 +882,13 @@ def try_install_mokutil(self): def try_update_certs(self): """ Attempts to update certificate status """ pass + + def is_hibernation_enabled_for_cert_update(self): + """Checks whether hibernation is enabled.""" + return False + + def is_reboot_required_before_cert_update(self): + """ Checks if a reboot is required before updating certificates """ + return False # endregion diff --git a/src/core/tests/Test_AptitudePackageManager.py b/src/core/tests/Test_AptitudePackageManager.py index 3f76efec..a33ee5a0 100644 --- a/src/core/tests/Test_AptitudePackageManager.py +++ b/src/core/tests/Test_AptitudePackageManager.py @@ -37,9 +37,6 @@ def setUp(self): self.argument_composer = ArgumentComposer().get_composed_arguments() self.runtime = RuntimeCompositor(self.argument_composer, True, Constants.APT) self.container = self.runtime.container - self.latest_certs_check_attempts = 0 - self.latest_apt_update_cmd_attempt = 0 - self.get_installed_fwupd_version_check_attempts = 0 def tearDown(self): self.runtime.stop() @@ -98,6 +95,10 @@ def mock_are_latest_certs_present_with_different_output_across_multiple_attempts return True + def mock_are_latest_certs_present_return_true_once(self): + self.latest_certs_check_attempts += 1 + return True + def mock_get_installed_fwupd_version_with_different_output_across_multiple_attempts(self): self.get_installed_fwupd_version_check_attempts += 1 @@ -108,7 +109,6 @@ def mock_get_installed_fwupd_version_with_different_output_across_multiple_attem return "2.0.15" def mock_run_command_output_fwupd_version_not_found_in_first_attempt_and_found_later(self, cmd, no_output=False, chk_err=True): - if cmd.find(self.runtime.package_manager.get_installed_fwupd_version_cmd) > -1: self.get_installed_fwupd_version_check_attempts += 1 @@ -135,19 +135,106 @@ def mock_run_command_output_fwupd_version_not_found_in_first_attempt_and_found_l return 0, "" def mock_run_command_output_fwupd_refresh_fails(self, cmd, no_output=False, chk_err=True): - if cmd.find(self.runtime.package_manager.fwupd_refresh_cmd) > -1: - return 1, "Error" - return 0, "" + self.assertGreater(cmd.find(self.runtime.package_manager.fwupd_refresh_cmd), -1) + return 1, "Error" def mock_run_command_output_apt_update_cmd_fails(self, cmd, no_output=False, chk_err=True): if cmd.find(self.runtime.package_manager.apt_update_cmd) > -1: - self.latest_apt_update_cmd_attempt +=1 + self.latest_apt_update_cmd_attempt += 1 if self.latest_apt_update_cmd_attempt == 2: return 1, "Error" return 0, "" def mock_is_mokutil_installed_return_false(self): return False + + def mock_are_latest_certs_present_return_true(self): + return True + + def mock_run_command_output_uptime_above_threshold(self, cmd, no_output=False, chk_err=True): + self.assertGreater(cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd), -1) + return 0, "700000.12 123.45" + + def mock_run_command_output_uptime_below_threshold(self, cmd, no_output=False, chk_err=True): + self.assertGreater(cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd), -1) + return 0, "120.12 123.45" + + def mock_run_command_output_uptime_cmd_fails(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate uptime command failure""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd), -1) + return 1, "Error: cannot read /proc/uptime" + + def mock_run_command_output_uptime_empty_output(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate uptime command returning empty output""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd), -1) + return 0, "" + + def mock_run_command_output_uptime_invalid_output(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate uptime command returning unparseable output""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd), -1) + return 0, "invalid unparseable output" + + def mock_run_command_output_hibernation_enabled(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate hibernation enabled state""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd), -1) + return 0, "platform [suspend] disk" + + def mock_run_command_output_hibernation_disabled(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate hibernation disabled state""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd), -1) + return 0, "platform suspend [disabled]" + + def mock_run_command_output_hibernation_cmd_fails(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate hibernation check command failure""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd), -1) + return 1, "Error: cannot read /sys/power/disk" + + def mock_run_command_output_hibernation_empty_output(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate empty output for hibernation check""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd), -1) + return 0, "" + + def mock_run_command_output_hibernation_none_output(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate None output for hibernation check""" + self.assertGreater(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd), -1) + return 0, None + + def mock_ensure_fwupd_installation_returns_true(self): + """Mock __ensure_fwupd_installation to return True""" + return True + + def mock_run_cert_shell_command_to_track_step_calls(self, command, step_name, raise_on_error=False): + """Mock __run_cert_shell_command to track step names""" + self.shell_calls_tracking.append(step_name) + + def mock_run_cert_apt_command_to_track_step_calls(self, command, step_name, raise_on_error=False): + """Mock __run_cert_apt_command to track step names""" + self.apt_calls_tracking.append(step_name) + + def mock_try_install_mokutil_returns_false(self): + """Mock try_install_mokutil to return False""" + return False + + def mock_try_install_mokutil_returns_true(self): + """Mock try_install_mokutil to return True""" + return True + + def mock_fetch_current_certs_should_not_be_called(self, cert_type, get_cert_status_cmd, raise_on_exception=False): + """Mock fetch_current_certs to fail if called in paths where it should not run""" + self.fail("fetch_current_certs should not be called for this use case") + + def mock_fetch_current_certs_returns_latest_cert_output(self, cert_type, get_cert_status_cmd, raise_on_exception=False): + """Mock fetch_current_certs to return latest cert output""" + cert_output = "{0}\n{1}".format("issuer: test", Constants.LATEST_CERTIFICATE_TIMESTAMP) + return (0, cert_output) + + def mock_add_error_to_status_to_capture_errors(self, error_msg, error_code=None): + """Mock add_error_to_status to capture errors""" + self.captured_errors_tracking.append((error_msg, error_code)) + + def mock_ensure_mokutil_available_returns_false(self): + """Mock ensure_mokutil_available_for_cert_checks to return False""" + return False # endregion Mocks # region Utility Functions @@ -1154,246 +1241,370 @@ def test_try_install_mokutil_failure(self): package_manager = self.container.get('package_manager') self.assertFalse(package_manager.try_install_mokutil()) - def test_are_latest_certs_present_returns_true_when_both_kek_and_db_are_latest(self): - self.runtime.set_legacy_test_type('SuccessInstallPath') - package_manager = self.container.get('package_manager') - self.assertTrue(package_manager.are_latest_certs_present()) - - def test_are_latest_certs_present_returns_false_when_kek_is_not_latest(self): + def test_are_latest_certs_present__with_various_use_cases(self): package_manager = self.container.get('package_manager') - backup_fetch_current_certs = package_manager.fetch_current_certs - - package_manager.fetch_current_certs = self.mock_fetch_current_certs_only_db_latest - self.assertFalse(package_manager.are_latest_certs_present()) - package_manager.fetch_current_certs = backup_fetch_current_certs - - def test_are_latest_certs_present_returns_false_when_db_fetch_fails(self): - package_manager = self.container.get('package_manager') + backup_is_mokutil_installed = package_manager.is_mokutil_installed + backup_try_install_mokutil = package_manager.try_install_mokutil backup_fetch_current_certs = package_manager.fetch_current_certs + backup_add_error_to_status = package_manager.status_handler.add_error_to_status + backup_ensure_mokutil_available = package_manager.ensure_mokutil_available_for_cert_checks - package_manager.fetch_current_certs = self.mock_fetch_current_certs_kek_latest_db_fails - - self.assertFalse(package_manager.are_latest_certs_present()) - package_manager.fetch_current_certs = backup_fetch_current_certs - - def test_try_update_certs_returns_true_when_latest_certs_already_present(self): + use_cases = [ + { + "name": "both_kek_and_db_are_latest", + "set_legacy_test_type": "SuccessInstallPath", + "expected": True + }, + { + "name": "kek_is_not_latest", + "fetch_current_certs": self.mock_fetch_current_certs_only_db_latest, + "expected": False + }, + { + "name": "db_fetch_fails", + "fetch_current_certs": self.mock_fetch_current_certs_kek_latest_db_fails, + "expected": False + }, + { + "name": "mokutil_missing_and_install_fails", + "mock_mokutil_install_failure": True, + "expected": False, + "validate_error_capture": True + }, + { + "name": "ensure_mokutil_check_fails", + "force_ensure_mokutil_false": True, + "expected": False + }, + { + "name": "mokutil_install_succeeds", + "mock_mokutil_install_success": True, + "expected": True + } + ] + + try: + for use_case in use_cases: + package_manager.is_mokutil_installed = backup_is_mokutil_installed + package_manager.try_install_mokutil = backup_try_install_mokutil + package_manager.fetch_current_certs = backup_fetch_current_certs + package_manager.status_handler.add_error_to_status = backup_add_error_to_status + package_manager.ensure_mokutil_available_for_cert_checks = backup_ensure_mokutil_available + self.captured_errors_tracking = [] + + if "set_legacy_test_type" in use_case: + self.runtime.set_legacy_test_type(use_case["set_legacy_test_type"]) + + if "fetch_current_certs" in use_case: + package_manager.fetch_current_certs = use_case["fetch_current_certs"] + + if use_case.get("mock_mokutil_install_failure", False): + package_manager.is_mokutil_installed = self.mock_is_mokutil_installed_return_false + package_manager.try_install_mokutil = self.mock_try_install_mokutil_returns_false + package_manager.fetch_current_certs = self.mock_fetch_current_certs_should_not_be_called + package_manager.status_handler.add_error_to_status = self.mock_add_error_to_status_to_capture_errors + + if use_case.get("force_ensure_mokutil_false", False): + package_manager.ensure_mokutil_available_for_cert_checks = self.mock_ensure_mokutil_available_returns_false + + if use_case.get("mock_mokutil_install_success", False): + package_manager.is_mokutil_installed = self.mock_is_mokutil_installed_return_false + package_manager.try_install_mokutil = self.mock_try_install_mokutil_returns_true + package_manager.fetch_current_certs = self.mock_fetch_current_certs_returns_latest_cert_output + + self.assertEqual( + package_manager.are_latest_certs_present_with_mokutil_check(), + use_case["expected"], + "Failed use case: {0}".format(use_case["name"]) + ) + + if use_case.get("validate_error_capture", False): + self.assertEqual(len(self.captured_errors_tracking), 1) + self.assertIn("Mokutil is not installed and could not be installed", self.captured_errors_tracking[0][0]) + self.assertEqual(self.captured_errors_tracking[0][1], Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + finally: + package_manager.is_mokutil_installed = backup_is_mokutil_installed + package_manager.try_install_mokutil = backup_try_install_mokutil + package_manager.fetch_current_certs = backup_fetch_current_certs + package_manager.status_handler.add_error_to_status = backup_add_error_to_status + package_manager.ensure_mokutil_available_for_cert_checks = backup_ensure_mokutil_available + + def test_try_update_certs_attempts_commands_when_called_directly(self): self.runtime.set_legacy_test_type('SuccessInstallPath') package_manager = self.container.get('package_manager') - shell_calls = [] - apt_calls = [] - package_manager._AptitudePackageManager__run_cert_shell_command = lambda command, step_name, raise_on_error=False: shell_calls.append(step_name) - package_manager._AptitudePackageManager__run_cert_apt_command = lambda command, step_name, raise_on_error=False: apt_calls.append(step_name) - - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(shell_calls, []) - self.assertEqual(apt_calls, []) - - def test_try_update_certs_success(self): - package_manager = self.container.get('package_manager') - - backup_is_reboot_pending = package_manager.is_reboot_pending backup_are_latest_certs_present = package_manager.are_latest_certs_present - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts - - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) - - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present - - def test_try_update_certs_when_fwupd_meets_minimum_version(self): + backup_ensure_fwupd_installation = package_manager._AptitudePackageManager__ensure_fwupd_installation + backup_run_cert_shell_command = package_manager._AptitudePackageManager__run_cert_shell_command + backup_run_cert_apt_command = package_manager._AptitudePackageManager__run_cert_apt_command + + try: + self.shell_calls_tracking = [] + self.apt_calls_tracking = [] + package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_return_true + package_manager._AptitudePackageManager__ensure_fwupd_installation = self.mock_ensure_fwupd_installation_returns_true + package_manager._AptitudePackageManager__run_cert_shell_command = self.mock_run_cert_shell_command_to_track_step_calls + package_manager._AptitudePackageManager__run_cert_apt_command = self.mock_run_cert_apt_command_to_track_step_calls + + self.assertTrue(package_manager.try_update_certs()) + self.assertEqual(self.shell_calls_tracking, ["FwupdRefresh", "FwupdUpdate"]) + self.assertEqual(self.apt_calls_tracking, ["AptUpdate"]) + finally: + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager._AptitudePackageManager__ensure_fwupd_installation = backup_ensure_fwupd_installation + package_manager._AptitudePackageManager__run_cert_shell_command = backup_run_cert_shell_command + package_manager._AptitudePackageManager__run_cert_apt_command = backup_run_cert_apt_command + + def test_should_reboot_before_cert_update__with_various_use_cases(self): package_manager = self.container.get('package_manager') - backup_is_reboot_pending = package_manager.is_reboot_pending backup_are_latest_certs_present = package_manager.are_latest_certs_present - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts - - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) - - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present + backup_run_command_output = package_manager.env_layer.run_command_output - def test_try_update_certs_removes_old_fwupd_before_install(self): + use_cases = [ + { + "name": "uptime_above_threshold_even_when_certs_are_latest", + "are_latest_certs_present": self.mock_are_latest_certs_present_return_true, + "run_command_output": self.mock_run_command_output_uptime_above_threshold, + "expected": True + }, + { + "name": "uptime_above_threshold_and_certs_stale", + "are_latest_certs_present": self.mock_is_mokutil_installed_return_false, + "run_command_output": self.mock_run_command_output_uptime_above_threshold, + "expected": True + }, + { + "name": "uptime_below_threshold_and_certs_stale", + "are_latest_certs_present": self.mock_is_mokutil_installed_return_false, + "run_command_output": self.mock_run_command_output_uptime_below_threshold, + "expected": False + }, + { + "name": "uptime_cmd_fails_unable_to_determine_uptime", + "are_latest_certs_present": self.mock_is_mokutil_installed_return_false, + "run_command_output": self.mock_run_command_output_uptime_cmd_fails, + "expected": False + }, + { + "name": "uptime_empty_output_unable_to_determine_uptime", + "are_latest_certs_present": self.mock_is_mokutil_installed_return_false, + "run_command_output": self.mock_run_command_output_uptime_empty_output, + "expected": False + }, + { + "name": "uptime_invalid_output_unable_to_parse_uptime", + "are_latest_certs_present": self.mock_is_mokutil_installed_return_false, + "run_command_output": self.mock_run_command_output_uptime_invalid_output, + "expected": False + } + ] + + try: + for use_case in use_cases: + package_manager.are_latest_certs_present = use_case["are_latest_certs_present"] + package_manager.env_layer.run_command_output = use_case["run_command_output"] + self.assertEqual( + package_manager.is_reboot_required_before_cert_update(), + use_case["expected"], + "Failed use case: {0}".format(use_case["name"]) + ) + finally: + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager.env_layer.run_command_output = backup_run_command_output + + def test_try_update_certs_success_paths__with_various_use_cases(self): package_manager = self.container.get('package_manager') backup_is_reboot_pending = package_manager.is_reboot_pending backup_are_latest_certs_present = package_manager.are_latest_certs_present backup_get_installed_fwupd_version = package_manager._AptitudePackageManager__get_installed_fwupd_version - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts - package_manager._AptitudePackageManager__get_installed_fwupd_version = self.mock_get_installed_fwupd_version_with_different_output_across_multiple_attempts - - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) - - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present - package_manager._AptitudePackageManager__get_installed_fwupd_version = backup_get_installed_fwupd_version - - def test_try_update_certs_when_fwupd_not_pre_installed(self): - package_manager = self.container.get('package_manager') - - backup_is_reboot_pending = package_manager.is_reboot_pending - backup_are_latest_certs_present = package_manager.are_latest_certs_present backup_run_command_output = package_manager.env_layer.run_command_output - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts - package_manager.env_layer.run_command_output = self.mock_run_command_output_fwupd_version_not_found_in_first_attempt_and_found_later - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + use_cases = [ + { + "name": "success", + "are_latest_certs_present": self.mock_are_latest_certs_present_return_true_once + }, + { + "name": "fwupd_meets_minimum_version", + "are_latest_certs_present": self.mock_are_latest_certs_present_return_true_once + }, + { + "name": "removes_old_fwupd_before_install", + "are_latest_certs_present": self.mock_are_latest_certs_present_return_true_once, + "mock_get_installed_fwupd_version": self.mock_get_installed_fwupd_version_with_different_output_across_multiple_attempts + }, + { + "name": "fwupd_not_pre_installed", + "are_latest_certs_present": self.mock_are_latest_certs_present_return_true_once, + "mock_run_command_output": self.mock_run_command_output_fwupd_version_not_found_in_first_attempt_and_found_later + } + ] - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present - package_manager.env_layer.run_command_output = backup_run_command_output + try: + for use_case in use_cases: + self.latest_certs_check_attempts = 0 + self.get_installed_fwupd_version_check_attempts = 0 - def test_try_update_certs_when_fwupd_install_failed(self): - self.runtime.set_legacy_test_type('FailInstallPath') - package_manager = self.container.get('package_manager') + package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager._AptitudePackageManager__get_installed_fwupd_version = backup_get_installed_fwupd_version + package_manager.env_layer.run_command_output = backup_run_command_output - backup_is_reboot_pending = package_manager.is_reboot_pending - backup_are_latest_certs_present = package_manager.are_latest_certs_present - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts + package_manager.are_latest_certs_present = use_case["are_latest_certs_present"] - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + if "mock_get_installed_fwupd_version" in use_case: + package_manager._AptitudePackageManager__get_installed_fwupd_version = use_case["mock_get_installed_fwupd_version"] - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present + if "mock_run_command_output" in use_case: + package_manager.env_layer.run_command_output = use_case["mock_run_command_output"] - def test_try_update_certs_when_fwupd_version_normalization_fails(self): - self.runtime.set_legacy_test_type('SuccessInstallPath') - package_manager = self.container.get('package_manager') + self.assertTrue(package_manager.try_update_certs(), "Failed use case: {0}".format(use_case["name"])) + self.assertEqual(package_manager.status_handler.is_reboot_pending, False, "Failed use case: {0}".format(use_case["name"])) + self.assertEqual(self.latest_certs_check_attempts, 1, "Failed use case: {0}".format(use_case["name"])) + finally: + package_manager.is_reboot_pending = backup_is_reboot_pending + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager._AptitudePackageManager__get_installed_fwupd_version = backup_get_installed_fwupd_version + package_manager.env_layer.run_command_output = backup_run_command_output - backup_is_reboot_pending = package_manager.is_reboot_pending - backup_are_latest_certs_present = package_manager.are_latest_certs_present - backup_min_fwupd_version = package_manager.min_fwupd_version - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts - package_manager.min_fwupd_version = "test" - - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) - - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present - package_manager.min_fwupd_version = backup_min_fwupd_version - - def test_try_update_certs_when_older_fwupd_installed_by_azgps(self): - self.runtime.set_legacy_test_type('SuccessInstallPath') + def test_try_update_certs_failure_paths__with_various_use_cases(self): package_manager = self.container.get('package_manager') backup_is_reboot_pending = package_manager.is_reboot_pending backup_are_latest_certs_present = package_manager.are_latest_certs_present - package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False - package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts + backup_min_fwupd_version = package_manager.min_fwupd_version - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + use_cases = [ + { + "name": "fwupd_install_failed", + "set_legacy_test_type": "FailInstallPath", + "are_latest_certs_present": self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts + }, + { + "name": "fwupd_version_normalization_fails", + "set_legacy_test_type": "SuccessInstallPath", + "are_latest_certs_present": self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts, + "min_fwupd_version": "test" + }, + { + "name": "older_fwupd_installed_by_azgps", + "set_legacy_test_type": "SuccessInstallPath", + "are_latest_certs_present": self.mock_are_latest_certs_present_with_different_output_across_multiple_attempts + } + ] - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present + try: + for use_case in use_cases: + package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager.min_fwupd_version = backup_min_fwupd_version - def test_try_update_certs_all_commands_succeed_but_certs_not_updated(self): - """ Test when all commands to update certs succeed but certs are still not updated, - which should return False and not change reboot pending status """ - package_manager = self.container.get('package_manager') - previous_reboot_pending_status = package_manager.status_handler.is_reboot_pending - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, previous_reboot_pending_status) + self.runtime.set_legacy_test_type(use_case["set_legacy_test_type"]) + package_manager.are_latest_certs_present = use_case["are_latest_certs_present"] - def test_try_update_certs_shell_command_fail_raises_exception(self): - """ Test when a command to update certs fails, which should return False and not change reboot pending status""" - self.runtime.set_legacy_test_type('SadPath') - package_manager = self.container.get('package_manager') - previous_reboot_pending_status = package_manager.status_handler.is_reboot_pending - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, previous_reboot_pending_status) + if "min_fwupd_version" in use_case: + package_manager.min_fwupd_version = use_case["min_fwupd_version"] - def test_try_update_certs_apt_command_fail_raises_exception(self): - """ Test when a command to update certs fails, which should return False and not change reboot pending status""" - self.runtime.set_legacy_test_type('FailInstallPath') - package_manager = self.container.get('package_manager') - previous_reboot_pending_status = package_manager.status_handler.is_reboot_pending - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, previous_reboot_pending_status) + self.assertFalse(package_manager.try_update_certs(), "Failed use case: {0}".format(use_case["name"])) + self.assertEqual(package_manager.status_handler.is_reboot_pending, False, "Failed use case: {0}".format(use_case["name"])) + finally: + package_manager.is_reboot_pending = backup_is_reboot_pending + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager.min_fwupd_version = backup_min_fwupd_version - def test_try_update_certs_shell_command_fail_no_exception_raised(self): - """ Test when a command to update certs fails, which should return False and not change reboot pending status""" - self.runtime.set_legacy_test_type('HappyPath') + def test_try_update_certs_command_failure_paths__with_various_use_cases(self): + """Test when cert update command paths fail and reboot pending status must not change.""" package_manager = self.container.get('package_manager') - backup_run_command_output = package_manager.env_layer.run_command_output - package_manager.env_layer.run_command_output = self.mock_run_command_output_fwupd_refresh_fails - previous_reboot_pending_status = package_manager.status_handler.is_reboot_pending - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, previous_reboot_pending_status) - - package_manager.env_layer.run_command_output = backup_run_command_output - - def test_try_update_certs_apt_command_fail_no_exception_raised(self): - """ Test when a command to update certs fails, which should return False and not change reboot pending status""" - self.runtime.set_legacy_test_type('HappyPath') + use_cases = [ + { + "name": "all_commands_succeed_but_certs_not_updated" + }, + { + "name": "shell_command_fail_raises_exception", + "set_legacy_test_type": "SadPath" + }, + { + "name": "apt_command_fail_raises_exception", + "set_legacy_test_type": "FailInstallPath" + }, + { + "name": "shell_command_fail_no_exception_raised", + "set_legacy_test_type": "HappyPath", + "mock_run_command_output": self.mock_run_command_output_fwupd_refresh_fails + }, + { + "name": "apt_command_fail_no_exception_raised", + "set_legacy_test_type": "HappyPath", + "mock_run_command_output": self.mock_run_command_output_apt_update_cmd_fails + } + ] + + try: + for use_case in use_cases: + package_manager.env_layer.run_command_output = backup_run_command_output + + if "set_legacy_test_type" in use_case: + self.runtime.set_legacy_test_type(use_case["set_legacy_test_type"]) + + if "mock_run_command_output" in use_case: + package_manager.env_layer.run_command_output = use_case["mock_run_command_output"] + + previous_reboot_pending_status = package_manager.status_handler.is_reboot_pending + self.assertFalse(package_manager.try_update_certs(), "Failed use case: {0}".format(use_case["name"])) + self.assertEqual( + package_manager.status_handler.is_reboot_pending, + previous_reboot_pending_status, + "Failed use case: {0}".format(use_case["name"]) + ) + finally: + package_manager.env_layer.run_command_output = backup_run_command_output + + def test_is_hibernation_enabled_for_cert_update__with_various_use_cases(self): + """Test is_hibernation_enabled_for_cert_update with various hibernation states""" package_manager = self.container.get('package_manager') - backup_run_command_output = package_manager.env_layer.run_command_output - package_manager.env_layer.run_command_output = self.mock_run_command_output_apt_update_cmd_fails - - previous_reboot_pending_status = package_manager.status_handler.is_reboot_pending - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, previous_reboot_pending_status) - - package_manager.env_layer.run_command_output = backup_run_command_output - - def test_update_certs_calls_try_update_when_mokutil_already_installed(self): - package_manager = self.container.get('package_manager') - backup_try_update_certs = package_manager.try_update_certs - - calls = [] - package_manager.try_update_certs = lambda: calls.append("try_update_certs") - package_manager.update_certs() - - self.assertEqual(calls, ["try_update_certs"]) - package_manager.try_update_certs = backup_try_update_certs - - def test_update_certs_calls_try_update_when_mokutil_install_succeeds(self): - package_manager = self.container.get('package_manager') - backup_is_mokutil_installed = package_manager.is_mokutil_installed - backup_try_update_certs = package_manager.try_update_certs - - calls = [] - package_manager.is_mokutil_installed = self.mock_is_mokutil_installed_return_false - package_manager.try_update_certs = lambda: calls.append("try_update_certs") - - package_manager.update_certs() - self.assertEqual(calls, ["try_update_certs"]) - - package_manager.is_mokutil_installed = backup_is_mokutil_installed - package_manager.try_update_certs = backup_try_update_certs - - def test_update_certs_raises_when_mokutil_install_fails(self): - self.runtime.set_legacy_test_type('SadPath') - package_manager = self.container.get('package_manager') - backup_add_error_to_status = package_manager.status_handler.add_error_to_status - - captured_errors = [] - package_manager.status_handler.add_error_to_status = lambda error_msg, error_code=None: captured_errors.append((error_msg, error_code)) - self.assertRaises(Exception, package_manager.update_certs) - self.assertEqual(len(captured_errors), 2) - self.assertIn("Customer environment error:", captured_errors[0][0]) - self.assertEqual(captured_errors[0][1], Constants.PatchOperationErrorCodes.PACKAGE_MANAGER_FAILURE) - self.assertIn("Mokutil is not installed and could not be installed", captured_errors[1][0]) - self.assertEqual(captured_errors[1][1], Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) - - package_manager.status_handler.add_error_to_status = backup_add_error_to_status + use_cases = [ + { + "name": "hibernation_enabled", + "run_command_output": self.mock_run_command_output_hibernation_enabled, + "expected_computed_is_hibernation_enabled": True + }, + { + "name": "hibernation_disabled", + "run_command_output": self.mock_run_command_output_hibernation_disabled, + "expected_computed_is_hibernation_enabled": False + }, + { + "name": "hibernation_cmd_fails", + "run_command_output": self.mock_run_command_output_hibernation_cmd_fails, + "expected_computed_is_hibernation_enabled": True + }, + { + "name": "hibernation_empty_output", + "run_command_output": self.mock_run_command_output_hibernation_empty_output, + "expected_computed_is_hibernation_enabled": True + }, + { + "name": "hibernation_none_output", + "run_command_output": self.mock_run_command_output_hibernation_none_output, + "expected_computed_is_hibernation_enabled": True + } + ] + + try: + for use_case in use_cases: + package_manager.env_layer.run_command_output = use_case["run_command_output"] + self.assertEqual( + package_manager.is_hibernation_enabled_for_cert_update(), use_case["expected_computed_is_hibernation_enabled"], "Failed use case: {0}".format(use_case["name"]) + ) + finally: + package_manager.env_layer.run_command_output = backup_run_command_output # endregion diff --git a/src/core/tests/Test_PatchInstaller.py b/src/core/tests/Test_PatchInstaller.py index affbe97d..a3fe6cf9 100644 --- a/src/core/tests/Test_PatchInstaller.py +++ b/src/core/tests/Test_PatchInstaller.py @@ -33,21 +33,56 @@ def tearDown(self): pass # region Mocks - def mock_update_certs_raise_exception(self): + def mock_try_update_certs_raise_exception(self): raise Exception("Simulated cert update failure") + def mock_update_certs_returns_false(self): + return False + def mock_detect_confidential_vm_raises_exception(self): raise Exception("Simulated VM detection failure") def mock_detect_confidential_vm_by_imds_returns_true(self): return True, 'IMDS:ConfidentialVM' + + def mock_detect_confidential_vm_returns_false(self): + return False, str() + + def mock_should_reboot_before_cert_update_returns_true(self): + return True + + def mock_should_reboot_before_cert_update_returns_false(self): + return False + + def mock_start_reboot_if_required_and_time_available_returns_false(self, current_time): + return False + + def mock_start_reboot_if_required_and_time_available_raises_exception(self, current_time): + raise Exception("Simulated reboot failure") + + def mock_is_hibernation_enabled_for_cert_update_returns_true(self): + return True + + def mock_is_hibernation_enabled_for_cert_update_returns_false(self): + return False + + def mock_is_hibernation_enabled_for_cert_update_raises_exception(self): + raise Exception("Simulated hibernation detection failure") + + def mock_are_latest_certs_present_with_mokutil_check_returns_true(self): + return True + + def mock_are_latest_certs_present_with_mokutil_check_returns_false(self): + return False # endregion # region Utility functions (update cert tests) - def _create_update_certs_runtime(self, enable_uefi_cert_update=True, health_store_id=None, operation=Constants.INSTALLATION): + def _create_update_certs_runtime(self, enable_uefi_cert_update=True, health_store_id=None, operation=Constants.INSTALLATION, reboot_setting=None): argument_composer = ArgumentComposer() argument_composer.health_store_id = health_store_id argument_composer.operation = operation + if reboot_setting is not None: + argument_composer.reboot_setting = reboot_setting config_file_path = Constants.AzGPSPaths.UEFI_SETTINGS config_settings = { "EnabledBy": "TestSetup", @@ -64,10 +99,14 @@ def __write_config_settings_to_file(config_settings, config_file_path): f.write(json.dumps(config_settings)) @staticmethod - def _track_method_call(obj, method_name): - call_count = [] - setattr(obj, method_name, lambda: call_count.append(True)) - return call_count + def _get_installation_error_messages(runtime): + 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"] + + installation_substatus = [item for item in substatus_file_data if item["name"] == Constants.PATCH_INSTALLATION_SUMMARY][0] + installation_message = json.loads(installation_substatus["formattedMessage"]["message"]) + error_details = installation_message["errors"]["details"] + return [error["message"] for error in error_details] # endregion def test_yum_install_updates_maintenance_window_exceeded(self): @@ -754,98 +793,355 @@ def test_get_max_batch_size(self): runtime.stop() # region test update certs - def test_try_update_certificates_for_default_patching(self): - """ Test update certificates code path depending on different configurations """ - - # Use case 1: Feature flag is off - enable_uefi_cert_update_usecase1 = False - health_store_id_usecase1 = "pub_off_sku_2025.01.01" - operation_usecase1 = Constants.INSTALLATION - method_to_track_usecase1 = 'try_update_certificates_for_default_patching' - is_package_manager_method_usecase1 = False - number_of_times_method_called_usecase1 = 0 - - # Use case 2: update_certs should NOT be called when health_store_id is None i.e. not a default patching operation - enable_uefi_cert_update_usecase2 = True - health_store_id_usecase2 = None - operation_usecase2 = Constants.INSTALLATION - method_to_track_usecase2 = 'update_certs' - is_package_manager_method_usecase2 = True - number_of_times_method_called_usecase2= 0 - - # Use case 3: """update_certs should NOT be called when health_store_id is an empty string i.e. not a default patching operation - enable_uefi_cert_update_usecase3 = True - health_store_id_usecase3 = str() - operation_usecase3 = Constants.INSTALLATION - method_to_track_usecase3 = 'update_certs' - is_package_manager_method_usecase3 = True - number_of_times_method_called_usecase3 = 0 - - # Use case 4: update_certs should NOT be called when operation is Assessment (not Installation). i.e. not default patching operation - enable_uefi_cert_update_usecase4 = True - health_store_id_usecase4 = None - operation_usecase4 = Constants.ASSESSMENT - method_to_track_usecase4 = 'update_certs' - is_package_manager_method_usecase4 = True - number_of_times_method_called_usecase4 = 0 - - # Use case 5: update_certs SHOULD be called when feature flag is on, health_store_id is set, and operation is Installation (for default patching) - enable_uefi_cert_update_usecase5 = True - health_store_id_usecase5 = "pub_off_sku_2025.01.01" - operation_usecase5 = Constants.INSTALLATION - method_to_track_usecase5 = 'update_certs' - is_package_manager_method_usecase5 = True - number_of_times_method_called_usecase5 = 1 + def test_try_update_certificates__with_various_use_cases(self): + """Test update certificate flow using consolidated use cases without losing scenario coverage.""" + + use_cases = [ + # Use case 1: Feature flag is off + { + "name": "feature_flag_off", + "enable_uefi_cert_update": False, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": None, + "expected_present": [], + "expected_absent": ["Certificates may not have been updated"] + }, + # Use case 2: try_update_certs should NOT be called when health_store_id is None (not a default patching operation) + { + "name": "not_default_patching_health_store_id_none", + "enable_uefi_cert_update": True, + "health_store_id": None, + "operation": Constants.INSTALLATION, + "reboot_setting": None, + "expected_present": [], + "expected_absent": ["Certificates may not have been updated"] + }, + # Use case 3: try_update_certs should NOT be called when health_store_id is an empty string (not a default patching operation) + { + "name": "not_default_patching_health_store_id_empty", + "enable_uefi_cert_update": True, + "health_store_id": str(), + "operation": Constants.INSTALLATION, + "reboot_setting": None, + "expected_present": [], + "expected_absent": ["Certificates may not have been updated"] + }, + # Use case 4: try_update_certs should NOT be called during Assessment (not a default patching operation) + { + "name": "not_default_patching_assessment_operation", + "enable_uefi_cert_update": True, + "health_store_id": None, + "operation": Constants.ASSESSMENT, + "reboot_setting": None, + "expected_present": [], + "expected_absent": ["Certificates may not have been updated"] + }, + # Use case 5: try_update_certs SHOULD be called when this is default patching and all prerequisites are met + { + "name": "default_patching_prereqs_met_update_certs_returns_false", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": "IfRequired", + "mock_should_reboot": self.mock_should_reboot_before_cert_update_returns_false, + "mock_detect_confidential_vm": self.mock_detect_confidential_vm_returns_false, + "mock_hibernation": self.mock_is_hibernation_enabled_for_cert_update_returns_false, + "mock_latest_certs": self.mock_are_latest_certs_present_with_mokutil_check_returns_false, + "expected_present": ["Certificates may not have been updated"], + "expected_absent": [] + }, + # Use case 6: update_certs should NOT be called when reboot is required but reboot setting is 'Never' + { + "name": "reboot_required_but_reboot_setting_never", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": "Never", + "mock_should_reboot": self.mock_should_reboot_before_cert_update_returns_true, + "mock_detect_confidential_vm": self.mock_detect_confidential_vm_returns_false, + "mock_hibernation": self.mock_is_hibernation_enabled_for_cert_update_returns_false, + "mock_latest_certs": self.mock_are_latest_certs_present_with_mokutil_check_returns_false, + "expected_present": ["reboot first"], + "expected_absent": [] + }, + # Use case 7: update_certs should NOT be called when VM is a CVM + { + "name": "skip_when_confidential_vm", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": "IfRequired", + "mock_should_reboot": self.mock_should_reboot_before_cert_update_returns_false, + "mock_detect_confidential_vm_by_imds": self.mock_detect_confidential_vm_by_imds_returns_true, + "expected_present": [], + "expected_absent": ["Confidential VM", "Certificates may not have been updated"] + }, + # Use case 8: update_certs should NOT be called when hibernation is enabled + { + "name": "hibernation_enabled", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": "IfRequired", + "mock_should_reboot": self.mock_should_reboot_before_cert_update_returns_false, + "mock_detect_confidential_vm": self.mock_detect_confidential_vm_returns_false, + "mock_hibernation": self.mock_is_hibernation_enabled_for_cert_update_returns_true, + "expected_present": ["Turn off hibernation"], + "expected_absent": [] + }, + # Use case 9: update_certs should NOT be called when latest certs are already present + { + "name": "latest_certs_already_present", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": "IfRequired", + "mock_should_reboot": self.mock_should_reboot_before_cert_update_returns_true, + "mock_detect_confidential_vm": self.mock_detect_confidential_vm_returns_false, + "mock_hibernation": self.mock_is_hibernation_enabled_for_cert_update_returns_false, + "mock_latest_certs": self.mock_are_latest_certs_present_with_mokutil_check_returns_true, + "expected_present": [], + "expected_absent": ["Certificates may not have been updated"] + }, + # Use case 10: exception from update_certs should be swallowed and recorded in status + { + "name": "update_certs_raises_exception_is_swallowed_and_reported", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": "IfRequired", + "mock_should_reboot": self.mock_should_reboot_before_cert_update_returns_false, + "mock_detect_confidential_vm": self.mock_detect_confidential_vm_returns_false, + "mock_hibernation": self.mock_is_hibernation_enabled_for_cert_update_returns_false, + "mock_latest_certs": self.mock_are_latest_certs_present_with_mokutil_check_returns_false, + "mock_try_update_certs": self.mock_try_update_certs_raise_exception, + "expected_present": ["attempting to update certificates"], + "expected_absent": [] + }, + # Use case 11: if confidential-VM detection throws, cert update should be skipped without cert-update failure status + { + "name": "skip_when_detect_confidential_vm_raises_exception", + "enable_uefi_cert_update": True, + "health_store_id": "pub_off_sku_2025.01.01", + "operation": Constants.INSTALLATION, + "reboot_setting": None, + "mock_detect_confidential_vm": self.mock_detect_confidential_vm_raises_exception, + "expected_present": [], + "expected_absent": ["Unable to determine whether the VM is a Confidential VM", "Certificates may not have been updated"] + } + ] + + for use_case in use_cases: + runtime = self._create_update_certs_runtime( + enable_uefi_cert_update=bool(use_case["enable_uefi_cert_update"]), + health_store_id=use_case["health_store_id"], + operation=use_case["operation"], + reboot_setting=use_case["reboot_setting"] + ) + + backup_should_reboot = runtime.package_manager.is_reboot_required_before_cert_update + backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm + backup_detect_confidential_vm_by_imds = runtime.env_layer.detect_confidential_vm_by_imds + backup_hibernation = runtime.package_manager.is_hibernation_enabled_for_cert_update + backup_latest_certs = runtime.package_manager.are_latest_certs_present + backup_try_update_certs = runtime.package_manager.try_update_certs + + if "mock_should_reboot" in use_case: + runtime.package_manager.is_reboot_required_before_cert_update = use_case["mock_should_reboot"] + if "mock_detect_confidential_vm" in use_case: + runtime.env_layer.detect_confidential_vm = use_case["mock_detect_confidential_vm"] + if "mock_detect_confidential_vm_by_imds" in use_case: + runtime.env_layer.detect_confidential_vm_by_imds = use_case["mock_detect_confidential_vm_by_imds"] + if "mock_hibernation" in use_case: + runtime.package_manager.is_hibernation_enabled_for_cert_update = use_case["mock_hibernation"] + if "mock_latest_certs" in use_case: + runtime.package_manager.are_latest_certs_present = use_case["mock_latest_certs"] + if "mock_try_update_certs" in use_case: + runtime.package_manager.try_update_certs = use_case["mock_try_update_certs"] + + runtime.patch_installer.start_installation(simulate=True) + error_messages = self._get_installation_error_messages(runtime) + + for expected_message in use_case["expected_present"]: + self.assertTrue( + any(expected_message in message for message in error_messages), + "Expected '{0}' for use case '{1}'".format(expected_message, use_case["name"]) + ) + + for forbidden_message in use_case["expected_absent"]: + self.assertFalse( + any(forbidden_message in message for message in error_messages), + "Did not expect '{0}' for use case '{1}'".format(forbidden_message, use_case["name"]) + ) + + runtime.package_manager.is_reboot_required_before_cert_update = backup_should_reboot + runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm + runtime.env_layer.detect_confidential_vm_by_imds = backup_detect_confidential_vm_by_imds + runtime.package_manager.is_hibernation_enabled_for_cert_update = backup_hibernation + runtime.package_manager.are_latest_certs_present = backup_latest_certs + runtime.package_manager.try_update_certs = backup_try_update_certs + runtime.stop() + + def test_can_continue_cert_update_after_reboot_check(self): + """ Test all branches of can_continue_cert_update_after_reboot_check() """ + + # Use case 1: Reboot required but reboot setting is 'Never' - should not reboot, should return False + reboot_setting_uc1 = 'Never' + mock_should_reboot_uc1 = self.mock_should_reboot_before_cert_update_returns_true + mock_start_reboot_uc1 = None + expected_result_uc1 = False + expected_reboot_status_uc1 = Constants.RebootStatus.NOT_NEEDED + + # Use case 2: Reboot required and reboot setting allows it - reboot is initiated, should return False + reboot_setting_uc2 = 'IfRequired' + mock_should_reboot_uc2 = self.mock_should_reboot_before_cert_update_returns_true + mock_start_reboot_uc2 = None + expected_result_uc2 = False + expected_reboot_status_uc2 = Constants.RebootStatus.STARTED + + # Use case 3: Reboot not required - cert update can continue, should return True + reboot_setting_uc3 = 'IfRequired' + mock_should_reboot_uc3 = self.mock_should_reboot_before_cert_update_returns_false + mock_start_reboot_uc3 = None + expected_result_uc3 = True + expected_reboot_status_uc3 = Constants.RebootStatus.NOT_NEEDED + + # Use case 4: Reboot required but reboot could not be initiated (returns False) - should return False + reboot_setting_uc4 = 'IfRequired' + mock_should_reboot_uc4 = self.mock_should_reboot_before_cert_update_returns_true + mock_start_reboot_uc4 = self.mock_start_reboot_if_required_and_time_available_returns_false + expected_result_uc4 = False + expected_reboot_status_uc4 = Constants.RebootStatus.NOT_NEEDED + + # Use case 5: Reboot required but reboot raises an exception - should return False + reboot_setting_uc5 = 'IfRequired' + mock_should_reboot_uc5 = self.mock_should_reboot_before_cert_update_returns_true + mock_start_reboot_uc5 = self.mock_start_reboot_if_required_and_time_available_raises_exception + expected_result_uc5 = False + expected_reboot_status_uc5 = Constants.RebootStatus.NOT_NEEDED test_input_output_table = [ - [enable_uefi_cert_update_usecase1, health_store_id_usecase1, operation_usecase1, method_to_track_usecase1, is_package_manager_method_usecase1, number_of_times_method_called_usecase1], - [enable_uefi_cert_update_usecase2, health_store_id_usecase2, operation_usecase2, method_to_track_usecase2, is_package_manager_method_usecase2, number_of_times_method_called_usecase2], - [enable_uefi_cert_update_usecase3, health_store_id_usecase3, operation_usecase3, method_to_track_usecase3, is_package_manager_method_usecase3, number_of_times_method_called_usecase3], - [enable_uefi_cert_update_usecase4, health_store_id_usecase4, operation_usecase4, method_to_track_usecase4, is_package_manager_method_usecase4, number_of_times_method_called_usecase4], - [enable_uefi_cert_update_usecase5, health_store_id_usecase5, operation_usecase5, method_to_track_usecase5, is_package_manager_method_usecase5, number_of_times_method_called_usecase5], + [reboot_setting_uc1, mock_should_reboot_uc1, mock_start_reboot_uc1, expected_result_uc1, expected_reboot_status_uc1], + [reboot_setting_uc2, mock_should_reboot_uc2, mock_start_reboot_uc2, expected_result_uc2, expected_reboot_status_uc2], + [reboot_setting_uc3, mock_should_reboot_uc3, mock_start_reboot_uc3, expected_result_uc3, expected_reboot_status_uc3], + [reboot_setting_uc4, mock_should_reboot_uc4, mock_start_reboot_uc4, expected_result_uc4, expected_reboot_status_uc4], + [reboot_setting_uc5, mock_should_reboot_uc5, mock_start_reboot_uc5, expected_result_uc5, expected_reboot_status_uc5], ] for row in test_input_output_table: - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=bool(row[0]), health_store_id=row[1], operation=row[2]) - method_called = self._track_method_call(runtime.patch_installer.package_manager if row[4] == True else runtime.patch_installer, row[3]) - runtime.patch_installer.start_installation(simulate=True) - self.assertEqual(len(method_called), row[5], "Failed for row: {row}") + reboot_setting, mock_should_reboot, mock_start_reboot, expected_result, expected_reboot_status = row + + runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting=reboot_setting) + backup_should_reboot_before_cert_update = runtime.package_manager.is_reboot_required_before_cert_update + backup_start_reboot_if_required_and_time_available = runtime.patch_installer.reboot_manager.start_reboot_if_required_and_time_available + + original_force_reboot = runtime.package_manager.force_reboot + runtime.package_manager.is_reboot_required_before_cert_update = mock_should_reboot + if mock_start_reboot is not None: + runtime.patch_installer.reboot_manager.start_reboot_if_required_and_time_available = mock_start_reboot + + self.assertEqual(runtime.patch_installer.ensure_pre_cert_update_reboot_completed(), expected_result, "Failed for use case with reboot_setting={0}, should_reboot={1}".format(reboot_setting, mock_should_reboot.__name__)) + self.assertEqual(runtime.status_handler.get_installation_reboot_status(), expected_reboot_status, "Unexpected reboot status for use case with reboot_setting={0}".format(reboot_setting)) + self.assertEqual(runtime.package_manager.force_reboot, original_force_reboot, "force_reboot was not restored for use case with reboot_setting={0}".format(reboot_setting)) + + runtime.package_manager.is_reboot_required_before_cert_update = backup_should_reboot_before_cert_update + runtime.patch_installer.reboot_manager.start_reboot_if_required_and_time_available = backup_start_reboot_if_required_and_time_available runtime.stop() - def test_try_update_certs_swallows_exception_from_update_certs(self): - """An exception raised by update_certs should be swallowed and not propagate.""" - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01") - backup_up_update_certs = runtime.patch_installer.package_manager.update_certs + def test_can_continue_cert_update_after_hibernation_check(self): + """Test all branches of can_continue_cert_update_after_hibernation_check().""" - runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_raise_exception - runtime.patch_installer.start_installation(simulate=True) + # Use case 1: Hibernation enabled - cert update should not continue and an error should be recorded. + hibernation_check_uc1 = self.mock_is_hibernation_enabled_for_cert_update_returns_true + expected_result_uc1 = False + expected_hibernation_error_present_uc1 = True - runtime.patch_installer.package_manager.update_certs = backup_up_update_certs - runtime.stop() + # Use case 2: Hibernation disabled - cert update can continue. + hibernation_check_uc2 = self.mock_is_hibernation_enabled_for_cert_update_returns_false + expected_result_uc2 = True + expected_hibernation_error_present_uc2 = False - def test_try_update_certificates_skips_confidential_vm(self): - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01") - backup_detect_confidential_vm_by_imds = runtime.env_layer.detect_confidential_vm_by_imds + # Use case 3: Hibernation check throws - cert update should not continue and an error should be recorded. + hibernation_check_uc3 = self.mock_is_hibernation_enabled_for_cert_update_raises_exception + expected_result_uc3 = False + expected_hibernation_error_present_uc3 = True - runtime.env_layer.detect_confidential_vm_by_imds = self.mock_detect_confidential_vm_by_imds_returns_true - method_called = self._track_method_call(runtime.patch_installer.package_manager, 'update_certs') - runtime.patch_installer.start_installation(simulate=True) - self.assertEqual(len(method_called), 0) + test_input_output_table = [ + [hibernation_check_uc1, expected_result_uc1, expected_hibernation_error_present_uc1, "Turn off hibernation"], + [hibernation_check_uc2, expected_result_uc2, expected_hibernation_error_present_uc2, ""], + [hibernation_check_uc3, expected_result_uc3, expected_hibernation_error_present_uc3, "Unable to determine hibernation state"], + ] - runtime.env_layer.detect_confidential_vm_by_imds = backup_detect_confidential_vm_by_imds - runtime.stop() + for row in test_input_output_table: + mock_hibernation_check, expected_result, expected_hibernation_error_present, expected_error_substring = row - def test_try_update_certificates_skips_when_detect_confidential_vm_raises_exception(self): - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01") - backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm + runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting='IfRequired') + backup_is_hibernation_enabled_for_cert_update = runtime.package_manager.is_hibernation_enabled_for_cert_update - runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_raises_exception - method_called = self._track_method_call(runtime.patch_installer.package_manager, 'update_certs') - runtime.patch_installer.start_installation(simulate=True) - self.assertEqual(len(method_called), 0) + runtime.package_manager.is_hibernation_enabled_for_cert_update = mock_hibernation_check + runtime.status_handler.set_current_operation(Constants.INSTALLATION) + runtime.status_handler.set_installation_substatus_json() - runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm - runtime.stop() + self.assertEqual(runtime.patch_installer.can_continue_cert_update_after_hibernation_check(), expected_result, + "Failed for hibernation check use case: {0}".format(mock_hibernation_check.__name__)) + + 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"] + + installation_substatus = [item for item in substatus_file_data if item["name"] == Constants.PATCH_INSTALLATION_SUMMARY][0] + installation_message = json.loads(installation_substatus["formattedMessage"]["message"]) + error_details = installation_message["errors"]["details"] + is_hibernation_error_present = any(expected_error_substring in error["message"] for error in error_details) + self.assertEqual(is_hibernation_error_present, expected_hibernation_error_present, + "Unexpected hibernation error status for use case: {0}".format(mock_hibernation_check.__name__)) + + runtime.package_manager.is_hibernation_enabled_for_cert_update = backup_is_hibernation_enabled_for_cert_update + runtime.stop() + + def test_can_continue_cert_update_after_latest_cert_check(self): + """Test all branches of can_continue_cert_update_after_latest_cert_check().""" + + # Use case 1: Latest certs already present - cert update should not continue. + latest_certs_check_uc1 = self.mock_are_latest_certs_present_with_mokutil_check_returns_true + expected_result_uc1 = False + expected_latest_cert_error_present_uc1 = False + + # Use case 2: Latest certs not present - cert update can continue. + latest_certs_check_uc2 = self.mock_are_latest_certs_present_with_mokutil_check_returns_false + expected_result_uc2 = True + expected_latest_cert_error_present_uc2 = False + + test_input_output_table = [ + [latest_certs_check_uc1, expected_result_uc1, expected_latest_cert_error_present_uc1], + [latest_certs_check_uc2, expected_result_uc2, expected_latest_cert_error_present_uc2], + ] + + for row in test_input_output_table: + mock_latest_cert_check, expected_result, expected_latest_cert_error_present = row + + runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting='IfRequired') + backup_are_latest_certs_present = runtime.package_manager.are_latest_certs_present_with_mokutil_check + + runtime.package_manager.are_latest_certs_present_with_mokutil_check = mock_latest_cert_check + runtime.status_handler.set_current_operation(Constants.INSTALLATION) + runtime.status_handler.set_installation_substatus_json() + + self.assertEqual(runtime.patch_installer.can_continue_cert_update_after_latest_cert_check(), expected_result, + "Failed for latest cert check use case: {0}".format(getattr(mock_latest_cert_check, '__name__', 'lambda'))) + + 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"] + + installation_substatus = [item for item in substatus_file_data if item["name"] == Constants.PATCH_INSTALLATION_SUMMARY][0] + installation_message = json.loads(installation_substatus["formattedMessage"]["message"]) + error_details = installation_message["errors"]["details"] + is_latest_cert_error_present = any("latest certs" in error["message"].lower() for error in error_details) + self.assertEqual(is_latest_cert_error_present, expected_latest_cert_error_present, + "Unexpected latest cert error status for use case: {0}".format(getattr(mock_latest_cert_check, '__name__', 'lambda'))) + + runtime.package_manager.are_latest_certs_present_with_mokutil_check = backup_are_latest_certs_present + runtime.stop() # endregion