Merge "Implement force parameter"

This commit is contained in:
Zuul 2023-02-17 12:51:26 +00:00 committed by Gerrit Code Review
commit a2c3d54c82
5 changed files with 42 additions and 7 deletions

View File

@ -619,19 +619,23 @@ class BarbicanKeyManager(key_manager.KeyManager):
else: else:
raise exception.KeyManagerError(reason=e) raise exception.KeyManagerError(reason=e)
def delete(self, context, managed_object_id): def delete(self, context, managed_object_id, force=False):
"""Deletes the specified managed object. """Deletes the specified managed object.
:param context: contains information of the user and the environment :param context: contains information of the user and the environment
for the request (castellan/context.py) for the request (castellan/context.py)
:param managed_object_id: the UUID of the object to delete :param managed_object_id: the UUID of the object to delete
:param force: specifies if the secret must be deleted even when they
have consumers.
:raises ValueError: if the secret has consumers but no force parameter
is provided or if force equals False.
:raises KeyManagerError: if object deletion fails :raises KeyManagerError: if object deletion fails
:raises ManagedObjectNotFoundError: if the object could not be found :raises ManagedObjectNotFoundError: if the object could not be found
""" """
barbican_client = self._get_barbican_client(context) barbican_client = self._get_barbican_client(context)
try: try:
secret_ref = self._create_secret_ref(managed_object_id) secret_ref = self._create_secret_ref(managed_object_id)
barbican_client.secrets.delete(secret_ref) barbican_client.secrets.delete(secret_ref, force)
except (barbican_exceptions.HTTPAuthError, except (barbican_exceptions.HTTPAuthError,
barbican_exceptions.HTTPClientError, barbican_exceptions.HTTPClientError,
barbican_exceptions.HTTPServerError) as e: barbican_exceptions.HTTPServerError) as e:

View File

@ -106,7 +106,7 @@ class KeyManager(object, metaclass=abc.ABCMeta):
pass pass
@abc.abstractmethod @abc.abstractmethod
def delete(self, context, managed_object_id): def delete(self, context, managed_object_id, force=False):
"""Deletes the specified managed object. """Deletes the specified managed object.
Implementations should verify that the caller has permission to delete Implementations should verify that the caller has permission to delete
@ -119,6 +119,10 @@ class KeyManager(object, metaclass=abc.ABCMeta):
UUIDs of objects that belong to other users by repeatedly calling this UUIDs of objects that belong to other users by repeatedly calling this
method. That is, objects that belong to other users should be method. That is, objects that belong to other users should be
considered "non-existent" and completely invisible. considered "non-existent" and completely invisible.
Implementations that block the deletion of secrets with consumers
without making the "force" parameter equals True should raise
an exception.
""" """
pass pass

View File

@ -48,7 +48,7 @@ class NotImplementedKeyManager(key_manager.KeyManager):
def list(self, context, object_type=None): def list(self, context, object_type=None):
raise NotImplementedError() raise NotImplementedError()
def delete(self, context, managed_object_id): def delete(self, context, managed_object_id, force=False):
raise NotImplementedError() raise NotImplementedError()
def add_consumer(self, context, managed_object_id, consumer_data): def add_consumer(self, context, managed_object_id, consumer_data):

View File

@ -347,8 +347,12 @@ class VaultKeyManager(key_manager.KeyManager):
record['created'], record['created'],
key_id) key_id)
def delete(self, context, key_id): def delete(self, context, key_id, force=False):
"""Represents deleting the key.""" """Represents deleting the key.
The 'force' parameter is not used whatsoever and only kept to allow
consistency with the Barbican implementation.
"""
if not key_id: if not key_id:
raise exception.KeyManagerError('key identifier not provided') raise exception.KeyManagerError('key identifier not provided')

View File

@ -383,7 +383,30 @@ class BarbicanKeyManagerTestCase(test_key_manager.KeyManagerTestCase):
def test_delete_key(self): def test_delete_key(self):
self.key_mgr.delete(self.ctxt, self.key_id) self.key_mgr.delete(self.ctxt, self.key_id)
self.delete.assert_called_once_with(self.secret_ref) self.delete.assert_called_once_with(self.secret_ref, False)
def test_delete_secret_with_consumers_no_force_parameter(self):
self.mock_barbican.secrets.delete = mock.Mock(
side_effect=exception.KeyManagerError(
"Secret has consumers! Use the 'force' parameter."))
self.assertRaises(exception.KeyManagerError,
self.key_mgr.delete, self.ctxt, self.key_id)
self.mock_barbican.secrets.delete.assert_called_once_with(
self.secret_ref, False)
def test_delete_secret_with_consumers_force_parameter_false(self):
self.mock_barbican.secrets.delete = mock.Mock(
side_effect=barbican_exceptions.HTTPClientError(
"Secret has consumers! Use the 'force' parameter."))
self.assertRaises(exception.KeyManagerError,
self.key_mgr.delete, self.ctxt, self.key_id,
force=False)
self.mock_barbican.secrets.delete.assert_called_once_with(
self.secret_ref, False)
def test_delete_secret_with_consumers_force_parameter_true(self):
self.key_mgr.delete(self.ctxt, self.key_id, force=True)
self.delete.assert_called_once_with(self.secret_ref, True)
def test_delete_unknown_key(self): def test_delete_unknown_key(self):
self.assertRaises(exception.KeyManagerError, self.assertRaises(exception.KeyManagerError,