"Fossies" - the Fresh Open Source Software Archive

Member "keystone-18.0.0/keystone/tests/unit/test_v3_catalog.py" (14 Oct 2020, 42581 Bytes) of package /linux/misc/openstack/keystone-18.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_v3_catalog.py": 17.0.0_vs_18.0.0.

    1 # Copyright 2013 OpenStack Foundation
    2 #
    3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #    not use this file except in compliance with the License. You may obtain
    5 #    a copy of the License at
    6 #
    7 #         http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 #    Unless required by applicable law or agreed to in writing, software
   10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #    License for the specific language governing permissions and limitations
   13 #    under the License.
   14 
   15 import copy
   16 import uuid
   17 
   18 import http.client
   19 from testtools import matchers
   20 
   21 from keystone.common import provider_api
   22 from keystone.tests import unit
   23 from keystone.tests.unit import default_fixtures
   24 from keystone.tests.unit.ksfixtures import database
   25 from keystone.tests.unit import test_v3
   26 
   27 PROVIDERS = provider_api.ProviderAPIs
   28 
   29 
   30 class CatalogTestCase(test_v3.RestfulTestCase):
   31     """Test service & endpoint CRUD."""
   32 
   33     # region crud tests
   34 
   35     def test_create_region_with_id(self):
   36         """Call ``PUT /regions/{region_id}`` w/o an ID in the request body."""
   37         ref = unit.new_region_ref()
   38         region_id = ref.pop('id')
   39         r = self.put(
   40             '/regions/%s' % region_id,
   41             body={'region': ref},
   42             expected_status=http.client.CREATED)
   43         self.assertValidRegionResponse(r, ref)
   44         # Double-check that the region ID was kept as-is and not
   45         # populated with a UUID, as is the case with POST /v3/regions
   46         self.assertEqual(region_id, r.json['region']['id'])
   47 
   48     def test_create_region_with_matching_ids(self):
   49         """Call ``PUT /regions/{region_id}`` with an ID in the request body."""
   50         ref = unit.new_region_ref()
   51         region_id = ref['id']
   52         r = self.put(
   53             '/regions/%s' % region_id,
   54             body={'region': ref},
   55             expected_status=http.client.CREATED)
   56         self.assertValidRegionResponse(r, ref)
   57         # Double-check that the region ID was kept as-is and not
   58         # populated with a UUID, as is the case with POST /v3/regions
   59         self.assertEqual(region_id, r.json['region']['id'])
   60 
   61     def test_create_region_with_duplicate_id(self):
   62         """Call ``PUT /regions/{region_id}``."""
   63         ref = unit.new_region_ref()
   64         region_id = ref['id']
   65         self.put(
   66             '/regions/%s' % region_id,
   67             body={'region': ref}, expected_status=http.client.CREATED)
   68         # Create region again with duplicate id
   69         self.put(
   70             '/regions/%s' % region_id,
   71             body={'region': ref}, expected_status=http.client.CONFLICT)
   72 
   73     def test_create_region(self):
   74         """Call ``POST /regions`` with an ID in the request body."""
   75         # the ref will have an ID defined on it
   76         ref = unit.new_region_ref()
   77         r = self.post(
   78             '/regions',
   79             body={'region': ref})
   80         self.assertValidRegionResponse(r, ref)
   81 
   82         # we should be able to get the region, having defined the ID ourselves
   83         r = self.get(
   84             '/regions/%(region_id)s' % {
   85                 'region_id': ref['id']})
   86         self.assertValidRegionResponse(r, ref)
   87 
   88     def test_create_region_with_empty_id(self):
   89         """Call ``POST /regions`` with an empty ID in the request body."""
   90         ref = unit.new_region_ref(id='')
   91 
   92         r = self.post('/regions', body={'region': ref})
   93         self.assertValidRegionResponse(r, ref)
   94         self.assertNotEmpty(r.result['region'].get('id'))
   95 
   96     def test_create_region_without_id(self):
   97         """Call ``POST /regions`` without an ID in the request body."""
   98         ref = unit.new_region_ref()
   99 
  100         # instead of defining the ID ourselves...
  101         del ref['id']
  102 
  103         # let the service define the ID
  104         r = self.post('/regions', body={'region': ref})
  105         self.assertValidRegionResponse(r, ref)
  106 
  107     def test_create_region_without_description(self):
  108         """Call ``POST /regions`` without description in the request body."""
  109         ref = unit.new_region_ref(description=None)
  110 
  111         del ref['description']
  112 
  113         r = self.post('/regions', body={'region': ref})
  114         # Create the description in the reference to compare to since the
  115         # response should now have a description, even though we didn't send
  116         # it with the original reference.
  117         ref['description'] = ''
  118         self.assertValidRegionResponse(r, ref)
  119 
  120     def test_create_regions_with_same_description_string(self):
  121         """Call ``POST /regions`` with duplicate descriptions."""
  122         # NOTE(lbragstad): Make sure we can create two regions that have the
  123         # same description.
  124         region_desc = 'Some Region Description'
  125 
  126         ref1 = unit.new_region_ref(description=region_desc)
  127         ref2 = unit.new_region_ref(description=region_desc)
  128 
  129         resp1 = self.post('/regions', body={'region': ref1})
  130         self.assertValidRegionResponse(resp1, ref1)
  131 
  132         resp2 = self.post('/regions', body={'region': ref2})
  133         self.assertValidRegionResponse(resp2, ref2)
  134 
  135     def test_create_regions_without_descriptions(self):
  136         """Call ``POST /regions`` with no description."""
  137         # NOTE(lbragstad): Make sure we can create two regions that have
  138         # no description in the request body. The description should be
  139         # populated by Catalog Manager.
  140         ref1 = unit.new_region_ref()
  141         ref2 = unit.new_region_ref()
  142 
  143         del ref1['description']
  144         ref2['description'] = None
  145 
  146         resp1 = self.post('/regions', body={'region': ref1})
  147 
  148         resp2 = self.post('/regions', body={'region': ref2})
  149         # Create the descriptions in the references to compare to since the
  150         # responses should now have descriptions, even though we didn't send
  151         # a description with the original references.
  152         ref1['description'] = ''
  153         ref2['description'] = ''
  154         self.assertValidRegionResponse(resp1, ref1)
  155         self.assertValidRegionResponse(resp2, ref2)
  156 
  157     def test_create_region_with_conflicting_ids(self):
  158         """Call ``PUT /regions/{region_id}`` with conflicting region IDs."""
  159         # the region ref is created with an ID
  160         ref = unit.new_region_ref()
  161 
  162         # but instead of using that ID, make up a new, conflicting one
  163         self.put(
  164             '/regions/%s' % uuid.uuid4().hex,
  165             body={'region': ref},
  166             expected_status=http.client.BAD_REQUEST)
  167 
  168     def test_list_head_regions(self):
  169         """Call ``GET & HEAD /regions``."""
  170         resource_url = '/regions'
  171         r = self.get(resource_url)
  172         self.assertValidRegionListResponse(r, ref=self.region)
  173         self.head(resource_url, expected_status=http.client.OK)
  174 
  175     def _create_region_with_parent_id(self, parent_id=None):
  176         ref = unit.new_region_ref(parent_region_id=parent_id)
  177         return self.post(
  178             '/regions',
  179             body={'region': ref})
  180 
  181     def test_list_regions_filtered_by_parent_region_id(self):
  182         """Call ``GET /regions?parent_region_id={parent_region_id}``."""
  183         new_region = self._create_region_with_parent_id()
  184         parent_id = new_region.result['region']['id']
  185 
  186         new_region = self._create_region_with_parent_id(parent_id)
  187         new_region = self._create_region_with_parent_id(parent_id)
  188 
  189         r = self.get('/regions?parent_region_id=%s' % parent_id)
  190 
  191         for region in r.result['regions']:
  192             self.assertEqual(parent_id, region['parent_region_id'])
  193 
  194     def test_get_head_region(self):
  195         """Call ``GET & HEAD /regions/{region_id}``."""
  196         resource_url = '/regions/%(region_id)s' % {
  197             'region_id': self.region_id}
  198         r = self.get(resource_url)
  199         self.assertValidRegionResponse(r, self.region)
  200         self.head(resource_url, expected_status=http.client.OK)
  201 
  202     def test_update_region(self):
  203         """Call ``PATCH /regions/{region_id}``."""
  204         region = unit.new_region_ref()
  205         del region['id']
  206         r = self.patch('/regions/%(region_id)s' % {
  207             'region_id': self.region_id},
  208             body={'region': region})
  209         self.assertValidRegionResponse(r, region)
  210 
  211     def test_update_region_without_description_keeps_original(self):
  212         """Call ``PATCH /regions/{region_id}``."""
  213         region_ref = unit.new_region_ref()
  214 
  215         resp = self.post('/regions', body={'region': region_ref})
  216 
  217         region_updates = {
  218             # update with something that's not the description
  219             'parent_region_id': self.region_id,
  220         }
  221         resp = self.patch('/regions/%s' % region_ref['id'],
  222                           body={'region': region_updates})
  223 
  224         # NOTE(dstanek): Keystone should keep the original description.
  225         self.assertEqual(region_ref['description'],
  226                          resp.result['region']['description'])
  227 
  228     def test_update_region_with_null_description(self):
  229         """Call ``PATCH /regions/{region_id}``."""
  230         region = unit.new_region_ref(description=None)
  231         del region['id']
  232         r = self.patch('/regions/%(region_id)s' % {
  233             'region_id': self.region_id},
  234             body={'region': region})
  235 
  236         # NOTE(dstanek): Keystone should turn the provided None value into
  237         # an empty string before storing in the backend.
  238         region['description'] = ''
  239         self.assertValidRegionResponse(r, region)
  240 
  241     def test_delete_region(self):
  242         """Call ``DELETE /regions/{region_id}``."""
  243         ref = unit.new_region_ref()
  244         r = self.post(
  245             '/regions',
  246             body={'region': ref})
  247         self.assertValidRegionResponse(r, ref)
  248 
  249         self.delete('/regions/%(region_id)s' % {
  250             'region_id': ref['id']})
  251 
  252     # service crud tests
  253 
  254     def test_create_service(self):
  255         """Call ``POST /services``."""
  256         ref = unit.new_service_ref()
  257         r = self.post(
  258             '/services',
  259             body={'service': ref})
  260         self.assertValidServiceResponse(r, ref)
  261 
  262     def test_create_service_no_name(self):
  263         """Call ``POST /services``."""
  264         ref = unit.new_service_ref()
  265         del ref['name']
  266         r = self.post(
  267             '/services',
  268             body={'service': ref})
  269         ref['name'] = ''
  270         self.assertValidServiceResponse(r, ref)
  271 
  272     def test_create_service_no_enabled(self):
  273         """Call ``POST /services``."""
  274         ref = unit.new_service_ref()
  275         del ref['enabled']
  276         r = self.post(
  277             '/services',
  278             body={'service': ref})
  279         ref['enabled'] = True
  280         self.assertValidServiceResponse(r, ref)
  281         self.assertIs(True, r.result['service']['enabled'])
  282 
  283     def test_create_service_enabled_false(self):
  284         """Call ``POST /services``."""
  285         ref = unit.new_service_ref(enabled=False)
  286         r = self.post(
  287             '/services',
  288             body={'service': ref})
  289         self.assertValidServiceResponse(r, ref)
  290         self.assertIs(False, r.result['service']['enabled'])
  291 
  292     def test_create_service_enabled_true(self):
  293         """Call ``POST /services``."""
  294         ref = unit.new_service_ref(enabled=True)
  295         r = self.post(
  296             '/services',
  297             body={'service': ref})
  298         self.assertValidServiceResponse(r, ref)
  299         self.assertIs(True, r.result['service']['enabled'])
  300 
  301     def test_create_service_enabled_str_true(self):
  302         """Call ``POST /services``."""
  303         ref = unit.new_service_ref(enabled='True')
  304         self.post('/services', body={'service': ref},
  305                   expected_status=http.client.BAD_REQUEST)
  306 
  307     def test_create_service_enabled_str_false(self):
  308         """Call ``POST /services``."""
  309         ref = unit.new_service_ref(enabled='False')
  310         self.post('/services', body={'service': ref},
  311                   expected_status=http.client.BAD_REQUEST)
  312 
  313     def test_create_service_enabled_str_random(self):
  314         """Call ``POST /services``."""
  315         ref = unit.new_service_ref(enabled='puppies')
  316         self.post('/services', body={'service': ref},
  317                   expected_status=http.client.BAD_REQUEST)
  318 
  319     def test_list_head_services(self):
  320         """Call ``GET & HEAD /services``."""
  321         resource_url = '/services'
  322         r = self.get(resource_url)
  323         self.assertValidServiceListResponse(r, ref=self.service)
  324         self.head(resource_url, expected_status=http.client.OK)
  325 
  326     def _create_random_service(self):
  327         ref = unit.new_service_ref()
  328         response = self.post(
  329             '/services',
  330             body={'service': ref})
  331         return response.json['service']
  332 
  333     def test_filter_list_services_by_type(self):
  334         """Call ``GET /services?type=<some type>``."""
  335         target_ref = self._create_random_service()
  336 
  337         # create unrelated services
  338         self._create_random_service()
  339         self._create_random_service()
  340 
  341         response = self.get('/services?type=' + target_ref['type'])
  342         self.assertValidServiceListResponse(response, ref=target_ref)
  343 
  344         filtered_service_list = response.json['services']
  345         self.assertEqual(1, len(filtered_service_list))
  346 
  347         filtered_service = filtered_service_list[0]
  348         self.assertEqual(target_ref['type'], filtered_service['type'])
  349 
  350     def test_filter_list_services_by_name(self):
  351         """Call ``GET /services?name=<some name>``."""
  352         # create unrelated services
  353         self._create_random_service()
  354         self._create_random_service()
  355 
  356         # create the desired service
  357         target_ref = self._create_random_service()
  358 
  359         response = self.get('/services?name=' + target_ref['name'])
  360         self.assertValidServiceListResponse(response, ref=target_ref)
  361 
  362         filtered_service_list = response.json['services']
  363         self.assertEqual(1, len(filtered_service_list))
  364 
  365         filtered_service = filtered_service_list[0]
  366         self.assertEqual(target_ref['name'], filtered_service['name'])
  367 
  368     def test_filter_list_services_by_name_with_list_limit(self):
  369         """Call ``GET /services?name=<some name>``."""
  370         self.config_fixture.config(list_limit=1)
  371 
  372         self.test_filter_list_services_by_name()
  373 
  374     def test_get_head_service(self):
  375         """Call ``GET & HEAD /services/{service_id}``."""
  376         resource_url = '/services/%(service_id)s' % {
  377             'service_id': self.service_id}
  378         r = self.get(resource_url)
  379         self.assertValidServiceResponse(r, self.service)
  380         self.head(resource_url, expected_status=http.client.OK)
  381 
  382     def test_update_service(self):
  383         """Call ``PATCH /services/{service_id}``."""
  384         service = unit.new_service_ref()
  385         del service['id']
  386         r = self.patch('/services/%(service_id)s' % {
  387             'service_id': self.service_id},
  388             body={'service': service})
  389         self.assertValidServiceResponse(r, service)
  390 
  391     def test_delete_service(self):
  392         """Call ``DELETE /services/{service_id}``."""
  393         self.delete('/services/%(service_id)s' % {
  394             'service_id': self.service_id})
  395 
  396     # endpoint crud tests
  397 
  398     def test_list_head_endpoints(self):
  399         """Call ``GET & HEAD /endpoints``."""
  400         resource_url = '/endpoints'
  401         r = self.get(resource_url)
  402         self.assertValidEndpointListResponse(r, ref=self.endpoint)
  403         self.head(resource_url, expected_status=http.client.OK)
  404 
  405     def _create_random_endpoint(self, interface='public',
  406                                 parent_region_id=None):
  407         region = self._create_region_with_parent_id(
  408             parent_id=parent_region_id)
  409         service = self._create_random_service()
  410         ref = unit.new_endpoint_ref(
  411             service_id=service['id'],
  412             interface=interface,
  413             region_id=region.result['region']['id'])
  414 
  415         response = self.post(
  416             '/endpoints',
  417             body={'endpoint': ref})
  418         return response.json['endpoint']
  419 
  420     def test_list_endpoints_filtered_by_interface(self):
  421         """Call ``GET /endpoints?interface={interface}``."""
  422         ref = self._create_random_endpoint(interface='internal')
  423 
  424         response = self.get('/endpoints?interface=%s' % ref['interface'])
  425         self.assertValidEndpointListResponse(response, ref=ref)
  426 
  427         for endpoint in response.json['endpoints']:
  428             self.assertEqual(ref['interface'], endpoint['interface'])
  429 
  430     def test_list_endpoints_filtered_by_service_id(self):
  431         """Call ``GET /endpoints?service_id={service_id}``."""
  432         ref = self._create_random_endpoint()
  433 
  434         response = self.get('/endpoints?service_id=%s' % ref['service_id'])
  435         self.assertValidEndpointListResponse(response, ref=ref)
  436 
  437         for endpoint in response.json['endpoints']:
  438             self.assertEqual(ref['service_id'], endpoint['service_id'])
  439 
  440     def test_list_endpoints_filtered_by_region_id(self):
  441         """Call ``GET /endpoints?region_id={region_id}``."""
  442         ref = self._create_random_endpoint()
  443 
  444         response = self.get('/endpoints?region_id=%s' % ref['region_id'])
  445         self.assertValidEndpointListResponse(response, ref=ref)
  446 
  447         for endpoint in response.json['endpoints']:
  448             self.assertEqual(ref['region_id'], endpoint['region_id'])
  449 
  450     def test_list_endpoints_filtered_by_parent_region_id(self):
  451         """Call ``GET /endpoints?region_id={region_id}``.
  452 
  453         Ensure passing the parent_region_id as filter returns an
  454         empty list.
  455 
  456         """
  457         parent_region = self._create_region_with_parent_id()
  458         parent_region_id = parent_region.result['region']['id']
  459         self._create_random_endpoint(parent_region_id=parent_region_id)
  460 
  461         response = self.get('/endpoints?region_id=%s' % parent_region_id)
  462         self.assertEqual(0, len(response.json['endpoints']))
  463 
  464     def test_list_endpoints_with_multiple_filters(self):
  465         """Call ``GET /endpoints?interface={interface}...``.
  466 
  467         Ensure passing different combinations of interface, region_id and
  468         service_id as filters will return the correct result.
  469 
  470         """
  471         # interface and region_id specified
  472         ref = self._create_random_endpoint(interface='internal')
  473         response = self.get('/endpoints?interface=%s&region_id=%s' %
  474                             (ref['interface'], ref['region_id']))
  475         self.assertValidEndpointListResponse(response, ref=ref)
  476 
  477         for endpoint in response.json['endpoints']:
  478             self.assertEqual(ref['interface'], endpoint['interface'])
  479             self.assertEqual(ref['region_id'], endpoint['region_id'])
  480 
  481         # interface and service_id specified
  482         ref = self._create_random_endpoint(interface='internal')
  483         response = self.get('/endpoints?interface=%s&service_id=%s' %
  484                             (ref['interface'], ref['service_id']))
  485         self.assertValidEndpointListResponse(response, ref=ref)
  486 
  487         for endpoint in response.json['endpoints']:
  488             self.assertEqual(ref['interface'], endpoint['interface'])
  489             self.assertEqual(ref['service_id'], endpoint['service_id'])
  490 
  491         # region_id and service_id specified
  492         ref = self._create_random_endpoint(interface='internal')
  493         response = self.get('/endpoints?region_id=%s&service_id=%s' %
  494                             (ref['region_id'], ref['service_id']))
  495         self.assertValidEndpointListResponse(response, ref=ref)
  496 
  497         for endpoint in response.json['endpoints']:
  498             self.assertEqual(ref['region_id'], endpoint['region_id'])
  499             self.assertEqual(ref['service_id'], endpoint['service_id'])
  500 
  501         # interface, region_id and service_id specified
  502         ref = self._create_random_endpoint(interface='internal')
  503         response = self.get(('/endpoints?interface=%s&region_id=%s'
  504                              '&service_id=%s') %
  505                             (ref['interface'], ref['region_id'],
  506                              ref['service_id']))
  507         self.assertValidEndpointListResponse(response, ref=ref)
  508 
  509         for endpoint in response.json['endpoints']:
  510             self.assertEqual(ref['interface'], endpoint['interface'])
  511             self.assertEqual(ref['region_id'], endpoint['region_id'])
  512             self.assertEqual(ref['service_id'], endpoint['service_id'])
  513 
  514     def test_list_endpoints_with_random_filter_values(self):
  515         """Call ``GET /endpoints?interface={interface}...``.
  516 
  517         Ensure passing random values for: interface, region_id and
  518         service_id will return an empty list.
  519 
  520         """
  521         self._create_random_endpoint(interface='internal')
  522 
  523         response = self.get('/endpoints?interface=%s' % uuid.uuid4().hex)
  524         self.assertEqual(0, len(response.json['endpoints']))
  525 
  526         response = self.get('/endpoints?region_id=%s' % uuid.uuid4().hex)
  527         self.assertEqual(0, len(response.json['endpoints']))
  528 
  529         response = self.get('/endpoints?service_id=%s' % uuid.uuid4().hex)
  530         self.assertEqual(0, len(response.json['endpoints']))
  531 
  532     def test_create_endpoint_no_enabled(self):
  533         """Call ``POST /endpoints``."""
  534         ref = unit.new_endpoint_ref(service_id=self.service_id,
  535                                     interface='public',
  536                                     region_id=self.region_id)
  537         r = self.post('/endpoints', body={'endpoint': ref})
  538         ref['enabled'] = True
  539         self.assertValidEndpointResponse(r, ref)
  540 
  541     def test_create_endpoint_enabled_true(self):
  542         """Call ``POST /endpoints`` with enabled: true."""
  543         ref = unit.new_endpoint_ref(service_id=self.service_id,
  544                                     interface='public',
  545                                     region_id=self.region_id,
  546                                     enabled=True)
  547         r = self.post('/endpoints', body={'endpoint': ref})
  548         self.assertValidEndpointResponse(r, ref)
  549 
  550     def test_create_endpoint_enabled_false(self):
  551         """Call ``POST /endpoints`` with enabled: false."""
  552         ref = unit.new_endpoint_ref(service_id=self.service_id,
  553                                     interface='public',
  554                                     region_id=self.region_id,
  555                                     enabled=False)
  556         r = self.post('/endpoints', body={'endpoint': ref})
  557         self.assertValidEndpointResponse(r, ref)
  558 
  559     def test_create_endpoint_enabled_str_true(self):
  560         """Call ``POST /endpoints`` with enabled: 'True'."""
  561         ref = unit.new_endpoint_ref(service_id=self.service_id,
  562                                     interface='public',
  563                                     region_id=self.region_id,
  564                                     enabled='True')
  565         self.post('/endpoints', body={'endpoint': ref},
  566                   expected_status=http.client.BAD_REQUEST)
  567 
  568     def test_create_endpoint_enabled_str_false(self):
  569         """Call ``POST /endpoints`` with enabled: 'False'."""
  570         ref = unit.new_endpoint_ref(service_id=self.service_id,
  571                                     interface='public',
  572                                     region_id=self.region_id,
  573                                     enabled='False')
  574         self.post('/endpoints', body={'endpoint': ref},
  575                   expected_status=http.client.BAD_REQUEST)
  576 
  577     def test_create_endpoint_enabled_str_random(self):
  578         """Call ``POST /endpoints`` with enabled: 'puppies'."""
  579         ref = unit.new_endpoint_ref(service_id=self.service_id,
  580                                     interface='public',
  581                                     region_id=self.region_id,
  582                                     enabled='puppies')
  583         self.post('/endpoints', body={'endpoint': ref},
  584                   expected_status=http.client.BAD_REQUEST)
  585 
  586     def test_create_endpoint_with_invalid_region_id(self):
  587         """Call ``POST /endpoints``."""
  588         ref = unit.new_endpoint_ref(service_id=self.service_id)
  589         self.post('/endpoints', body={'endpoint': ref},
  590                   expected_status=http.client.BAD_REQUEST)
  591 
  592     def test_create_endpoint_with_region(self):
  593         """EndpointV3 creates the region before creating the endpoint.
  594 
  595         This occurs when endpoint is provided with 'region' and no 'region_id'.
  596         """
  597         ref = unit.new_endpoint_ref_with_region(service_id=self.service_id,
  598                                                 region=uuid.uuid4().hex)
  599         self.post('/endpoints', body={'endpoint': ref})
  600         # Make sure the region is created
  601         self.get('/regions/%(region_id)s' % {'region_id': ref["region"]})
  602 
  603     def test_create_endpoint_with_no_region(self):
  604         """EndpointV3 allows to creates the endpoint without region."""
  605         ref = unit.new_endpoint_ref(service_id=self.service_id, region_id=None)
  606         del ref['region_id']  # cannot just be None, it needs to not exist
  607         self.post('/endpoints', body={'endpoint': ref})
  608 
  609     def test_create_endpoint_with_empty_url(self):
  610         """Call ``POST /endpoints``."""
  611         ref = unit.new_endpoint_ref(service_id=self.service_id, url='')
  612         self.post('/endpoints', body={'endpoint': ref},
  613                   expected_status=http.client.BAD_REQUEST)
  614 
  615     def test_get_head_endpoint(self):
  616         """Call ``GET & HEAD /endpoints/{endpoint_id}``."""
  617         resource_url = '/endpoints/%(endpoint_id)s' % {
  618             'endpoint_id': self.endpoint_id}
  619         r = self.get(resource_url)
  620         self.assertValidEndpointResponse(r, self.endpoint)
  621         self.head(resource_url, expected_status=http.client.OK)
  622 
  623     def test_update_endpoint(self):
  624         """Call ``PATCH /endpoints/{endpoint_id}``."""
  625         ref = unit.new_endpoint_ref(service_id=self.service_id,
  626                                     interface='public',
  627                                     region_id=self.region_id)
  628         del ref['id']
  629         r = self.patch(
  630             '/endpoints/%(endpoint_id)s' % {
  631                 'endpoint_id': self.endpoint_id},
  632             body={'endpoint': ref})
  633         ref['enabled'] = True
  634         self.assertValidEndpointResponse(r, ref)
  635 
  636     def test_update_endpoint_enabled_true(self):
  637         """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: True."""
  638         r = self.patch(
  639             '/endpoints/%(endpoint_id)s' % {
  640                 'endpoint_id': self.endpoint_id},
  641             body={'endpoint': {'enabled': True}})
  642         self.assertValidEndpointResponse(r, self.endpoint)
  643 
  644     def test_update_endpoint_enabled_false(self):
  645         """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: False."""
  646         r = self.patch(
  647             '/endpoints/%(endpoint_id)s' % {
  648                 'endpoint_id': self.endpoint_id},
  649             body={'endpoint': {'enabled': False}})
  650         exp_endpoint = copy.copy(self.endpoint)
  651         exp_endpoint['enabled'] = False
  652         self.assertValidEndpointResponse(r, exp_endpoint)
  653 
  654     def test_update_endpoint_enabled_str_true(self):
  655         """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: 'True'."""
  656         self.patch(
  657             '/endpoints/%(endpoint_id)s' % {
  658                 'endpoint_id': self.endpoint_id},
  659             body={'endpoint': {'enabled': 'True'}},
  660             expected_status=http.client.BAD_REQUEST)
  661 
  662     def test_update_endpoint_enabled_str_false(self):
  663         """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: 'False'."""
  664         self.patch(
  665             '/endpoints/%(endpoint_id)s' % {
  666                 'endpoint_id': self.endpoint_id},
  667             body={'endpoint': {'enabled': 'False'}},
  668             expected_status=http.client.BAD_REQUEST)
  669 
  670     def test_update_endpoint_enabled_str_random(self):
  671         """Call ``PATCH /endpoints/{endpoint_id}`` with enabled: 'kitties'."""
  672         self.patch(
  673             '/endpoints/%(endpoint_id)s' % {
  674                 'endpoint_id': self.endpoint_id},
  675             body={'endpoint': {'enabled': 'kitties'}},
  676             expected_status=http.client.BAD_REQUEST)
  677 
  678     def test_delete_endpoint(self):
  679         """Call ``DELETE /endpoints/{endpoint_id}``."""
  680         self.delete(
  681             '/endpoints/%(endpoint_id)s' % {
  682                 'endpoint_id': self.endpoint_id})
  683 
  684     def test_deleting_endpoint_with_space_in_url(self):
  685         # add a space to all urls (intentional "i d" to test bug)
  686         url_with_space = "http://127.0.0.1:8774 /v1.1/\\$(tenant_i d)s"
  687 
  688         # create a v3 endpoint ref
  689         ref = unit.new_endpoint_ref(service_id=self.service['id'],
  690                                     region_id=None,
  691                                     publicurl=url_with_space,
  692                                     internalurl=url_with_space,
  693                                     adminurl=url_with_space,
  694                                     url=url_with_space)
  695 
  696         # add the endpoint to the database
  697         PROVIDERS.catalog_api.create_endpoint(ref['id'], ref)
  698 
  699         # delete the endpoint
  700         self.delete('/endpoints/%s' % ref['id'])
  701 
  702         # make sure it's deleted (GET should return Not Found)
  703         self.get('/endpoints/%s' % ref['id'],
  704                  expected_status=http.client.NOT_FOUND)
  705 
  706     def test_endpoint_create_with_valid_url(self):
  707         """Create endpoint with valid url should be tested,too."""
  708         # list one valid url is enough, no need to list too much
  709         valid_url = 'http://127.0.0.1:8774/v1.1/$(project_id)s'
  710 
  711         ref = unit.new_endpoint_ref(self.service_id,
  712                                     interface='public',
  713                                     region_id=self.region_id,
  714                                     url=valid_url)
  715         self.post('/endpoints', body={'endpoint': ref})
  716 
  717     def test_endpoint_create_with_valid_url_project_id(self):
  718         """Create endpoint with valid url should be tested,too."""
  719         valid_url = 'http://127.0.0.1:8774/v1.1/$(project_id)s'
  720 
  721         ref = unit.new_endpoint_ref(self.service_id,
  722                                     interface='public',
  723                                     region_id=self.region_id,
  724                                     url=valid_url)
  725         self.post('/endpoints', body={'endpoint': ref})
  726 
  727     def test_endpoint_create_with_invalid_url(self):
  728         """Test the invalid cases: substitutions is not exactly right."""
  729         invalid_urls = [
  730             # using a substitution that is not whitelisted - KeyError
  731             'http://127.0.0.1:8774/v1.1/$(nonexistent)s',
  732 
  733             # invalid formatting - ValueError
  734             'http://127.0.0.1:8774/v1.1/$(project_id)',
  735             'http://127.0.0.1:8774/v1.1/$(project_id)t',
  736             'http://127.0.0.1:8774/v1.1/$(project_id',
  737 
  738             # invalid type specifier - TypeError
  739             # admin_url is a string not an int
  740             'http://127.0.0.1:8774/v1.1/$(admin_url)d',
  741         ]
  742 
  743         ref = unit.new_endpoint_ref(self.service_id)
  744 
  745         for invalid_url in invalid_urls:
  746             ref['url'] = invalid_url
  747             self.post('/endpoints',
  748                       body={'endpoint': ref},
  749                       expected_status=http.client.BAD_REQUEST)
  750 
  751 
  752 class TestMultiRegion(test_v3.RestfulTestCase):
  753 
  754     def test_catalog_with_multi_region_reports_all_endpoints(self):
  755 
  756         # Create two separate regions
  757         first_region = self.post(
  758             '/regions',
  759             body={'region': unit.new_region_ref()}
  760         ).json_body['region']
  761 
  762         second_region = self.post(
  763             '/regions',
  764             body={'region': unit.new_region_ref()}
  765         ).json_body['region']
  766 
  767         # Create two services with the same type but separate name.
  768         first_service = self.post(
  769             '/services',
  770             body={'service': unit.new_service_ref(type='foobar')}
  771         ).json_body['service']
  772 
  773         second_service = self.post(
  774             '/services',
  775             body={'service': unit.new_service_ref(type='foobar')}
  776         ).json_body['service']
  777 
  778         # Create an endpoint for each service
  779         first_endpoint = self.post(
  780             '/endpoints',
  781             body={
  782                 'endpoint': unit.new_endpoint_ref(
  783                     first_service['id'],
  784                     interface='public',
  785                     region_id=first_region['id']
  786                 )
  787             }
  788         ).json_body['endpoint']
  789 
  790         second_endpoint = self.post(
  791             '/endpoints',
  792             body={
  793                 'endpoint': unit.new_endpoint_ref(
  794                     second_service['id'],
  795                     interface='public',
  796                     region_id=second_region['id']
  797                 )
  798             }
  799         ).json_body['endpoint']
  800 
  801         # Assert the endpoints and services from each region are in the
  802         # catalog.
  803         found_first_endpoint = False
  804         found_second_endpoint = False
  805         catalog = self.get('/auth/catalog/').json_body['catalog']
  806         for service in catalog:
  807             if service['id'] == first_service['id']:
  808                 endpoint = service['endpoints'][0]
  809                 self.assertEqual(endpoint['id'], first_endpoint['id'])
  810                 self.assertEqual(endpoint['region_id'], first_region['id'])
  811                 found_first_endpoint = True
  812             elif service['id'] == second_service['id']:
  813                 endpoint = service['endpoints'][0]
  814                 self.assertEqual(endpoint['id'], second_endpoint['id'])
  815                 self.assertEqual(endpoint['region_id'], second_region['id'])
  816                 found_second_endpoint = True
  817 
  818         self.assertTrue(found_first_endpoint)
  819         self.assertTrue(found_second_endpoint)
  820 
  821 
  822 class TestCatalogAPISQL(unit.TestCase):
  823     """Test for the catalog Manager against the SQL backend."""
  824 
  825     def setUp(self):
  826         super(TestCatalogAPISQL, self).setUp()
  827         self.useFixture(database.Database())
  828         self.load_backends()
  829 
  830         service = unit.new_service_ref()
  831         self.service_id = service['id']
  832         PROVIDERS.catalog_api.create_service(self.service_id, service)
  833 
  834         self.create_endpoint(service_id=self.service_id)
  835 
  836         PROVIDERS.resource_api.create_domain(
  837             default_fixtures.ROOT_DOMAIN['id'], default_fixtures.ROOT_DOMAIN)
  838 
  839     def create_endpoint(self, service_id, **kwargs):
  840         endpoint = unit.new_endpoint_ref(service_id=service_id,
  841                                          region_id=None, **kwargs)
  842 
  843         PROVIDERS.catalog_api.create_endpoint(endpoint['id'], endpoint)
  844         return endpoint
  845 
  846     def config_overrides(self):
  847         super(TestCatalogAPISQL, self).config_overrides()
  848         self.config_fixture.config(group='catalog', driver='sql')
  849 
  850     def test_get_catalog_ignores_endpoints_with_invalid_urls(self):
  851         user_id = uuid.uuid4().hex
  852 
  853         # create a project since the project should exist if we want to
  854         # filter the catalog by the project or replace the url with a
  855         # valid project id.
  856         domain = unit.new_domain_ref()
  857         PROVIDERS.resource_api.create_domain(domain['id'], domain)
  858         project = unit.new_project_ref(domain_id=domain['id'])
  859         PROVIDERS.resource_api.create_project(project['id'], project)
  860 
  861         # the only endpoint in the catalog is the one created in setUp
  862         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project['id'])
  863         self.assertEqual(1, len(catalog[0]['endpoints']))
  864         # it's also the only endpoint in the backend
  865         self.assertEqual(1, len(PROVIDERS.catalog_api.list_endpoints()))
  866 
  867         # create a new, invalid endpoint - malformed type declaration
  868         self.create_endpoint(self.service_id,
  869                              url='http://keystone/%(project_id)')
  870 
  871         # create a new, invalid endpoint - nonexistent key
  872         self.create_endpoint(self.service_id,
  873                              url='http://keystone/%(you_wont_find_me)s')
  874 
  875         # verify that the invalid endpoints don't appear in the catalog
  876         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project['id'])
  877         self.assertEqual(1, len(catalog[0]['endpoints']))
  878         # all three appear in the backend
  879         self.assertEqual(3, len(PROVIDERS.catalog_api.list_endpoints()))
  880 
  881         # create another valid endpoint - project_id will be replaced
  882         self.create_endpoint(self.service_id,
  883                              url='http://keystone/%(project_id)s')
  884 
  885         # there are two valid endpoints, positive check
  886         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project['id'])
  887         self.assertThat(catalog[0]['endpoints'], matchers.HasLength(2))
  888 
  889         # If the URL has no 'project_id' to substitute, we will skip the
  890         # endpoint which contains this kind of URL, negative check.
  891         project_id = None
  892         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project_id)
  893         self.assertThat(catalog[0]['endpoints'], matchers.HasLength(1))
  894 
  895     def test_get_catalog_always_returns_service_name(self):
  896         user_id = uuid.uuid4().hex
  897         # create a project since the project should exist if we want to
  898         # filter the catalog by the project or replace the url with a
  899         # valid project id.
  900         domain = unit.new_domain_ref()
  901         PROVIDERS.resource_api.create_domain(domain['id'], domain)
  902         project = unit.new_project_ref(domain_id=domain['id'])
  903         PROVIDERS.resource_api.create_project(project['id'], project)
  904 
  905         # create a service, with a name
  906         named_svc = unit.new_service_ref()
  907         PROVIDERS.catalog_api.create_service(named_svc['id'], named_svc)
  908         self.create_endpoint(service_id=named_svc['id'])
  909 
  910         # create a service, with no name
  911         unnamed_svc = unit.new_service_ref(name=None)
  912         del unnamed_svc['name']
  913         PROVIDERS.catalog_api.create_service(unnamed_svc['id'], unnamed_svc)
  914         self.create_endpoint(service_id=unnamed_svc['id'])
  915 
  916         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project['id'])
  917 
  918         named_endpoint = [ep for ep in catalog
  919                           if ep['type'] == named_svc['type']][0]
  920         self.assertEqual(named_svc['name'], named_endpoint['name'])
  921 
  922         unnamed_endpoint = [ep for ep in catalog
  923                             if ep['type'] == unnamed_svc['type']][0]
  924         self.assertEqual('', unnamed_endpoint['name'])
  925 
  926 
  927 # TODO(dstanek): this needs refactoring with the test above, but we are in a
  928 # crunch so that will happen in a future patch.
  929 class TestCatalogAPISQLRegions(unit.TestCase):
  930     """Test for the catalog Manager against the SQL backend."""
  931 
  932     def setUp(self):
  933         super(TestCatalogAPISQLRegions, self).setUp()
  934         self.useFixture(database.Database())
  935         self.load_backends()
  936         PROVIDERS.resource_api.create_domain(
  937             default_fixtures.ROOT_DOMAIN['id'], default_fixtures.ROOT_DOMAIN)
  938 
  939     def config_overrides(self):
  940         super(TestCatalogAPISQLRegions, self).config_overrides()
  941         self.config_fixture.config(group='catalog', driver='sql')
  942 
  943     def test_get_catalog_returns_proper_endpoints_with_no_region(self):
  944         service = unit.new_service_ref()
  945         service_id = service['id']
  946         PROVIDERS.catalog_api.create_service(service_id, service)
  947 
  948         endpoint = unit.new_endpoint_ref(service_id=service_id,
  949                                          region_id=None)
  950         del endpoint['region_id']
  951         PROVIDERS.catalog_api.create_endpoint(endpoint['id'], endpoint)
  952 
  953         # create a project since the project should exist if we want to
  954         # filter the catalog by the project or replace the url with a
  955         # valid project id.
  956         domain = unit.new_domain_ref()
  957         PROVIDERS.resource_api.create_domain(domain['id'], domain)
  958         project = unit.new_project_ref(domain_id=domain['id'])
  959         PROVIDERS.resource_api.create_project(project['id'], project)
  960         user_id = uuid.uuid4().hex
  961         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project['id'])
  962         self.assertValidCatalogEndpoint(
  963             catalog[0]['endpoints'][0], ref=endpoint)
  964 
  965     def test_get_catalog_returns_proper_endpoints_with_region(self):
  966         service = unit.new_service_ref()
  967         service_id = service['id']
  968         PROVIDERS.catalog_api.create_service(service_id, service)
  969 
  970         endpoint = unit.new_endpoint_ref(service_id=service_id)
  971         region = unit.new_region_ref(id=endpoint['region_id'])
  972         PROVIDERS.catalog_api.create_region(region)
  973         PROVIDERS.catalog_api.create_endpoint(endpoint['id'], endpoint)
  974 
  975         endpoint = PROVIDERS.catalog_api.get_endpoint(endpoint['id'])
  976         user_id = uuid.uuid4().hex
  977         # create a project since the project should exist if we want to
  978         # filter the catalog by the project or replace the url with a
  979         # valid project id.
  980         domain = unit.new_domain_ref()
  981         PROVIDERS.resource_api.create_domain(domain['id'], domain)
  982         project = unit.new_project_ref(domain_id=domain['id'])
  983         PROVIDERS.resource_api.create_project(project['id'], project)
  984 
  985         catalog = PROVIDERS.catalog_api.get_v3_catalog(user_id, project['id'])
  986         self.assertValidCatalogEndpoint(
  987             catalog[0]['endpoints'][0], ref=endpoint)
  988 
  989     def assertValidCatalogEndpoint(self, entity, ref=None):
  990         keys = ['description', 'id', 'interface', 'name', 'region_id', 'url']
  991         for k in keys:
  992             self.assertEqual(ref.get(k), entity[k], k)
  993         self.assertEqual(entity['region_id'], entity['region'])
  994 
  995 
  996 class TestCatalogAPITemplatedProject(test_v3.RestfulTestCase):
  997     """Templated Catalog doesn't support full API.
  998 
  999     Eg. No region/endpoint creation.
 1000 
 1001     """
 1002 
 1003     def config_overrides(self):
 1004         super(TestCatalogAPITemplatedProject, self).config_overrides()
 1005         self.config_fixture.config(group='catalog', driver='templated')
 1006 
 1007     def load_fixtures(self, fixtures):
 1008         self.load_sample_data(create_region_and_endpoints=False)
 1009 
 1010     def test_project_delete(self):
 1011         """Deleting a project should not result in an 500 ISE.
 1012 
 1013         Deleting a project will create a notification, which the EndpointFilter
 1014         functionality will use to clean up any project->endpoint and
 1015         project->endpoint_group relationships. The templated catalog does not
 1016         support such relationships, but the act of attempting to delete them
 1017         should not cause a NotImplemented exception to be exposed to an API
 1018         caller.
 1019 
 1020         Deleting an endpoint has a similar notification and clean up
 1021         mechanism, but since we do not allow deletion of endpoints with the
 1022         templated catalog, there is no testing to do for that action.
 1023         """
 1024         self.delete(
 1025             '/projects/%(project_id)s' % {
 1026                 'project_id': self.project_id})