Address Pending Comments from Dnf5 Original PR

src/core/src/bootstrap/EnvLayer.py

@@ -81,6 +81,18 @@ 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
+        if code != 0 or not out:
+            return code, out, None
+        version = str(out).split()[-1]
+        return code, out, version
+
     def get_package_manager(self):
         # type: () -> str
         """ Detects package manager type """
@@ -99,13 +111,19 @@ def get_package_manager(self):
 
         # Check for Azure Linux 4 or Above( uses dnf5)
         if self.is_distro_azure_linux_4(str(os_name)):
-            code, out = self.run_command_output('which dnf', False, False)
-            if code == 0:
-                return Constants.DNF5
-            else:
-                print("Error: Expected package manager dnf5 not found on this Azure Linux4 VM.")
+            if not self.__is_dnf_available():
+                print("Error: Expected package manager dnf not found on this Azure Linux4 VM.")
                 return str()
 
+            code, out, version = self.__get_dnf_version()
+            if version:
+                if version.startswith('5'):
+                    return Constants.DNF5
+                print("Error: Expected dnf version 5 on this Azure Linux4 VM. Found: {0}".format(version))
+                return str()
+            print("Error: Unable to determine dnf version. Code={0}, Output={1}".format(code, out))
+            return str()
+
         # Check for Azure Linux (3 and below use TDNF)
         if self.is_distro_azure_linux(str(os_name)):
             code, out = self.run_command_output('which tdnf', False, False)

src/core/src/package_managers/Dnf5PackageManager.py

@@ -263,9 +263,9 @@ def get_dependent_list(self, packages):
         code, output = self.env_layer.run_command_output(cmd, False, False)
         self.composite_logger.log_verbose("[DNF5] Dependency simulation. [Command={0}][Code={1}]".format(cmd, str(code)))
         if code not in self.dnf5_simulation_valid_exit_codes:
-            self.composite_logger.log_error("[DNF5] Unexpected failure. [Command={0}][Code={1}][Output={2}]".format(cmd, str(code), output))
+            self.composite_logger.log_error("[DNF5] Unexpected failure during dependency simulation. [Command={0}][Code={1}][Output={2}]".format(cmd, str(code), output))
             error_msg = "DNF5 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.DEFAULT_ERROR)
+            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)

src/core/tests/Test_EnvLayer.py

@@ -45,6 +45,9 @@ def mock_platform_system_windows(self):
     def mock_linux_distribution(self):
         return ['test', 'test', 'test']
 
+    def mock_linux_distribution_to_return_azure_linux_4(self):
+        return ['Microsoft Azure Linux', '4.0', '']
+
     def mock_linux_distribution_to_return_azure_linux_3(self):
         return ['Microsoft Azure Linux', '3.0', '']
 
@@ -71,6 +74,33 @@ def mock_run_command_for_tdnf(self, cmd, no_output=False, chk_err=False):
             return 0, ''
         return -1, ''
 
+    def mock_run_command_for_dnf5(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, 'dnf5 version 5.2.18.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 4.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_4(self, attribute):
+        return '4.0.0'
+
     def mock_distro_os_release_attr_return_azure_linux_3(self, attribute):
         return '3.0.0'
 
@@ -96,20 +126,22 @@ def test_get_package_manager(self):
         self.backup_distro_os_release_attr = distro.os_release_attr
 
         test_input_output_table = [
-            [self.mock_run_command_for_apt, self.mock_linux_distribution, Constants.APT],
-            [self.mock_run_command_for_tdnf, self.mock_linux_distribution_to_return_azure_linux_3, Constants.TDNF],
-            [self.mock_run_command_for_yum, self.mock_linux_distribution_to_return_azure_linux_3, str()],  # check for Azure Linux machine which does not have tdnf
-            [self.mock_run_command_for_tdnf, self.mock_linux_distribution_to_return_azure_linux_2, Constants.TDNF],
-            [self.mock_run_command_for_yum, self.mock_linux_distribution, Constants.YUM],
-            [self.mock_run_command_for_zypper, self.mock_linux_distribution, Constants.ZYPPER],
-            [lambda cmd, no_output, chk_err: (-1, ''), self.mock_linux_distribution, str()],    # no matches for any package manager
+            [self.mock_run_command_for_apt, self.mock_linux_distribution, Constants.APT, self.mock_distro_os_release_attr_return_none],
+            [self.mock_run_command_for_dnf5, self.mock_linux_distribution_to_return_azure_linux_4, Constants.DNF5, self.mock_distro_os_release_attr_return_azure_linux_4],
+            [self.mock_run_command_for_tdnf, self.mock_linux_distribution_to_return_azure_linux_3, Constants.TDNF, self.mock_distro_os_release_attr_return_azure_linux_3],
+            [self.mock_run_command_for_yum, self.mock_linux_distribution_to_return_azure_linux_3, str(), self.mock_distro_os_release_attr_return_none],  # check for Azure Linux machine which does not have tdnf
+            [self.mock_run_command_for_tdnf, self.mock_linux_distribution_to_return_azure_linux_2, Constants.TDNF, self.mock_distro_os_release_attr_return_azure_linux_2],
+            [self.mock_run_command_for_yum, self.mock_linux_distribution, Constants.YUM, self.mock_distro_os_release_attr_return_none],
+            [self.mock_run_command_for_zypper, self.mock_linux_distribution, Constants.ZYPPER, self.mock_distro_os_release_attr_return_none],
+            [lambda cmd, no_output, chk_err: (-1, ''), self.mock_linux_distribution, str(), self.mock_distro_os_release_attr_return_none],    # no matches for any package manager
         ]
 
         for row in test_input_output_table:
             self.envlayer.run_command_output = row[0]
             self.envlayer.platform.linux_distribution = row[1]
+            distro.os_release_attr = row[3]
             package_manager = self.envlayer.get_package_manager()
-            self.assertTrue(package_manager is row[2])
+            self.assertEqual(package_manager, row[2])
 
         # test for Windows
         platform.system = self.mock_platform_system_windows
@@ -118,6 +150,7 @@ def test_get_package_manager(self):
         # 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_is_distro_azure_linux_3(self):
@@ -138,6 +171,51 @@ def test_is_distro_azure_linux_3(self):
         # restore original methods
         distro.os_release_attr = self.backup_envlayer_distro_os_release_attr
 
+    def test_is_distro_azure_linux_4(self):
+        self.backup_envlayer_distro_os_release_attr = distro.os_release_attr
+
+        test_input_output_table = [
+            [self.mock_linux_distribution_to_return_azure_linux_4, self.mock_distro_os_release_attr_return_azure_linux_4, True],
+            [self.mock_linux_distribution_to_return_azure_linux_4, 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_azure_linux_4(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_dnf5_error_cases(self):
+        """Test dnf5 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_azure_linux_4
+        distro.os_release_attr = self.mock_distro_os_release_attr_return_azure_linux_4
+
+        test_input_output_table = [
+            [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:
+            self.envlayer.run_command_output = row[0]
+            result = self.envlayer.get_package_manager()
+            self.assertEqual(result, row[1])
+
+        # 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_filesystem(self):
         # only validates if these invocable without exceptions
         backup_retry_count = Constants.MAX_FILE_OPERATION_RETRY_COUNT
@@ -152,7 +230,7 @@ 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):
+    def test_get_package_manager_rhel10_not_supported(self):
         """Test for RHEL 10 log unsupported message"""
         self.backup_platform_system = platform.system
         self.backup_linux_distribution = self.envlayer.platform.linux_distribution
@@ -177,6 +255,23 @@ def test_get_package_manager_azure_linux_4_and_rhel10_not_supported(self):
         # restore
         self.__restore_mocks()
 
+    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_dnf5('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)
+
     def __restore_mocks(self):
         """Restore backed up mocks to their original state"""
         distro.os_release_attr = self.backup_distro_os_release_attr