openstack
/
deb-cinder
Code Issues Proposed changes

Merge "VMware: Allow deletion of snapshot in error state"

This commit is contained in:
Jenkins 2017-01-23 23:36:01 +00:00 committed by Gerrit Code Review
parent 9b450ac24b 9a889a90f0
commit 964a8d2914
2 changed files with 45 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
cinder/tests/unit/volume/drivers/vmware/test_vmware_vmdk.py
View File

@ -28,6 +28,7 @@ import six
from cinder import context
from cinder import exception as cinder_exceptions
from cinder import test
from cinder.tests.unit import fake_snapshot
from cinder.tests.unit import fake_volume
from cinder.volume import configuration
from cinder.volume.drivers.vmware import datastore as hub
@ -261,11 +262,13 @@ class VMwareVcVmdkDriverTestCase(test.TestCase):
vops.get_backing.return_value = None
volume = self._create_volume_dict()
snapshot = self._create_snapshot_dict(volume)
snapshot = fake_snapshot.fake_snapshot_obj(self._context,
volume=volume)
self._driver.delete_snapshot(snapshot)
vops.get_backing.assert_called_once_with(snapshot['volume_name'])
self.assertFalse(vops.delete_snapshot.called)
vops.get_backing.assert_called_once_with(snapshot.volume_name)
vops.get_snapshot.assert_not_called()
vops.delete_snapshot.assert_not_called()
@mock.patch.object(VMDK_DRIVER, 'volumeops')
def test_delete_snapshot_with_backing(self, vops):
@ -273,20 +276,40 @@ class VMwareVcVmdkDriverTestCase(test.TestCase):
vops.get_backing.return_value = backing
volume = self._create_volume_dict()
snapshot = self._create_snapshot_dict(volume)
snapshot = fake_snapshot.fake_snapshot_obj(self._context,
volume=volume)
self._driver.delete_snapshot(snapshot)
vops.get_backing.assert_called_once_with(snapshot['volume_name'])
vops.get_backing.assert_called_once_with(snapshot.volume_name)
vops.get_snapshot.assert_called_once_with(backing, snapshot.name)
vops.delete_snapshot.assert_called_once_with(
backing, snapshot['name'])
backing, snapshot.name)
def test_delete_snapshot_when_attached(self):
@mock.patch.object(VMDK_DRIVER, 'volumeops')
def test_delete_snapshot_when_attached(self, vops):
volume = self._create_volume_dict(status='in-use')
snapshot = self._create_snapshot_dict(volume)
snapshot = fake_snapshot.fake_snapshot_obj(self._context,
volume=volume)
self.assertRaises(cinder_exceptions.InvalidVolume,
self.assertRaises(cinder_exceptions.InvalidSnapshot,
self._driver.delete_snapshot, snapshot)
@mock.patch.object(VMDK_DRIVER, 'volumeops')
def test_delete_snapshot_without_backend_snapshot(self, vops):
backing = mock.sentinel.backing
vops.get_backing.return_value = backing
vops.get_snapshot.return_value = None
volume = self._create_volume_dict(status='in-use')
snapshot = fake_snapshot.fake_snapshot_obj(self._context,
volume=volume)
self._driver.delete_snapshot(snapshot)
vops.get_backing.assert_called_once_with(snapshot.volume_name)
vops.get_snapshot.assert_called_once_with(backing, snapshot.name)
vops.delete_snapshot.assert_not_called()
@ddt.data('vmdk', 'VMDK', None)
def test_validate_disk_format(self, disk_format):
self._driver._validate_disk_format(disk_format)

28
cinder/volume/drivers/vmware/vmdk.py
View File

@ -670,26 +670,24 @@ class VMwareVcVmdkDriver(driver.VolumeDriver):
"""Delete snapshot.
If the volume does not have a backing or the snapshot does not exist
then simply pass, else delete the snapshot.
Snapshot deletion of only available volume is supported.
then simply pass, else delete the snapshot. Snapshot deletion of only
available volume is supported.
:param snapshot: Snapshot object
"""
volume = snapshot['volume']
if volume['status'] != 'available':
msg = _("Delete snapshot of volume not supported in "
"state: %s.") % volume['status']
LOG.error(msg)
raise exception.InvalidVolume(msg)
backing = self.volumeops.get_backing(snapshot['volume_name'])
backing = self.volumeops.get_backing(snapshot.volume_name)
if not backing:
LOG.info(_LI("There is no backing, and so there is no "
"snapshot: %s."), snapshot['name'])
LOG.debug("Backing does not exist for volume.",
resource=snapshot.volume)
elif not self.volumeops.get_snapshot(backing, snapshot.name):
LOG.debug("Snapshot does not exist in backend.", resource=snapshot)
elif snapshot.volume.status != 'available':
msg = _("Delete snapshot of volume not supported in "
"state: %s.") % snapshot.volume.status
LOG.error(msg)
raise exception.InvalidSnapshot(reason=msg)
else:
self.volumeops.delete_snapshot(backing, snapshot['name'])
LOG.info(_LI("Successfully deleted snapshot: %s."),
snapshot['name'])
self.volumeops.delete_snapshot(backing, snapshot.name)
def delete_snapshot(self, snapshot):
"""Delete snapshot.