diff --git a/shade/tests/unit/base.py b/shade/tests/unit/base.py index f54f102d8..452f029be 100644 --- a/shade/tests/unit/base.py +++ b/shade/tests/unit/base.py @@ -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 = [] diff --git a/shade/tests/unit/test_identity_roles.py b/shade/tests/unit/test_identity_roles.py index 3582bb334..d7f676c83 100644 --- a/shade/tests/unit/test_identity_roles.py +++ b/shade/tests/unit/test_identity_roles.py @@ -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()