Part 1: Adding ACL support for Client API.
Added ACL client API support. Added ACL client API specific unit tests. Refactored client code as per meetup comments. Modified client get call to return data which can be looked by operation_type value. Moved add and remove users calls to cli as for client side, client will make updates directly in acl entity and submits changes to server. Change-Id: Ieb6d6a60919e2fdc311df605a51e88b87baa3c67
This commit is contained in:
Arun Kant
Arun Kant
parent
5700a0fea3
commit
5455c28853
|
|
@ -0,0 +1,475 @@
|
|||
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import logging
|
||||
|
||||
from oslo_utils.timeutils import parse_isotime
|
||||
|
||||
from barbicanclient import base
|
||||
from barbicanclient import formatter
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_OPERATION_TYPE = 'read'
|
||||
|
||||
VALID_ACL_OPERATIONS = ['read', 'write', 'delete', 'list']
|
||||
|
||||
|
||||
class ACLFormatter(formatter.EntityFormatter):
|
||||
|
||||
columns = ("Operation Type",
|
||||
"Project Access",
|
||||
"Users",
|
||||
"Created",
|
||||
"Updated",
|
||||
)
|
||||
|
||||
def _get_formatted_data(self):
|
||||
data = (self.operation_type,
|
||||
self.project_access,
|
||||
self.users,
|
||||
self.created,
|
||||
self.updated,
|
||||
self.acl_ref,
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
class _PerOperationACL(ACLFormatter):
|
||||
|
||||
def __init__(self, parent_acl, entity_ref=None, users=None,
|
||||
project_access=None, operation_type=None,
|
||||
created=None, updated=None):
|
||||
"""Per Operation ACL data instance for secret or container.
|
||||
|
||||
This class not to be instantiated outside of this module.
|
||||
|
||||
:param parent_acl: acl entity to this per operation data belongs to
|
||||
:param str entity_ref: Full HATEOAS reference to a secret or container
|
||||
:param users: List of Keystone userid(s) to be used for ACL.
|
||||
:type users: List or None
|
||||
:param bool project_access: Flag indicating project access behavior
|
||||
:param str operation_type: Type indicating which class of Barbican
|
||||
operations this ACL is defined for e.g. 'read' operations
|
||||
:param str created: Time string indicating ACL create timestamp. This
|
||||
is populated only when populating data from api response. Not
|
||||
needed in client input.
|
||||
:param str updated: Time string indicating ACL last update timestamp.
|
||||
This is populated only when populating data from api response. Not
|
||||
needed in client input.
|
||||
"""
|
||||
self._parent_acl = parent_acl
|
||||
self._entity_ref = entity_ref
|
||||
self._users = users if users else list()
|
||||
self._project_access = project_access
|
||||
self._operation_type = operation_type
|
||||
self._created = parse_isotime(created) if created else None
|
||||
self._updated = parse_isotime(updated) if updated else None
|
||||
|
||||
@property
|
||||
def acl_ref(self):
|
||||
return ACL.get_acl_ref_from_entity_ref(self.entity_ref)
|
||||
|
||||
@property
|
||||
def entity_ref(self):
|
||||
return self._entity_ref
|
||||
|
||||
@property
|
||||
def project_access(self):
|
||||
"""Flag indicating project access behavior is enabled or not"""
|
||||
return self._project_access
|
||||
|
||||
@property
|
||||
def users(self):
|
||||
"""List of users for this ACL setting"""
|
||||
return self._users
|
||||
|
||||
@property
|
||||
def operation_type(self):
|
||||
"""Type indicating class of Barbican operations for this ACL"""
|
||||
return self._operation_type
|
||||
|
||||
@property
|
||||
def created(self):
|
||||
return self._created
|
||||
|
||||
@property
|
||||
def updated(self):
|
||||
return self._updated
|
||||
|
||||
@operation_type.setter
|
||||
def operation_type(self, value):
|
||||
self._operation_type = value
|
||||
|
||||
@project_access.setter
|
||||
def project_access(self, value):
|
||||
self._project_access = value
|
||||
|
||||
@users.setter
|
||||
def users(self, value):
|
||||
self._users = value
|
||||
|
||||
def remove(self):
|
||||
"""
|
||||
Remove operation specific setting defined for a secret or container
|
||||
|
||||
:raises barbicanclient.exceptions.HTTPAuthError: 401 Responses
|
||||
:raises barbicanclient.exceptions.HTTPClientError: 4xx Responses
|
||||
"""
|
||||
LOG.debug('Removing {0} operation specific ACL for href: {1}'
|
||||
.format(self.operation_type, self.acl_ref))
|
||||
|
||||
self._parent_acl.load_acls_data()
|
||||
acl_entity = self._parent_acl
|
||||
|
||||
# Find matching operation specific acl entry and remove from list
|
||||
per_op_acl = acl_entity.get(self.operation_type)
|
||||
if per_op_acl:
|
||||
acl_entity.operation_acls.remove(per_op_acl)
|
||||
|
||||
# after above operation specific acl removal, check if there are
|
||||
# any remaining acls. If yes, then submit updates to server.
|
||||
# If not, then remove/delete acls from server.
|
||||
if acl_entity.operation_acls:
|
||||
acl_entity.submit()
|
||||
else:
|
||||
acl_entity.remove()
|
||||
|
||||
def _validate_users_type(self):
|
||||
if self.users and not (type(self.users) is list or
|
||||
type(self.users) is set):
|
||||
raise ValueError('Users value is expected to be provided'
|
||||
' as list/set.')
|
||||
|
||||
|
||||
class ACL(object):
|
||||
|
||||
_resource_name = 'acl'
|
||||
|
||||
def __init__(self, api, entity_ref, users=None, project_access=None,
|
||||
operation_type=DEFAULT_OPERATION_TYPE, created=None,
|
||||
updated=None):
|
||||
"""Base ACL entity instance for secret or container.
|
||||
|
||||
Provide ACL data arguments to set ACL setting for given operation_type.
|
||||
|
||||
To add ACL setting for other operation types, use `add_operation_acl`
|
||||
method.
|
||||
|
||||
:param api: client instance reference
|
||||
:param str entity_ref: Full HATEOAS reference to a secret or container
|
||||
:param users: List of Keystone userid(s) to be used for ACL.
|
||||
:type users: str List or None
|
||||
:param bool project_access: Flag indicating project access behavior
|
||||
:param str operation_type: Type indicating which class of Barbican
|
||||
operations this ACL is defined for e.g. 'read' operations
|
||||
:param str created: Time string indicating ACL create timestamp. This
|
||||
is populated only when populating data from api response. Not
|
||||
needed in client input.
|
||||
:param str updated: Time string indicating ACL last update timestamp.
|
||||
This is populated only when populating data from api response. Not
|
||||
needed in client input.
|
||||
"""
|
||||
|
||||
self._api = api
|
||||
self._entity_ref = entity_ref
|
||||
self._operation_acls = []
|
||||
|
||||
# create per operation ACL data entity only when client has set users
|
||||
# or project_access flag.
|
||||
if users is not None or project_access is not None:
|
||||
acl = _PerOperationACL(parent_acl=self, entity_ref=entity_ref,
|
||||
users=users, project_access=project_access,
|
||||
operation_type=operation_type,
|
||||
created=created, updated=updated)
|
||||
self._operation_acls.append(acl)
|
||||
|
||||
@property
|
||||
def entity_ref(self):
|
||||
"""Entity URI reference."""
|
||||
return self._entity_ref
|
||||
|
||||
@property
|
||||
def operation_acls(self):
|
||||
"""List of operation specific ACL settings."""
|
||||
return self._operation_acls
|
||||
|
||||
@property
|
||||
def acl_ref(self):
|
||||
return ACL.get_acl_ref_from_entity_ref(self.entity_ref)
|
||||
|
||||
def add_operation_acl(self, users=None, project_access=None,
|
||||
operation_type=None, created=None,
|
||||
updated=None,):
|
||||
"""Add ACL settings to entity for specific operation type.
|
||||
|
||||
If matching operation_type ACL already exists, then it replaces it with
|
||||
new PerOperationACL object using provided inputs. Otherwise it appends
|
||||
new PerOperationACL object to existing per operation ACL list.
|
||||
|
||||
This just adds to local entity and have not yet applied these changes
|
||||
to server.
|
||||
|
||||
:param users: List of Keystone userid(s) to be used in ACL.
|
||||
:type users: List or None
|
||||
:param bool project_access: Flag indicating project access behavior
|
||||
:param str operation_type: Type indicating which class of Barbican
|
||||
operations this ACL is defined for e.g. 'read' operations
|
||||
:param str created: Time string indicating ACL create timestamp. This
|
||||
is populated only when populating data from api response. Not
|
||||
needed in client input.
|
||||
:param str updated: Time string indicating ACL last update timestamp.
|
||||
This is populated only when populating data from api response. Not
|
||||
needed in client input.
|
||||
"""
|
||||
new_acl = _PerOperationACL(parent_acl=self, entity_ref=self.entity_ref,
|
||||
users=users, project_access=project_access,
|
||||
operation_type=operation_type,
|
||||
created=created, updated=updated)
|
||||
|
||||
for i, acl in enumerate(self._operation_acls):
|
||||
if acl.operation_type == operation_type:
|
||||
# replace with new ACL setting
|
||||
self._operation_acls[i] = new_acl
|
||||
break
|
||||
else:
|
||||
self._operation_acls.append(new_acl)
|
||||
|
||||
def _get_operation_acl(self, operation_type):
|
||||
return next((acl for acl in self._operation_acls
|
||||
if acl.operation_type == operation_type), None)
|
||||
|
||||
def get(self, operation_type):
|
||||
"""Get operation specific ACL instance.
|
||||
|
||||
:param str operation_type: Type indicating which operation's ACL
|
||||
setting is needed.
|
||||
"""
|
||||
return self._get_operation_acl(operation_type)
|
||||
|
||||
def __getattr__(self, name):
|
||||
if name in VALID_ACL_OPERATIONS:
|
||||
return self._get_operation_acl(name)
|
||||
else:
|
||||
raise AttributeError(name)
|
||||
|
||||
def submit(self):
|
||||
"""Submits ACLs for a secret or a container defined in server
|
||||
|
||||
In existing ACL case, this overwrites the existing ACL setting with
|
||||
provided inputs. If input users are None or empty list, this will
|
||||
remove existing ACL users if there. If input project_access flag is
|
||||
None, then default project access behavior is enabled.
|
||||
|
||||
:returns: str acl_ref: Full HATEOAS reference to a secret or container
|
||||
ACL.
|
||||
:raises barbicanclient.exceptions.HTTPAuthError: 401 Responses
|
||||
:raises barbicanclient.exceptions.HTTPClientError: 4xx Responses
|
||||
:raises barbicanclient.exceptions.HTTPServerError: 5xx Responses
|
||||
"""
|
||||
LOG.debug('Submitting complete {0} ACL for href: {1}'
|
||||
.format(self.acl_type, self.entity_ref))
|
||||
if not self.operation_acls:
|
||||
raise ValueError('ACL data for {0} is not provided.'.
|
||||
format(self._acl_type))
|
||||
|
||||
self.validate_input_ref()
|
||||
|
||||
acl_dict = {}
|
||||
|
||||
for per_op_acl in self.operation_acls:
|
||||
per_op_acl._validate_users_type()
|
||||
op_type = per_op_acl.operation_type
|
||||
acl_data = {}
|
||||
if per_op_acl.project_access is not None:
|
||||
acl_data['project-access'] = per_op_acl.project_access
|
||||
if per_op_acl.users is not None:
|
||||
acl_data['users'] = per_op_acl.users
|
||||
acl_dict[op_type] = acl_data
|
||||
|
||||
response = self._api.put(self.acl_ref, json=acl_dict)
|
||||
|
||||
return response.json().get('acl_ref')
|
||||
|
||||
def remove(self):
|
||||
"""
|
||||
Remove Barbican ACLs setting defined for a secret or container
|
||||
|
||||
:raises barbicanclient.exceptions.HTTPAuthError: 401 Responses
|
||||
:raises barbicanclient.exceptions.HTTPClientError: 4xx Responses
|
||||
"""
|
||||
self.validate_input_ref()
|
||||
LOG.debug('Removing ACL for {0} for href: {1}'
|
||||
.format(self.acl_type, self.entity_ref))
|
||||
self._api.delete(self.acl_ref)
|
||||
|
||||
def load_acls_data(self):
|
||||
"""Loads ACL entity from Barbican server using its acl_ref
|
||||
|
||||
Clears the existing list of per operation ACL settings if there.
|
||||
Populates current ACL entity with ACL settings received from Barbican
|
||||
server.
|
||||
|
||||
:raises barbicanclient.exceptions.HTTPAuthError: 401 Responses
|
||||
:raises barbicanclient.exceptions.HTTPClientError: 4xx Responses
|
||||
:raises barbicanclient.exceptions.HTTPServerError: 5xx Responses
|
||||
"""
|
||||
|
||||
response = self._api.get(self.acl_ref)
|
||||
|
||||
del self.operation_acls[:] # clearing list for all of its references
|
||||
for op_type in response:
|
||||
acl_dict = response.get(op_type)
|
||||
proj_access = acl_dict.get('project-access')
|
||||
users = acl_dict.get('users')
|
||||
created = acl_dict.get('created')
|
||||
updated = acl_dict.get('updated')
|
||||
self.add_operation_acl(operation_type=op_type,
|
||||
project_access=proj_access,
|
||||
users=users, created=created,
|
||||
updated=updated)
|
||||
|
||||
def validate_input_ref(self):
|
||||
res_title = self._acl_type.title()
|
||||
if not self.entity_ref:
|
||||
raise ValueError('{0} href is required.'.format(res_title))
|
||||
if self._parent_entity_path in self.entity_ref:
|
||||
if '/acl' in self.entity_ref:
|
||||
raise ValueError('{0} ACL URI provided. Expecting {0} URI.'
|
||||
.format(res_title))
|
||||
ref_type = self._acl_type
|
||||
else:
|
||||
raise ValueError('{0} URI is not specified.'.format(res_title))
|
||||
|
||||
base.validate_ref(self.entity_ref, ref_type)
|
||||
return ref_type
|
||||
|
||||
@staticmethod
|
||||
def get_acl_ref_from_entity_ref(entity_ref):
|
||||
# Utility for converting entity ref to acl ref
|
||||
if entity_ref:
|
||||
entity_ref = entity_ref.rstrip('/')
|
||||
return '{0}/{1}'.format(entity_ref, ACL._resource_name)
|
||||
|
||||
@staticmethod
|
||||
def identify_ref_type(entity_ref):
|
||||
# Utility for identifying ACL type from given entity URI.
|
||||
if not entity_ref:
|
||||
raise ValueError('Secret or container href is required.')
|
||||
if '/secrets' in entity_ref:
|
||||
ref_type = 'secret'
|
||||
elif '/containers' in entity_ref:
|
||||
ref_type = 'container'
|
||||
else:
|
||||
raise ValueError('Secret or container URI is not specified.')
|
||||
|
||||
return ref_type
|
||||
|
||||
|
||||
class SecretACL(ACL):
|
||||
"""ACL entity for a secret"""
|
||||
|
||||
columns = ACLFormatter.columns + ("Secret ACL Ref",)
|
||||
_acl_type = 'secret'
|
||||
_parent_entity_path = '/secrets'
|
||||
|
||||
@property
|
||||
def acl_type(self):
|
||||
return self._acl_type
|
||||
|
||||
|
||||
class ContainerACL(ACL):
|
||||
"""ACL entity for a container"""
|
||||
|
||||
columns = ACLFormatter.columns + ("Container ACL Ref",)
|
||||
_acl_type = 'container'
|
||||
_parent_entity_path = '/containers'
|
||||
|
||||
@property
|
||||
def acl_type(self):
|
||||
return self._acl_type
|
||||
|
||||
|
||||
class ACLManager(base.BaseEntityManager):
|
||||
"""Entity Manager for Secret or Container ACL entities"""
|
||||
|
||||
acl_class_map = {
|
||||
'secret': SecretACL,
|
||||
'container': ContainerACL
|
||||
}
|
||||
|
||||
def __init__(self, api):
|
||||
super(ACLManager, self).__init__(api, ACL._resource_name)
|
||||
|
||||
def create(self, entity_ref=None, users=None, project_access=None,
|
||||
operation_type=DEFAULT_OPERATION_TYPE):
|
||||
"""
|
||||
Factory method for creating `ACL` entity.
|
||||
|
||||
`ACL` object returned by this method have not yet been
|
||||
stored in Barbican.
|
||||
|
||||
Input entity_ref is used to determine whether
|
||||
ACL object type needs to be :class:`barbicanclient.acls.SecretACL`
|
||||
or :class:`barbicanclient.acls.ContainerACL`.
|
||||
|
||||
:param str entity_ref: Full HATEOAS reference to a secret or container
|
||||
:param users: List of Keystone userid(s) to be used in ACL.
|
||||
:type users: List or None
|
||||
:param bool project_access: Flag indicating project access behavior
|
||||
:param str operation_type: Type indicating which class of Barbican
|
||||
operations this ACL is defined for e.g. 'read' operations
|
||||
:returns: ACL object instance
|
||||
:rtype: :class:`barbicanclient.acls.SecretACL` or
|
||||
:class:`barbicanclient.acls.ContainerACL`
|
||||
"""
|
||||
entity_type = ACL.identify_ref_type(entity_ref)
|
||||
|
||||
entity_class = ACLManager.acl_class_map.get(entity_type)
|
||||
# entity_class cannot be None as entity_ref is already validated above
|
||||
return entity_class(api=self._api, entity_ref=entity_ref, users=users,
|
||||
project_access=project_access,
|
||||
operation_type=operation_type)
|
||||
|
||||
def get(self, entity_ref):
|
||||
"""
|
||||
Retrieve existing ACLs for a secret or container defined in Barbican
|
||||
|
||||
:param str entity_ref: Full HATEOAS reference to a secret or container.
|
||||
:returns: ACL entity object instance
|
||||
:rtype: :class:`barbicanclient.acls.SecretACL` or
|
||||
:class:`barbicanclient.acls.ContainerACL`
|
||||
:raises barbicanclient.exceptions.HTTPAuthError: 401 Responses
|
||||
:raises barbicanclient.exceptions.HTTPClientError: 4xx Responses
|
||||
"""
|
||||
entity = self._validate_acl_ref(entity_ref)
|
||||
LOG.debug('Getting ACL for {0} href: {1}'
|
||||
.format(entity.acl_type, entity.acl_ref))
|
||||
entity.load_acls_data()
|
||||
return entity
|
||||
|
||||
def _validate_acl_ref(self, entity_ref):
|
||||
if entity_ref is None:
|
||||
raise ValueError('Expected secret or container URI is not '
|
||||
'specified.')
|
||||
|
||||
entity_ref = entity_ref.rstrip('/')
|
||||
entity_type = ACL.identify_ref_type(entity_ref)
|
||||
|
||||
entity_class = ACLManager.acl_class_map.get(entity_type)
|
||||
acl_entity = entity_class(api=self._api, entity_ref=entity_ref)
|
||||
acl_entity.validate_input_ref()
|
||||
return acl_entity
|
||||
|
|
@ -20,6 +20,7 @@ from keystoneclient import adapter
|
|||
from keystoneclient.auth.base import BaseAuthPlugin
|
||||
from keystoneclient import session as ks_session
|
||||
|
||||
from barbicanclient import acls
|
||||
from barbicanclient import cas
|
||||
from barbicanclient import containers
|
||||
from barbicanclient import exceptions
|
||||
|
|
@ -173,6 +174,7 @@ class Client(object):
|
|||
self.orders = orders.OrderManager(httpclient)
|
||||
self.containers = containers.ContainerManager(httpclient)
|
||||
self.cas = cas.CAManager(httpclient)
|
||||
self.acls = acls.ACLManager(httpclient)
|
||||
|
||||
|
||||
def env(*vars, **kwargs):
|
||||
|
|
|
|||
|
|
@ -0,0 +1,457 @@
|
|||
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from oslo_utils import timeutils
|
||||
import requests_mock
|
||||
|
||||
from barbicanclient import acls
|
||||
from barbicanclient.tests import test_client
|
||||
|
||||
|
||||
class ACLTestCase(test_client.BaseEntityResource):
|
||||
|
||||
def setUp(self):
|
||||
self._setUp('acl', entity_id='d9f95d61-8863-49d3-a045-5c2cb77130b5')
|
||||
|
||||
self.secret_ref = (self.endpoint +
|
||||
'/secrets/8a3108ec-88fc-4f5c-86eb-f37b8ae8358e')
|
||||
|
||||
self.secret_acl_ref = '{0}/acl'.format(self.secret_ref)
|
||||
self.container_ref = (self.endpoint + '/containers/'
|
||||
'83c302c7-86fe-4f07-a277-c4962f121f19')
|
||||
self.container_acl_ref = '{0}/acl'.format(self.container_ref)
|
||||
|
||||
self.manager = self.client.acls
|
||||
self.users1 = ['2d0ee7c681cc4549b6d76769c320d91f',
|
||||
'721e27b8505b499e8ab3b38154705b9e']
|
||||
self.users2 = ['2d0ee7c681cc4549b6d76769c320d91f']
|
||||
self.created = str(timeutils.utcnow())
|
||||
|
||||
def get_acl_response_data(self, operation_type='read',
|
||||
users=None,
|
||||
project_access=False):
|
||||
if users is None:
|
||||
users = self.users1
|
||||
op_data = {'users': users}
|
||||
op_data['project-access'] = project_access
|
||||
op_data['created'] = self.created
|
||||
op_data['updated'] = str(timeutils.utcnow())
|
||||
acl_data = {operation_type: op_data}
|
||||
return acl_data
|
||||
|
||||
|
||||
class WhenTestingACLManager(ACLTestCase):
|
||||
|
||||
def test_should_get_secret_acl(self):
|
||||
self.responses.get(self.secret_acl_ref,
|
||||
json=self.get_acl_response_data())
|
||||
|
||||
api_resp = self.manager.get(entity_ref=self.secret_ref)
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertEqual(False, api_resp.get('read').project_access)
|
||||
self.assertEqual('read', api_resp.get('read').operation_type)
|
||||
self.assertEqual(self.secret_acl_ref, api_resp.get('read').acl_ref)
|
||||
|
||||
def test_should_get_secret_acl_with_extra_trailing_slashes(self):
|
||||
self.responses.get(requests_mock.ANY,
|
||||
json=self.get_acl_response_data())
|
||||
# check if trailing slashes are corrected in get call.
|
||||
self.manager.get(entity_ref=self.secret_ref + '///')
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
|
||||
def test_should_get_container_acl(self):
|
||||
self.responses.get(self.container_acl_ref,
|
||||
json=self.get_acl_response_data())
|
||||
|
||||
api_resp = self.manager.get(entity_ref=self.container_ref)
|
||||
self.assertEqual(self.container_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertEqual(False, api_resp.get('read').project_access)
|
||||
self.assertEqual('read', api_resp.get('read').operation_type)
|
||||
self.assertEqual(self.container_acl_ref, api_resp.get('read').acl_ref)
|
||||
|
||||
def test_should_get_container_acl_with_trailing_slashes(self):
|
||||
self.responses.get(requests_mock.ANY,
|
||||
json=self.get_acl_response_data())
|
||||
# check if trailing slashes are corrected in get call.
|
||||
self.manager.get(entity_ref=self.container_ref + '///')
|
||||
self.assertEqual(self.container_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
|
||||
def test_should_fail_get_no_href(self):
|
||||
self.assertRaises(ValueError, self.manager.get, None)
|
||||
|
||||
def test_should_fail_get_invalid_uri(self):
|
||||
# secret_acl URI expected and not secret URI
|
||||
self.assertRaises(ValueError, self.manager.get, self.secret_acl_ref)
|
||||
|
||||
self.assertRaises(ValueError, self.manager.get,
|
||||
self.endpoint + '/containers/consumers')
|
||||
|
||||
def test_should_create_secret_acl(self):
|
||||
entity = self.manager.create(entity_ref=self.secret_ref + '///',
|
||||
users=self.users1, project_access=True)
|
||||
self.assertIsInstance(entity, acls.SecretACL)
|
||||
|
||||
read_acl = entity.read
|
||||
# entity ref is kept same as provided input.
|
||||
self.assertEqual(self.secret_ref + '///', read_acl.entity_ref)
|
||||
self.assertEqual(True, read_acl.project_access)
|
||||
self.assertEqual(self.users1, read_acl.users)
|
||||
self.assertEqual(acls.DEFAULT_OPERATION_TYPE, read_acl.operation_type)
|
||||
# acl ref removes extra trailing slashes if there
|
||||
self.assertIn(self.secret_ref, read_acl.acl_ref,
|
||||
'ACL ref has additional /acl')
|
||||
self.assertIsNone(read_acl.created)
|
||||
self.assertIsNone(read_acl.updated)
|
||||
|
||||
read_acl_via_get = entity.get('read')
|
||||
self.assertEqual(read_acl, read_acl_via_get)
|
||||
|
||||
def test_should_create_acl_with_users(self):
|
||||
entity = self.manager.create(entity_ref=self.container_ref + '///',
|
||||
users=self.users2, project_access=False)
|
||||
self.assertIsInstance(entity, acls.ContainerACL)
|
||||
# entity ref is kept same as provided input.
|
||||
self.assertEqual(self.container_ref + '///', entity.entity_ref)
|
||||
|
||||
read_acl = entity.read
|
||||
self.assertEqual(False, read_acl.project_access)
|
||||
self.assertEqual(self.users2, read_acl.users)
|
||||
self.assertEqual(acls.DEFAULT_OPERATION_TYPE, read_acl.operation_type)
|
||||
# acl ref removes extra trailing slashes if there
|
||||
self.assertIn(self.container_ref, read_acl.acl_ref,
|
||||
'ACL ref has additional /acl')
|
||||
|
||||
def test_should_create_acl_with_no_users(self):
|
||||
entity = self.manager.create(entity_ref=self.container_ref, users=[])
|
||||
read_acl = entity.read
|
||||
self.assertEqual([], read_acl.users)
|
||||
self.assertEqual(acls.DEFAULT_OPERATION_TYPE, read_acl.operation_type)
|
||||
self.assertIsNone(read_acl.project_access)
|
||||
|
||||
read_acl_via_get = entity.get('read')
|
||||
self.assertEqual(read_acl, read_acl_via_get)
|
||||
|
||||
def test_create_no_acl_settings(self):
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref)
|
||||
self.assertEqual([], entity.operation_acls)
|
||||
self.assertEqual(self.container_ref, entity.entity_ref)
|
||||
self.assertEqual(self.container_ref + '/acl', entity.acl_ref)
|
||||
|
||||
def test_should_fail_create_invalid_uri(self):
|
||||
|
||||
self.assertRaises(ValueError, self.manager.create,
|
||||
self.endpoint + '/orders')
|
||||
self.assertRaises(ValueError, self.manager.create, None)
|
||||
|
||||
|
||||
class WhenTestingACLEntity(ACLTestCase):
|
||||
|
||||
def test_should_submit_acl_with_users_project_access_set(self):
|
||||
data = {'acl_ref': self.secret_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.secret_acl_ref, json=data)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref + '///',
|
||||
users=self.users1, project_access=True)
|
||||
api_resp = entity.submit()
|
||||
self.assertEqual(self.secret_acl_ref, api_resp)
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
|
||||
def test_should_submit_acl_with_project_access_set_but_no_users(self):
|
||||
data = {'acl_ref': self.secret_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.secret_acl_ref, json=data)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
project_access=False)
|
||||
api_resp = entity.submit()
|
||||
self.assertEqual(self.secret_acl_ref, api_resp)
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertFalse(entity.read.project_access)
|
||||
self.assertEqual([], entity.get('read').users)
|
||||
|
||||
def test_should_submit_acl_with_user_set_but_not_project_access(self):
|
||||
data = {'acl_ref': self.container_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.container_acl_ref, json=data)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref,
|
||||
users=self.users2)
|
||||
api_resp = entity.submit()
|
||||
self.assertEqual(self.container_acl_ref, api_resp)
|
||||
self.assertEqual(self.container_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertEqual(self.users2, entity.read.users)
|
||||
self.assertIsNone(entity.get('read').project_access)
|
||||
|
||||
def test_should_fail_submit_acl_invalid_secret_uri(self):
|
||||
data = {'acl_ref': self.secret_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.secret_acl_ref, json=data)
|
||||
entity = self.manager.create(entity_ref=self.secret_acl_ref + '///',
|
||||
users=self.users1, project_access=True)
|
||||
# Submit checks provided URI is entity URI and not ACL URI.
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users=self.users1, project_access=True)
|
||||
entity._entity_ref = None
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users=self.users1, project_access=True)
|
||||
entity._entity_ref = self.container_ref # expected secret uri here
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
def test_should_fail_submit_acl_invalid_container_uri(self):
|
||||
"""Adding tests for container URI validation.
|
||||
|
||||
Container URI validation is different from secret URI validation.
|
||||
That's why adding separate tests for code coverage.
|
||||
"""
|
||||
|
||||
data = {'acl_ref': self.container_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.container_acl_ref, json=data)
|
||||
entity = self.manager.create(entity_ref=self.container_acl_ref + '///',
|
||||
users=self.users1, project_access=True)
|
||||
# Submit checks provided URI is entity URI and not ACL URI.
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref,
|
||||
users=self.users1, project_access=True)
|
||||
entity._entity_ref = None
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref,
|
||||
users=self.users1, project_access=True)
|
||||
entity._entity_ref = self.secret_ref # expected container uri here
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
def test_should_fail_submit_acl_no_acl_data(self):
|
||||
data = {'acl_ref': self.secret_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.secret_acl_ref, json=data)
|
||||
entity = self.manager.create(entity_ref=self.secret_ref + '///')
|
||||
# Submit checks that ACL setting data is there or not.
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
def test_should_fail_submit_acl_input_users_as_not_list(self):
|
||||
data = {'acl_ref': self.secret_acl_ref}
|
||||
# register put acl URI with expected acl ref in response
|
||||
self.responses.put(self.secret_acl_ref, json=data)
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users='u1')
|
||||
# Submit checks that input users are provided as list or not
|
||||
self.assertRaises(ValueError, entity.submit)
|
||||
|
||||
def test_should_load_acls_data(self):
|
||||
self.responses.get(
|
||||
self.container_acl_ref, json=self.get_acl_response_data(
|
||||
users=self.users2, project_access=True))
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref,
|
||||
users=self.users1)
|
||||
self.assertEqual(self.container_ref, entity.entity_ref)
|
||||
self.assertEqual(self.container_acl_ref, entity.acl_ref)
|
||||
|
||||
entity.load_acls_data()
|
||||
|
||||
self.assertEqual(self.users2, entity.read.users)
|
||||
self.assertTrue(entity.get('read').project_access)
|
||||
self.assertEqual(timeutils.parse_isotime(self.created),
|
||||
entity.read.created)
|
||||
self.assertEqual(timeutils.parse_isotime(self.created),
|
||||
entity.get('read').created)
|
||||
|
||||
self.assertEqual(1, len(entity.operation_acls))
|
||||
self.assertEqual(self.container_acl_ref, entity.get('read').acl_ref)
|
||||
self.assertEqual(self.container_ref, entity.read.entity_ref)
|
||||
|
||||
def test_should_add_operation_acl(self):
|
||||
entity = self.manager.create(entity_ref=self.secret_ref + '///',
|
||||
users=self.users1, project_access=True)
|
||||
self.assertIsInstance(entity, acls.SecretACL)
|
||||
|
||||
entity.add_operation_acl(users=self.users2, project_access=False,
|
||||
operation_type='read')
|
||||
|
||||
read_acl = entity.read
|
||||
# entity ref is kept same as provided input.
|
||||
self.assertEqual(self.secret_ref + '/acl', read_acl.acl_ref)
|
||||
self.assertFalse(read_acl.project_access)
|
||||
self.assertEqual(self.users2, read_acl.users)
|
||||
self.assertEqual(acls.DEFAULT_OPERATION_TYPE, read_acl.operation_type)
|
||||
|
||||
entity.add_operation_acl(users=[], project_access=False,
|
||||
operation_type='dummy')
|
||||
dummy_acl = entity.get('dummy')
|
||||
self.assertFalse(dummy_acl.project_access)
|
||||
self.assertEqual([], dummy_acl.users)
|
||||
|
||||
def test_acl_entity_properties(self):
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users=self.users2)
|
||||
self.assertEqual(self.secret_ref, entity.entity_ref)
|
||||
self.assertEqual(self.secret_acl_ref, entity.acl_ref)
|
||||
|
||||
self.assertEqual(self.users2, entity.read.users)
|
||||
self.assertEqual(self.users2, entity.get('read').users)
|
||||
self.assertIsNone(entity.read.project_access)
|
||||
self.assertIsNone(entity.get('read').project_access)
|
||||
self.assertIsNone(entity.read.created)
|
||||
self.assertIsNone(entity.get('read').created)
|
||||
self.assertEqual('read', entity.read.operation_type)
|
||||
self.assertEqual('read', entity.get('read').operation_type)
|
||||
|
||||
self.assertEqual(1, len(entity.operation_acls))
|
||||
self.assertEqual(self.secret_acl_ref, entity.read.acl_ref)
|
||||
self.assertEqual(self.secret_acl_ref, entity.get('read').acl_ref)
|
||||
self.assertEqual(self.secret_ref, entity.read.entity_ref)
|
||||
|
||||
self.assertIsNone(entity.get('dummyOperation'))
|
||||
|
||||
entity.read.users = ['u1']
|
||||
entity.read.project_access = False
|
||||
entity.read.operation_type = 'my_operation'
|
||||
self.assertFalse(entity.get('my_operation').project_access)
|
||||
self.assertEqual(['u1'], entity.get('my_operation').users)
|
||||
|
||||
self.assertRaises(AttributeError, lambda x: x.dummy_operation, entity)
|
||||
|
||||
def test_get_formatted_data(self):
|
||||
|
||||
s_entity = acls.SecretACL(api=None,
|
||||
entity_ref=self.secret_ref,
|
||||
users=self.users1)
|
||||
|
||||
data = s_entity.read._get_formatted_data()
|
||||
|
||||
self.assertEqual(acls.DEFAULT_OPERATION_TYPE, data[0])
|
||||
self.assertIsNone(data[1])
|
||||
self.assertEqual(self.users1, data[2])
|
||||
self.assertIsNone(data[3]) # created
|
||||
self.assertIsNone(data[4]) # updated
|
||||
self.assertEqual(self.secret_acl_ref, data[5])
|
||||
|
||||
c_entity = acls.ContainerACL(api=None,
|
||||
entity_ref=self.container_ref,
|
||||
users=self.users2, created=self.created)
|
||||
|
||||
data = c_entity.get('read')._get_formatted_data()
|
||||
|
||||
self.assertEqual(acls.DEFAULT_OPERATION_TYPE, data[0])
|
||||
self.assertIsNone(data[1])
|
||||
self.assertEqual(self.users2, data[2])
|
||||
self.assertEqual(timeutils.parse_isotime(self.created),
|
||||
data[3]) # created
|
||||
self.assertIsNone(data[4]) # updated
|
||||
self.assertEqual(self.container_acl_ref, data[5])
|
||||
|
||||
def test_should_secret_acl_remove(self):
|
||||
self.responses.delete(self.secret_acl_ref)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users=self.users2)
|
||||
|
||||
api_resp = entity.remove()
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertIsNone(api_resp)
|
||||
|
||||
def test_should_secret_acl_remove_uri_with_slashes(self):
|
||||
self.responses.delete(self.secret_acl_ref)
|
||||
|
||||
# check if trailing slashes are corrected in delete call.
|
||||
entity = self.manager.create(entity_ref=self.secret_ref + '///',
|
||||
users=self.users2)
|
||||
entity.remove()
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
|
||||
self.responses.delete(self.container_acl_ref)
|
||||
|
||||
def test_should_container_acl_remove(self):
|
||||
self.responses.delete(self.container_acl_ref)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref)
|
||||
entity.remove()
|
||||
self.assertEqual(self.container_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
|
||||
def test_should_fail_acl_remove_invalid_uri(self):
|
||||
# secret_acl URI expected and not secret acl URI
|
||||
entity = self.manager.create(entity_ref=self.secret_acl_ref)
|
||||
self.assertRaises(ValueError, entity.remove)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_acl_ref)
|
||||
self.assertRaises(ValueError, entity.remove)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.container_ref +
|
||||
'/consumers')
|
||||
self.assertRaises(ValueError, entity.remove)
|
||||
|
||||
# check to make sure UUID is passed in
|
||||
entity = self.manager.create(entity_ref=self.endpoint + '/secrets' +
|
||||
'/consumers')
|
||||
self.assertRaises(ValueError, entity.remove)
|
||||
|
||||
def test_should_per_operation_acl_remove(self):
|
||||
self.responses.get(self.secret_acl_ref,
|
||||
json=self.get_acl_response_data(users=self.users2,
|
||||
project_access=True)
|
||||
)
|
||||
self.responses.delete(self.secret_acl_ref)
|
||||
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users=self.users2)
|
||||
|
||||
api_resp = entity.read.remove()
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertIsNone(api_resp)
|
||||
self.assertEqual(0, len(entity.operation_acls))
|
||||
|
||||
# now try case where there are 2 operation acls defined
|
||||
# and one operation specific acl is removed. In that case,
|
||||
# entity.submit() is called instead of remove to update rest of entity
|
||||
|
||||
acl_data = self.get_acl_response_data(users=self.users2,
|
||||
project_access=True)
|
||||
|
||||
data = self.get_acl_response_data(users=self.users1,
|
||||
operation_type='write',
|
||||
project_access=False)
|
||||
acl_data['write'] = data['write']
|
||||
|
||||
self.responses.get(self.secret_acl_ref, json=acl_data)
|
||||
self.responses.put(self.secret_acl_ref, json={})
|
||||
# check if trailing slashes are corrected in delete call.
|
||||
entity = self.manager.create(entity_ref=self.secret_ref,
|
||||
users=self.users2)
|
||||
entity.read.remove()
|
||||
self.assertEqual(self.secret_acl_ref,
|
||||
self.responses.last_request.url)
|
||||
self.assertEqual(1, len(entity.operation_acls))
|
||||
self.assertEqual('write', entity.operation_acls[0].operation_type)
|
||||
self.assertEqual(self.users1, entity.operation_acls[0].users)
|
||||
Loading…
Reference in New Issue