diff --git a/src/core/src/core_logic/PatchInstaller.py b/src/core/src/core_logic/PatchInstaller.py index 81bb0976..efc00a49 100644 --- a/src/core/src/core_logic/PatchInstaller.py +++ b/src/core/src/core_logic/PatchInstaller.py @@ -808,24 +808,139 @@ 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: + certs_updated = self.package_manager.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""" + + if self.__is_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.can_continue_cert_update_after_reboot_check(): + return False + + return True + + def __is_cvm(self): + # type: () -> bool + """ Returns true if this is a Confidential VM (CVM) """ try: 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 True 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 is_confidential_vm + + def can_continue_cert_update_after_reboot_check(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. + """ + + should_reboot_before_cert_update = self.package_manager.should_reboot_before_cert_update() + if should_reboot_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.""" + try: + latest_certs_present = self.package_manager.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. Not continuing with certificate update. [Error: {0}]".format(str(e))) + return False + + 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..1a539b96 100644 --- a/src/core/src/package_managers/AptitudePackageManager.py +++ b/src/core/src/package_managers/AptitudePackageManager.py @@ -102,6 +102,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 # region Sources Management def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_classification=str()): @@ -957,6 +960,57 @@ 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 should_reboot_before_cert_update(self): + # type: () -> bool + """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: + self.composite_logger.log_debug('[APM][Certs] Unable to determine hibernation state for cert update. Assuming disabled. [Code={0}][Output={1}]'.format(str(code), out)) + return False + + # 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 +1023,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: diff --git a/src/core/src/package_managers/Dnf5PackageManager.py b/src/core/src/package_managers/Dnf5PackageManager.py index d2942a1c..a3960198 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 should_reboot_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..d7108b81 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 @@ -498,31 +497,27 @@ 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 - + # type: () -> bool + """ Updates the certificates if the required ones do not exist. Returns True if certs were updated successfully, False otherwise. """ 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() + return self.try_update_certs() - 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) -> 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 @@ -534,6 +529,9 @@ def is_mokutil_installed(self): return code == 0 def are_latest_certs_present(self): + if not self.ensure_mokutil_available_for_cert_checks(): + return False + kek_code, kek_out = self.fetch_current_certs(Constants.Certificates.KEK, self.get_kek_cert_status_cmd, raise_on_exception=False) 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) @@ -577,5 +575,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 should_reboot_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..a4047cea 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 should_reboot_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..7c8cccff 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 should_reboot_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..268185a7 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 should_reboot_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..f6bd4c44 100644 --- a/src/core/tests/Test_AptitudePackageManager.py +++ b/src/core/tests/Test_AptitudePackageManager.py @@ -98,6 +98,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 +112,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 +138,69 @@ 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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(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.assertTrue(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1) + return 0, None # endregion Mocks # region Utility Functions @@ -1154,246 +1207,429 @@ 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') + def test_are_latest_certs_present__with_various_use_cases(self): 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): - 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 + cert_output = "{0}\n{1}".format("issuer: test", Constants.LATEST_CERTIFICATE_TIMESTAMP) - 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, + "expect_exception": True, + "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 + + 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"] + + context = {"fetch_calls": [], "captured_errors": []} + + 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 = lambda: False + package_manager.fetch_current_certs = lambda cert_type, get_cert_status_cmd, raise_on_exception=False: context["fetch_calls"].append(cert_type) + package_manager.status_handler.add_error_to_status = lambda error_msg, error_code=None: context["captured_errors"].append((error_msg, error_code)) + + if use_case.get("force_ensure_mokutil_false", False): + package_manager.ensure_mokutil_available_for_cert_checks = lambda: 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 = lambda: True + package_manager.fetch_current_certs = lambda cert_type, get_cert_status_cmd, raise_on_exception=False: (0, cert_output) + + if use_case.get("expect_exception", False): + self.assertRaises(Exception, package_manager.are_latest_certs_present) + else: + self.assertEqual( + package_manager.are_latest_certs_present(), + use_case["expected"], + "Failed use case: {0}".format(use_case["name"]) + ) + + if use_case.get("validate_error_capture", False): + self.assertEqual(context["fetch_calls"], []) + self.assertEqual(len(context["captured_errors"]), 1) + self.assertIn("Mokutil is not installed and could not be installed", context["captured_errors"][0][0]) + self.assertEqual(context["captured_errors"][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') + backup_are_latest_certs_present = package_manager.are_latest_certs_present + backup_ensure_fwupd_installation = package_manager._AptitudePackageManager__ensure_fwupd_installation shell_calls = [] apt_calls = [] + package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_return_true + package_manager._AptitudePackageManager__ensure_fwupd_installation = lambda: True 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) + self.assertEqual(shell_calls, ["FwupdRefresh", "FwupdUpdate"]) + self.assertEqual(apt_calls, ["AptUpdate"]) - package_manager.is_reboot_pending = backup_is_reboot_pending package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager._AptitudePackageManager__ensure_fwupd_installation = backup_ensure_fwupd_installation - def test_try_update_certs_when_fwupd_meets_minimum_version(self): + 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.should_reboot_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) - 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 - - def test_try_update_certs_when_fwupd_install_failed(self): - self.runtime.set_legacy_test_type('FailInstallPath') - package_manager = self.container.get('package_manager') + 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 + } + ] - 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 + try: + for use_case in use_cases: + self.latest_certs_check_attempts = 0 + self.get_installed_fwupd_version_check_attempts = 0 - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + 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 - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager.are_latest_certs_present = use_case["are_latest_certs_present"] - 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') + if "mock_get_installed_fwupd_version" in use_case: + package_manager._AptitudePackageManager__get_installed_fwupd_version = use_case["mock_get_installed_fwupd_version"] - 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" + if "mock_run_command_output" in use_case: + package_manager.env_layer.run_command_output = use_case["mock_run_command_output"] - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + 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 - 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 - - 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 - - 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) - - 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) + backup_min_fwupd_version = package_manager.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) + 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 + } + ] - 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') - package_manager = self.container.get('package_manager') + 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 - 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 + self.runtime.set_legacy_test_type(use_case["set_legacy_test_type"]) + package_manager.are_latest_certs_present = use_case["are_latest_certs_present"] - 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"] - package_manager.env_layer.run_command_output = backup_run_command_output + 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_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') + 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_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) + 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 + } + ] - package_manager.env_layer.run_command_output = backup_run_command_output + try: + for use_case in use_cases: + 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 + if "set_legacy_test_type" in use_case: + self.runtime.set_legacy_test_type(use_case["set_legacy_test_type"]) - calls = [] - package_manager.try_update_certs = lambda: calls.append("try_update_certs") - package_manager.update_certs() + if "mock_run_command_output" in use_case: + package_manager.env_layer.run_command_output = use_case["mock_run_command_output"] - self.assertEqual(calls, ["try_update_certs"]) - package_manager.try_update_certs = backup_try_update_certs + 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_update_certs_calls_try_update_when_mokutil_install_succeeds(self): + def test_update_certs__with_various_use_cases(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') + use_cases = [ + { + "name": "calls_try_update_when_mokutil_already_installed", + "mock_try_update_to_track_calls": True, + "expected_calls": ["try_update_certs"] + }, + { + "name": "calls_try_update_when_mokutil_install_succeeds", + "force_mokutil_not_installed": True, + "mock_try_update_to_track_calls": True, + "expected_calls": ["try_update_certs"] + }, + { + "name": "returns_false_when_try_update_reports_failure", + "mock_try_update_returns_false": True, + "expected_result": False + } + ] + + try: + for use_case in use_cases: + package_manager.is_mokutil_installed = backup_is_mokutil_installed + package_manager.try_update_certs = backup_try_update_certs + + calls = [] + if use_case.get("force_mokutil_not_installed", False): + package_manager.is_mokutil_installed = self.mock_is_mokutil_installed_return_false + + if use_case.get("mock_try_update_to_track_calls", False): + package_manager.try_update_certs = lambda: calls.append("try_update_certs") + + if use_case.get("mock_try_update_returns_false", False): + package_manager.try_update_certs = lambda: False + + result = package_manager.update_certs() + + if "expected_calls" in use_case: + self.assertEqual( + calls, + use_case["expected_calls"], + "Failed use case: {0}".format(use_case["name"]) + ) + + if "expected_result" in use_case: + self.assertEqual( + result, + use_case["expected_result"], + "Failed use case: {0}".format(use_case["name"]) + ) + finally: + package_manager.is_mokutil_installed = backup_is_mokutil_installed + package_manager.try_update_certs = backup_try_update_certs + + 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_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) + backup_run_command_output = package_manager.env_layer.run_command_output - 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": True + }, + { + "name": "hibernation_disabled", + "run_command_output": self.mock_run_command_output_hibernation_disabled, + "expected": False + }, + { + "name": "hibernation_cmd_fails", + "run_command_output": self.mock_run_command_output_hibernation_cmd_fails, + "expected": False + }, + { + "name": "hibernation_empty_output", + "run_command_output": self.mock_run_command_output_hibernation_empty_output, + "expected": True + }, + { + "name": "hibernation_none_output", + "run_command_output": self.mock_run_command_output_hibernation_none_output, + "expected": 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"], "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..ae8cf821 100644 --- a/src/core/tests/Test_PatchInstaller.py +++ b/src/core/tests/Test_PatchInstaller.py @@ -36,18 +36,53 @@ def tearDown(self): def mock_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_returns_true(self): + return True + + def mock_are_latest_certs_present_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,369 @@ 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: 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: 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: 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: 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_returns_false, + "mock_update_certs": self.mock_update_certs_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_returns_false, + "mock_update_certs": self.mock_update_certs_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, + "mock_update_certs": self.mock_update_certs_returns_false, + "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, + "mock_update_certs": self.mock_update_certs_returns_false, + "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_returns_true, + "mock_update_certs": self.mock_update_certs_returns_false, + "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_returns_false, + "mock_update_certs": self.mock_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, + "mock_update_certs": self.mock_update_certs_returns_false, + "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.should_reboot_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_update_certs = runtime.patch_installer.package_manager.update_certs + + if "mock_should_reboot" in use_case: + runtime.package_manager.should_reboot_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_update_certs" in use_case: + runtime.patch_installer.package_manager.update_certs = use_case["mock_update_certs"] + else: + runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_returns_false + + 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.should_reboot_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.patch_installer.package_manager.update_certs = backup_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.should_reboot_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.should_reboot_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.can_continue_cert_update_after_reboot_check(), 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.should_reboot_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_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_returns_false + expected_result_uc2 = True + expected_latest_cert_error_present_uc2 = False + + # Use case 3: Latest cert check throws - cert update should not continue, but this is only warned/logged. + latest_certs_check_uc3 = lambda: (_ for _ in ()).throw(Exception("Simulated latest cert detection failure")) + expected_result_uc3 = False + expected_latest_cert_error_present_uc3 = 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], + [latest_certs_check_uc3, expected_result_uc3, expected_latest_cert_error_present_uc3], + ] + + 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 + + runtime.package_manager.are_latest_certs_present = 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 = backup_are_latest_certs_present + runtime.stop() # endregion