openstack/deb-manila
Code Issues Proposed changes

Fix force-delete on snapshot resource

Browse Source 
When using snapshot-delete API, if the share driver
logs an exception, the status of the snapshot is set
to 'error_deleting'. If an administrator wants to remove
this snapshot, he/she would expect to use the
snapshot-force-delete API. However, if the share
driver still cannot delete the resource, it will remain
in 'error_deleting' state; leaving the administrator
with no choice but to resort to making changes to
the snapshot record in Manila's database.

Propagate the force flag to the share manager and log
driver exceptions, while ignoring them to obliterate
the snapshot on Manila's database if the delete
operation is forced.

Also fix data being sent in create/delete snapshot
driver interfaces and refactor corresponding unit tests.

Change-Id: Ic2494e1f9eb9104f96ef3a50f12d0f304735ee4f
Closes-Bug: #1560119
This commit is contained in:
Goutham Pacha Ravi 2016-03-21 13:28:16 -04:00
parent 9e06481ed5
commit 35648819aa
8 changed files with 342 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
manila/share/api.py
View File

@ -1039,8 +1039,8 @@ class API(base.Base):
context, snapshot, share['instance']['host'],
share_id=share['id'], force=force)
else:
self.share_rpcapi.delete_snapshot(context, snapshot,
share['instance']['host'])
self.share_rpcapi.delete_snapshot(
context, snapshot, share['instance']['host'], force=force)
@policy.wrap_check_policy('share')
def update(self, context, share, fields):

63
manila/share/manager.py
View File

@ -1882,13 +1882,14 @@ class ShareManager(manager.SchedulerDependentManager):
)
snapshot_instance_id = snapshot_instance['id']
try:
model_update = self.driver.create_snapshot(
context, snapshot_instance, share_server=share_server)
snapshot_instance = self._get_snapshot_instance_dict(
context, snapshot_instance)
try:
model_update = self.driver.create_snapshot(
context, snapshot_instance, share_server=share_server) or {}
if model_update:
self.db.share_snapshot_instance_update(
context, snapshot_instance_id, model_update)
except Exception:
with excutils.save_and_reraise_exception():
self.db.share_snapshot_instance_update(
@ -1896,17 +1897,16 @@ class ShareManager(manager.SchedulerDependentManager):
snapshot_instance_id,
{'status': constants.STATUS_ERROR})
if model_update.get('status') in (None, constants.STATUS_AVAILABLE):
model_update['status'] = constants.STATUS_AVAILABLE
model_update['progress'] = '100%'
self.db.share_snapshot_instance_update(
context,
snapshot_instance_id,
{'status': constants.STATUS_AVAILABLE,
'progress': '100%'}
)
return snapshot_id
context, snapshot_instance_id, model_update)
@add_hooks
@utils.require_driver_initialized
def delete_snapshot(self, context, snapshot_id):
def delete_snapshot(self, context, snapshot_id, force=False):
"""Delete share snapshot."""
context = context.elevated()
snapshot_ref = self.db.share_snapshot_get(context, snapshot_id)
@ -1914,8 +1914,7 @@ class ShareManager(manager.SchedulerDependentManager):
share_server = self._get_share_server(
context, snapshot_ref['share']['instance'])
snapshot_instance = self.db.share_snapshot_instance_get(
context, snapshot_ref.instance['id'], with_share_data=True
)
context, snapshot_ref.instance['id'], with_share_data=True)
snapshot_instance_id = snapshot_instance['id']
if context.project_id != snapshot_ref['project_id']:
@ -1923,23 +1922,46 @@ class ShareManager(manager.SchedulerDependentManager):
else:
project_id = context.project_id
snapshot_instance = self._get_snapshot_instance_dict(
context, snapshot_instance)
try:
self.driver.delete_snapshot(context, snapshot_instance,
share_server=share_server)
except exception.ShareSnapshotIsBusy:
with excutils.save_and_reraise_exception() as exc:
if force:
msg = _("The driver reported that the snapshot %s "
"was busy on the backend. Since this "
"operation was forced, the snapshot will "
"be deleted from Manila's database. A "
"cleanup on the backend may be necessary.")
LOG.exception(msg, snapshot_id)
exc.reraise = False
else:
self.db.share_snapshot_instance_update(
context,
snapshot_instance_id,
{'status': constants.STATUS_AVAILABLE})
except Exception:
with excutils.save_and_reraise_exception():
with excutils.save_and_reraise_exception() as exc:
if force:
msg = _("The driver was unable to delete the "
"snapshot %s on the backend. Since this "
"operation is forced, the snapshot will "
"be deleted from Manila's database. A cleanup on "
"the backend may be necessary.")
LOG.exception(msg, snapshot_id)
exc.reraise = False
else:
self.db.share_snapshot_instance_update(
context,
snapshot_instance_id,
{'status': constants.STATUS_ERROR_DELETING})
else:
self.db.share_snapshot_instance_delete(
context, snapshot_instance_id)
self.db.share_snapshot_instance_delete(context, snapshot_instance_id)
try:
reservations = QUOTAS.reserve(
context, project_id=project_id, snapshots=-1,
@ -1947,7 +1969,8 @@ class ShareManager(manager.SchedulerDependentManager):
user_id=snapshot_ref['user_id'])
except Exception:
reservations = None
LOG.exception(_LE("Failed to update usages deleting snapshot"))
LOG.exception(_LE("Failed to update quota usages while deleting "
"snapshot %s."), snapshot_id)
if reservations:
QUOTAS.commit(context, reservations, project_id=project_id,

5
manila/share/rpcapi.py
View File

@ -161,12 +161,13 @@ class ShareAPI(object):
share_id=share['id'],
snapshot_id=snapshot['id'])
def delete_snapshot(self, context, snapshot, host):
def delete_snapshot(self, context, snapshot, host, force=False):
new_host = utils.extract_host(host)
call_context = self.client.prepare(server=new_host)
call_context.cast(context,
'delete_snapshot',
snapshot_id=snapshot['id'])
snapshot_id=snapshot['id'],
force=force)
def create_replicated_snapshot(self, context, share, replicated_snapshot):
host = utils.extract_host(share['instance']['host'])

11
manila/tests/fake_share.py
View File

@ -89,7 +89,8 @@ def fake_snapshot(create_instance=False, **kwargs):
instance_keys = ('instance_id', 'snapshot_id', 'share_instance_id',
'status', 'progress', 'provider_location')
snapshot_keys = ('id', 'share_name', 'share_id', 'name', 'share_size',
'share_proto', 'instance', 'aggregate_status')
'share_proto', 'instance', 'aggregate_status', 'share',
'project_id', 'size')
instance_kwargs = {k: kwargs.get(k) for k in instance_keys if k in kwargs}
snapshot_kwargs = {k: kwargs.get(k) for k in snapshot_keys if k in kwargs}
@ -108,6 +109,9 @@ def fake_snapshot(create_instance=False, **kwargs):
'instance': None,
'share': 'fake_share',
'aggregate_status': aggregate_status,
'project_id': 'fakeprojectid',
'size': 1,
'user_id': 'xyzzy',
}
snapshot.update(snapshot_kwargs)
if create_instance:
@ -141,6 +145,11 @@ def fake_snapshot_instance(base_snapshot=None, **kwargs):
'share_name': 'fakename',
'share_id': 'fakeshareinstanceid',
'share_instance_id': 'fakeshareinstanceid',
'deleted': False,
'updated_at': datetime.datetime(2016, 3, 21, 0, 5, 58),
'created_at': datetime.datetime(2016, 3, 21, 0, 5, 58),
'deleted_at': None,
'share': fake_share(),
}
snapshot_instance.update(kwargs)
return db_fakes.FakeModel(snapshot_instance)

26
manila/tests/share/test_api.py
View File

@ -1069,7 +1069,7 @@ class ShareAPITestCase(test.TestCase):
mock.Mock(return_value=share)):
self.api.delete_snapshot(self.context, snapshot)
self.share_rpcapi.delete_snapshot.assert_called_once_with(
self.context, snapshot, share['host'])
self.context, snapshot, share['host'], force=False)
share_api.policy.check_policy.assert_called_once_with(
self.context, 'share', 'delete_snapshot', snapshot)
db_api.share_snapshot_instance_update.assert_called_once_with(
@ -1090,6 +1090,30 @@ class ShareAPITestCase(test.TestCase):
share_api.policy.check_policy.assert_called_once_with(
self.context, 'share', 'delete_snapshot', snapshot)
@ddt.data(constants.STATUS_MANAGING, constants.STATUS_ERROR_DELETING,
constants.STATUS_CREATING, constants.STATUS_AVAILABLE)
def test_delete_snapshot_force_delete(self, status):
share = fakes.fake_share(id=uuid.uuid4(), has_replicas=False)
snapshot = fakes.fake_snapshot(aggregate_status=status, share=share)
snapshot_instance = fakes.fake_snapshot_instance(
base_snapshot=snapshot)
self.mock_object(db_api, 'share_get', mock.Mock(return_value=share))
self.mock_object(
db_api, 'share_snapshot_instance_get_all_with_filters',
mock.Mock(return_value=[snapshot_instance]))
mock_instance_update_call = self.mock_object(
db_api, 'share_snapshot_instance_update')
mock_rpc_call = self.mock_object(self.share_rpcapi, 'delete_snapshot')
retval = self.api.delete_snapshot(self.context, snapshot, force=True)
self.assertIsNone(retval)
mock_instance_update_call.assert_called_once_with(
self.context, snapshot_instance['id'],
{'status': constants.STATUS_DELETING})
mock_rpc_call.assert_called_once_with(
self.context, snapshot, share['instance']['host'], force=True)
@ddt.data(True, False)
def test_delete_snapshot_replicated_snapshot(self, force):
share = fakes.fake_share(has_replicas=True)

314
manila/tests/share/test_manager.py
View File

@ -1363,116 +1363,268 @@ class ShareManagerTestCase(test.TestCase):
self.assertIsNone(retval)
self.assertTrue(replica_update_call.called)
def test_create_delete_share_snapshot(self):
"""Test share's snapshot can be created and deleted."""
def _get_snapshot_instance_dict(self, snapshot_instance, share):
expected_snapshot_instance_dict = {
'status': constants.STATUS_CREATING,
'share_id': share['id'],
'share_name': snapshot_instance['share_name'],
'deleted': snapshot_instance['deleted'],
'share': share,
'updated_at': snapshot_instance['updated_at'],
'snapshot_id': snapshot_instance['snapshot_id'],
'id': snapshot_instance['id'],
'name': snapshot_instance['name'],
'created_at': snapshot_instance['created_at'],
'share_instance_id': snapshot_instance['share_instance_id'],
'progress': snapshot_instance['progress'],
'deleted_at': snapshot_instance['deleted_at'],
'provider_location': snapshot_instance['provider_location'],
}
return expected_snapshot_instance_dict
def _fake_create_snapshot(self, snapshot, **kwargs):
snapshot['progress'] = '99%'
return snapshot.to_dict()
self.mock_object(self.share_manager.driver, "create_snapshot",
_fake_create_snapshot)
share = db_utils.create_share()
share_id = share['id']
snapshot = db_utils.create_snapshot(share_id=share_id)
snapshot_id = snapshot['id']
self.share_manager.create_snapshot(self.context, share_id,
snapshot_id)
self.assertEqual(share_id,
db.share_snapshot_get(context.get_admin_context(),
snapshot_id).share_id)
snap = db.share_snapshot_get(self.context, snapshot_id)
self.assertEqual(constants.STATUS_AVAILABLE, snap['status'])
self.share_manager.delete_snapshot(self.context, snapshot_id)
self.assertRaises(exception.NotFound,
db.share_snapshot_get,
self.context,
snapshot_id)
def test_create_delete_share_snapshot_error(self):
"""Test snapshot can be created and deleted with error."""
def test_create_snapshot_driver_exception(self):
def _raise_not_found(self, *args, **kwargs):
raise exception.NotFound()
share_id = 'FAKE_SHARE_ID'
share = fakes.fake_share(id=share_id, instance={'id': 'fake_id'})
snapshot_instance = fakes.fake_snapshot_instance(
share_id=share_id, share=share, name='fake_snapshot')
snapshot = fakes.fake_snapshot(
share_id=share_id, share=share, instance=snapshot_instance,
project_id=self.context.project_id)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager.driver, "create_snapshot",
mock.Mock(side_effect=_raise_not_found))
self.mock_object(self.share_manager.driver, "delete_snapshot",
mock.Mock(side_effect=_raise_not_found))
share = db_utils.create_share()
share_id = share['id']
snapshot = db_utils.create_snapshot(share_id=share_id)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager, '_get_share_server',
mock.Mock(return_value=None))
self.mock_object(self.share_manager.db, 'share_snapshot_instance_get',
mock.Mock(return_value=snapshot_instance))
self.mock_object(self.share_manager.db, 'share_snapshot_get',
mock.Mock(return_value=snapshot))
db_update = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_update')
expected_snapshot_instance_dict = self._get_snapshot_instance_dict(
snapshot_instance, share)
self.assertRaises(exception.NotFound,
self.share_manager.create_snapshot,
self.context, share_id, snapshot_id)
db_update.assert_called_once_with(self.context,
snapshot_instance['id'],
{'status': constants.STATUS_ERROR})
snap = db.share_snapshot_get(self.context, snapshot_id)
self.assertEqual(constants.STATUS_ERROR, snap['status'])
self.share_manager.driver.create_snapshot.assert_called_once_with(
self.context, expected_snapshot_instance_dict, share_server=None)
def test_create_snapshot(self):
share_id = 'FAKE_SHARE_ID'
share = fakes.fake_share(id=share_id, instance={'id': 'fake_id'})
snapshot_instance = fakes.fake_snapshot_instance(
share_id=share_id, share=share, name='fake_snapshot')
snapshot = fakes.fake_snapshot(
share_id=share_id, share=share, instance=snapshot_instance)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager.db, 'share_snapshot_get',
mock.Mock(return_value=snapshot))
self.mock_object(self.share_manager.db, 'share_snapshot_instance_get',
mock.Mock(return_value=snapshot_instance))
self.mock_object(self.share_manager, '_get_share_server',
mock.Mock(return_value=None))
expected_update_calls = [
mock.call(self.context, snapshot_instance['id'],
{'status': constants.STATUS_AVAILABLE,
'progress': '100%'})
]
expected_snapshot_instance_dict = self._get_snapshot_instance_dict(
snapshot_instance, share)
self.mock_object(self.share_manager.driver, 'create_snapshot',
mock.Mock(return_value=None))
db_update = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_update')
return_value = self.share_manager.create_snapshot(
self.context, share_id, snapshot_id)
self.assertEqual(None, return_value)
self.share_manager.driver.create_snapshot.assert_called_once_with(
self.context, expected_snapshot_instance_dict, share_server=None)
db_update.assert_has_calls(expected_update_calls, any_order=True)
def test_delete_snapshot_driver_other_exception(self):
def _raise_not_found(self, *args, **kwargs):
raise exception.NotFound()
share_id = 'FAKE_SHARE_ID'
share = fakes.fake_share(id=share_id, instance={'id': 'fake_id'})
snapshot_instance = fakes.fake_snapshot_instance(
share_id=share_id, share=share, name='fake_snapshot')
snapshot = fakes.fake_snapshot(
share_id=share_id, share=share, instance=snapshot_instance,
project_id=self.context.project_id)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager.driver, "delete_snapshot",
mock.Mock(side_effect=_raise_not_found))
self.mock_object(self.share_manager, '_get_share_server',
mock.Mock(return_value=None))
self.mock_object(self.share_manager.db, 'share_snapshot_instance_get',
mock.Mock(return_value=snapshot_instance))
self.mock_object(self.share_manager.db, 'share_snapshot_get',
mock.Mock(return_value=snapshot))
self.mock_object(
self.share_manager.db, 'share_get', mock.Mock(return_value=share))
db_update = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_update')
db_destroy_call = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_delete')
expected_snapshot_instance_dict = self._get_snapshot_instance_dict(
snapshot_instance, share)
mock_exception_log = self.mock_object(manager.LOG, 'exception')
self.assertRaises(exception.NotFound,
self.share_manager.delete_snapshot,
self.context, snapshot_id)
self.assertEqual(
constants.STATUS_ERROR_DELETING,
db.share_snapshot_get(self.context, snapshot_id).status)
self.share_manager.driver.create_snapshot.assert_called_once_with(
self.context, utils.IsAMatcher(models.ShareSnapshotInstance),
share_server=None)
db_update.assert_called_once_with(
mock.ANY, snapshot_instance['id'],
{'status': constants.STATUS_ERROR_DELETING})
self.share_manager.driver.delete_snapshot.assert_called_once_with(
utils.IsAMatcher(context.RequestContext),
utils.IsAMatcher(models.ShareSnapshotInstance),
mock.ANY, expected_snapshot_instance_dict,
share_server=None)
self.assertFalse(db_destroy_call.called)
self.assertFalse(mock_exception_log.called)
def test_delete_snapshot_quota_error(self):
share = db_utils.create_share()
share_id = share['id']
snapshot = db_utils.create_snapshot(share_id=share_id)
snapshot_id = snapshot['id']
snapshot = db_utils.create_snapshot(
with_share=True, status=constants.STATUS_AVAILABLE)
self.mock_object(quota.QUOTAS, 'reserve',
mock.Mock(side_effect=exception.QuotaError('fake')))
self.mock_object(quota.QUOTAS, 'commit')
self.share_manager.delete_snapshot(self.context, snapshot_id)
quota.QUOTAS.reserve.assert_called_once_with(
mock.ANY,
project_id=six.text_type(snapshot['project_id']),
snapshots=-1,
snapshot_gigabytes=-snapshot['size'],
user_id=six.text_type(snapshot['user_id'])
)
self.assertFalse(quota.QUOTAS.commit.called)
def test_delete_share_instance_if_busy(self):
"""Test snapshot could not be deleted if busy."""
def test_delete_snapshot_exception_snapshot_is_busy(self):
"""Test snapshot should not be deleted if busy."""
def _raise_share_snapshot_is_busy(self, *args, **kwargs):
raise exception.ShareSnapshotIsBusy(snapshot_name='fakename')
share_id = 'FAKE_SHARE_ID'
share = fakes.fake_share(
id=share_id, has_replicas=True, instance={'id': 'fake_id'})
snapshot_instance = fakes.fake_snapshot_instance(
share_id=share_id, share=share, name='fake_snapshot')
snapshot = fakes.fake_snapshot(
share_id=share_id, share=share, instance=snapshot_instance,
project_id=self.context.project_id)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager, '_get_share_server',
mock.Mock(return_value=None))
self.mock_object(self.share_manager.db, 'share_snapshot_instance_get',
mock.Mock(return_value=snapshot_instance))
self.mock_object(self.share_manager.db, 'share_snapshot_get',
mock.Mock(return_value=snapshot))
self.mock_object(self.share_manager.driver, "delete_snapshot",
mock.Mock(side_effect=_raise_share_snapshot_is_busy))
share = db_utils.create_share(status=constants.STATUS_ACTIVE)
snapshot = db_utils.create_snapshot(share_id=share['id'])
snapshot_update_call = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_update')
self.mock_object(
self.share_manager.db, 'share_get', mock.Mock(return_value=share))
self.assertRaises(exception.ShareSnapshotIsBusy,
self.share_manager.delete_snapshot,
self.context, snapshot_id)
snapshot_update_call.assert_called_once_with(
mock.ANY, snapshot_instance['id'],
{'status': constants.STATUS_AVAILABLE})
@ddt.data(True, False)
def test_delete_snapshot_with_quota_error(self, quota_error):
share_id = 'FAKE_SHARE_ID'
share = fakes.fake_share(id=share_id, instance={'id': 'fake_id'})
snapshot_instance = fakes.fake_snapshot_instance(
share_id=share_id, share=share, name='fake_snapshot')
snapshot = fakes.fake_snapshot(
share_id=share_id, share=share, instance=snapshot_instance,
project_id=self.context.project_id, size=1)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager.db, 'share_snapshot_get',
mock.Mock(return_value=snapshot))
self.mock_object(self.share_manager.db, 'share_snapshot_instance_get',
mock.Mock(return_value=snapshot_instance))
self.mock_object(self.share_manager.db, 'share_get',
mock.Mock(return_value=share))
self.mock_object(self.share_manager, '_get_share_server',
mock.Mock(return_value=None))
mock_exception_log = self.mock_object(manager.LOG, 'exception')
expected_exc_count = 1 if quota_error else 0
self.share_manager.delete_snapshot(self.context, snapshot_id)
expected_snapshot_instance_dict = self._get_snapshot_instance_dict(
snapshot_instance, share)
snap = db.share_snapshot_get(self.context, snapshot_id)
self.assertEqual(constants.STATUS_AVAILABLE, snap['status'])
self.mock_object(self.share_manager.driver, 'delete_snapshot')
db_update_call = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_update')
snapshot_destroy_call = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_delete')
side_effect = exception.QuotaError(code=500) if quota_error else None
self.mock_object(quota.QUOTAS, 'reserve',
mock.Mock(side_effect=side_effect))
quota_commit_call = self.mock_object(quota.QUOTAS, 'commit')
retval = self.share_manager.delete_snapshot(
self.context, snapshot_id)
self.assertIsNone(retval)
self.share_manager.driver.delete_snapshot.assert_called_once_with(
utils.IsAMatcher(context.RequestContext),
utils.IsAMatcher(models.ShareSnapshotInstance),
share_server=None)
mock.ANY, expected_snapshot_instance_dict, share_server=None)
self.assertFalse(db_update_call.called)
self.assertTrue(snapshot_destroy_call.called)
self.assertTrue(manager.QUOTAS.reserve.called)
quota.QUOTAS.reserve.assert_called_once_with(
mock.ANY, project_id=self.context.project_id, snapshots=-1,
snapshot_gigabytes=-snapshot['size'], user_id=snapshot['user_id'])
self.assertEqual(not quota_error, quota_commit_call.called)
self.assertEqual(quota_error, mock_exception_log.called)
self.assertEqual(expected_exc_count, mock_exception_log.call_count)
@ddt.data(exception.ShareSnapshotIsBusy, exception.ManilaException)
def test_delete_snapshot_ignore_exceptions_with_the_force(self, exc):
def _raise_quota_error():
raise exception.QuotaError(code='500')
share_id = 'FAKE_SHARE_ID'
share = fakes.fake_share(id=share_id, instance={'id': 'fake_id'})
snapshot_instance = fakes.fake_snapshot_instance(
share_id=share_id, share=share, name='fake_snapshot')
snapshot = fakes.fake_snapshot(
share_id=share_id, share=share, instance=snapshot_instance,
project_id=self.context.project_id, size=1)
snapshot_id = snapshot['id']
self.mock_object(self.share_manager.db, 'share_snapshot_get',
mock.Mock(return_value=snapshot))
self.mock_object(self.share_manager.db, 'share_snapshot_instance_get',
mock.Mock(return_value=snapshot_instance))
self.mock_object(self.share_manager.db, 'share_get',
mock.Mock(return_value=share))
self.mock_object(self.share_manager, '_get_share_server',
mock.Mock(return_value=None))
mock_exception_log = self.mock_object(manager.LOG, 'exception')
self.mock_object(self.share_manager.driver, 'delete_snapshot',
mock.Mock(side_effect=exc))
db_update_call = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_update')
snapshot_destroy_call = self.mock_object(
self.share_manager.db, 'share_snapshot_instance_delete')
self.mock_object(quota.QUOTAS, 'reserve',
mock.Mock(side_effect=_raise_quota_error))
quota_commit_call = self.mock_object(quota.QUOTAS, 'commit')
retval = self.share_manager.delete_snapshot(
self.context, snapshot_id, force=True)
self.assertIsNone(retval)
self.assertEqual(2, mock_exception_log.call_count)
snapshot_destroy_call.assert_called_once_with(
mock.ANY, snapshot_instance['id'])
self.assertFalse(quota_commit_call.called)
self.assertFalse(db_update_call.called)
def test_create_share_instance_with_share_network_dhss_false(self):
manager.CONF.set_default('driver_handles_share_servers', False)

3
manila/tests/share/test_rpcapi.py
View File

@ -196,7 +196,8 @@ class ShareRpcAPITestCase(test.TestCase):
self._test_share_api('delete_snapshot',
rpc_method='cast',
snapshot=self.fake_snapshot,
host='fake_host')
host='fake_host',
force=False)
def test_delete_share_server(self):
self._test_share_api('delete_share_server',

5
releasenotes/notes/snapshot-force-delete-4432bebfb5a0bbc9.yaml Normal file
View File

@ -0,0 +1,5 @@
---
fixes:
- force-delete API requests for snapshots are now propagated to the
manila-share service and will not fail even if share drivers cannot remove
the snapshots on the storage backend.