diff options
Diffstat (limited to 'keystone-moon/keystone/tests/unit/test_associate_project_endpoint_extension.py')
-rw-r--r-- | keystone-moon/keystone/tests/unit/test_associate_project_endpoint_extension.py | 1391 |
1 files changed, 0 insertions, 1391 deletions
diff --git a/keystone-moon/keystone/tests/unit/test_associate_project_endpoint_extension.py b/keystone-moon/keystone/tests/unit/test_associate_project_endpoint_extension.py deleted file mode 100644 index 79065863..00000000 --- a/keystone-moon/keystone/tests/unit/test_associate_project_endpoint_extension.py +++ /dev/null @@ -1,1391 +0,0 @@ -# Copyright 2013 OpenStack Foundation -# -# 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 copy -import uuid - -import mock -from oslo_log import versionutils -from six.moves import http_client -from testtools import matchers - -from keystone.contrib.endpoint_filter import routers -from keystone.tests import unit -from keystone.tests.unit import test_v3 - - -class EndpointFilterTestCase(test_v3.RestfulTestCase): - - def config_overrides(self): - super(EndpointFilterTestCase, self).config_overrides() - self.config_fixture.config( - group='catalog', driver='endpoint_filter.sql') - - def setUp(self): - super(EndpointFilterTestCase, self).setUp() - self.default_request_url = ( - '/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': self.endpoint_id}) - - -class EndpointFilterDeprecateTestCase(test_v3.RestfulTestCase): - - @mock.patch.object(versionutils, 'report_deprecated_feature') - def test_exception_happens(self, mock_deprecator): - routers.EndpointFilterExtension(mock.ANY) - mock_deprecator.assert_called_once_with(mock.ANY, mock.ANY) - args, _kwargs = mock_deprecator.call_args - self.assertIn("Remove endpoint_filter_extension from", args[1]) - - -class EndpointFilterCRUDTestCase(EndpointFilterTestCase): - - def test_create_endpoint_project_association(self): - """PUT /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Valid endpoint and project id test case. - - """ - self.put(self.default_request_url) - - def test_create_endpoint_project_association_with_invalid_project(self): - """PUT OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Invalid project id test case. - - """ - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': uuid.uuid4().hex, - 'endpoint_id': self.endpoint_id}, - expected_status=http_client.NOT_FOUND) - - def test_create_endpoint_project_association_with_invalid_endpoint(self): - """PUT /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Invalid endpoint id test case. - - """ - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': uuid.uuid4().hex}, - expected_status=http_client.NOT_FOUND) - - def test_create_endpoint_project_association_with_unexpected_body(self): - """PUT /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Unexpected body in request. The body should be ignored. - - """ - self.put(self.default_request_url, - body={'project_id': self.default_domain_project_id}) - - def test_check_endpoint_project_association(self): - """HEAD /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Valid project and endpoint id test case. - - """ - self.put(self.default_request_url) - self.head('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': self.endpoint_id}) - - def test_check_endpoint_project_association_with_invalid_project(self): - """HEAD /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Invalid project id test case. - - """ - self.put(self.default_request_url) - self.head('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': uuid.uuid4().hex, - 'endpoint_id': self.endpoint_id}, - expected_status=http_client.NOT_FOUND) - - def test_check_endpoint_project_association_with_invalid_endpoint(self): - """HEAD /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Invalid endpoint id test case. - - """ - self.put(self.default_request_url) - self.head('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': uuid.uuid4().hex}, - expected_status=http_client.NOT_FOUND) - - def test_list_endpoints_associated_with_valid_project(self): - """GET /OS-EP-FILTER/projects/{project_id}/endpoints - - Valid project and endpoint id test case. - - """ - self.put(self.default_request_url) - resource_url = '/OS-EP-FILTER/projects/%(project_id)s/endpoints' % { - 'project_id': self.default_domain_project_id} - r = self.get(resource_url) - self.assertValidEndpointListResponse(r, self.endpoint, - resource_url=resource_url) - - def test_list_endpoints_associated_with_invalid_project(self): - """GET /OS-EP-FILTER/projects/{project_id}/endpoints - - Invalid project id test case. - - """ - self.put(self.default_request_url) - self.get('/OS-EP-FILTER/projects/%(project_id)s/endpoints' % { - 'project_id': uuid.uuid4().hex}, - expected_status=http_client.NOT_FOUND) - - def test_list_projects_associated_with_endpoint(self): - """GET /OS-EP-FILTER/endpoints/{endpoint_id}/projects - - Valid endpoint-project association test case. - - """ - self.put(self.default_request_url) - resource_url = '/OS-EP-FILTER/endpoints/%(endpoint_id)s/projects' % { - 'endpoint_id': self.endpoint_id} - r = self.get(resource_url) - self.assertValidProjectListResponse(r, self.default_domain_project, - resource_url=resource_url) - - def test_list_projects_with_no_endpoint_project_association(self): - """GET /OS-EP-FILTER/endpoints/{endpoint_id}/projects - - Valid endpoint id but no endpoint-project associations test case. - - """ - r = self.get('/OS-EP-FILTER/endpoints/%(endpoint_id)s/projects' % - {'endpoint_id': self.endpoint_id}) - self.assertValidProjectListResponse(r, expected_length=0) - - def test_list_projects_associated_with_invalid_endpoint(self): - """GET /OS-EP-FILTER/endpoints/{endpoint_id}/projects - - Invalid endpoint id test case. - - """ - self.get('/OS-EP-FILTER/endpoints/%(endpoint_id)s/projects' % - {'endpoint_id': uuid.uuid4().hex}, - expected_status=http_client.NOT_FOUND) - - def test_remove_endpoint_project_association(self): - """DELETE /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Valid project id and endpoint id test case. - - """ - self.put(self.default_request_url) - self.delete('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': self.endpoint_id}) - - def test_remove_endpoint_project_association_with_invalid_project(self): - """DELETE /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Invalid project id test case. - - """ - self.put(self.default_request_url) - self.delete('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': uuid.uuid4().hex, - 'endpoint_id': self.endpoint_id}, - expected_status=http_client.NOT_FOUND) - - def test_remove_endpoint_project_association_with_invalid_endpoint(self): - """DELETE /OS-EP-FILTER/projects/{project_id}/endpoints/{endpoint_id} - - Invalid endpoint id test case. - - """ - self.put(self.default_request_url) - self.delete('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': uuid.uuid4().hex}, - expected_status=http_client.NOT_FOUND) - - def test_endpoint_project_association_cleanup_when_project_deleted(self): - self.put(self.default_request_url) - association_url = ('/OS-EP-FILTER/endpoints/%(endpoint_id)s/projects' % - {'endpoint_id': self.endpoint_id}) - r = self.get(association_url) - self.assertValidProjectListResponse(r, expected_length=1) - - self.delete('/projects/%(project_id)s' % { - 'project_id': self.default_domain_project_id}) - - r = self.get(association_url) - self.assertValidProjectListResponse(r, expected_length=0) - - def test_endpoint_project_association_cleanup_when_endpoint_deleted(self): - self.put(self.default_request_url) - association_url = '/OS-EP-FILTER/projects/%(project_id)s/endpoints' % { - 'project_id': self.default_domain_project_id} - r = self.get(association_url) - self.assertValidEndpointListResponse(r, expected_length=1) - - self.delete('/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}) - - r = self.get(association_url) - self.assertValidEndpointListResponse(r, expected_length=0) - - @unit.skip_if_cache_disabled('catalog') - def test_create_endpoint_project_association_invalidates_cache(self): - # NOTE(davechen): create another endpoint which will be added to - # default project, this should be done at first since - # `create_endpoint` will also invalidate cache. - endpoint_id2 = uuid.uuid4().hex - endpoint2 = unit.new_endpoint_ref(service_id=self.service_id, - region_id=self.region_id, - interface='public', - id=endpoint_id2) - self.catalog_api.create_endpoint(endpoint_id2, endpoint2.copy()) - - # create endpoint project association. - self.put(self.default_request_url) - - # should get back only one endpoint that was just created. - user_id = uuid.uuid4().hex - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - # there is only one endpoints associated with the default project. - self.assertEqual(1, len(catalog[0]['endpoints'])) - self.assertEqual(self.endpoint_id, catalog[0]['endpoints'][0]['id']) - - # add the second endpoint to default project, bypassing - # catalog_api API manager. - self.catalog_api.driver.add_endpoint_to_project( - endpoint_id2, - self.default_domain_project_id) - - # but, we can just get back one endpoint from the cache, since the - # catalog is pulled out from cache and its haven't been invalidated. - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertEqual(1, len(catalog[0]['endpoints'])) - - # remove the endpoint2 from the default project, and add it again via - # catalog_api API manager. - self.catalog_api.driver.remove_endpoint_from_project( - endpoint_id2, - self.default_domain_project_id) - - # add second endpoint to default project, this can be done by calling - # the catalog_api API manager directly but call the REST API - # instead for consistency. - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': endpoint_id2}) - - # should get back two endpoints since the cache has been - # invalidated when the second endpoint was added to default project. - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertEqual(2, len(catalog[0]['endpoints'])) - - ep_id_list = [catalog[0]['endpoints'][0]['id'], - catalog[0]['endpoints'][1]['id']] - self.assertItemsEqual([self.endpoint_id, endpoint_id2], ep_id_list) - - @unit.skip_if_cache_disabled('catalog') - def test_remove_endpoint_from_project_invalidates_cache(self): - endpoint_id2 = uuid.uuid4().hex - endpoint2 = unit.new_endpoint_ref(service_id=self.service_id, - region_id=self.region_id, - interface='public', - id=endpoint_id2) - self.catalog_api.create_endpoint(endpoint_id2, endpoint2.copy()) - # create endpoint project association. - self.put(self.default_request_url) - - # add second endpoint to default project. - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': endpoint_id2}) - - # should get back only one endpoint that was just created. - user_id = uuid.uuid4().hex - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - # there are two endpoints associated with the default project. - ep_id_list = [catalog[0]['endpoints'][0]['id'], - catalog[0]['endpoints'][1]['id']] - self.assertEqual(2, len(catalog[0]['endpoints'])) - self.assertItemsEqual([self.endpoint_id, endpoint_id2], ep_id_list) - - # remove the endpoint2 from the default project, bypassing - # catalog_api API manager. - self.catalog_api.driver.remove_endpoint_from_project( - endpoint_id2, - self.default_domain_project_id) - - # but, we can just still get back two endpoints from the cache, - # since the catalog is pulled out from cache and its haven't - # been invalidated. - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertEqual(2, len(catalog[0]['endpoints'])) - - # add back the endpoint2 to the default project, and remove it by - # catalog_api API manage. - self.catalog_api.driver.add_endpoint_to_project( - endpoint_id2, - self.default_domain_project_id) - - # remove the endpoint2 from the default project, this can be done - # by calling the catalog_api API manager directly but call - # the REST API instead for consistency. - self.delete('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.default_domain_project_id, - 'endpoint_id': endpoint_id2}) - - # should only get back one endpoint since the cache has been - # invalidated after the endpoint project association was removed. - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertEqual(1, len(catalog[0]['endpoints'])) - self.assertEqual(self.endpoint_id, catalog[0]['endpoints'][0]['id']) - - -class EndpointFilterTokenRequestTestCase(EndpointFilterTestCase): - - def test_project_scoped_token_using_endpoint_filter(self): - """Verify endpoints from project scoped token filtered.""" - # create a project to work with - ref = unit.new_project_ref(domain_id=self.domain_id) - r = self.post('/projects', body={'project': ref}) - project = self.assertValidProjectResponse(r, ref) - - # grant the user a role on the project - self.put( - '/projects/%(project_id)s/users/%(user_id)s/roles/%(role_id)s' % { - 'user_id': self.user['id'], - 'project_id': project['id'], - 'role_id': self.role['id']}) - - # set the user's preferred project - body = {'user': {'default_project_id': project['id']}} - r = self.patch('/users/%(user_id)s' % { - 'user_id': self.user['id']}, - body=body) - self.assertValidUserResponse(r) - - # add one endpoint to the project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': project['id'], - 'endpoint_id': self.endpoint_id}) - - # attempt to authenticate without requesting a project - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password']) - r = self.post('/auth/tokens', body=auth_data) - self.assertValidProjectScopedTokenResponse( - r, - require_catalog=True, - endpoint_filter=True, - ep_filter_assoc=1) - self.assertEqual(project['id'], r.result['token']['project']['id']) - - def test_default_scoped_token_using_endpoint_filter(self): - """Verify endpoints from default scoped token filtered.""" - # add one endpoint to default project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': self.endpoint_id}) - - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password'], - project_id=self.project['id']) - r = self.post('/auth/tokens', body=auth_data) - self.assertValidProjectScopedTokenResponse( - r, - require_catalog=True, - endpoint_filter=True, - ep_filter_assoc=1) - self.assertEqual(self.project['id'], - r.result['token']['project']['id']) - - # Ensure name of the service exists - self.assertIn('name', r.result['token']['catalog'][0]) - - # region and region_id should be the same in endpoints - endpoint = r.result['token']['catalog'][0]['endpoints'][0] - self.assertIn('region', endpoint) - self.assertIn('region_id', endpoint) - self.assertEqual(endpoint['region'], endpoint['region_id']) - - def test_scoped_token_with_no_catalog_using_endpoint_filter(self): - """Verify endpoint filter does not affect no catalog.""" - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': self.endpoint_id}) - - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password'], - project_id=self.project['id']) - r = self.post('/auth/tokens?nocatalog', body=auth_data) - self.assertValidProjectScopedTokenResponse( - r, - require_catalog=False) - self.assertEqual(self.project['id'], - r.result['token']['project']['id']) - - def test_invalid_endpoint_project_association(self): - """Verify an invalid endpoint-project association is handled.""" - # add first endpoint to default project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': self.endpoint_id}) - - # create a second temporary endpoint - endpoint_id2 = uuid.uuid4().hex - endpoint2 = unit.new_endpoint_ref(service_id=self.service_id, - region_id=self.region_id, - interface='public', - id=endpoint_id2) - self.catalog_api.create_endpoint(endpoint_id2, endpoint2.copy()) - - # add second endpoint to default project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': endpoint_id2}) - - # remove the temporary reference - # this will create inconsistency in the endpoint filter table - # which is fixed during the catalog creation for token request - self.catalog_api.delete_endpoint(endpoint_id2) - - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password'], - project_id=self.project['id']) - r = self.post('/auth/tokens', body=auth_data) - self.assertValidProjectScopedTokenResponse( - r, - require_catalog=True, - endpoint_filter=True, - ep_filter_assoc=1) - self.assertEqual(self.project['id'], - r.result['token']['project']['id']) - - def test_disabled_endpoint(self): - """Test that a disabled endpoint is handled.""" - # Add an enabled endpoint to the default project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': self.endpoint_id}) - - # Add a disabled endpoint to the default project. - - # Create a disabled endpoint that's like the enabled one. - disabled_endpoint_ref = copy.copy(self.endpoint) - disabled_endpoint_id = uuid.uuid4().hex - disabled_endpoint_ref.update({ - 'id': disabled_endpoint_id, - 'enabled': False, - 'interface': 'internal' - }) - self.catalog_api.create_endpoint(disabled_endpoint_id, - disabled_endpoint_ref) - - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': disabled_endpoint_id}) - - # Authenticate to get token with catalog - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password'], - project_id=self.project['id']) - r = self.post('/auth/tokens', body=auth_data) - - endpoints = r.result['token']['catalog'][0]['endpoints'] - endpoint_ids = [ep['id'] for ep in endpoints] - self.assertEqual([self.endpoint_id], endpoint_ids) - - def test_multiple_endpoint_project_associations(self): - - def _create_an_endpoint(): - endpoint_ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id) - r = self.post('/endpoints', body={'endpoint': endpoint_ref}) - return r.result['endpoint']['id'] - - # create three endpoints - endpoint_id1 = _create_an_endpoint() - endpoint_id2 = _create_an_endpoint() - _create_an_endpoint() - - # only associate two endpoints with project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': endpoint_id1}) - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': endpoint_id2}) - - # there should be only two endpoints in token catalog - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password'], - project_id=self.project['id']) - r = self.post('/auth/tokens', body=auth_data) - self.assertValidProjectScopedTokenResponse( - r, - require_catalog=True, - endpoint_filter=True, - ep_filter_assoc=2) - - def test_get_auth_catalog_using_endpoint_filter(self): - # add one endpoint to default project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': self.endpoint_id}) - - auth_data = self.build_authentication_request( - user_id=self.user['id'], - password=self.user['password'], - project_id=self.project['id']) - token_data = self.post('/auth/tokens', body=auth_data) - self.assertValidProjectScopedTokenResponse( - token_data, - require_catalog=True, - endpoint_filter=True, - ep_filter_assoc=1) - - auth_catalog = self.get('/auth/catalog', - token=token_data.headers['X-Subject-Token']) - self.assertEqual(token_data.result['token']['catalog'], - auth_catalog.result['catalog']) - - -class JsonHomeTests(EndpointFilterTestCase, test_v3.JsonHomeTestMixin): - JSON_HOME_DATA = { - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/endpoint_projects': { - 'href-template': '/OS-EP-FILTER/endpoints/{endpoint_id}/projects', - 'href-vars': { - 'endpoint_id': - 'http://docs.openstack.org/api/openstack-identity/3/param/' - 'endpoint_id', - }, - }, - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/endpoint_groups': { - 'href': '/OS-EP-FILTER/endpoint_groups', - }, - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/endpoint_group': { - 'href-template': '/OS-EP-FILTER/endpoint_groups/' - '{endpoint_group_id}', - 'href-vars': { - 'endpoint_group_id': - 'http://docs.openstack.org/api/openstack-identity/3/' - 'ext/OS-EP-FILTER/1.0/param/endpoint_group_id', - }, - }, - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/endpoint_group_to_project_association': { - 'href-template': '/OS-EP-FILTER/endpoint_groups/' - '{endpoint_group_id}/projects/{project_id}', - 'href-vars': { - 'project_id': - 'http://docs.openstack.org/api/openstack-identity/3/param/' - 'project_id', - 'endpoint_group_id': - 'http://docs.openstack.org/api/openstack-identity/3/' - 'ext/OS-EP-FILTER/1.0/param/endpoint_group_id', - }, - }, - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/projects_associated_with_endpoint_group': { - 'href-template': '/OS-EP-FILTER/endpoint_groups/' - '{endpoint_group_id}/projects', - 'href-vars': { - 'endpoint_group_id': - 'http://docs.openstack.org/api/openstack-identity/3/' - 'ext/OS-EP-FILTER/1.0/param/endpoint_group_id', - }, - }, - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/endpoints_in_endpoint_group': { - 'href-template': '/OS-EP-FILTER/endpoint_groups/' - '{endpoint_group_id}/endpoints', - 'href-vars': { - 'endpoint_group_id': - 'http://docs.openstack.org/api/openstack-identity/3/' - 'ext/OS-EP-FILTER/1.0/param/endpoint_group_id', - }, - }, - 'http://docs.openstack.org/api/openstack-identity/3/ext/OS-EP-FILTER/' - '1.0/rel/project_endpoint_groups': { - 'href-template': '/OS-EP-FILTER/projects/{project_id}/' - 'endpoint_groups', - 'href-vars': { - 'project_id': - 'http://docs.openstack.org/api/openstack-identity/3/param/' - 'project_id', - }, - }, - } - - -class EndpointGroupCRUDTestCase(EndpointFilterTestCase): - - DEFAULT_ENDPOINT_GROUP_BODY = { - 'endpoint_group': { - 'description': 'endpoint group description', - 'filters': { - 'interface': 'admin' - }, - 'name': 'endpoint_group_name' - } - } - - DEFAULT_ENDPOINT_GROUP_URL = '/OS-EP-FILTER/endpoint_groups' - - def test_create_endpoint_group(self): - """POST /OS-EP-FILTER/endpoint_groups - - Valid endpoint group test case. - - """ - r = self.post(self.DEFAULT_ENDPOINT_GROUP_URL, - body=self.DEFAULT_ENDPOINT_GROUP_BODY) - expected_filters = (self.DEFAULT_ENDPOINT_GROUP_BODY - ['endpoint_group']['filters']) - expected_name = (self.DEFAULT_ENDPOINT_GROUP_BODY - ['endpoint_group']['name']) - self.assertEqual(expected_filters, - r.result['endpoint_group']['filters']) - self.assertEqual(expected_name, r.result['endpoint_group']['name']) - self.assertThat( - r.result['endpoint_group']['links']['self'], - matchers.EndsWith( - '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': r.result['endpoint_group']['id']})) - - def test_create_invalid_endpoint_group(self): - """POST /OS-EP-FILTER/endpoint_groups - - Invalid endpoint group creation test case. - - """ - invalid_body = copy.deepcopy(self.DEFAULT_ENDPOINT_GROUP_BODY) - invalid_body['endpoint_group']['filters'] = {'foobar': 'admin'} - self.post(self.DEFAULT_ENDPOINT_GROUP_URL, - body=invalid_body, - expected_status=http_client.BAD_REQUEST) - - def test_get_endpoint_group(self): - """GET /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Valid endpoint group test case. - - """ - # create an endpoint group to work with - response = self.post(self.DEFAULT_ENDPOINT_GROUP_URL, - body=self.DEFAULT_ENDPOINT_GROUP_BODY) - endpoint_group_id = response.result['endpoint_group']['id'] - endpoint_group_filters = response.result['endpoint_group']['filters'] - endpoint_group_name = response.result['endpoint_group']['name'] - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.get(url) - self.assertEqual(endpoint_group_id, - response.result['endpoint_group']['id']) - self.assertEqual(endpoint_group_filters, - response.result['endpoint_group']['filters']) - self.assertEqual(endpoint_group_name, - response.result['endpoint_group']['name']) - self.assertThat(response.result['endpoint_group']['links']['self'], - matchers.EndsWith(url)) - - def test_get_invalid_endpoint_group(self): - """GET /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Invalid endpoint group test case. - - """ - endpoint_group_id = 'foobar' - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_check_endpoint_group(self): - """HEAD /OS-EP-FILTER/endpoint_groups/{endpoint_group_id} - - Valid endpoint_group_id test case. - - """ - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.head(url, expected_status=http_client.OK) - - def test_check_invalid_endpoint_group(self): - """HEAD /OS-EP-FILTER/endpoint_groups/{endpoint_group_id} - - Invalid endpoint_group_id test case. - - """ - endpoint_group_id = 'foobar' - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.head(url, expected_status=http_client.NOT_FOUND) - - def test_patch_endpoint_group(self): - """PATCH /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Valid endpoint group patch test case. - - """ - body = copy.deepcopy(self.DEFAULT_ENDPOINT_GROUP_BODY) - body['endpoint_group']['filters'] = {'region_id': 'UK'} - body['endpoint_group']['name'] = 'patch_test' - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - r = self.patch(url, body=body) - self.assertEqual(endpoint_group_id, - r.result['endpoint_group']['id']) - self.assertEqual(body['endpoint_group']['filters'], - r.result['endpoint_group']['filters']) - self.assertThat(r.result['endpoint_group']['links']['self'], - matchers.EndsWith(url)) - - def test_patch_nonexistent_endpoint_group(self): - """PATCH /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Invalid endpoint group patch test case. - - """ - body = { - 'endpoint_group': { - 'name': 'patch_test' - } - } - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': 'ABC'} - self.patch(url, body=body, expected_status=http_client.NOT_FOUND) - - def test_patch_invalid_endpoint_group(self): - """PATCH /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Valid endpoint group patch test case. - - """ - body = { - 'endpoint_group': { - 'description': 'endpoint group description', - 'filters': { - 'region': 'UK' - }, - 'name': 'patch_test' - } - } - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.patch(url, body=body, expected_status=http_client.BAD_REQUEST) - - # Perform a GET call to ensure that the content remains - # the same (as DEFAULT_ENDPOINT_GROUP_BODY) after attempting to update - # with an invalid filter - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - r = self.get(url) - del r.result['endpoint_group']['id'] - del r.result['endpoint_group']['links'] - self.assertDictEqual(self.DEFAULT_ENDPOINT_GROUP_BODY, r.result) - - def test_delete_endpoint_group(self): - """GET /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Valid endpoint group test case. - - """ - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.delete(url) - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_delete_invalid_endpoint_group(self): - """GET /OS-EP-FILTER/endpoint_groups/{endpoint_group} - - Invalid endpoint group test case. - - """ - endpoint_group_id = 'foobar' - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.delete(url, expected_status=http_client.NOT_FOUND) - - def test_add_endpoint_group_to_project(self): - """Create a valid endpoint group and project association.""" - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - self._create_endpoint_group_project_association(endpoint_group_id, - self.project_id) - - def test_add_endpoint_group_to_project_with_invalid_project_id(self): - """Create an invalid endpoint group and project association.""" - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # associate endpoint group with project - project_id = uuid.uuid4().hex - url = self._get_project_endpoint_group_url( - endpoint_group_id, project_id) - self.put(url, expected_status=http_client.NOT_FOUND) - - def test_get_endpoint_group_in_project(self): - """Test retrieving project endpoint group association.""" - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # associate endpoint group with project - url = self._get_project_endpoint_group_url( - endpoint_group_id, self.project_id) - self.put(url) - response = self.get(url) - self.assertEqual( - endpoint_group_id, - response.result['project_endpoint_group']['endpoint_group_id']) - self.assertEqual( - self.project_id, - response.result['project_endpoint_group']['project_id']) - - def test_get_invalid_endpoint_group_in_project(self): - """Test retrieving project endpoint group association.""" - endpoint_group_id = uuid.uuid4().hex - project_id = uuid.uuid4().hex - url = self._get_project_endpoint_group_url( - endpoint_group_id, project_id) - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_list_endpoint_groups_in_project(self): - """GET /OS-EP-FILTER/projects/{project_id}/endpoint_groups.""" - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # associate endpoint group with project - url = self._get_project_endpoint_group_url( - endpoint_group_id, self.project_id) - self.put(url) - - url = ('/OS-EP-FILTER/projects/%(project_id)s/endpoint_groups' % - {'project_id': self.project_id}) - response = self.get(url) - - self.assertEqual( - endpoint_group_id, - response.result['endpoint_groups'][0]['id']) - - def test_list_endpoint_groups_in_invalid_project(self): - """Test retrieving from invalid project.""" - project_id = uuid.uuid4().hex - url = ('/OS-EP-FILTER/projects/%(project_id)s/endpoint_groups' % - {'project_id': project_id}) - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_empty_endpoint_groups_in_project(self): - """Test when no endpoint groups associated with the project.""" - url = ('/OS-EP-FILTER/projects/%(project_id)s/endpoint_groups' % - {'project_id': self.project_id}) - response = self.get(url) - - self.assertEqual(0, len(response.result['endpoint_groups'])) - - def test_check_endpoint_group_to_project(self): - """Test HEAD with a valid endpoint group and project association.""" - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - self._create_endpoint_group_project_association(endpoint_group_id, - self.project_id) - url = self._get_project_endpoint_group_url( - endpoint_group_id, self.project_id) - self.head(url, expected_status=http_client.OK) - - def test_check_endpoint_group_to_project_with_invalid_project_id(self): - """Test HEAD with an invalid endpoint group and project association.""" - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # create an endpoint group to project association - url = self._get_project_endpoint_group_url( - endpoint_group_id, self.project_id) - self.put(url) - - # send a head request with an invalid project id - project_id = uuid.uuid4().hex - url = self._get_project_endpoint_group_url( - endpoint_group_id, project_id) - self.head(url, expected_status=http_client.NOT_FOUND) - - def test_list_endpoint_groups(self): - """GET /OS-EP-FILTER/endpoint_groups.""" - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # recover all endpoint groups - url = '/OS-EP-FILTER/endpoint_groups' - r = self.get(url) - self.assertNotEmpty(r.result['endpoint_groups']) - self.assertEqual(endpoint_group_id, - r.result['endpoint_groups'][0].get('id')) - - def test_list_projects_associated_with_endpoint_group(self): - """GET /OS-EP-FILTER/endpoint_groups/{endpoint_group}/projects - - Valid endpoint group test case. - - """ - # create an endpoint group to work with - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # associate endpoint group with project - self._create_endpoint_group_project_association(endpoint_group_id, - self.project_id) - - # recover list of projects associated with endpoint group - url = ('/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' - '/projects' % - {'endpoint_group_id': endpoint_group_id}) - self.get(url) - - def test_list_endpoints_associated_with_endpoint_group(self): - """GET /OS-EP-FILTER/endpoint_groups/{endpoint_group}/endpoints - - Valid endpoint group test case. - - """ - # create a service - service_ref = unit.new_service_ref() - response = self.post( - '/services', - body={'service': service_ref}) - - service_id = response.result['service']['id'] - - # create an endpoint - endpoint_ref = unit.new_endpoint_ref(service_id=service_id, - interface='public', - region_id=self.region_id) - response = self.post('/endpoints', body={'endpoint': endpoint_ref}) - endpoint_id = response.result['endpoint']['id'] - - # create an endpoint group - body = copy.deepcopy(self.DEFAULT_ENDPOINT_GROUP_BODY) - body['endpoint_group']['filters'] = {'service_id': service_id} - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, body) - - # create association - self._create_endpoint_group_project_association(endpoint_group_id, - self.project_id) - - # recover list of endpoints associated with endpoint group - url = ('/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' - '/endpoints' % {'endpoint_group_id': endpoint_group_id}) - r = self.get(url) - self.assertNotEmpty(r.result['endpoints']) - self.assertEqual(endpoint_id, r.result['endpoints'][0].get('id')) - - def test_list_endpoints_associated_with_project_endpoint_group(self): - """GET /OS-EP-FILTER/projects/{project_id}/endpoints - - Valid project, endpoint id, and endpoint group test case. - - """ - # create a temporary service - service_ref = unit.new_service_ref() - response = self.post('/services', body={'service': service_ref}) - service_id2 = response.result['service']['id'] - - # create additional endpoints - self._create_endpoint_and_associations( - self.default_domain_project_id, service_id2) - self._create_endpoint_and_associations( - self.default_domain_project_id) - - # create project and endpoint association with default endpoint: - self.put(self.default_request_url) - - # create an endpoint group that contains a different endpoint - body = copy.deepcopy(self.DEFAULT_ENDPOINT_GROUP_BODY) - body['endpoint_group']['filters'] = {'service_id': service_id2} - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, body) - - # associate endpoint group with project - self._create_endpoint_group_project_association( - endpoint_group_id, self.default_domain_project_id) - - # Now get a list of the filtered endpoints - endpoints_url = '/OS-EP-FILTER/projects/%(project_id)s/endpoints' % { - 'project_id': self.default_domain_project_id} - r = self.get(endpoints_url) - endpoints = self.assertValidEndpointListResponse(r) - self.assertEqual(2, len(endpoints)) - - # Ensure catalog includes the endpoints from endpoint_group project - # association, this is needed when a project scoped token is issued - # and "endpoint_filter.sql" backend driver is in place. - user_id = uuid.uuid4().hex - catalog_list = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - self.assertEqual(2, len(catalog_list)) - - # Now remove project endpoint group association - url = self._get_project_endpoint_group_url( - endpoint_group_id, self.default_domain_project_id) - self.delete(url) - - # Now remove endpoint group - url = '/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' % { - 'endpoint_group_id': endpoint_group_id} - self.delete(url) - - r = self.get(endpoints_url) - endpoints = self.assertValidEndpointListResponse(r) - self.assertEqual(1, len(endpoints)) - - catalog_list = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - self.assertEqual(1, len(catalog_list)) - - def test_endpoint_group_project_cleanup_with_project(self): - # create endpoint group - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # create new project and associate with endpoint_group - project_ref = unit.new_project_ref(domain_id=self.domain_id) - r = self.post('/projects', body={'project': project_ref}) - project = self.assertValidProjectResponse(r, project_ref) - url = self._get_project_endpoint_group_url(endpoint_group_id, - project['id']) - self.put(url) - - # check that we can recover the project endpoint group association - self.get(url) - - # Now delete the project and then try and retrieve the project - # endpoint group association again - self.delete('/projects/%(project_id)s' % { - 'project_id': project['id']}) - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_endpoint_group_project_cleanup_with_endpoint_group(self): - # create endpoint group - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # create new project and associate with endpoint_group - project_ref = unit.new_project_ref(domain_id=self.domain_id) - r = self.post('/projects', body={'project': project_ref}) - project = self.assertValidProjectResponse(r, project_ref) - url = self._get_project_endpoint_group_url(endpoint_group_id, - project['id']) - self.put(url) - - # check that we can recover the project endpoint group association - self.get(url) - - # now remove the project endpoint group association - self.delete(url) - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_removing_an_endpoint_group_project(self): - # create an endpoint group - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # create an endpoint_group project - url = self._get_project_endpoint_group_url( - endpoint_group_id, self.default_domain_project_id) - self.put(url) - - # remove the endpoint group project - self.delete(url) - self.get(url, expected_status=http_client.NOT_FOUND) - - def test_remove_endpoint_group_with_project_association(self): - # create an endpoint group - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # create an endpoint_group project - project_endpoint_group_url = self._get_project_endpoint_group_url( - endpoint_group_id, self.default_domain_project_id) - self.put(project_endpoint_group_url) - - # remove endpoint group, the associated endpoint_group project will - # be removed as well. - endpoint_group_url = ('/OS-EP-FILTER/endpoint_groups/' - '%(endpoint_group_id)s' - % {'endpoint_group_id': endpoint_group_id}) - self.delete(endpoint_group_url) - self.get(endpoint_group_url, expected_status=http_client.NOT_FOUND) - self.get(project_endpoint_group_url, - expected_status=http_client.NOT_FOUND) - - @unit.skip_if_cache_disabled('catalog') - def test_add_endpoint_group_to_project_invalidates_catalog_cache(self): - # create another endpoint with 'admin' interface which matches - # 'filters' definition in endpoint group, then there should be two - # endpoints returned when retrieving v3 catalog if cache works as - # expected. - # this should be done at first since `create_endpoint` will also - # invalidate cache. - endpoint_id2 = uuid.uuid4().hex - endpoint2 = unit.new_endpoint_ref(service_id=self.service_id, - region_id=self.region_id, - interface='admin', - id=endpoint_id2) - self.catalog_api.create_endpoint(endpoint_id2, endpoint2) - - # create a project and endpoint association. - self.put(self.default_request_url) - - # there is only one endpoint associated with the default project. - user_id = uuid.uuid4().hex - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertThat(catalog[0]['endpoints'], matchers.HasLength(1)) - - # create an endpoint group. - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # add the endpoint group to default project, bypassing - # catalog_api API manager. - self.catalog_api.driver.add_endpoint_group_to_project( - endpoint_group_id, - self.default_domain_project_id) - - # can get back only one endpoint from the cache, since the catalog - # is pulled out from cache. - invalid_catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertThat(invalid_catalog[0]['endpoints'], - matchers.HasLength(1)) - self.assertEqual(catalog, invalid_catalog) - - # remove the endpoint group from default project, and add it again via - # catalog_api API manager. - self.catalog_api.driver.remove_endpoint_group_from_project( - endpoint_group_id, - self.default_domain_project_id) - - # add the endpoint group to default project. - self.catalog_api.add_endpoint_group_to_project( - endpoint_group_id, - self.default_domain_project_id) - - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - # now, it will return 2 endpoints since the cache has been - # invalidated. - self.assertThat(catalog[0]['endpoints'], matchers.HasLength(2)) - - ep_id_list = [catalog[0]['endpoints'][0]['id'], - catalog[0]['endpoints'][1]['id']] - self.assertItemsEqual([self.endpoint_id, endpoint_id2], ep_id_list) - - @unit.skip_if_cache_disabled('catalog') - def test_remove_endpoint_group_from_project_invalidates_cache(self): - # create another endpoint with 'admin' interface which matches - # 'filters' definition in endpoint group, then there should be two - # endpoints returned when retrieving v3 catalog. But only one - # endpoint will return after the endpoint group's deletion if cache - # works as expected. - # this should be done at first since `create_endpoint` will also - # invalidate cache. - endpoint_id2 = uuid.uuid4().hex - endpoint2 = unit.new_endpoint_ref(service_id=self.service_id, - region_id=self.region_id, - interface='admin', - id=endpoint_id2) - self.catalog_api.create_endpoint(endpoint_id2, endpoint2) - - # create project and endpoint association. - self.put(self.default_request_url) - - # create an endpoint group. - endpoint_group_id = self._create_valid_endpoint_group( - self.DEFAULT_ENDPOINT_GROUP_URL, self.DEFAULT_ENDPOINT_GROUP_BODY) - - # add the endpoint group to default project. - self.catalog_api.add_endpoint_group_to_project( - endpoint_group_id, - self.default_domain_project_id) - - # should get back two endpoints, one from endpoint project - # association, the other one is from endpoint_group project - # association. - user_id = uuid.uuid4().hex - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertThat(catalog[0]['endpoints'], matchers.HasLength(2)) - - ep_id_list = [catalog[0]['endpoints'][0]['id'], - catalog[0]['endpoints'][1]['id']] - self.assertItemsEqual([self.endpoint_id, endpoint_id2], ep_id_list) - - # remove endpoint_group project association, bypassing - # catalog_api API manager. - self.catalog_api.driver.remove_endpoint_group_from_project( - endpoint_group_id, - self.default_domain_project_id) - - # still get back two endpoints, since the catalog is pulled out - # from cache and the cache haven't been invalidated. - invalid_catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertThat(invalid_catalog[0]['endpoints'], - matchers.HasLength(2)) - self.assertEqual(catalog, invalid_catalog) - - # add back the endpoint_group project association and remove it from - # manager. - self.catalog_api.driver.add_endpoint_group_to_project( - endpoint_group_id, - self.default_domain_project_id) - - self.catalog_api.remove_endpoint_group_from_project( - endpoint_group_id, - self.default_domain_project_id) - - # should only get back one endpoint since the cache has been - # invalidated after the endpoint_group project association was - # removed. - catalog = self.catalog_api.get_v3_catalog( - user_id, - self.default_domain_project_id) - - self.assertThat(catalog[0]['endpoints'], matchers.HasLength(1)) - self.assertEqual(self.endpoint_id, catalog[0]['endpoints'][0]['id']) - - def _create_valid_endpoint_group(self, url, body): - r = self.post(url, body=body) - return r.result['endpoint_group']['id'] - - def _create_endpoint_group_project_association(self, - endpoint_group_id, - project_id): - url = self._get_project_endpoint_group_url(endpoint_group_id, - project_id) - self.put(url) - - def _get_project_endpoint_group_url(self, - endpoint_group_id, - project_id): - return ('/OS-EP-FILTER/endpoint_groups/%(endpoint_group_id)s' - '/projects/%(project_id)s' % - {'endpoint_group_id': endpoint_group_id, - 'project_id': project_id}) - - def _create_endpoint_and_associations(self, project_id, service_id=None): - """Creates an endpoint associated with service and project.""" - if not service_id: - # create a new service - service_ref = unit.new_service_ref() - response = self.post( - '/services', body={'service': service_ref}) - service_id = response.result['service']['id'] - - # create endpoint - endpoint_ref = unit.new_endpoint_ref(service_id=service_id, - interface='public', - region_id=self.region_id) - response = self.post('/endpoints', body={'endpoint': endpoint_ref}) - endpoint = response.result['endpoint'] - - # now add endpoint to project - self.put('/OS-EP-FILTER/projects/%(project_id)s' - '/endpoints/%(endpoint_id)s' % { - 'project_id': self.project['id'], - 'endpoint_id': endpoint['id']}) - return endpoint |