"Fossies" - the Fresh Open Source Software Archive

Member "keystone-17.0.0/keystone/tests/unit/server/test_keystone_flask.py" (13 May 2020, 31540 Bytes) of package /linux/misc/openstack/keystone-17.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_keystone_flask.py": 16.0.1_vs_17.0.0.

    1 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 # not use this file except in compliance with the License. You may obtain
    3 # a copy of the License at
    4 #
    5 #      http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 # Unless required by applicable law or agreed to in writing, software
    8 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 # License for the specific language governing permissions and limitations
   11 # under the License.
   12 
   13 import uuid
   14 
   15 import fixtures
   16 import flask
   17 import flask_restful
   18 import functools
   19 from oslo_policy import policy
   20 from oslo_serialization import jsonutils
   21 from testtools import matchers
   22 
   23 from keystone.common import context
   24 from keystone.common import json_home
   25 from keystone.common import rbac_enforcer
   26 import keystone.conf
   27 from keystone import exception
   28 from keystone.server.flask import common as flask_common
   29 from keystone.server.flask.request_processing import json_body
   30 from keystone.tests.unit import rest
   31 
   32 
   33 CONF = keystone.conf.CONF
   34 
   35 
   36 class _TestResourceWithCollectionInfo(flask_common.ResourceBase):
   37     collection_key = 'arguments'
   38     member_key = 'argument'
   39     __shared_state__ = {}
   40     _storage_dict = {}
   41 
   42     def __init__(self):
   43         super(_TestResourceWithCollectionInfo, self).__init__()
   44         # Share State, this is for "dummy" backend storage.
   45         self.__dict__ = self.__shared_state__
   46 
   47     @classmethod
   48     def _reset(cls):
   49         # Used after a test to ensure clean-state
   50         cls._storage_dict.clear()
   51         cls.__shared_state__.clear()
   52 
   53     def _list_arguments(self):
   54         return self.wrap_collection(list(self._storage_dict.values()))
   55 
   56     def get(self, argument_id=None):
   57         # List with no argument, get resource with id, used for HEAD as well.
   58         rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
   59             action='example:allowed')
   60         if argument_id is None:
   61             # List
   62             return self._list_arguments()
   63         else:
   64             # get resource with id
   65             try:
   66                 return self.wrap_member(self._storage_dict[argument_id])
   67             except KeyError:
   68                 raise exception.NotFound(target=argument_id)
   69 
   70     def post(self):
   71         rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
   72             action='example:allowed')
   73         ref = flask.request.get_json(force=True)
   74         ref = self._assign_unique_id(ref)
   75         self._storage_dict[ref['id']] = ref
   76         return self.wrap_member(self._storage_dict[ref['id']]), 201
   77 
   78     def put(self, argument_id):
   79         rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
   80             action='example:allowed')
   81         try:
   82             self._storage_dict[argument_id]
   83         except KeyError:
   84             raise exception.NotFound(target=argument_id)
   85         ref = flask.request.get_json(force=True)
   86         self._require_matching_id(ref)
   87         # Maintain the ref id
   88         ref['id'] = argument_id
   89         self._storage_dict[argument_id] = ref
   90         return '', 204
   91 
   92     def patch(self, argument_id):
   93         rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
   94             action='example:allowed')
   95         try:
   96             self._storage_dict[argument_id]
   97         except KeyError:
   98             raise exception.NotFound(target=argument_id)
   99         ref = flask.request.get_json(force=True)
  100         self._require_matching_id(ref)
  101         self._storage_dict[argument_id].update(ref)
  102         return self.wrap_member(self._storage_dict[argument_id])
  103 
  104     def delete(self, argument_id):
  105         rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
  106             action='example:allowed')
  107         try:
  108             del self._storage_dict[argument_id]
  109         except KeyError:
  110             raise exception.NotFound(target=argument_id)
  111         return '', 204
  112 
  113 
  114 class _TestRestfulAPI(flask_common.APIBase):
  115     _name = 'test_api_base'
  116     _import_name = __name__
  117     resources = []
  118     resource_mapping = []
  119 
  120     def __init__(self, *args, **kwargs):
  121         self.resource_mapping = kwargs.pop('resource_mapping', [])
  122         self.resources = kwargs.pop('resources',
  123                                     [_TestResourceWithCollectionInfo])
  124         super(_TestRestfulAPI, self).__init__(*args, **kwargs)
  125 
  126 
  127 class TestKeystoneFlaskCommon(rest.RestfulTestCase):
  128 
  129     _policy_rules = [
  130         policy.RuleDefault(
  131             name='example:allowed',
  132             check_str=''
  133         ),
  134         policy.RuleDefault(
  135             name='example:deny',
  136             check_str='false:false'
  137         )
  138     ]
  139 
  140     def setUp(self):
  141         super(TestKeystoneFlaskCommon, self).setUp()
  142         enf = rbac_enforcer.enforcer.RBACEnforcer()
  143 
  144         def register_rules(enf_obj):
  145             enf_obj.register_defaults(self._policy_rules)
  146 
  147         self.useFixture(fixtures.MockPatchObject(
  148             enf, 'register_rules', register_rules))
  149         self.useFixture(fixtures.MockPatchObject(
  150             rbac_enforcer.enforcer, '_POSSIBLE_TARGET_ACTIONS',
  151             {r.name for r in self._policy_rules}))
  152 
  153         enf._reset()
  154         self.addCleanup(enf._reset)
  155         self.addCleanup(
  156             _TestResourceWithCollectionInfo._reset)
  157 
  158     def _get_token(self):
  159         auth_json = {
  160             'auth': {
  161                 'identity': {
  162                     'methods': ['password'],
  163                     'password': {
  164                         'user': {
  165                             'name': self.user_req_admin['name'],
  166                             'password': self.user_req_admin['password'],
  167                             'domain': {
  168                                 'id': self.user_req_admin['domain_id']
  169                             }
  170                         }
  171                     }
  172                 },
  173                 'scope': {
  174                     'project': {
  175                         'id': self.project_service['id']
  176                     }
  177                 }
  178             }
  179         }
  180         return self.test_client().post(
  181             '/v3/auth/tokens',
  182             json=auth_json,
  183             expected_status_code=201).headers['X-Subject-Token']
  184 
  185     def _setup_flask_restful_api(self, **options):
  186 
  187         self.restful_api_opts = options.copy()
  188         orig_value = _TestResourceWithCollectionInfo.api_prefix
  189         setattr(_TestResourceWithCollectionInfo,
  190                 'api_prefix', options.get('api_url_prefix', ''))
  191         self.addCleanup(setattr, _TestResourceWithCollectionInfo, 'api_prefix',
  192                         orig_value)
  193         self.restful_api = _TestRestfulAPI(**options)
  194         self.public_app.app.register_blueprint(self.restful_api.blueprint)
  195         self.cleanup_instance('restful_api')
  196         self.cleanup_instance('restful_api_opts')
  197 
  198     def _make_requests(self):
  199         path_base = '/arguments'
  200         api_prefix = self.restful_api_opts.get('api_url_prefix', '')
  201         blueprint_prefix = self.restful_api._blueprint_url_prefix.rstrip('/')
  202         url = ''.join(
  203             [x for x in [blueprint_prefix, api_prefix, path_base] if x])
  204         headers = {'X-Auth-Token': self._get_token()}
  205         with self.test_client() as c:
  206             # GET LIST
  207             resp = c.get(url, headers=headers)
  208             self.assertEqual(
  209                 _TestResourceWithCollectionInfo.wrap_collection(
  210                     []), resp.json)
  211             unknown_id = uuid.uuid4().hex
  212 
  213             # GET non-existent ref
  214             c.get('%s/%s' % (url, unknown_id), headers=headers,
  215                   expected_status_code=404)
  216 
  217             # HEAD non-existent ref
  218             c.head('%s/%s' % (url, unknown_id), headers=headers,
  219                    expected_status_code=404)
  220 
  221             # PUT non-existent ref
  222             c.put('%s/%s' % (url, unknown_id), json={}, headers=headers,
  223                   expected_status_code=404)
  224 
  225             # PATCH non-existent ref
  226             c.patch('%s/%s' % (url, unknown_id), json={}, headers=headers,
  227                     expected_status_code=404)
  228 
  229             # DELETE non-existent ref
  230             c.delete('%s/%s' % (url, unknown_id), headers=headers,
  231                      expected_status_code=404)
  232 
  233             # POST new ref
  234             new_argument_resource = {'testing': uuid.uuid4().hex}
  235             new_argument_resp = c.post(
  236                 url,
  237                 json=new_argument_resource,
  238                 headers=headers).json['argument']
  239 
  240             # POST second new ref
  241             new_argument2_resource = {'testing': uuid.uuid4().hex}
  242             new_argument2_resp = c.post(
  243                 url,
  244                 json=new_argument2_resource,
  245                 headers=headers).json['argument']
  246 
  247             # GET list
  248             get_list_resp = c.get(url, headers=headers).json
  249             self.assertIn(new_argument_resp,
  250                           get_list_resp['arguments'])
  251             self.assertIn(new_argument2_resp,
  252                           get_list_resp['arguments'])
  253 
  254             # GET first ref
  255             get_resp = c.get('%s/%s' % (url, new_argument_resp['id']),
  256                              headers=headers).json['argument']
  257             self.assertEqual(new_argument_resp, get_resp)
  258 
  259             # HEAD first ref
  260             head_resp = c.head(
  261                 '%s/%s' % (url, new_argument_resp['id']),
  262                 headers=headers).data
  263             # NOTE(morgan): For python3 compat, explicitly binary type
  264             self.assertEqual(head_resp, b'')
  265 
  266             # PUT update first ref
  267             replacement_argument = {'new_arg': True, 'id': uuid.uuid4().hex}
  268             c.put('%s/%s' % (url, new_argument_resp['id']), headers=headers,
  269                   json=replacement_argument, expected_status_code=400)
  270             replacement_argument.pop('id')
  271             c.put('%s/%s' % (url, new_argument_resp['id']),
  272                   headers=headers,
  273                   json=replacement_argument)
  274             put_resp = c.get('%s/%s' % (url, new_argument_resp['id']),
  275                              headers=headers).json['argument']
  276             self.assertNotIn(new_argument_resp['testing'],
  277                              put_resp)
  278             self.assertTrue(put_resp['new_arg'])
  279 
  280             # GET first ref (check for replacement)
  281             get_replacement_resp = c.get(
  282                 '%s/%s' % (url, new_argument_resp['id']),
  283                 headers=headers).json['argument']
  284             self.assertEqual(put_resp,
  285                              get_replacement_resp)
  286 
  287             # PATCH update first ref
  288             patch_ref = {'uuid': uuid.uuid4().hex}
  289             patch_resp = c.patch('%s/%s' % (url, new_argument_resp['id']),
  290                                  headers=headers,
  291                                  json=patch_ref).json['argument']
  292             self.assertTrue(patch_resp['new_arg'])
  293             self.assertEqual(patch_ref['uuid'], patch_resp['uuid'])
  294 
  295             # GET first ref (check for update)
  296             get_patched_ref_resp = c.get(
  297                 '%s/%s' % (url, new_argument_resp['id']),
  298                 headers=headers).json['argument']
  299             self.assertEqual(patch_resp,
  300                              get_patched_ref_resp)
  301 
  302             # DELETE first ref
  303             c.delete(
  304                 '%s/%s' % (url, new_argument_resp['id']),
  305                 headers=headers)
  306             # Check that it was in-fact deleted
  307             c.get(
  308                 '%s/%s' % (url, new_argument_resp['id']),
  309                 headers=headers, expected_status_code=404)
  310 
  311     def test_api_url_prefix(self):
  312         url_prefix = '/%s' % uuid.uuid4().hex
  313         self._setup_flask_restful_api(
  314             api_url_prefix=url_prefix)
  315         self._make_requests()
  316 
  317     def test_blueprint_url_prefix(self):
  318         url_prefix = '/%s' % uuid.uuid4().hex
  319         self._setup_flask_restful_api(
  320             blueprint_url_prefix=url_prefix)
  321         self._make_requests()
  322 
  323     def test_build_restful_api_no_prefix(self):
  324         self._setup_flask_restful_api()
  325         self._make_requests()
  326 
  327     def test_cannot_add_before_request_functions_twice(self):
  328 
  329         class TestAPIDuplicateBefore(_TestRestfulAPI):
  330             def __init__(self):
  331                 super(TestAPIDuplicateBefore, self).__init__()
  332                 self._register_before_request_functions()
  333 
  334         self.assertRaises(AssertionError, TestAPIDuplicateBefore)
  335 
  336     def test_cannot_add_after_request_functions_twice(self):
  337 
  338         class TestAPIDuplicateAfter(_TestRestfulAPI):
  339             def __init__(self):
  340                 super(TestAPIDuplicateAfter, self).__init__()
  341                 self._register_after_request_functions()
  342 
  343         self.assertRaises(AssertionError, TestAPIDuplicateAfter)
  344 
  345     def test_after_request_functions_must_be_added(self):
  346 
  347         class TestAPINoAfter(_TestRestfulAPI):
  348             def _register_after_request_functions(self, functions=None):
  349                 pass
  350 
  351         self.assertRaises(AssertionError, TestAPINoAfter)
  352 
  353     def test_before_request_functions_must_be_added(self):
  354 
  355         class TestAPINoBefore(_TestRestfulAPI):
  356             def _register_before_request_functions(self, functions=None):
  357                 pass
  358 
  359         self.assertRaises(AssertionError, TestAPINoBefore)
  360 
  361     def test_before_request_functions(self):
  362         # Test additional "before" request functions fire.
  363         attr = uuid.uuid4().hex
  364 
  365         def do_something():
  366             setattr(flask.g, attr, True)
  367 
  368         class TestAPI(_TestRestfulAPI):
  369             def _register_before_request_functions(self, functions=None):
  370                 functions = functions or []
  371                 functions.append(do_something)
  372                 super(TestAPI, self)._register_before_request_functions(
  373                     functions)
  374 
  375         api = TestAPI(resources=[_TestResourceWithCollectionInfo])
  376         self.public_app.app.register_blueprint(api.blueprint)
  377         token = self._get_token()
  378         with self.test_client() as c:
  379             c.get('/v3/arguments', headers={'X-Auth-Token': token})
  380             self.assertTrue(getattr(flask.g, attr, False))
  381 
  382     def test_after_request_functions(self):
  383         # Test additional "after" request functions fire. In this case, we
  384         # alter the response code to 420
  385         attr = uuid.uuid4().hex
  386 
  387         def do_something(resp):
  388             setattr(flask.g, attr, True)
  389             resp.status_code = 420
  390             return resp
  391 
  392         class TestAPI(_TestRestfulAPI):
  393             def _register_after_request_functions(self, functions=None):
  394                 functions = functions or []
  395                 functions.append(do_something)
  396                 super(TestAPI, self)._register_after_request_functions(
  397                     functions)
  398 
  399         api = TestAPI(resources=[_TestResourceWithCollectionInfo])
  400         self.public_app.app.register_blueprint(api.blueprint)
  401         token = self._get_token()
  402         with self.test_client() as c:
  403             c.get('/v3/arguments', headers={'X-Auth-Token': token},
  404                   expected_status_code=420)
  405 
  406     def test_construct_resource_map(self):
  407         resource_name = 'arguments'
  408         param_relation = json_home.build_v3_parameter_relation(
  409             'argument_id')
  410         alt_rel_func = functools.partial(
  411             json_home.build_v3_extension_resource_relation,
  412             extension_name='extension', extension_version='1.0')
  413         url = '/v3/arguments/<string:argument_id>'
  414         old_url = [dict(
  415             url='/v3/old_arguments/<string:argument_id>',
  416             json_home=flask_common.construct_json_home_data(
  417                 rel='arguments',
  418                 resource_relation_func=alt_rel_func)
  419         )]
  420 
  421         mapping = flask_common.construct_resource_map(
  422             resource=_TestResourceWithCollectionInfo,
  423             url=url,
  424             resource_kwargs={},
  425             alternate_urls=old_url,
  426             rel=resource_name,
  427             status=json_home.Status.EXPERIMENTAL,
  428             path_vars={'argument_id': param_relation},
  429             resource_relation_func=json_home.build_v3_resource_relation)
  430         self.assertEqual(_TestResourceWithCollectionInfo,
  431                          mapping.resource)
  432         self.assertEqual(url, mapping.url)
  433         self.assertEqual(json_home.build_v3_resource_relation(resource_name),
  434                          mapping.json_home_data.rel)
  435         self.assertEqual(json_home.Status.EXPERIMENTAL,
  436                          mapping.json_home_data.status)
  437         self.assertEqual({'argument_id': param_relation},
  438                          mapping.json_home_data.path_vars)
  439         # Check the alternate URL data is populated sanely
  440         self.assertEqual(1, len(mapping.alternate_urls))
  441         alt_url_data = mapping.alternate_urls[0]
  442         self.assertEqual(old_url[0]['url'], alt_url_data['url'])
  443         self.assertEqual(old_url[0]['json_home'], alt_url_data['json_home'])
  444 
  445     def test_instantiate_and_register_to_app(self):
  446         # Test that automatic instantiation and registration to app works.
  447         self.restful_api_opts = {}
  448         self.restful_api = _TestRestfulAPI.instantiate_and_register_to_app(
  449             self.public_app.app)
  450         self.cleanup_instance('restful_api_opts')
  451         self.cleanup_instance('restful_api')
  452         self._make_requests()
  453 
  454     def test_unenforced_api_decorator(self):
  455         # Test unenforced decorator works as expected
  456 
  457         class MappedResource(flask_restful.Resource):
  458             @flask_common.unenforced_api
  459             def post(self):
  460                 post_body = flask.request.get_json()
  461                 return {'post_body': post_body}, 201
  462 
  463         resource_map = flask_common.construct_resource_map(
  464             resource=MappedResource,
  465             url='test_api',
  466             alternate_urls=[],
  467             resource_kwargs={},
  468             rel='test',
  469             status=json_home.Status.STABLE,
  470             path_vars=None,
  471             resource_relation_func=json_home.build_v3_resource_relation)
  472 
  473         restful_api = _TestRestfulAPI(resource_mapping=[resource_map],
  474                                       resources=[])
  475         self.public_app.app.register_blueprint(restful_api.blueprint)
  476         token = self._get_token()
  477         with self.test_client() as c:
  478             body = {'test_value': uuid.uuid4().hex}
  479             # Works with token
  480             resp = c.post('/v3/test_api', json=body,
  481                           headers={'X-Auth-Token': token})
  482             self.assertEqual(body, resp.json['post_body'])
  483             # Works without token
  484             resp = c.post('/v3/test_api', json=body)
  485             self.assertEqual(body, resp.json['post_body'])
  486 
  487     def test_HTTP_OPTIONS_is_unenforced(self):
  488         # Standup a test mapped resource and call OPTIONS on it. This will
  489         # return a header "Allow" with the valid methods, in this case
  490         # OPTIONS and POST. Ensure that the response otherwise conforms
  491         # as expected.
  492 
  493         class MappedResource(flask_restful.Resource):
  494             def post(self):
  495                 # we don't actually use this or call it.
  496                 pass
  497 
  498         resource_map = flask_common.construct_resource_map(
  499             resource=MappedResource,
  500             url='test_api',
  501             alternate_urls=[],
  502             resource_kwargs={},
  503             rel='test',
  504             status=json_home.Status.STABLE,
  505             path_vars=None,
  506             resource_relation_func=json_home.build_v3_resource_relation)
  507 
  508         restful_api = _TestRestfulAPI(resource_mapping=[resource_map],
  509                                       resources=[])
  510         self.public_app.app.register_blueprint(restful_api.blueprint)
  511         with self.test_client() as c:
  512             r = c.options('/v3/test_api')
  513             # make sure we split the data and left/right strip off whitespace
  514             # The use of a SET here is to ensure the exact values are in place
  515             # even if hash-seeds change order. `r.data` will be an empty
  516             # byte-string. `Content-Length` will be 0.
  517             self.assertEqual(
  518                 set(['OPTIONS', 'POST']),
  519                 set([v.lstrip().rstrip()
  520                      for v in r.headers['Allow'].split(',')]))
  521             self.assertEqual(r.headers['Content-Length'], '0')
  522             self.assertEqual(r.data, b'')
  523 
  524     def test_mapped_resource_routes(self):
  525         # Test non-standard URL routes ("mapped") function as expected
  526 
  527         class MappedResource(flask_restful.Resource):
  528             def post(self):
  529                 rbac_enforcer.enforcer.RBACEnforcer().enforce_call(
  530                     action='example:allowed')
  531                 post_body = flask.request.get_json()
  532                 return {'post_body': post_body}, 201
  533 
  534         resource_map = flask_common.construct_resource_map(
  535             resource=MappedResource,
  536             url='test_api',
  537             alternate_urls=[],
  538             resource_kwargs={},
  539             rel='test',
  540             status=json_home.Status.STABLE,
  541             path_vars=None,
  542             resource_relation_func=json_home.build_v3_resource_relation)
  543 
  544         restful_api = _TestRestfulAPI(resource_mapping=[resource_map],
  545                                       resources=[])
  546         self.public_app.app.register_blueprint(restful_api.blueprint)
  547         token = self._get_token()
  548         with self.test_client() as c:
  549             body = {'test_value': uuid.uuid4().hex}
  550             resp = c.post('/v3/test_api', json=body,
  551                           headers={'X-Auth-Token': token})
  552             self.assertEqual(body, resp.json['post_body'])
  553 
  554     def test_correct_json_home_document(self):
  555         class MappedResource(flask_restful.Resource):
  556             def post(self):
  557                 rbac_enforcer.enforcer.RBACEnforcer().enforce_call(
  558                     action='example:allowed')
  559                 post_body = flask.request.get_json()
  560                 return {'post_body': post_body}
  561 
  562         # NOTE(morgan): totally fabricated json_home data based upon our TEST
  563         # restful_apis.
  564         json_home_data = {
  565             'https://docs.openstack.org/api/openstack-identity/3/'
  566             'rel/argument': {
  567                 'href-template': '/v3/arguments/{argument_id}',
  568                 'href-vars': {
  569                     'argument_id': 'https://docs.openstack.org/api/'
  570                                    'openstack-identity/3/param/argument_id'
  571                 }
  572             },
  573             'https://docs.openstack.org/api/openstack-identity/3/'
  574             'rel/arguments': {
  575                 'href': '/v3/arguments'
  576             },
  577             'https://docs.openstack.org/api/openstack-identity/3/'
  578             'rel/test': {
  579                 'href': '/v3/test_api'
  580             },
  581         }
  582 
  583         resource_map = flask_common.construct_resource_map(
  584             resource=MappedResource,
  585             url='test_api',
  586             alternate_urls=[],
  587             resource_kwargs={},
  588             rel='test',
  589             status=json_home.Status.STABLE,
  590             path_vars=None,
  591             resource_relation_func=json_home.build_v3_resource_relation)
  592 
  593         restful_api = _TestRestfulAPI(resource_mapping=[resource_map])
  594         self.public_app.app.register_blueprint(restful_api.blueprint)
  595 
  596         with self.test_client() as c:
  597             headers = {'Accept': 'application/json-home'}
  598             resp = c.get('/', headers=headers)
  599             resp_data = jsonutils.loads(resp.data)
  600             for rel in json_home_data:
  601                 self.assertThat(resp_data['resources'][rel],
  602                                 matchers.Equals(json_home_data[rel]))
  603 
  604     def test_normalize_domain_id_extracts_domain_id_if_needed(self):
  605         self._setup_flask_restful_api()
  606         blueprint_prefix = self.restful_api._blueprint_url_prefix.rstrip('/')
  607         url = ''.join([blueprint_prefix, '/arguments'])
  608         headers = {'X-Auth-Token': self._get_token()}
  609         ref_with_domain_id = {'domain_id': uuid.uuid4().hex}
  610         ref_without_domain_id = {}
  611         with self.test_client() as c:
  612             # Make a dummy request.. ANY request is fine to push the whole
  613             # context stack.
  614             c.get('%s/%s' % (url, uuid.uuid4().hex), headers=headers,
  615                   expected_status_code=404)
  616 
  617             oslo_context = flask.request.environ[
  618                 context.REQUEST_CONTEXT_ENV]
  619 
  620             # Normal Project Scope Form
  621             # ---------------------------
  622             # test that _normalize_domain_id does something sane
  623             domain_id = ref_with_domain_id['domain_id']
  624             # Ensure we don't change the domain if it is specified
  625             flask_common.ResourceBase._normalize_domain_id(ref_with_domain_id)
  626             self.assertEqual(domain_id, ref_with_domain_id['domain_id'])
  627             # Ensure (deprecated) we add default domain if needed
  628             flask_common.ResourceBase._normalize_domain_id(
  629                 ref_without_domain_id)
  630             self.assertEqual(
  631                 CONF.identity.default_domain_id,
  632                 ref_without_domain_id['domain_id'])
  633             ref_without_domain_id.clear()
  634 
  635             # Domain Scoped Form
  636             # --------------------
  637             # Just set oslo_context domain_id to a value. This is how we
  638             # communicate domain scope. No need to explicitly
  639             # do a domain-scoped request, this is a synthetic text anyway
  640             oslo_context.domain_id = uuid.uuid4().hex
  641             # Ensure we don't change the domain if it is specified
  642             flask_common.ResourceBase._normalize_domain_id(ref_with_domain_id)
  643             self.assertEqual(domain_id, ref_with_domain_id['domain_id'])
  644             flask_common.ResourceBase._normalize_domain_id(
  645                 ref_without_domain_id)
  646             self.assertEqual(oslo_context.domain_id,
  647                              ref_without_domain_id['domain_id'])
  648             ref_without_domain_id.clear()
  649 
  650             # "Admin" Token form
  651             # -------------------
  652             # Explicitly set "is_admin" to true, no new request is needed
  653             # as we simply check "is_admin" value everywhere
  654             oslo_context.is_admin = True
  655             oslo_context.domain_id = None
  656             # Ensure we don't change the domain if it is specified
  657             flask_common.ResourceBase._normalize_domain_id(ref_with_domain_id)
  658             self.assertEqual(domain_id, ref_with_domain_id['domain_id'])
  659             # Ensure we raise an appropriate exception with the inferred
  660             # domain_id
  661             self.assertRaises(exception.ValidationError,
  662                               flask_common.ResourceBase._normalize_domain_id,
  663                               ref=ref_without_domain_id)
  664 
  665     def test_api_prefix_self_referential_link_substitution(self):
  666 
  667         view_arg = uuid.uuid4().hex
  668 
  669         class TestResource(flask_common.ResourceBase):
  670             api_prefix = '/<string:test_value>/nothing'
  671 
  672         # use a dummy request context, no enforcement is happening
  673         # therefore we don't need the heavy lifting of a full request
  674         # run.
  675         with self.test_request_context(
  676                 path='/%s/nothing/values' % view_arg,
  677                 base_url='https://localhost/'):
  678             # explicitly set the view_args, this is a special case
  679             # for a synthetic test case, usually one would rely on
  680             # a full request stack to set these.
  681             flask.request.view_args = {'test_value': view_arg}
  682 
  683             # create dummy ref
  684             ref = {'id': uuid.uuid4().hex}
  685 
  686             # add the self referential link
  687             TestResource._add_self_referential_link(
  688                 ref, collection_name='values')
  689 
  690             # Check that the link in fact starts with what we expect
  691             # including the explicit view arg.
  692             self.assertTrue(ref['links']['self'].startswith(
  693                 'https://localhost/v3/%s' % view_arg)
  694             )
  695 
  696     def test_json_body_before_req_func_valid_json(self):
  697         with self.test_request_context(
  698                 headers={'Content-Type': 'application/json'},
  699                 data='{"key": "value"}'):
  700             # No exception should be raised, everything is happy.
  701             json_body.json_body_before_request()
  702 
  703     def test_json_body_before_req_func_invalid_json(self):
  704         with self.test_request_context(
  705                 headers={'Content-Type': 'application/json'},
  706                 data='invalid JSON'):
  707             # keystone.exception.ValidationError should be raised
  708             self.assertRaises(exception.ValidationError,
  709                               json_body.json_body_before_request)
  710 
  711     def test_json_body_before_req_func_no_content_type(self):
  712         # Unset
  713         with self.test_request_context(data='{"key": "value"}'):
  714             # No exception should be raised, everything is happy.
  715             json_body.json_body_before_request()
  716 
  717         # Explicitly set to ''
  718         with self.test_request_context(
  719                 headers={'Content-Type': ''}, data='{"key": "value"}'):
  720             # No exception should be raised, everything is happy.
  721             json_body.json_body_before_request()
  722 
  723     def test_json_body_before_req_func_unrecognized_content_type(self):
  724         with self.test_request_context(
  725                 headers={'Content-Type': 'unrecognized/content-type'},
  726                 data='{"key": "value"'):
  727             # keystone.exception.ValidationError should be raised
  728             self.assertRaises(exception.ValidationError,
  729                               json_body.json_body_before_request)
  730 
  731     def test_json_body_before_req_func_unrecognized_conten_type_no_body(self):
  732         with self.test_request_context(
  733                 headers={'Content-Type': 'unrecognized/content-type'}):
  734             # No exception should be raised, everything is happy.
  735             json_body.json_body_before_request()
  736 
  737     def test_resource_collection_key_raises_exception_if_unset(self):
  738         class TestResource(flask_common.ResourceBase):
  739             """A Test Resource."""
  740 
  741         class TestResourceWithKey(flask_common.ResourceBase):
  742             collection_key = uuid.uuid4().hex
  743 
  744         r = TestResource()
  745         self.assertRaises(ValueError, getattr, r, 'collection_key')
  746 
  747         r = TestResourceWithKey()
  748         self.assertEqual(
  749             TestResourceWithKey.collection_key, r.collection_key)
  750 
  751     def test_resource_member_key_raises_exception_if_unset(self):
  752         class TestResource(flask_common.ResourceBase):
  753             """A Test Resource."""
  754 
  755         class TestResourceWithKey(flask_common.ResourceBase):
  756             member_key = uuid.uuid4().hex
  757 
  758         r = TestResource()
  759         self.assertRaises(ValueError, getattr, r, 'member_key')
  760 
  761         r = TestResourceWithKey()
  762         self.assertEqual(
  763             TestResourceWithKey.member_key, r.member_key)
  764 
  765 
  766 class TestKeystoneFlaskUnrouted404(rest.RestfulTestCase):
  767     def setUp(self):
  768         super(TestKeystoneFlaskUnrouted404, self).setUp()
  769         # unregister the 404 handler we explicitly set in loadapp. This
  770         # makes the 404 error fallback to a standard werkzeug handling.
  771         self.public_app.app.error_handler_spec[None].pop(404)
  772 
  773     def test_unrouted_path_is_not_jsonified_404(self):
  774         with self.test_client() as c:
  775             path = '/{unrouted_path}'.format(unrouted_path=uuid.uuid4())
  776             resp = c.get(path, expected_status_code=404)
  777             # Make sure we're emitting a html error
  778             self.assertIn('text/html', resp.headers['Content-Type'])
  779             # Ensure the more generic flask/werkzeug 404 response is emitted
  780             self.assertTrue(b'404 Not Found' in resp.data)