diff --git a/src/core/src/bootstrap/ConfigurationFactory.py b/src/core/src/bootstrap/ConfigurationFactory.py index dd92491d..e0731a8b 100644 --- a/src/core/src/bootstrap/ConfigurationFactory.py +++ b/src/core/src/bootstrap/ConfigurationFactory.py @@ -38,6 +38,7 @@ from core.src.package_managers.AptitudePackageManager import AptitudePackageManager from core.src.package_managers.AzL3TdnfPackageManager import AzL3TdnfPackageManager +from core.src.package_managers.Dnf4PackageManager import Dnf4PackageManager from core.src.package_managers.Dnf5PackageManager import Dnf5PackageManager from core.src.package_managers.YumPackageManager import YumPackageManager from core.src.package_managers.ZypperPackageManager import ZypperPackageManager @@ -71,20 +72,23 @@ def __init__(self, log_file_path, events_folder, telemetry_supported): self.configurations = { 'apt_prod_config': self.new_prod_configuration(Constants.APT, AptitudePackageManager), - 'dnf5_prod_config': self.new_prod_configuration(Constants.DNF5, Dnf5PackageManager), - 'tdnf_prod_config': self.new_prod_configuration(Constants.TDNF, AzL3TdnfPackageManager), + 'dnf4_prod_config': self.new_prod_configuration(Constants.DNF4, Dnf4PackageManager), + 'dnf5_prod_config': self.new_prod_configuration(Constants.DNF5, Dnf5PackageManager), + 'tdnf_prod_config': self.new_prod_configuration(Constants.TDNF, AzL3TdnfPackageManager), 'yum_prod_config': self.new_prod_configuration(Constants.YUM, YumPackageManager), 'zypper_prod_config': self.new_prod_configuration(Constants.ZYPPER, ZypperPackageManager), 'apt_dev_config': self.new_dev_configuration(Constants.APT, AptitudePackageManager), - 'dnf5_dev_config': self.new_dev_configuration(Constants.DNF5, Dnf5PackageManager), - 'tdnf_dev_config': self.new_dev_configuration(Constants.TDNF, AzL3TdnfPackageManager), + 'dnf4_dev_config': self.new_dev_configuration(Constants.DNF4, Dnf4PackageManager), + 'dnf5_dev_config': self.new_dev_configuration(Constants.DNF5, Dnf5PackageManager), + 'tdnf_dev_config': self.new_dev_configuration(Constants.TDNF, AzL3TdnfPackageManager), 'yum_dev_config': self.new_dev_configuration(Constants.YUM, YumPackageManager), 'zypper_dev_config': self.new_dev_configuration(Constants.ZYPPER, ZypperPackageManager), 'apt_test_config': self.new_test_configuration(Constants.APT, AptitudePackageManager), - 'dnf5_test_config': self.new_test_configuration(Constants.DNF5, Dnf5PackageManager), - 'tdnf_test_config': self.new_test_configuration(Constants.TDNF, AzL3TdnfPackageManager), + 'dnf4_test_config': self.new_test_configuration(Constants.DNF4, Dnf4PackageManager), + 'dnf5_test_config': self.new_test_configuration(Constants.DNF5, Dnf5PackageManager), + 'tdnf_test_config': self.new_test_configuration(Constants.TDNF, AzL3TdnfPackageManager), 'yum_test_config': self.new_test_configuration(Constants.YUM, YumPackageManager), 'zypper_test_config': self.new_test_configuration(Constants.ZYPPER, ZypperPackageManager) } @@ -120,7 +124,7 @@ def get_configuration(self, env, package_manager_name): print ("Error: Environment configuration not supported - " + str(env)) return None - if str(package_manager_name) not in [Constants.APT, Constants.DNF5, Constants.TDNF, Constants.YUM, Constants.ZYPPER]: + if str(package_manager_name) not in [Constants.APT, Constants.DNF4, Constants.DNF5, Constants.TDNF, Constants.YUM, Constants.ZYPPER]: print ("Error: Package manager configuration not supported - " + str(package_manager_name)) return None diff --git a/src/core/src/bootstrap/Constants.py b/src/core/src/bootstrap/Constants.py index 0946b2b5..873007dd 100644 --- a/src/core/src/bootstrap/Constants.py +++ b/src/core/src/bootstrap/Constants.py @@ -212,6 +212,7 @@ class StatusTruncationConfig(EnumBackport): # Package Managers APT = 'apt' + DNF4 = 'dnf4' DNF5 = 'dnf5' TDNF = 'tdnf' YUM = 'yum' diff --git a/src/core/src/bootstrap/EnvLayer.py b/src/core/src/bootstrap/EnvLayer.py index 4c01ea12..e4ba477f 100644 --- a/src/core/src/bootstrap/EnvLayer.py +++ b/src/core/src/bootstrap/EnvLayer.py @@ -81,6 +81,21 @@ def is_distro_rhel_10(self, distro_name): """ Checks if the current distro is RHEL 10 """ return self.__is_matching_distro_and_version(distro_name, Constants.RED_HAT, version_to_match=10) + def __is_dnf_available(self): + code, _ = self.run_command_output('which dnf', False, False) + return code == 0 + + def __get_dnf_version(self): + code, out = self.run_command_output('dnf --version', False, False) + # Output : dnf5 version 5.2.18.0 + # Output : 4.20.0 + if code != 0 or not out: + return code, out, None + + first_line = str(out).splitlines()[0].strip() + version = first_line.split()[-1] + return code, out, version + def get_package_manager(self): # type: () -> str """ Detects package manager type """ @@ -91,9 +106,20 @@ def get_package_manager(self): # Example: ['Azure Linux', '4.0', ''] os_name, os_version, os_code = self.platform.linux_distribution() - # Check for unsupported distros + # Check for Rhel 10 ( uses dnf4) if self.is_distro_rhel_10(os_name): - error_msg = "This distro is not yet supported in your region. Please review https://aka.ms/VMGuestPatchingCompatibility for more information. [Distro={0}][Version={1}][Code={2}]".format(str(os_name), os_version, os_code) + if not self.__is_dnf_available(): + error_msg = "Expected package manager dnf not found on this rhel 10 VM" + print("Error: {0}".format(error_msg)) + return str() + code, out, version = self.__get_dnf_version() + if version: + if version.startswith('4'): + return Constants.DNF4 + error_msg = "Expected dnf version 4 on this rhel 10 VM. Found: {0}".format(version) + print("Error: {0}".format(error_msg)) + return str() + error_msg = "Unable to determine dnf version. Code={0}, Output={1}".format(code, out) print("Error: {0}".format(error_msg)) return str() diff --git a/src/core/src/package_managers/Dnf4PackageManager.py b/src/core/src/package_managers/Dnf4PackageManager.py new file mode 100644 index 00000000..964da673 --- /dev/null +++ b/src/core/src/package_managers/Dnf4PackageManager.py @@ -0,0 +1,714 @@ +# Copyright 2026 Microsoft Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# Requires Python 2.7+ + +"""Dnf4PackageManager for Rhel 10""" +import json +import re + +from core.src.core_logic.VersionComparator import VersionComparator +from core.src.bootstrap.Constants import Constants +from core.src.package_managers.PackageManager import PackageManager + + +class Dnf4PackageManager(PackageManager): + """Implementation of dnf4 package management operations""" + + def __init__(self, env_layer, execution_config, composite_logger, telemetry_writer, status_handler): + super(Dnf4PackageManager, self).__init__(env_layer, execution_config, composite_logger, telemetry_writer, status_handler) + # Repo refresh + self.cmd_clean_cache = "sudo dnf4 -q clean expire-cache" + self.cmd_repo_refresh = self.cmd_get_all_updates = "sudo dnf4 -q check-update" + + # Support to get updates and their dependencies + self.single_package_check_versions = 'sudo dnf4 list --available ' + self.single_package_check_installed = 'sudo dnf4 list --installed ' + self.single_package_upgrade_simulation_cmd = 'sudo dnf4 install --assumeno --skip-broken ' + + # Install update + self.single_package_upgrade_cmd = 'sudo dnf4 -y install ' + + # Support to check for processes requiring restart + self.needs_restarting_with_flag = 'sudo LANG=en_US.UTF8 needs-restarting -r' + + # Package manager exit code(s) + self.dnf4_exitcode_ok = [0, 100] + # DNF4 valid exit codes for simulation commands + self.dnf4_simulation_valid_exit_codes = [0, 1] + self.dnf4_not_installed_exit_code = 1 + + # Package manager success/failure text + self.dnf4_not_installed_text = "No matching packages to list" + self.dnf4_dependency_success_text = ["Installing dependencies:", "Upgrading:", "Dependencies resolved."] + self.dnf4_dependency_exit_text = "Transaction Summary" + self.dnf4_dependency_failure_text = "Skipping packages with broken dependencies" + self.dnf4_subscription_failure_texts = ["Unable to read consumer identity", "This system is not registered with an entitlement server"] + self.dnf4_list_installed_command_patterns = "list --installed" + + # Auto OS updates + self.current_auto_os_update_service = None + self.os_patch_configuration_settings_file_path = '' + self.auto_update_service_enabled = False + self.auto_update_config_pattern_match_text = "" + self.download_updates_identifier_text = "" + self.apply_updates_identifier_text = "" + self.enable_on_reboot_identifier_text = "" + self.enable_on_reboot_check_cmd = '' + self.enable_on_reboot_cmd = '' + self.installation_state_identifier_text = "" + self.install_check_cmd = "" + self.apply_updates_enabled = "Enabled" + self.apply_updates_disabled = "Disabled" + self.apply_updates_unknown = "Unknown" + + # commands for DNF Automatic updates service + self.__init_constants_for_dnf4_automatic() + + # Miscellaneous + self.set_package_manager_setting(Constants.PKG_MGR_SETTING_IDENTITY, Constants.DNF4) + self.STR_TOTAL_DOWNLOAD_SIZE = "Total download size: " + self.version_comparator = VersionComparator() + + self.package_install_expected_avg_time_in_seconds = 90 # Setting a default value of 90 seconds as the avg time to install a package using dnf4. + + def refresh_repo(self): + self.composite_logger.log("[DNF4] Refreshing local repo...") + self.invoke_package_manager(self.cmd_clean_cache) + self.invoke_package_manager(self.cmd_repo_refresh) + + # region Get Available Updates + def invoke_package_manager_advanced(self, command, raise_on_exception=True): + """Get missing updates using the command input""" + self.composite_logger.log_verbose("[DNF4] Invoking package manager. [Command={0}]".format(str(command))) + code, out = self.env_layer.run_command_output(command, False, False) + self.validate_dnf4_output(out) + is_valid_not_installed = (self.dnf4_list_installed_command_patterns in command and code == self.dnf4_not_installed_exit_code and self.dnf4_not_installed_text in (out or "")) + + if code in self.dnf4_exitcode_ok or is_valid_not_installed: + self.composite_logger.log_debug('[DNF4] Invoked package manager. [Command={0}][Code={1}][Output={2}]'.format(command, str(code), str(out))) + else: + self.composite_logger.log_warning('[ERROR] Customer environment error. [Command={0}][Code={1}][Output={2}]'.format(command, str(code), str(out))) + error_msg = "Customer environment error: Investigate and resolve unexpected return code ({0}) from package manager on command: {1}".format(str(code), command) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.PACKAGE_MANAGER_FAILURE) + if raise_on_exception: + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + + return out, code + + def validate_dnf4_output(self, output): + for failure_text in self.dnf4_subscription_failure_texts: + if failure_text in output: + self.composite_logger.log_error("[DNF4] Subscription/entitlement failure detected. [{0}]".format(failure_text)) + raise Exception("System is not properly registered with subscription service.") + + # region Classification-based (incl. All) update check + def get_all_updates(self, cached=False): + """Get all missing updates""" + self.composite_logger.log_verbose("[DNF4] Discovering all packages...") + if cached and not len(self.all_updates_cached) == 0: + self.composite_logger.log_debug("[DNF4] Get all updates : [Cached={0}][PackagesCount={1}]]".format(str(cached), len(self.all_updates_cached))) + return self.all_updates_cached, self.all_update_versions_cached # allows for high performance reuse in areas of the code explicitly aware of the cache + + out = self.invoke_package_manager(self.cmd_get_all_updates) + self.all_updates_cached, self.all_update_versions_cached = self.extract_packages_and_versions(out) + self.composite_logger.log_debug("[DNF4] Get all updates : [Cached={0}][PackagesCount={1}]]".format(str(False), len(self.all_updates_cached))) + return self.all_updates_cached, self.all_update_versions_cached + + def get_security_updates(self): + """Get missing security updates. NOTE: Classification based categorization of patches is not available in Rhel 10 as of now""" + self.composite_logger.log_verbose("[DNF4] Discovering all packages as 'security' packages, since DNF4 does not support package classification...") + security_packages, security_package_versions = self.get_all_updates(cached=False) + self.composite_logger.log_debug("[DNF4] Discovered 'security' packages. [Count={0}]".format(len(security_packages))) + return security_packages, security_package_versions + + def get_other_updates(self): + """Get missing other updates.""" + self.composite_logger.log_verbose("[DNF4] Discovering 'other' packages...") + return [], [] + + def set_max_patch_publish_date(self, max_patch_publish_date=str()): + pass + # endregion + + # region Output Parser(s) + def extract_packages_and_versions(self, output): + """Returns packages and versions from given output""" + packages, versions = self.extract_packages_and_versions_including_duplicates(output) + packages, versions = self.dedupe_update_packages_to_get_latest_versions(packages, versions) + return packages, versions + + def extract_packages_and_versions_including_duplicates(self, output): + """Returns packages and versions from given output""" + # DNF check-update returns 3-column format: package.arch version repo + # Sample output : python3-rpm.x86_64 4.19.1.1-23.el10 rhel-10-baseos-rhui-rpms + self.composite_logger.log_verbose("[DNF4] Extracting package and version data...") + packages, versions = [], [] + lines = output.strip().split('\n') + for line_index in range(0, len(lines)): + line = lines[line_index].strip() + # Do not install Obsoleting Packages. The obsoleting packages list comes towards end in the output. + if line.startswith("Obsoleting"): + break + + filtered_line = re.split(r'\s+', lines[line_index].strip()) + if len(filtered_line) == 3 and self.__is_package(filtered_line[0]): + packages.append(self.get_product_name(filtered_line[0])) + versions.append(filtered_line[1]) + else: + self.composite_logger.log_verbose("[DNF4] > Inapplicable line ({0}): {1}".format(line_index, lines[line_index])) + + return packages, versions + + def dedupe_update_packages_to_get_latest_versions(self, packages, package_versions): + """Remove duplicate packages and returns the latest/highest version of each package""" + deduped_packages = [] + deduped_package_versions = [] + + for index, package in enumerate(packages): + if package in deduped_packages: + deduped_package_version = deduped_package_versions[deduped_packages.index(package)] + duplicate_package_version = package_versions[index] + # use custom comparator output 0 (equal), -1 (deduped package version is the lower one), +1 (deduped package version is the greater one) + is_deduped_package_latest = self.version_comparator.compare_versions(deduped_package_version, + duplicate_package_version) + if is_deduped_package_latest < 0: + deduped_package_versions[deduped_packages.index(package)] = duplicate_package_version + continue + + deduped_packages.append(package) + deduped_package_versions.append(package_versions[index]) + + return deduped_packages, deduped_package_versions + + @staticmethod + def __is_package(chunk): + # Using a list comprehension to determine if chunk is a package + return any(chunk.endswith(ext) for ext in Constants.SUPPORTED_PACKAGE_ARCH) + # endregion + + # region Install Update + def get_composite_package_identifier(self, package, package_version): + package_without_arch, arch = self.get_product_name_and_arch(package) + package_identifier = package_without_arch + '-' + str(package_version) + if arch is not None: + package_identifier += arch + return package_identifier + + def install_updates_fail_safe(self, excluded_packages): + return + + def install_security_updates_azgps_coordinated(self): + """This is not applicable for dnf4 yet. DNF4 will not have this method implemented""" + pass + + def try_meet_azgps_coordinated_requirements(self): + """This is not applicable for dnf4 yet. DNF4 will not have this method implemented""" + pass + # endregion + + # region Package Information + def get_all_available_versions_of_package(self, package_name): + """Returns a list of all the available versions of a package""" + # Sample output format ( plugin : kernel) + # Updating Subscription Management repositories. + # Last metadata expiration check: 0:46:06 ago on Tue Jun 30 19:39:22 2026. + # Available Packages + # kernel.x86_64 6.12.0-211.28.1.el10_2 rhel-10-baseos-rhui-rpms + cmd = self.single_package_check_versions.replace('', package_name) + output = self.invoke_package_manager(cmd) + packages, package_versions = self.extract_packages_and_versions_including_duplicates(output) + return package_versions + + def is_package_version_installed(self, package_name, package_version): + """Returns true if the specific package version is installed""" + # Sample output for the command `dnf4 list --installed kernel`: + # Updating Subscription Management repositories. + # Installed Packages + # kernel.x86_64 6.12.0-211.22.1.el10_2 @System + self.composite_logger.log_verbose("[DNF4] Checking package install status. [PackageName={0}][PackageVersion={1}]".format(str(package_name), str(package_version))) + cmd = self.single_package_check_installed.replace('', package_name) + output = self.invoke_package_manager(cmd) + packages, package_versions = self.extract_packages_and_versions_including_duplicates(output) + + for index, package in enumerate(packages): + if package == package_name and (package_versions[index] == package_version): + self.composite_logger.log_debug("[DNF4] Installed version match found. [PackageName={0}][PackageVersion={1}]".format(str(package_name), str(package_version))) + return True + else: + self.composite_logger.log_verbose("[DNF4] Did not match: " + package + " (" + package_versions[index] + ")") + + # If no matching package name and version are found in the package manager output, the requested version is not installed (it may have been replaced, upgraded, or removed) + self.composite_logger.log_debug("[DNF4] Installed version match NOT found. [PackageName={0}][PackageVersion={1}]".format(str(package_name), str(package_version))) + return False + + def extract_dependencies(self, output, packages): + # Extracts dependent packages from output. Refer dnf4_output_expected_format.txt for examples of output formats. + dependencies = [] + + # Handle non-blocking dependency failure / nothing-to-do cases + if self.dnf4_dependency_failure_text in output: + self.composite_logger.log_warning("[DNF4] Packages skipped due to broken dependencies (non-blocking)") + return dependencies + + package_arch_to_look_for = ["x86_64", "noarch", "i686", "aarch64"] + + lines = output.strip().splitlines() + in_dependency_section = False + + for line_index in range(0, len(lines)): + line_str = lines[line_index].strip() + # Detect start of dependency section + if any(line_str.startswith(text) for text in self.dnf4_dependency_success_text): + in_dependency_section = True + + # Detect exit of dependency section + if in_dependency_section and line_str.startswith(self.dnf4_dependency_exit_text): + self.composite_logger.log_verbose("[DNF4] Exiting dependency section. Remaining output lines are skipped.") + break + + line = re.split(r'\s+', line_str) + dependent_package_name = "" + + if self.is_valid_update(line, package_arch_to_look_for): + dependent_package_name = self.get_product_name_with_arch(line, package_arch_to_look_for) + else: + self.composite_logger.log_verbose("[DNF4] Inapplicable line: " + str(line)) + continue + + # Remove input packages (support both pkg and pkg.arch) + if len(dependent_package_name) != 0 and dependent_package_name not in packages and dependent_package_name not in dependencies: + self.composite_logger.log_verbose("[DNF4] Dependency detected: " + dependent_package_name) + dependencies.append(dependent_package_name) + + return dependencies + + def add_arch_dependencies(self, package_manager, package, version, packages, package_versions, package_and_dependencies, package_and_dependency_versions): + """Not needed since it already supports multi-architecture. Refer to PR description for details.""" + pass + + def is_valid_update(self, package_details_in_output, package_arch_to_look_for): + # Verifies whether the line under consideration (i.e. package_details_in_output) contains relevant package details. + # package_details_in_output will be of the following format if it is valid + # Sample package details in DNF4: + # perl-lib x86_64 0.65-512.2.el10_0 rhel-10-appstream-rhui-rpms 16 k + return len(package_details_in_output) == 6 and self.is_arch_in_package_details(package_details_in_output[1], package_arch_to_look_for) + + @staticmethod + def is_arch_in_package_details(package_detail, package_arch_to_look_for): + # Using a list comprehension to determine if chunk is a package + return len([p for p in package_arch_to_look_for if p in package_detail]) == 1 + + def get_dependent_list(self, packages): + """Returns dependent List for the list of packages""" + package_names = " ".join(packages) + cmd = self.single_package_upgrade_simulation_cmd + package_names + code, output = self.env_layer.run_command_output(cmd, False, False) + self.composite_logger.log_verbose("[DNF4] Dependency simulation. [Command={0}][Code={1}]".format(cmd, str(code))) + if code not in self.dnf4_simulation_valid_exit_codes: + self.composite_logger.log_error("[DNF4] Unexpected failure during dependency simulation. [Command={0}][Code={1}][Output={2}]".format(cmd, str(code), output)) + error_msg = "DNF4 dependency simulation failed. Investigate and resolve unexpected return code({0}) from package manager on command: {1} ".format(str(code), cmd) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.PACKAGE_MANAGER_FAILURE) + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + + dependencies = self.extract_dependencies(output, packages) + self.composite_logger.log_verbose("[DNF4] Resolved dependencies. [Command={0}][Packages={1}][DependencyCount={2}]".format(str(cmd), str(packages), len(dependencies))) + return dependencies + + def get_product_name(self, package_name): + """Retrieve package name""" + return package_name + + def get_product_name_and_arch(self, package_name): + """Splits out product name and architecture - if this is changed, modify in PackageFilter also""" + architectures = Constants.SUPPORTED_PACKAGE_ARCH + for arch in architectures: + if package_name.endswith(arch): + return package_name[:-len(arch)], arch + return package_name, None + + def get_product_name_with_arch(self, package_detail, package_arch_to_look_for): + """Retrieve product name with arch separated by '.'. Note: This format is default in dnf4. Refer samples noted within func extract_dependencies() for more clarity""" + return package_detail[0] + "." + package_detail[1] if package_detail[1] in package_arch_to_look_for else \ + package_detail[1] + + def get_package_size(self, output): + """Retrieve package size from installation output string""" + # Sample output line: + # Total download size: 15 M + if "Nothing to do" not in output: + lines = output.strip().split('\n') + for line in lines: + if line.find(self.STR_TOTAL_DOWNLOAD_SIZE) >= 0: + return line.replace(self.STR_TOTAL_DOWNLOAD_SIZE, "") + return Constants.UNKNOWN_PACKAGE_SIZE + # endregion + + # region auto OS updates + def __init_constants_for_dnf4_automatic(self): + self.dnf4_automatic_configuration_file_path = '/etc/dnf/automatic.conf' + self.dnf4_automatic_install_check_cmd = ' rpm -qa | grep dnf-automatic' + self.dnf4_automatic_enable_on_reboot_check_cmd = 'systemctl is-enabled dnf-automatic.timer' + self.dnf4_automatic_disable_on_reboot_cmd = 'systemctl disable --now dnf-automatic.timer' + self.dnf4_automatic_enable_on_reboot_cmd = 'systemctl enable --now dnf-automatic.timer' + self.dnf4_automatic_config_pattern_match_text = ' = (no|yes)' + self.dnf4_automatic_download_updates_identifier_text = 'download_updates' + self.dnf4_automatic_apply_updates_identifier_text = 'apply_updates' + self.dnf4_automatic_enable_on_reboot_identifier_text = "enable_on_reboot" + self.dnf4_automatic_installation_state_identifier_text = "installation_state" + self.dnf4_auto_os_update_service = "dnf-automatic" + + def get_current_auto_os_patch_state(self): + """Gets the current auto OS update patch state on the machine""" + self.composite_logger.log("[DNF4] Fetching the current automatic OS patch state on the machine...") + current_auto_os_patch_state_for_dnf_automatic = self.__get_current_auto_os_patch_state_for_dnf_automatic() + self.composite_logger.log("[DNF4] OS patch state per auto OS update service: [dnf-automatic={0}]".format(str(current_auto_os_patch_state_for_dnf_automatic))) + + if current_auto_os_patch_state_for_dnf_automatic == Constants.AutomaticOSPatchStates.ENABLED: + current_auto_os_patch_state = Constants.AutomaticOSPatchStates.ENABLED + elif current_auto_os_patch_state_for_dnf_automatic == Constants.AutomaticOSPatchStates.DISABLED: + current_auto_os_patch_state = Constants.AutomaticOSPatchStates.DISABLED + else: + current_auto_os_patch_state = Constants.AutomaticOSPatchStates.UNKNOWN + + self.composite_logger.log_debug("[DNF4] Overall Auto OS Patch State based on all auto OS update service states [OverallAutoOSPatchState={0}]".format(str(current_auto_os_patch_state))) + return current_auto_os_patch_state + + def __get_current_auto_os_patch_state_for_dnf_automatic(self): + """Gets current auto OS update patch state for dnf-automatic""" + self.composite_logger.log_debug("[DNF4] Fetching current automatic OS patch state in dnf-automatic service. This includes checks on whether the service is installed, current auto patch enable state and whether it is set to enable on reboot") + self.__init_auto_update_for_dnf_automatic() + is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() + + apply_updates = self.__get_extension_standard_value_for_apply_updates(apply_updates_value) + + if apply_updates == self.apply_updates_enabled or enable_on_reboot_value: + return Constants.AutomaticOSPatchStates.ENABLED + # OS patch state is considered to be disabled: a) if it was successfully disabled or b) if the service is not installed + elif not is_service_installed or (apply_updates == self.apply_updates_disabled and not enable_on_reboot_value): + return Constants.AutomaticOSPatchStates.DISABLED + else: + return Constants.AutomaticOSPatchStates.UNKNOWN + + def __init_auto_update_for_dnf_automatic(self): + """Initializes all generic auto OS update variables with the config values for dnf automatic service""" + self.os_patch_configuration_settings_file_path = self.dnf4_automatic_configuration_file_path + self.download_updates_identifier_text = self.dnf4_automatic_download_updates_identifier_text + self.apply_updates_identifier_text = self.dnf4_automatic_apply_updates_identifier_text + self.enable_on_reboot_identifier_text = self.dnf4_automatic_enable_on_reboot_identifier_text + self.installation_state_identifier_text = self.dnf4_automatic_installation_state_identifier_text + self.auto_update_config_pattern_match_text = self.dnf4_automatic_config_pattern_match_text + self.enable_on_reboot_check_cmd = self.dnf4_automatic_enable_on_reboot_check_cmd + self.enable_on_reboot_cmd = self.dnf4_automatic_enable_on_reboot_cmd + self.install_check_cmd = self.dnf4_automatic_install_check_cmd + self.current_auto_os_update_service = self.dnf4_auto_os_update_service + + def __get_current_auto_os_updates_setting_on_machine(self): + """Gets all the update settings related to auto OS updates currently set on the machine""" + try: + download_updates_value = "" + apply_updates_value = "" + is_service_installed = False + enable_on_reboot_value = False + + # get install state + if not self.is_auto_update_service_installed(self.install_check_cmd): + return is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value + + is_service_installed = True + enable_on_reboot_value = self.is_service_set_to_enable_on_reboot(self.enable_on_reboot_check_cmd) + + self.composite_logger.log_debug("[DNF4] Checking if auto updates are currently enabled...") + image_default_patch_configuration = self.env_layer.file_system.read_with_retry(self.os_patch_configuration_settings_file_path, raise_if_not_found=False) + if image_default_patch_configuration is not None: + settings = image_default_patch_configuration.strip().split('\n') + for setting in settings: + match = re.search(self.download_updates_identifier_text + self.auto_update_config_pattern_match_text, str(setting)) + if match is not None: + download_updates_value = match.group(1) + + match = re.search(self.apply_updates_identifier_text + self.auto_update_config_pattern_match_text, str(setting)) + if match is not None: + apply_updates_value = match.group(1) + + if download_updates_value == "": + self.composite_logger.log_verbose("[DNF4] Machine did not have any value set for [Setting={0}]".format(str(self.download_updates_identifier_text))) + else: + self.composite_logger.log_verbose("[DNF4] Current value set for [{0}={1}]".format(str(self.download_updates_identifier_text) , str(download_updates_value))) + + if apply_updates_value == "": + self.composite_logger.log_verbose("[DNF4] Machine did not have any value set for [Setting={0}]".format(str(self.apply_updates_identifier_text))) + else: + self.composite_logger.log_verbose("[DNF4] Current value set for [{0}={1}]".format(str(self.apply_updates_identifier_text), str(apply_updates_value))) + return is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value + + except Exception as error: + raise Exception("[DNF4] Error occurred in fetching current auto OS update settings from the machine. [Exception={0}]".format(repr(error))) + + def is_auto_update_service_installed(self, install_check_cmd): + """Checks if the auto update service is enable_on_reboot on the VM""" + self.composite_logger.log_verbose("[DNF4] Checking if auto update service is installed. [Command={0}]".format(install_check_cmd)) + code, out = self.env_layer.run_command_output(install_check_cmd, False, False) + is_installed = len(out.strip()) > 0 and code == 0 + self.composite_logger.log_debug("[DNF4] Auto update service check completed. [Command={0}][Code={1}][Output={2}][Installed={3}]".format(install_check_cmd, str(code), out, str(is_installed))) + return is_installed + + def is_service_set_to_enable_on_reboot(self, command): + """Checking if auto update is enable_on_reboot on the machine. An enable_on_reboot service will be activated (if currently inactive) on machine reboot""" + self.composite_logger.log_verbose("[DNF4] Checking if auto update service is set to enable on reboot. [Command={0}]".format(command)) + code, out = self.env_layer.run_command_output(command, False, False) + is_enable_on_reboot = len(out.strip()) > 0 and code == 0 and out.strip().lower() == "enabled" + self.composite_logger.log_debug("[DNF4] Auto update service enable on reboot check completed. [Command={0}][Code={1}][EnabledOnReboot={2}]".format(command, str(code), str(is_enable_on_reboot))) + return is_enable_on_reboot + + def __get_extension_standard_value_for_apply_updates(self, apply_updates_value): + if apply_updates_value.lower() == 'yes': + return self.apply_updates_enabled + elif apply_updates_value.lower() == 'no': + return self.apply_updates_disabled + else: + return self.apply_updates_unknown + + def disable_auto_os_update(self): + """Disables auto OS updates on the machine only if they are enabled and logs the default settings the machine comes with""" + try: + self.composite_logger.log_verbose("[DNF4] Disabling auto OS updates in all identified services...") + self.disable_auto_os_update_for_dnf_automatic() + self.composite_logger.log_debug("[DNF4] Successfully disabled auto OS updates") + except Exception as error: + self.composite_logger.log_error("[DNF4] Could not disable auto OS updates. [Error={0}]".format(repr(error))) + raise + + def disable_auto_os_update_for_dnf_automatic(self): + """Disables auto OS updates, using dnf-automatic service, and logs the default settings the machine comes with""" + self.composite_logger.log_verbose("[DNF4] Disabling auto OS updates using dnf-automatic") + self.__init_auto_update_for_dnf_automatic() + + self.backup_image_default_patch_configuration_if_not_exists() + + if not self.is_auto_update_service_installed(self.dnf4_automatic_install_check_cmd): + self.composite_logger.log_debug("[DNF4] Cannot disable as dnf-automatic is not installed on the machine") + return + + self.composite_logger.log_verbose("[DNF4] Preemptively disabling auto OS updates using dnf-automatic") + self.update_os_patch_configuration_sub_setting(self.download_updates_identifier_text, "no", self.dnf4_automatic_config_pattern_match_text) + self.update_os_patch_configuration_sub_setting(self.apply_updates_identifier_text, "no", self.dnf4_automatic_config_pattern_match_text) + self.disable_auto_update_on_reboot(self.dnf4_automatic_disable_on_reboot_cmd) + self.composite_logger.log_debug("[DNF4] Successfully disabled auto OS updates using dnf-automatic") + + def disable_auto_update_on_reboot(self, command): + self.composite_logger.log_verbose("[DNF4] Disabling auto update on reboot. [Command={0}] ".format(command)) + code, out = self.env_layer.run_command_output(command, False, False) + + if code != 0: + self.composite_logger.log_error("[DNF4][ERROR] Error disabling auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + error_msg = 'Unexpected return code (' + str(code) + ') on command: ' + command + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.OPERATION_FAILED) + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + else: + self.composite_logger.log_debug("[DNF4] Disabled auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + + def backup_image_default_patch_configuration_if_not_exists(self): + """Records the default system settings for auto OS updates within patch extension artifacts for future reference. + We only log the default system settings a VM comes with, any subsequent updates will not be recorded""" + """ JSON format for backup file: + { + "dnf-automatic": { + "apply_updates": "yes/no/empty string", + "download_updates": "yes/no/empty string", + "enable_on_reboot": true/false, + "installation_state": true/false + } + } """ + try: + self.composite_logger.log_debug("[DNF4] Ensuring there is a backup of the default patch state for [AutoOSUpdateService={0}]".format(str(self.current_auto_os_update_service))) + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() + + # verify if existing backup is valid if not, write to backup + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + if is_backup_valid: + self.composite_logger.log_debug("[DNF4] Since extension has a valid backup, no need to log the current settings again. [Default Auto OS update settings={0}] [File path={1}]".format(str(image_default_patch_configuration_backup), self.image_default_patch_configuration_backup_path)) + else: + self.composite_logger.log_debug("[DNF4] Since the backup is invalid, will add a new backup with the current auto OS update settings") + self.composite_logger.log_debug("[DNF4] Fetching current auto OS update settings for [AutoOSUpdateService={0}]".format(str(self.current_auto_os_update_service))) + is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() + + backup_image_default_patch_configuration_json_to_add = { + self.current_auto_os_update_service: { + self.download_updates_identifier_text: download_updates_value, + self.apply_updates_identifier_text: apply_updates_value, + self.enable_on_reboot_identifier_text: enable_on_reboot_value, + self.installation_state_identifier_text: is_service_installed + } + } + image_default_patch_configuration_backup.update(backup_image_default_patch_configuration_json_to_add) + self.composite_logger.log_debug("[DNF4] Logging default system configuration settings for auto OS updates. [Settings={0}] [Log file path={1}]" + .format(str(image_default_patch_configuration_backup), self.image_default_patch_configuration_backup_path)) + self.env_layer.file_system.write_with_retry(self.image_default_patch_configuration_backup_path, '{0}'.format(json.dumps(image_default_patch_configuration_backup)), mode='w+') + except Exception as error: + error_message = "[DNF4] Exception during fetching and logging default auto update settings on the machine. [Exception={0}]".format(repr(error)) + self.composite_logger.log_error(error_message) + self.status_handler.add_error_to_status(error_message, Constants.PatchOperationErrorCodes.DEFAULT_ERROR) + raise + + def is_image_default_patch_configuration_backup_valid(self, image_default_patch_configuration_backup): + """Verifies if default auto update configurations, for a service under consideration, are saved in backup""" + return self.is_backup_valid_for_dnf_automatic(image_default_patch_configuration_backup) + + def is_backup_valid_for_dnf_automatic(self, image_default_patch_configuration_backup): + if self.dnf4_auto_os_update_service in image_default_patch_configuration_backup \ + and self.dnf4_automatic_download_updates_identifier_text in image_default_patch_configuration_backup[self.dnf4_auto_os_update_service] \ + and self.dnf4_automatic_apply_updates_identifier_text in image_default_patch_configuration_backup[self.dnf4_auto_os_update_service] \ + and self.dnf4_automatic_enable_on_reboot_identifier_text in image_default_patch_configuration_backup[self.dnf4_auto_os_update_service] \ + and self.dnf4_automatic_installation_state_identifier_text in image_default_patch_configuration_backup[self.dnf4_auto_os_update_service]: + self.composite_logger.log_debug("[DNF4] Extension has a valid backup for default dnf-automatic configuration settings") + return True + else: + self.composite_logger.log_debug("[DNF4] Extension does not have a valid backup for default dnf-automatic configuration settings") + return False + + def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_setting, value="no", config_pattern_match_text=""): + """Updates (or adds if it doesn't exist) the given patch_configuration_sub_setting with the given value in os_patch_configuration_settings_file""" + try: + # note: adding space between the patch_configuration_sub_setting and value since, we will have to do that if we have to add a patch_configuration_sub_setting that did not exist before + self.composite_logger.log_debug("[DNF4] Updating system configuration settings for auto OS updates. [Patch Configuration Sub Setting={0}] [Value={1}]".format(str(patch_configuration_sub_setting), value)) + os_patch_configuration_settings = self.env_layer.file_system.read_with_retry(self.os_patch_configuration_settings_file_path) + patch_configuration_sub_setting_to_update = patch_configuration_sub_setting + ' = ' + value + patch_configuration_sub_setting_found_in_file = False + updated_patch_configuration_sub_setting = "" + settings = os_patch_configuration_settings.strip().split('\n') + + # update value of existing setting + for i in range(len(settings)): + match = re.search(patch_configuration_sub_setting + config_pattern_match_text, settings[i]) + if match is not None: + settings[i] = patch_configuration_sub_setting_to_update + patch_configuration_sub_setting_found_in_file = True + updated_patch_configuration_sub_setting += settings[i] + "\n" + + # add setting to configuration file, since it doesn't exist + if not patch_configuration_sub_setting_found_in_file: + updated_patch_configuration_sub_setting += patch_configuration_sub_setting_to_update + "\n" + + self.env_layer.file_system.write_with_retry(self.os_patch_configuration_settings_file_path, '{0}'.format(updated_patch_configuration_sub_setting.lstrip()), mode='w+') + except Exception as error: + error_msg = "[DNF4] Error occurred while updating system configuration settings for auto OS updates. [Patch Configuration={0}] [Error={1}]".format(str(patch_configuration_sub_setting), repr(error)) + self.composite_logger.log_error(error_msg) + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR) + raise + + def revert_auto_os_update_to_system_default(self): + """Reverts the auto OS update patch state on the machine to its system default value, if one exists in our backup file""" + # type () -> None + self.composite_logger.log("[DNF4] Reverting the current automatic OS patch state on the machine to its system default value before patchmode was set to 'AutomaticByPlatform'") + self.revert_auto_os_update_to_system_default_for_dnf_automatic() + self.composite_logger.log_debug("[DNF4] Successfully reverted auto OS updates to system default config") + + def revert_auto_os_update_to_system_default_for_dnf_automatic(self): + """Reverts the auto OS update patch state on the machine to its system default value for given service, if applicable""" + # type () -> None + self.__init_auto_update_for_dnf_automatic() + self.composite_logger.log("[DNF4] Reverting the current automatic OS patch state on the machine to its system default value for [Service={0}]".format(str(self.current_auto_os_update_service))) + is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value = self.__get_current_auto_os_updates_setting_on_machine() + + if not is_service_installed: + self.composite_logger.log_debug("[DNF4] Machine default auto OS update service is not installed on the VM and hence no config to revert. [Service={0}]".format(str(self.current_auto_os_update_service))) + return + + self.composite_logger.log_verbose("[DNF4] Logging current configuration settings for auto OS updates [Service={0}][Is_Service_Installed={1}][Machine_default_update_enable_on_reboot={2}][{3}={4}]][{5}={6}]".format(str(self.current_auto_os_update_service), str(is_service_installed), str(enable_on_reboot_value), + str(self.download_updates_identifier_text), str(download_updates_value), str(self.apply_updates_identifier_text), str(apply_updates_value))) + + image_default_patch_configuration_backup = self.__get_image_default_patch_configuration_backup() + self.composite_logger.log_verbose("[DNF4] Logging system default configuration settings for auto OS updates. [Settings={0}]".format(str(image_default_patch_configuration_backup))) + is_backup_valid = self.is_image_default_patch_configuration_backup_valid(image_default_patch_configuration_backup) + + if is_backup_valid: + download_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.download_updates_identifier_text] + apply_updates_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.apply_updates_identifier_text] + enable_on_reboot_value_from_backup = image_default_patch_configuration_backup[self.current_auto_os_update_service][self.enable_on_reboot_identifier_text] + + self.update_os_patch_configuration_sub_setting(self.download_updates_identifier_text, download_updates_value_from_backup, self.auto_update_config_pattern_match_text) + self.update_os_patch_configuration_sub_setting(self.apply_updates_identifier_text, apply_updates_value_from_backup, self.auto_update_config_pattern_match_text) + if str(enable_on_reboot_value_from_backup).lower() == 'true': + self.enable_auto_update_on_reboot() + else: + self.composite_logger.log_debug("[DNF4] Backup is invalid or does not exist for current service. Unable to revert auto OS patch settings to system default value. [Service={0}]".format(str(self.current_auto_os_update_service))) + + def enable_auto_update_on_reboot(self): + """Enables machine default auto update on reboot""" + # type () -> None + command = self.enable_on_reboot_cmd + self.composite_logger.log_verbose("[DNF4] Enabling auto update on reboot. [Command={0}] ".format(command)) + code, out = self.env_layer.run_command_output(command, False, False) + + if code != 0: + self.composite_logger.log_error("[DNF4][ERROR] Error enabling auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + error_msg = 'Unexpected return code (' + str(code) + ') on command: ' + command + self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.OPERATION_FAILED) + raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS)) + else: + self.composite_logger.log_debug("[DNF4] Enabled auto update on reboot. [Command={0}][Code={1}][Output={2}]".format(command, str(code), out)) + + def __get_image_default_patch_configuration_backup(self): + """Get image_default_patch_configuration_backup file""" + image_default_patch_configuration_backup = {} + # Read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file + if self.image_default_patch_configuration_backup_exists(): + try: + image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path)) + except Exception as error: + self.composite_logger.log_error("[DNF4] Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error))) + return image_default_patch_configuration_backup + # endregion + + # region Reboot Management + def is_reboot_pending(self): + """Checks if there is a pending reboot on the machine.""" + self.composite_logger.log_verbose("[DNF4] Checking if reboot is required [Command={0}]".format(self.needs_restarting_with_flag)) + code, output = self.env_layer.run_command_output(self.needs_restarting_with_flag, False, False) + reboot_required = (code == 1) + self.composite_logger.log_debug("[DNF4] > Outcome of reboot required check. [Command={0}][Code={1}][ComputedValueOfRebootRequired={2}]".format(str(self.needs_restarting_with_flag), str(code), str(reboot_required))) + return reboot_required + + def do_processes_require_restart(self): + """Fulfilling base class contract. Not needed for DNF4""" + pass + # endregion + + def set_security_esm_package_status(self, operation, packages): + """Set the security-ESM classification for the esm packages. Only needed for apt. No-op for dnf4, yum and zypper.""" + return + + def separate_out_esm_packages(self, packages, package_versions): + """Filter out packages from the list where the version matches the UA_ESM_REQUIRED string. + Only needed for apt. No-op for dnf4, dnf5 yum and zypper""" + esm_packages = [] + esm_package_versions = [] + esm_packages_found = False + + return packages, package_versions, esm_packages, esm_package_versions, esm_packages_found + + 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 try_install_mokutil(self): + """Attempts to install mokutil""" + pass + + def try_update_certs(self): + """Attempts to update certificate status""" + pass + # endregion + diff --git a/src/core/tests/Test_CoreMain.py b/src/core/tests/Test_CoreMain.py index 86cdf3f2..4420cd26 100644 --- a/src/core/tests/Test_CoreMain.py +++ b/src/core/tests/Test_CoreMain.py @@ -54,6 +54,9 @@ def mock_linux_distribution_to_return_azure_linux(self): def mock_linux_distribution_to_return_azure_linux4(self): return ['Microsoft Azure Linux', '4.0', ''] + def mock_linux_distribution_to_return_rhel_10(self): + return ['Red Hat', '10.0', 'abc'] + def mock_os_remove(self, file_to_remove): raise Exception("File could not be deleted") @@ -1375,6 +1378,56 @@ def test_install_all_packages_for_azure_linux4_autopatching(self): LegacyEnvLayerExtensions.LegacyPlatform.linux_distribution = backup_envlayer_platform_linux_distribution + def test_install_all_packages_for_rhel_10_autopatching(self): + """Unit test for auto patching request on rhel10, should install all patches irrespective of classification""" + + backup_envlayer_platform_linux_distribution = LegacyEnvLayerExtensions.LegacyPlatform.linux_distribution + LegacyEnvLayerExtensions.LegacyPlatform.linux_distribution = self.mock_linux_distribution_to_return_rhel_10 + + argument_composer = ArgumentComposer() + classifications_to_include = ["Security", "Critical"] + argument_composer.health_store_id = str("pub_off_sku_2025.03.24") + argument_composer.classifications_to_include = classifications_to_include + argument_composer.reboot_setting = 'Always' + runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.DNF4) + runtime.set_legacy_test_type("HappyPath") + CoreMain(argument_composer.get_composed_arguments()) + + # check telemetry events + self.__check_telemetry_events(runtime) + + # check status file + 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"] + self.assertEqual(len(substatus_file_data), 4) + self.assertEqual(substatus_file_data[0]["name"], Constants.PATCH_ASSESSMENT_SUMMARY) + self.assertEqual(substatus_file_data[0]["status"].lower(), Constants.STATUS_SUCCESS.lower()) + self.assertEqual(substatus_file_data[1]["name"], Constants.PATCH_INSTALLATION_SUMMARY) + self.assertEqual(substatus_file_data[1]["status"].lower(), Constants.STATUS_SUCCESS.lower()) + self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["installedPatchCount"], 4) + self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][1]["name"], "dracut-config-generic.x86_64") + self.assertIn("Security", str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][1]["classifications"])) + self.assertEqual("Installed", json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][1]["patchInstallationState"]) + self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][2]["name"], "dracut-network.x86_64") + self.assertIn("Security", str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][2]["classifications"])) + self.assertEqual("Installed", json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][2]["patchInstallationState"]) + self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][0]["name"], "dracut.x86_64") + self.assertIn("Security", str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][0]["classifications"])) + self.assertEqual("Installed", json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][0]["patchInstallationState"]) + self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][3]["name"], "dracut-squash.x86_64") + self.assertIn("Security", str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][3]["classifications"])) + self.assertEqual("Installed", json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][3]["patchInstallationState"]) + self.assertEqual(substatus_file_data[2]["name"], Constants.PATCH_METADATA_FOR_HEALTHSTORE) + self.assertEqual(substatus_file_data[2]["status"].lower(), Constants.STATUS_SUCCESS.lower()) + substatus_file_data_patch_metadata_summary = json.loads(substatus_file_data[2]["formattedMessage"]["message"]) + self.assertEqual(substatus_file_data_patch_metadata_summary["patchVersion"], "pub_off_sku_2025.03.24") + self.assertTrue(substatus_file_data_patch_metadata_summary["shouldReportToHealthStore"]) + self.assertEqual(substatus_file_data[3]["name"], Constants.CONFIGURE_PATCHING_SUMMARY) + self.assertEqual(substatus_file_data[3]["status"].lower(), Constants.STATUS_SUCCESS.lower()) + runtime.stop() + + LegacyEnvLayerExtensions.LegacyPlatform.linux_distribution = backup_envlayer_platform_linux_distribution + def __check_telemetry_events(self, runtime): all_events = os.listdir(runtime.telemetry_writer.events_folder_path) self.assertTrue(len(all_events) > 0) diff --git a/src/core/tests/Test_Dnf4PackageManager.py b/src/core/tests/Test_Dnf4PackageManager.py new file mode 100644 index 00000000..0a8c5b96 --- /dev/null +++ b/src/core/tests/Test_Dnf4PackageManager.py @@ -0,0 +1,721 @@ +# Copyright 2026 Microsoft Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# Requires Python 2.7+ +import json +import os +import sys +import unittest + +# Conditional import for StringIO +try: + from StringIO import StringIO # Python 2 +except ImportError: + from io import StringIO # Python 3 + +from core.tests.library.ArgumentComposer import ArgumentComposer +from core.src.bootstrap.Constants import Constants +from core.tests.library.RuntimeCompositor import RuntimeCompositor + +class TestDnf4PackageManager(unittest.TestCase): + def setUp(self): + self.runtime = RuntimeCompositor(ArgumentComposer().get_composed_arguments(), True, Constants.DNF4) + self.container = self.runtime.container + + def tearDown(self): + self.runtime.stop() + + def __assert_std_io(self, captured_output, expected_output=''): + output = captured_output.getvalue() + self.assertIn(expected_output, output) + + def mock_write_with_retry_raise_exception(self, file_path_or_handle, data, mode='a+'): + raise Exception + + def mock_run_command_output_check_update(self, cmd, no_output=False, chk_err=True): + if "check-update" in cmd: + return 0, "" + return None + + def mock_run_command_output_no_reboot(self, cmd, no_output=False, chk_err=True): + if "needs-restarting" in cmd: + return 0, ("No core libraries or services have been updated since boot-up.\n" + "Reboot should not be necessary.\n") + return 0, "" + + # region Utility Functions + def __assert_reverted_automatic_patch_configuration_settings(self, package_manager, config_exists=True, config_value_expected=''): + if config_exists: + reverted_dnf4_automatic_patch_configuration_settings = self.runtime.env_layer.file_system.read_with_retry( + package_manager.dnf4_automatic_configuration_file_path) + self.assertIsNotNone(reverted_dnf4_automatic_patch_configuration_settings) + else: + self.assertFalse(os.path.exists(package_manager.dnf4_automatic_configuration_file_path)) + + @staticmethod + def __capture_std_io(): + # arrange capture std IO + captured_output = StringIO() + original_stdout = sys.stdout + sys.stdout = captured_output + return captured_output, original_stdout + + def __setup_config_and_invoke_revert_auto_os_to_system_default(self, package_manager, create_current_auto_os_config=True, create_backup_for_system_default_config=True, current_auto_os_update_config_value='', apply_updates_value="", + download_updates_value="", + enable_on_reboot_value=False, + installation_state_value=False, + set_installation_state=True): + """ Sets up current auto OS update config, backup for system default config (if requested) and invoke revert to system default """ + # setup current auto OS update config + if create_current_auto_os_config: + self.__setup_current_auto_os_update_config(package_manager, current_auto_os_update_config_value) + + # setup backup for system default auto OS update config + if create_backup_for_system_default_config: + self.__setup_backup_for_system_default_OS_update_config(package_manager, apply_updates_value=apply_updates_value, download_updates_value=download_updates_value, enable_on_reboot_value=enable_on_reboot_value, + installation_state_value=installation_state_value, set_installation_state=set_installation_state) + package_manager.revert_auto_os_update_to_system_default() + + def __setup_current_auto_os_update_config(self, package_manager, config_value='', + config_file_name="automatic.conf"): + # setup current auto OS update config + package_manager.dnf4_automatic_configuration_file_path = os.path.join(self.runtime.execution_config.config_folder, config_file_name) + self.runtime.write_to_file(package_manager.dnf4_automatic_configuration_file_path, config_value) + + def __setup_backup_for_system_default_OS_update_config(self, package_manager, apply_updates_value="", download_updates_value="", enable_on_reboot_value=False, installation_state_value=False, set_installation_state=True): + # setup backup for system default auto OS update config + package_manager.image_default_patch_configuration_backup_path = os.path.join(self.runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + backup_image_default_patch_configuration_json = { + "dnf-automatic": { + "apply_updates": apply_updates_value, + "download_updates": download_updates_value, + "enable_on_reboot": enable_on_reboot_value + } + } + if set_installation_state: + backup_image_default_patch_configuration_json["dnf-automatic"]["installation_state" ] = installation_state_value + self.runtime.write_to_file(package_manager.image_default_patch_configuration_backup_path,'{0}'.format(json.dumps(backup_image_default_patch_configuration_json))) + #endregion + + def test_refresh_repo(self): + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + package_manager.refresh_repo_safely() + # When no updates available and exit code 0 + self.runtime.env_layer.run_command_output = self.mock_run_command_output_check_update + package_manager.refresh_repo_safely() + + def test_disable_auto_os_updates_with_uninstalled_services(self): + # no services are installed on the machine. expected o/p: function will complete successfully. Backup file will be created with default values, no auto OS update configuration settings will be updated as there are none + self.runtime.set_legacy_test_type('SadPath') + package_manager = self.container.get('package_manager') + package_manager.disable_auto_os_update() + self.assertTrue(package_manager.image_default_patch_configuration_backup_exists()) + image_default_patch_configuration_backup = json.loads(self.runtime.env_layer.file_system.read_with_retry(package_manager.image_default_patch_configuration_backup_path)) + self.assertIsNotNone(image_default_patch_configuration_backup) + + # validating backup for dnf-automatic + self.assertIn(package_manager.dnf4_auto_os_update_service, image_default_patch_configuration_backup) + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_download_updates_identifier_text], "") + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_apply_updates_identifier_text], "") + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_enable_on_reboot_identifier_text], False) + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_installation_state_identifier_text], False) + + def test_disable_auto_os_updates_with_installed_services(self): + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + + package_manager.dnf4_automatic_configuration_file_path = os.path.join(self.runtime.execution_config.config_folder, "automatic.conf") + dnf4_automatic_os_patch_configuration_settings = 'apply_updates = yes\ndownload_updates = yes\n' + self.runtime.write_to_file(package_manager.dnf4_automatic_configuration_file_path, dnf4_automatic_os_patch_configuration_settings) + + package_manager.disable_auto_os_update() + self.assertTrue(package_manager.image_default_patch_configuration_backup_exists()) + image_default_patch_configuration_backup = json.loads(self.runtime.env_layer.file_system.read_with_retry(package_manager.image_default_patch_configuration_backup_path)) + self.assertIsNot(image_default_patch_configuration_backup, None) + + # validating backup for dnf-automatic + self.assertIn(package_manager.dnf4_auto_os_update_service, image_default_patch_configuration_backup) + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_download_updates_identifier_text], "yes") + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_apply_updates_identifier_text], "yes") + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_enable_on_reboot_identifier_text], False) + self.assertEqual(image_default_patch_configuration_backup[package_manager.dnf4_auto_os_update_service][package_manager.dnf4_automatic_installation_state_identifier_text], True) + + def test_disable_auto_os_update_failure(self): + package_manager = self.container.get('package_manager') + + self.assertRaises(Exception, package_manager.disable_auto_os_update) + self.assertTrue(package_manager.image_default_patch_configuration_backup_exists()) + + def test_get_current_auto_os_patch_state_with_uninstalled_services(self): + """Test get_current_auto_os_patch_state when dnf-automatic is not installed""" + self.runtime.set_legacy_test_type('SadPath') + package_manager = self.container.get('package_manager') + package_manager.get_current_auto_os_patch_state = self.runtime.backup_get_current_auto_os_patch_state + current_auto_os_patch_state = package_manager.get_current_auto_os_patch_state() + + self.assertFalse(package_manager.image_default_patch_configuration_backup_exists()) + self.assertEqual(current_auto_os_patch_state, Constants.AutomaticOSPatchStates.DISABLED) + + def test_get_current_auto_os_patch_state_with_installed_services_and_state_enabled(self): + # with enable on reboot set to false + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + package_manager.get_current_auto_os_patch_state = self.runtime.backup_get_current_auto_os_patch_state + + package_manager.dnf4_automatic_configuration_file_path = os.path.join(self.runtime.execution_config.config_folder, "automatic.conf") + dnf4_automatic_os_patch_configuration_settings = 'apply_updates = yes\ndownload_updates = yes\n' + self.runtime.write_to_file(package_manager.dnf4_automatic_configuration_file_path, dnf4_automatic_os_patch_configuration_settings) + + current_auto_os_patch_state = package_manager.get_current_auto_os_patch_state() + + self.assertFalse(package_manager.image_default_patch_configuration_backup_exists()) + self.assertEqual(current_auto_os_patch_state, Constants.AutomaticOSPatchStates.ENABLED) + + # with enable on reboot set to true + self.runtime.set_legacy_test_type('AnotherSadPath') + package_manager = self.container.get('package_manager') + package_manager.get_current_auto_os_patch_state = self.runtime.backup_get_current_auto_os_patch_state + + package_manager.dnf4_automatic_configuration_file_path = os.path.join( + self.runtime.execution_config.config_folder, "automatic.conf") + dnf4_automatic_os_patch_configuration_settings = 'apply_updates = no\ndownload_updates = yes\n' + self.runtime.write_to_file(package_manager.dnf4_automatic_configuration_file_path, + dnf4_automatic_os_patch_configuration_settings) + + current_auto_os_patch_state = package_manager.get_current_auto_os_patch_state() + + self.assertFalse(package_manager.image_default_patch_configuration_backup_exists()) + self.assertEqual(current_auto_os_patch_state, Constants.AutomaticOSPatchStates.ENABLED) + + def test_get_current_auto_os_patch_state_with_installed_services_and_state_disabled(self): + self.runtime.set_legacy_test_type('SadPath') + package_manager = self.container.get('package_manager') + + current_auto_os_patch_state = package_manager.get_current_auto_os_patch_state() + + self.assertFalse(package_manager.image_default_patch_configuration_backup_exists()) + self.assertEqual(current_auto_os_patch_state, Constants.AutomaticOSPatchStates.DISABLED) + + def test_get_current_auto_os_patch_state_disabled(self): + self.runtime.set_legacy_test_type('SadPath') + + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + + current_auto_os_patch_state = package_manager.get_current_auto_os_patch_state() + + self.assertFalse(package_manager.image_default_patch_configuration_backup_exists()) + self.assertEqual(current_auto_os_patch_state, Constants.AutomaticOSPatchStates.DISABLED) + + def test_revert_auto_os_update_to_system_default_with_service_not_installed(self): + """Test revert when dnf-automatic is not installed - should be no-op""" + self.runtime.set_legacy_test_type('SadPath') + package_manager = self.container.get('package_manager') + + # Create backup with service marked as not installed + package_manager.image_default_patch_configuration_backup_path = os.path.join( + self.runtime.execution_config.config_folder, Constants.IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH) + backup_config = { + "dnf-automatic": { + "enable_on_reboot": False, + "installation_state": False + } + } + self.runtime.write_to_file(package_manager.image_default_patch_configuration_backup_path,json.dumps(backup_config)) + + # Should complete without error even when service is not installed + package_manager.revert_auto_os_update_to_system_default() + + def test_revert_auto_os_update_to_system_default(self): + revert_success_testcase = { + "legacy_type": 'HappyPath', + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'apply_updates = no\ndownload_updates = no\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "yes", + "download_updates_value": "yes", + "enable_on_reboot_value": True, + "installation_state_value": True, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": 'apply_updates = yes\ndownload_updates = yes\n', + "config_exists": True + } + } + + revert_success_with_dnf_not_installed_testcase = { + "legacy_type": 'SadPath', + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": False, + "current_auto_os_update_config_value": '' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "", + "download_updates_value": "", + "enable_on_reboot_value": False, + "installation_state_value": False, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": "", + "config_exists": False + } + } + + revert_success_with_dnf_installed_but_no_config_value_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": False, + "expected_output": None + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'test_value = yes\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "", + "download_updates_value": "", + "enable_on_reboot_value": False, + "installation_state_value": False, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": 'download_updates =\napply_updates = \n', + "config_exists": True + } + } + + revert_success_backup_config_does_not_exist_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": True, + "expected_output": "[DNF4] Backup is invalid or does not exist for current service. Unable to revert auto OS patch settings to system default value. [Service=dnf-automatic]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'apply_updates = no\ndownload_updates = no\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": False, + "apply_updates_value": "", + "download_updates_value": "", + "enable_on_reboot_value": False, + "installation_state_value": False, + "set_installation_state": True + } + }, + "assertions": { + "config_value_expected": 'apply_updates = no\ndownload_updates = no\n', + "config_exists": True + } + } + + revert_success_default_backup_config_invalid_testcase = { + "legacy_type": 'RevertToImageDefault', + "stdio": { + "capture_output": True, + "expected_output": "[DNF4] Backup is invalid or does not exist for current service. Unable to revert auto OS patch settings to system default value. [Service=dnf-automatic]" + }, + "config": { + "current_auto_update_config": { + "create_current_auto_os_config": True, + "current_auto_os_update_config_value": 'apply_updates = no\ndownload_updates = no\n' + }, + "backup_system_default_config": { + "create_backup_for_system_default_config": True, + "apply_updates_value": "yes", + "download_updates_value": "yes", + "enable_on_reboot_value": True, + "installation_state_value": False, + "set_installation_state": False + } + }, + "assertions": { + "config_value_expected": 'apply_updates = no\ndownload_updates = no\n', + "config_exists": True + } + } + + all_testcases = [revert_success_testcase, revert_success_with_dnf_not_installed_testcase, + revert_success_with_dnf_installed_but_no_config_value_testcase, + revert_success_backup_config_does_not_exist_testcase, + revert_success_default_backup_config_invalid_testcase] + + for testcase in all_testcases: + self.tearDown() + self.setUp() + captured_output, original_stdout = None, None + if testcase["stdio"]["capture_output"]: + # arrange capture std IO + captured_output, original_stdout = self.__capture_std_io() + + self.runtime.set_legacy_test_type(testcase["legacy_type"]) + package_manager = self.container.get('package_manager') + + # setup current auto OS update config, backup for system default config and invoke revert to system default + self.__setup_config_and_invoke_revert_auto_os_to_system_default(package_manager,create_current_auto_os_config=bool(testcase["config"]["current_auto_update_config"]["create_current_auto_os_config"]), + current_auto_os_update_config_value=testcase["config"]["current_auto_update_config"]["current_auto_os_update_config_value"], + create_backup_for_system_default_config=bool(testcase["config"]["backup_system_default_config"]["create_backup_for_system_default_config"]), + apply_updates_value=testcase["config"]["backup_system_default_config"]["apply_updates_value"], + download_updates_value=testcase["config"]["backup_system_default_config"]["download_updates_value"], + enable_on_reboot_value=bool(testcase["config"]["backup_system_default_config"]["enable_on_reboot_value"]), + installation_state_value=bool(testcase["config"]["backup_system_default_config"]["installation_state_value"]), + set_installation_state=bool(testcase["config"]["backup_system_default_config"]["set_installation_state"])) + # assert + if testcase["stdio"]["capture_output"]: + # restore sys.stdout output + sys.stdout = original_stdout + self.__assert_std_io(captured_output=captured_output,expected_output=testcase["stdio"]["expected_output"]) + self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_exists=bool( + testcase["assertions"]["config_exists"]), config_value_expected=testcase["assertions"]["config_value_expected"]) + + def test_dedupe_update_packages_to_get_latest_versions(self): + packages = [] + package_versions = [] + + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + deduped_packages, deduped_package_versions = package_manager.dedupe_update_packages_to_get_latest_versions( + packages, package_versions) + self.assertEqual(deduped_packages, []) + self.assertEqual(deduped_package_versions, []) + + packages = ['python3.x86_64', 'dracut.x86_64', 'libxml2.x86_64', 'azurelinux-release.noarch', 'python3.noarch', + 'python3.x86_64', 'python3.x86_64', 'hypervvssd.x86_64', 'python3.x86_64', 'python3.x86_64'] + package_versions = ['3.12.3-1.azl3', '102-7.azl3 ', '2.11.5-1.azl3', '3.0-16.azl3', '3.12.9-2.azl3', + '3.12.9-1.azl3', '3.12.3-4.azl3', '6.6.78.1-1.azl3', '3.12.3-5.azl3', '3.12.3-5.azl3'] + deduped_packages, deduped_package_versions = package_manager.dedupe_update_packages_to_get_latest_versions( + packages, package_versions) + + self.assertIsNotNone(deduped_packages) + self.assertNotEqual(deduped_packages, []) + self.assertIsNotNone(deduped_package_versions) + self.assertNotEqual(deduped_package_versions, []) + self.assertEqual(len(deduped_packages), 6) + self.assertEqual(deduped_packages[0], 'python3.x86_64') + self.assertEqual(deduped_package_versions[0], '3.12.9-1.azl3') + + def test_obsolete_packages_should_not_considered_in_available_updates(self): + self.runtime.set_legacy_test_type('ObsoletePackages') + package_manager = self.container.get('package_manager') + + # test for all available versions + package_versions = package_manager.get_all_available_versions_of_package("kernel") + self.assertEqual(len(package_versions), 1) + self.assertEqual(package_versions[0], '6.12.0-211.28.1.el10_2') + + def test_install_package_failure(self): + """Unit test for install package failure""" + self.runtime.set_legacy_test_type('FailInstallPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + # test for unsuccessfully installing a package + self.assertEqual(package_manager.install_update_and_dependencies_and_get_status('hyperv-daemons-license.noarch','6.6.78.1-1.azl3',simulate=True),Constants.FAILED) + + def test_get_product_name(self): + """Unit test for retrieving product Name""" + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + self.assertEqual(package_manager.get_product_name("bash.x86_64"), "bash.x86_64") + self.assertEqual(package_manager.get_product_name("firefox.x86_64"), "firefox.x86_64") + self.assertEqual(package_manager.get_product_name("test.noarch"), "test.noarch") + self.assertEqual(package_manager.get_product_name("noextension"), "noextension") + self.assertEqual(package_manager.get_product_name("noextension.ext"), "noextension.ext") + + def test_package_manager_no_updates(self): + """Unit test for dnf4 package manager with no updates""" + self.runtime.set_legacy_test_type('SadPath') + + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + + available_updates, package_versions = package_manager.get_available_updates(package_filter) + self.assertEqual(len(available_updates), 0) + self.assertEqual(len(package_versions), 0) + + def test_package_manager(self): + """Unit test for dnf4 package manager""" + self.runtime.set_legacy_test_type('HappyPath') + + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + + # Test: get_available_updates (do NOT assert exact count unless mock supports it) + available_updates, package_versions = package_manager.get_available_updates(package_filter) + self.assertIsNotNone(available_updates) + self.assertIsNotNone(package_versions) + self.assertEqual(len(available_updates), len(package_versions)) + + cmd = package_manager.single_package_upgrade_simulation_cmd + "hyperv-daemons.x86_64" + code, out = self.runtime.env_layer.run_command_output(cmd, False, False) + size = package_manager.get_package_size(out) + self.assertEqual(size, "91 k") + + # test for get_package_size when size is not available + cmd = package_manager.single_package_upgrade_cmd + "systemd" + code, out = self.runtime.env_layer.run_command_output(cmd, False, False) + size = package_manager.get_package_size(out) + self.assertEqual(size, Constants.UNKNOWN_PACKAGE_SIZE) + + # Test: get_all_available_versions_of_package (ONLY python3 since mock exists) + versions = package_manager.get_all_available_versions_of_package("python3") + self.assertIsNotNone(versions) + self.assertEqual(len(versions), 1) + self.assertEqual(versions[0], '3.12.13-2.el10_2') + + # Test: install command generation (pure logic, safe to test) + # Test exception handling scenarios + self.runtime.stop() + self.runtime = RuntimeCompositor(ArgumentComposer().get_composed_arguments(), True,Constants.DNF4) + + self.container = self.runtime.container + self.runtime.set_legacy_test_type('ExceptionPath') + + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + + # Exception test: get_available_updates + try: + package_manager.get_available_updates(package_filter) + except Exception as exception: + self.assertTrue(str(exception)) + else: + self.assertFalse(True, "Exception did not occur and test failed.") + + # Exception test: get_dependent_list + try: + package_manager.get_dependent_list(["hyperv-daemons.x86_64"]) + except Exception as exception: + self.assertTrue(str(exception)) + else: + self.assertFalse(True, "Exception did not occur and test failed.") + + self.runtime.set_legacy_test_type('SuccessInstallPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + + # Test: get_dependent_list + dependent_list = package_manager.get_dependent_list(["hyperv-daemons.x86_64"]) + self.assertIsNotNone(dependent_list) + + self.runtime.set_legacy_test_type('AnotherSadPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + sad_dependent_list = package_manager.get_dependent_list(["openssl-999.999"]) + self.assertIsNotNone(sad_dependent_list) + self.assertEqual(len(sad_dependent_list), 0) + + def test_install_package_success(self): + """Unit test for install package success""" + self.runtime.set_legacy_test_type('SuccessInstallPath') + + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + + # test for successfully installing a package + self.assertEqual(package_manager.install_update_and_dependencies_and_get_status('rubygem-json.x86_64','2.7.2-12.el10_1',simulate=True),Constants.INSTALLED) + + def test_inclusion_type_other(self): + """Unit test for dnf4 package manager with inclusion and Classification = Other. All packages are considered are 'Security' since DNF does not have patch classification""" + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + self.runtime.stop() + + argument_composer = ArgumentComposer() + argument_composer.classifications_to_include = [Constants.PackageClassification.OTHER] + argument_composer.patches_to_include = ["ssh", "tcpdump"] + argument_composer.patches_to_exclude = ["ssh*", "test"] + self.runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.DNF4) + self.container = self.runtime.container + + package_filter = self.container.get('package_filter') + self.assertIsNotNone(package_filter) + + # test for get_available_updates + available_updates, package_versions = package_manager.get_available_updates(package_filter) + self.assertIsNotNone(available_updates) + self.assertIsNotNone(package_versions) + self.assertEqual(0, len(available_updates)) + self.assertEqual(0, len(package_versions)) + + def test_update_image_default_patch_mode(self): + package_manager = self.container.get('package_manager') + package_manager.os_patch_configuration_settings_file_path = package_manager.dnf4_automatic_configuration_file_path = os.path.join(self.runtime.execution_config.config_folder, "automatic.conf") + + # disable apply_updates when enabled by default + dnf4_automatic_os_patch_configuration_settings = 'apply_updates = yes\ndownload_updates = yes\n' + self.runtime.write_to_file(package_manager.dnf4_automatic_configuration_file_path,dnf4_automatic_os_patch_configuration_settings) + + package_manager.update_os_patch_configuration_sub_setting(package_manager.dnf4_automatic_apply_updates_identifier_text, "no",package_manager.dnf4_automatic_config_pattern_match_text) + dnf4_automatic_os_patch_configuration_settings_file_path_read = self.runtime.env_layer.file_system.read_with_retry(package_manager.os_patch_configuration_settings_file_path) + self.assertIsNotNone(dnf4_automatic_os_patch_configuration_settings_file_path_read) + self.assertIn('apply_updates = no', dnf4_automatic_os_patch_configuration_settings_file_path_read) + self.assertIn('download_updates = yes', dnf4_automatic_os_patch_configuration_settings_file_path_read) + + # disable download_updates when enabled by default + dnf4_automatic_os_patch_configuration_settings = 'apply_updates = yes\ndownload_updates = yes\n' + self.runtime.write_to_file(package_manager.os_patch_configuration_settings_file_path,dnf4_automatic_os_patch_configuration_settings) + package_manager.update_os_patch_configuration_sub_setting(package_manager.dnf4_automatic_download_updates_identifier_text, "no",package_manager.dnf4_automatic_config_pattern_match_text) + dnf4_automatic_os_patch_configuration_settings_file_path_read = self.runtime.env_layer.file_system.read_with_retry(package_manager.os_patch_configuration_settings_file_path) + self.assertIsNotNone(dnf4_automatic_os_patch_configuration_settings_file_path_read is not None) + self.assertIn('apply_updates = yes', dnf4_automatic_os_patch_configuration_settings_file_path_read) + self.assertIn('download_updates = no', dnf4_automatic_os_patch_configuration_settings_file_path_read) + + # disable apply_updates when default patch mode settings file is empty + dnf4_automatic_os_patch_configuration_settings = '' + self.runtime.write_to_file(package_manager.os_patch_configuration_settings_file_path,dnf4_automatic_os_patch_configuration_settings) + package_manager.update_os_patch_configuration_sub_setting(package_manager.dnf4_automatic_apply_updates_identifier_text, "no",package_manager.dnf4_automatic_config_pattern_match_text) + dnf4_automatic_os_patch_configuration_settings_file_path_read = self.runtime.env_layer.file_system.read_with_retry(package_manager.os_patch_configuration_settings_file_path) + self.assertIsNotNone(dnf4_automatic_os_patch_configuration_settings_file_path_read) + self.assertNotIn('download_updates', dnf4_automatic_os_patch_configuration_settings_file_path_read) + self.assertIn('apply_updates = no' , dnf4_automatic_os_patch_configuration_settings_file_path_read) + + def test_is_reboot_pending(self): + """Unit test for dnf4 package manager reboot detection""" + # Restart required (needs-restarting returns code=1) + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + self.assertTrue(package_manager.is_reboot_pending()) + + # Restart not required (needs-restarting returns code=0) + self.runtime.set_legacy_test_type('SadPath') + self.runtime.env_layer.run_output_command = self.mock_run_command_output_no_reboot + self.assertFalse(package_manager.is_reboot_pending()) + + # Exception Path + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + self.runtime.env_layer.file_system.write_with_retry = self.mock_write_with_retry_raise_exception + self.assertRaises(Exception, package_manager.is_reboot_pending()) + + def test_disable_auto_os_update_on_reboot(self): + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertTrue(package_manager) + command = "systemctl disable --now dnf-automatic.timer" + + package_manager.disable_auto_update_on_reboot(command) + + self.runtime.set_legacy_test_type('AnotherSadPath') + package_manager = self.container.get('package_manager') + command = "systemctl enable --nows dnf-automatic.timer" + self.assertRaises(Exception, package_manager.disable_auto_update_on_reboot, command) + + def test_enable_auto_os_update_on_reboot(self): + self.runtime.set_legacy_test_type('SadPath') + package_manager = self.container.get('package_manager') + self.assertTrue(package_manager) + + package_manager.enable_on_reboot_cmd = "systemctl enable --now dnf-automatic.timer" + self.assertRaises(Exception, package_manager.enable_auto_update_on_reboot) + + def test_do_processes_require_restart(self): + """Unit test for dnf4 package manager""" + # Restart required + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertTrue(package_manager) + self.assertTrue(package_manager.is_reboot_pending()) + + # Restart not required + self.runtime.set_legacy_test_type('SadPath') + package_manager = self.container.get('package_manager') + self.assertIsNotNone(package_manager) + self.assertFalse(package_manager.is_reboot_pending()) + + def test_get_security_updates(self): + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertTrue(package_manager) + + security_packages, security_package_versions = package_manager.get_security_updates() + self.assertTrue(5, security_packages) + self.assertTrue(5, security_package_versions) + + def test_update_os_patch_configuration_sub_setting_exception_handling(self): + """Test exception handling when override file write fails""" + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + # Mock file_system.write_with_retry to raise exception + self.runtime.env_layer.file_system.write_with_retry = self.mock_write_with_retry_raise_exception + self.assertRaises(Exception, package_manager.update_os_patch_configuration_sub_setting, ) + + def test_backup_image_default_patch_configuration_if_not_exists_exception_handling(self): + """Test exception handling when override file write fails""" + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + # Mock file_system.write_with_retry to raise exception + self.runtime.env_layer.file_system.write_with_retry = self.mock_write_with_retry_raise_exception + self.assertRaises(Exception, package_manager.backup_image_default_patch_configuration_if_not_exists, ) + + def test_get_package_install_expected_avg_time_in_seconds(self): + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + self.assertEqual(package_manager.get_package_install_expected_avg_time_in_seconds(), 90) + + def test_no_op_methods(self): + """Test all no-op methods execute without error""" + self.runtime.set_legacy_test_type('HappyPath') + package_manager = self.container.get('package_manager') + + # These should all execute without raising exceptions + package_manager.do_processes_require_restart() + package_manager.set_max_patch_publish_date() + package_manager.add_arch_dependencies(package_manager, "pkg", "1.0", [], [], [], []) + package_manager.set_security_esm_package_status("op", []) + package_manager.separate_out_esm_packages([], []) + +if __name__ == '__main__': + unittest.main() + diff --git a/src/core/tests/Test_EnvLayer.py b/src/core/tests/Test_EnvLayer.py index cecc0ec8..fcd11aa5 100644 --- a/src/core/tests/Test_EnvLayer.py +++ b/src/core/tests/Test_EnvLayer.py @@ -71,6 +71,30 @@ def mock_run_command_for_tdnf(self, cmd, no_output=False, chk_err=False): return 0, '' return -1, '' + def mock_run_command_for_dnf4(self, cmd, no_output=False, chk_err=False): + if "which dnf" in cmd: + return 0, '/usr/bin/dnf' + if "dnf --version" in cmd: + return 0, '4.20.0' + return -1, '' + + def mock_run_command_for_dnf_not_found(self, cmd, no_output=False, chk_err=False): + return -1, '' + + def mock_run_command_for_dnf_wrong_version(self, cmd, no_output=False, chk_err=False): + if "which dnf" in cmd: + return 0, '/usr/bin/dnf' + if "dnf --version" in cmd: + return 0, 'dnf version 6.14.0' + return -1, '' + + def mock_run_command_for_dnf_version_command_failure(self, cmd, no_output=False, chk_err=False): + if "which dnf" in cmd: + return 0, '/usr/bin/dnf' + if "dnf --version" in cmd: + return -1, 'dnf version command failure' + return -1, '' + def mock_distro_os_release_attr_return_azure_linux_3(self, attribute): return '3.0.0' @@ -152,36 +176,70 @@ def test_platform(self): self.envlayer.platform.cpu_arch() self.envlayer.platform.vm_name() - def test_get_package_manager_azure_linux_4_and_rhel10_not_supported(self): - """Test for RHEL 10 log unsupported message""" + + def test_is_distro_rhel_10(self): + self.backup_envlayer_distro_os_release_attr = distro.os_release_attr + + test_input_output_table = [ + [self.mock_linux_distribution_to_return_rhel_10, + self.mock_distro_os_release_attr_return_rhel_10, True], + [self.mock_linux_distribution_to_return_rhel_10, self.mock_distro_os_release_attr_return_none, False], + ] + + for row in test_input_output_table: + distro_name = row[0]()[0] # Extract distro name from tuple (first element) + distro.os_release_attr = row[1] + result = self.envlayer.is_distro_rhel_10(distro_name) + self.assertEqual(result, row[2]) + + # restore original methods + distro.os_release_attr = self.backup_envlayer_distro_os_release_attr + + def test_get_package_manager_dnf4_error_cases(self): + """Test dnf4 error cases in get_package_manager""" self.backup_platform_system = platform.system self.backup_linux_distribution = self.envlayer.platform.linux_distribution + self.backup_run_command_output = self.envlayer.run_command_output self.backup_distro_os_release_attr = distro.os_release_attr platform.system = self.mock_platform_system + self.envlayer.platform.linux_distribution = self.mock_linux_distribution_to_return_rhel_10 + distro.os_release_attr = self.mock_distro_os_release_attr_return_rhel_10 + test_input_output_table = [ - [self.mock_linux_distribution_to_return_rhel_10, self.mock_distro_os_release_attr_return_rhel_10, "Error: This distro is not yet supported in your region. Please review https://aka.ms/VMGuestPatchingCompatibility for more information. [Distro=Red Hat][Version=10.0][Code=abc]\n"], + [self.mock_run_command_for_dnf4, "dnf4"], + [self.mock_run_command_for_dnf_not_found, str()], + [self.mock_run_command_for_dnf_wrong_version, str()], + [self.mock_run_command_for_dnf_version_command_failure, str()] ] - for row in test_input_output_table: - captured_output = StringIO() - original_output = sys.stdout - sys.stdout = captured_output - self.envlayer.platform.linux_distribution = row[0] - distro.os_release_attr = row[1] + for row in test_input_output_table: + self.envlayer.run_command_output = row[0] result = self.envlayer.get_package_manager() - sys.stdout = original_output - self.assertEqual(row[2], captured_output.getvalue()) - self.assertEqual(result, "") - - # restore - self.__restore_mocks() + self.assertEqual(result, row[1]) - def __restore_mocks(self): - """Restore backed up mocks to their original state""" - distro.os_release_attr = self.backup_distro_os_release_attr + # restore original methods + self.envlayer.run_command_output = self.backup_run_command_output self.envlayer.platform.linux_distribution = self.backup_linux_distribution + distro.os_release_attr = self.backup_distro_os_release_attr platform.system = self.backup_platform_system + def test_mock_command_fallback_paths(self): + """Test that mock commands return -1 for unexpected commands""" + code, out = self.mock_run_command_for_apt('which apt') + self.assertEqual(code, -1) + + code, out = self.mock_run_command_for_dnf4('which not-dnf') + self.assertEqual(code, -1) + + code, out = self.mock_run_command_for_dnf_wrong_version('dnf --v') + self.assertEqual(code, -1) + + code, out = self.mock_run_command_for_dnf_version_command_failure('dnf --v') + self.assertEqual(code, -1) + + code, out = self.mock_run_command_for_tdnf('which not-tdnf') + self.assertEqual(code, -1) + if __name__ == '__main__': - unittest.main() \ No newline at end of file + unittest.main() diff --git a/src/core/tests/library/LegacyEnvLayerExtensions.py b/src/core/tests/library/LegacyEnvLayerExtensions.py index fb0fa6ce..e2efffc2 100644 --- a/src/core/tests/library/LegacyEnvLayerExtensions.py +++ b/src/core/tests/library/LegacyEnvLayerExtensions.py @@ -688,9 +688,83 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): "dracut x86_64 102-7.azl3 azurelinux-official-base 382.51k 258.06k\n\n" + \ "hyperv-daemons-license noarch 6.6.78.1-1.azl3 azurelinux-official-base 847.91k 403.29k\n" + \ "hypervvssd x86_64 6.6.78.1-1.azl3 azurelinux-official-base 382.51k 258.06k\n\n" + \ - "hypervkvpd x86_64 6.6.78.1-1.azl3 azurelinux-official-base 847.91k 403.29k\n" + \ - "Total installed size: 1.20M\n" + \ - "Total download size: 661.34k\n" + "hypervkvpd x86_64 6.6.78.1-1.azl3 azurelinux-official-base 847.91k 403.29k\n" + \ + "Total installed size: 1.20M\n" + \ + "Total download size: 661.34k\n" + elif self.legacy_package_manager_name is Constants.DNF4: + if cmd.find("check-update") > -1: + code = 100 + output = ( + "Updating Subscription Management repositories.\n" + "Last metadata expiration check: 2:44:12 ago on Tue Jun 30 17:24:19 2026.\n" + "\n" + "dracut.x86_64 107-8.el10_2 rhel-10-baseos-rhui-rpms\n" + "dracut-config-generic.x86_64 107-8.el10_2 rhel-10-baseos-rhui-rpms\n" + "dracut-network.x86_64 107-8.el10_2 rhel-10-baseos-rhui-rpms\n" + "dracut-squash.x86_64 107-8.el10_2 rhel-10-baseos-rhui-rpms\n" + ) + elif cmd.find("dnf4 -y install") > -1: + code = 0 + output = "Complete!\n" + elif "needs-restarting" in cmd: + code = 1 + output = ( + "Updating Subscription Management repositories.\n" + "Core libraries or services have been updated since boot-up:\n" + " * glibc\n\n" + "Reboot is required to fully utilize these updates.\n") + elif cmd.find("dnf4 list --available python3") > -1: + code = 0 + output = ( + "Updating Subscription Management repositories.\n" + "Available Packages\n" + "python3.x86_64 3.12.13-2.el10_2 rhel-10-baseos-rhui-rpms\n") + elif cmd.find("dnf4 install --assumeno --skip-broken") > -1 and "hyperv-daemons" in cmd: + code = 1 + output = ( + "Updating Subscription Management repositories.\n" + "Dependencies resolved.\n" + "==================================================================================================================\n" + " Package Architecture Version Repository Size\n" + "==================================================================================================================\n" + "Installing:\n" + " hyperv-daemons x86_64 0-0.50.20220731git.el10 rhel-10-appstream-rhui-rpms 6.6 k\n" + "Installing dependencies:\n" + " hyperv-daemons-license noarch 0-0.50.20220731git.el10 rhel-10-appstream-rhui-rpms 18 k\n" + " hypervfcopyd x86_64 0-0.50.20220731git.el10 rhel-10-appstream-rhui-rpms 19 k\n" + " hypervkvpd x86_64 0-0.50.20220731git.el10 rhel-10-appstream-rhui-rpms 28 k\n" + " hypervvssd x86_64 0-0.50.20220731git.el10 rhel-10-appstream-rhui-rpms 19 k\n" + "\nTransaction Summary\n" + "==================================================================================================================\n" + "Install 5 Packages\n" + "\nTotal download size: 91 k\n" + "Installed size: 94 k\n" + "Operation aborted.\n") + elif cmd.find("systemctl is-enabled ") > -1: + code = 0 + output = "disabled" + + elif cmd.find("systemctl disable ") > -1: + code = 0 + output = "Removed /etc/systemd/system/multi-user.target.wants/dnf-automatic.service." + + elif cmd.find("rpm -qa") > -1: + code = 0 + output = ( + "dnf-automatic\n" + "hypervkvpd x86_64 6.6.78.1-1.azl3 azurelinux-official-base 847.91k 403.29k\n" + "Total installed size: 1.20M\n" + "Total download size: 661.34k\n") + elif cmd.find("dnf4 list --installed") > -1: + code = 0 + package = cmd.replace('sudo dnf4 list --installed ','').strip() + whitelisted_versions = ['107-8.el10_2','8.12.1-4.el10', '9.0.1-3.el10','2.39-124.el10_2' ] + output = ("Updating Subscription Management repositories.\n""Installed Packages\n") + template = " @System\n" + for version in whitelisted_versions: + entry = template.replace('', package) + entry = entry.replace('', version) + output += entry elif self.legacy_package_manager_name is Constants.DNF5: if cmd.find("check-update") > -1: code = 100 @@ -820,6 +894,12 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): output = '' else: code = 0 + elif self.legacy_package_manager_name is Constants.DNF4: + if cmd.find("systemctl enable --now dnf-automatic.timer") > -1: + code = 1 + output = '' + else: + code = 0 elif cmd.find("systemctl") > -1: code = 1 output = '' @@ -889,6 +969,25 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): elif cmd.find("systemctl is-enabled ") > -1: code = 0 output = 'enabled' + if self.legacy_package_manager_name is Constants.DNF4: + if "systemctl is-enabled" in cmd: + code = 0 + output = 'enabled' + elif cmd.find("rpm -qa") > -1: + code = 0 + output = 'dnf-automatic-4.20.0-22.el10_2.noarch' + elif "systemctl enable --nows dnf-automatic.timer" in cmd: + code = 1 + output = 'systemctl: unrecognized option --nows' + elif cmd.find("dnf4 install --assumeno") > -1 and "openssl-999.999" in cmd: + code = 1 + output = ( + "Updating and loading repositories:\n" + "Repositories loaded.\n" + "Failed to resolve the transaction:\n" + "No match for argument: openssl-999.999\n" + "You can try to add to command line:\n" + " --skip-unavailable to skip unavailable packages\n") if self.legacy_package_manager_name is Constants.DNF5: if cmd.find("systemctl cat dnf5-automatic.service") > -1: code = 0 @@ -1078,6 +1177,34 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): code = 0 output = "Loaded plugin: tdnfrepogpgcheck\n" + \ "hyperv-daemons-license.noarch 6.6.78.1-1.azl3 @System\n" + elif self.legacy_package_manager_name is Constants.DNF4: + if "rubygem-json" in cmd and "--assumeno" in cmd: + code = 1 + output = ("Updating Subscription Management repositories.\n" + "Dependencies resolved.\n" + "=============================================================================\n" + "Installing:\n" + " rubygem-json x86_64 2.7.2-12.el10_1 rhel-10-appstream-rhui-rpms\n" + "Installing dependencies:\n" + " ruby x86_64 3.3.10-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " ruby-default-gems noarch 3.3.10-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " ruby-libs x86_64 3.3.10-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " rubygem-io-console x86_64 0.7.1-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " rubygem-psych x86_64 5.1.2-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " rubygems noarch 3.5.22-12.el10_1 rhel-10-appstream-rhui-rpms\n" + "Installing weak dependencies:\n" + " rubygem-bigdecimal x86_64 3.1.5-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " rubygem-bundler noarch 2.5.22-12.el10_1 rhel-10-appstream-rhui-rpms\n" + " rubygem-rdoc noarch 6.6.3.1-12.el10_1 rhel-10-appstream-rhui-rpms\n" + "\nTransaction Summary\n" + "=============================================================================\n" + "Install 10 Packages\n" + "\nOperation aborted.\n") + elif cmd.find("dnf4 list --installed rubygem-json.x86_64") > -1: + code = 0 + output = ( + "Installed Packages\n" + "rubygem-json.x86_64 2.7.2-12.el10_1 @rhel-10-appstream-rhui-rpms\n") elif self.legacy_package_manager_name is Constants.DNF5: if "rubygem-json" in cmd and "--assumeno" in cmd: code = 0 @@ -1253,6 +1380,21 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): if cmd.find("simulate-install") > -1 or cmd.find("sudo tdnf install --assumeno --skip-broken hyperv-daemons-license") > -1: code = 100 output = "Failed to install package" + elif self.legacy_package_manager_name is Constants.DNF4: + if cmd.find("simulate-install") > -1 or cmd.find("dnf4 install --assumeno --skip-broken hyperv-daemons-license") > -1: + code = 1 + output = ( + "Updating Subscription Management repositories.\n" + "Dependencies resolved.\n" + "=============================================================================================\n" + " Package Arch Version Repository Size\n" + "=============================================================================================\n" + "Installing:\n" + " hyperv-daemons-license noarch 0-0.50.20220731git.el10 rhel-10-appstream-rhui-rpms 18 k\n" + "\nTransaction Summary\n" + "=============================================================================================\n" + "Install 1 Package\n" + "\nOperation aborted.\n") elif self.legacy_package_manager_name is Constants.DNF5: if cmd.find("simulate-install") > -1 or cmd.find("sudo dnf5 install --assumeno --skip-broken hyperv-daemons-license") > -1: code = 1 @@ -1572,6 +1714,13 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): "python3.x86_64 3.12.9-1.azl3 azurelinux-official-base\n" + \ "Obsoleting:\n" + \ "python.x86_64 2.7.9-1.azl3 azurelinux-official-base\n" + elif self.legacy_package_manager_name is Constants.DNF4: + if cmd.find("dnf4 list --available kernel") > -1: + code = 0 + output = "Updating Subscription Management repositories.\n" + \ + "Last metadata expiration check: 0:43:21 ago on Mon Jun 29 14:49:21 2026.\n" + \ + "Available Packages\n" + \ + "kernel.x86_64 6.12.0-211.28.1.el10_2 rhel-10-baseos-rhui-rpms\n" elif self.legacy_package_manager_name is Constants.DNF5: if cmd.find("dnf5 list --available python3") > -1: code = 0 @@ -1684,6 +1833,10 @@ def run_command_output(self, cmd, no_output=False, chk_err=True): if cmd.find("systemctl list-unit-files --type=service | grep dnf-automatic.service") > -1: code = 0 output = 'Auto update service installed' + elif self.legacy_package_manager_name is Constants.DNF4: + if "rpm -qa | grep dnf-automatic" in cmd: + code = 0 + output = 'dnf-automatic' elif self.legacy_package_manager_name is Constants.DNF5: if "rpm -qa | grep dnf5-plugin-automatic" in cmd: code = 0 diff --git a/src/tools/references/cmd_output_references/dnf4_ouput_expected_formats b/src/tools/references/cmd_output_references/dnf4_ouput_expected_formats new file mode 100644 index 00000000..cb741d33 --- /dev/null +++ b/src/tools/references/cmd_output_references/dnf4_ouput_expected_formats @@ -0,0 +1,50 @@ +Example of different output formats for DNF 4 commands: + +In DNF4 : Sample Output for the command 'dnf4 install --assumeno git' is as follows(Success use case): + # Updating Subscription Management repositories. + # Last metadata expiration check: 1:13:33 ago on Wed Jun 10 12:37:56 2026. + # Dependencies resolved. + # ======================================================================================================================================================================================================================================================================================== + # Package Architecture Version Repository Size + # ======================================================================================================================================================================================================================================================================================== + # Installing: + # git x86_64 2.52.0-1.el10 rhel-10-appstream-rhui-rpms 46 k + # Installing dependencies: + # git-core x86_64 2.52.0-1.el10 rhel-10-appstream-rhui-rpms 5.2 M + # git-core-doc noarch 2.52.0-1.el10 rhel-10-appstream-rhui-rpms 3.2 M + # perl-Error noarch 1:0.17029-18.el10 rhel-10-appstream-rhui-rpms 46 k + # perl-Git noarch 2.52.0-1.el10 rhel-10-appstream-rhui-rpms 43 k + # perl-TermReadKey x86_64 2.38-24.el10 rhel-10-appstream-rhui-rpms 40 k + # perl-lib x86_64 0.65-512.2.el10_0 rhel-10-appstream-rhui-rpms 16 k + # Transaction Summary + # ======================================================================================================================================================================================================================================================================================== + # Install 7 Packages + # Total download size: 8.6 M + # Installed size: 40 M + # Operation aborted. + +In DNF4 : Sample output for command 'dnf4 install --assumeno python3' is as follows (arch dependencies): + # Updating Subscription Management repositories. + # Last metadata expiration check: 1:05:28 ago on Tue Jun 9 15:10:49 2026. + # Package python3-3.12.12-3.el10_1.2.x86_64 is already installed. + # Dependencies resolved. + # ======================================================================================================================================================================================================================================================================================== + # Package Architecture Version Repository Size + # ======================================================================================================================================================================================================================================================================================== + # Upgrading: + # expat x86_64 2.7.3-1.el10_2.1 rhel-10-baseos-rhui-rpms 126 k + # python-unversioned-command noarch 3.12.13-2.el10_2 rhel-10-appstream-rhui-rpms 17 k + # python3 x86_64 3.12.13-2.el10_2 rhel-10-baseos-rhui-rpms 35 k + # python3-libs x86_64 3.12.13-2.el10_2 rhel-10-baseos-rhui-rpms 9.5 M + # Transaction Summary + # ======================================================================================================================================================================================================================================================================================== + # Upgrade 4 Packages + # Total download size: 9.6 M + # Operation aborted. + # + # 2nd Sample output for the command `dnf4 install --assumeno glibc.i686` (Failure case - package not found): + # Updating Subscription Management repositories. + # Last metadata expiration check: 1:05:52 ago on Tue Jun 9 15:10:49 2026. + # No match for argument: glibc.i686 + # Error: Unable to find a match: glibc.i686 +