diff options
Diffstat (limited to 'keystone-moon/keystone/tests/unit/test_v3_catalog.py')
| -rw-r--r-- | keystone-moon/keystone/tests/unit/test_v3_catalog.py | 924 |
1 files changed, 0 insertions, 924 deletions
diff --git a/keystone-moon/keystone/tests/unit/test_v3_catalog.py b/keystone-moon/keystone/tests/unit/test_v3_catalog.py deleted file mode 100644 index 2eb9db14..00000000 --- a/keystone-moon/keystone/tests/unit/test_v3_catalog.py +++ /dev/null @@ -1,924 +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 - -from six.moves import http_client -from testtools import matchers - -from keystone import catalog -from keystone.tests import unit -from keystone.tests.unit.ksfixtures import database -from keystone.tests.unit import test_v3 - - -class CatalogTestCase(test_v3.RestfulTestCase): - """Test service & endpoint CRUD.""" - - # region crud tests - - def test_create_region_with_id(self): - """Call ``PUT /regions/{region_id}`` w/o an ID in the request body.""" - ref = unit.new_region_ref() - region_id = ref.pop('id') - r = self.put( - '/regions/%s' % region_id, - body={'region': ref}, - expected_status=http_client.CREATED) - self.assertValidRegionResponse(r, ref) - # Double-check that the region ID was kept as-is and not - # populated with a UUID, as is the case with POST /v3/regions - self.assertEqual(region_id, r.json['region']['id']) - - def test_create_region_with_matching_ids(self): - """Call ``PUT /regions/{region_id}`` with an ID in the request body.""" - ref = unit.new_region_ref() - region_id = ref['id'] - r = self.put( - '/regions/%s' % region_id, - body={'region': ref}, - expected_status=http_client.CREATED) - self.assertValidRegionResponse(r, ref) - # Double-check that the region ID was kept as-is and not - # populated with a UUID, as is the case with POST /v3/regions - self.assertEqual(region_id, r.json['region']['id']) - - def test_create_region_with_duplicate_id(self): - """Call ``PUT /regions/{region_id}``.""" - ref = dict(description="my region") - self.put( - '/regions/myregion', - body={'region': ref}, expected_status=http_client.CREATED) - # Create region again with duplicate id - self.put( - '/regions/myregion', - body={'region': ref}, expected_status=http_client.CONFLICT) - - def test_create_region(self): - """Call ``POST /regions`` with an ID in the request body.""" - # the ref will have an ID defined on it - ref = unit.new_region_ref() - r = self.post( - '/regions', - body={'region': ref}) - self.assertValidRegionResponse(r, ref) - - # we should be able to get the region, having defined the ID ourselves - r = self.get( - '/regions/%(region_id)s' % { - 'region_id': ref['id']}) - self.assertValidRegionResponse(r, ref) - - def test_create_region_with_empty_id(self): - """Call ``POST /regions`` with an empty ID in the request body.""" - ref = unit.new_region_ref(id='') - - r = self.post('/regions', body={'region': ref}) - self.assertValidRegionResponse(r, ref) - self.assertNotEmpty(r.result['region'].get('id')) - - def test_create_region_without_id(self): - """Call ``POST /regions`` without an ID in the request body.""" - ref = unit.new_region_ref() - - # instead of defining the ID ourselves... - del ref['id'] - - # let the service define the ID - r = self.post('/regions', body={'region': ref}) - self.assertValidRegionResponse(r, ref) - - def test_create_region_without_description(self): - """Call ``POST /regions`` without description in the request body.""" - ref = unit.new_region_ref(description=None) - - del ref['description'] - - r = self.post('/regions', body={'region': ref}) - # Create the description in the reference to compare to since the - # response should now have a description, even though we didn't send - # it with the original reference. - ref['description'] = '' - self.assertValidRegionResponse(r, ref) - - def test_create_regions_with_same_description_string(self): - """Call ``POST /regions`` with duplicate descriptions.""" - # NOTE(lbragstad): Make sure we can create two regions that have the - # same description. - region_desc = 'Some Region Description' - - ref1 = unit.new_region_ref(description=region_desc) - ref2 = unit.new_region_ref(description=region_desc) - - resp1 = self.post('/regions', body={'region': ref1}) - self.assertValidRegionResponse(resp1, ref1) - - resp2 = self.post('/regions', body={'region': ref2}) - self.assertValidRegionResponse(resp2, ref2) - - def test_create_regions_without_descriptions(self): - """Call ``POST /regions`` with no description.""" - # NOTE(lbragstad): Make sure we can create two regions that have - # no description in the request body. The description should be - # populated by Catalog Manager. - ref1 = unit.new_region_ref() - ref2 = unit.new_region_ref() - - del ref1['description'] - ref2['description'] = None - - resp1 = self.post('/regions', body={'region': ref1}) - - resp2 = self.post('/regions', body={'region': ref2}) - # Create the descriptions in the references to compare to since the - # responses should now have descriptions, even though we didn't send - # a description with the original references. - ref1['description'] = '' - ref2['description'] = '' - self.assertValidRegionResponse(resp1, ref1) - self.assertValidRegionResponse(resp2, ref2) - - def test_create_region_with_conflicting_ids(self): - """Call ``PUT /regions/{region_id}`` with conflicting region IDs.""" - # the region ref is created with an ID - ref = unit.new_region_ref() - - # but instead of using that ID, make up a new, conflicting one - self.put( - '/regions/%s' % uuid.uuid4().hex, - body={'region': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_list_regions(self): - """Call ``GET /regions``.""" - r = self.get('/regions') - self.assertValidRegionListResponse(r, ref=self.region) - - def _create_region_with_parent_id(self, parent_id=None): - ref = unit.new_region_ref(parent_region_id=parent_id) - return self.post( - '/regions', - body={'region': ref}) - - def test_list_regions_filtered_by_parent_region_id(self): - """Call ``GET /regions?parent_region_id={parent_region_id}``.""" - new_region = self._create_region_with_parent_id() - parent_id = new_region.result['region']['id'] - - new_region = self._create_region_with_parent_id(parent_id) - new_region = self._create_region_with_parent_id(parent_id) - - r = self.get('/regions?parent_region_id=%s' % parent_id) - - for region in r.result['regions']: - self.assertEqual(parent_id, region['parent_region_id']) - - def test_get_region(self): - """Call ``GET /regions/{region_id}``.""" - r = self.get('/regions/%(region_id)s' % { - 'region_id': self.region_id}) - self.assertValidRegionResponse(r, self.region) - - def test_update_region(self): - """Call ``PATCH /regions/{region_id}``.""" - region = unit.new_region_ref() - del region['id'] - r = self.patch('/regions/%(region_id)s' % { - 'region_id': self.region_id}, - body={'region': region}) - self.assertValidRegionResponse(r, region) - - def test_update_region_without_description_keeps_original(self): - """Call ``PATCH /regions/{region_id}``.""" - region_ref = unit.new_region_ref() - - resp = self.post('/regions', body={'region': region_ref}) - - region_updates = { - # update with something that's not the description - 'parent_region_id': self.region_id, - } - resp = self.patch('/regions/%s' % region_ref['id'], - body={'region': region_updates}) - - # NOTE(dstanek): Keystone should keep the original description. - self.assertEqual(region_ref['description'], - resp.result['region']['description']) - - def test_update_region_with_null_description(self): - """Call ``PATCH /regions/{region_id}``.""" - region = unit.new_region_ref(description=None) - del region['id'] - r = self.patch('/regions/%(region_id)s' % { - 'region_id': self.region_id}, - body={'region': region}) - - # NOTE(dstanek): Keystone should turn the provided None value into - # an empty string before storing in the backend. - region['description'] = '' - self.assertValidRegionResponse(r, region) - - def test_delete_region(self): - """Call ``DELETE /regions/{region_id}``.""" - ref = unit.new_region_ref() - r = self.post( - '/regions', - body={'region': ref}) - self.assertValidRegionResponse(r, ref) - - self.delete('/regions/%(region_id)s' % { - 'region_id': ref['id']}) - - # service crud tests - - def test_create_service(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref() - r = self.post( - '/services', - body={'service': ref}) - self.assertValidServiceResponse(r, ref) - - def test_create_service_no_name(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref() - del ref['name'] - r = self.post( - '/services', - body={'service': ref}) - ref['name'] = '' - self.assertValidServiceResponse(r, ref) - - def test_create_service_no_enabled(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref() - del ref['enabled'] - r = self.post( - '/services', - body={'service': ref}) - ref['enabled'] = True - self.assertValidServiceResponse(r, ref) - self.assertIs(True, r.result['service']['enabled']) - - def test_create_service_enabled_false(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref(enabled=False) - r = self.post( - '/services', - body={'service': ref}) - self.assertValidServiceResponse(r, ref) - self.assertIs(False, r.result['service']['enabled']) - - def test_create_service_enabled_true(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref(enabled=True) - r = self.post( - '/services', - body={'service': ref}) - self.assertValidServiceResponse(r, ref) - self.assertIs(True, r.result['service']['enabled']) - - def test_create_service_enabled_str_true(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref(enabled='True') - self.post('/services', body={'service': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_create_service_enabled_str_false(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref(enabled='False') - self.post('/services', body={'service': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_create_service_enabled_str_random(self): - """Call ``POST /services``.""" - ref = unit.new_service_ref(enabled='puppies') - self.post('/services', body={'service': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_list_services(self): - """Call ``GET /services``.""" - r = self.get('/services') - self.assertValidServiceListResponse(r, ref=self.service) - - def _create_random_service(self): - ref = unit.new_service_ref() - response = self.post( - '/services', - body={'service': ref}) - return response.json['service'] - - def test_filter_list_services_by_type(self): - """Call ``GET /services?type=<some type>``.""" - target_ref = self._create_random_service() - - # create unrelated services - self._create_random_service() - self._create_random_service() - - response = self.get('/services?type=' + target_ref['type']) - self.assertValidServiceListResponse(response, ref=target_ref) - - filtered_service_list = response.json['services'] - self.assertEqual(1, len(filtered_service_list)) - - filtered_service = filtered_service_list[0] - self.assertEqual(target_ref['type'], filtered_service['type']) - - def test_filter_list_services_by_name(self): - """Call ``GET /services?name=<some name>``.""" - target_ref = self._create_random_service() - - # create unrelated services - self._create_random_service() - self._create_random_service() - - response = self.get('/services?name=' + target_ref['name']) - self.assertValidServiceListResponse(response, ref=target_ref) - - filtered_service_list = response.json['services'] - self.assertEqual(1, len(filtered_service_list)) - - filtered_service = filtered_service_list[0] - self.assertEqual(target_ref['name'], filtered_service['name']) - - def test_get_service(self): - """Call ``GET /services/{service_id}``.""" - r = self.get('/services/%(service_id)s' % { - 'service_id': self.service_id}) - self.assertValidServiceResponse(r, self.service) - - def test_update_service(self): - """Call ``PATCH /services/{service_id}``.""" - service = unit.new_service_ref() - del service['id'] - r = self.patch('/services/%(service_id)s' % { - 'service_id': self.service_id}, - body={'service': service}) - self.assertValidServiceResponse(r, service) - - def test_delete_service(self): - """Call ``DELETE /services/{service_id}``.""" - self.delete('/services/%(service_id)s' % { - 'service_id': self.service_id}) - - # endpoint crud tests - - def test_list_endpoints(self): - """Call ``GET /endpoints``.""" - r = self.get('/endpoints') - self.assertValidEndpointListResponse(r, ref=self.endpoint) - - def _create_random_endpoint(self, interface='public', - parent_region_id=None): - region = self._create_region_with_parent_id( - parent_id=parent_region_id) - service = self._create_random_service() - ref = unit.new_endpoint_ref( - service_id=service['id'], - interface=interface, - region_id=region.result['region']['id']) - - response = self.post( - '/endpoints', - body={'endpoint': ref}) - return response.json['endpoint'] - - def test_list_endpoints_filtered_by_interface(self): - """Call ``GET /endpoints?interface={interface}``.""" - ref = self._create_random_endpoint(interface='internal') - - response = self.get('/endpoints?interface=%s' % ref['interface']) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['interface'], endpoint['interface']) - - def test_list_endpoints_filtered_by_service_id(self): - """Call ``GET /endpoints?service_id={service_id}``.""" - ref = self._create_random_endpoint() - - response = self.get('/endpoints?service_id=%s' % ref['service_id']) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['service_id'], endpoint['service_id']) - - def test_list_endpoints_filtered_by_region_id(self): - """Call ``GET /endpoints?region_id={region_id}``.""" - ref = self._create_random_endpoint() - - response = self.get('/endpoints?region_id=%s' % ref['region_id']) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['region_id'], endpoint['region_id']) - - def test_list_endpoints_filtered_by_parent_region_id(self): - """Call ``GET /endpoints?region_id={region_id}``. - - Ensure passing the parent_region_id as filter returns an - empty list. - - """ - parent_region = self._create_region_with_parent_id() - parent_region_id = parent_region.result['region']['id'] - self._create_random_endpoint(parent_region_id=parent_region_id) - - response = self.get('/endpoints?region_id=%s' % parent_region_id) - self.assertEqual(0, len(response.json['endpoints'])) - - def test_list_endpoints_with_multiple_filters(self): - """Call ``GET /endpoints?interface={interface}...``. - - Ensure passing different combinations of interface, region_id and - service_id as filters will return the correct result. - - """ - # interface and region_id specified - ref = self._create_random_endpoint(interface='internal') - response = self.get('/endpoints?interface=%s®ion_id=%s' % - (ref['interface'], ref['region_id'])) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['interface'], endpoint['interface']) - self.assertEqual(ref['region_id'], endpoint['region_id']) - - # interface and service_id specified - ref = self._create_random_endpoint(interface='internal') - response = self.get('/endpoints?interface=%s&service_id=%s' % - (ref['interface'], ref['service_id'])) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['interface'], endpoint['interface']) - self.assertEqual(ref['service_id'], endpoint['service_id']) - - # region_id and service_id specified - ref = self._create_random_endpoint(interface='internal') - response = self.get('/endpoints?region_id=%s&service_id=%s' % - (ref['region_id'], ref['service_id'])) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['region_id'], endpoint['region_id']) - self.assertEqual(ref['service_id'], endpoint['service_id']) - - # interface, region_id and service_id specified - ref = self._create_random_endpoint(interface='internal') - response = self.get(('/endpoints?interface=%s®ion_id=%s' - '&service_id=%s') % - (ref['interface'], ref['region_id'], - ref['service_id'])) - self.assertValidEndpointListResponse(response, ref=ref) - - for endpoint in response.json['endpoints']: - self.assertEqual(ref['interface'], endpoint['interface']) - self.assertEqual(ref['region_id'], endpoint['region_id']) - self.assertEqual(ref['service_id'], endpoint['service_id']) - - def test_list_endpoints_with_random_filter_values(self): - """Call ``GET /endpoints?interface={interface}...``. - - Ensure passing random values for: interface, region_id and - service_id will return an empty list. - - """ - self._create_random_endpoint(interface='internal') - - response = self.get('/endpoints?interface=%s' % uuid.uuid4().hex) - self.assertEqual(0, len(response.json['endpoints'])) - - response = self.get('/endpoints?region_id=%s' % uuid.uuid4().hex) - self.assertEqual(0, len(response.json['endpoints'])) - - response = self.get('/endpoints?service_id=%s' % uuid.uuid4().hex) - self.assertEqual(0, len(response.json['endpoints'])) - - def test_create_endpoint_no_enabled(self): - """Call ``POST /endpoints``.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id) - r = self.post('/endpoints', body={'endpoint': ref}) - ref['enabled'] = True - self.assertValidEndpointResponse(r, ref) - - def test_create_endpoint_enabled_true(self): - """Call ``POST /endpoints`` with enabled: true.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id, - enabled=True) - r = self.post('/endpoints', body={'endpoint': ref}) - self.assertValidEndpointResponse(r, ref) - - def test_create_endpoint_enabled_false(self): - """Call ``POST /endpoints`` with enabled: false.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id, - enabled=False) - r = self.post('/endpoints', body={'endpoint': ref}) - self.assertValidEndpointResponse(r, ref) - - def test_create_endpoint_enabled_str_true(self): - """Call ``POST /endpoints`` with enabled: 'True'.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id, - enabled='True') - self.post('/endpoints', body={'endpoint': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_create_endpoint_enabled_str_false(self): - """Call ``POST /endpoints`` with enabled: 'False'.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id, - enabled='False') - self.post('/endpoints', body={'endpoint': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_create_endpoint_enabled_str_random(self): - """Call ``POST /endpoints`` with enabled: 'puppies'.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id, - enabled='puppies') - self.post('/endpoints', body={'endpoint': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_create_endpoint_with_invalid_region_id(self): - """Call ``POST /endpoints``.""" - ref = unit.new_endpoint_ref(service_id=self.service_id) - self.post('/endpoints', body={'endpoint': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_create_endpoint_with_region(self): - """EndpointV3 creates the region before creating the endpoint. - - This occurs when endpoint is provided with 'region' and no 'region_id'. - """ - ref = unit.new_endpoint_ref_with_region(service_id=self.service_id, - region=uuid.uuid4().hex) - self.post('/endpoints', body={'endpoint': ref}) - # Make sure the region is created - self.get('/regions/%(region_id)s' % {'region_id': ref["region"]}) - - def test_create_endpoint_with_no_region(self): - """EndpointV3 allows to creates the endpoint without region.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, region_id=None) - del ref['region_id'] # cannot just be None, it needs to not exist - self.post('/endpoints', body={'endpoint': ref}) - - def test_create_endpoint_with_empty_url(self): - """Call ``POST /endpoints``.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, url='') - self.post('/endpoints', body={'endpoint': ref}, - expected_status=http_client.BAD_REQUEST) - - def test_get_endpoint(self): - """Call ``GET /endpoints/{endpoint_id}``.""" - r = self.get( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}) - self.assertValidEndpointResponse(r, self.endpoint) - - def test_update_endpoint(self): - """Call ``PATCH /endpoints/{endpoint_id}``.""" - ref = unit.new_endpoint_ref(service_id=self.service_id, - interface='public', - region_id=self.region_id) - del ref['id'] - r = self.patch( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}, - body={'endpoint': ref}) - ref['enabled'] = True - self.assertValidEndpointResponse(r, ref) - - def test_update_endpoint_enabled_true(self): - """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: True.""" - r = self.patch( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}, - body={'endpoint': {'enabled': True}}) - self.assertValidEndpointResponse(r, self.endpoint) - - def test_update_endpoint_enabled_false(self): - """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: False.""" - r = self.patch( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}, - body={'endpoint': {'enabled': False}}) - exp_endpoint = copy.copy(self.endpoint) - exp_endpoint['enabled'] = False - self.assertValidEndpointResponse(r, exp_endpoint) - - def test_update_endpoint_enabled_str_true(self): - """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: 'True'.""" - self.patch( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}, - body={'endpoint': {'enabled': 'True'}}, - expected_status=http_client.BAD_REQUEST) - - def test_update_endpoint_enabled_str_false(self): - """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: 'False'.""" - self.patch( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}, - body={'endpoint': {'enabled': 'False'}}, - expected_status=http_client.BAD_REQUEST) - - def test_update_endpoint_enabled_str_random(self): - """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: 'kitties'.""" - self.patch( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}, - body={'endpoint': {'enabled': 'kitties'}}, - expected_status=http_client.BAD_REQUEST) - - def test_delete_endpoint(self): - """Call ``DELETE /endpoints/{endpoint_id}``.""" - self.delete( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}) - - def test_create_endpoint_on_v2(self): - # clear the v3 endpoint so we only have endpoints created on v2 - self.delete( - '/endpoints/%(endpoint_id)s' % { - 'endpoint_id': self.endpoint_id}) - - # create a v3 endpoint ref, and then tweak it back to a v2-style ref - ref = unit.new_endpoint_ref_with_region(service_id=self.service['id'], - region=uuid.uuid4().hex, - internalurl=None) - del ref['id'] - del ref['interface'] - ref['publicurl'] = ref.pop('url') - # don't set adminurl to ensure it's absence is handled like internalurl - - # create the endpoint on v2 (using a v3 token) - r = self.admin_request( - method='POST', - path='/v2.0/endpoints', - token=self.get_scoped_token(), - body={'endpoint': ref}) - endpoint_v2 = r.result['endpoint'] - - # test the endpoint on v3 - r = self.get('/endpoints') - endpoints = self.assertValidEndpointListResponse(r) - self.assertEqual(1, len(endpoints)) - endpoint_v3 = endpoints.pop() - - # these attributes are identical between both APIs - self.assertEqual(ref['region'], endpoint_v3['region_id']) - self.assertEqual(ref['service_id'], endpoint_v3['service_id']) - self.assertEqual(ref['description'], endpoint_v3['description']) - - # a v2 endpoint is not quite the same concept as a v3 endpoint, so they - # receive different identifiers - self.assertNotEqual(endpoint_v2['id'], endpoint_v3['id']) - - # v2 has a publicurl; v3 has a url + interface type - self.assertEqual(ref['publicurl'], endpoint_v3['url']) - self.assertEqual('public', endpoint_v3['interface']) - - # tests for bug 1152632 -- these attributes were being returned by v3 - self.assertNotIn('publicurl', endpoint_v3) - self.assertNotIn('adminurl', endpoint_v3) - self.assertNotIn('internalurl', endpoint_v3) - - # test for bug 1152635 -- this attribute was being returned by v3 - self.assertNotIn('legacy_endpoint_id', endpoint_v3) - - self.assertEqual(endpoint_v2['region'], endpoint_v3['region_id']) - - def test_deleting_endpoint_with_space_in_url(self): - # add a space to all urls (intentional "i d" to test bug) - url_with_space = "http://127.0.0.1:8774 /v1.1/\$(tenant_i d)s" - - # create a v3 endpoint ref - ref = unit.new_endpoint_ref(service_id=self.service['id'], - region_id=None, - publicurl=url_with_space, - internalurl=url_with_space, - adminurl=url_with_space, - url=url_with_space) - - # add the endpoint to the database - self.catalog_api.create_endpoint(ref['id'], ref) - - # delete the endpoint - self.delete('/endpoints/%s' % ref['id']) - - # make sure it's deleted (GET should return Not Found) - self.get('/endpoints/%s' % ref['id'], - expected_status=http_client.NOT_FOUND) - - def test_endpoint_create_with_valid_url(self): - """Create endpoint with valid url should be tested,too.""" - # list one valid url is enough, no need to list too much - valid_url = 'http://127.0.0.1:8774/v1.1/$(tenant_id)s' - - ref = unit.new_endpoint_ref(self.service_id, - interface='public', - region_id=self.region_id, - url=valid_url) - self.post('/endpoints', body={'endpoint': ref}) - - def test_endpoint_create_with_valid_url_project_id(self): - """Create endpoint with valid url should be tested,too.""" - valid_url = 'http://127.0.0.1:8774/v1.1/$(project_id)s' - - ref = unit.new_endpoint_ref(self.service_id, - interface='public', - region_id=self.region_id, - url=valid_url) - self.post('/endpoints', body={'endpoint': ref}) - - def test_endpoint_create_with_invalid_url(self): - """Test the invalid cases: substitutions is not exactly right.""" - invalid_urls = [ - # using a substitution that is not whitelisted - KeyError - 'http://127.0.0.1:8774/v1.1/$(nonexistent)s', - - # invalid formatting - ValueError - 'http://127.0.0.1:8774/v1.1/$(tenant_id)', - 'http://127.0.0.1:8774/v1.1/$(tenant_id)t', - 'http://127.0.0.1:8774/v1.1/$(tenant_id', - - # invalid type specifier - TypeError - # admin_url is a string not an int - 'http://127.0.0.1:8774/v1.1/$(admin_url)d', - ] - - ref = unit.new_endpoint_ref(self.service_id) - - for invalid_url in invalid_urls: - ref['url'] = invalid_url - self.post('/endpoints', - body={'endpoint': ref}, - expected_status=http_client.BAD_REQUEST) - - -class TestCatalogAPISQL(unit.TestCase): - """Tests for the catalog Manager against the SQL backend.""" - - def setUp(self): - super(TestCatalogAPISQL, self).setUp() - self.useFixture(database.Database()) - self.catalog_api = catalog.Manager() - - service = unit.new_service_ref() - self.service_id = service['id'] - self.catalog_api.create_service(self.service_id, service) - - self.create_endpoint(service_id=self.service_id) - - def create_endpoint(self, service_id, **kwargs): - endpoint = unit.new_endpoint_ref(service_id=service_id, - region_id=None, **kwargs) - - self.catalog_api.create_endpoint(endpoint['id'], endpoint) - return endpoint - - def config_overrides(self): - super(TestCatalogAPISQL, self).config_overrides() - self.config_fixture.config(group='catalog', driver='sql') - - def test_get_catalog_ignores_endpoints_with_invalid_urls(self): - user_id = uuid.uuid4().hex - tenant_id = uuid.uuid4().hex - - # the only endpoint in the catalog is the one created in setUp - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - self.assertEqual(1, len(catalog[0]['endpoints'])) - # it's also the only endpoint in the backend - self.assertEqual(1, len(self.catalog_api.list_endpoints())) - - # create a new, invalid endpoint - malformed type declaration - self.create_endpoint(self.service_id, - url='http://keystone/%(tenant_id)') - - # create a new, invalid endpoint - nonexistent key - self.create_endpoint(self.service_id, - url='http://keystone/%(you_wont_find_me)s') - - # verify that the invalid endpoints don't appear in the catalog - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - self.assertEqual(1, len(catalog[0]['endpoints'])) - # all three appear in the backend - self.assertEqual(3, len(self.catalog_api.list_endpoints())) - - # create another valid endpoint - tenant_id will be replaced - self.create_endpoint(self.service_id, - url='http://keystone/%(tenant_id)s') - - # there are two valid endpoints, positive check - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - self.assertThat(catalog[0]['endpoints'], matchers.HasLength(2)) - - # If the URL has no 'tenant_id' to substitute, we will skip the - # endpoint which contains this kind of URL, negative check. - tenant_id = None - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - self.assertThat(catalog[0]['endpoints'], matchers.HasLength(1)) - - def test_get_catalog_always_returns_service_name(self): - user_id = uuid.uuid4().hex - tenant_id = uuid.uuid4().hex - - # create a service, with a name - named_svc = unit.new_service_ref() - self.catalog_api.create_service(named_svc['id'], named_svc) - self.create_endpoint(service_id=named_svc['id']) - - # create a service, with no name - unnamed_svc = unit.new_service_ref(name=None) - del unnamed_svc['name'] - self.catalog_api.create_service(unnamed_svc['id'], unnamed_svc) - self.create_endpoint(service_id=unnamed_svc['id']) - - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - - named_endpoint = [ep for ep in catalog - if ep['type'] == named_svc['type']][0] - self.assertEqual(named_svc['name'], named_endpoint['name']) - - unnamed_endpoint = [ep for ep in catalog - if ep['type'] == unnamed_svc['type']][0] - self.assertEqual('', unnamed_endpoint['name']) - - -# TODO(dstanek): this needs refactoring with the test above, but we are in a -# crunch so that will happen in a future patch. -class TestCatalogAPISQLRegions(unit.TestCase): - """Tests for the catalog Manager against the SQL backend.""" - - def setUp(self): - super(TestCatalogAPISQLRegions, self).setUp() - self.useFixture(database.Database()) - self.catalog_api = catalog.Manager() - - def config_overrides(self): - super(TestCatalogAPISQLRegions, self).config_overrides() - self.config_fixture.config(group='catalog', driver='sql') - - def test_get_catalog_returns_proper_endpoints_with_no_region(self): - service = unit.new_service_ref() - service_id = service['id'] - self.catalog_api.create_service(service_id, service) - - endpoint = unit.new_endpoint_ref(service_id=service_id, - region_id=None) - del endpoint['region_id'] - self.catalog_api.create_endpoint(endpoint['id'], endpoint) - - user_id = uuid.uuid4().hex - tenant_id = uuid.uuid4().hex - - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - self.assertValidCatalogEndpoint( - catalog[0]['endpoints'][0], ref=endpoint) - - def test_get_catalog_returns_proper_endpoints_with_region(self): - service = unit.new_service_ref() - service_id = service['id'] - self.catalog_api.create_service(service_id, service) - - endpoint = unit.new_endpoint_ref(service_id=service_id) - region = unit.new_region_ref(id=endpoint['region_id']) - self.catalog_api.create_region(region) - self.catalog_api.create_endpoint(endpoint['id'], endpoint) - - endpoint = self.catalog_api.get_endpoint(endpoint['id']) - user_id = uuid.uuid4().hex - tenant_id = uuid.uuid4().hex - - catalog = self.catalog_api.get_v3_catalog(user_id, tenant_id) - self.assertValidCatalogEndpoint( - catalog[0]['endpoints'][0], ref=endpoint) - - def assertValidCatalogEndpoint(self, entity, ref=None): - keys = ['description', 'id', 'interface', 'name', 'region_id', 'url'] - for k in keys: - self.assertEqual(ref.get(k), entity[k], k) - self.assertEqual(entity['region_id'], entity['region']) |