Refactor API utilities into api_utils.py

This collects up utils that are only used in the API code,
and moves them from cinder/utils.py to here, for better code
organization.

Change-Id: Iecd909ef4e24d5597dcccb80a671a27f361e4b4d
This commit is contained in:
Eric Harney 2019-08-05 15:56:00 -04:00
parent f2b0de24c6
commit 90f962553a
16 changed files with 203 additions and 189 deletions

132
cinder/api/api_utils.py Normal file
View File

@ -0,0 +1,132 @@
# 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.
from oslo_log import log as logging
from oslo_utils import strutils
import six
import webob
from webob import exc
from cinder.i18n import _
LOG = logging.getLogger(__name__)
def _parse_is_public(is_public):
"""Parse is_public into something usable.
* True: List public volume types only
* False: List private volume types only
* None: List both public and private volume types
"""
if is_public is None:
# preserve default value of showing only public types
return True
elif is_none_string(is_public):
return None
else:
try:
return strutils.bool_from_string(is_public, strict=True)
except ValueError:
msg = _('Invalid is_public filter [%s]') % is_public
raise exc.HTTPBadRequest(explanation=msg)
def is_none_string(val):
"""Check if a string represents a None value."""
if not isinstance(val, six.string_types):
return False
return val.lower() == 'none'
def remove_invalid_filter_options(context, filters,
allowed_search_options):
"""Remove search options that are not valid for non-admin API/context."""
if context.is_admin:
# Allow all options
return
# Otherwise, strip out all unknown options
unknown_options = [opt for opt in filters
if opt not in allowed_search_options]
bad_options = ", ".join(unknown_options)
LOG.debug("Removing options '%s' from query.", bad_options)
for opt in unknown_options:
del filters[opt]
_visible_admin_metadata_keys = ['readonly', 'attached_mode']
def add_visible_admin_metadata(volume):
"""Add user-visible admin metadata to regular metadata.
Extracts the admin metadata keys that are to be made visible to
non-administrators, and adds them to the regular metadata structure for the
passed-in volume.
"""
visible_admin_meta = {}
if volume.get('volume_admin_metadata'):
if isinstance(volume['volume_admin_metadata'], dict):
volume_admin_metadata = volume['volume_admin_metadata']
for key in volume_admin_metadata:
if key in _visible_admin_metadata_keys:
visible_admin_meta[key] = volume_admin_metadata[key]
else:
for item in volume['volume_admin_metadata']:
if item['key'] in _visible_admin_metadata_keys:
visible_admin_meta[item['key']] = item['value']
# avoid circular ref when volume is a Volume instance
elif (volume.get('admin_metadata') and
isinstance(volume.get('admin_metadata'), dict)):
for key in _visible_admin_metadata_keys:
if key in volume['admin_metadata'].keys():
visible_admin_meta[key] = volume['admin_metadata'][key]
if not visible_admin_meta:
return
# NOTE(zhiyan): update visible administration metadata to
# volume metadata, administration metadata will rewrite existing key.
if volume.get('volume_metadata'):
orig_meta = list(volume.get('volume_metadata'))
for item in orig_meta:
if item['key'] in visible_admin_meta.keys():
item['value'] = visible_admin_meta.pop(item['key'])
for key, value in visible_admin_meta.items():
orig_meta.append({'key': key, 'value': value})
volume['volume_metadata'] = orig_meta
# avoid circular ref when vol is a Volume instance
elif (volume.get('metadata') and
isinstance(volume.get('metadata'), dict)):
volume['metadata'].update(visible_admin_meta)
else:
volume['metadata'] = visible_admin_meta
def validate_integer(value, name, min_value=None, max_value=None):
"""Make sure that value is a valid integer, potentially within range.
:param value: the value of the integer
:param name: the name of the integer
:param min_length: the min_length of the integer
:param max_length: the max_length of the integer
:returns: integer
"""
try:
value = strutils.validate_integer(value, name, min_value, max_value)
return value
except ValueError as e:
raise webob.exc.HTTPBadRequest(explanation=six.text_type(e))

View File

@ -24,11 +24,11 @@ from oslo_log import log as logging
from six.moves import urllib
import webob
from cinder.api import api_utils
from cinder.api import microversions as mv
from cinder.common import constants
from cinder import exception
from cinder.i18n import _
from cinder import utils
api_common_opts = [
@ -106,7 +106,10 @@ def _get_marker_param(params):
def _get_offset_param(params):
"""Extract offset id from request's dictionary (defaults to 0) or fail."""
offset = params.pop('offset', 0)
return utils.validate_integer(offset, 'offset', 0, constants.DB_MAX_INT)
return api_utils.validate_integer(offset,
'offset',
0,
constants.DB_MAX_INT)
def limited(items, request, max_limit=None):

View File

@ -22,6 +22,7 @@ from oslo_utils import strutils
from six.moves import http_client
from webob import exc
from cinder.api import api_utils
from cinder.api import common
from cinder.api import extensions
from cinder.api import microversions as mv
@ -88,9 +89,10 @@ class BackupsController(wsgi.Controller):
@common.process_general_filtering('backup')
def _process_backup_filtering(self, context=None, filters=None,
req_version=None):
utils.remove_invalid_filter_options(context,
filters,
self._get_backup_filter_options())
api_utils.remove_invalid_filter_options(
context,
filters,
self._get_backup_filter_options())
def _convert_sort_name(self, req_version, sort_keys):
"""Convert sort key "name" to "display_name". """

View File

@ -21,6 +21,7 @@ import six
from six.moves import http_client
import webob
from cinder.api import api_utils
from cinder.api import common
from cinder.api import extensions
from cinder.api.openstack import wsgi
@ -66,8 +67,8 @@ class QoSSpecsController(wsgi.Controller):
sort_keys, sort_dirs = common.get_sort_params(params)
filters = params
allowed_search_options = ('id', 'name', 'consumer')
utils.remove_invalid_filter_options(context, filters,
allowed_search_options)
api_utils.remove_invalid_filter_options(context, filters,
allowed_search_options)
specs = qos_specs.get_all_specs(context, filters=filters,
marker=marker, limit=limit,

View File

@ -16,6 +16,7 @@ from oslo_log import log as logging
from oslo_utils import strutils
from six.moves import http_client
from cinder.api import api_utils
from cinder.api import common
from cinder.api.contrib import resource_common_manage
from cinder.api import extensions
@ -28,7 +29,6 @@ from cinder.api.views import manageable_volumes as list_manageable_view
from cinder import exception
from cinder.i18n import _
from cinder.policies import manageable_volumes as policy
from cinder import utils
from cinder import volume as cinder_volume
from cinder.volume import volume_types
@ -147,7 +147,7 @@ class VolumeManageController(wsgi.Controller):
'value': host or cluster_name}
raise exception.ServiceUnavailable(message=msg)
utils.add_visible_admin_metadata(new_volume)
api_utils.add_visible_admin_metadata(new_volume)
return self._view_builder.detail(req, new_volume)

View File

@ -20,12 +20,12 @@ from oslo_utils import strutils
from six.moves import http_client
import webob
from cinder.api import api_utils
from cinder.api import common
from cinder.api.openstack import wsgi
from cinder.api.schemas import snapshots as snapshot
from cinder.api import validation
from cinder.api.views import snapshots as snapshot_views
from cinder import utils
from cinder import volume
from cinder.volume import utils as volume_utils
@ -84,8 +84,8 @@ class SnapshotsController(wsgi.Controller):
# Filter out invalid options
allowed_search_options = ('status', 'volume_id', 'name')
utils.remove_invalid_filter_options(context, search_opts,
allowed_search_options)
api_utils.remove_invalid_filter_options(context, search_opts,
allowed_search_options)
# NOTE(thingee): v2 API allows name instead of display_name
if 'name' in search_opts:

View File

@ -16,11 +16,10 @@
"""The volume type & volume types extra specs extension."""
import ast
from webob import exc
from oslo_log import log as logging
from oslo_utils import strutils
from cinder.api import api_utils
from cinder.api import common
from cinder.api import microversions as mv
from cinder.api.openstack import wsgi
@ -28,7 +27,6 @@ from cinder.api.v2.views import types as views_types
from cinder import exception
from cinder.i18n import _
from cinder.policies import volume_type as type_policy
from cinder import utils
from cinder.volume import volume_types
LOG = logging.getLogger(__name__)
@ -66,33 +64,13 @@ class VolumeTypesController(wsgi.Controller):
context.authorize(type_policy.GET_POLICY, target_obj=vol_type)
return self._view_builder.show(req, vol_type)
def _parse_is_public(self, is_public):
"""Parse is_public into something usable.
* True: List public volume types only
* False: List private volume types only
* None: List both public and private volume types
"""
if is_public is None:
# preserve default value of showing only public types
return True
elif utils.is_none_string(is_public):
return None
else:
try:
return strutils.bool_from_string(is_public, strict=True)
except ValueError:
msg = _('Invalid is_public filter [%s]') % is_public
raise exc.HTTPBadRequest(explanation=msg)
@common.process_general_filtering('volume_type')
def _process_volume_type_filtering(self, context=None, filters=None,
req_version=None):
utils.remove_invalid_filter_options(context,
filters,
self._get_vol_type_filter_options()
)
api_utils.remove_invalid_filter_options(
context,
filters,
self._get_vol_type_filter_options())
def _get_volume_types(self, req):
"""Helper function that returns a list of type dicts."""
@ -107,11 +85,11 @@ class VolumeTypesController(wsgi.Controller):
filters=filters,
req_version=req_version)
else:
utils.remove_invalid_filter_options(
api_utils.remove_invalid_filter_options(
context, filters, self._get_vol_type_filter_options())
if context.is_admin:
# Only admin has query access to all volume types
filters['is_public'] = self._parse_is_public(
filters['is_public'] = api_utils._parse_is_public(
req.params.get('is_public', None))
else:
filters['is_public'] = True

View File

@ -24,6 +24,7 @@ from six.moves import http_client
import webob
from webob import exc
from cinder.api import api_utils
from cinder.api import common
from cinder.api.contrib import scheduler_hints
from cinder.api import microversions as mv
@ -64,7 +65,7 @@ class VolumeController(wsgi.Controller):
vol = self.volume_api.get(context, id, viewable_admin_meta=True)
req.cache_db_volume(vol)
utils.add_visible_admin_metadata(vol)
api_utils.add_visible_admin_metadata(vol)
return self._view_builder.detail(req, vol)
@ -102,9 +103,10 @@ class VolumeController(wsgi.Controller):
# NOTE(wanghao): Always removing glance_metadata since we support it
# only in API version >= VOLUME_LIST_GLANCE_METADATA.
filters.pop('glance_metadata', None)
utils.remove_invalid_filter_options(context,
filters,
self._get_volume_filter_options())
api_utils.remove_invalid_filter_options(
context,
filters,
self._get_volume_filter_options())
# NOTE(thingee): v2 API allows name instead of display_name
if 'name' in sort_keys:
@ -122,7 +124,7 @@ class VolumeController(wsgi.Controller):
offset=offset)
for volume in volumes:
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
req.cache_db_volumes(volumes.objects)
@ -307,7 +309,7 @@ class VolumeController(wsgi.Controller):
volume.update(update_dict)
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
volume_utils.notify_about_volume_usage(context, volume,
'update.end')

View File

@ -16,6 +16,7 @@ from oslo_log import log as logging
from six.moves import http_client
import webob
from cinder.api import api_utils
from cinder.api import common
from cinder.api import microversions as mv
from cinder.api.openstack import wsgi
@ -27,7 +28,6 @@ from cinder.i18n import _
from cinder import objects
from cinder.objects import fields
from cinder.policies import attachments as attachment_policy
from cinder import utils
from cinder.volume import api as volume_api
from cinder.volume import utils as volume_utils
@ -70,8 +70,8 @@ class AttachmentsController(wsgi.Controller):
@common.process_general_filtering('attachment')
def _process_attachment_filtering(self, context=None, filters=None,
req_version=None):
utils.remove_invalid_filter_options(context, filters,
self.allowed_filters)
api_utils.remove_invalid_filter_options(context, filters,
self.allowed_filters)
def _items(self, req):
"""Return a list of attachments, transformed through view builder."""

View File

@ -20,6 +20,7 @@ from six.moves import http_client
import webob
from webob import exc
from cinder.api import api_utils
from cinder.api import common
from cinder.api import microversions as mv
from cinder.api.openstack import wsgi
@ -189,26 +190,6 @@ class GroupTypesController(wsgi.Controller):
return self._view_builder.show(req, grp_type)
def _parse_is_public(self, is_public):
"""Parse is_public into something usable.
* True: List public group types only
* False: List private group types only
* None: List both public and private group types
"""
if is_public is None:
# preserve default value of showing only public types
return True
elif utils.is_none_string(is_public):
return None
else:
try:
return strutils.bool_from_string(is_public, strict=True)
except ValueError:
msg = _('Invalid is_public filter [%s]') % is_public
raise exc.HTTPBadRequest(explanation=msg)
def _get_group_types(self, req):
"""Helper function that returns a list of type dicts."""
params = req.params.copy()
@ -218,14 +199,14 @@ class GroupTypesController(wsgi.Controller):
context = req.environ['cinder.context']
if context.is_admin:
# Only admin has query access to all group types
filters['is_public'] = self._parse_is_public(
filters['is_public'] = api_utils._parse_is_public(
req.params.get('is_public', None))
else:
filters['is_public'] = True
utils.remove_invalid_filter_options(context,
filters,
self._get_grp_type_filter_options()
)
api_utils.remove_invalid_filter_options(
context,
filters,
self._get_grp_type_filter_options())
limited_types = group_types.get_all_group_types(context,
filters=filters,
marker=marker,

View File

@ -19,6 +19,7 @@ import ast
from oslo_log import log as logging
from cinder.api import api_utils
from cinder.api import common
from cinder.api import microversions as mv
from cinder.api.openstack import wsgi
@ -66,8 +67,8 @@ class SnapshotsController(snapshots_v2.SnapshotsController):
# Filter out invalid options
allowed_search_options = self._get_snapshot_filter_options()
utils.remove_invalid_filter_options(context, filters,
allowed_search_options)
api_utils.remove_invalid_filter_options(context, filters,
allowed_search_options)
def _items(self, req, is_detail=True):
"""Returns a list of snapshots, transformed through view builder."""

View File

@ -20,6 +20,7 @@ from six.moves import http_client
import webob
from webob import exc
from cinder.api import api_utils
from cinder.api import common
from cinder.api.contrib import scheduler_hints
from cinder.api import microversions as mv
@ -88,7 +89,7 @@ class VolumeController(volumes_v2.VolumeController):
if req_version.matches(None, mv.BACKUP_UPDATE):
filters.pop('group_id', None)
utils.remove_invalid_filter_options(
api_utils.remove_invalid_filter_options(
context, filters,
self._get_volume_filter_options())
@ -135,7 +136,7 @@ class VolumeController(volumes_v2.VolumeController):
context, 'volume', filters)
for volume in volumes:
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
req.cache_db_volumes(volumes.objects)
@ -154,8 +155,10 @@ class VolumeController(volumes_v2.VolumeController):
context = req.environ['cinder.context']
filters = req.params.copy()
utils.remove_invalid_filter_options(context, filters,
self._get_volume_filter_options())
api_utils.remove_invalid_filter_options(
context,
filters,
self._get_volume_filter_options())
num_vols, sum_size, metadata = self.volume_api.get_volume_summary(
context, filters=filters)

View File

@ -29,6 +29,7 @@ from oslo_utils import uuidutils
import six
import webob.exc
from cinder.api import api_utils
from cinder import db
from cinder import exception
from cinder.i18n import _
@ -247,8 +248,8 @@ def _validate_quota_set(quota_set):
if key in NON_QUOTA_KEYS:
continue
utils.validate_integer(value, key, min_value=-1,
max_value=db.MAX_INT)
api_utils.validate_integer(value, key, min_value=-1,
max_value=db.MAX_INT)
if len(bad_keys) > 0:
msg = _("Bad key(s) in quota set: %s") % ", ".join(bad_keys)

View File

@ -23,6 +23,7 @@ from oslo_utils import strutils
from six.moves import http_client
import webob
from cinder.api import api_utils
from cinder.api import common
from cinder.api import extensions
from cinder.api import microversions as mv
@ -43,7 +44,6 @@ from cinder.tests.unit.api.v2 import test_volumes as v2_test_volumes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit.image import fake as fake_image
from cinder.tests.unit import utils as test_utils
from cinder import utils
from cinder.volume import api as volume_api
from cinder.volume import api as vol_get
@ -774,7 +774,7 @@ class VolumeApiTest(test.TestCase):
@ddt.data(mv.get_prior_version(mv.RESOURCE_FILTER), mv.RESOURCE_FILTER,
mv.LIKE_FILTER)
@mock.patch.object(volume_api.API, 'check_volume_filters', mock.Mock())
@mock.patch.object(utils, 'add_visible_admin_metadata', mock.Mock())
@mock.patch.object(api_utils, 'add_visible_admin_metadata', mock.Mock())
@mock.patch('cinder.api.common.reject_invalid_filters')
def test_list_volume_with_general_filter(self, version, mock_update):
req = fakes.HTTPRequest.blank('/v3/volumes', version=version)

View File

@ -28,6 +28,7 @@ from six.moves import range
import webob.exc
import cinder
from cinder.api import api_utils
from cinder import exception
from cinder import test
from cinder.tests.unit import fake_constants as fake
@ -820,7 +821,7 @@ class AddVisibleAdminMetadataTestCase(test.TestCase):
{"key": "readonly", "value": "existing"}]
volume = {'volume_admin_metadata': admin_metadata,
'volume_metadata': metadata}
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
self.assertEqual([{"key": "key", "value": "value"},
{"key": "readonly", "value": "visible"},
{"key": "attached_mode", "value": "visible"}],
@ -832,7 +833,7 @@ class AddVisibleAdminMetadataTestCase(test.TestCase):
metadata = {"key": "value", "readonly": "existing"}
volume = {'admin_metadata': admin_metadata,
'metadata': metadata}
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
self.assertEqual({'key': 'value',
'attached_mode': 'visible',
'readonly': 'visible'},
@ -846,7 +847,7 @@ class AddVisibleAdminMetadataTestCase(test.TestCase):
metadata = [{"key": "key", "value": "value"}]
volume = {'volume_admin_metadata': admin_metadata,
'volume_metadata': metadata}
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
self.assertEqual([{"key": "key", "value": "value"}],
volume['volume_metadata'])
@ -856,7 +857,7 @@ class AddVisibleAdminMetadataTestCase(test.TestCase):
metadata = {"key": "value"}
volume = {'admin_metadata': admin_metadata,
'metadata': metadata}
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
self.assertEqual({'key': 'value'}, volume['metadata'])
def test_add_visible_admin_metadata_no_existing_metadata(self):
@ -864,7 +865,7 @@ class AddVisibleAdminMetadataTestCase(test.TestCase):
{"key": "readonly", "value": "visible"},
{"key": "attached_mode", "value": "visible"}]
volume = {'volume_admin_metadata': admin_metadata}
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
self.assertEqual({'attached_mode': 'visible', 'readonly': 'visible'},
volume['metadata'])
@ -872,7 +873,7 @@ class AddVisibleAdminMetadataTestCase(test.TestCase):
"readonly": "visible",
"attached_mode": "visible"}
volume = {'admin_metadata': admin_metadata}
utils.add_visible_admin_metadata(volume)
api_utils.add_visible_admin_metadata(volume)
self.assertEqual({'attached_mode': 'visible', 'readonly': 'visible'},
volume['metadata'])
@ -887,8 +888,8 @@ class InvalidFilterTestCase(test.TestCase):
allowed_search_options = ('allowed1', 'allowed2')
allowed_orig = ('allowed1', 'allowed2')
utils.remove_invalid_filter_options(ctxt, filters,
allowed_search_options)
api_utils.remove_invalid_filter_options(ctxt, filters,
allowed_search_options)
self.assertEqual(allowed_orig, allowed_search_options)
self.assertEqual(fltrs_orig, filters)
@ -902,8 +903,8 @@ class InvalidFilterTestCase(test.TestCase):
allowed_search_options = ('allowed1', 'allowed2')
allowed_orig = ('allowed1', 'allowed2')
utils.remove_invalid_filter_options(ctxt, filters,
allowed_search_options)
api_utils.remove_invalid_filter_options(ctxt, filters,
allowed_search_options)
self.assertEqual(allowed_orig, allowed_search_options)
self.assertNotEqual(fltrs_orig, filters)
@ -1483,7 +1484,7 @@ class TestValidateInteger(test.TestCase):
)
def test_validate_integer_raise_assert(self, value):
self.assertRaises(webob.exc.HTTPBadRequest,
utils.validate_integer,
api_utils.validate_integer,
value, 'limit', min_value=-1, max_value=(2 ** 31))
@ddt.data(
@ -1492,8 +1493,8 @@ class TestValidateInteger(test.TestCase):
u"123" # integer in unicode format
)
def test_validate_integer(self, value):
res = utils.validate_integer(value, 'limit', min_value=-1,
max_value=(2 ** 31))
res = api_utils.validate_integer(value, 'limit', min_value=-1,
max_value=(2 ** 31))
self.assertEqual(123, res)

View File

@ -53,7 +53,6 @@ from oslo_utils import strutils
from oslo_utils import timeutils
import retrying
import six
import webob.exc
from cinder import exception
from cinder.i18n import _
@ -292,14 +291,6 @@ def time_format(at=None):
return date_string
def is_none_string(val):
"""Check if a string represents a None value."""
if not isinstance(val, six.string_types):
return False
return val.lower() == 'none'
def monkey_patch():
"""Patches decorators for all functions in a specified module.
@ -673,72 +664,6 @@ def check_string_length(value, name, min_length=0, max_length=None,
raise exception.InvalidInput(reason=msg)
_visible_admin_metadata_keys = ['readonly', 'attached_mode']
def add_visible_admin_metadata(volume):
"""Add user-visible admin metadata to regular metadata.
Extracts the admin metadata keys that are to be made visible to
non-administrators, and adds them to the regular metadata structure for the
passed-in volume.
"""
visible_admin_meta = {}
if volume.get('volume_admin_metadata'):
if isinstance(volume['volume_admin_metadata'], dict):
volume_admin_metadata = volume['volume_admin_metadata']
for key in volume_admin_metadata:
if key in _visible_admin_metadata_keys:
visible_admin_meta[key] = volume_admin_metadata[key]
else:
for item in volume['volume_admin_metadata']:
if item['key'] in _visible_admin_metadata_keys:
visible_admin_meta[item['key']] = item['value']
# avoid circular ref when volume is a Volume instance
elif (volume.get('admin_metadata') and
isinstance(volume.get('admin_metadata'), dict)):
for key in _visible_admin_metadata_keys:
if key in volume['admin_metadata'].keys():
visible_admin_meta[key] = volume['admin_metadata'][key]
if not visible_admin_meta:
return
# NOTE(zhiyan): update visible administration metadata to
# volume metadata, administration metadata will rewrite existing key.
if volume.get('volume_metadata'):
orig_meta = list(volume.get('volume_metadata'))
for item in orig_meta:
if item['key'] in visible_admin_meta.keys():
item['value'] = visible_admin_meta.pop(item['key'])
for key, value in visible_admin_meta.items():
orig_meta.append({'key': key, 'value': value})
volume['volume_metadata'] = orig_meta
# avoid circular ref when vol is a Volume instance
elif (volume.get('metadata') and
isinstance(volume.get('metadata'), dict)):
volume['metadata'].update(visible_admin_meta)
else:
volume['metadata'] = visible_admin_meta
def remove_invalid_filter_options(context, filters,
allowed_search_options):
"""Remove search options that are not valid for non-admin API/context."""
if context.is_admin:
# Allow all options
return
# Otherwise, strip out all unknown options
unknown_options = [opt for opt in filters
if opt not in allowed_search_options]
bad_options = ", ".join(unknown_options)
LOG.debug("Removing options '%s' from query.", bad_options)
for opt in unknown_options:
del filters[opt]
def is_blk_device(dev):
try:
if stat.S_ISBLK(os.stat(dev).st_mode):
@ -1132,22 +1057,6 @@ def calculate_max_over_subscription_ratio(capability,
return max_over_subscription_ratio
def validate_integer(value, name, min_value=None, max_value=None):
"""Make sure that value is a valid integer, potentially within range.
:param value: the value of the integer
:param name: the name of the integer
:param min_length: the min_length of the integer
:param max_length: the max_length of the integer
:returns: integer
"""
try:
value = strutils.validate_integer(value, name, min_value, max_value)
return value
except ValueError as e:
raise webob.exc.HTTPBadRequest(explanation=six.text_type(e))
def validate_dictionary_string_length(specs):
"""Check the length of each key and value of dictionary."""
if not isinstance(specs, dict):