Merge "Add exception SnapshotIsBusy to be handled as VolumeIsBusy."

nova/exception.py

@@ -451,6 +451,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."""