Merge "Flesh out and add testing for flask_RESTful scaffolding"

This commit is contained in:
Zuul 2018-07-16 21:44:19 +00:00 committed by Gerrit Code Review
commit 19e28d0cd3
4 changed files with 1049 additions and 19 deletions

View File

@ -13,31 +13,127 @@
import abc
import collections
import functools
import itertools
import re
import uuid
import wsgiref.util
import flask
from flask import blueprints
from flask import g
import flask_restful
from oslo_log import log
import six
from keystone.common import driver_hints
from keystone.common import json_home
from keystone.common.rbac_enforcer import enforcer
from keystone.common import utils
import keystone.conf
from keystone import exception
# NOTE(morgan): Capture the relevant part of the flask url route rule for
# substitution. In flask arguments (e.g. url elements to be passed to the
# "resource" method, e.g. user_id, are specified like `<string:user_id>`
# we use this regex to replace the <> with {} for JSON Home purposes and
# remove the argument type. Use of this is done like
# _URL_SUBST.sub('{\\1}', entity_path), which replaces the whole match
# match rule bit with the capture group (this is a greedy sub).
_URL_SUBST = re.compile(r'<[^\s:]+:([^>]+)>')
CONF = keystone.conf.CONF
LOG = log.getLogger(__name__)
ResourceMap = collections.namedtuple('resource_map', 'resource, urls, kwargs')
ResourceMap = collections.namedtuple(
'resource_map', 'resource, url, alternate_urls, kwargs, json_home_data')
JsonHomeData = collections.namedtuple(
'json_home_data', 'rel, status, path_vars')
_v3_resource_relation = json_home.build_v3_resource_relation
def construct_resource_map(resource, url, resource_kwargs, alternate_urls=None,
rel=None, status=json_home.Status.STABLE,
path_vars=None,
resource_relation_func=_v3_resource_relation):
"""Construct the ResourceMap Named Tuple.
:param resource: The flask-RESTful resource class implementing the methods
for the API.
:type resource: :class:`ResourceMap`
:param url: Flask-standard url route, all flask url routing rules apply.
url variables will be passed to the Resource methods as
arguments.
:type url: str
:param resource_kwargs: a dict of optional value(s) that can further modify
the handling of the routing.
* endpoint: endpoint name (defaults to
:meth:`Resource.__name__.lower`
Can be used to reference this route in
:class:`fields.Url` fields (str)
* resource_class_args: args to be forwarded to the
constructor of the resource.
(tuple)
* resource_class_kwargs: kwargs to be forwarded to
the constructor of the
resource. (dict)
Additional keyword arguments not specified above
will be passed as-is to
:meth:`flask.Flask.add_url_rule`.
:param alternate_urls: An iterable (list) of urls that also map to the
resource. These are used to ensure API compat when
a "new" path is more correct for the API but old
paths must continue to work. Example:
`/auth/domains` being the new path for
`/OS-FEDERATION/domains`. The `OS-FEDERATION` part
would be listed as an alternate url. These are not
added to the JSON Home Document.
:type: any iterable or None
:param rel:
:type rel: str or None
:param status: JSON Home API Status, e.g. "STABLE"
:type status: str
:param path_vars: JSON Home Path Var Data (arguments)
:type path_vars: dict or None
:param resource_relation_func: function to build expected resource rel data
:type resource_relation_func: callable
:return:
"""
if rel is not None:
jh_data = construct_json_home_data(
rel=rel, status=status, path_vars=path_vars,
resource_relation_func=resource_relation_func)
else:
jh_data = None
if not url.startswith('/'):
url = '/%s' % url
return ResourceMap(
resource=resource, url=url, alternate_urls=alternate_urls,
kwargs=resource_kwargs, json_home_data=jh_data)
def construct_json_home_data(rel, status=json_home.Status.STABLE,
path_vars=None,
resource_relation_func=_v3_resource_relation):
rel = resource_relation_func(rel)
return JsonHomeData(rel=rel, status=status, path_vars=(path_vars or {}))
def _initialize_rbac_enforcement_check():
setattr(g, enforcer._ENFORCEMENT_CHECK_ATTR, False)
def _assert_rbac_enforcement_called():
def _assert_rbac_enforcement_called(resp):
# assert is intended to be used to ensure code during development works
# as expected, it is fine to be optimized out with `python -O`
msg = ('PROGRAMMING ERROR: enforcement (`keystone.common.rbac_enforcer.'
'enforcer.RBACKEnforcer.enforce_call()`) has not been called; API '
'is unenforced.')
assert getattr(g, enforcer._ENFORCEMENT_CHECK_ATTR, False), msg # nosec
return resp
@six.add_metaclass(abc.ABCMeta)
@ -64,10 +160,18 @@ class APIBase(object):
* resource: a :class:`flask_restful.Resource` class or subclass
* urls: a url route or iterable of url routes to match for the
resource, standard flask routing rules apply. Any url
variables will be passed to the resource method as args.
(str)
* url: a url route to match for the resource, standard flask
routing rules apply. Any url variables will be passed
to the resource method as args. (str)
* alternate_urls: an iterable of url routes to match for the
resource, standard flask routing rules apply.
These rules are in addition (for API compat) to
the primary url. Any url variables will be
passed to the resource method as args. (iterable)
* json_home_data: :class:`JsonHomeData` populated with relevant
info for populated JSON Home Documents or None.
* kwargs: a dict of optional value(s) that can further modify the
handling of the routing.
@ -90,36 +194,60 @@ class APIBase(object):
"""
raise NotImplementedError()
@property
def resources(self):
return []
@staticmethod
def _build_bp_url_prefix(prefix):
# NOTE(morgan): Keystone only has a V3 API, this is here for future
# proofing and exceptional cases such as root discovery API object(s)
parts = ['/v3']
if prefix:
parts.append(prefix)
return '/'.join(parts)
parts.append(prefix.lstrip('/'))
return '/'.join(parts).rstrip('/')
@property
def api(self):
# The API may be directly accessed via this property
return self.__api
@property
def blueprint(self):
# The API Blueprint may be directly accessed via this property
return self.__api_bp
return self.__blueprint
def __init__(self, blueprint_url_prefix='', api_url_prefix='',
default_mediatype='application/json', decorators=None,
errors=None):
self.__before_request_functions_added = False
self.__after_request_functions_added = False
self._blueprint_url_prefix = blueprint_url_prefix
self._default_mediatype = default_mediatype
self._api_url_prefix = api_url_prefix
blueprint_url_prefix = blueprint_url_prefix.rstrip('/')
api_url_prefix = api_url_prefix.rstrip('/')
if api_url_prefix and not api_url_prefix.startswith('/'):
self._api_url_prefix = '/%s' % api_url_prefix
else:
self._api_url_prefix = api_url_prefix
if blueprint_url_prefix and not blueprint_url_prefix.startswith('/'):
self._blueprint_url_prefix = self._build_bp_url_prefix(
'/%s' % blueprint_url_prefix)
else:
self._blueprint_url_prefix = self._build_bp_url_prefix(
blueprint_url_prefix)
self.__blueprint = blueprints.Blueprint(
name=self._name, import_name=self._import_name,
url_prefix=self._build_bp_url_prefix(self._blueprint_url_prefix))
self.__api_bp = flask_restful.Api(
url_prefix=self._blueprint_url_prefix)
self.__api = flask_restful.Api(
app=self.__blueprint, prefix=self._api_url_prefix,
default_mediatype=self._default_mediatype,
decorators=decorators, errors=errors)
self._add_resources()
self._add_mapped_resources()
# Apply Before and After request functions
self._register_before_request_functions()
@ -131,12 +259,91 @@ class APIBase(object):
assert self.__after_request_functions_added, msg % 'after' # nosec
def _add_resources(self):
# Add resources that are standardized. Each resource implements a
# base set of handling for a collection of entities such as
# `users`. Resources are sourced from self.resources. Each resource
# should have an attribute/property containing the `collection_key`
# which is typically the "plural" form of the entity, e.g. `users` and
# `member_key` which is typically the "singular" of the entity, e.g.
# `user`. Resources are sourced from self.resources, each element is
# simply a :class:`flask_restful.Resource`.
for r in self.resources:
c_key = getattr(r, 'collection_key', None)
m_key = getattr(r, 'member_key', None)
if not c_key or not m_key:
LOG.debug('Unable to add resource %(resource)s to API '
'%(name)s, both `member_key` and `collection_key` '
'must be implemented. [collection_key(%(col_key)s) '
'member_key(%(m_key)s)]',
{'resource': r.__class__.view_class.__name__,
'name': self._name, 'col_key': c_key,
'm_key': m_key})
continue
collection_path = '/%s' % c_key
entity_path = '/%(collection_key)s/<string:%(member_key)s_id>' % {
'collection_key': c_key, 'member_key': m_key}
# NOTE(morgan): The json-home form of the entity path is different
# from the flask-url routing form.
jh_e_path = _URL_SUBST.sub('{\\1}', entity_path)
LOG.debug(
'Adding standard routes to API %(name)s for `%(resource)s` '
'[%(collection_path)s, %(entity_path)s]', {
'name': self._name, 'resource': r.__class__.__name__,
'collection_path': collection_path,
'entity_path': entity_path})
self.api.add_resource(r, collection_path, entity_path)
# Add JSON Home data
collection_rel = json_home.build_v3_resource_relation(c_key)
rel_data = {'href': collection_path}
entity_rel = json_home.build_v3_resource_relation(m_key)
id_str = '%s_id' % m_key
id_param_rel = json_home.build_v3_parameter_relation(id_str)
entity_rel_data = {'href-template': jh_e_path,
'href-vars': {id_str: id_param_rel}}
json_home.JsonHomeResources.append_resource(
collection_rel, rel_data)
json_home.JsonHomeResources.append_resource(
entity_rel, entity_rel_data)
def _add_mapped_resources(self):
# Add resource mappings, non-standard resource connections
for r in self.resource_mapping:
LOG.debug(
'Adding resource routes to API %(name)s: '
'[%(urls)r %(kwargs)r]',
{'name': self._name, 'urls': r.urls, 'kwargs': r.kwargs})
self.blueprint.add_resource(r.resource, *r.urls, **r.kwargs)
'[%(url)r %(kwargs)r]',
{'name': self._name, 'url': r.url, 'kwargs': r.kwargs})
self.api.add_resource(r.resource, r.url, **r.kwargs)
if r.alternate_urls is not None:
LOG.debug(
'Adding additional resource routes (alternate) to API'
'%(name)s: [%(urls)r %(kwargs)r]',
{'name': self._name, 'urls': r.alternate_urls,
'kwargs': r.kwargs})
self.api.add_resource(r.resource, *r.alternate_urls,
**r.kwargs)
# Build the JSON Home data and add it to the relevant JSON Home
# Documents for explicit JSON Home data.
if r.json_home_data:
resource_data = {}
# NOTE(morgan): JSON Home form of the URL is different
# from FLASK, do the conversion here.
conv_url = _URL_SUBST.sub('{\\1}', r.url)
if r.json_home_data.path_vars:
resource_data['href-template'] = conv_url
resource_data['href-vars'] = r.json_home_data.path_vars
else:
resource_data['href'] = conv_url
json_home.Status.update_resource_data(
resource_data, r.json_home_data.status)
json_home.JsonHomeResources.append_resource(
r.json_home_data.rel,
resource_data)
def _register_before_request_functions(self, functions=None):
"""Register functions to be executed in the `before request` phase.
@ -226,5 +433,271 @@ class APIBase(object):
blueprint is loaded. Anything beyond defaults should be done
explicitly via normal instantiation where more values may be passed
via :meth:`__init__`.
:returns: :class:`keystone.server.flask.common.APIBase`
"""
flask_app.register_blueprint(cls().blueprint)
inst = cls()
flask_app.register_blueprint(inst.blueprint)
return inst
class ResourceBase(flask_restful.Resource):
collection_key = None
member_key = None
method_decorators = []
def __init__(self):
super(ResourceBase, self).__init__()
if self.collection_key is None:
raise ValueError('PROGRAMMING ERROR: `self.collection_key` '
'cannot be `None`.')
if self.member_key is None:
raise ValueError('PROGRAMMING ERROR: `self.member_key` cannot '
'be `None`.')
@staticmethod
def _assign_unique_id(ref):
ref = ref.copy()
ref['id'] = uuid.uuid4().hex
return ref
@classmethod
def _require_matching_id(cls, ref):
"""Ensure the value matches the reference's ID, if any."""
id_arg = None
if cls.member_key is not None:
id_arg = flask.request.view_args.get('%s_id' % cls.member_key)
if ref.get('id') is not None and id_arg != ref['id']:
raise exception.ValidationError('Cannot change ID')
@classmethod
def wrap_collection(cls, refs, hints=None):
"""Wrap a collection, checking for filtering and pagination.
Returns the wrapped collection, which includes:
- Executing any filtering not already carried out
- Truncate to a set limit if necessary
- Adds 'self' links in every member
- Adds 'next', 'self' and 'prev' links for the whole collection.
:param refs: the list of members of the collection
:param hints: list hints, containing any relevant filters and limit.
Any filters already satisfied by managers will have been
removed
"""
# Check if there are any filters in hints that were not handled by
# the drivers. The driver will not have paginated or limited the
# output if it found there were filters it was unable to handle
if hints:
refs = cls.filter_by_attributes(refs, hints)
list_limited, refs = cls.limit(refs, hints)
for ref in refs:
cls._add_self_referential_link(ref)
container = {cls.collection_key: refs}
self_url = full_url()
container['links'] = {
'next': None,
'self': self_url,
'previous': None
}
if list_limited:
container['truncated'] = True
return container
@classmethod
def wrap_member(cls, ref):
cls._add_self_referential_link(ref)
return {cls.member_key: ref}
@classmethod
def _add_self_referential_link(cls, ref):
self_link = '/'.join([base_url(), 'v3', cls.collection_key])
ref.setdefault('links', {})['self'] = self_link
@classmethod
def filter_by_attributes(cls, refs, hints):
"""Filter a list of references by filter values."""
def _attr_match(ref_attr, val_attr):
"""Matche attributes allowing for booleans as strings.
We test explicitly for a value that defines it as 'False',
which also means that the existence of the attribute with
no value implies 'True'
"""
if type(ref_attr) is bool:
return ref_attr == utils.attr_as_boolean(val_attr)
else:
return ref_attr == val_attr
def _inexact_attr_match(inexact_filter, ref):
"""Apply an inexact filter to a result dict.
:param inexact_filter: the filter in question
:param ref: the dict to check
:returns: True if there is a match
"""
comparator = inexact_filter['comparator']
key = inexact_filter['name']
if key in ref:
filter_value = inexact_filter['value']
target_value = ref[key]
if not inexact_filter['case_sensitive']:
# We only support inexact filters on strings so
# it's OK to use lower()
filter_value = filter_value.lower()
target_value = target_value.lower()
if comparator == 'contains':
return (filter_value in target_value)
elif comparator == 'startswith':
return target_value.startswith(filter_value)
elif comparator == 'endswith':
return target_value.endswith(filter_value)
else:
# We silently ignore unsupported filters
return True
return False
for f in hints.filters:
if f['comparator'] == 'equals':
attr = f['name']
value = f['value']
refs = [r for r in refs if _attr_match(
utils.flatten_dict(r).get(attr), value)]
else:
# It might be an inexact filter
refs = [r for r in refs if _inexact_attr_match(f, r)]
return refs
@staticmethod
def build_driver_hints(supported_filters):
"""Build list hints based on the context query string.
:param supported_filters: list of filters supported, so ignore any
keys in query_dict that are not in this list.
"""
hints = driver_hints.Hints()
if not flask.request.args:
return hints
for key, value in flask.request.args.items():
# Check if this is an exact filter
if supported_filters is None or key in supported_filters:
hints.add_filter(key, value)
continue
# Check if it is an inexact filter
for valid_key in supported_filters:
# See if this entry in query_dict matches a known key with an
# inexact suffix added. If it doesn't match, then that just
# means that there is no inexact filter for that key in this
# query.
if not key.startswith(valid_key + '__'):
continue
base_key, comparator = key.split('__', 1)
# We map the query-style inexact of, for example:
#
# {'email__contains', 'myISP'}
#
# into a list directive add filter call parameters of:
#
# name = 'email'
# value = 'myISP'
# comparator = 'contains'
# case_sensitive = True
case_sensitive = True
if comparator.startswith('i'):
case_sensitive = False
comparator = comparator[1:]
hints.add_filter(base_key, value,
comparator=comparator,
case_sensitive=case_sensitive)
# NOTE(henry-nash): If we were to support pagination, we would pull any
# pagination directives out of the query_dict here, and add them into
# the hints list.
return hints
@classmethod
def limit(cls, refs, hints):
"""Limit a list of entities.
The underlying driver layer may have already truncated the collection
for us, but in case it was unable to handle truncation we check here.
:param refs: the list of members of the collection
:param hints: hints, containing, among other things, the limit
requested
:returns: boolean indicating whether the list was truncated, as well
as the list of (truncated if necessary) entities.
"""
NOT_LIMITED = False
LIMITED = True
if hints is None or hints.limit is None:
# No truncation was requested
return NOT_LIMITED, refs
if hints.limit.get('truncated', False):
# The driver did truncate the list
return LIMITED, refs
if len(refs) > hints.limit['limit']:
# The driver layer wasn't able to truncate it for us, so we must
# do it here
return LIMITED, refs[:hints.limit['limit']]
return NOT_LIMITED, refs
def base_url():
url = CONF['public_endpoint']
if url:
substitutions = dict(
itertools.chain(CONF.items(), CONF.eventlet_server.items()))
url = url % substitutions
elif flask.request.environ:
url = wsgiref.util.application_uri(flask.request.environ)
# remove version from the URL as it may be part of SCRIPT_NAME but
# it should not be part of base URL
url = re.sub(r'/v(3|(2\.0))/*$', '', url)
# now remove the standard port
url = utils.remove_standard_port(url)
else:
# if we don't have enough information to come up with a base URL,
# then fall back to localhost. This should never happen in
# production environment.
url = 'http://localhost:%d' % CONF.eventlet_server.public_port
return url.rstrip('/')
def full_url():
subs = {'url': base_url(), 'query_string': ''}
qs = flask.request.environ.get('QUERY_STRING')
if qs:
subs['query_string'] = '?%s' % qs
return '%(url)s%(query_string)s' % subs

View File

@ -37,6 +37,7 @@ from oslo_log import fixture as log_fixture
from oslo_log import log
from oslo_utils import timeutils
import six
from six.moves import http_client
from sqlalchemy import exc
import testtools
from testtools import testcase
@ -489,13 +490,30 @@ def _assert_expected_status(f):
`expected_status_code` must be passed as a kwarg.
"""
TEAPOT_HTTP_STATUS = 418
_default_expected_responses = {
'get': http_client.OK,
'head': http_client.OK,
'post': http_client.CREATED,
'put': http_client.NO_CONTENT,
'patch': http_client.OK,
'delete': http_client.NO_CONTENT,
}
@functools.wraps(f)
def inner(*args, **kwargs):
expected_status_code = kwargs.pop('expected_status_code', 200)
# Get the "expected_status_code" kwarg if supplied. If not supplied use
# the `_default_expected_response` mapping, or fall through to
# "HTTP OK" if the method is somehow unknown.
expected_status_code = kwargs.pop(
'expected_status_code',
_default_expected_responses.get(
f.__name__.lower(), http_client.OK))
response = f(*args, **kwargs)
# Logic to verify the response object is sane. Expand as needed
if response.status_code == 418:
if response.status_code == TEAPOT_HTTP_STATUS:
# NOTE(morgan): We use 418 internally during tests to indicate
# an un-routed HTTP call was made. This allows us to avoid
# misinterpreting HTTP 404 from Flask and HTTP 404 from a

View File

View File

@ -0,0 +1,539 @@
# 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 uuid
import fixtures
import flask
import flask_restful
from oslo_policy import policy
from oslo_serialization import jsonutils
from testtools import matchers
from keystone.common import json_home
from keystone.common import rbac_enforcer
from keystone import exception
from keystone.server.flask import common as flask_common
from keystone.tests.unit import rest
class _TestResourceWithCollectionInfo(flask_common.ResourceBase):
collection_key = 'arguments'
member_key = 'argument'
__shared_state__ = {}
_storage_dict = {}
def __init__(self):
super(_TestResourceWithCollectionInfo, self).__init__()
# Share State, this is for "dummy" backend storage.
self.__dict__ = self.__shared_state__
@classmethod
def _reset(cls):
# Used after a test to ensure clean-state
cls._storage_dict.clear()
cls.__shared_state__.clear()
def _list_arguments(self):
return self.wrap_collection(list(self._storage_dict.values()))
def get(self, argument_id=None):
# List with no argument, get resource with id, used for HEAD as well.
rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
action='example:allowed')
if argument_id is None:
# List
return self._list_arguments()
else:
# get resource with id
try:
return self.wrap_member(self._storage_dict[argument_id])
except KeyError:
raise exception.NotFound(target=argument_id)
def post(self):
rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
action='example:allowed')
ref = flask.request.get_json(force=True)
ref = self._assign_unique_id(ref)
self._storage_dict[ref['id']] = ref
return self.wrap_member(self._storage_dict[ref['id']]), 201
def put(self, argument_id):
rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
action='example:allowed')
try:
self._storage_dict[argument_id]
except KeyError:
raise exception.NotFound(target=argument_id)
ref = flask.request.get_json(force=True)
self._require_matching_id(ref)
self._storage_dict[argument_id] = ref
return '', 204
def patch(self, argument_id):
rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
action='example:allowed')
try:
self._storage_dict[argument_id]
except KeyError:
raise exception.NotFound(target=argument_id)
ref = flask.request.get_json(force=True)
self._require_matching_id(ref)
self._storage_dict[argument_id].update(ref)
return self.wrap_member(self._storage_dict[argument_id])
def delete(self, argument_id):
rbac_enforcer.enforcer.RBACEnforcer.enforce_call(
action='example:allowed')
try:
del self._storage_dict[argument_id]
except KeyError:
raise exception.NotFound(target=argument_id)
return '', 204
class _TestRestfulAPI(flask_common.APIBase):
_name = 'test_api_base'
_import_name = __name__
resources = []
resource_mapping = []
def __init__(self, *args, **kwargs):
self.resource_mapping = kwargs.pop('resource_mapping', [])
self.resources = kwargs.pop('resources',
[_TestResourceWithCollectionInfo])
super(_TestRestfulAPI, self).__init__(*args, **kwargs)
class TestKeystoneFlaskCommon(rest.RestfulTestCase):
_policy_rules = [
policy.RuleDefault(
name='example:allowed',
check_str=''
),
policy.RuleDefault(
name='example:deny',
check_str='false:false'
)
]
def setUp(self):
super(TestKeystoneFlaskCommon, self).setUp()
enf = rbac_enforcer.enforcer.RBACEnforcer()
def register_rules(enf_obj):
enf_obj.register_defaults(self._policy_rules)
self.useFixture(fixtures.MockPatchObject(
enf, 'register_rules', register_rules))
self.useFixture(fixtures.MockPatchObject(
rbac_enforcer.enforcer, '_POSSIBLE_TARGET_ACTIONS',
{r.name for r in self._policy_rules}))
enf._reset()
self.addCleanup(enf._reset)
self.addCleanup(
_TestResourceWithCollectionInfo._reset)
def _get_token(self):
auth_json = {
'auth': {
'identity': {
'methods': ['password'],
'password': {
'user': {
'name': self.user_req_admin['name'],
'password': self.user_req_admin['password'],
'domain': {
'id': self.user_req_admin['domain_id']
}
}
}
},
'scope': {
'project': {
'id': self.tenant_service['id']
}
}
}
}
return self.test_client().post(
'/v3/auth/tokens',
json=auth_json,
expected_status_code=201).headers['X-Subject-Token']
def _setup_flask_restful_api(self, **options):
self.restful_api_opts = options.copy()
self.restful_api = _TestRestfulAPI(**options)
self.public_app.app.register_blueprint(self.restful_api.blueprint)
self.cleanup_instance('restful_api')
self.cleanup_instance('restful_api_opts')
def _make_requests(self):
path_base = '/arguments'
api_prefix = self.restful_api_opts.get('api_url_prefix', '')
blueprint_prefix = self.restful_api._blueprint_url_prefix.rstrip('/')
url = ''.join(
[x for x in [blueprint_prefix, api_prefix, path_base] if x])
headers = {'X-Auth-Token': self._get_token()}
with self.test_client() as c:
# GET LIST
resp = c.get(url, headers=headers)
self.assertEqual(
_TestResourceWithCollectionInfo.wrap_collection(
[]), resp.json)
unknown_id = uuid.uuid4().hex
# GET non-existent ref
c.get('%s/%s' % (url, unknown_id), headers=headers,
expected_status_code=404)
# HEAD non-existent ref
c.head('%s/%s' % (url, unknown_id), headers=headers,
expected_status_code=404)
# PUT non-existent ref
c.put('%s/%s' % (url, unknown_id), json={}, headers=headers,
expected_status_code=404)
# PATCH non-existent ref
c.patch('%s/%s' % (url, unknown_id), json={}, headers=headers,
expected_status_code=404)
# DELETE non-existent ref
c.delete('%s/%s' % (url, unknown_id), headers=headers,
expected_status_code=404)
# POST new ref
new_argument_resource = {'testing': uuid.uuid4().hex}
new_argument_resp = c.post(
url,
json=new_argument_resource,
headers=headers).json['argument']
# POST second new ref
new_argument2_resource = {'testing': uuid.uuid4().hex}
new_argument2_resp = c.post(
url,
json=new_argument2_resource,
headers=headers).json['argument']
# GET list
get_list_resp = c.get(url, headers=headers).json
self.assertIn(new_argument_resp,
get_list_resp['arguments'])
self.assertIn(new_argument2_resp,
get_list_resp['arguments'])
# GET first ref
get_resp = c.get('%s/%s' % (url, new_argument_resp['id']),
headers=headers).json['argument']
self.assertEqual(new_argument_resp, get_resp)
# HEAD first ref
head_resp = c.head(
'%s/%s' % (url, new_argument_resp['id']),
headers=headers).data
# NOTE(morgan): For python3 compat, explicitly binary type
self.assertEqual(head_resp, b'')
# PUT update first ref
replacement_argument = {'new_arg': True, 'id': uuid.uuid4().hex}
c.put('%s/%s' % (url, new_argument_resp['id']), headers=headers,
json=replacement_argument, expected_status_code=400)
replacement_argument.pop('id')
c.put('%s/%s' % (url, new_argument_resp['id']),
headers=headers,
json=replacement_argument)
put_resp = c.get('%s/%s' % (url, new_argument_resp['id']),
headers=headers).json['argument']
self.assertNotIn(new_argument_resp['testing'],
put_resp)
self.assertTrue(put_resp['new_arg'])
# GET first ref (check for replacement)
get_replacement_resp = c.get(
'%s/%s' % (url, new_argument_resp['id']),
headers=headers).json['argument']
self.assertEqual(put_resp,
get_replacement_resp)
# PATCH update first ref
patch_ref = {'uuid': uuid.uuid4().hex}
patch_resp = c.patch('%s/%s' % (url, new_argument_resp['id']),
headers=headers,
json=patch_ref).json['argument']
self.assertTrue(patch_resp['new_arg'])
self.assertEqual(patch_ref['uuid'], patch_resp['uuid'])
# GET first ref (check for update)
get_patched_ref_resp = c.get(
'%s/%s' % (url, new_argument_resp['id']),
headers=headers).json['argument']
self.assertEqual(patch_resp,
get_patched_ref_resp)
# DELETE first ref
c.delete(
'%s/%s' % (url, new_argument_resp['id']),
headers=headers)
# Check that it was in-fact deleted
c.get(
'%s/%s' % (url, new_argument_resp['id']),
headers=headers, expected_status_code=404)
def test_api_url_prefix(self):
url_prefix = '/%s' % uuid.uuid4().hex
self._setup_flask_restful_api(
api_url_prefix=url_prefix)
self._make_requests()
def test_blueprint_url_prefix(self):
url_prefix = '/%s' % uuid.uuid4().hex
self._setup_flask_restful_api(
blueprint_url_prefix=url_prefix)
self._make_requests()
def test_build_restful_api_no_prefix(self):
self._setup_flask_restful_api()
self._make_requests()
def test_cannot_add_before_request_functions_twice(self):
class TestAPIDuplicateBefore(_TestRestfulAPI):
def __init__(self):
super(TestAPIDuplicateBefore, self).__init__()
self._register_before_request_functions()
self.assertRaises(AssertionError, TestAPIDuplicateBefore)
def test_cannot_add_after_request_functions_twice(self):
class TestAPIDuplicateAfter(_TestRestfulAPI):
def __init__(self):
super(TestAPIDuplicateAfter, self).__init__()
self._register_after_request_functions()
self.assertRaises(AssertionError, TestAPIDuplicateAfter)
def test_after_request_functions_must_be_added(self):
class TestAPINoAfter(_TestRestfulAPI):
def _register_after_request_functions(self, functions=None):
pass
self.assertRaises(AssertionError, TestAPINoAfter)
def test_before_request_functions_must_be_added(self):
class TestAPINoBefore(_TestRestfulAPI):
def _register_before_request_functions(self, functions=None):
pass
self.assertRaises(AssertionError, TestAPINoBefore)
def test_before_request_functions(self):
# Test additional "before" request functions fire.
attr = uuid.uuid4().hex
def do_something():
setattr(flask.g, attr, True)
class TestAPI(_TestRestfulAPI):
def _register_before_request_functions(self, functions=None):
functions = functions or []
functions.append(do_something)
super(TestAPI, self)._register_before_request_functions(
functions)
api = TestAPI(resources=[_TestResourceWithCollectionInfo])
self.public_app.app.register_blueprint(api.blueprint)
token = self._get_token()
with self.test_client() as c:
c.get('/v3/arguments', headers={'X-Auth-Token': token})
self.assertTrue(getattr(flask.g, attr, False))
def test_after_request_functions(self):
# Test additional "after" request functions fire. In this case, we
# alter the response code to 420
attr = uuid.uuid4().hex
def do_something(resp):
setattr(flask.g, attr, True)
resp.status_code = 420
return resp
class TestAPI(_TestRestfulAPI):
def _register_after_request_functions(self, functions=None):
functions = functions or []
functions.append(do_something)
super(TestAPI, self)._register_after_request_functions(
functions)
api = TestAPI(resources=[_TestResourceWithCollectionInfo])
self.public_app.app.register_blueprint(api.blueprint)
token = self._get_token()
with self.test_client() as c:
c.get('/v3/arguments', headers={'X-Auth-Token': token},
expected_status_code=420)
def test_construct_resource_map(self):
param_relation = json_home.build_v3_parameter_relation(
'argument_id')
url = '/v3/arguments/<string:argument_id>'
old_url = ['/v3/old_arguments/<string:argument_id>']
resource_name = 'arguments'
mapping = flask_common.construct_resource_map(
resource=_TestResourceWithCollectionInfo,
url=url,
resource_kwargs={},
alternate_urls=old_url,
rel=resource_name,
status=json_home.Status.EXPERIMENTAL,
path_vars={'argument_id': param_relation},
resource_relation_func=json_home.build_v3_resource_relation)
self.assertEqual(_TestResourceWithCollectionInfo,
mapping.resource)
self.assertEqual(url, mapping.url)
self.assertEqual(old_url, mapping.alternate_urls)
self.assertEqual(json_home.build_v3_resource_relation(resource_name),
mapping.json_home_data.rel)
self.assertEqual(json_home.Status.EXPERIMENTAL,
mapping.json_home_data.status)
self.assertEqual({'argument_id': param_relation},
mapping.json_home_data.path_vars)
def test_instantiate_and_register_to_app(self):
# Test that automatic instantiation and registration to app works.
self.restful_api_opts = {}
self.restful_api = _TestRestfulAPI.instantiate_and_register_to_app(
self.public_app.app)
self.cleanup_instance('restful_api_opts')
self.cleanup_instance('restful_api')
self._make_requests()
def test_unenforced_api_decorator(self):
# Test unenforced decorator works as expected
class MappedResource(flask_restful.Resource):
@_TestRestfulAPI.unenforced_api
def post(self):
post_body = flask.request.get_json()
return {'post_body': post_body}, 201
resource_map = flask_common.construct_resource_map(
resource=MappedResource,
url='test_api',
alternate_urls=[],
resource_kwargs={},
rel='test',
status=json_home.Status.STABLE,
path_vars=None,
resource_relation_func=json_home.build_v3_resource_relation)
restful_api = _TestRestfulAPI(resource_mapping=[resource_map],
resources=[])
self.public_app.app.register_blueprint(restful_api.blueprint)
token = self._get_token()
with self.test_client() as c:
body = {'test_value': uuid.uuid4().hex}
# Works with token
resp = c.post('/v3/test_api', json=body,
headers={'X-Auth-Token': token})
self.assertEqual(body, resp.json['post_body'])
# Works without token
resp = c.post('/v3/test_api', json=body)
self.assertEqual(body, resp.json['post_body'])
def test_mapped_resource_routes(self):
# Test non-standard URL routes ("mapped") function as expected
class MappedResource(flask_restful.Resource):
def post(self):
rbac_enforcer.enforcer.RBACEnforcer().enforce_call(
action='example:allowed')
post_body = flask.request.get_json()
return {'post_body': post_body}, 201
resource_map = flask_common.construct_resource_map(
resource=MappedResource,
url='test_api',
alternate_urls=[],
resource_kwargs={},
rel='test',
status=json_home.Status.STABLE,
path_vars=None,
resource_relation_func=json_home.build_v3_resource_relation)
restful_api = _TestRestfulAPI(resource_mapping=[resource_map],
resources=[])
self.public_app.app.register_blueprint(restful_api.blueprint)
token = self._get_token()
with self.test_client() as c:
body = {'test_value': uuid.uuid4().hex}
resp = c.post('/v3/test_api', json=body,
headers={'X-Auth-Token': token})
self.assertEqual(body, resp.json['post_body'])
def test_correct_json_home_document(self):
class MappedResource(flask_restful.Resource):
def post(self):
rbac_enforcer.enforcer.RBACEnforcer().enforce_call(
action='example:allowed')
post_body = flask.request.get_json()
return {'post_body': post_body}
# NOTE(morgan): totally fabricated json_home data based upon our TEST
# restful_apis.
json_home_data = {
'https://docs.openstack.org/api/openstack-identity/3/'
'rel/argument': {
'href-template': '/v3/arguments/{argument_id}',
'href-vars': {
'argument_id': 'https://docs.openstack.org/api/'
'openstack-identity/3/param/argument_id'
}
},
'https://docs.openstack.org/api/openstack-identity/3/'
'rel/arguments': {
'href': '/v3/arguments'
},
'https://docs.openstack.org/api/openstack-identity/3/'
'rel/test': {
'href': '/v3/test_api'
},
}
resource_map = flask_common.construct_resource_map(
resource=MappedResource,
url='test_api',
alternate_urls=[],
resource_kwargs={},
rel='test',
status=json_home.Status.STABLE,
path_vars=None,
resource_relation_func=json_home.build_v3_resource_relation)
restful_api = _TestRestfulAPI(resource_mapping=[resource_map])
self.public_app.app.register_blueprint(restful_api.blueprint)
with self.test_client() as c:
headers = {'Accept': 'application/json-home'}
resp = c.get('/', headers=headers)
resp_data = jsonutils.loads(resp.data)
for rel in json_home_data:
self.assertThat(resp_data['resources'][rel],
matchers.Equals(json_home_data[rel]))