Add delete support for volume type encryption

This modification adds delete support to the volume type
encryption API extension.  The delete operation is not
permitted when the encrypted volume type is in use by an
existing volume.

Change-Id: Ic490fd3f0f1b1282cd24c65886c92beb88d30560
Implements: blueprint encrypt-cinder-volumes

diff --git a/cinder/api/contrib/volume_type_encryption.py b/cinder/api/contrib/volume_type_encryption.py
index a403b2dc7d5345b795983875d74e16b04765f621..e6d28f07a254e19a364ed8818adf266926582cc1 100644 (file)
--- a/cinder/api/contrib/volume_type_encryption.py
+++ b/cinder/api/contrib/volume_type_encryption.py
@@ -85,6 +85,16 @@ class VolumeTypeEncryptionController(wsgi.Controller):
                 msg = _("Valid control location are: %s") % CONTROL_LOCATION
                 raise exception.InvalidInput(reason=msg)
 
+    def _encrypted_type_in_use(self, context, volume_type_id):
+        volume_list = db.volume_type_encryption_volume_get(context,
+                                                           volume_type_id)
+        # If there is at least one volume in the list
+        # returned, this type is in use by a volume.
+        if len(volume_list) > 0:
+            return True
+        else:
+            return False
+
     @wsgi.serializers(xml=VolumeTypeEncryptionTemplate)
     def index(self, req, type_id):
         """Returns the encryption specs for a given volume type."""
@@ -136,6 +146,19 @@ class VolumeTypeEncryptionController(wsgi.Controller):
 
         return {id: encryption_specs[id]}
 
+    def delete(self, req, type_id, id):
+        """Delete encryption specs for a given volume type."""
+        context = req.environ['cinder.context']
+        authorize(context)
+
+        if self._encrypted_type_in_use(context, type_id):
+            expl = _('Cannot delete encryption specs. Volume type in use.')
+            raise webob.exc.HTTPBadRequest(explanation=expl)
+        else:
+            db.volume_type_encryption_delete(context, type_id)
+
+        return webob.Response(status_int=202)
+
 
 class Volume_type_encryption(extensions.ExtensionDescriptor):
     """Encryption support for volume types."""

diff --git a/cinder/tests/api/contrib/test_volume_type_encryption.py b/cinder/tests/api/contrib/test_volume_type_encryption.py
index b6b87a57501444d6253a7bedca1672cb4be963a5..058f87b82b439670d217207a7c6a2c4192774286 100644 (file)
--- a/cinder/tests/api/contrib/test_volume_type_encryption.py
+++ b/cinder/tests/api/contrib/test_volume_type_encryption.py
@@ -376,3 +376,127 @@ class VolumeTypeEncryptionTest(test.TestCase):
                                'volume_type_id': 'volume_type'}}
         msg = ("Invalid input received: provider must be defined")
         self._encryption_create_bad_body(body=body, msg=msg)
+
+    def test_delete(self):
+        volume_type = {
+            'id': 'fake_type_id',
+            'name': 'fake_type',
+        }
+        db.volume_type_create(context.get_admin_context(), volume_type)
+
+        # Test that before create, there's nothing with a get
+        res = self._get_response(volume_type, req_method='GET',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption')
+        self.assertEqual(200, res.status_code)
+        res_dict = json.loads(res.body)
+        self.assertEqual({}, res_dict)
+
+        body = {"encryption": {'cipher': 'cipher',
+                               'key_size': 128,
+                               'control_location': 'front-end',
+                               'provider': 'fake_provider',
+                               'volume_type_id': volume_type['id']}}
+
+        # Create, and test that get returns something
+        res = self._get_response(volume_type, req_method='POST',
+                                 req_body=json.dumps(body),
+                                 req_headers='application/json')
+        res_dict = json.loads(res.body)
+
+        res = self._get_response(volume_type, req_method='GET',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption')
+        self.assertEqual(200, res.status_code)
+        res_dict = json.loads(res.body)
+        self.assertEqual(volume_type['id'], res_dict['volume_type_id'])
+
+        # Delete, and test that get returns nothing
+        res = self._get_response(volume_type, req_method='DELETE',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption/provider')
+        self.assertEqual(202, res.status_code)
+        self.assertEqual(0, len(res.body))
+        res = self._get_response(volume_type, req_method='GET',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption')
+        self.assertEqual(200, res.status_code)
+        res_dict = json.loads(res.body)
+        self.assertEqual({}, res_dict)
+
+        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.
+        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',
+                               'provider': 'fake_provider',
+                               'volume_type_id': volume_type['id']}}
+
+        # Create encryption with volume type, and test with GET
+        res = self._get_response(volume_type, req_method='POST',
+                                 req_body=json.dumps(body),
+                                 req_headers='application/json')
+        res = self._get_response(volume_type, req_method='GET',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption')
+        self.assertEqual(200, res.status_code)
+        res_dict = json.loads(res.body)
+        self.assertEqual(volume_type['id'], res_dict['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',
+                                 url='/v2/fake/types/%s/encryption/provider')
+        self.assertEqual(400, res.status_code)
+        res_dict = json.loads(res.body)
+        expected = {
+            'badRequest': {
+                'code': 400,
+                'message': 'Cannot delete encryption specs. '
+                           'Volume type in use.'
+            }
+        }
+        self.assertEqual(expected, res_dict)
+
+        # Delete the volumes
+        db.volume_destroy(context.get_admin_context(), 'fake_id')
+        db.volume_destroy(context.get_admin_context(), 'fake_id2')
+
+        # Delete, and test that get returns nothing
+        res = self._get_response(volume_type, req_method='DELETE',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption/provider')
+        self.assertEqual(202, res.status_code)
+        self.assertEqual(0, len(res.body))
+        res = self._get_response(volume_type, req_method='GET',
+                                 req_headers='application/json',
+                                 url='/v2/fake/types/%s/encryption')
+        self.assertEqual(200, res.status_code)
+        res_dict = json.loads(res.body)
+        self.assertEqual({}, res_dict)
+
+        db.volume_type_destroy(context.get_admin_context(), volume_type['id'])