Reapply OSSA-2024-001 patches for Yoga

nova/tests/unit/virt/libvirt/test_utils.py

@@ -118,24 +118,86 @@ def test_create_image(self, mock_execute):
     @mock.patch('oslo_concurrency.processutils.execute')
     @mock.patch('nova.virt.images.qemu_img_info')
     @mock.patch('nova.image.format_inspector.detect_file_format')
-    def test_create_cow_image(self, mock_detect, mock_info, mock_execute,
-                              mock_exists):
+    def _test_create_cow_image(
+        self, mock_detect, mock_info, mock_execute,
+        mock_exists, backing_file=None, safety_check=True
+    ):
+        if isinstance(backing_file, dict):
+            backing_info = backing_file
+            backing_file = backing_info.pop('file', None)
+        else:
+            backing_info = {}
+        backing_backing_file = backing_info.pop('backing_file', None)
+        backing_fmt = backing_info.pop('backing_fmt',
+                                       mock.sentinel.backing_fmt)
+
         mock_execute.return_value = ('stdout', None)
         mock_info.return_value = mock.Mock(
-            file_format=mock.sentinel.backing_fmt,
+            file_format=backing_fmt,
             cluster_size=mock.sentinel.cluster_size,
-            backing_file=None)
-        mock_detect.return_value.safety_check.return_value = True
+            backing_file=backing_backing_file,
+            format_specific=backing_info)
+
+        mock_detect.return_value.safety_check.return_value = safety_check
+
         libvirt_utils.create_cow_image(mock.sentinel.backing_path,
                                        mock.sentinel.new_path)
         mock_info.assert_called_once_with(mock.sentinel.backing_path)
         mock_execute.assert_has_calls([mock.call(
             'qemu-img', 'create', '-f', 'qcow2', '-o',
             'backing_file=%s,backing_fmt=%s,cluster_size=%s' % (
-                mock.sentinel.backing_path, mock.sentinel.backing_fmt,
+                mock.sentinel.backing_path, backing_fmt,
                 mock.sentinel.cluster_size),
              mock.sentinel.new_path)])
-        mock_detect.return_value.safety_check.assert_called_once_with()
+        if backing_file:
+            mock_detect.return_value.safety_check.assert_called_once_with()
+
+    def test_create_image_qcow2(self):
+        self._test_create_cow_image()
+
+    def test_create_image_backing_file(self):
+        self._test_create_cow_image(
+            backing_file=mock.sentinel.backing_file
+        )
+
+    def test_create_image_base_has_backing_file(self):
+        self.assertRaises(
+            exception.InvalidDiskInfo,
+            self._test_create_cow_image,
+            backing_file={'file': mock.sentinel.backing_file,
+                          'backing_file': mock.sentinel.backing_backing_file},
+        )
+
+    def test_create_image_base_has_data_file(self):
+        self.assertRaises(
+            exception.InvalidDiskInfo,
+            self._test_create_cow_image,
+            backing_file={'file': mock.sentinel.backing_file,
+                          'backing_file': mock.sentinel.backing_backing_file,
+                          'data': {'data-file': mock.sentinel.data_file}},
+        )
+
+    def test_create_image_size_none(self):
+        self._test_create_cow_image(
+            backing_file=mock.sentinel.backing_file,
+        )
+
+    def test_create_image_vmdk(self):
+        self._test_create_cow_image(
+            backing_file={'file': mock.sentinel.backing_file,
+                          'backing_fmt': 'vmdk',
+                          'backing_file': None,
+                          'data': {'create-type': 'monolithicSparse'}}
+        )
+
+    def test_create_image_vmdk_invalid_type(self):
+        self.assertRaises(exception.ImageUnacceptable,
+            self._test_create_cow_image,
+            backing_file={'file': mock.sentinel.backing_file,
+                          'backing_fmt': 'vmdk',
+                          'backing_file': None,
+                          'data': {'create-type': 'monolithicFlat'}}
+        )
 
     @ddt.unpack
     @ddt.data({'fs_type': 'some_fs_type',

nova/tests/unit/virt/test_images.py

@@ -301,6 +301,24 @@ def test_fetch_inspect_ami(self, imginfo, glance):
         # Make sure 'ami was translated into 'raw' before we call qemu-img
         imginfo.assert_called_once_with('/no.path.part', format='raw')
 
+    @mock.patch.object(images, 'IMAGE_API')
+    @mock.patch.object(images, 'qemu_img_info')
+    def test_fetch_inspect_aki(self, imginfo, glance):
+        glance.get.return_value = {'disk_format': 'aki'}
+        self.assertRaises(exception.ImageUnacceptable,
+                          images.fetch_to_raw, None, 'href123', '/no.path')
+        # Make sure 'aki was translated into 'raw' before we call qemu-img
+        imginfo.assert_called_once_with('/no.path.part', format='raw')
+
+    @mock.patch.object(images, 'IMAGE_API')
+    @mock.patch.object(images, 'qemu_img_info')
+    def test_fetch_inspect_ari(self, imginfo, glance):
+        glance.get.return_value = {'disk_format': 'ari'}
+        self.assertRaises(exception.ImageUnacceptable,
+                          images.fetch_to_raw, None, 'href123', '/no.path')
+        # Make sure 'aki was translated into 'raw' before we call qemu-img
+        imginfo.assert_called_once_with('/no.path.part', format='raw')
+
     @mock.patch.object(images, 'IMAGE_API')
     @mock.patch.object(images, 'qemu_img_info')
     def test_fetch_inspect_unknown_format(self, imginfo, glance):

nova/virt/images.py

@@ -160,7 +160,7 @@ def do_image_deep_inspection(img, image_href, path):
         # No inspector was found
         LOG.warning('Unable to perform deep image inspection on type %r',
                     img['disk_format'])
-        if disk_format == 'ami':
+        if disk_format in ('ami', 'aki', 'ari'):
             # A lot of things can be in a UEC, although it is typically a raw
             # filesystem. We really have nothing we can do other than treat it
             # like a 'raw', which is what qemu-img will detect a filesystem as