import six
import webob
+from cinder.api import common
from cinder.api import extensions
from cinder.api.openstack import wsgi
from cinder.api.views import qos_specs as view_qos_specs
"""Returns the list of qos_specs."""
context = req.environ['cinder.context']
authorize(context)
- specs = qos_specs.get_all_specs(context)
+
+ params = req.params.copy()
+
+ marker, limit, offset = common.get_pagination_params(params)
+ 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)
+
+ specs = qos_specs.get_all_specs(context, filters=filters,
+ marker=marker, limit=limit,
+ offset=offset, sort_keys=sort_keys,
+ sort_dirs=sort_dirs)
return self._view_builder.summary_list(req, specs)
@wsgi.serializers(xml=QoSSpecsTemplate)
class ViewBuilder(common.ViewBuilder):
"""Model QoS specs API responses as a python dictionary."""
- _collection_name = "qos_specs"
+ _collection_name = "qos-specs"
def __init__(self):
"""Initialize view builder."""
super(ViewBuilder, self).__init__()
- def summary_list(self, request, qos_specs):
+ def summary_list(self, request, qos_specs, qos_count=None):
"""Show a list of qos_specs without many details."""
- return self._list_view(self.detail, request, qos_specs)
+ return self._list_view(self.detail, request, qos_specs, qos_count)
def summary(self, request, qos_spec):
"""Generic, non-detailed view of a qos_specs."""
'qos_associations': associates
}
- def _list_view(self, func, request, qos_specs):
+ def _list_view(self, func, request, qos_specs, qos_count=None):
"""Provide a view for a list of qos_specs."""
specs_list = [func(request, specs)['qos_specs'] for specs in qos_specs]
+ specs_links = self._get_collection_links(request, qos_specs,
+ self._collection_name,
+ qos_count)
specs_dict = dict(qos_specs=specs_list)
+ if specs_links:
+ specs_dict['qos_specs_links'] = specs_links
return specs_dict
return IMPL.qos_specs_get(context, qos_specs_id)
-def qos_specs_get_all(context, inactive=False, filters=None):
+def qos_specs_get_all(context, filters=None, marker=None, limit=None,
+ offset=None, sort_keys=None, sort_dirs=None):
"""Get all qos_specs."""
- return IMPL.qos_specs_get_all(context, inactive, filters)
+ return IMPL.qos_specs_get_all(context, filters=filters, marker=marker,
+ limit=limit, offset=offset,
+ sort_keys=sort_keys, sort_dirs=sort_dirs)
def qos_specs_get_by_name(context, name):
if query is None:
return None
- marker_volume = None
+ marker_object = None
if marker is not None:
- marker_volume = get(context, marker, session)
+ marker_object = get(context, marker, session)
return sqlalchemyutils.paginate_query(query, paginate_type, limit,
sort_keys,
- marker=marker_volume,
+ marker=marker_object,
sort_dirs=sort_dirs,
offset=offset)
@require_admin_context
-def qos_specs_get_all(context, inactive=False, filters=None):
+def qos_specs_get_all(context, filters=None, marker=None, limit=None,
+ offset=None, sort_keys=None, sort_dirs=None):
"""Returns a list of all qos_specs.
Results is like:
},
]
"""
- filters = filters or {}
- # TODO(zhiteng) Add filters for 'consumer'
+ session = get_session()
+ with session.begin():
+ # Generate the query
+ query = _generate_paginate_query(context, session, marker, limit,
+ sort_keys, sort_dirs, filters,
+ offset, models.QualityOfServiceSpecs)
+ # No Qos specs would match, return empty list
+ if query is None:
+ return []
+ rows = query.all()
+ return _dict_with_qos_specs(rows)
- read_deleted = "yes" if inactive else "no"
+
+@require_admin_context
+def _qos_specs_get_query(context, session):
rows = model_query(context, models.QualityOfServiceSpecs,
- read_deleted=read_deleted). \
- options(joinedload_all('specs')).all()
+ session=session,
+ read_deleted='no').\
+ options(joinedload_all('specs')).filter_by(key='QoS_Specs_Name')
+ return rows
+
- return _dict_with_qos_specs(rows)
+def _process_qos_specs_filters(query, filters):
+ if filters:
+ # Ensure that filters' keys exist on the model
+ if not is_valid_model_filters(models.QualityOfServiceSpecs, filters):
+ return
+ query = query.filter_by(**filters)
+ return query
+
+
+@require_admin_context
+def _qos_specs_get(context, qos_spec_id, session=None):
+ result = model_query(context, models.QualityOfServiceSpecs,
+ session=session,
+ read_deleted='no').\
+ filter_by(id=qos_spec_id).filter_by(key='QoS_Specs_Name').first()
+
+ if not result:
+ raise exception.QoSSpecsNotFound(specs_id=qos_spec_id)
+
+ return result
@require_admin_context
PAGINATION_HELPERS = {
models.Volume: (_volume_get_query, _process_volume_filters, _volume_get),
models.Snapshot: (_snaps_get_query, _process_snaps_filters, _snapshot_get),
- models.Backup: (_backups_get_query, _process_backups_filters, _backup_get)
+ models.Backup: (_backups_get_query, _process_backups_filters, _backup_get),
+ models.QualityOfServiceSpecs: (_qos_specs_get_query,
+ _process_qos_specs_filters, _qos_specs_get)
}
from cinder.api.contrib import qos_specs_manage
from cinder.api import xmlutil
+from cinder import context
+from cinder import db
from cinder import exception
from cinder import test
from cinder.tests.unit.api import fakes
'id': 'FakeVolTypeID'}]
-def return_qos_specs_get_all(context):
+def return_qos_specs_get_all(context, filters=None, marker=None, limit=None,
+ offset=None, sort_keys=None, sort_dirs=None):
return [
stub_qos_specs(1),
stub_qos_specs(2),
class QoSSpecManageApiTest(test.TestCase):
+
+ def _create_qos_specs(self, name, values=None):
+ """Create a transfer object."""
+ if values:
+ specs = dict(name=name, qos_specs=values)
+ else:
+ specs = {'name': name,
+ 'qos_specs': {
+ 'consumer': 'back-end',
+ 'key1': 'value1',
+ 'key2': 'value2'}}
+ return db.qos_specs_create(self.ctxt, specs)['id']
+
def setUp(self):
super(QoSSpecManageApiTest, self).setUp()
self.flags(host='fake')
self.controller = qos_specs_manage.QoSSpecsController()
+ self.ctxt = context.RequestContext(user_id='user_id',
+ project_id='project_id',
+ is_admin=True)
+ self.qos_id1 = self._create_qos_specs("Qos_test_1")
+ self.qos_id2 = self._create_qos_specs("Qos_test_2")
+ self.qos_id3 = self._create_qos_specs("Qos_test_3")
+ self.qos_id4 = self._create_qos_specs("Qos_test_4")
@mock.patch('cinder.volume.qos_specs.get_all_specs',
side_effect=return_qos_specs_get_all)
expected_names = ['qos_specs_1', 'qos_specs_2', 'qos_specs_3']
self.assertEqual(set(expected_names), names)
+ def test_index_with_limit(self):
+ url = '/v2/fake/qos-specs?limit=2'
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ res = self.controller.index(req)
+
+ self.assertEqual(2, len(res['qos_specs']))
+ self.assertEqual(self.qos_id4, res['qos_specs'][0]['id'])
+ self.assertEqual(self.qos_id3, res['qos_specs'][1]['id'])
+
+ expect_next_link = ('http://localhost/v2/fakeproject/qos-specs?limit'
+ '=2&marker=%s') % res['qos_specs'][1]['id']
+ self.assertEqual(expect_next_link, res['qos_specs_links'][0]['href'])
+
+ def test_index_with_offset(self):
+ url = '/v2/fake/qos-specs?offset=1'
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ res = self.controller.index(req)
+
+ self.assertEqual(3, len(res['qos_specs']))
+
+ 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)
+ res = self.controller.index(req)
+
+ self.assertEqual(2, len(res['qos_specs']))
+ self.assertEqual(self.qos_id3, res['qos_specs'][0]['id'])
+ self.assertEqual(self.qos_id2, res['qos_specs'][1]['id'])
+
+ def test_index_with_marker(self):
+ url = '/v2/fake/qos-specs?marker=%s' % self.qos_id4
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ res = self.controller.index(req)
+
+ self.assertEqual(3, len(res['qos_specs']))
+
+ def test_index_with_filter(self):
+ url = '/v2/fake/qos-specs?id=%s' % self.qos_id4
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ res = self.controller.index(req)
+
+ self.assertEqual(1, len(res['qos_specs']))
+ self.assertEqual(self.qos_id4, res['qos_specs'][0]['id'])
+
+ def test_index_with_sort_keys(self):
+ url = '/v2/fake/qos-specs?sort=id'
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ res = self.controller.index(req)
+ self.assertEqual(4, len(res['qos_specs']))
+ expect_result = [self.qos_id1, self.qos_id2,
+ self.qos_id3, self.qos_id4]
+ expect_result.sort(reverse=True)
+
+ self.assertEqual(expect_result[0], res['qos_specs'][0]['id'])
+ self.assertEqual(expect_result[1], res['qos_specs'][1]['id'])
+ self.assertEqual(expect_result[2], res['qos_specs'][2]['id'])
+ self.assertEqual(expect_result[3], res['qos_specs'][3]['id'])
+
+ def test_index_with_sort_keys_and_sort_dirs(self):
+ url = '/v2/fake/qos-specs?sort=id:asc'
+ req = fakes.HTTPRequest.blank(url, use_admin_context=True)
+ res = self.controller.index(req)
+ self.assertEqual(4, len(res['qos_specs']))
+ expect_result = [self.qos_id1, self.qos_id2,
+ self.qos_id3, self.qos_id4]
+ expect_result.sort()
+
+ self.assertEqual(expect_result[0], res['qos_specs'][0]['id'])
+ self.assertEqual(expect_result[1], res['qos_specs'][1]['id'])
+ self.assertEqual(expect_result[2], res['qos_specs'][2]['id'])
+ self.assertEqual(expect_result[3], res['qos_specs'][3]['id'])
+
@mock.patch('cinder.volume.qos_specs.get_qos_specs',
side_effect=return_qos_specs_get_qos_specs)
@mock.patch('cinder.volume.qos_specs.delete',
type_id=None)
-def get_all_specs(context, inactive=False, search_opts=None):
- """Get all non-deleted qos specs.
-
- Pass inactive=True as argument and deleted volume types would return
- as well.
- """
- search_opts = search_opts or {}
- qos_specs = db.qos_specs_get_all(context, inactive)
-
- if search_opts:
- LOG.debug("Searching by: %s", search_opts)
-
- def _check_specs_match(qos_specs, searchdict):
- for k, v in searchdict.items():
- if ((k not in qos_specs['specs'].keys() or
- qos_specs['specs'][k] != v)):
- return False
- return True
-
- # search_option to filter_name mapping.
- filter_mapping = {'qos_specs': _check_specs_match}
-
- result = {}
- for name, args in qos_specs.items():
- # go over all filters in the list
- for opt, values in search_opts.items():
- try:
- filter_func = filter_mapping[opt]
- except KeyError:
- # no such filter - ignore it, go to next filter
- continue
- else:
- if filter_func(args, values):
- result[name] = args
- break
- qos_specs = result
+def get_all_specs(context, filters=None, marker=None, limit=None, offset=None,
+ sort_keys=None, sort_dirs=None):
+ """Get all non-deleted qos specs."""
+ qos_specs = db.qos_specs_get_all(context, filters=filters, marker=marker,
+ limit=limit, offset=offset,
+ sort_keys=sort_keys, sort_dirs=sort_dirs)
return qos_specs
Code Review / openstack-build / cinder-build.git / commitdiff