UT Coverage

Author: yashnap

src/core/src/package_managers/Dnf5PackageManager.py

@@ -45,7 +45,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
 
         # Support to check if reboot is required
         # dnf-utils not required (needs-restarting is built into dnf5)
-        self.needs_restarting_with_flag = 'sudo LANG=en_US.UTF8 dnf5 needs-restarting'
+        self.needs_restarting_with_flag = 'sudo LANG=en_US.UTF8 dnf5 needs-restarting -r'
 
         # DNF5 exit codes
         self.dnf_exitcode_ok = 0
@@ -552,9 +552,8 @@ def backup_image_default_patch_configuration_if_not_exists(self):
                 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 = "[DNF5] 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)
+            self.composite_logger.log_error( "[DNF5] Exception during fetching and logging default auto update settings on the machine. [Exception={0}]".format(repr(error)))
+            self.status_handler.add_error_to_status( "[DNF5] Exception during fetching and logging default auto update settings on the machine. [Exception={0}]".format(repr(error)), Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
             raise
 
     def is_image_default_patch_configuration_backup_valid(self, image_default_patch_configuration_backup):
@@ -845,25 +844,15 @@ def is_service_set_to_enable_on_reboot(self, command):
 
     def __set_dnf5_automatic_execstart_flags(self, download_updates=None, apply_updates=None):
         try:
-            if download_updates is None and apply_updates is None:
-                self.__remove_dnf5_automatic_execstart_override()
-                return
-
             flags = ["/usr/bin/dnf5", "automatic", "--timer"]
 
             if apply_updates is True:
                 flags.append("--installupdates")
-
-            elif apply_updates is False:
-                if download_updates is True:
-                    flags.append("--downloadupdates")
+                flags.append("--downloadupdates")
+            else:  # apply_updates is False OR None
+                if apply_updates is False:
                     flags.append("--no-installupdates")
-                elif download_updates is False:
-                    flags.append("--no-downloadupdates")
-                    flags.append("--no-installupdates")
-                else:
-                    flags.append("--no-installupdates")
-            else:
+
                 if download_updates is True:
                     flags.append("--downloadupdates")
                 elif download_updates is False:
@@ -877,16 +866,14 @@ def __set_dnf5_automatic_execstart_flags(self, download_updates=None, apply_upda
             self.composite_logger.log_debug("[DNF5] Wrote override. [ExecStart={0}]".format(" ".join(flags)))
 
         except Exception as error:
-            error_msg = "[DNF5] Error writing override. [Exception={0}]".format(repr(error))
-            self.composite_logger.log_error(error_msg)
-            self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
+            self.composite_logger.log_error("[DNF5] Error writing override. [Exception={0}]".format(repr(error)))
+            self.status_handler.add_error_to_status("[DNF5] Error writing override. [Exception={0}]".format(repr(error)), Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
             raise
 
     def __remove_dnf5_automatic_execstart_override(self):
         """Removes systemd override file for dnf5-automatic if it exists."""
         try:
             try:
-                # self.env_layer.file_system.write_with_retry(self.dnf5_automatic_override_file,"",mode='w+')
                 if os.path.exists(self.dnf5_automatic_override_file):
                     os.remove(self.dnf5_automatic_override_file)
                     self.composite_logger.log_debug("[DNF5] Removed override file. [File={0}]".format(self.dnf5_automatic_override_file))
@@ -900,9 +887,8 @@ def __remove_dnf5_automatic_execstart_override(self):
             self.composite_logger.log_debug("[DNF5] Cleared dnf5 automatic override file. [File={0}]".format(self.dnf5_automatic_override_file))
 
         except Exception as error:
-            error_msg = "[DNF5] Error removing dnf5 automatic override. [Exception={0}]".format(repr(error))
-            self.composite_logger.log_error(error_msg)
-            self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
+            self.composite_logger.log_error("[DNF5] Error removing dnf5 automatic override. [Exception={0}]".format(repr(error)))
+            self.status_handler.add_error_to_status("[DNF5] Error removing dnf5 automatic override. [Exception={0}]".format(repr(error)), Constants.PatchOperationErrorCodes.DEFAULT_ERROR)
             raise
 
     # endregion

src/core/tests/Test_Dnf5PackageManager.py

@@ -17,6 +17,8 @@
 import os
 import sys
 import unittest
+from unittest.mock import patch
+
 # Conditional import for StringIO
 try:
     from StringIO import StringIO  # Python 2
@@ -643,6 +645,12 @@ def test_package_manager_dnf5(self):
         size = package_manager.get_package_size(out)
         self.assertEqual(size, "135.09k")
 
+        # 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.assertTrue(versions is not None)
@@ -801,9 +809,7 @@ def test_update_image_default_patch_mode_dnf5(self):
 
         override_read = self.runtime.env_layer.file_system.read_with_retry(override_file)
         self.assertTrue(override_read is not None)
-        self.assertTrue("--no-installupdates" in override_read)
         self.assertTrue("--downloadupdates" in override_read)
-        self.assertTrue("--no-downloadupdates" not in override_read)
         self.assertTrue("--installupdates" not in override_read)
 
         # -----------------------------
@@ -837,9 +843,7 @@ def test_update_image_default_patch_mode_dnf5(self):
 
         override_read = self.runtime.env_layer.file_system.read_with_retry(override_file)
         self.assertTrue(override_read is not None)
-        self.assertTrue("--no-installupdates" in override_read)
         self.assertTrue("--downloadupdates"  in override_read)
-        self.assertTrue("--no-downloadupdates" not in override_read)
         self.assertTrue("--installupdates" not in override_read)
 
 
@@ -883,6 +887,77 @@ def test_install_security_updates_azgps_coordinated(self):
         self.assertTrue(code == 0)
         self.assertTrue("Complete!", out)
 
+    def test_backup_image_default_patch_configuration_with_downloadupdates_only_execstart(self):
+        """
+        DNF5:
+        Validates backup creation when current ExecStart contains only --downloadupdates
+        and no explicit apply flag.
+
+        Expected backup:
+          - installation_state = True
+          - enable_on_reboot = True
+          - download_updates = "yes"
+          - apply_updates = ""
+        """
+        self.runtime.set_legacy_test_type('AnotherHappyPath')
+        package_manager = self.container.get('package_manager')
+        self.assertIsNotNone(package_manager)
+
+        # Arrange
+        package_manager._Dnf5PackageManager__init_auto_update_for_dnf5_automatic()
+
+        # Make sure no stale backup exists from previous runs
+        if os.path.exists(package_manager.image_default_patch_configuration_backup_path):
+            os.remove(package_manager.image_default_patch_configuration_backup_path)
+
+        # Make sure service override dir exists
+        override_dir = os.path.dirname(package_manager.dnf5_automatic_override_file)
+        os.makedirs(override_dir, exist_ok=True)
+
+        # Write override with ONLY --downloadupdates
+        override_text = "[Service]\nExecStart=\nExecStart=/usr/bin/dnf5 automatic --timer --downloadupdates\n"
+        self.runtime.env_layer.file_system.write_with_retry(
+            package_manager.dnf5_automatic_override_file,
+            override_text,
+            mode='w+'
+        )
+
+        # Act
+        package_manager.backup_image_default_patch_configuration_if_not_exists()
+
+        # Assert
+        self.assertTrue(os.path.exists(package_manager.image_default_patch_configuration_backup_path))
+
+        backup_text = self.runtime.env_layer.file_system.read_with_retry(package_manager.image_default_patch_configuration_backup_path)
+        backup_json = json.loads(backup_text)
+
+        self.assertIn(package_manager.current_auto_os_update_service, backup_json)
+
+        service_backup = backup_json[package_manager.current_auto_os_update_service]
+
+        self.assertEqual("yes", service_backup[package_manager.download_updates_identifier_text])
+        self.assertEqual("", service_backup[package_manager.apply_updates_identifier_text])
+        self.assertFalse(service_backup[package_manager.enable_on_reboot_identifier_text])
+        self.assertTrue(service_backup[package_manager.installation_state_identifier_text])
+
+
+    def test_set_dnf5_automatic_execstart_flags_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
+        with patch.object(package_manager.env_layer.file_system, 'write_with_retry') as mock_write:
+            mock_write.side_effect = IOError("Permission denied")
+
+            # Should raise exception caught by outer try-except
+            self.assertRaises(Exception, package_manager._Dnf5PackageManager__set_dnf5_automatic_execstart_flags,
+                              download_updates=False, apply_updates=False)
+
+    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.assertTrue(package_manager.get_package_install_expected_avg_time_in_seconds(), 90)
 
 if __name__ == '__main__':
     unittest.main()

src/core/tests/library/LegacyEnvLayerExtensions.py

@@ -678,14 +678,6 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
                                  "python3.x86_64 3.12.3-4.azl4~20260501 azurelinux-base\n" + \
                                  "python3.x86_64 3.12.3-5.azl4~20260501 azurelinux-base\n" + \
                                  "python3.x86_64 3.12.3-6.azl4~20260501 azurelinux-base\n"
-
-                    elif ("list available" in cmd) and ("hyperv-daemons.x86_64" in cmd):
-                        code = 0
-                        output = "Updating and loading repositories:\n" + \
-                                 "Repositories loaded.\n" + \
-                                 "Available packages\n" + \
-                                 "hyperv-daemons.x86_64 6.10-3.azl4~20260501 azurelinux-base\n"
-
                     elif cmd.find("dnf5 install --assumeno --skip-broken") > -1 and "hyperv-daemons" in cmd:
                         code = 1
                         output = "Updating and loading repositories:\n" + \
@@ -761,10 +753,7 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
                         code = 0
                         output = ''
                 elif self.legacy_package_manager_name is Constants.DNF:
-                    if cmd.find("systemctl cat dnf5-automatic.service") > -1:
-                        code = 0
-                        output = "ExecStart=/usr/bin/dnf5 automatic --timer --downloadupdates --no-installupdates"
-                    elif cmd.find("systemctl enable --now dnf5-automatic.timer") > -1:
+                    if cmd.find("systemctl enable --now dnf5-automatic.timer") > -1:
                         code = 1
                         output = ''
                     else:
@@ -854,6 +843,16 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
             elif self.legacy_test_type == 'ExceptionPath':
                 code = -1
                 output = ''
+            elif self.legacy_test_type == 'AnotherHappyPath':
+                if cmd.find("systemctl cat dnf5-automatic.service") > -1:
+                    code = 0
+                    output = "ExecStart=/usr/bin/dnf5 automatic --timer --downloadupdates"
+                elif cmd.find("rpm -qa") > -1:
+                    code = 0
+                    output = 'dnf5-plugin-automatic'
+                elif "systemctl is-enabled" in cmd:
+                    code = 0
+                    output = "disabled"
             elif self.legacy_test_type == 'SuccessInstallPath':
                 if cmd.find("cat /proc/cpuinfo | grep name") > -1:
                     code = 0
@@ -1618,10 +1617,6 @@ 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.DNF:
-                    if cmd.find("systemctl list-unit-files --type=service | grep dnf5-automatic.service") > -1:
-                        code = 0
-                        output = 'Auto update service installed'
             major_version = self.get_python_major_version()
             if major_version == 2:
                 return code, output.decode('utf8', 'ignore').encode('ascii', 'ignore')