openstack
/
glance
Code Issues Proposed changes

Port v1.test_registry_api to Python 3

Browse Source 
* urlsafe_b64decode(): on Python 3, decode from ASCII to get Unicode
  string.
* urlsafe_decrypt(): encode Unicode to UTF-8.
* in some loops, copy dictionary keys or items because the dictionary
  is modified in the loop body.
* test_api: use byte string for image content, not Unicode
* test_registry_api: HTTP body type is bytes, use byte strings and
  jsonutils.dump_as_bytes(), instead of native strings and
  jsonutils.dumps()
* tox.ini: add glance.tests.unit.v1.test_registry_api to Python 3.4

Change-Id: Ib9d18dce6e5728b9adf094b5aae64a86a3fea71a
changes/95/238595/2
Victor Stinner 7 years ago
parent 6f68059e5c
commit f85590f4d8
8 changed files with 53 additions and 44 deletions
Show Stats Download Patch File Download Diff File

9
glance/common/crypt.py
Unescape View File

@ -57,7 +57,10 @@ def urlsafe_encrypt(key, plaintext, blocksize=16):
init_vector = Random.get_random_bytes(16)
cypher = AES.new(key, AES.MODE_CBC, init_vector)
padded = cypher.encrypt(pad(six.binary_type(plaintext)))
return base64.urlsafe_b64encode(init_vector + padded)
encoded = base64.urlsafe_b64encode(init_vector + padded)
if six.PY3:
encoded = encoded.decode('ascii')
return encoded
def urlsafe_decrypt(key, ciphertext):
@ -71,7 +74,9 @@ def urlsafe_decrypt(key, ciphertext):
:returns : Resulting plaintext
"""
# Cast from unicode
ciphertext = base64.urlsafe_b64decode(six.binary_type(ciphertext))
if isinstance(ciphertext, six.text_type):
ciphertext = ciphertext.encode('utf-8')
ciphertext = base64.urlsafe_b64decode(ciphertext)
cypher = AES.new(key, AES.MODE_CBC, ciphertext[:16])
padded = cypher.decrypt(ciphertext[16:])
text = padded[:padded.rfind(b'\0')]

8
glance/db/sqlalchemy/api.py
Unescape View File

@ -455,7 +455,10 @@ def _make_conditions_from_filters(filters, is_public=None):
filters = {k: v for k, v in filters.items() if v is not None}
for (k, v) in filters.items():
# need to copy items because filters is modified in the loop body
# (filters.pop(k))
keys = list(filters.keys())
for k in keys:
key = k
if k.endswith('_min') or k.endswith('_max'):
key = key[0:-4]
@ -471,8 +474,7 @@ def _make_conditions_from_filters(filters, is_public=None):
if k.endswith('_max'):
image_conditions.append(getattr(models.Image, key) <= v)
for (k, v) in filters.items():
value = filters.pop(k)
for (k, value) in filters.items():
if hasattr(models.Image, k):
image_conditions.append(getattr(models.Image, k) == value)
else:

4
glance/registry/api/v1/images.py
Unescape View File

@ -206,7 +206,9 @@ class Controller(object):
# Only admin gets to look for non-public images
params['is_public'] = self._get_is_public(req)
for key, value in params.items():
# need to coy items because the params is modified in the loop body
items = list(params.items())
for key, value in items:
if value is None:
del params[key]

7
glance/tests/unit/test_misc.py
Unescape View File

@ -43,11 +43,8 @@ class UtilsTestCase(test_utils.BaseTestCase):
for key in key_list:
for plaintext in plaintext_list:
ciphertext = crypt.urlsafe_encrypt(key, plaintext, blocksize)
self.assertIsInstance(ciphertext, bytes)
if six.PY3:
self.assertNotEqual(ciphertext, plaintext.encode('utf-8'))
else:
self.assertNotEqual(ciphertext, plaintext)
self.assertIsInstance(ciphertext, str)
self.assertNotEqual(ciphertext, plaintext)
text = crypt.urlsafe_decrypt(key, ciphertext)
self.assertIsInstance(text, str)
self.assertEqual(plaintext, text)

2
glance/tests/unit/v1/test_api.py
Unescape View File

@ -127,7 +127,7 @@ class TestGlanceAPI(base.IsolatedUnitTest):
db_api.image_create(self.context, fixture)
# We write a fake image file to the filesystem
with open("%s/%s" % (self.test_dir, fixture['id']), 'wb') as image:
image.write("chunk00000remainder")
image.write(b"chunk00000remainder")
image.flush()
def destroy_fixtures(self):

64
glance/tests/unit/v1/test_registry_api.py
Unescape View File

@ -208,7 +208,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
when a malformed marker is provided
"""
res = self.get_api_response_ext(400, url='/images?marker=4')
self.assertIn('marker', res.body)
self.assertIn(b'marker', res.body)
def test_get_index_forbidden_marker(self):
"""
@ -711,7 +711,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
when a malformed marker is provided
"""
res = self.get_api_response_ext(400, url='/images/detail?marker=4')
self.assertIn('marker', res.body)
self.assertIn(b'marker', res.body)
def test_get_details_forbidden_marker(self):
"""
@ -1206,7 +1206,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture()
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(200, body=body,
method='POST', content_type='json')
@ -1221,7 +1221,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
def test_create_image_with_min_disk(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture(min_disk=5)
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(200, body=body,
method='POST', content_type='json')
@ -1232,7 +1232,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
def test_create_image_with_min_ram(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture(min_ram=256)
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(200, body=body,
method='POST', content_type='json')
@ -1243,7 +1243,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
def test_create_image_with_min_ram_default(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture()
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(200, body=body,
method='POST', content_type='json')
@ -1254,7 +1254,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
def test_create_image_with_min_disk_default(self):
"""Tests that the /images POST registry API creates the image"""
fixture = self.get_minimal_fixture()
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(200, body=body,
method='POST', content_type='json')
@ -1265,18 +1265,19 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
def test_create_image_with_bad_status(self):
"""Tests proper exception is raised if a bad status is set"""
fixture = self.get_minimal_fixture(id=_gen_uuid(), status='bad status')
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(400, body=body,
method='POST', content_type='json')
self.assertIn('Invalid image status', res.body)
self.assertIn(b'Invalid image status', res.body)
def test_create_image_with_bad_id(self):
"""Tests proper exception is raised if a bad disk_format is set"""
fixture = self.get_minimal_fixture(id='asdf')
body = jsonutils.dump_as_bytes(dict(image=fixture))
self.get_api_response_ext(400, content_type='json', method='POST',
body=jsonutils.dumps(dict(image=fixture)))
body=body)
def test_create_image_with_image_id_in_log(self):
"""Tests correct image id in log message when creating image"""
@ -1291,8 +1292,9 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
self.stubs.Set(rserver.images.LOG, 'info', fake_log_info)
body = jsonutils.dump_as_bytes(dict(image=fixture))
self.get_api_response_ext(200, content_type='json', method='POST',
body=jsonutils.dumps(dict(image=fixture)))
body=body)
self.assertTrue(self.log_image_id)
def test_update_image(self):
@ -1301,7 +1303,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
'min_disk': 5,
'min_ram': 256,
'disk_format': 'raw'}
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(200, url='/images/%s' % UUID2,
body=body, method='PUT',
@ -1330,7 +1332,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
'min_ram': 256,
'disk_format': 'raw',
'location': 'fake://image'}
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
log_debug.side_effect = fake_log_debug
@ -1352,7 +1354,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
non-existing image
"""
fixture = {'status': 'killed'}
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
self.get_api_response_ext(404, url='/images/%s' % _gen_uuid(),
method='PUT', body=body, content_type='json')
@ -1360,12 +1362,12 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
def test_update_image_with_bad_status(self):
"""Tests that exception raised trying to set a bad status"""
fixture = {'status': 'invalid'}
body = jsonutils.dumps(dict(image=fixture))
body = jsonutils.dump_as_bytes(dict(image=fixture))
res = self.get_api_response_ext(400, method='PUT', body=body,
url='/images/%s' % UUID2,
content_type='json')
self.assertIn('Invalid image status', res.body)
self.assertIn(b'Invalid image status', res.body)
def test_update_private_image_no_admin(self):
"""
@ -1379,7 +1381,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
db_api.image_create(self.context, extra_fixture)
test_rserv = rserver.API(self.mapper)
api = test_utils.FakeAuthMiddleware(test_rserv, is_admin=False)
body = jsonutils.dumps(dict(image=extra_fixture))
body = jsonutils.dump_as_bytes(dict(image=extra_fixture))
self.get_api_response_ext(404, body=body, api=api,
url='/images/%s' % UUID8, method='PUT',
content_type='json')
@ -1507,7 +1509,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
self.api = test_utils.FakeAuthMiddleware(rserver.API(self.mapper),
is_admin=False)
fixture = dict(member_id='pattieblack')
body = jsonutils.dumps(dict(image_memberships=fixture))
body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
self.get_api_response_ext(401, method='PUT', body=body,
url='/images/%s/members' % UUID2,
@ -1523,7 +1525,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
req.method = 'PUT'
self.context.tenant = 'test2'
req.content_type = 'application/json'
req.body = jsonutils.dumps(dict(image_memberships=fixture))
req.body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
res = req.get_response(self.api)
self.assertEqual(404, res.status_int)
@ -1550,7 +1552,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
self.assertEqual(1, num_members)
fixture = dict(member_id='test1')
body = jsonutils.dumps(dict(image_memberships=fixture))
body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
self.get_api_response_ext(400, url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
@ -1570,7 +1572,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
req.headers['X-Auth-Token'] = 'test1:test1:'
req.method = 'PUT'
req.content_type = 'application/json'
req.body = jsonutils.dumps(dict(image_memberships=fixture))
req.body = jsonutils.dump_as_bytes(dict(image_memberships=fixture))
res = req.get_response(api)
self.assertEqual(403, res.status_int)
@ -1591,7 +1593,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
req.get_response(self.api)
fixture = [dict(member_id='test2', can_share=True)]
body = jsonutils.dumps(dict(memberships=fixture))
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(204, url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
@ -1612,7 +1614,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
req.method = 'PUT'
req.get_response(self.api)
fixture = dict(member_id='test3')
body = jsonutils.dumps(dict(memberships=fixture))
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(400, url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
@ -1633,7 +1635,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
req.get_response(self.api)
fixture = [dict(member_id='test1', can_share=False)]
body = jsonutils.dumps(dict(memberships=fixture))
body = jsonutils.dump_as_bytes(dict(memberships=fixture))
self.get_api_response_ext(204, url='/images/%s/members' % UUID8,
method='PUT', body=body,
content_type='json')
@ -1659,7 +1661,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
db_api.image_create(self.context, extra_fixture)
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_positive'
body = jsonutils.dumps(dict(member=fixture))
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(204, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
@ -1671,7 +1673,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
"""
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_positive'
body = jsonutils.dumps(dict(member=fixture))
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(404, url=test_uri % _gen_uuid(),
method='PUT', body=body,
content_type='json')
@ -1692,7 +1694,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
req.headers['X-Auth-Token'] = 'test1:test1:'
req.method = 'PUT'
req.content_type = 'application/json'
req.body = jsonutils.dumps(dict(member=fixture))
req.body = jsonutils.dump_as_bytes(dict(member=fixture))
res = req.get_response(api)
self.assertEqual(403, res.status_int)
@ -1709,7 +1711,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
fixture = [dict(can_share=True)]
test_uri = '/images/%s/members/test_add_member_bad_request'
body = jsonutils.dumps(dict(member=fixture))
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(400, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
@ -1733,7 +1735,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
res = self.get_api_response_ext(404, method='DELETE',
url=('/images/%s/members/pattieblack' %
UUID2))
self.assertIn('Membership could not be found', res.body)
self.assertIn(b'Membership could not be found', res.body)
def test_delete_member_from_non_exist_image(self):
"""
@ -1779,7 +1781,7 @@ class TestRegistryAPI(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
db_api.image_create(self.context, extra_fixture)
fixture = dict(can_share=True)
test_uri = '/images/%s/members/test_add_member_delete_create'
body = jsonutils.dumps(dict(member=fixture))
body = jsonutils.dump_as_bytes(dict(member=fixture))
self.get_api_response_ext(204, url=test_uri % UUID8,
method='PUT', body=body,
content_type='json')
@ -1927,7 +1929,7 @@ class TestRegistryAPILocations(base.IsolatedUnitTest,
req = webob.Request.blank('/images')
req.method = 'POST'
req.content_type = 'application/json'
req.body = jsonutils.dumps(dict(image=fixture))
req.body = jsonutils.dump_as_bytes(dict(image=fixture))
res = req.get_response(self.api)
self.assertEqual(200, res.status_int)

2
glance/tests/utils.py
Unescape View File

@ -475,7 +475,7 @@ class RegistryAPIMixIn(object):
db_api.image_create(self.context, fixture)
with open(os.path.join(self.test_dir, fixture['id']),
'wb') as image:
image.write("chunk00000remainder")
image.write(b"chunk00000remainder")
def destroy_fixtures(self):
db_models.unregister_models(db_api.get_engine())

1
tox.ini
Unescape View File

@ -68,6 +68,7 @@ commands =
glance.tests.unit.test_store_image \
glance.tests.unit.test_store_location \
glance.tests.unit.test_versions \
glance.tests.unit.v1.test_registry_api \
glance.tests.unit.v1.test_upload_utils \
glance.tests.unit.v2.test_image_actions_resource \
glance.tests.unit.v2.test_image_tags_resource \

Write Preview
Loading…
Cancel
Save