from cinder.api.openstack import wsgi
from cinder.api import xmlutil
+import cinder.db
from cinder import exception
from cinder.i18n import _
import cinder.policy
def _get_offset_param(params):
"""Extract offset id from request's dictionary (defaults to 0) or fail."""
- try:
- offset = int(params.pop('offset', 0))
- except ValueError:
- msg = _('offset param must be an integer')
- raise webob.exc.HTTPBadRequest(explanation=msg)
-
- if offset < 0:
- msg = _('offset param must be positive')
- raise webob.exc.HTTPBadRequest(explanation=msg)
-
- return offset
+ offset = params.pop('offset', 0)
+ return utils.validate_integer(offset, 'offset', 0, cinder.db.MAX_INT)
def limited(items, request, max_limit=None):
from cinder import exception
from cinder.i18n import _
from cinder import quota
+from cinder import utils
QUOTAS = quota.QUOTAS
for key, value in body['quota_class_set'].items():
if key in QUOTAS:
try:
- value = self.validate_integer(value, key, min_value=-1,
- max_value=db.MAX_INT)
+ value = utils.validate_integer(value, key, min_value=-1,
+ max_value=db.MAX_INT)
db.quota_class_update(context, quota_class, key, value)
except exception.QuotaClassNotFound:
db.quota_class_create(context, quota_class, key, value)
if key in NON_QUOTA_KEYS:
continue
- value = self.validate_integer(
+ value = utils.validate_integer(
body['quota_set'][key], key, min_value=-1,
max_value=db.MAX_INT)
from cinder import exception
from cinder.i18n import _
from cinder import rpc
+from cinder import utils
from cinder.volume import volume_types
authorize = extensions.extension_authorizer('volume',
def _check_encryption_input(self, encryption, create=True):
if encryption.get('key_size') is not None:
- encryption['key_size'] = self.validate_integer(
+ encryption['key_size'] = utils.validate_integer(
encryption['key_size'], 'key_size',
min_value=0, max_value=db.MAX_INT)
except exception.InvalidInput as error:
raise webob.exc.HTTPBadRequest(explanation=error.msg)
- @staticmethod
- 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 = int(value)
- except (TypeError, ValueError, UnicodeEncodeError):
- raise webob.exc.HTTPBadRequest(explanation=(
- _('%s must be an integer.') % name))
-
- if min_value is not None and value < min_value:
- raise webob.exc.HTTPBadRequest(
- explanation=(_('%(value_name)s must be >= %(min_value)d') %
- {'value_name': name, 'min_value': min_value}))
- if max_value is not None and value > max_value:
- raise webob.exc.HTTPBadRequest(
- explanation=(_('%(value_name)s must be <= %(max_value)d') %
- {'value_name': name, 'max_value': max_value}))
-
- return value
-
class Fault(webob.exc.HTTPException):
"""Wrap webob.exc.HTTPException to provide API friendly response."""
db.backup_destroy(context.get_admin_context(), backup_id2)
db.backup_destroy(context.get_admin_context(), backup_id1)
+ def test_list_backups_with_offset_out_of_range(self):
+ url = '/v2/fake/backups?offset=252452434242342434'
+ req = webob.Request.blank(url)
+ req.method = 'GET'
+ req.headers['Content-Type'] = 'application/json'
+ res = req.get_response(fakes.wsgi_app())
+ self.assertEqual(400, res.status_int)
+
def test_list_backups_with_marker(self):
backup_id1 = self._create_backup()
backup_id2 = self._create_backup()
db.backup_destroy(context.get_admin_context(), backup_id2)
db.backup_destroy(context.get_admin_context(), backup_id1)
+ def test_list_backups_detail_with_offset_out_of_range(self):
+ url = '/v2/fake/backups/detail?offset=234534543657634523'
+ req = webob.Request.blank(url)
+ req.method = 'GET'
+ req.headers['Content-Type'] = 'application/json'
+ res = req.get_response(fakes.wsgi_app())
+ self.assertEqual(400, res.status_int)
+
@mock.patch('cinder.db.service_get_all_by_topic')
@mock.patch(
'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
consistencygroup2.destroy()
consistencygroup3.destroy()
+ @ddt.data(False, True)
+ def test_list_consistencygroups_with_offset_out_of_range(self, is_detail):
+ url = '/v2/fake/consistencygroups?offset=234523423455454'
+ if is_detail:
+ url = '/v2/fake/consistencygroups/detail?offset=234523423455454'
+ req = webob.Request.blank(url)
+ req.method = 'GET'
+ req.headers['Content-Type'] = 'application/json'
+ res = req.get_response(fakes.wsgi_app())
+ self.assertEqual(400, res.status_int)
+
@ddt.data(False, True)
def test_list_consistencygroups_with_limit_and_offset(self, is_detail):
consistencygroup1 = self._create_consistencygroup()
self.assertEqual(3, len(res['qos_specs']))
+ def test_index_with_offset_out_of_range(self):
+ url = '/v2/fake/qos-specs?offset=356576877698707'
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ self.assertRaises(webob.exc.HTTPBadRequest, self.controller.index,
+ req)
+
def test_index_with_limit_and_offset(self):
url = '/v2/fake/qos-specs?limit=2&offset=1'
req = fakes.HTTPRequest.blank(url, use_admin_context=True)
@mock.patch(
'cinder.api.openstack.wsgi.Controller.validate_string_length')
@mock.patch(
- 'cinder.api.openstack.wsgi.Controller.validate_integer')
+ 'cinder.utils.validate_integer')
def test_update_limit(self, mock_validate_integer, mock_validate):
mock_validate_integer.return_value = 10
self.assertDictMatch(body, result)
@mock.patch('cinder.api.openstack.wsgi.Controller.validate_string_length')
- @mock.patch('cinder.api.openstack.wsgi.Controller.validate_integer')
+ @mock.patch('cinder.utils.validate_integer')
def test_update_limit(self, mock_validate_integer, mock_validate):
mock_validate_integer.return_value = 5
volume_types.create(self.ctxt, 'fake_type')
from oslo_serialization import jsonutils
import webob
-from cinder.api.openstack import wsgi
from cinder import context
from cinder import db
from cinder import test
from cinder.tests.unit.api import fakes
+from cinder import utils
def return_volume_type_encryption(context, volume_type_id):
db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
def test_create_json(self):
- with mock.patch.object(wsgi.Controller,
+ with mock.patch.object(utils,
'validate_integer') as mock_validate_integer:
mock_validate_integer.return_value = 128
self._create('fake_cipher', 'front-end', 128, 'fake_encryptor')
self.assertEqual(expected, jsonutils.loads(res.body))
db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
- @mock.patch('cinder.api.openstack.wsgi.Controller.validate_integer')
+ @mock.patch('cinder.utils.validate_integer')
def test_update_item(self, mock_validate_integer):
mock_validate_integer.return_value = 512
volume_type = self._default_volume_type
body = {'name': 'a' * 255 + " "}
self.controller.validate_name_and_description(body)
self.assertEqual('a' * 255, body['name'])
-
- def test_validate_integer_greater_than_max_int_limit(self):
- value = (2 ** 31) + 1
- self.assertRaises(webob.exc.HTTPBadRequest,
- self.controller.validate_integer,
- value, 'limit', min_value=-1, max_value=(2 ** 31))
-
- def test_validate_integer_less_than_min_int_limit(self):
- value = -12
- self.assertRaises(webob.exc.HTTPBadRequest,
- self.controller.validate_integer,
- value, 'limit', min_value=-1, max_value=(2 ** 31))
-
- def test_validate_integer_invalid_limit(self):
- value = "should_be_int"
- self.assertRaises(webob.exc.HTTPBadRequest,
- self.controller.validate_integer,
- value, 'limit', min_value=-1, max_value=(2 ** 31))
self.assertRaises(
webob.exc.HTTPBadRequest, common.limited, self.tiny, req)
+ def test_limiter_offset_out_of_range(self):
+ """Test offset key works with a offset out of range."""
+ req = webob.Request.blank('/?offset=123456789012346456')
+ self.assertRaises(
+ webob.exc.HTTPBadRequest, common.limited, self.tiny, req)
+
def test_limiter_offset_bad(self):
"""Test offset key works with a BAD offset."""
req = webob.Request.blank(u'/?offset=\u0020aa')
self.assertEqual(items[3:1003], common.limited(items, req))
req = webob.Request.blank('/?offset=3000&limit=10')
self.assertEqual([], common.limited(items, req))
+ req = webob.Request.blank('/?offset=30034522235674530&limit=10')
+ self.assertRaises(
+ webob.exc.HTTPBadRequest, common.limited, items, req)
def test_limiter_custom_max_limit(self):
"""Test a max_limit other than 1000."""
self.controller.index,
req)
+ # Test that we get an exception HTTPBadRequest(400) with an offset
+ # greater than the maximum offset value.
+ url = '/v2/snapshots?limit=1&offset=323245324356534235'
+ req = fakes.HTTPRequest.blank(url)
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ self.controller.index, req)
+
def _assert_list_next(self, expected_query=None, project='fakeproject',
**kwargs):
"""Check a page of snapshots list."""
self.assertEqual(2, len(res['volume_types']))
+ def test_volume_types_index_with_offset_out_of_range(self):
+ url = '/v2/fake/types?offset=424366766556787'
+ req = fakes.HTTPRequest.blank(url)
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ self.controller.index, req)
+
def test_volume_types_index_with_limit_and_offset(self):
req = fakes.HTTPRequest.blank('/v2/fake/types?limit=2&offset=1')
req.environ['cinder.context'] = self.ctxt
self.controller.index,
req)
+ # Test that we get an exception HTTPBadRequest(400) with an offset
+ # greater than the maximum offset value.
+ url = '/v2/volumes?limit=2&offset=43543564546567575'
+ req = fakes.HTTPRequest.blank(url)
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ self.controller.index,
+ req)
+
def test_volume_detail_with_marker(self):
def stub_volume_get_all_by_project(context, project_id, marker, limit,
sort_keys=None, sort_dirs=None,
self.controller.detail,
req)
+ url = '/v2/volumes/detail?limit=2&offset=4536546546546467'
+ req = fakes.HTTPRequest.blank(url)
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ self.controller.detail,
+ req)
+
def test_volume_with_limit_zero(self):
def stub_volume_get_all(context, marker, limit, **kwargs):
return []
from oslo_utils import timeutils
import six
from six.moves import range
+import webob.exc
import cinder
from cinder import exception
def test_compare(self):
self.assertEqual(NotImplemented,
self.one._compare(1, self.one._cmpkey))
+
+
+class TestValidateInteger(test.TestCase):
+
+ def test_validate_integer_greater_than_max_int_limit(self):
+ value = (2 ** 31) + 1
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ utils.validate_integer,
+ value, 'limit', min_value=-1, max_value=(2 ** 31))
+
+ def test_validate_integer_less_than_min_int_limit(self):
+ value = -12
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ utils.validate_integer,
+ value, 'limit', min_value=-1, max_value=(2 ** 31))
+
+ def test_validate_integer_invalid_limit(self):
+ value = "should_be_int"
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ utils.validate_integer,
+ value, 'limit', min_value=-1, max_value=(2 ** 31))
from oslo_utils import timeutils
import retrying
import six
+import webob.exc
from cinder import exception
from cinder.i18n import _, _LE, _LW
# account the reserved space.
free = free_capacity - math.floor(total * reserved)
return free
+
+
+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 = int(value)
+ except (TypeError, ValueError, UnicodeEncodeError):
+ raise webob.exc.HTTPBadRequest(explanation=(
+ _('%s must be an integer.') % name))
+
+ if min_value is not None and value < min_value:
+ raise webob.exc.HTTPBadRequest(
+ explanation=(_('%(value_name)s must be >= %(min_value)d') %
+ {'value_name': name, 'min_value': min_value}))
+ if max_value is not None and value > max_value:
+ raise webob.exc.HTTPBadRequest(
+ explanation=(_('%(value_name)s must be <= %(max_value)d') %
+ {'value_name': name, 'max_value': max_value}))
+
+ return value
Code Review / openstack-build / cinder-build.git / commitdiff