From ea24502755a3c15be734205fce1204565ebc3961 Mon Sep 17 00:00:00 2001 From: Rajasi Rane Date: Fri, 10 Jul 2026 15:38:00 -0700 Subject: [PATCH 1/3] Feature: [Ubuntu, no-CVM] Adding reboot, hibernate checks and aligning all pre-reqs together --- src/core/src/core_logic/PatchInstaller.py | 133 +++++- .../AptitudePackageManager.py | 57 ++- .../package_managers/Dnf5PackageManager.py | 8 + .../src/package_managers/PackageManager.py | 55 ++- .../package_managers/TdnfPackageManager.py | 8 + .../src/package_managers/YumPackageManager.py | 8 + .../package_managers/ZypperPackageManager.py | 8 + src/core/tests/Test_AptitudePackageManager.py | 142 +++++- src/core/tests/Test_PatchInstaller.py | 428 ++++++++++++++++-- 9 files changed, 756 insertions(+), 91 deletions(-) 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..82e30208 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 @@ -148,6 +152,19 @@ def mock_run_command_output_apt_update_cmd_fails(self, cmd, no_output=False, chk 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): + if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: + return 0, "700000.12 123.45" + return 0, "" + + def mock_run_command_output_uptime_below_threshold(self, cmd, no_output=False, chk_err=True): + if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: + return 0, "120.12 123.45" + return 0, "" # endregion Mocks # region Utility Functions @@ -1177,18 +1194,107 @@ def test_are_latest_certs_present_returns_false_when_db_fetch_fails(self): 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): + def test_are_latest_certs_present_returns_false_when_mokutil_missing_and_install_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 + + fetch_calls = [] + captured_errors = [] + 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: fetch_calls.append(cert_type) + 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.are_latest_certs_present) + self.assertEqual(fetch_calls, []) + self.assertEqual(len(captured_errors), 1) + self.assertIn("Mokutil is not installed and could not be installed", captured_errors[0][0]) + self.assertEqual(captured_errors[0][1], Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) + + 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 + + def test_are_latest_certs_present_checks_certs_when_mokutil_install_succeeds(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 + + cert_output = "{0}\n{1}".format("issuer: test", Constants.LATEST_CERTIFICATE_TIMESTAMP) + 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) + + self.assertTrue(package_manager.are_latest_certs_present()) + + 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 + + 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, []) + self.assertEqual(shell_calls, ["FwupdRefresh", "FwupdUpdate"]) + self.assertEqual(apt_calls, ["AptUpdate"]) + + package_manager.are_latest_certs_present = backup_are_latest_certs_present + package_manager._AptitudePackageManager__ensure_fwupd_installation = backup_ensure_fwupd_installation + + def test_should_reboot_before_cert_update__with_various_use_cases(self): + package_manager = self.container.get('package_manager') + + backup_are_latest_certs_present = package_manager.are_latest_certs_present + backup_run_command_output = package_manager.env_layer.run_command_output + + 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 + } + ] + + 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(self): package_manager = self.container.get('package_manager') @@ -1196,10 +1302,11 @@ def test_try_update_certs_success(self): 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 = self.mock_are_latest_certs_present_return_true_once self.assertTrue(package_manager.try_update_certs()) self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + self.assertEqual(self.latest_certs_check_attempts, 1) package_manager.is_reboot_pending = backup_is_reboot_pending package_manager.are_latest_certs_present = backup_are_latest_certs_present @@ -1210,10 +1317,11 @@ def test_try_update_certs_when_fwupd_meets_minimum_version(self): 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 = self.mock_are_latest_certs_present_return_true_once self.assertTrue(package_manager.try_update_certs()) self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + self.assertEqual(self.latest_certs_check_attempts, 1) package_manager.is_reboot_pending = backup_is_reboot_pending package_manager.are_latest_certs_present = backup_are_latest_certs_present @@ -1225,11 +1333,12 @@ def test_try_update_certs_removes_old_fwupd_before_install(self): 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.are_latest_certs_present = self.mock_are_latest_certs_present_return_true_once 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) + self.assertEqual(self.latest_certs_check_attempts, 1) package_manager.is_reboot_pending = backup_is_reboot_pending package_manager.are_latest_certs_present = backup_are_latest_certs_present @@ -1242,11 +1351,12 @@ def test_try_update_certs_when_fwupd_not_pre_installed(self): 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.are_latest_certs_present = self.mock_are_latest_certs_present_return_true_once 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) + self.assertEqual(self.latest_certs_check_attempts, 1) package_manager.is_reboot_pending = backup_is_reboot_pending package_manager.are_latest_certs_present = backup_are_latest_certs_present @@ -1378,22 +1488,14 @@ def test_update_certs_calls_try_update_when_mokutil_install_succeeds(self): 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') + def test_update_certs_returns_false_when_try_update_reports_failure(self): 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)) + backup_try_update_certs = package_manager.try_update_certs - 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.try_update_certs = lambda: False + self.assertFalse(package_manager.update_certs()) - package_manager.status_handler.add_error_to_status = backup_add_error_to_status + package_manager.try_update_certs = backup_try_update_certs # endregion diff --git a/src/core/tests/Test_PatchInstaller.py b/src/core/tests/Test_PatchInstaller.py index affbe97d..1d28b7c7 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): @@ -761,91 +800,408 @@ def test_try_update_certificates_for_default_patching(self): 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 + reboot_setting_usecase1 = None + mock_should_reboot_usecase1 = None + expected_error_message_usecase1 = None + mock_detect_confidential_vm_usecase1 = None + # hibernation check not reached (feature flag off, returns before prereq checks) # 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 + reboot_setting_usecase2 = None + mock_should_reboot_usecase2 = None + expected_error_message_usecase2 = None + mock_detect_confidential_vm_usecase2 = None + # hibernation check not reached (not a default patching operation) - # Use case 3: """update_certs should NOT be called when health_store_id is an empty string i.e. not a default patching operation + # 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 + reboot_setting_usecase3 = None + mock_should_reboot_usecase3 = None + expected_error_message_usecase3 = None + mock_detect_confidential_vm_usecase3 = None + # hibernation check not reached (not a default patching operation) # 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 + reboot_setting_usecase4 = None + mock_should_reboot_usecase4 = None + expected_error_message_usecase4 = None + mock_detect_confidential_vm_usecase4 = None + # hibernation check not reached (not a default patching operation) - # 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) + # Use case 5: update_certs SHOULD be called only when this is default patching and all prerequisites are met. 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 + reboot_setting_usecase5 = 'IfRequired' + mock_should_reboot_usecase5 = self.mock_should_reboot_before_cert_update_returns_false + expected_error_message_usecase5 = "Certificates may not have been updated" + mock_detect_confidential_vm_usecase5 = self.mock_detect_confidential_vm_returns_false + mock_hibernation_usecase5 = self.mock_is_hibernation_enabled_for_cert_update_returns_false # explicitly disable hibernation so this prereq passes + mock_latest_certs_usecase5 = self.mock_are_latest_certs_present_returns_false + + # Use case 6: update_certs should NOT be called when a reboot is required but the reboot setting is 'Never' + enable_uefi_cert_update_usecase6 = True + health_store_id_usecase6 = "pub_off_sku_2025.01.01" + operation_usecase6 = Constants.INSTALLATION + reboot_setting_usecase6 = 'Never' + mock_should_reboot_usecase6 = self.mock_should_reboot_before_cert_update_returns_true + expected_error_message_usecase6 = "reboot first" + mock_detect_confidential_vm_usecase6 = self.mock_detect_confidential_vm_returns_false + mock_hibernation_usecase6 = self.mock_is_hibernation_enabled_for_cert_update_returns_false + mock_latest_certs_usecase6 = self.mock_are_latest_certs_present_returns_false + + # Use case 7: update_certs should NOT be called when VM is a CVM. + enable_uefi_cert_update_usecase7 = True + health_store_id_usecase7 = "pub_off_sku_2025.01.01" + operation_usecase7 = Constants.INSTALLATION + reboot_setting_usecase7 = 'IfRequired' + mock_should_reboot_usecase7 = self.mock_should_reboot_before_cert_update_returns_false + expected_error_message_usecase7 = None + mock_detect_confidential_vm_usecase7 = self.mock_detect_confidential_vm_by_imds_returns_true + mock_hibernation_usecase7 = None # hibernation check not reached (CVM check fails first) + mock_latest_certs_usecase7 = None + + # Use case 8: update_certs should NOT be called when hibernation is enabled. + enable_uefi_cert_update_usecase8 = True + health_store_id_usecase8 = "pub_off_sku_2025.01.01" + operation_usecase8 = Constants.INSTALLATION + reboot_setting_usecase8 = 'IfRequired' + mock_should_reboot_usecase8 = self.mock_should_reboot_before_cert_update_returns_false + expected_error_message_usecase8 = "Turn off hibernation" + mock_detect_confidential_vm_usecase8 = self.mock_detect_confidential_vm_returns_false + mock_hibernation_usecase8 = self.mock_is_hibernation_enabled_for_cert_update_returns_true + mock_latest_certs_usecase8 = None # latest cert check not reached when hibernation is enabled + + # Use case 9: update_certs should NOT be called when latest certs are already present. + enable_uefi_cert_update_usecase9 = True + health_store_id_usecase9 = "pub_off_sku_2025.01.01" + operation_usecase9 = Constants.INSTALLATION + reboot_setting_usecase9 = 'IfRequired' + mock_should_reboot_usecase9 = self.mock_should_reboot_before_cert_update_returns_true + expected_error_message_usecase9 = None + mock_detect_confidential_vm_usecase9 = self.mock_detect_confidential_vm_returns_false + mock_hibernation_usecase9 = self.mock_is_hibernation_enabled_for_cert_update_returns_false + mock_latest_certs_usecase9 = self.mock_are_latest_certs_present_returns_true 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], + [enable_uefi_cert_update_usecase1, health_store_id_usecase1, operation_usecase1, reboot_setting_usecase1, mock_should_reboot_usecase1, expected_error_message_usecase1, mock_detect_confidential_vm_usecase1, None, None], + [enable_uefi_cert_update_usecase2, health_store_id_usecase2, operation_usecase2, reboot_setting_usecase2, mock_should_reboot_usecase2, expected_error_message_usecase2, mock_detect_confidential_vm_usecase2, None, None], + [enable_uefi_cert_update_usecase3, health_store_id_usecase3, operation_usecase3, reboot_setting_usecase3, mock_should_reboot_usecase3, expected_error_message_usecase3, mock_detect_confidential_vm_usecase3, None, None], + [enable_uefi_cert_update_usecase4, health_store_id_usecase4, operation_usecase4, reboot_setting_usecase4, mock_should_reboot_usecase4, expected_error_message_usecase4, mock_detect_confidential_vm_usecase4, None, None], + [enable_uefi_cert_update_usecase5, health_store_id_usecase5, operation_usecase5, reboot_setting_usecase5, mock_should_reboot_usecase5, expected_error_message_usecase5, mock_detect_confidential_vm_usecase5, mock_hibernation_usecase5, mock_latest_certs_usecase5], + [enable_uefi_cert_update_usecase6, health_store_id_usecase6, operation_usecase6, reboot_setting_usecase6, mock_should_reboot_usecase6, expected_error_message_usecase6, mock_detect_confidential_vm_usecase6, mock_hibernation_usecase6, mock_latest_certs_usecase6], + [enable_uefi_cert_update_usecase7, health_store_id_usecase7, operation_usecase7, reboot_setting_usecase7, mock_should_reboot_usecase7, expected_error_message_usecase7, mock_detect_confidential_vm_usecase7, mock_hibernation_usecase7, mock_latest_certs_usecase7], + [enable_uefi_cert_update_usecase8, health_store_id_usecase8, operation_usecase8, reboot_setting_usecase8, mock_should_reboot_usecase8, expected_error_message_usecase8, mock_detect_confidential_vm_usecase8, mock_hibernation_usecase8, mock_latest_certs_usecase8], + [enable_uefi_cert_update_usecase9, health_store_id_usecase9, operation_usecase9, reboot_setting_usecase9, mock_should_reboot_usecase9, expected_error_message_usecase9, mock_detect_confidential_vm_usecase9, mock_hibernation_usecase9, mock_latest_certs_usecase9], ] 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]) + enable_uefi_cert_update_row, health_store_id_row, operation_row, reboot_setting_row, mock_should_reboot_row, expected_error_message_row, mock_detect_confidential_vm_row, mock_hibernation_row, mock_latest_certs_row = row + + runtime = self._create_update_certs_runtime(enable_uefi_cert_update=bool(enable_uefi_cert_update_row), health_store_id=health_store_id_row, operation=operation_row, reboot_setting=reboot_setting_row) + + backup_should_reboot_before_cert_update = runtime.package_manager.should_reboot_before_cert_update + backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm + backup_hibernation_check = runtime.package_manager.is_hibernation_enabled_for_cert_update + backup_latest_certs_check = runtime.package_manager.are_latest_certs_present + backup_update_certs = runtime.patch_installer.package_manager.update_certs + if mock_should_reboot_row is not None: + runtime.package_manager.should_reboot_before_cert_update = mock_should_reboot_row + if mock_detect_confidential_vm_row is not None: + runtime.env_layer.detect_confidential_vm = mock_detect_confidential_vm_row + if mock_hibernation_row is not None: + runtime.package_manager.is_hibernation_enabled_for_cert_update = mock_hibernation_row + if mock_latest_certs_row is not None: + runtime.package_manager.are_latest_certs_present = mock_latest_certs_row + runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_returns_false + runtime.patch_installer.start_installation(simulate=True) - self.assertEqual(len(method_called), row[5], "Failed for row: {row}") + error_messages = self._get_installation_error_messages(runtime) + if expected_error_message_row is None: + self.assertFalse(any("Certificates may not have been updated" in message for message in error_messages), + "Did not expect certificate update failure status for row: {0}".format(row)) + else: + self.assertTrue(any(expected_error_message_row in message for message in error_messages), + "Expected certificate-related status not found for row: {0}".format(row)) + + runtime.package_manager.should_reboot_before_cert_update = backup_should_reboot_before_cert_update + runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm + runtime.package_manager.is_hibernation_enabled_for_cert_update = backup_hibernation_check + runtime.package_manager.are_latest_certs_present = backup_latest_certs_check + runtime.patch_installer.package_manager.update_certs = backup_update_certs 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") + """An exception raised by update_certs should be swallowed and not propagate, but an error should be recorded in the installation status.""" + runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting='IfRequired') backup_up_update_certs = runtime.patch_installer.package_manager.update_certs + backup_should_reboot = runtime.package_manager.should_reboot_before_cert_update + backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm + backup_hibernation = runtime.package_manager.is_hibernation_enabled_for_cert_update + backup_latest_certs = runtime.package_manager.are_latest_certs_present + runtime.package_manager.should_reboot_before_cert_update = self.mock_should_reboot_before_cert_update_returns_false + runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_returns_false + runtime.package_manager.is_hibernation_enabled_for_cert_update = self.mock_is_hibernation_enabled_for_cert_update_returns_false + runtime.package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_returns_false runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_raise_exception + runtime.patch_installer.start_installation(simulate=True) + 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_cert_error_present = any("attempting to update certificates" in error["message"] for error in error_details) + self.assertTrue(is_cert_error_present, "Expected certificate update failure error to be recorded in installation status.") + runtime.patch_installer.package_manager.update_certs = backup_up_update_certs + runtime.package_manager.should_reboot_before_cert_update = backup_should_reboot + runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm + runtime.package_manager.is_hibernation_enabled_for_cert_update = backup_hibernation + runtime.package_manager.are_latest_certs_present = backup_latest_certs + runtime.stop() + + def test_try_update_certs_adds_error_to_status_when_update_certs_returns_false(self): + """When update_certs returns False, an error should be recorded in the installation status.""" + runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting='IfRequired') + backup_should_reboot = runtime.package_manager.should_reboot_before_cert_update + backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm + 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 + + runtime.package_manager.should_reboot_before_cert_update = self.mock_should_reboot_before_cert_update_returns_false + runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_returns_false + runtime.package_manager.is_hibernation_enabled_for_cert_update = self.mock_is_hibernation_enabled_for_cert_update_returns_false + runtime.package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_returns_false + runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_returns_false + + runtime.patch_installer.start_installation(simulate=True) + + 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_cert_update_failure_present = any("Certificates may not have been updated" in error["message"] for error in error_details) + self.assertTrue(is_cert_update_failure_present, "Expected a certificate update failure error to be recorded in the installation status.") + + runtime.package_manager.should_reboot_before_cert_update = backup_should_reboot + runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm + 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_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 + backup_update_certs = runtime.patch_installer.package_manager.update_certs 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.package_manager.update_certs = self.mock_update_certs_returns_false runtime.patch_installer.start_installation(simulate=True) - self.assertEqual(len(method_called), 0) + error_messages = self._get_installation_error_messages(runtime) + self.assertFalse(any("Confidential VM" in message for message in error_messages)) + self.assertFalse(any("Certificates may not have been updated" in message for message in error_messages)) runtime.env_layer.detect_confidential_vm_by_imds = backup_detect_confidential_vm_by_imds + runtime.patch_installer.package_manager.update_certs = backup_update_certs runtime.stop() 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 + backup_update_certs = runtime.patch_installer.package_manager.update_certs 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.package_manager.update_certs = self.mock_update_certs_returns_false runtime.patch_installer.start_installation(simulate=True) - self.assertEqual(len(method_called), 0) + error_messages = self._get_installation_error_messages(runtime) + self.assertFalse(any("Unable to determine whether the VM is a Confidential VM" in message for message in error_messages)) + self.assertFalse(any("Certificates may not have been updated" in message for message in error_messages)) runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm + 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 = [ + [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: + 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_can_continue_cert_update_after_hibernation_check(self): + """Test all branches of can_continue_cert_update_after_hibernation_check().""" + + # 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 + + # 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 + + # 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 + + 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"], + ] + + for row in test_input_output_table: + mock_hibernation_check, expected_result, expected_hibernation_error_present, expected_error_substring = 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_is_hibernation_enabled_for_cert_update = runtime.package_manager.is_hibernation_enabled_for_cert_update + + 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() + + 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 From 3a99f8a665a0ab86b5b5a510835f3ed821efa4ea Mon Sep 17 00:00:00 2001 From: Rajasi Rane Date: Fri, 10 Jul 2026 16:50:15 -0700 Subject: [PATCH 2/3] Adding missing UTs and re-formatting existing UTs --- src/core/tests/Test_AptitudePackageManager.py | 572 +++++++++++------- src/core/tests/Test_PatchInstaller.py | 416 ++++++------- 2 files changed, 544 insertions(+), 444 deletions(-) diff --git a/src/core/tests/Test_AptitudePackageManager.py b/src/core/tests/Test_AptitudePackageManager.py index 82e30208..9281bba2 100644 --- a/src/core/tests/Test_AptitudePackageManager.py +++ b/src/core/tests/Test_AptitudePackageManager.py @@ -165,6 +165,54 @@ def mock_run_command_output_uptime_below_threshold(self, cmd, no_output=False, c if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: return 0, "120.12 123.45" return 0, "" + + 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""" + if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: + return 1, "Error: cannot read /proc/uptime" + return 0, "" + + 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""" + if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: + return 0, "" + 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""" + if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: + return 0, "invalid unparseable output" + return 0, "" + + def mock_run_command_output_hibernation_enabled(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate hibernation enabled state""" + if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: + return 0, "platform [suspend] disk" + return 0, "" + + def mock_run_command_output_hibernation_disabled(self, cmd, no_output=False, chk_err=True): + """Mock run_command_output to simulate hibernation disabled state""" + if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: + return 0, "platform suspend [disabled]" + return 0, "" + + 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""" + if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: + return 1, "Error: cannot read /sys/power/disk" + return 0, "" + + 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""" + if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: + return 0, "" + 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""" + if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: + return 0, None + return 0, "" # endregion Mocks # region Utility Functions @@ -1171,70 +1219,101 @@ 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_fetch_current_certs = package_manager.fetch_current_certs - 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_are_latest_certs_present_returns_false_when_mokutil_missing_and_install_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 - - fetch_calls = [] - captured_errors = [] - 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: fetch_calls.append(cert_type) - 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.are_latest_certs_present) - self.assertEqual(fetch_calls, []) - self.assertEqual(len(captured_errors), 1) - self.assertIn("Mokutil is not installed and could not be installed", captured_errors[0][0]) - self.assertEqual(captured_errors[0][1], Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) - - 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 - - def test_are_latest_certs_present_checks_certs_when_mokutil_install_succeeds(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_ensure_mokutil_available = package_manager.ensure_mokutil_available_for_cert_checks cert_output = "{0}\n{1}".format("issuer: test", Constants.LATEST_CERTIFICATE_TIMESTAMP) - 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) - self.assertTrue(package_manager.are_latest_certs_present()) + 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 + } + ] - 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 + 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') @@ -1280,6 +1359,24 @@ def test_should_reboot_before_cert_update__with_various_use_cases(self): "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 } ] @@ -1296,206 +1393,255 @@ def test_should_reboot_before_cert_update__with_various_use_cases(self): 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(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_return_true_once - - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) - self.assertEqual(self.latest_certs_check_attempts, 1) - - 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): - 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_return_true_once - - self.assertTrue(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) - self.assertEqual(self.latest_certs_check_attempts, 1) - - 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_removes_old_fwupd_before_install(self): + 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_return_true_once - 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) - self.assertEqual(self.latest_certs_check_attempts, 1) - - 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_return_true_once - 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) - self.assertEqual(self.latest_certs_check_attempts, 1) - 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') - - 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 + 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 + } + ] - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + try: + for use_case in use_cases: + self.latest_certs_check_attempts = 0 + self.get_installed_fwupd_version_check_attempts = 0 - package_manager.is_reboot_pending = backup_is_reboot_pending - package_manager.are_latest_certs_present = backup_are_latest_certs_present + 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 - 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') + package_manager.are_latest_certs_present = use_case["are_latest_certs_present"] - 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_get_installed_fwupd_version" in use_case: + package_manager._AptitudePackageManager__get_installed_fwupd_version = use_case["mock_get_installed_fwupd_version"] - self.assertFalse(package_manager.try_update_certs()) - self.assertEqual(package_manager.status_handler.is_reboot_pending, False) + if "mock_run_command_output" in use_case: + package_manager.env_layer.run_command_output = use_case["mock_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 + 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 - 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"]) + 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 + } + ] - package_manager.is_mokutil_installed = backup_is_mokutil_installed - package_manager.try_update_certs = backup_try_update_certs + 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_update_certs_returns_false_when_try_update_reports_failure(self): + 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_try_update_certs = package_manager.try_update_certs + backup_run_command_output = package_manager.env_layer.run_command_output - package_manager.try_update_certs = lambda: False - self.assertFalse(package_manager.update_certs()) + 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 + } + ] - package_manager.try_update_certs = backup_try_update_certs + 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 1d28b7c7..ae8cf821 100644 --- a/src/core/tests/Test_PatchInstaller.py +++ b/src/core/tests/Test_PatchInstaller.py @@ -793,250 +793,204 @@ 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 - reboot_setting_usecase1 = None - mock_should_reboot_usecase1 = None - expected_error_message_usecase1 = None - mock_detect_confidential_vm_usecase1 = None - # hibernation check not reached (feature flag off, returns before prereq checks) - - # 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 - reboot_setting_usecase2 = None - mock_should_reboot_usecase2 = None - expected_error_message_usecase2 = None - mock_detect_confidential_vm_usecase2 = None - # hibernation check not reached (not a default patching operation) - - # 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 - reboot_setting_usecase3 = None - mock_should_reboot_usecase3 = None - expected_error_message_usecase3 = None - mock_detect_confidential_vm_usecase3 = None - # hibernation check not reached (not a default patching operation) - - # 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 - reboot_setting_usecase4 = None - mock_should_reboot_usecase4 = None - expected_error_message_usecase4 = None - mock_detect_confidential_vm_usecase4 = None - # hibernation check not reached (not a default patching operation) - - # Use case 5: update_certs SHOULD be called only when this is default patching and all prerequisites are met. - enable_uefi_cert_update_usecase5 = True - health_store_id_usecase5 = "pub_off_sku_2025.01.01" - operation_usecase5 = Constants.INSTALLATION - reboot_setting_usecase5 = 'IfRequired' - mock_should_reboot_usecase5 = self.mock_should_reboot_before_cert_update_returns_false - expected_error_message_usecase5 = "Certificates may not have been updated" - mock_detect_confidential_vm_usecase5 = self.mock_detect_confidential_vm_returns_false - mock_hibernation_usecase5 = self.mock_is_hibernation_enabled_for_cert_update_returns_false # explicitly disable hibernation so this prereq passes - mock_latest_certs_usecase5 = self.mock_are_latest_certs_present_returns_false - - # Use case 6: update_certs should NOT be called when a reboot is required but the reboot setting is 'Never' - enable_uefi_cert_update_usecase6 = True - health_store_id_usecase6 = "pub_off_sku_2025.01.01" - operation_usecase6 = Constants.INSTALLATION - reboot_setting_usecase6 = 'Never' - mock_should_reboot_usecase6 = self.mock_should_reboot_before_cert_update_returns_true - expected_error_message_usecase6 = "reboot first" - mock_detect_confidential_vm_usecase6 = self.mock_detect_confidential_vm_returns_false - mock_hibernation_usecase6 = self.mock_is_hibernation_enabled_for_cert_update_returns_false - mock_latest_certs_usecase6 = self.mock_are_latest_certs_present_returns_false - - # Use case 7: update_certs should NOT be called when VM is a CVM. - enable_uefi_cert_update_usecase7 = True - health_store_id_usecase7 = "pub_off_sku_2025.01.01" - operation_usecase7 = Constants.INSTALLATION - reboot_setting_usecase7 = 'IfRequired' - mock_should_reboot_usecase7 = self.mock_should_reboot_before_cert_update_returns_false - expected_error_message_usecase7 = None - mock_detect_confidential_vm_usecase7 = self.mock_detect_confidential_vm_by_imds_returns_true - mock_hibernation_usecase7 = None # hibernation check not reached (CVM check fails first) - mock_latest_certs_usecase7 = None - - # Use case 8: update_certs should NOT be called when hibernation is enabled. - enable_uefi_cert_update_usecase8 = True - health_store_id_usecase8 = "pub_off_sku_2025.01.01" - operation_usecase8 = Constants.INSTALLATION - reboot_setting_usecase8 = 'IfRequired' - mock_should_reboot_usecase8 = self.mock_should_reboot_before_cert_update_returns_false - expected_error_message_usecase8 = "Turn off hibernation" - mock_detect_confidential_vm_usecase8 = self.mock_detect_confidential_vm_returns_false - mock_hibernation_usecase8 = self.mock_is_hibernation_enabled_for_cert_update_returns_true - mock_latest_certs_usecase8 = None # latest cert check not reached when hibernation is enabled - - # Use case 9: update_certs should NOT be called when latest certs are already present. - enable_uefi_cert_update_usecase9 = True - health_store_id_usecase9 = "pub_off_sku_2025.01.01" - operation_usecase9 = Constants.INSTALLATION - reboot_setting_usecase9 = 'IfRequired' - mock_should_reboot_usecase9 = self.mock_should_reboot_before_cert_update_returns_true - expected_error_message_usecase9 = None - mock_detect_confidential_vm_usecase9 = self.mock_detect_confidential_vm_returns_false - mock_hibernation_usecase9 = self.mock_is_hibernation_enabled_for_cert_update_returns_false - mock_latest_certs_usecase9 = self.mock_are_latest_certs_present_returns_true - - test_input_output_table = [ - [enable_uefi_cert_update_usecase1, health_store_id_usecase1, operation_usecase1, reboot_setting_usecase1, mock_should_reboot_usecase1, expected_error_message_usecase1, mock_detect_confidential_vm_usecase1, None, None], - [enable_uefi_cert_update_usecase2, health_store_id_usecase2, operation_usecase2, reboot_setting_usecase2, mock_should_reboot_usecase2, expected_error_message_usecase2, mock_detect_confidential_vm_usecase2, None, None], - [enable_uefi_cert_update_usecase3, health_store_id_usecase3, operation_usecase3, reboot_setting_usecase3, mock_should_reboot_usecase3, expected_error_message_usecase3, mock_detect_confidential_vm_usecase3, None, None], - [enable_uefi_cert_update_usecase4, health_store_id_usecase4, operation_usecase4, reboot_setting_usecase4, mock_should_reboot_usecase4, expected_error_message_usecase4, mock_detect_confidential_vm_usecase4, None, None], - [enable_uefi_cert_update_usecase5, health_store_id_usecase5, operation_usecase5, reboot_setting_usecase5, mock_should_reboot_usecase5, expected_error_message_usecase5, mock_detect_confidential_vm_usecase5, mock_hibernation_usecase5, mock_latest_certs_usecase5], - [enable_uefi_cert_update_usecase6, health_store_id_usecase6, operation_usecase6, reboot_setting_usecase6, mock_should_reboot_usecase6, expected_error_message_usecase6, mock_detect_confidential_vm_usecase6, mock_hibernation_usecase6, mock_latest_certs_usecase6], - [enable_uefi_cert_update_usecase7, health_store_id_usecase7, operation_usecase7, reboot_setting_usecase7, mock_should_reboot_usecase7, expected_error_message_usecase7, mock_detect_confidential_vm_usecase7, mock_hibernation_usecase7, mock_latest_certs_usecase7], - [enable_uefi_cert_update_usecase8, health_store_id_usecase8, operation_usecase8, reboot_setting_usecase8, mock_should_reboot_usecase8, expected_error_message_usecase8, mock_detect_confidential_vm_usecase8, mock_hibernation_usecase8, mock_latest_certs_usecase8], - [enable_uefi_cert_update_usecase9, health_store_id_usecase9, operation_usecase9, reboot_setting_usecase9, mock_should_reboot_usecase9, expected_error_message_usecase9, mock_detect_confidential_vm_usecase9, mock_hibernation_usecase9, mock_latest_certs_usecase9], + 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 row in test_input_output_table: - enable_uefi_cert_update_row, health_store_id_row, operation_row, reboot_setting_row, mock_should_reboot_row, expected_error_message_row, mock_detect_confidential_vm_row, mock_hibernation_row, mock_latest_certs_row = row - - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=bool(enable_uefi_cert_update_row), health_store_id=health_store_id_row, operation=operation_row, reboot_setting=reboot_setting_row) + 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_before_cert_update = runtime.package_manager.should_reboot_before_cert_update + backup_should_reboot = runtime.package_manager.should_reboot_before_cert_update backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm - backup_hibernation_check = runtime.package_manager.is_hibernation_enabled_for_cert_update - backup_latest_certs_check = runtime.package_manager.are_latest_certs_present + 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_row is not None: - runtime.package_manager.should_reboot_before_cert_update = mock_should_reboot_row - if mock_detect_confidential_vm_row is not None: - runtime.env_layer.detect_confidential_vm = mock_detect_confidential_vm_row - if mock_hibernation_row is not None: - runtime.package_manager.is_hibernation_enabled_for_cert_update = mock_hibernation_row - if mock_latest_certs_row is not None: - runtime.package_manager.are_latest_certs_present = mock_latest_certs_row - runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_returns_false + + 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) - if expected_error_message_row is None: - self.assertFalse(any("Certificates may not have been updated" in message for message in error_messages), - "Did not expect certificate update failure status for row: {0}".format(row)) - else: - self.assertTrue(any(expected_error_message_row in message for message in error_messages), - "Expected certificate-related status not found for row: {0}".format(row)) - runtime.package_manager.should_reboot_before_cert_update = backup_should_reboot_before_cert_update + 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.package_manager.is_hibernation_enabled_for_cert_update = backup_hibernation_check - runtime.package_manager.are_latest_certs_present = backup_latest_certs_check + 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_try_update_certs_swallows_exception_from_update_certs(self): - """An exception raised by update_certs should be swallowed and not propagate, but an error should be recorded in the installation status.""" - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting='IfRequired') - backup_up_update_certs = runtime.patch_installer.package_manager.update_certs - backup_should_reboot = runtime.package_manager.should_reboot_before_cert_update - backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm - backup_hibernation = runtime.package_manager.is_hibernation_enabled_for_cert_update - backup_latest_certs = runtime.package_manager.are_latest_certs_present - - runtime.package_manager.should_reboot_before_cert_update = self.mock_should_reboot_before_cert_update_returns_false - runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_returns_false - runtime.package_manager.is_hibernation_enabled_for_cert_update = self.mock_is_hibernation_enabled_for_cert_update_returns_false - runtime.package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_returns_false - runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_raise_exception - - runtime.patch_installer.start_installation(simulate=True) - - 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_cert_error_present = any("attempting to update certificates" in error["message"] for error in error_details) - self.assertTrue(is_cert_error_present, "Expected certificate update failure error to be recorded in installation status.") - - runtime.patch_installer.package_manager.update_certs = backup_up_update_certs - runtime.package_manager.should_reboot_before_cert_update = backup_should_reboot - runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm - runtime.package_manager.is_hibernation_enabled_for_cert_update = backup_hibernation - runtime.package_manager.are_latest_certs_present = backup_latest_certs - runtime.stop() - - def test_try_update_certs_adds_error_to_status_when_update_certs_returns_false(self): - """When update_certs returns False, an error should be recorded in the installation status.""" - runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01", reboot_setting='IfRequired') - backup_should_reboot = runtime.package_manager.should_reboot_before_cert_update - backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm - 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 - - runtime.package_manager.should_reboot_before_cert_update = self.mock_should_reboot_before_cert_update_returns_false - runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_returns_false - runtime.package_manager.is_hibernation_enabled_for_cert_update = self.mock_is_hibernation_enabled_for_cert_update_returns_false - runtime.package_manager.are_latest_certs_present = self.mock_are_latest_certs_present_returns_false - runtime.patch_installer.package_manager.update_certs = self.mock_update_certs_returns_false - - runtime.patch_installer.start_installation(simulate=True) - - 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_cert_update_failure_present = any("Certificates may not have been updated" in error["message"] for error in error_details) - self.assertTrue(is_cert_update_failure_present, "Expected a certificate update failure error to be recorded in the installation status.") - - runtime.package_manager.should_reboot_before_cert_update = backup_should_reboot - runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm - 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_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 - backup_update_certs = runtime.patch_installer.package_manager.update_certs - - runtime.env_layer.detect_confidential_vm_by_imds = self.mock_detect_confidential_vm_by_imds_returns_true - 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) - self.assertFalse(any("Confidential VM" in message for message in error_messages)) - self.assertFalse(any("Certificates may not have been updated" in message for message in error_messages)) - - runtime.env_layer.detect_confidential_vm_by_imds = backup_detect_confidential_vm_by_imds - runtime.patch_installer.package_manager.update_certs = backup_update_certs - runtime.stop() - - 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 - backup_update_certs = runtime.patch_installer.package_manager.update_certs - - runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_raises_exception - 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) - self.assertFalse(any("Unable to determine whether the VM is a Confidential VM" in message for message in error_messages)) - self.assertFalse(any("Certificates may not have been updated" in message for message in error_messages)) - - runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm - 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() """ From c7c8b4702c5e658cb099c1a1e4d913499bb3293a Mon Sep 17 00:00:00 2001 From: Rajasi Rane Date: Fri, 10 Jul 2026 18:14:02 -0700 Subject: [PATCH 3/3] Increasing code coverage --- src/core/tests/Test_AptitudePackageManager.py | 54 ++++++++----------- 1 file changed, 21 insertions(+), 33 deletions(-) diff --git a/src/core/tests/Test_AptitudePackageManager.py b/src/core/tests/Test_AptitudePackageManager.py index 9281bba2..f6bd4c44 100644 --- a/src/core/tests/Test_AptitudePackageManager.py +++ b/src/core/tests/Test_AptitudePackageManager.py @@ -112,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 @@ -139,13 +138,12 @@ 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, "" @@ -157,62 +155,52 @@ 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): - if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: - return 0, "700000.12 123.45" - return 0, "" + 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): - if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: - return 0, "120.12 123.45" - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: - return 1, "Error: cannot read /proc/uptime" - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_uptime_seconds_cmd) > -1: - return 0, "invalid unparseable output" - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: - return 0, "platform [suspend] disk" - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: - return 0, "platform suspend [disabled]" - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: - return 1, "Error: cannot read /sys/power/disk" - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: - return 0, "" + 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""" - if cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1: - return 0, None - return 0, "" + self.assertTrue(cmd.find(self.runtime.package_manager.get_hibernation_state_cmd) > -1) + return 0, None # endregion Mocks # region Utility Functions