Flesh out and add testing for flask_RESTful scaffolding
Add in support for JSON Home documents, a ResourceBase implementing basic functionality, and full testing of the new flask_RESTful scaffolding. Change-Id: I5bcc8660b68c0b39a2110089f6c67531769d14ef Parital-Bug: #1776504
This commit is contained in:
parent
6d1456061d
commit
16be22b428
|
@ -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
|
||||
|
|
|
@ -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` 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
|
||||
|
|
|
@ -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]))
|
Loading…
Reference in New Issue