Add new force_disk parameter for cloud_virtual_instance module (ansible-collections#224)

- Add new force_disk parameter to force deletion and recreation of existing disk files.

Author: thinkdoggie

plugins/doc_fragments/virt_install.py

@@ -1313,6 +1313,7 @@ class ModuleDocFragment(object):
         description:
           - Additional options for the direct attached macvtap interface.
           - The dictionary contains key/value pairs that define individual properties.
+        version_added: '2.1.0'
       target:
         type: dict
         description:

plugins/modules/virt_cloud_instance.py

@@ -56,6 +56,15 @@
       - Only used when O(base_image) is a URL.
       - Useful for updating cached images to their latest versions.
     default: false
+  force_disk:
+    type: bool
+    description:
+      - Force deletion and recreation of existing disk files.
+      - When set to V(true), any existing disk files specified in O(disks) will be removed and recreated.
+      - When set to V(false), the module will fail if any disk file already exists.
+      - This parameter only affects disk file creation, not VM definition.
+      - Use with O(recreate=true) to recreate both the VM and its disk files.
+    default: false
   image_checksum:
     type: str
     description:
@@ -245,6 +254,21 @@
       meta_data:
         instance-id: web-server-01
         local-hostname: web-server-01
+
+# Force overwrite existing disk files
+- name: Create VM and force overwrite existing disk
+  community.libvirt.virt_cloud_instance:
+    name: test-vm
+    base_image: /srv/images/ubuntu-22.04-server-cloudimg-amd64.img
+    force_disk: true
+    disks:
+      - path: /var/lib/libvirt/images/test-vm.qcow2
+        size: 20
+        format: qcow2
+    memory: 2048
+    vcpus: 2
+    networks:
+      - network: default
 """
 
 RETURN = r"""
@@ -431,20 +455,30 @@ def _fetch_checksum_from_url(self, checksum_url, algorithm):
         self.module.fail_json(
             msg="Unable to find a checksum for file '%s' in '%s'" % (filename, checksum_url))
 
-    def build_system_disk(self, disk_param):
+    def build_system_disk(self, disk_param, force_disk=False):
         """
         Build system disk from base image.
 
         This operation creates and modifies disk files, so it respects check_mode.
         In check_mode, validation is performed but no files are created or modified.
+
+        Args:
+            disk_param: Disk configuration dictionary
+            force_disk: If True, remove existing disk file before creating new one
         """
         system_disk = self._resolve_system_disk(disk_param)
         disk_path = self.system_disk_path
         base_image_path = self.base_image_path
 
         if os.path.exists(disk_path):
-            self.module.fail_json(
-                msg="The system disk file already exists: %s" % disk_path)
+            if force_disk:
+                # In check_mode, don't actually delete the file
+                if not self.module.check_mode:
+                    os.remove(disk_path)
+                    self.module.log("Removed existing disk file: %s" % disk_path)
+            else:
+                self.module.fail_json(
+                    msg="The system disk file already exists: %s" % disk_path)
 
         qemuImgTool = QemuImgTool(self.module)
 
@@ -563,6 +597,7 @@ def core(module):
     base_image = module.params.get('base_image')
     image_cache_dir = module.params.get('image_cache_dir')
     force_pull = module.params.get('force_pull', False)
+    force_disk = module.params.get('force_disk', False)
     disks = module.params.get('disks', [])
     image_checksum = module.params.get('image_checksum')
     url_timeout = module.params.get('url_timeout')
@@ -614,7 +649,7 @@ def core(module):
         if not image_operator.validate_checksum():
             module.fail_json(
                 msg="The checksum of the base image does not match the expected value")
-        system_disk = image_operator.build_system_disk(disks[0])
+        system_disk = image_operator.build_system_disk(disks[0], force_disk=force_disk)
 
         # Add base_image_path to result
         result['base_image_path'] = image_operator.base_image_path
@@ -672,6 +707,7 @@ def main():
         base_image=dict(type='str', required=True),
         image_cache_dir=dict(type='path'),
         force_pull=dict(type='bool', default=False),
+        force_disk=dict(type='bool', default=False),
         image_checksum=dict(type='str'),
         url_timeout=dict(type='int', default=60),
         url_username=dict(type='str'),

tests/unit/modules/test_virt_cloud_instance.py

@@ -699,6 +699,89 @@ def test_build_system_disk_size_too_small(self, mock_exists, mock_qemu_tool_clas
             msg=f"The system disk size is too small to import the base image: {expected_system_size} < 21474836480"
         )
 
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.QemuImgTool')
+    @mock.patch('os.path.exists')
+    @mock.patch('os.remove')
+    def test_build_system_disk_force_disk_removes_existing(self, mock_remove, mock_exists, mock_qemu_tool_class):
+        """Test building system disk with force_disk=True removes existing file"""
+        mock_exists.return_value = True  # System disk exists
+
+        # Configure the mocked QemuImgTool
+        mock_qemu_instance = mock.Mock()
+        mock_qemu_tool_class.return_value = mock_qemu_instance
+        mock_qemu_instance.info.return_value = (0, {
+            'format': 'qcow2',
+            'virtual-size': 1073741824  # 1GB
+        }, '')
+        mock_qemu_instance.resize.return_value = (0, '', '')
+
+        disk_param = {
+            'path': '/var/lib/libvirt/images/vm.qcow2',
+            'size': 20,
+            'format': 'qcow2'
+        }
+
+        result = self.operator.build_system_disk(disk_param, force_disk=True)
+
+        # Verify the existing file was removed
+        mock_remove.assert_called_once_with('/var/lib/libvirt/images/vm.qcow2')
+        self.assertEqual(result, disk_param)
+
+        # Verify disk was created after removal
+        self.mock_module.preserved_copy.assert_called_once_with(
+            '/path/to/base.qcow2', '/var/lib/libvirt/images/vm.qcow2'
+        )
+        mock_qemu_instance.resize.assert_called_once()
+
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.QemuImgTool')
+    @mock.patch('os.path.exists')
+    @mock.patch('os.remove')
+    def test_build_system_disk_force_disk_check_mode(self, mock_remove, mock_exists, mock_qemu_tool_class):
+        """Test building system disk with force_disk=True in check_mode doesn't remove file"""
+        self.mock_module.check_mode = True
+        # System disk exists, base image doesn't exist
+        mock_exists.side_effect = lambda path: path == '/var/lib/libvirt/images/vm.qcow2'
+
+        # Configure the mocked QemuImgTool
+        mock_qemu_instance = mock.Mock()
+        mock_qemu_tool_class.return_value = mock_qemu_instance
+
+        disk_param = {
+            'path': '/var/lib/libvirt/images/vm.qcow2',
+            'size': 20,
+            'format': 'qcow2'
+        }
+
+        result = self.operator.build_system_disk(disk_param, force_disk=True)
+
+        # In check_mode, file should NOT be removed
+        mock_remove.assert_not_called()
+
+        # Should return disk config without creating files
+        self.assertEqual(result, disk_param)
+        self.mock_module.preserved_copy.assert_not_called()
+        mock_qemu_instance.resize.assert_not_called()
+
+    @mock.patch('os.path.exists')
+    def test_build_system_disk_force_disk_false_existing_fails(self, mock_exists):
+        """Test building system disk with force_disk=False fails when disk exists"""
+        mock_exists.return_value = True  # System disk exists
+
+        # Make fail_json raise an exception to simulate real behavior
+        self.mock_module.fail_json.side_effect = Exception("Module failed")
+
+        disk_param = {
+            'path': '/var/lib/libvirt/images/vm.qcow2',
+            'size': 20
+        }
+
+        with self.assertRaises(Exception):
+            self.operator.build_system_disk(disk_param, force_disk=False)
+
+        self.mock_module.fail_json.assert_called_once_with(
+            msg="The system disk file already exists: /var/lib/libvirt/images/vm.qcow2"
+        )
+
 
 class TestBaseImageOperatorCheckMode(unittest.TestCase):
     """Test BaseImageOperator check_mode support"""
@@ -928,9 +1011,13 @@ def setUp(self):
             'base_image': 'https://example.com/image.qcow2',
             'image_cache_dir': '/tmp/cache',
             'force_pull': False,
+            'force_disk': False,
             'disks': [{'path': '/var/lib/libvirt/images/test.qcow2', 'size': 20}],
             'image_checksum': None,
-            'url_timeout': None
+            'url_timeout': None,
+            'wait_for_cloud_init_reboot': True,
+            'cloud_init_auto_reboot': True,
+            'cloud_init_reboot_timeout': 600
         }
 
     @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.BaseImageOperator')
@@ -973,7 +1060,7 @@ def test_core_vm_not_exists_create_success(self, mock_update_params, mock_valida
         mock_operator.validate_checksum.assert_called_once()
         mock_operator.build_system_disk.assert_called_once()
         mock_update_params.assert_called_once()
-        mock_virt_install.execute.assert_called_once_with(dryrun=False, wait_timeout=None)
+        mock_virt_install.execute.assert_called_once_with(dryrun=False, wait_timeout=600)
 
         # VM should not be destroyed (doesn't exist)
         mock_virt_conn.destroy.assert_not_called()
@@ -1068,7 +1155,7 @@ def test_core_vm_exists_recreate_active(self, mock_update_params, mock_validate_
         mock_operator.fetch_image.assert_called_once()
         mock_operator.validate_checksum.assert_called_once()
         mock_operator.build_system_disk.assert_called_once()
-        mock_virt_install.execute.assert_called_once_with(dryrun=False, wait_timeout=None)
+        mock_virt_install.execute.assert_called_once_with(dryrun=False, wait_timeout=600)
 
         # Should return success
         self.assertEqual(rc, VIRT_SUCCESS)
@@ -1258,12 +1345,104 @@ def test_core_check_mode_vm_recreation(self, mock_update_params, mock_validate_d
         mock_virt_conn.undefine.assert_not_called()
 
         # But virt-install should be called with dryrun=True
-        mock_virt_install.execute.assert_called_once_with(dryrun=True, wait_timeout=None)
+        mock_virt_install.execute.assert_called_once_with(dryrun=True, wait_timeout=600)
 
         # Should still return success
         self.assertEqual(rc, VIRT_SUCCESS)
         self.assertTrue(result['changed'])
 
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.BaseImageOperator')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.VirtInstallTool')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.LibvirtWrapper')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.validate_disks')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.update_virtinst_params')
+    def test_core_force_disk_passed_to_build_system_disk(self, mock_update_params, mock_validate_disks,
+                                                         mock_libvirt_wrapper_class, mock_virt_install_class,
+                                                         mock_base_image_operator_class):
+        """Test that force_disk parameter is passed to build_system_disk"""
+        from ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance import core, VIRT_SUCCESS
+        from ansible_collections.community.libvirt.plugins.module_utils.libvirt import VMNotFound
+
+        # Set force_disk to True
+        self.mock_module.params['force_disk'] = True
+
+        # Mock LibvirtWrapper - VM doesn't exist
+        mock_virt_conn = mock.Mock()
+        mock_virt_conn.find_vm.side_effect = VMNotFound("VM not found")
+        mock_libvirt_wrapper_class.return_value = mock_virt_conn
+
+        # Mock VirtInstallTool
+        mock_virt_install = mock.Mock()
+        mock_virt_install.execute.return_value = (
+            True, VIRT_SUCCESS, {'some': 'data'})
+        mock_virt_install_class.return_value = mock_virt_install
+
+        # Mock BaseImageOperator
+        mock_operator = mock.Mock()
+        mock_operator.validate_checksum.return_value = True
+        mock_operator.build_system_disk.return_value = {
+            'path': '/var/lib/libvirt/images/test.qcow2'}
+        mock_base_image_operator_class.return_value = mock_operator
+
+        # Execute core function
+        rc, result = core(self.mock_module)
+
+        # Verify force_disk=True was passed to build_system_disk
+        mock_operator.build_system_disk.assert_called_once_with(
+            {'path': '/var/lib/libvirt/images/test.qcow2', 'size': 20},
+            force_disk=True
+        )
+
+        # Should return success
+        self.assertEqual(rc, VIRT_SUCCESS)
+        self.assertTrue(result['changed'])
+
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.BaseImageOperator')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.VirtInstallTool')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.LibvirtWrapper')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.validate_disks')
+    @mock.patch('ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance.update_virtinst_params')
+    def test_core_force_disk_false_passed_to_build_system_disk(self, mock_update_params, mock_validate_disks,
+                                                               mock_libvirt_wrapper_class, mock_virt_install_class,
+                                                               mock_base_image_operator_class):
+        """Test that force_disk=False parameter is passed to build_system_disk"""
+        from ansible_collections.community.libvirt.plugins.modules.virt_cloud_instance import core, VIRT_SUCCESS
+        from ansible_collections.community.libvirt.plugins.module_utils.libvirt import VMNotFound
+
+        # Ensure force_disk is False (default)
+        self.mock_module.params['force_disk'] = False
+
+        # Mock LibvirtWrapper - VM doesn't exist
+        mock_virt_conn = mock.Mock()
+        mock_virt_conn.find_vm.side_effect = VMNotFound("VM not found")
+        mock_libvirt_wrapper_class.return_value = mock_virt_conn
+
+        # Mock VirtInstallTool
+        mock_virt_install = mock.Mock()
+        mock_virt_install.execute.return_value = (
+            True, VIRT_SUCCESS, {'some': 'data'})
+        mock_virt_install_class.return_value = mock_virt_install
+
+        # Mock BaseImageOperator
+        mock_operator = mock.Mock()
+        mock_operator.validate_checksum.return_value = True
+        mock_operator.build_system_disk.return_value = {
+            'path': '/var/lib/libvirt/images/test.qcow2'}
+        mock_base_image_operator_class.return_value = mock_operator
+
+        # Execute core function
+        rc, result = core(self.mock_module)
+
+        # Verify force_disk=False was passed to build_system_disk
+        mock_operator.build_system_disk.assert_called_once_with(
+            {'path': '/var/lib/libvirt/images/test.qcow2', 'size': 20},
+            force_disk=False
+        )
+
+        # Should return success
+        self.assertEqual(rc, VIRT_SUCCESS)
+        self.assertTrue(result['changed'])
+
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)