# 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=``.""" 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=``.""" 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'])