Merge "Convert test_identity_roles to requests mock"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
67806a7ea9
@ -58,6 +58,7 @@ _ServiceData = collections.namedtuple(
|
||||
'service_id, service_name, service_type, description, enabled, '
|
||||
'json_response_v3, json_response_v2, json_request')
|
||||
|
||||
|
||||
_EndpointDataV3 = collections.namedtuple(
|
||||
'EndpointData',
|
||||
'endpoint_id, service_id, interface, region, url, enabled, '
|
||||
@ -71,6 +72,13 @@ _EndpointDataV2 = collections.namedtuple(
|
||||
'json_request')
|
||||
|
||||
|
||||
# NOTE(notmorgan): Shade does not support domain-specific roles
|
||||
# This should eventually be fixed if it becomes a main-stream feature.
|
||||
_RoleData = collections.namedtuple(
|
||||
'RoleData',
|
||||
'role_id, role_name, json_response, json_request')
|
||||
|
||||
|
||||
class BaseTestCase(base.TestCase):
|
||||
|
||||
def setUp(self, cloud_config_fixture='clouds.yaml'):
|
||||
@ -166,16 +174,20 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
self.__register_uris_called = False
|
||||
|
||||
def get_mock_url(self, service_type, interface, resource=None,
|
||||
append=None, base_url_append=None):
|
||||
append=None, base_url_append=None,
|
||||
qs_elements=None):
|
||||
endpoint_url = self.cloud.endpoint_for(
|
||||
service_type=service_type, interface=interface)
|
||||
to_join = [endpoint_url]
|
||||
qs = ''
|
||||
if base_url_append:
|
||||
to_join.append(base_url_append)
|
||||
if resource:
|
||||
to_join.append(resource)
|
||||
to_join.extend(append or [])
|
||||
return '/'.join(to_join)
|
||||
if qs_elements is not None:
|
||||
qs = '?%s' % '&'.join(qs_elements)
|
||||
return '%(uri)s%(qs)s' % {'uri': '/'.join(to_join), 'qs': qs}
|
||||
|
||||
def mock_for_keystone_projects(self, project=None, v3=True,
|
||||
list_get=False, id_get=False,
|
||||
@ -228,12 +240,12 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
return project_list
|
||||
|
||||
def _get_project_data(self, project_name=None, enabled=None,
|
||||
description=None, v3=True):
|
||||
domain_id=None, description=None, v3=True):
|
||||
project_name = project_name or self.getUniqueString('projectName')
|
||||
project_id = uuid.uuid4().hex
|
||||
response = {'id': project_id, 'name': project_name}
|
||||
request = {'name': project_name}
|
||||
domain_id = uuid.uuid4().hex if v3 else None
|
||||
domain_id = (domain_id or uuid.uuid4().hex) if v3 else None
|
||||
if domain_id:
|
||||
request['domain_id'] = domain_id
|
||||
response['domain_id'] = domain_id
|
||||
@ -374,6 +386,15 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
internal_url, admin_url, v3_endpoints,
|
||||
{'endpoint': response}, {'endpoint': request})
|
||||
|
||||
def _get_role_data(self, role_name=None):
|
||||
role_id = uuid.uuid4().hex
|
||||
role_name = role_name or uuid.uuid4().hex
|
||||
request = {'name': role_name}
|
||||
response = request.copy()
|
||||
response['id'] = role_id
|
||||
return _RoleData(role_id, role_name, {'role': response},
|
||||
{'role': request})
|
||||
|
||||
def use_keystone_v3(self):
|
||||
self.adapter = self.useFixture(rm_fixture.Fixture())
|
||||
self.calls = []
|
||||
|
||||
@ -11,15 +11,11 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import mock
|
||||
import testtools
|
||||
|
||||
import os_client_config as occ
|
||||
import shade
|
||||
from shade import meta
|
||||
from shade import _utils
|
||||
from shade.tests.unit import base
|
||||
from shade.tests import fakes
|
||||
from testtools import matchers
|
||||
|
||||
|
||||
RAW_ROLE_ASSIGNMENTS = [
|
||||
@ -38,135 +34,264 @@ RAW_ROLE_ASSIGNMENTS = [
|
||||
]
|
||||
|
||||
|
||||
class TestIdentityRoles(base.TestCase):
|
||||
class TestIdentityRoles(base.RequestsMockTestCase):
|
||||
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_roles(self, mock_keystone):
|
||||
def get_mock_url(self, service_type='identity', interface='admin',
|
||||
resource='roles', append=None, base_url_append='v3',
|
||||
qs_elements=None):
|
||||
return super(TestIdentityRoles, self).get_mock_url(
|
||||
service_type, interface, resource, append, base_url_append,
|
||||
qs_elements)
|
||||
|
||||
def test_list_roles(self):
|
||||
self._add_discovery_uri_call()
|
||||
role_data = self._get_role_data()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'roles': [role_data.json_response['role']]})
|
||||
])
|
||||
self.op_cloud.list_roles()
|
||||
self.assertTrue(mock_keystone.roles.list.called)
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_get_role(self, mock_keystone):
|
||||
role_obj = fakes.FakeRole(id='1234', name='fake_role')
|
||||
mock_keystone.roles.list.return_value = [role_obj]
|
||||
def test_get_role_by_name(self):
|
||||
self._add_discovery_uri_call()
|
||||
role_data = self._get_role_data()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'roles': [role_data.json_response['role']]})
|
||||
])
|
||||
role = self.op_cloud.get_role(role_data.role_name)
|
||||
|
||||
role = self.op_cloud.get_role('fake_role')
|
||||
|
||||
self.assertTrue(mock_keystone.roles.list.called)
|
||||
self.assertIsNotNone(role)
|
||||
self.assertEqual('1234', role['id'])
|
||||
self.assertEqual('fake_role', role['name'])
|
||||
self.assertThat(role.id, matchers.Equals(role_data.role_id))
|
||||
self.assertThat(role.name, matchers.Equals(role_data.role_name))
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_create_role(self, mock_keystone):
|
||||
role_name = 'tootsie_roll'
|
||||
role_obj = fakes.FakeRole(id='1234', name=role_name)
|
||||
mock_keystone.roles.create.return_value = role_obj
|
||||
def test_get_role_by_id(self):
|
||||
self._add_discovery_uri_call()
|
||||
role_data = self._get_role_data()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'roles': [role_data.json_response['role']]})
|
||||
])
|
||||
role = self.op_cloud.get_role(role_data.role_id)
|
||||
|
||||
role = self.op_cloud.create_role(role_name)
|
||||
|
||||
mock_keystone.roles.create.assert_called_once_with(
|
||||
name=role_name
|
||||
)
|
||||
self.assertIsNotNone(role)
|
||||
self.assertEqual(role_name, role['name'])
|
||||
self.assertThat(role.id, matchers.Equals(role_data.role_id))
|
||||
self.assertThat(role.name, matchers.Equals(role_data.role_name))
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(shade.OperatorCloud, 'get_role')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_delete_role(self, mock_keystone, mock_get):
|
||||
role_obj = fakes.FakeRole(id='1234', name='aaa')
|
||||
mock_get.return_value = meta.obj_to_dict(role_obj)
|
||||
self.assertTrue(self.op_cloud.delete_role('1234'))
|
||||
self.assertTrue(mock_keystone.roles.delete.called)
|
||||
def test_create_role(self):
|
||||
self._add_discovery_uri_call()
|
||||
role_data = self._get_role_data()
|
||||
self.register_uris([
|
||||
dict(method='POST',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json=role_data.json_response,
|
||||
validate=role_data.json_request),
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(append=[role_data.role_id]),
|
||||
status_code=200,
|
||||
json=role_data.json_response)
|
||||
])
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments(self, mock_keystone, mock_api_version):
|
||||
mock_api_version.return_value = '3'
|
||||
mock_keystone.role_assignments.list.return_value = RAW_ROLE_ASSIGNMENTS
|
||||
role = self.op_cloud.create_role(role_data.role_name)
|
||||
|
||||
self.assertIsNotNone(role)
|
||||
self.assertThat(role.name, matchers.Equals(role_data.role_name))
|
||||
self.assertThat(role.id, matchers.Equals(role_data.role_id))
|
||||
self.assert_calls()
|
||||
|
||||
def test_delete_role_by_id(self):
|
||||
self._add_discovery_uri_call()
|
||||
role_data = self._get_role_data()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'roles': [role_data.json_response['role']]}),
|
||||
dict(method='DELETE',
|
||||
uri=self.get_mock_url(append=[role_data.role_id]),
|
||||
status_code=204)
|
||||
])
|
||||
role = self.op_cloud.delete_role(role_data.role_id)
|
||||
self.assertThat(role, matchers.Equals(True))
|
||||
self.assert_calls()
|
||||
|
||||
def test_delete_role_by_name(self):
|
||||
self._add_discovery_uri_call()
|
||||
role_data = self._get_role_data()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'roles': [role_data.json_response['role']]}),
|
||||
dict(method='DELETE',
|
||||
uri=self.get_mock_url(append=[role_data.role_id]),
|
||||
status_code=204)
|
||||
])
|
||||
role = self.op_cloud.delete_role(role_data.role_name)
|
||||
self.assertThat(role, matchers.Equals(True))
|
||||
self.assert_calls()
|
||||
|
||||
def test_list_role_assignments(self):
|
||||
self._add_discovery_uri_call()
|
||||
domain_data = self._get_domain_data()
|
||||
user_data = self._get_user_data(domain_id=domain_data.domain_id)
|
||||
group_data = self._get_group_data(domain_id=domain_data.domain_id)
|
||||
project_data = self._get_project_data(domain_id=domain_data.domain_id)
|
||||
role_data = self._get_role_data()
|
||||
response = [
|
||||
{'links': 'https://example.com',
|
||||
'role': {'id': role_data.role_id},
|
||||
'scope': {'domain': {'id': domain_data.domain_id}},
|
||||
'user': {'id': user_data.user_id}},
|
||||
{'links': 'https://example.com',
|
||||
'role': {'id': role_data.role_id},
|
||||
'scope': {'project': {'id': project_data.project_id}},
|
||||
'group': {'id': group_data.group_id}},
|
||||
]
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource='role_assignments'),
|
||||
status_code=200,
|
||||
json={'role_assignments': response},
|
||||
complete_qs=True)
|
||||
])
|
||||
ret = self.op_cloud.list_role_assignments()
|
||||
mock_keystone.role_assignments.list.assert_called_once_with()
|
||||
normalized_assignments = _utils.normalize_role_assignments(
|
||||
RAW_ROLE_ASSIGNMENTS
|
||||
)
|
||||
self.assertEqual(normalized_assignments, ret)
|
||||
self.assertThat(len(ret), matchers.Equals(2))
|
||||
self.assertThat(ret[0].user, matchers.Equals(user_data.user_id))
|
||||
self.assertThat(ret[0].id, matchers.Equals(role_data.role_id))
|
||||
self.assertThat(ret[0].domain, matchers.Equals(domain_data.domain_id))
|
||||
self.assertThat(ret[1].group, matchers.Equals(group_data.group_id))
|
||||
self.assertThat(ret[1].id, matchers.Equals(role_data.role_id))
|
||||
self.assertThat(ret[1].project,
|
||||
matchers.Equals(project_data.project_id))
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments_filters(self, mock_keystone,
|
||||
mock_api_version):
|
||||
mock_api_version.return_value = '3'
|
||||
params = dict(user='123', domain='456', effective=True)
|
||||
self.op_cloud.list_role_assignments(filters=params)
|
||||
mock_keystone.role_assignments.list.assert_called_once_with(**params)
|
||||
def test_list_role_assignments_filters(self):
|
||||
self._add_discovery_uri_call()
|
||||
domain_data = self._get_domain_data()
|
||||
user_data = self._get_user_data(domain_id=domain_data.domain_id)
|
||||
role_data = self._get_role_data()
|
||||
response = [
|
||||
{'links': 'https://example.com',
|
||||
'role': {'id': role_data.role_id},
|
||||
'scope': {'domain': {'id': domain_data.domain_id}},
|
||||
'user': {'id': user_data.user_id}}
|
||||
]
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments_exception(self, mock_keystone,
|
||||
mock_api_version):
|
||||
mock_api_version.return_value = '3'
|
||||
mock_keystone.role_assignments.list.side_effect = Exception()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource='role_assignments',
|
||||
qs_elements=['scope.domain.id=%s' % domain_data.domain_id,
|
||||
'user.id=%s' % user_data.user_id,
|
||||
'effective=True']),
|
||||
status_code=200,
|
||||
json={'role_assignments': response},
|
||||
complete_qs=True)
|
||||
])
|
||||
params = dict(user=user_data.user_id, domain=domain_data.domain_id,
|
||||
effective=True)
|
||||
ret = self.op_cloud.list_role_assignments(filters=params)
|
||||
self.assertThat(len(ret), matchers.Equals(1))
|
||||
self.assertThat(ret[0].user, matchers.Equals(user_data.user_id))
|
||||
self.assertThat(ret[0].id, matchers.Equals(role_data.role_id))
|
||||
self.assertThat(ret[0].domain, matchers.Equals(domain_data.domain_id))
|
||||
|
||||
def test_list_role_assignments_exception(self):
|
||||
self._add_discovery_uri_call()
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(resource='role_assignments'),
|
||||
status_code=403)
|
||||
])
|
||||
with testtools.ExpectedException(
|
||||
shade.OpenStackCloudException,
|
||||
"Failed to list role assignments"
|
||||
):
|
||||
self.op_cloud.list_role_assignments()
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments_keystone_v2(self, mock_keystone,
|
||||
mock_api_version):
|
||||
fake_role = fakes.FakeRole(id='1234', name='fake_role')
|
||||
mock_api_version.return_value = '2.0'
|
||||
mock_keystone.roles.roles_for_user.return_value = [fake_role]
|
||||
def test_list_role_assignments_keystone_v2(self):
|
||||
self.use_keystone_v2()
|
||||
role_data = self._get_role_data()
|
||||
user_data = self._get_user_data()
|
||||
project_data = self._get_project_data(v3=False)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource='tenants',
|
||||
append=[project_data.project_id,
|
||||
'users',
|
||||
user_data.user_id,
|
||||
'roles'],
|
||||
base_url_append=None),
|
||||
status_code=200,
|
||||
json={'roles': [role_data.json_response['role']]})
|
||||
])
|
||||
ret = self.op_cloud.list_role_assignments(
|
||||
filters={
|
||||
'user': '2222',
|
||||
'project': '3333'})
|
||||
self.assertEqual(
|
||||
ret, [{
|
||||
'id': fake_role.id,
|
||||
'project': '3333',
|
||||
'user': '2222'}])
|
||||
'user': user_data.user_id,
|
||||
'project': project_data.project_id})
|
||||
self.assertThat(len(ret), matchers.Equals(1))
|
||||
self.assertThat(ret[0].project,
|
||||
matchers.Equals(project_data.project_id))
|
||||
self.assertThat(ret[0].id, matchers.Equals(role_data.role_id))
|
||||
self.assertThat(ret[0].user, matchers.Equals(user_data.user_id))
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments_keystone_v2_with_role(self, mock_keystone,
|
||||
mock_api_version):
|
||||
fake_role1 = fakes.FakeRole(id='1234', name='fake_role')
|
||||
fake_role2 = fakes.FakeRole(id='4321', name='fake_role')
|
||||
mock_api_version.return_value = '2.0'
|
||||
mock_keystone.roles.roles_for_user.return_value = [fake_role1,
|
||||
fake_role2]
|
||||
def test_list_role_assignments_keystone_v2_with_role(self):
|
||||
self.use_keystone_v2()
|
||||
roles_data = [self._get_role_data() for r in range(0, 2)]
|
||||
user_data = self._get_user_data()
|
||||
project_data = self._get_project_data(v3=False)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource='tenants',
|
||||
append=[project_data.project_id,
|
||||
'users',
|
||||
user_data.user_id,
|
||||
'roles'],
|
||||
base_url_append=None),
|
||||
status_code=200,
|
||||
json={'roles': [r.json_response['role'] for r in roles_data]})
|
||||
])
|
||||
ret = self.op_cloud.list_role_assignments(
|
||||
filters={
|
||||
'role': fake_role1.id,
|
||||
'user': '2222',
|
||||
'project': '3333'})
|
||||
self.assertEqual(
|
||||
ret, [{
|
||||
'id': fake_role1.id,
|
||||
'project': '3333',
|
||||
'user': '2222'}])
|
||||
'role': roles_data[0].role_id,
|
||||
'user': user_data.user_id,
|
||||
'project': project_data.project_id})
|
||||
self.assertThat(len(ret), matchers.Equals(1))
|
||||
self.assertThat(ret[0].project,
|
||||
matchers.Equals(project_data.project_id))
|
||||
self.assertThat(ret[0].id, matchers.Equals(roles_data[0].role_id))
|
||||
self.assertThat(ret[0].user, matchers.Equals(user_data.user_id))
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments_exception_v2(self, mock_keystone,
|
||||
mock_api_version):
|
||||
mock_api_version.return_value = '2.0'
|
||||
def test_list_role_assignments_exception_v2(self):
|
||||
self.use_keystone_v2()
|
||||
with testtools.ExpectedException(
|
||||
shade.OpenStackCloudException,
|
||||
"Must provide project and user for keystone v2"
|
||||
):
|
||||
self.op_cloud.list_role_assignments()
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(occ.cloud_config.CloudConfig, 'get_api_version')
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_role_assignments_exception_v2_no_project(self, mock_keystone,
|
||||
mock_api_version):
|
||||
mock_api_version.return_value = '2.0'
|
||||
def test_list_role_assignments_exception_v2_no_project(self):
|
||||
self.use_keystone_v2()
|
||||
with testtools.ExpectedException(
|
||||
shade.OpenStackCloudException,
|
||||
"Must provide project and user for keystone v2"
|
||||
):
|
||||
self.op_cloud.list_role_assignments(filters={'user': '12345'})
|
||||
self.assert_calls()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user