Add exception SnapshotIsBusy to be handled as VolumeIsBusy.

This is helpful when driver can not delete snapshot if it has volumes
created from it.
Add tests for both VolumeIsBusy and SnapshotIsBusy cases.

Change-Id: I3fcc6b1ffdfbc7e22ed30ac832db500eb3260d04

nova/exception.py

@@ -447,6 +447,11 @@ class VolumeIsBusy(NovaException):
     message = _("deleting volume %(volume_name)s that has snapshot")
 
 
+class SnapshotIsBusy(NovaException):
+    message = _("deleting snapshot %(snapshot_name)s that has "
+                "dependent volumes")
+
+
 class ISCSITargetNotFoundForVolume(NotFound):
     message = _("No target id found for volume %(volume_id)s.")
 

nova/tests/test_volume.py

@@ -22,6 +22,8 @@ Tests for Volume Code.
 
 import cStringIO
 
+import mox
+
 from nova import context
 from nova import exception
 from nova import db
@@ -79,6 +81,25 @@ class VolumeTestCase(test.TestCase):
                           self.context,
                           volume_id)
 
+    def test_delete_busy_volume(self):
+        """Test volume survives deletion if driver reports it as busy."""
+        volume = self._create_volume()
+        volume_id = volume['id']
+        self.volume.create_volume(self.context, volume_id)
+
+        self.mox.StubOutWithMock(self.volume.driver, 'delete_volume')
+        self.volume.driver.delete_volume(mox.IgnoreArg()) \
+                                              .AndRaise(exception.VolumeIsBusy)
+        self.mox.ReplayAll()
+        res = self.volume.delete_volume(self.context, volume_id)
+        self.assertEqual(True, res)
+        volume_ref = db.volume_get(context.get_admin_context(), volume_id)
+        self.assertEqual(volume_id, volume_ref.id)
+        self.assertEqual("available", volume_ref.status)
+
+        self.mox.UnsetStubs()
+        self.volume.delete_volume(self.context, volume_id)
+
     def test_create_volume_from_snapshot(self):
         """Test volume can be created from a snapshot."""
         volume_src = self._create_volume()
@@ -260,6 +281,27 @@ class VolumeTestCase(test.TestCase):
         db.snapshot_destroy(self.context, snapshot_ref['id'])
         db.volume_destroy(self.context, volume['id'])
 
+    def test_delete_busy_snapshot(self):
+        """Test snapshot can be created and deleted."""
+        volume = self._create_volume()
+        volume_id = volume['id']
+        self.volume.create_volume(self.context, volume_id)
+        snapshot_id = self._create_snapshot(volume_id)
+        self.volume.create_snapshot(self.context, volume_id, snapshot_id)
+
+        self.mox.StubOutWithMock(self.volume.driver, 'delete_snapshot')
+        self.volume.driver.delete_snapshot(mox.IgnoreArg()) \
+                                            .AndRaise(exception.SnapshotIsBusy)
+        self.mox.ReplayAll()
+        self.volume.delete_snapshot(self.context, snapshot_id)
+        snapshot_ref = db.snapshot_get(self.context, snapshot_id)
+        self.assertEqual(snapshot_id, snapshot_ref.id)
+        self.assertEqual("available", snapshot_ref.status)
+
+        self.mox.UnsetStubs()
+        self.volume.delete_snapshot(self.context, snapshot_id)
+        self.volume.delete_volume(self.context, volume_id)
+
 
 class DriverTestCase(test.TestCase):
     """Base Test class for Drivers."""