Stop volume_type_encryption creation when in use

This bug fix addresses bug #1274252.  It requires a volume
type to have no volumes with the type before allowing it
to be made encrypted.  This prevents a situation where
non-encrypted volumes could have an encrypted volume type.
This fix also adds a unit test to confirm functionality.

Change-Id: Iab0954c2d0af73bce28be7fa319b8b34ca20e720
Closes-Bug: #1274252

diff --git a/cinder/api/contrib/volume_type_encryption.py b/cinder/api/contrib/volume_type_encryption.py
index 85f10a1e3d11e76205413b45e94bf152e2d7e4dd..12107cb9578325cce59fe5afea25432329b37497 100644 (file)
--- a/cinder/api/contrib/volume_type_encryption.py
+++ b/cinder/api/contrib/volume_type_encryption.py
@@ -107,6 +107,10 @@ class VolumeTypeEncryptionController(wsgi.Controller):
         context = req.environ['cinder.context']
         authorize(context)
 
+        if self._encrypted_type_in_use(context, type_id):
+            expl = _('Cannot create encryption specs. Volume type in use.')
+            raise webob.exc.HTTPBadRequest(explanation=expl)
+
         if not self.is_valid_body(body, 'encryption'):
             expl = _('Create body is not valid.')
             raise webob.exc.HTTPBadRequest(explanation=expl)

diff --git a/cinder/tests/api/contrib/test_volume_type_encryption.py b/cinder/tests/api/contrib/test_volume_type_encryption.py
index 96ceb24326a258be2b706dab9c8ce6c2d7d0c681..fc6bacc50d89921908cbe5b0439a11b9c1f7975d 100644 (file)
--- a/cinder/tests/api/contrib/test_volume_type_encryption.py
+++ b/cinder/tests/api/contrib/test_volume_type_encryption.py
@@ -313,6 +313,46 @@ class VolumeTypeEncryptionTest(test.TestCase):
         self.assertEqual(expected, res_dict)
         db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
 
+    def test_create_volume_exists(self):
+        # Create the volume type and a volume with the volume type.
+        volume_type = {
+            'id': 'fake_type_id',
+            'name': 'fake_type',
+        }
+        db.volume_type_create(context.get_admin_context(), volume_type)
+        db.volume_create(context.get_admin_context(),
+                         {'id': 'fake_id',
+                          'display_description': 'Test Desc',
+                          'size': 20,
+                          'status': 'creating',
+                          'instance_uuid': None,
+                          'host': 'dummy',
+                          'volume_type_id': volume_type['id']})
+
+        body = {"encryption": {'cipher': 'cipher',
+                               'key_size': 128,
+                               'control_location': 'front-end',
+                               'provider': 'fake_provider',
+                               'volume_type_id': volume_type['id']}}
+
+        # Try to create encryption specs for a volume type
+        # with a volume.
+        res = self._get_response(volume_type, req_method='POST',
+                                 req_body=json.dumps(body),
+                                 req_headers='application/json')
+        res_dict = json.loads(res.body)
+
+        expected = {
+            'badRequest': {
+                'code': 400,
+                'message': ('Cannot create encryption specs. '
+                            'Volume type in use.')
+            }
+        }
+        self.assertEqual(expected, res_dict)
+        db.volume_destroy(context.get_admin_context(), 'fake_id')
+        db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
+
     def _encryption_create_bad_body(self, body,
                                     msg='Create body is not valid.'):
         volume_type = {
@@ -418,29 +458,13 @@ class VolumeTypeEncryptionTest(test.TestCase):
         db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
 
     def test_delete_with_volume_in_use(self):
-        # Create the volume type and volumes with the volume type.
+        # Create the volume type
         volume_type = {
             'id': 'fake_type_id',
             'name': 'fake_type',
         }
         db.volume_type_create(context.get_admin_context(), volume_type)
-        db.volume_create(context.get_admin_context(),
-                         {'id': 'fake_id',
-                          'display_description': 'Test Desc',
-                          'size': 20,
-                          'status': 'creating',
-                          'instance_uuid': None,
-                          'host': 'dummy',
-                          'volume_type_id': volume_type['id']})
 
-        db.volume_create(context.get_admin_context(),
-                         {'id': 'fake_id2',
-                          'display_description': 'Test Desc2',
-                          'size': 2,
-                          'status': 'creating',
-                          'instance_uuid': None,
-                          'host': 'dummy',
-                          'volume_type_id': volume_type['id']})
         body = {"encryption": {'cipher': 'cipher',
                                'key_size': 128,
                                'control_location': 'front-end',
@@ -458,6 +482,25 @@ class VolumeTypeEncryptionTest(test.TestCase):
         res_dict = json.loads(res.body)
         self.assertEqual(volume_type['id'], res_dict['volume_type_id'])
 
+        # Create volumes with the volume type
+        db.volume_create(context.get_admin_context(),
+                         {'id': 'fake_id',
+                          'display_description': 'Test Desc',
+                          'size': 20,
+                          'status': 'creating',
+                          'instance_uuid': None,
+                          'host': 'dummy',
+                          'volume_type_id': volume_type['id']})
+
+        db.volume_create(context.get_admin_context(),
+                         {'id': 'fake_id2',
+                          'display_description': 'Test Desc2',
+                          'size': 2,
+                          'status': 'creating',
+                          'instance_uuid': None,
+                          'host': 'dummy',
+                          'volume_type_id': volume_type['id']})
+
         # Delete, and test that there is an error since volumes exist
         res = self._get_response(volume_type, req_method='DELETE',
                                  req_headers='application/json',