Merge pull request #148 from stackhpc/upstream/2025.1-2026-06-22

Synchronise 2025.1 with upstream

Author: priteau

cinder/tests/unit/volume/drivers/test_quobyte.py

@@ -1021,7 +1021,9 @@ def test_copy_volume_from_snapshot(self):
         self.mock_object(image_utils, 'qemu_img_info', return_value=img_info)
         drv._set_rw_permissions = mock.Mock()
 
-        drv._copy_volume_from_snapshot(snapshot, dest_volume, size)
+        drv._copy_volume_from_snapshot(snapshot, dest_volume, size,
+                                       src_encryption_key_id=None,
+                                       new_encryption_key_id=None)
 
         drv._read_info_file.assert_called_once_with(info_path)
         image_utils.qemu_img_info.assert_called_once_with(
@@ -1074,14 +1076,17 @@ def test_copy_volume_from_snapshot_cached(self, os_ac_mock,
 
         # mocking and testing starts here
         mock_convert = self.mock_object(image_utils, 'convert_image')
-        drv._read_info_file = mock.Mock(return_value=
-                                        {'active': snap_file,
-                                         snapshot['id']: snap_file})
+        drv._read_info_file = mock.Mock(return_value={
+            'active': snap_file,
+            snapshot['id']: snap_file
+        })
         self.mock_object(image_utils, 'qemu_img_info', return_value=img_info)
         drv._set_rw_permissions = mock.Mock()
         shutil.copyfile = mock.Mock()
 
-        drv._copy_volume_from_snapshot(snapshot, dest_volume, size)
+        drv._copy_volume_from_snapshot(snapshot, dest_volume, size,
+                                       src_encryption_key_id=None,
+                                       new_encryption_key_id=None)
 
         drv._read_info_file.assert_called_once_with(info_path)
         image_utils.qemu_img_info.assert_called_once_with(
@@ -1138,14 +1143,17 @@ def test_copy_volume_from_snapshot_not_cached_overlay(self, os_ac_mock,
 
         # mocking and testing starts here
         mock_convert = self.mock_object(image_utils, 'convert_image')
-        drv._read_info_file = mock.Mock(return_value=
-                                        {'active': snap_file,
-                                         snapshot['id']: snap_file})
+        drv._read_info_file = mock.Mock(return_value={
+            'active': snap_file,
+            snapshot['id']: snap_file
+        })
         self.mock_object(image_utils, 'qemu_img_info', return_value=img_info)
         drv._set_rw_permissions = mock.Mock()
         drv._create_overlay_volume_from_snapshot = mock.Mock()
 
-        drv._copy_volume_from_snapshot(snapshot, dest_volume, size)
+        drv._copy_volume_from_snapshot(snapshot, dest_volume, size,
+                                       src_encryption_key_id=None,
+                                       new_encryption_key_id=None)
 
         drv._read_info_file.assert_called_once_with(info_path)
         os_ac_mock.assert_called_once_with(
@@ -1212,7 +1220,9 @@ def test_copy_volume_from_snapshot_not_cached(self, qb_falloc_mock):
         drv._set_rw_permissions = mock.Mock()
         self.mock_object(shutil, 'copyfile')
 
-        drv._copy_volume_from_snapshot(snapshot, dest_volume, size)
+        drv._copy_volume_from_snapshot(snapshot, dest_volume, size,
+                                       src_encryption_key_id=None,
+                                       new_encryption_key_id=None)
 
         drv._read_info_file.assert_called_once_with(info_path)
         image_utils.qemu_img_info.assert_called_once_with(
@@ -1229,6 +1239,25 @@ def test_copy_volume_from_snapshot_not_cached(self, qb_falloc_mock):
         shutil.copyfile.assert_called_once_with(cache_path, dest_vol_path)
         drv._set_rw_permissions.assert_called_once_with(dest_vol_path)
 
+    def test_copy_volume_from_snapshot_with_encr(self):
+        # setup vars
+        drv = self._driver
+        src_volume = self._simple_volume()
+        snapshot = self._get_fake_snapshot(src_volume)
+        dest_volume = self._simple_volume(
+            id='c1073000-0000-0000-0000-0000000c1073')
+        size = dest_volume['size']
+
+        # run test
+        self.assertRaises(exception.NotSupportedOperation,
+                          drv._copy_volume_from_snapshot,
+                          snapshot,
+                          dest_volume,
+                          size,
+                          src_encryption_key_id=mock.sentinel.src_key,
+                          new_encryption_key_id=mock.sentinel.dest_key
+                          )
+
     @ddt.data(['available', True], ['backing-up', True],
               ['creating', False], ['deleting', False])
     @ddt.unpack

cinder/volume/drivers/quobyte.py

@@ -38,7 +38,7 @@
 from cinder.volume import configuration
 from cinder.volume.drivers import remotefs as remotefs_drv
 
-VERSION = '1.1.13'
+VERSION = '1.1.14'
 
 LOG = logging.getLogger(__name__)
 
@@ -119,6 +119,7 @@ class QuobyteDriver(remotefs_drv.RemoteFSSnapDriverDistributed):
         1.1.11 - NAS secure ownership & permissions are now False by default
         1.1.12 - Ensure the currently configured volume url is always used
         1.1.13 - Allow creating volumes from snapshots in state 'backing-up'
+        1.1.14 - Fixes regression from encryption being added in parent class
 
     """
 
@@ -383,7 +384,9 @@ def create_volume_from_snapshot(self, volume, snapshot):
         return self._create_volume_from_snapshot(volume, snapshot)
 
     @coordination.synchronized('{self.driver_prefix}-{volume.id}')
-    def _copy_volume_from_snapshot(self, snapshot, volume, volume_size):
+    def _copy_volume_from_snapshot(self, snapshot, volume, volume_size,
+                                   src_encryption_key_id=None,
+                                   new_encryption_key_id=None):
         """Copy data from snapshot to destination volume.
 
         This is done with a qemu-img convert to raw/qcow2 from the snapshot
@@ -392,6 +395,12 @@ def _copy_volume_from_snapshot(self, snapshot, volume, volume_size):
         snapshot id are created directly from the cache.
         """
 
+        if new_encryption_key_id:
+            msg = _("Encryption key %s was requested. Volume "
+                    "encryption is not supported.")
+            raise exception.NotSupportedOperation(
+                message=msg % new_encryption_key_id)
+
         LOG.debug("snapshot: %(snap)s, volume: %(vol)s, ",
                   {'snap': snapshot.id,
                    'vol': volume.id,