Restrict Volume type deletion with volumes assoc

Updated volume_type_destroy method to throw exception
for volume type delete with associated volumes.

Updated volume_type unit tests

Closes-Bug: #1215329

tag:doc-impact

Change-Id: I7a5d4b473588757d21b461337df493e8046e1d09

cinder/api/contrib/types_manage.py

@@ -97,6 +97,13 @@ def _delete(self, req, id):
             notifier_api.notify(context, 'volumeType',
                                 'volume_type.delete',
                                 notifier_api.INFO, notifier_info)
+        except exception.VolumeTypeInUse as err:
+            notifier_err = dict(id=id, error_message=str(err))
+            self._notify_voloume_type_error(context,
+                                            'volume_type.delete',
+                                            notifier_err)
+            msg = 'Target volume type is still in use.'
+            raise webob.exc.HTTPBadRequest(explanation=msg)
         except exception.NotFound as err:
             notifier_err = dict(id=id, error_message=str(err))
             self._notify_voloume_type_error(context,

cinder/db/sqlalchemy/api.py

@@ -1884,7 +1884,12 @@ def volume_type_destroy(context, id):
     session = get_session()
     with session.begin():
         _volume_type_get(context, id, session)
-
+        results = model_query(context, models.Volume, session=session). \
+            filter_by(volume_type_id=id).all()
+        if results:
+            msg = _('VolumeType %s deletion failed, VolumeType in use.') % id
+            LOG.error(msg)
+            raise exception.VolumeTypeInUse(volume_type_id=id)
         session.query(models.VolumeTypes).\
             filter_by(id=id).\
             update({'deleted': True,

cinder/exception.py

@@ -274,6 +274,11 @@ class VolumeTypeExtraSpecsNotFound(NotFound):
                 "key %(extra_specs_key)s.")
 
 
+class VolumeTypeInUse(CinderException):
+    message = _("Volume Type %(volume_type_id)s deletion is not allowed with "
+                "volumes present with the type.")
+
+
 class SnapshotNotFound(NotFound):
     message = _("Snapshot %(snapshot_id)s could not be found.")
 

cinder/tests/api/contrib/test_types_manage.py

@@ -45,6 +45,12 @@ def return_volume_types_destroy(context, name):
     pass
 
 
+def return_volume_types_with_volumes_destroy(context, id):
+    if id == "1":
+        raise exception.VolumeTypeInUse(volume_type_id=id)
+    pass
+
+
 def return_volume_types_create(context, name, specs):
     pass
 
@@ -93,6 +99,16 @@ def test_volume_types_delete_not_found(self):
                           req, '777')
         self.assertEquals(len(test_notifier.NOTIFICATIONS), 1)
 
+    def test_volume_types_with_volumes_destroy(self):
+        self.stubs.Set(volume_types, 'get_volume_type',
+                       return_volume_types_get_volume_type)
+        self.stubs.Set(volume_types, 'destroy',
+                       return_volume_types_with_volumes_destroy)
+        req = fakes.HTTPRequest.blank('/v2/fake/types/1')
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 0)
+        self.controller._delete(req, 1)
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 1)
+
     def test_create(self):
         self.stubs.Set(volume_types, 'create',
                        return_volume_types_create)

cinder/tests/test_volume_types.py

@@ -18,9 +18,11 @@
 """
 
 
+import datetime
 import time
 
 from cinder import context
+from cinder import db
 from cinder.db.sqlalchemy import api as db_api
 from cinder.db.sqlalchemy import models
 from cinder import exception
@@ -104,6 +106,19 @@ def test_non_existent_vol_type_shouldnt_delete(self):
         self.assertRaises(exception.VolumeTypeNotFound,
                           volume_types.destroy, self.ctxt, "sfsfsdfdfs")
 
+    def test_volume_type_with_volumes_shouldnt_delete(self):
+        """Ensures volume type deletion with associated volumes fail."""
+        type_ref = volume_types.create(self.ctxt, self.vol_type1_name)
+        db.volume_create(self.ctxt,
+                         {'id': '1',
+                          'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1),
+                          'display_description': 'Test Desc',
+                          'size': 20,
+                          'status': 'available',
+                          'volume_type_id': type_ref['id']})
+        self.assertRaises(exception.VolumeTypeInUse,
+                          volume_types.destroy, self.ctxt, type_ref['id'])
+
     def test_repeated_vol_types_shouldnt_raise(self):
         """Ensures that volume duplicates don't raise."""
         new_name = self.vol_type1_name + "dup"