Use utils.prepare_query_string instead of duplicated code

There are some duplicated code in nova client for generating
query string.

The 'prepare_query_string' method can convert dict params to
query string(it transforms the dict to a sequence of two-element
tuples in fixed order, then the encoded string will be consistent
in Python 2&3.)

This patch use utils.prepare_query_string instead of these and
plus some notes in the 'prepare_query_string' method.

Change-Id: Idb3c5e97f8bbcd5ec5446f776c10fa8c84b54d5d
Closes-Bug: 1727968
This commit is contained in:
Yikun Jiang 2017-10-27 15:28:15 +08:00
parent c9e7a64ca8
commit 50460bddfc
8 changed files with 47 additions and 40 deletions

View File

@ -247,7 +247,10 @@ class Manager(HookableMixin):
def api_version(self):
return self.api.api_version
def _list(self, url, response_key, obj_class=None, body=None):
def _list(self, url, response_key, obj_class=None, body=None,
filters=None):
if filters:
url = utils.get_url_with_filter(url, filters)
if body:
resp, body = self.api.client.post(url, body=body)
else:
@ -347,7 +350,9 @@ class Manager(HookableMixin):
if cache:
cache.write("%s\n" % val)
def _get(self, url, response_key):
def _get(self, url, response_key, filters=None):
if filters:
url = utils.get_url_with_filter(url, filters)
resp, body = self.api.client.get(url)
if response_key is not None:
content = body[response_key]

View File

@ -443,19 +443,31 @@ class RecordTimeTestCase(test_utils.TestCase):
class PrepareQueryStringTestCase(test_utils.TestCase):
def test_convert_dict_to_string(self):
ustr = b'?\xd0\xbf=1&\xd1\x80=2'
def setUp(self):
super(PrepareQueryStringTestCase, self).setUp()
self.ustr = b'?\xd0\xbf=1&\xd1\x80=2'
if six.PY3:
# in py3 real unicode symbols will be urlencoded
ustr = ustr.decode('utf8')
cases = (
self.ustr = self.ustr.decode('utf8')
self.cases = (
({}, ''),
(None, ''),
({'2': 2, '10': 1}, '?10=1&2=2'),
({'abc': 1, 'abc1': 2}, '?abc=1&abc1=2'),
({b'\xd0\xbf': 1, b'\xd1\x80': 2}, ustr),
({b'\xd0\xbf': 1, b'\xd1\x80': 2}, self.ustr),
({(1, 2): '1', (3, 4): '2'}, '?(1, 2)=1&(3, 4)=2')
)
for case in cases:
def test_convert_dict_to_string(self):
for case in self.cases:
self.assertEqual(
case[1],
parse.unquote_plus(utils.prepare_query_string(case[0])))
def test_get_url_with_filter(self):
url = '/fake'
for case in self.cases:
self.assertEqual(
'%s%s' % (url, case[1]),
parse.unquote_plus(utils.get_url_with_filter(url, case[0])))

View File

@ -454,5 +454,15 @@ def record_time(times, enabled, *args):
def prepare_query_string(params):
"""Convert dict params to query string"""
# Transform the dict to a sequence of two-element tuples in fixed
# order, then the encoded string will be consistent in Python 2&3.
if not params:
return ''
params = sorted(params.items(), key=lambda x: x[0])
return '?%s' % parse.urlencode(params) if params else ''
def get_url_with_filter(url, filters):
query_string = prepare_query_string(filters)
url = "%s%s" % (url, query_string)
return url

View File

@ -17,7 +17,6 @@ Flavor interface.
"""
from oslo_utils import strutils
from six.moves.urllib import parse
from novaclient import base
from novaclient import exceptions
@ -128,14 +127,11 @@ class FlavorManager(base.ManagerWithFind):
qparams['sort_dir'] = str(sort_dir)
if not is_public:
qparams['is_public'] = is_public
qparams = sorted(qparams.items(), key=lambda x: x[0])
query_string = "?%s" % parse.urlencode(qparams) if qparams else ""
detail = ""
if detailed:
detail = "/detail"
return self._list("/flavors%s%s" % (detail, query_string), "flavors")
return self._list("/flavors%s" % detail, "flavors", filters=qparams)
def get(self, flavor):
"""Get a specific flavor.

View File

@ -19,7 +19,6 @@ Keypair interface
from novaclient import api_versions
from novaclient import base
from novaclient import utils
class Keypair(base.Resource):
@ -170,9 +169,11 @@ class KeypairManager(base.ManagerWithFind):
:param user_id: Id of key-pairs owner (Admin only).
"""
query_string = "?user_id=%s" % user_id if user_id else ""
url = '/%s%s' % (self.keypair_prefix, query_string)
return self._list(url, 'keypairs')
params = {}
if user_id:
params['user_id'] = user_id
return self._list('/%s' % self.keypair_prefix, 'keypairs',
filters=params)
@api_versions.wraps("2.35")
def list(self, user_id=None, marker=None, limit=None):
@ -192,6 +193,5 @@ class KeypairManager(base.ManagerWithFind):
params['limit'] = int(limit)
if marker:
params['marker'] = str(marker)
query_string = utils.prepare_query_string(params)
url = '/%s%s' % (self.keypair_prefix, query_string)
return self._list(url, 'keypairs')
return self._list('/%s' % self.keypair_prefix, 'keypairs',
filters=params)

View File

@ -12,8 +12,6 @@
# License for the specific language governing permissions and limitations
# under the License.
from six.moves.urllib import parse
from novaclient import base
@ -95,6 +93,4 @@ class LimitsManager(base.Manager):
opts['reserved'] = 1
if tenant_id:
opts['tenant_id'] = tenant_id
query_string = "?%s" % parse.urlencode(opts) if opts else ""
return self._get("/limits%s" % query_string, "limits")
return self._get("/limits", "limits", filters=opts)

View File

@ -14,8 +14,6 @@
migration interface
"""
from six.moves.urllib import parse
from novaclient import base
from novaclient.i18n import _
@ -50,10 +48,4 @@ class MigrationManager(base.ManagerWithFind):
if instance_uuid:
opts['instance_uuid'] = instance_uuid
# Transform the dict to a sequence of two-element tuples in fixed
# order, then the encoded string will be consistent in Python 2&3.
new_opts = sorted(opts.items(), key=lambda x: x[0])
query_string = "?%s" % parse.urlencode(new_opts) if new_opts else ""
return self._list("/os-migrations%s" % query_string, "migrations")
return self._list("/os-migrations", "migrations", filters=opts)

View File

@ -17,8 +17,6 @@
Server group interface.
"""
from six.moves.urllib import parse
from novaclient import base
@ -62,10 +60,8 @@ class ServerGroupsManager(base.ManagerWithFind):
qparams['limit'] = int(limit)
if offset:
qparams['offset'] = int(offset)
qparams = sorted(qparams.items(), key=lambda x: x[0])
query_string = "?%s" % parse.urlencode(qparams) if qparams else ""
return self._list('/os-server-groups%s' % query_string,
'server_groups')
return self._list('/os-server-groups', 'server_groups',
filters=qparams)
def get(self, id):
"""Get a specific server group.