return items[start_index:range_end]
+def get_sort_params(params, default_key='created_at', default_dir='desc'):
+ """Retrieves sort keys/directions parameters.
+
+ Processes the parameters to create a list of sort keys and sort directions
+ that correspond to either the 'sort' parameter or the 'sort_key' and
+ 'sort_dir' parameter values. The value of the 'sort' parameter is a comma-
+ separated list of sort keys, each key is optionally appended with
+ ':<sort_direction>'.
+
+ Note that the 'sort_key' and 'sort_dir' parameters are deprecated in kilo
+ and an exception is raised if they are supplied with the 'sort' parameter.
+
+ The sort parameters are removed from the request parameters by this
+ function.
+
+ :param params: webob.multidict of request parameters (from
+ cinder.api.openstack.wsgi.Request.params)
+ :param default_key: default sort key value, added to the list if no
+ sort keys are supplied
+ :param default_dir: default sort dir value, added to the list if the
+ corresponding key does not have a direction
+ specified
+ :returns: list of sort keys, list of sort dirs
+ :raise webob.exc.HTTPBadRequest: If both 'sort' and either 'sort_key' or
+ 'sort_dir' are supplied parameters
+ """
+ if 'sort' in params and ('sort_key' in params or 'sort_dir' in params):
+ msg = _("The 'sort_key' and 'sort_dir' parameters are deprecated and "
+ "cannot be used with the 'sort' parameter.")
+ raise webob.exc.HTTPBadRequest(explanation=msg)
+ sort_keys = []
+ sort_dirs = []
+ if 'sort' in params:
+ for sort in params.pop('sort').strip().split(','):
+ sort_key, _sep, sort_dir = sort.partition(':')
+ if not sort_dir:
+ sort_dir = default_dir
+ sort_keys.append(sort_key.strip())
+ sort_dirs.append(sort_dir.strip())
+ else:
+ sort_key = params.pop('sort_key', default_key)
+ sort_dir = params.pop('sort_dir', default_dir)
+ sort_keys.append(sort_key.strip())
+ sort_dirs.append(sort_dir.strip())
+ return sort_keys, sort_dirs
+
+
def remove_version_from_href(href):
"""Removes the first api version from the href.
return query
+def process_sort_params(sort_keys, sort_dirs, default_keys=None,
+ default_dir='asc'):
+ """Process the sort parameters to include default keys.
+
+ Creates a list of sort keys and a list of sort directions. Adds the default
+ keys to the end of the list if they are not already included.
+
+ When adding the default keys to the sort keys list, the associated
+ direction is:
+ 1) The first element in the 'sort_dirs' list (if specified), else
+ 2) 'default_dir' value (Note that 'asc' is the default value since this is
+ the default in sqlalchemy.utils.paginate_query)
+
+ :param sort_keys: List of sort keys to include in the processed list
+ :param sort_dirs: List of sort directions to include in the processed list
+ :param default_keys: List of sort keys that need to be included in the
+ processed list, they are added at the end of the list
+ if not already specified.
+ :param default_dir: Sort direction associated with each of the default
+ keys that are not supplied, used when they are added
+ to the processed list
+ :returns: list of sort keys, list of sort directions
+ :raise exception.InvalidInput: If more sort directions than sort keys
+ are specified or if an invalid sort
+ direction is specified
+ """
+ if default_keys is None:
+ default_keys = ['created_at', 'id']
+
+ # Determine direction to use for when adding default keys
+ if sort_dirs and len(sort_dirs):
+ default_dir_value = sort_dirs[0]
+ else:
+ default_dir_value = default_dir
+
+ # Create list of keys (do not modify the input list)
+ if sort_keys:
+ result_keys = list(sort_keys)
+ else:
+ result_keys = []
+
+ # If a list of directions is not provided, use the default sort direction
+ # for all provided keys.
+ if sort_dirs:
+ result_dirs = []
+ # Verify sort direction
+ for sort_dir in sort_dirs:
+ if sort_dir not in ('asc', 'desc'):
+ msg = _LE("Unknown sort direction, must be 'desc' or 'asc'.")
+ raise exception.InvalidInput(reason=msg)
+ result_dirs.append(sort_dir)
+ else:
+ result_dirs = [default_dir_value for _sort_key in result_keys]
+
+ # Ensure that the key and direction length match
+ while len(result_dirs) < len(result_keys):
+ result_dirs.append(default_dir_value)
+ # Unless more direction are specified, which is an error
+ if len(result_dirs) > len(result_keys):
+ msg = _LE("Sort direction array size exceeds sort key array size.")
+ raise exception.InvalidInput(reason=msg)
+
+ # Ensure defaults are included
+ for key in default_keys:
+ if key not in result_keys:
+ result_keys.append(key)
+ result_dirs.append(default_dir_value)
+
+ return result_keys, result_dirs
+
+
@require_admin_context
def volume_get_iscsi_target_num(context, volume_id):
result = model_query(context, models.IscsiTarget, read_deleted="yes").\
{'marker': marker, 'limit': 20})
+class SortParamUtilsTest(test.TestCase):
+
+ def test_get_sort_params_defaults(self):
+ '''Verifies the default sort key and direction.'''
+ sort_keys, sort_dirs = common.get_sort_params({})
+ self.assertEqual(['created_at'], sort_keys)
+ self.assertEqual(['desc'], sort_dirs)
+
+ def test_get_sort_params_override_defaults(self):
+ '''Verifies that the defaults can be overriden.'''
+ sort_keys, sort_dirs = common.get_sort_params({}, default_key='key1',
+ default_dir='dir1')
+ self.assertEqual(['key1'], sort_keys)
+ self.assertEqual(['dir1'], sort_dirs)
+
+ def test_get_sort_params_single_value_sort_param(self):
+ '''Verifies a single sort key and direction.'''
+ params = {'sort': 'key1:dir1'}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1'], sort_keys)
+ self.assertEqual(['dir1'], sort_dirs)
+
+ def test_get_sort_params_single_value_old_params(self):
+ '''Verifies a single sort key and direction.'''
+ params = {'sort_key': 'key1', 'sort_dir': 'dir1'}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1'], sort_keys)
+ self.assertEqual(['dir1'], sort_dirs)
+
+ def test_get_sort_params_single_with_default_sort_param(self):
+ '''Verifies a single sort value with a default direction.'''
+ params = {'sort': 'key1'}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1'], sort_keys)
+ # Direction should be defaulted
+ self.assertEqual(['desc'], sort_dirs)
+
+ def test_get_sort_params_single_with_default_old_params(self):
+ '''Verifies a single sort value with a default direction.'''
+ params = {'sort_key': 'key1'}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1'], sort_keys)
+ # Direction should be defaulted
+ self.assertEqual(['desc'], sort_dirs)
+
+ def test_get_sort_params_multiple_values(self):
+ '''Verifies multiple sort parameter values.'''
+ params = {'sort': 'key1:dir1,key2:dir2,key3:dir3'}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
+ self.assertEqual(['dir1', 'dir2', 'dir3'], sort_dirs)
+
+ def test_get_sort_params_multiple_not_all_dirs(self):
+ '''Verifies multiple sort keys without all directions.'''
+ params = {'sort': 'key1:dir1,key2,key3:dir3'}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
+ # Second key is missing the direction, should be defaulted
+ self.assertEqual(['dir1', 'desc', 'dir3'], sort_dirs)
+
+ def test_get_sort_params_multiple_override_default_dir(self):
+ '''Verifies multiple sort keys and overriding default direction.'''
+ params = {'sort': 'key1:dir1,key2,key3'}
+ sort_keys, sort_dirs = common.get_sort_params(params,
+ default_dir='foo')
+ self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
+ self.assertEqual(['dir1', 'foo', 'foo'], sort_dirs)
+
+ def test_get_sort_params_params_modified(self):
+ '''Verifies that the input sort parameter are modified.'''
+ params = {'sort': 'key1:dir1,key2:dir2,key3:dir3'}
+ common.get_sort_params(params)
+ self.assertEqual({}, params)
+
+ params = {'sort_dir': 'key1', 'sort_dir': 'dir1'}
+ common.get_sort_params(params)
+ self.assertEqual({}, params)
+
+ def test_get_sort_params_random_spaces(self):
+ '''Verifies that leading and trailing spaces are removed.'''
+ params = {'sort': ' key1 : dir1,key2: dir2 , key3 '}
+ sort_keys, sort_dirs = common.get_sort_params(params)
+ self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
+ self.assertEqual(['dir1', 'dir2', 'desc'], sort_dirs)
+
+ def test_get_params_mix_sort_and_old_params(self):
+ '''An exception is raised if both types of sorting params are given.'''
+ for params in ({'sort': 'k1', 'sort_key': 'k1'},
+ {'sort': 'k1', 'sort_dir': 'd1'},
+ {'sort': 'k1', 'sort_key': 'k1', 'sort_dir': 'd2'}):
+ self.assertRaises(webob.exc.HTTPBadRequest,
+ common.get_sort_params,
+ params)
+
+
class MiscFunctionsTest(test.TestCase):
def test_remove_major_version_from_href(self):
from cinder import context
from cinder import db
+from cinder.db.sqlalchemy import api as sqlalchemy_api
from cinder import exception
from cinder.quota import ReservableResource
from cinder import test
def test_backup_not_found(self):
self.assertRaises(exception.BackupNotFound, db.backup_get, self.ctxt,
'notinbase')
+
+
+class DBAPIProcessSortParamTestCase(test.TestCase):
+
+ def test_process_sort_params_defaults(self):
+ '''Verifies default sort parameters.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params([], [])
+ self.assertEqual(['created_at', 'id'], sort_keys)
+ self.assertEqual(['asc', 'asc'], sort_dirs)
+
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(None, None)
+ self.assertEqual(['created_at', 'id'], sort_keys)
+ self.assertEqual(['asc', 'asc'], sort_dirs)
+
+ def test_process_sort_params_override_default_keys(self):
+ '''Verifies that the default keys can be overridden.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ [], [], default_keys=['key1', 'key2', 'key3'])
+ self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
+ self.assertEqual(['asc', 'asc', 'asc'], sort_dirs)
+
+ def test_process_sort_params_override_default_dir(self):
+ '''Verifies that the default direction can be overridden.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ [], [], default_dir='dir1')
+ self.assertEqual(['created_at', 'id'], sort_keys)
+ self.assertEqual(['dir1', 'dir1'], sort_dirs)
+
+ def test_process_sort_params_override_default_key_and_dir(self):
+ '''Verifies that the default key and dir can be overridden.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ [], [], default_keys=['key1', 'key2', 'key3'],
+ default_dir='dir1')
+ self.assertEqual(['key1', 'key2', 'key3'], sort_keys)
+ self.assertEqual(['dir1', 'dir1', 'dir1'], sort_dirs)
+
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ [], [], default_keys=[], default_dir='dir1')
+ self.assertEqual([], sort_keys)
+ self.assertEqual([], sort_dirs)
+
+ def test_process_sort_params_non_default(self):
+ '''Verifies that non-default keys are added correctly.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['key1', 'key2'], ['asc', 'desc'])
+ self.assertEqual(['key1', 'key2', 'created_at', 'id'], sort_keys)
+ # First sort_dir in list is used when adding the default keys
+ self.assertEqual(['asc', 'desc', 'asc', 'asc'], sort_dirs)
+
+ def test_process_sort_params_default(self):
+ '''Verifies that default keys are added correctly.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2'], ['asc', 'desc'])
+ self.assertEqual(['id', 'key2', 'created_at'], sort_keys)
+ self.assertEqual(['asc', 'desc', 'asc'], sort_dirs)
+
+ # Include default key value, rely on default direction
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2'], [])
+ self.assertEqual(['id', 'key2', 'created_at'], sort_keys)
+ self.assertEqual(['asc', 'asc', 'asc'], sort_dirs)
+
+ def test_process_sort_params_default_dir(self):
+ '''Verifies that the default dir is applied to all keys.'''
+ # Direction is set, ignore default dir
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2'], ['desc'], default_dir='dir')
+ self.assertEqual(['id', 'key2', 'created_at'], sort_keys)
+ self.assertEqual(['desc', 'desc', 'desc'], sort_dirs)
+
+ # But should be used if no direction is set
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2'], [], default_dir='dir')
+ self.assertEqual(['id', 'key2', 'created_at'], sort_keys)
+ self.assertEqual(['dir', 'dir', 'dir'], sort_dirs)
+
+ def test_process_sort_params_unequal_length(self):
+ '''Verifies that a sort direction list is applied correctly.'''
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2', 'key3'], ['desc'])
+ self.assertEqual(['id', 'key2', 'key3', 'created_at'], sort_keys)
+ self.assertEqual(['desc', 'desc', 'desc', 'desc'], sort_dirs)
+
+ # Default direction is the first key in the list
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2', 'key3'], ['desc', 'asc'])
+ self.assertEqual(['id', 'key2', 'key3', 'created_at'], sort_keys)
+ self.assertEqual(['desc', 'asc', 'desc', 'desc'], sort_dirs)
+
+ sort_keys, sort_dirs = sqlalchemy_api.process_sort_params(
+ ['id', 'key2', 'key3'], ['desc', 'asc', 'asc'])
+ self.assertEqual(['id', 'key2', 'key3', 'created_at'], sort_keys)
+ self.assertEqual(['desc', 'asc', 'asc', 'desc'], sort_dirs)
+
+ def test_process_sort_params_extra_dirs_lengths(self):
+ '''InvalidInput raised if more directions are given.'''
+ self.assertRaises(exception.InvalidInput,
+ sqlalchemy_api.process_sort_params,
+ ['key1', 'key2'],
+ ['asc', 'desc', 'desc'])
+
+ def test_process_sort_params_invalid_sort_dir(self):
+ '''InvalidInput raised if invalid directions are given.'''
+ for dirs in [['foo'], ['asc', 'foo'], ['asc', 'desc', 'foo']]:
+ self.assertRaises(exception.InvalidInput,
+ sqlalchemy_api.process_sort_params,
+ ['key'],
+ dirs)
Code Review / openstack-build / cinder-build.git / commitdiff