aboutsummaryrefslogtreecommitdiffstats
path: root/keystone-moon/keystone/tests/unit/test_associate_project_endpoint_extension.py
diff options
context:
space:
mode:
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.py1391
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