openstack
/
barbican-tempest-plugin
Code Issues Proposed changes

Refactor secret cleanup 

This patch is the first in a refactor of the cleanup logic in our
tests.

This patch adds a new `cleanup()` method to the SecretClient that
attempts to delete all the secrets it creates.

Moving the responsibility of tracking which secrets to clean up down
to the client allows us more flexibility when cleaning up the resources.
e.g.  it should be fairly easy to clean up secrets across multiple projects
by just calling the new `cleanup()` method on each client used.

This patch will also allow us to get rid of the overloaded `do_request()`
method that is currently used as a proxy to the client to be able to track
entities.

The change also makes the test code more explicit and easier to read.

Change-Id: Id9be832a0f255410bd955d94c32001fec500f32f
This commit is contained in:
Douglas Mendizábal 2021-07-22 22:22:00 +00:00 committed by Douglas Mendizábal
parent 6d4e995e72
commit 62baff3fcc
4 changed files with 71 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
barbican_tempest_plugin/services/key_manager/json/base.py
View File

@ -20,3 +20,7 @@ class BarbicanTempestClient(rest_client.RestClient):
def __init__(self, *args, **kwargs):
kwargs['service'] = _DEFAULT_SERVICE_TYPE
super().__init__(*args, **kwargs)
@classmethod
def ref_to_uuid(cls, href):
return href.split('/')[-1]

26
barbican_tempest_plugin/services/key_manager/json/secret_client.py
View File

@ -18,6 +18,7 @@ import json
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import exceptions
from barbican_tempest_plugin.services.key_manager.json import base
@ -27,7 +28,11 @@ CONF = config.CONF
class SecretClient(base.BarbicanTempestClient):
def create_secret(self, **kwargs):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._secret_ids = set()
def create_secret(self, expected_status=201, **kwargs):
if 'name' not in kwargs:
kwargs['name'] = data_utils.rand_name("tempest-sec")
@ -37,10 +42,13 @@ class SecretClient(base.BarbicanTempestClient):
post_body = kwargs
body = json.dumps(post_body)
resp, body = self.post("v1/secrets", body)
self.expected_success(201, resp.status)
return self._parse_resp(body)
self.expected_success(expected_status, resp.status)
resp = self._parse_resp(body)
self._secret_ids.add(self.ref_to_uuid(resp['secret_ref']))
return resp
def delete_secret(self, secret_id):
self._secret_ids.discard(secret_id)
resp, body = self.delete("v1/secrets/%s" % secret_id)
self.expected_success(204, resp.status)
return body
@ -84,3 +92,15 @@ class SecretClient(base.BarbicanTempestClient):
headers=content_headers)
self.expected_success(204, resp.status)
return body
def queue_for_cleanup(self, secret_id):
self._secret_ids.add(secret_id)
def cleanup(self):
cleanup_ids = self._secret_ids
self._secret_ids = set()
for secret_id in cleanup_ids:
try:
self.delete_secret(secret_id)
except exceptions.NotFound:
pass

12
barbican_tempest_plugin/tests/rbac/v1/base.py
View File

@ -49,6 +49,7 @@ class BarbicanV1RbacBase(test.BaseTestCase):
credentials = ['system_admin']
# TODO(dmendiza): remove this and use the clients instead
@classmethod
def ref_to_uuid(cls, href):
return href.split('/')[-1]
@ -133,7 +134,6 @@ class BarbicanV1RbacBase(test.BaseTestCase):
cls.container_client = os.secret_v1.ContainerClient()
cls.order_client = os.secret_v1.OrderClient()
cls.quota_client = os.secret_v1.QuotaClient()
cls.secret_client = os.secret_v1.SecretClient()
cls.secret_metadata_client = os.secret_v1.SecretMetadataClient(
service='key-manager'
)
@ -151,9 +151,6 @@ class BarbicanV1RbacBase(test.BaseTestCase):
cls.admin_container_client = adm.secret_v1.ContainerClient()
cls.admin_order_client = adm.secret_v1.OrderClient()
cls.admin_quota_client = adm.secret_v1.QuotaClient()
cls.admin_secret_client = adm.secret_v1.SecretClient(
service='key-manager'
)
cls.admin_secret_metadata_client = adm.secret_v1.SecretMetadataClient(
service='key-manager'
)
@ -179,6 +176,8 @@ class BarbicanV1RbacBase(test.BaseTestCase):
for secret_uuid in list(cls.created_objects['secret']):
cls.admin_secret_client.delete_secret(secret_uuid)
cls.created_objects['secret'].remove(secret_uuid)
for client in [cls.secret_client, cls.admin_secret_client]:
client.cleanup()
finally:
super(BarbicanV1RbacBase, cls).resource_cleanup()
@ -203,6 +202,7 @@ class BarbicanV1RbacBase(test.BaseTestCase):
def delete_cleanup(cls, resource, uuid):
cls.created_objects[resource].remove(uuid)
# TODO(dmendiza): get rid of this helper method.
def do_request(self, method, client=None, expected_status=200,
cleanup=None, **args):
if client is None:
@ -220,9 +220,7 @@ class BarbicanV1RbacBase(test.BaseTestCase):
def create_empty_secret_admin(self, secret_name):
"""add empty secret as admin user """
return self.do_request(
'create_secret', client=self.admin_secret_client,
expected_status=201, cleanup='secret', name=secret_name)
return self.admin_secret_client.create_secret(name=secret_name)
def create_aes_secret_admin(self, secret_name):
key = create_aes_key()

78
barbican_tempest_plugin/tests/rbac/v1/test_secrets.py
View File

@ -96,13 +96,11 @@ class ProjectMemberTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
def test_create_secret(self):
"""Test add_secret policy."""
self.do_request('create_secret', expected_status=201, cleanup='secret',
name='test_create_secret')
self.client.create_secret(name='test_create_secret')
key = rbac_base.create_aes_key()
expire_time = (datetime.utcnow() + timedelta(days=5))
self.do_request(
'create_secret', expected_status=201, cleanup="secret",
self.client.create_secret(
name='test_create_secret2',
expiration=expire_time.isoformat(), algorithm="aes",
bit_length=256, mode="cbc", payload=key,
@ -117,55 +115,54 @@ class ProjectMemberTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
self.create_empty_secret_admin('test_list_secrets_2')
# list secrets with name secret_1
resp = self.do_request('list_secrets', name='test_list_secrets')
resp = self.client.list_secrets(name='test_list_secrets')
secrets = resp['secrets']
self.assertEqual('test_list_secrets', secrets[0]['name'])
# list secrets with name secret_2
resp = self.do_request('list_secrets', name='test_list_secrets_2')
resp = self.client.list_secrets(name='test_list_secrets_2')
secrets = resp['secrets']
self.assertEqual('test_list_secrets_2', secrets[0]['name'])
# list all secrets
resp = self.do_request('list_secrets')
resp = self.client.list_secrets()
secrets = resp['secrets']
self.assertGreaterEqual(len(secrets), 2)
def test_delete_secret(self):
"""Test delete_secrets policy."""
sec = self.create_empty_secret_admin('test_delete_secret_1')
uuid = self.ref_to_uuid(sec['secret_ref'])
self.do_request('delete_secret', secret_id=uuid)
self.delete_cleanup('secret', uuid)
uuid = self.client.ref_to_uuid(sec['secret_ref'])
self.client.delete_secret(uuid)
def test_get_secret(self):
"""Test get_secret policy."""
sec = self.create_empty_secret_admin('test_get_secret')
uuid = self.ref_to_uuid(sec['secret_ref'])
resp = self.do_request('get_secret_metadata', secret_id=uuid)
self.assertEqual(uuid, self.ref_to_uuid(resp['secret_ref']))
uuid = self.client.ref_to_uuid(sec['secret_ref'])
resp = self.client.get_secret_metadata(uuid)
self.assertEqual(uuid, self.client.ref_to_uuid(resp['secret_ref']))
def test_get_secret_payload(self):
"""Test get_secret payload policy."""
key, sec = self.create_aes_secret_admin('test_get_secret_payload')
uuid = self.ref_to_uuid(sec['secret_ref'])
uuid = self.client.ref_to_uuid(sec['secret_ref'])
# Retrieve the payload
payload = self.do_request('get_secret_payload', secret_id=uuid)
payload = self.client.get_secret_payload(uuid)
self.assertEqual(key, base64.b64encode(payload))
def test_put_secret_payload(self):
"""Test put_secret policy."""
sec = self.create_empty_secret_admin('test_put_secret_payload')
uuid = self.ref_to_uuid(sec['secret_ref'])
uuid = self.client.ref_to_uuid(sec['secret_ref'])
key = rbac_base.create_aes_key()
# Associate the payload with the created secret
self.do_request('put_secret_payload', secret_id=uuid, payload=key)
self.client.put_secret_payload(uuid, key)
# Retrieve the payload
payload = self.do_request('get_secret_payload', secret_id=uuid)
payload = self.client.get_secret_payload(uuid)
self.assertEqual(key, base64.b64encode(payload))
@ -185,15 +182,13 @@ class ProjectReaderTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
def test_create_secret(self):
"""Test add_secret policy."""
self.do_request(
'create_secret', expected_status=exceptions.Forbidden,
cleanup='secret')
self.assertRaises(exceptions.Forbidden, self.client.create_secret)
key = rbac_base.create_aes_key()
expire_time = (datetime.utcnow() + timedelta(days=5))
self.do_request(
'create_secret', expected_status=exceptions.Forbidden,
cleanup="secret",
self.assertRaises(
exceptions.Forbidden, self.client.create_secret,
expiration=expire_time.isoformat(), algorithm="aes",
bit_length=256, mode="cbc", payload=key,
payload_content_type="application/octet-stream",
@ -207,28 +202,30 @@ class ProjectReaderTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
self.create_empty_secret_admin('secret_2')
# list secrets with name secret_1
self.do_request(
'list_secrets', expected_status=exceptions.Forbidden,
self.assertRaises(
exceptions.Forbidden,
self.client.list_secrets,
name='secret_1'
)
# list secrets with name secret_2
self.do_request(
'list_secrets', expected_status=exceptions.Forbidden,
self.assertRaises(
exceptions.Forbidden,
self.client.list_secrets,
name='secret_2'
)
# list all secrets
self.do_request(
'list_secrets', expected_status=exceptions.Forbidden
)
self.assertRaises(exceptions.Forbidden, self.client.list_secrets)
def test_delete_secret(self):
"""Test delete_secrets policy."""
sec = self.create_empty_secret_admin('secret_1')
uuid = self.ref_to_uuid(sec['secret_ref'])
self.do_request(
'delete_secret', expected_status=exceptions.Forbidden,
self.assertRaises(
exceptions.Forbidden,
self.client.delete_secret,
secret_id=uuid
)
@ -236,8 +233,9 @@ class ProjectReaderTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
"""Test get_secret policy."""
sec = self.create_empty_secret_admin('secret_1')
uuid = self.ref_to_uuid(sec['secret_ref'])
self.do_request(
'get_secret_metadata', expected_status=exceptions.Forbidden,
self.assertRaises(
exceptions.Forbidden,
self.client.get_secret_metadata,
secret_id=uuid
)
@ -247,8 +245,9 @@ class ProjectReaderTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
uuid = self.ref_to_uuid(sec['secret_ref'])
# Retrieve the payload
self.do_request(
'get_secret_payload', expected_status=exceptions.Forbidden,
self.assertRaises(
exceptions.Forbidden,
self.client.get_secret_payload,
secret_id=uuid
)
@ -260,8 +259,9 @@ class ProjectReaderTests(rbac_base.BarbicanV1RbacBase, BarbicanV1RbacSecrets):
key = rbac_base.create_aes_key()
# Associate the payload with the created secret
self.do_request(
'put_secret_payload', expected_status=exceptions.Forbidden,
self.assertRaises(
exceptions.Forbidden,
self.client.put_secret_payload,
secret_id=uuid, payload=key
)