Merge "Prohibit creating volume from source with dif encryptions"

cinder/tests/unit/test_volume.py

@@ -1512,6 +1512,54 @@ class VolumeTestCase(base.BaseVolumeTestCase):
                               volume_type=foo_type,
                               snapshot=snapshot_obj)
 
+    def _test_create_from_source_snapshot_encryptions(
+            self, is_snapshot=False):
+        volume_api = cinder.volume.api.API()
+        foo_type = {
+            'name': 'foo',
+            'extra_specs': {'volume_backend_name': 'dev_1'},
+            'id': fake.VOLUME_TYPE_ID,
+            'description': None}
+
+        biz_type = {
+            'name': 'foo',
+            'extra_specs': {'volume_backend_name': 'dev_1'},
+            'id': fake.VOLUME_TYPE2_ID,
+            'description': None}
+        source_vol = {'id': fake.VOLUME_ID,
+                      'status': 'available',
+                      'volume_size': 1,
+                      'volume_type': biz_type,
+                      'volume_type_id': biz_type['id']}
+
+        snapshot = {'id': fake.SNAPSHOT_ID,
+                    'status': fields.SnapshotStatus.AVAILABLE,
+                    'volume_size': 1,
+                    'volume_type_id': biz_type['id']}
+        snapshot_obj = fake_snapshot.fake_snapshot_obj(self.context,
+                                                       **snapshot)
+
+        with mock.patch.object(
+                cinder.volume.volume_types,
+                'volume_types_encryption_changed') as mock_encryption_changed:
+            mock_encryption_changed.return_value = True
+            self.assertRaises(exception.InvalidInput,
+                              volume_api.create,
+                              self.context,
+                              size=1,
+                              name='fake_name',
+                              description='fake_desc',
+                              volume_type=foo_type,
+                              source_volume=(
+                                  source_vol if not is_snapshot else None),
+                              snapshot=snapshot_obj if is_snapshot else None)
+
+    def test_create_from_source_encryption_changed(self):
+        self._test_create_from_source_snapshot_encryptions()
+
+    def test_create_from_snapshot_encryption_changed(self):
+        self._test_create_from_source_snapshot_encryptions(is_snapshot=True)
+
     def test_create_snapshot_driver_not_initialized(self):
         volume_src = tests_utils.create_volume(self.context,
                                                **self.volume_params)

cinder/volume/api.py

@@ -174,10 +174,14 @@ class API(base.Base):
         return tuple(azs)
 
     def _retype_is_possible(self, context,
-                            first_type_id, second_type_id,
+                            source_type_id, target_type_id):
-                            first_type=None, second_type=None):
         safe = False
         elevated = context.elevated()
+        # If encryptions are different, it is not allowed
+        # to create volume from source volume or snapshot.
+        if volume_types.volume_types_encryption_changed(
+                elevated, source_type_id, target_type_id):
+            return False
         services = objects.ServiceList.get_all_by_topic(
             elevated,
             constants.VOLUME_TOPIC,
@@ -185,14 +189,14 @@ class API(base.Base):
         if len(services.objects) == 1:
             safe = True
         else:
-            type_a = first_type or volume_types.get_volume_type(
+            source_type = volume_types.get_volume_type(
                 elevated,
-                first_type_id)
+                source_type_id)
-            type_b = second_type or volume_types.get_volume_type(
+            target_type = volume_types.get_volume_type(
                 elevated,
-                second_type_id)
+                target_type_id)
-            if (volume_utils.matching_backend_name(type_a['extra_specs'],
+            if (volume_utils.matching_backend_name(
-                                                   type_b['extra_specs'])):
+                    source_type['extra_specs'], target_type['extra_specs'])):
                 safe = True
         return safe
 
@@ -266,9 +270,8 @@ class API(base.Base):
             if volume_type['id'] != source_volume['volume_type_id']:
                 if not self._retype_is_possible(
                         context,
-                        volume_type['id'],
                         source_volume['volume_type_id'],
-                        volume_type):
+                        volume_type['id']):
                     msg = _("Invalid volume_type provided: %s (requested type "
                             "is not compatible; either match source volume, "
                             "or omit type argument).") % volume_type['id']
@@ -283,9 +286,8 @@ class API(base.Base):
         if snapshot and volume_type:
             if volume_type['id'] != snapshot.volume_type_id:
                 if not self._retype_is_possible(context,
-                                                volume_type['id'],
                                                 snapshot.volume_type_id,
-                                                volume_type):
+                                                volume_type['id']):
                     msg = _("Invalid volume_type provided: %s (requested "
                             "type is not compatible; recommend omitting "
                             "the type argument).") % volume_type['id']