openstack
/
manila
Code Issues Proposed changes

Merge "Fix force-delete on snapshot resource" into stable/mitaka

This commit is contained in:
Jenkins 2016-08-12 15:44:05 +00:00 committed by Gerrit Code Review
parent 1dfebf0c4e eba4586ef3
commit 614418a2e4
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

@ -1045,8 +1045,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):

101
manila/share/manager.py
View File

@ -1887,13 +1887,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(
@ -1901,17 +1902,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)
@ -1919,8 +1919,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']:
@ -1928,35 +1927,59 @@ 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:
self.db.share_snapshot_instance_update(
context,
snapshot_instance_id,
{'status': constants.STATUS_AVAILABLE})
except Exception:
with excutils.save_and_reraise_exception():
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)
try:
reservations = QUOTAS.reserve(
context, project_id=project_id, snapshots=-1,
snapshot_gigabytes=-snapshot_ref['size'],
user_id=snapshot_ref['user_id'])
except Exception:
reservations = None
LOG.exception(_LE("Failed to update usages deleting snapshot"))
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})
if reservations:
QUOTAS.commit(context, reservations, project_id=project_id,
user_id=snapshot_ref['user_id'])
except 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})
self.db.share_snapshot_instance_delete(context, snapshot_instance_id)
try:
reservations = QUOTAS.reserve(
context, project_id=project_id, snapshots=-1,
snapshot_gigabytes=-snapshot_ref['size'],
user_id=snapshot_ref['user_id'])
except Exception:
reservations = None
LOG.exception(_LE("Failed to update quota usages while deleting "
"snapshot %s."), snapshot_id)
if reservations:
QUOTAS.commit(context, reservations, project_id=project_id,
user_id=snapshot_ref['user_id'])
@add_hooks
@utils.require_driver_initialized

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

@ -94,7 +94,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}
@ -113,6 +114,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:
@ -146,6 +150,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

@ -1106,7 +1106,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(
@ -1127,6 +1127,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.