openstack/python-barbicanclient
Code Issues Proposed changes

Fix consistency between Order/Secret/Container

Browse Source 
Order/Secret capitalized factory methods disappear and are replaced
by more appropriate get/create methods in their respective managers,
to match the revised Container code.

The higher level REST functions for get/get_raw/post/delete become
private so that the namespace for developers using the client is
much cleaner. For example, nothing is visible in the top level
client object besides the manager classes for container/secret/order
and nothing is visible inside the manager classes besides the
appropriate get/create/delete methods (or other applicable
functions).

Internally, the barbican_cli code uses at least one private property
from ContainerManager, but it is assumed to be OK since it is within
the same "package", and the code will be maintained simultaneously.

Change-Id: I89525506f3be26b77421a5b8efa49bb645169aaf
This commit is contained in:
Adam Harwell
2014-09-29 16:14:18 -05:00
parent 075e213391
commit 0966770a0e
12 changed files with 241 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
barbicanclient/barbican_cli/containers.py
View File

@@ -93,7 +93,7 @@ class CreateContainer(show.ShowOne):
return parser
def take_action(self, args):
container_type = self.app.client.containers.container_map.get(
container_type = self.app.client.containers._container_map.get(
args.type)
if not container_type:
raise ValueError('Invalid container type specified.')

6
barbicanclient/barbican_cli/orders.py
View File

@@ -48,7 +48,7 @@ class CreateOrder(show.ShowOne):
return parser
def take_action(self, args):
entity = self.app.client.orders.Order(
entity = self.app.client.orders.create(
name=args.name, payload_content_type=args.payload_content_type,
algorithm=args.algorithm, bit_length=args.bit_length,
mode=args.mode, expiration=args.expiration)
@@ -77,7 +77,7 @@ class GetOrder(show.ShowOne):
return parser
def take_action(self, args):
entity = self.app.client.orders.Order(order_ref=args.URI)
entity = self.app.client.orders.get(order_ref=args.URI)
return entity._get_formatted_entity()
@@ -99,4 +99,4 @@ class ListOrder(lister.Lister):
def take_action(self, args):
obj_list = self.app.client.orders.list(args.limit, args.offset)
return orders.OrderFormatter._list_objects(obj_list)
return orders.Order._list_objects(obj_list)

6
barbicanclient/barbican_cli/secrets.py
View File

@@ -57,7 +57,7 @@ class GetSecret(show.ShowOne):
return (('Secret',),
(entity,))
else:
entity = self.app.client.secrets.Secret(secret_ref=args.URI)
entity = self.app.client.secrets.get(secret_ref=args.URI)
return entity._get_formatted_entity()
@@ -95,7 +95,7 @@ class ListSecret(lister.Lister):
args.name, args.mode,
args.algorithm,
args.bit_length)
return secrets.SecretFormatter._list_objects(obj_list)
return secrets.Secret._list_objects(obj_list)
class StoreSecret(show.ShowOne):
@@ -133,7 +133,7 @@ class StoreSecret(show.ShowOne):
return parser
def take_action(self, args):
entity = self.app.client.secrets.Secret(
entity = self.app.client.secrets.create(
name=args.name, payload=args.payload,
payload_content_type=args.payload_content_type,
payload_content_encoding=args.payload_content_encoding,

8
barbicanclient/base.py
View File

@@ -42,15 +42,15 @@ class ImmutableException(Exception):
class BaseEntityManager(object):
def __init__(self, api, entity):
self.api = api
self.entity = entity
self._api = api
self._entity = entity
def total(self):
"""
Returns the total number of entities stored in Barbican.
"""
href = '{0}/{1}'.format(self.api.base_url, self.entity)
href = '{0}/{1}'.format(self._api._base_url, self._entity)
params = {'limit': 0, 'offset': 0}
resp = self.api.get(href, params)
resp = self._api._get(href, params)
return resp['total']

12
barbicanclient/client.py
View File

@@ -97,7 +97,7 @@ class Client(object):
self._barbican_url = self._get_normalized_endpoint(endpoint)
self._tenant_id = tenant_id
self.base_url = '{0}'.format(self._barbican_url)
self._base_url = '{0}'.format(self._barbican_url)
self.secrets = secrets.SecretManager(self)
self.orders = orders.OrderManager(self)
self.containers = containers.ContainerManager(self)
@@ -161,27 +161,27 @@ class Client(object):
if isinstance(headers, dict) and not self._session.auth:
headers['X-Project-Id'] = self._tenant_id
def get(self, href, params=None):
def _get(self, href, params=None):
headers = {'Accept': 'application/json'}
self._prepare_auth(headers)
resp = self._session.get(href, params=params, headers=headers)
self._check_status_code(resp)
return resp.json()
def get_raw(self, href, headers):
def _get_raw(self, href, headers):
self._prepare_auth(headers)
resp = self._session.get(href, headers=headers)
self._check_status_code(resp)
return resp.content
def delete(self, href, json=None):
def _delete(self, href, json=None):
headers = {}
self._prepare_auth(headers)
resp = self._session.delete(href, headers=headers, json=json)
self._check_status_code(resp)
def post(self, path, data):
url = '{0}/{1}/'.format(self.base_url, path)
def _post(self, path, data):
url = '{0}/{1}/'.format(self._base_url, path)
headers = {'content-type': 'application/json'}
self._prepare_auth(headers)
resp = self._session.post(url, data=json.dumps(data), headers=headers)

70
barbicanclient/containers.py
View File

@@ -71,7 +71,7 @@ class Container(ContainerFormatter):
"""
Containers are used to keep track of the data stored in Barbican.
"""
entity = 'containers'
_entity = 'containers'
_type = 'generic'
def __init__(self, api, name=None, secrets=None, consumers=None,
@@ -79,14 +79,20 @@ class Container(ContainerFormatter):
secret_refs=None):
self._api = api
self._name = name
self._container_ref = container_ref
self._secret_refs = secret_refs
self._cached_secrets = dict()
self._initialize_secrets(secrets)
self._consumers = consumers if consumers else list()
self._container_ref = container_ref
self._created = parse_isotime(created) if created else None
self._updated = parse_isotime(updated) if updated else None
self._status = status
if container_ref:
self._consumers = consumers if consumers else list()
self._created = parse_isotime(created) if created else None
self._updated = parse_isotime(updated) if updated else None
self._status = status
else:
self._consumers = list()
self._created = None
self._updated = None
self._status = None
def _initialize_secrets(self, secrets):
try:
@@ -105,7 +111,7 @@ class Container(ContainerFormatter):
def _fill_secrets_from_secret_refs(self):
if self._secret_refs:
self._cached_secrets = dict(
(name.lower(), self._api.secrets.Secret(secret_ref=secret_ref))
(name.lower(), self._api.secrets.get(secret_ref=secret_ref))
for name, secret_ref in six.iteritems(self._secret_refs)
)
@@ -185,14 +191,14 @@ class Container(ContainerFormatter):
LOG.debug("Request body: {0}".format(container_dict))
# Save, store container_ref and return
response = self._api.post(self.entity, container_dict)
response = self._api._post(self._entity, container_dict)
if response:
self._container_ref = response['container_ref']
return self.container_ref
def delete(self):
if self._container_ref:
self._api.delete(self._container_ref)
self._api._delete(self._container_ref)
self._container_ref = None
self._status = None
self._created = None
@@ -217,7 +223,7 @@ class Container(ContainerFormatter):
.format(self._container_ref))
base.validate_ref(self._container_ref, 'Container')
try:
response = self._api.get(self._container_ref)
response = self._api._get(self._container_ref)
except AttributeError:
raise LookupError('Container {0} could not be found.'
.format(self._container_ref))
@@ -486,7 +492,7 @@ class CertificateContainer(CertificateContainerFormatter, Container):
class ContainerManager(base.BaseEntityManager):
container_map = {
_container_map = {
'generic': Container,
'rsa': RSAContainer,
'certificate': CertificateContainer
@@ -496,7 +502,8 @@ class ContainerManager(base.BaseEntityManager):
super(ContainerManager, self).__init__(api, 'containers')
def get(self, container_ref):
"""Get a Container
"""
Get a Container
:param container_ref: Full HATEOAS reference to a Container
:returns: Container object or a subclass of the appropriate type
@@ -505,7 +512,7 @@ class ContainerManager(base.BaseEntityManager):
.format(container_ref))
base.validate_ref(container_ref, 'Container')
try:
response = self.api.get(container_ref)
response = self._api._get(container_ref)
except AttributeError:
raise LookupError('Container {0} could not be found.'
.format(container_ref))
@@ -513,7 +520,7 @@ class ContainerManager(base.BaseEntityManager):
def _generate_typed_container(self, response):
resp_type = response.get('type', '').lower()
container_type = self.container_map.get(resp_type)
container_type = self._container_map.get(resp_type)
if not container_type:
raise TypeError('Unknown container type "{0}".'
.format(resp_type))
@@ -533,7 +540,7 @@ class ContainerManager(base.BaseEntityManager):
private_key_ref = secret_refs.get('private_key')
private_key_pass_ref = secret_refs.get('private_key_passphrase')
return RSAContainer(
api=self.api,
api=self._api,
name=name,
consumers=consumers,
container_ref=container_ref,
@@ -550,7 +557,7 @@ class ContainerManager(base.BaseEntityManager):
private_key_ref = secret_refs.get('private_key')
private_key_pass_ref = secret_refs.get('private_key_passphrase')
return CertificateContainer(
api=self.api,
api=self._api,
name=name,
consumers=consumers,
container_ref=container_ref,
@@ -563,7 +570,7 @@ class ContainerManager(base.BaseEntityManager):
private_key_passphrase_ref=private_key_pass_ref,
)
return container_type(
api=self.api,
api=self._api,
name=name,
secret_refs=secret_refs,
consumers=consumers,
@@ -582,14 +589,14 @@ class ContainerManager(base.BaseEntityManager):
def create(self, name=None, secrets=None):
"""
Container creation method
Create a Container
:param name: A friendly name for the Container
:param secrets: Secrets to populate when creating a Container
:returns: Container
"""
return Container(
api=self.api,
api=self._api,
name=name,
secrets=secrets
)
@@ -597,7 +604,7 @@ class ContainerManager(base.BaseEntityManager):
def create_rsa(self, name=None, public_key=None, private_key=None,
private_key_passphrase=None):
"""
RSAContainer creation method
Create an RSAContainer
:param name: A friendly name for the RSAContainer
:param public_key: Secret object containing a Public Key
@@ -606,7 +613,7 @@ class ContainerManager(base.BaseEntityManager):
:returns: RSAContainer
"""
return RSAContainer(
api=self.api,
api=self._api,
name=name,
public_key=public_key,
private_key=private_key,
@@ -617,7 +624,7 @@ class ContainerManager(base.BaseEntityManager):
intermediates=None, private_key=None,
private_key_passphrase=None):
"""
CertificateContainer creation method
Create a CertificateContainer
:param name: A friendly name for the CertificateContainer
:param certificate: Secret object containing a Certificate
@@ -627,7 +634,7 @@ class ContainerManager(base.BaseEntityManager):
:returns: CertificateContainer
"""
return CertificateContainer(
api=self.api,
api=self._api,
name=name,
certificate=certificate,
intermediates=intermediates,
@@ -637,14 +644,14 @@ class ContainerManager(base.BaseEntityManager):
def delete(self, container_ref):
"""
Deletes a container
Delete a Container
:param container_ref: Full HATEOAS reference to a Container
"""
if not container_ref:
raise ValueError('container_ref is required.')
try:
self.api.delete(container_ref)
self._api._delete(container_ref)
except AttributeError:
raise LookupError('Container {0} could not be deleted. '
'Does it still exist?'.format(container_ref))
@@ -661,14 +668,14 @@ class ContainerManager(base.BaseEntityManager):
"""
LOG.debug('Listing containers - offset {0} limit {1} name {2} type {3}'
.format(offset, limit, name, type))
href = '{0}/{1}'.format(self.api.base_url, self.entity)
href = '{0}/{1}'.format(self._api._base_url, self._entity)
params = {'limit': limit, 'offset': offset}
if name:
params['name'] = name
if type:
params['type'] = type
response = self.api.get(href, params)
response = self._api._get(href, params)
return [self._generate_typed_container(container)
for container in response.get('containers', [])]
@@ -684,13 +691,13 @@ class ContainerManager(base.BaseEntityManager):
"""
LOG.debug('Creating consumer registration for container '
'{0} as {1}: {2}'.format(container_ref, name, url))
href = '{0}/{1}/consumers'.format(self.entity,
href = '{0}/{1}/consumers'.format(self._entity,
container_ref.split('/')[-1])
consumer_dict = dict()
consumer_dict['name'] = name
consumer_dict['URL'] = url
response = self.api.post(href, consumer_dict)
response = self._api._post(href, consumer_dict)
return self._generate_typed_container(response)
def remove_consumer(self, container_ref, name, url):
@@ -703,11 +710,12 @@ class ContainerManager(base.BaseEntityManager):
"""
LOG.debug('Deleting consumer registration for container '
'{0} as {1}: {2}'.format(container_ref, name, url))
href = '{0}/{1}/{2}/consumers'.format(self.api.base_url, self.entity,
href = '{0}/{1}/{2}/consumers'.format(self._api._base_url,
self._entity,
container_ref.split('/')[-1])
consumer_dict = {
'name': name,
'URL': url
}
self.api.delete(href, json=consumer_dict)
self._api._delete(href, json=consumer_dict)

48
barbicanclient/orders.py
View File

@@ -58,7 +58,7 @@ class Order(OrderFormatter):
"""
Orders are used to request the generation of a Secret in Barbican.
"""
entity = 'orders'
_entity = 'orders'
def __init__(self, api, name=None, algorithm=None, bit_length=None,
mode=None, payload_content_type='application/octet-stream',
@@ -207,14 +207,14 @@ class Order(OrderFormatter):
'secret': self._secret
})
LOG.debug("Request body: {0}".format(order_dict.get('secret')))
response = self._api.post(self.entity, order_dict)
response = self._api._post(self._entity, order_dict)
if response:
self._order_ref = response.get('order_ref')
return self._order_ref
def delete(self):
if self._order_ref:
self._api.delete(self._order_ref)
self._api._delete(self._order_ref)
self._order_ref = None
else:
raise LookupError("Order is not yet stored.")
@@ -228,45 +228,49 @@ class OrderManager(base.BaseEntityManager):
def __init__(self, api):
super(OrderManager, self).__init__(api, 'orders')
def Order(self, order_ref=None, name=None, payload_content_type=None,
algorithm=None, bit_length=None, mode=None, expiration=None):
def get(self, order_ref):
"""
Factory method that either retrieves an Order from Barbican if
given an order_ref, or creates a new Order if not, and returns
the Order object.
Get an Order
:param order_ref: Full HATEOAS reference to an Order
:returns: Order
"""
LOG.debug("Getting order - Order href: {0}".format(order_ref))
base.validate_ref(order_ref, 'Order')
response = self._api._get(order_ref)
return Order(api=self._api, **response)
def create(self, name=None, payload_content_type=None,
algorithm=None, bit_length=None, mode=None, expiration=None):
"""
Create an Order
:param order_ref: If provided, will do an Order GET in Barbican
:param name: A friendly name for the secret
:param payload_content_type: The format/type of the secret data
:param algorithm: The algorithm associated with this secret key
:param bit_length: The bit length of this secret key
:param mode: The algorithm mode used with this secret key
:param expiration: The expiration time of the secret in ISO 8601 format
:returns: Secret object
:returns: Order
"""
if order_ref:
LOG.debug("Getting order - Order href: {0}".format(order_ref))
base.validate_ref(order_ref, 'Order')
response = self.api.get(order_ref)
return Order(api=self.api, **response)
return Order(api=self.api, name=name,
return Order(api=self._api, name=name,
payload_content_type=payload_content_type,
algorithm=algorithm, bit_length=bit_length, mode=mode,
expiration=expiration)
def delete(self, order_ref):
"""
Deletes an order
Delete an Order
:param order_ref: The href for the order
"""
if not order_ref:
raise ValueError('order_ref is required.')
self.api.delete(order_ref)
self._api._delete(order_ref)
def list(self, limit=10, offset=0):
"""
Lists all orders for the tenant
List all Orders for the tenant
:param limit: Max number of orders returned
:param offset: Offset orders to begin list
@@ -274,8 +278,8 @@ class OrderManager(base.BaseEntityManager):
"""
LOG.debug('Listing orders - offset {0} limit {1}'.format(offset,
limit))
href = '{0}/{1}'.format(self.api.base_url, self.entity)
href = '{0}/{1}'.format(self._api._base_url, self._entity)
params = {'limit': limit, 'offset': offset}
response = self.api.get(href, params)
response = self._api._get(href, params)
return [Order(api=self.api, **o) for o in response.get('orders', [])]
return [Order(api=self._api, **o) for o in response.get('orders', [])]

58
barbicanclient/secrets.py
View File

@@ -65,7 +65,7 @@ class Secret(SecretFormatter):
"""
Secrets are used to keep track of the data stored in Barbican.
"""
entity = 'secrets'
_entity = 'secrets'
def __init__(self, api, name=None, expiration=None, algorithm=None,
bit_length=None, mode=None, payload=None,
@@ -210,7 +210,7 @@ class Secret(SecretFormatter):
"secret does not specify a 'default' "
"content-type.")
headers = {'Accept': self._payload_content_type}
self._payload = self._api.get_raw(self._secret_ref, headers)
self._payload = self._api._get_raw(self._secret_ref, headers)
@immutable_after_save
def store(self):
@@ -228,14 +228,14 @@ class Secret(SecretFormatter):
LOG.debug("Request body: {0}".format(secret_dict))
# Save, store secret_ref and return
response = self._api.post(self.entity, secret_dict)
response = self._api._post(self._entity, secret_dict)
if response:
self._secret_ref = response.get('secret_ref')
return self.secret_ref
def delete(self):
if self._secret_ref:
self._api.delete(self._secret_ref)
self._api._delete(self._secret_ref)
self._secret_ref = None
else:
raise LookupError("Secret is not yet stored.")
@@ -249,15 +249,29 @@ class SecretManager(base.BaseEntityManager):
def __init__(self, api):
super(SecretManager, self).__init__(api, 'secrets')
def Secret(self, secret_ref=None, name=None, payload=None,
def get(self, secret_ref, payload_content_type=None):
"""
Get a Secret
:param secret_ref: Full HATEOAS reference to a Secret
:param payload_content_type: Content type to use for payload decryption
:returns: Secret
"""
LOG.debug("Getting secret - Secret href: {0}".format(secret_ref))
base.validate_ref(secret_ref, 'Secret')
response = self._api._get(secret_ref)
return Secret(
api=self._api,
payload_content_type=payload_content_type,
**response
)
def create(self, name=None, payload=None,
payload_content_type=None, payload_content_encoding=None,
algorithm=None, bit_length=None, mode=None, expiration=None):
"""
Factory method that either retrieves a Secret from Barbican if
given a secret_ref, or creates a new Secret if not, and returns
the Secret object.
Create a Secret
:param secret_ref: If provided, will do a Secret GET in Barbican
:param name: A friendly name for the Secret
:param payload: The unencrypted secret data
:param payload_content_type: The format/type of the secret data
@@ -266,16 +280,9 @@ class SecretManager(base.BaseEntityManager):
:param bit_length: The bit length of this secret key
:param mode: The algorithm mode used with this secret key
:param expiration: The expiration time of the secret in ISO 8601 format
:returns: Secret object
:returns: Secret
"""
if secret_ref:
LOG.debug("Getting secret - Secret href: {0}".format(secret_ref))
base.validate_ref(secret_ref, 'Secret')
response = self.api.get(secret_ref)
return Secret(api=self.api,
payload_content_type=payload_content_type,
**response)
return Secret(api=self.api, name=name, payload=payload,
return Secret(api=self._api, name=name, payload=payload,
payload_content_type=payload_content_type,
payload_content_encoding=payload_content_encoding,
algorithm=algorithm, bit_length=bit_length, mode=mode,
@@ -283,18 +290,18 @@ class SecretManager(base.BaseEntityManager):
def delete(self, secret_ref):
"""
Deletes a secret
Delete a Secret
:param secret_ref: The href for the secret
"""
if not secret_ref:
raise ValueError('secret_ref is required.')
self.api.delete(secret_ref)
self._api._delete(secret_ref)
def list(self, limit=10, offset=0, name=None, algorithm=None,
mode=None, bits=0):
"""
List all secrets for the tenant
List all Secrets for the tenant
:param limit: Max number of secrets returned
:param offset: Offset secrets to begin list
@@ -306,7 +313,7 @@ class SecretManager(base.BaseEntityManager):
"""
LOG.debug('Listing secrets - offset {0} limit {1}'.format(offset,
limit))
href = '{0}/{1}'.format(self.api.base_url, self.entity)
href = '{0}/{1}'.format(self._api._base_url, self._entity)
params = {'limit': limit, 'offset': offset}
if name:
params['name'] = name
@@ -317,6 +324,9 @@ class SecretManager(base.BaseEntityManager):
if bits > 0:
params['bits'] = bits
response = self.api.get(href, params)
response = self._api._get(href, params)
return [Secret(api=self.api, **s) for s in response.get('secrets', [])]
return [
Secret(api=self._api, **s)
for s in response.get('secrets', [])
]

48
barbicanclient/test/test_client.py
View File

@@ -106,7 +106,7 @@ class WhenTestingClientInit(testtools.TestCase):
c = client.Client(auth_plugin=None, endpoint=self.endpoint,
tenant_id=self.tenant_id)
expected = self.endpoint.rstrip('/')
self.assertEqual(expected, c.base_url)
self.assertEqual(expected, c._base_url)
def test_auth_token_header_is_set_when_using_auth_plugin(self):
c = client.Client(auth_plugin=self.fake_auth)
@@ -127,11 +127,11 @@ class WhenTestingClientInit(testtools.TestCase):
def test_base_url_starts_with_endpoint_url(self):
c = client.Client(auth_plugin=self.fake_auth)
self.assertTrue(c.base_url.startswith(self.endpoint))
self.assertTrue(c._base_url.startswith(self.endpoint))
def test_base_url_has_no_tenant_id(self):
c = client.Client(auth_plugin=self.fake_auth)
self.assertNotIn(self.tenant_id, c.base_url)
self.assertNotIn(self.tenant_id, c._base_url)
def test_should_raise_for_unauthorized_response(self):
resp = self._mock_response(status_code=401)
@@ -176,7 +176,7 @@ class WhenTestingClientWithSession(testtools.TestCase):
self.session.request.return_value.json.return_value = {
'entity_ref': self.entity_href}
resp_dict = self.client.post(self.entity, self.entity_dict)
resp_dict = self.client._post(self.entity, self.entity_dict)
self.assertEqual(self.entity_href, resp_dict['entity_ref'])
@@ -193,7 +193,7 @@ class WhenTestingClientWithSession(testtools.TestCase):
self.session.request.return_value = mock.MagicMock(status_code=200)
self.session.request.return_value.json.return_value = {
'name': self.entity_name}
resp_dict = self.client.get(self.entity_href)
resp_dict = self.client._get(self.entity_href)
self.assertEqual(self.entity_name, resp_dict['name'])
@@ -211,7 +211,7 @@ class WhenTestingClientWithSession(testtools.TestCase):
content='content')
headers = {'Accept': 'application/octet-stream'}
content = self.client.get_raw(self.entity_href, headers)
content = self.client._get_raw(self.entity_href, headers)
self.assertEqual('content', content)
@@ -227,7 +227,7 @@ class WhenTestingClientWithSession(testtools.TestCase):
def test_should_delete(self):
self.session.request.return_value = mock.MagicMock(status_code=200)
self.client.delete(self.entity_href)
self.client._delete(self.entity_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.session.request.call_args
@@ -255,14 +255,14 @@ class WhenTestingClientWithKeystoneV2(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v2_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
list_secrets_url = '{0}/secrets'.format(c.base_url)
list_secrets_url = '{0}/secrets'.format(c._base_url)
httpretty.register_uri(
httpretty.GET,
list_secrets_url,
status=200,
body='{{"name": "{0}", "secret_ref": "{1}"}}'.format(
self.entity_name, self.entity_href))
resp = c.get(list_secrets_url)
resp = c._get(list_secrets_url)
self.assertEqual(self.entity_name, resp['name'])
self.assertEqual(self.entity_href, resp['secret_ref'])
@@ -281,14 +281,14 @@ class WhenTestingClientWithKeystoneV2(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v2_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
post_secret_url = '{0}/secrets/'.format(c.base_url)
post_secret_url = '{0}/secrets/'.format(c._base_url)
httpretty.register_uri(
httpretty.POST,
post_secret_url,
status=200,
body='{{"name": "{0}", "secret_ref": "{1}"}}'.format(
self.entity_name, self.entity_href))
resp = c.post('secrets', '{"name":"test"}')
resp = c._post('secrets', '{"name":"test"}')
self.assertEqual(self.entity_name, resp['name'])
self.assertEqual(self.entity_href, resp['secret_ref'])
@@ -307,13 +307,13 @@ class WhenTestingClientWithKeystoneV2(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v2_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
get_secret_url = '{0}/secrets/s1'.format(c.base_url)
get_secret_url = '{0}/secrets/s1'.format(c._base_url)
httpretty.register_uri(
httpretty.GET,
get_secret_url,
status=200, body='content')
headers = {"Content-Type": "application/json"}
resp = c.get_raw(get_secret_url, headers)
resp = c._get_raw(get_secret_url, headers)
self.assertEqual(b'content', resp)
@httpretty.activate
@@ -331,12 +331,12 @@ class WhenTestingClientWithKeystoneV2(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v2_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
delete_secret_url = '{0}/secrets/s1'.format(c.base_url)
delete_secret_url = '{0}/secrets/s1'.format(c._base_url)
httpretty.register_uri(
httpretty.DELETE,
delete_secret_url,
status=201)
c.delete(delete_secret_url)
c._delete(delete_secret_url)
class WhenTestingClientWithKeystoneV3(WhenTestingClientWithSession):
@@ -360,14 +360,14 @@ class WhenTestingClientWithKeystoneV3(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v3_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
list_secrets_url = '{0}/secrets'.format(c.base_url)
list_secrets_url = '{0}/secrets'.format(c._base_url)
httpretty.register_uri(
httpretty.GET,
list_secrets_url,
status=200,
body='{{"name": "{0}", "secret_ref": "{1}"}}'.format(
self.entity_name, self.entity_href))
resp = c.get(list_secrets_url)
resp = c._get(list_secrets_url)
self.assertEqual(self.entity_name, resp['name'])
self.assertEqual(self.entity_href, resp['secret_ref'])
@@ -388,7 +388,7 @@ class WhenTestingClientWithKeystoneV3(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v3_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
post_secret_url = '{0}/secrets/'.format(c.base_url)
post_secret_url = '{0}/secrets/'.format(c._base_url)
httpretty.register_uri(
httpretty.POST,
post_secret_url,
@@ -396,7 +396,7 @@ class WhenTestingClientWithKeystoneV3(WhenTestingClientWithSession):
x_subject_token=id,
body='{{"name": "{0}", "secret_ref": "{1}"}}'.format(
self.entity_name, self.entity_href))
resp = c.post('secrets', '{"name":"test"}')
resp = c._post('secrets', '{"name":"test"}')
self.assertEqual(self.entity_name, resp['name'])
self.assertEqual(self.entity_href, resp['secret_ref'])
@@ -417,13 +417,13 @@ class WhenTestingClientWithKeystoneV3(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v3_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
get_secret_url = '{0}/secrets/s1'.format(c.base_url)
get_secret_url = '{0}/secrets/s1'.format(c._base_url)
httpretty.register_uri(
httpretty.GET,
get_secret_url,
status=200, body='content')
headers = {"Content-Type": "application/json"}
resp = c.get_raw(get_secret_url, headers)
resp = c._get_raw(get_secret_url, headers)
self.assertEqual(b'content', resp)
@httpretty.activate
@@ -443,12 +443,12 @@ class WhenTestingClientWithKeystoneV3(WhenTestingClientWithSession):
auth_plugin = KeystonePasswordPlugins.get_v3_plugin()
c = client.Client(auth_plugin=auth_plugin)
# emulate list secrets
delete_secret_url = '{0}/secrets/s1'.format(c.base_url)
delete_secret_url = '{0}/secrets/s1'.format(c._base_url)
httpretty.register_uri(
httpretty.DELETE,
delete_secret_url,
status=201)
c.delete(delete_secret_url)
c._delete(delete_secret_url)
class BaseEntityResource(testtools.TestCase):
@@ -496,4 +496,4 @@ class BaseEntityResource(testtools.TestCase):
'abcd1234-eabc-5678-9abc-abcdef012345'
self.api = mock.MagicMock()
self.api.base_url = self.endpoint[:-1]
self.api._base_url = self.endpoint[:-1]

76
barbicanclient/test/test_client_containers.py
View File

@@ -131,7 +131,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
repr(container_obj))
def test_should_store_generic_via_constructor(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create(
name=self.container.name,
@@ -141,7 +141,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -153,7 +153,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container_req['secret_refs'])
def test_should_store_generic_via_attributes(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create()
container.name = self.container.name
@@ -163,7 +163,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -175,7 +175,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container_req['secret_refs'])
def test_should_store_certificate_via_attributes(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create_certificate()
container.name = self.container.name
@@ -188,7 +188,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -200,7 +200,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container_req['secret_refs'])
def test_should_store_certificate_via_constructor(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create_certificate(
name=self.container.name,
@@ -213,7 +213,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -225,7 +225,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container_req['secret_refs'])
def test_should_store_rsa_via_attributes(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create_rsa()
container.name = self.container.name
@@ -237,7 +237,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -249,7 +249,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container_req['secret_refs'])
def test_should_store_rsa_via_constructor(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create_rsa(
name=self.container.name,
@@ -262,7 +262,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -274,7 +274,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container_req['secret_refs'])
def test_should_get_secret_refs_when_created_using_secret_objects(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create(
name=self.container.name,
@@ -285,8 +285,8 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.container.generic_secret_refs)
def test_should_reload_attributes_after_store(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api.get.return_value = self.container.get_dict(self.entity_href)
self.api._post.return_value = {'container_ref': self.entity_href}
self.api._get.return_value = self.container.get_dict(self.entity_href)
container = self.manager.create(
name=self.container.name,
@@ -321,7 +321,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container.add(self.container.secret.name, self.container.secret)
def test_should_be_immutable_after_store(self):
self.api.post.return_value = {'container_ref': self.entity_href}
self.api._post.return_value = {'container_ref': self.entity_href}
container = self.manager.create(
name=self.container.name,
@@ -359,28 +359,29 @@ class WhenTestingContainers(test_client.BaseEntityResource):
pass
def test_should_get_generic_container(self):
self.api.get.return_value = self.container.get_dict(self.entity_href)
self.api._get.return_value = self.container.get_dict(self.entity_href)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
self.assertIsNotNone(container.secrets)
def test_should_get_certificate_container(self):
self.api.get.return_value = self.container.get_dict(self.entity_href,
type='certificate')
self.api._get.return_value = self.container.get_dict(
self.entity_href, type='certificate'
)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -392,15 +393,15 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertIsNotNone(container.intermediates)
def test_should_get_rsa_container(self):
self.api.get.return_value = self.container.get_dict(self.entity_href,
type='rsa')
self.api._get.return_value = self.container.get_dict(self.entity_href,
type='rsa')
container = self.manager.get(container_ref=self.entity_href)
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -414,12 +415,12 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.manager.delete(container_ref=self.entity_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.delete.call_args
args, kwargs = self.api._delete.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
def test_should_delete_from_object(self):
self.api.get.return_value = self.container.get_dict(self.entity_href)
self.api._get.return_value = self.container.get_dict(self.entity_href)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsNotNone(container.container_ref)
@@ -427,7 +428,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container.delete()
# Verify the correct URL was used to make the call.
args, kwargs = self.api.delete.call_args
args, kwargs = self.api._delete.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -435,7 +436,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertIsNone(container.container_ref)
def test_should_store_after_delete_from_object(self):
self.api.get.return_value = self.container.get_dict(self.entity_href)
self.api._get.return_value = self.container.get_dict(self.entity_href)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsNotNone(container.container_ref)
@@ -443,7 +444,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container.delete()
# Verify the correct URL was used to make the call.
args, kwargs = self.api.delete.call_args
args, kwargs = self.api._delete.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -457,8 +458,8 @@ class WhenTestingContainers(test_client.BaseEntityResource):
def test_should_get_list(self):
container_resp = self.container.get_dict(self.entity_href)
self.api.get.return_value = {"containers":
[container_resp for v in range(3)]}
self.api._get.return_value = {"containers":
[container_resp for v in range(3)]}
containers_list = self.manager.list(limit=10, offset=5)
self.assertTrue(len(containers_list) == 3)
@@ -466,7 +467,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, containers_list[0].container_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_base[:-1], url)
@@ -483,7 +484,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertRaises(ValueError, self.manager.delete, None)
def test_should_register_consumer(self):
self.api.post.return_value = self.container.get_dict(
self.api._post.return_value = self.container.get_dict(
self.entity_href, consumers=[self.container.consumer]
)
container = self.manager.register_consumer(
@@ -493,7 +494,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_ref)
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
url, body = args[0], args[1]
self.assertEqual(self.consumers_post_resource, url)
@@ -506,9 +507,14 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.container.consumer.get('URL')
)
args, kwargs = self.api.delete.call_args
args, kwargs = self.api._delete.call_args
url = args[0]
body = kwargs['json']
self.assertEqual(self.consumers_delete_resource, url)
self.assertEqual(self.container.consumer, body)
def test_should_get_total(self):
self.api._get.return_value = {'total': 1}
total = self.manager.total()
self.assertEqual(total, 1)

37
barbicanclient/test/test_client_orders.py
View File

@@ -63,9 +63,9 @@ class WhenTestingOrders(test_client.BaseEntityResource):
self.assertIn('order_ref=' + self.entity_href, repr(order_obj))
def test_should_submit_via_constructor(self):
self.api.post.return_value = {'order_ref': self.entity_href}
self.api._post.return_value = {'order_ref': self.entity_href}
order = self.manager.Order(
order = self.manager.create(
name=self.order.secret.name,
algorithm=self.order.secret.algorithm,
payload_content_type=self.order.secret.content
@@ -75,7 +75,7 @@ class WhenTestingOrders(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, order_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -88,9 +88,9 @@ class WhenTestingOrders(test_client.BaseEntityResource):
order_req['secret']['payload_content_type'])
def test_should_submit_via_attributes(self):
self.api.post.return_value = {'order_ref': self.entity_href}
self.api._post.return_value = {'order_ref': self.entity_href}
order = self.manager.Order()
order = self.manager.create()
order.name = self.order.secret.name
order.algorithm = self.order.secret.algorithm
order.payload_content_type = self.order.secret.content
@@ -99,7 +99,7 @@ class WhenTestingOrders(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, order_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -112,9 +112,9 @@ class WhenTestingOrders(test_client.BaseEntityResource):
order_req['secret']['payload_content_type'])
def test_should_be_immutable_after_submit(self):
self.api.post.return_value = {'order_ref': self.entity_href}
self.api._post.return_value = {'order_ref': self.entity_href}
order = self.manager.Order(
order = self.manager.create(
name=self.order.secret.name,
algorithm=self.order.secret.algorithm,
payload_content_type=self.order.secret.content
@@ -136,7 +136,7 @@ class WhenTestingOrders(test_client.BaseEntityResource):
pass
def test_should_not_be_able_to_set_generated_attributes(self):
order = self.manager.Order()
order = self.manager.create()
# Verify that generated attributes cannot be set.
attributes = [
@@ -151,14 +151,14 @@ class WhenTestingOrders(test_client.BaseEntityResource):
pass
def test_should_get(self):
self.api.get.return_value = self.order.get_dict(self.entity_href)
self.api._get.return_value = self.order.get_dict(self.entity_href)
order = self.manager.Order(order_ref=self.entity_href)
order = self.manager.get(order_ref=self.entity_href)
self.assertIsInstance(order, orders.Order)
self.assertEqual(self.entity_href, order.order_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -166,14 +166,14 @@ class WhenTestingOrders(test_client.BaseEntityResource):
self.manager.delete(order_ref=self.entity_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.delete.call_args
args, kwargs = self.api._delete.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
def test_should_get_list(self):
order_resp = self.order.get_dict(self.entity_href)
self.api.get.return_value = {"orders":
[order_resp for v in range(3)]}
self.api._get.return_value = {"orders":
[order_resp for v in range(3)]}
orders_list = self.manager.list(limit=10, offset=5)
self.assertTrue(len(orders_list) == 3)
@@ -181,7 +181,7 @@ class WhenTestingOrders(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, orders_list[0].order_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_base[:-1], url)
@@ -192,3 +192,8 @@ class WhenTestingOrders(test_client.BaseEntityResource):
def test_should_fail_delete_no_href(self):
self.assertRaises(ValueError, self.manager.delete, None)
def test_should_get_total(self):
self.api._get.return_value = {'total': 1}
total = self.manager.total()
self.assertEqual(total, 1)

73
barbicanclient/test/test_client_secrets.py
View File

@@ -52,24 +52,24 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.manager = secrets.SecretManager(self.api)
def test_should_entity_str(self):
secret_obj = self.manager.Secret(name=self.secret.name)
secret_obj = self.manager.create(name=self.secret.name)
self.assertIn(self.secret.name, str(secret_obj))
def test_should_entity_repr(self):
secret_obj = self.manager.Secret(name=self.secret.name)
secret_obj = self.manager.create(name=self.secret.name)
self.assertIn('name="{0}"'.format(self.secret.name), repr(secret_obj))
def test_should_store_via_constructor(self):
self.api.post.return_value = {'secret_ref': self.entity_href}
self.api._post.return_value = {'secret_ref': self.entity_href}
secret = self.manager.Secret(name=self.secret.name,
secret = self.manager.create(name=self.secret.name,
payload=self.secret.payload,
payload_content_type=self.secret.content)
secret_href = secret.store()
self.assertEqual(self.entity_href, secret_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -81,9 +81,9 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
secret_req['payload_content_type'])
def test_should_store_via_attributes(self):
self.api.post.return_value = {'secret_ref': self.entity_href}
self.api._post.return_value = {'secret_ref': self.entity_href}
secret = self.manager.Secret()
secret = self.manager.create()
secret.name = self.secret.name
secret.payload = self.secret.payload
secret.payload_content_type = self.secret.content
@@ -91,7 +91,7 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, secret_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.post.call_args
args, kwargs = self.api._post.call_args
entity_resp = args[0]
self.assertEqual(self.entity, entity_resp)
@@ -103,9 +103,9 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
secret_req['payload_content_type'])
def test_should_be_immutable_after_submit(self):
self.api.post.return_value = {'secret_ref': self.entity_href}
self.api._post.return_value = {'secret_ref': self.entity_href}
secret = self.manager.Secret(name=self.secret.name,
secret = self.manager.create(name=self.secret.name,
payload=self.secret.payload,
payload_content_type=self.secret.content)
secret_href = secret.store()
@@ -125,7 +125,7 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
pass
def test_should_not_be_able_to_set_generated_attributes(self):
secret = self.manager.Secret()
secret = self.manager.create()
# Verify that generated attributes cannot be set.
attributes = [
@@ -139,24 +139,24 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
pass
def test_should_get(self):
self.api.get.return_value = self.secret.get_dict(self.entity_href)
self.api._get.return_value = self.secret.get_dict(self.entity_href)
secret = self.manager.Secret(secret_ref=self.entity_href)
secret = self.manager.get(secret_ref=self.entity_href)
self.assertIsInstance(secret, secrets.Secret)
self.assertEqual(self.entity_href, secret.secret_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
def test_should_decrypt_with_content_type(self):
self.api.get.return_value = self.secret.get_dict(self.entity_href)
self.api._get.return_value = self.secret.get_dict(self.entity_href)
decrypted = 'decrypted text here'
self.api.get_raw.return_value = decrypted
self.api._get_raw.return_value = decrypted
secret = self.manager.Secret(
secret = self.manager.get(
secret_ref=self.entity_href,
payload_content_type='application/octet-stream'
)
@@ -164,7 +164,7 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(decrypted, secret_payload)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get_raw.call_args
args, kwargs = self.api._get_raw.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -174,22 +174,22 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
def test_should_decrypt_without_content_type(self):
content_types_dict = {'default': 'application/octet-stream'}
self.api.get.return_value = self.secret.get_dict(self.entity_href,
content_types_dict)
self.api._get.return_value = self.secret.get_dict(self.entity_href,
content_types_dict)
decrypted = 'decrypted text here'
self.api.get_raw.return_value = decrypted
self.api._get_raw.return_value = decrypted
secret = self.manager.Secret(secret_ref=self.entity_href)
secret = self.manager.get(secret_ref=self.entity_href)
secret_payload = secret.payload
self.assertEqual(decrypted, secret_payload)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get_raw.call_args
args, kwargs = self.api._get_raw.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
@@ -201,14 +201,14 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.manager.delete(secret_ref=self.entity_href)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.delete.call_args
args, kwargs = self.api._delete.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
def test_should_get_list(self):
secret_resp = self.secret.get_dict(self.entity_href)
self.api.get.return_value = {"secrets":
[secret_resp for v in range(3)]}
self.api._get.return_value = {"secrets":
[secret_resp for v in range(3)]}
secrets_list = self.manager.list(limit=10, offset=5)
self.assertTrue(len(secrets_list) == 3)
@@ -216,7 +216,7 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, secrets_list[0].secret_ref)
# Verify the correct URL was used to make the call.
args, kwargs = self.api.get.call_args
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_base[:-1], url)
@@ -226,12 +226,12 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(5, params['offset'])
def test_should_fail_get_invalid_secret(self):
self.assertRaises(ValueError, self.manager.Secret,
self.assertRaises(ValueError, self.manager.get,
**{'secret_ref': '12345'})
def test_should_fail_decrypt_no_content_types(self):
self.api.get.return_value = self.secret.get_dict(self.entity_href)
secret = self.manager.Secret(secret_ref=self.entity_href)
self.api._get.return_value = self.secret.get_dict(self.entity_href)
secret = self.manager.get(secret_ref=self.entity_href)
try:
secret.payload
self.fail("didn't raise a ValueError exception")
@@ -240,9 +240,9 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
def test_should_fail_decrypt_no_default_content_type(self):
content_types_dict = {'no-default': 'application/octet-stream'}
self.api.get.return_value = self.secret.get_dict(self.entity_href,
content_types_dict)
secret = self.manager.Secret(secret_ref=self.entity_href)
self.api._get.return_value = self.secret.get_dict(self.entity_href,
content_types_dict)
secret = self.manager.get(secret_ref=self.entity_href)
try:
secret.payload
self.fail("didn't raise a ValueError exception")
@@ -251,3 +251,8 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
def test_should_fail_delete_no_href(self):
self.assertRaises(ValueError, self.manager.delete, None)
def test_should_get_total(self):
self.api._get.return_value = {'total': 1}
total = self.manager.total()
self.assertEqual(total, 1)