openstack/glance
Code Issues Proposed changes

Merge "Validate size of 'min_ram' and 'min_disk'"

Browse Source
This commit is contained in:
Jenkins 2015-07-01 02:27:12 +00:00 committed by Gerrit Code Review
commit 737be8dd51
3 changed files with 92 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
glance/db/sqlalchemy/api.py
View File

@ -657,22 +657,24 @@ def _image_get_disk_usage_by_owner(owner, session, image_id=None):
return total
def _validate_image(values):
def _validate_image(values, mandatory_status=True):
"""
Validates the incoming data and raises a Invalid exception
if anything is out of order.
:param values: Mapping of image metadata to check
:param mandatory_status: Whether to validate status from values
"""
status = values.get('status')
if not status:
msg = "Image status is required."
raise exception.Invalid(msg)
if mandatory_status:
status = values.get('status')
if not status:
msg = "Image status is required."
raise exception.Invalid(msg)
if status not in STATUSES:
msg = "Invalid image status '%s' for image." % status
raise exception.Invalid(msg)
if status not in STATUSES:
msg = "Invalid image status '%s' for image." % status
raise exception.Invalid(msg)
# validate integer values to eliminate DBError on save
utils.validate_mysql_int(min_disk=values.get('min_disk'),
@ -752,8 +754,8 @@ def _image_update(context, values, image_id, purge_props=False,
if from_state:
query = query.filter_by(status=from_state)
if new_status:
_validate_image(values)
mandatory_status = True if new_status else False
_validate_image(values, mandatory_status=mandatory_status)
# Validate fields for Images table. This is similar to what is done
# for the query result update except that we need to do it prior

42
glance/tests/unit/v1/test_api.py
View File

@ -201,6 +201,27 @@ class TestGlanceAPI(base.IsolatedUnitTest):
self.assertEqual(400, res.status_int)
self.assertIn('Invalid value', res.body)
def test_invalid_min_disk_size_update(self):
fixture_headers = {'x-image-meta-disk-format': 'vhd',
'x-image-meta-container-format': 'ovf',
'x-image-meta-name': 'fake image #3'}
req = webob.Request.blank("/images")
req.method = 'POST'
for k, v in six.iteritems(fixture_headers):
req.headers[k] = v
res = req.get_response(self.api)
self.assertEqual(201, res.status_int)
res_body = jsonutils.loads(res.body)['image']
self.assertEqual('queued', res_body['status'])
image_id = res_body['id']
req = webob.Request.blank("/images/%s" % image_id)
req.method = 'PUT'
req.headers['x-image-meta-min-disk'] = str(2 ** 31 + 1)
res = req.get_response(self.api)
self.assertEqual(400, res.status_int)
def test_bad_min_ram_size_create(self):
fixture_headers = {'x-image-meta-store': 'file',
'x-image-meta-disk-format': 'vhd',
@ -238,6 +259,27 @@ class TestGlanceAPI(base.IsolatedUnitTest):
self.assertEqual(400, res.status_int)
self.assertIn('Invalid value', res.body)
def test_invalid_min_ram_size_update(self):
fixture_headers = {'x-image-meta-disk-format': 'vhd',
'x-image-meta-container-format': 'ovf',
'x-image-meta-name': 'fake image #3'}
req = webob.Request.blank("/images")
req.method = 'POST'
for k, v in six.iteritems(fixture_headers):
req.headers[k] = v
res = req.get_response(self.api)
self.assertEqual(201, res.status_int)
res_body = jsonutils.loads(res.body)['image']
self.assertEqual('queued', res_body['status'])
image_id = res_body['id']
req = webob.Request.blank("/images/%s" % image_id)
req.method = 'PUT'
req.headers['x-image-meta-min-ram'] = str(2 ** 31 + 1)
res = req.get_response(self.api)
self.assertEqual(400, res.status_int)
def test_bad_disk_format(self):
fixture_headers = {
'x-image-meta-store': 'bad',

38
glance/tests/unit/v2/test_registry_client.py
View File

@ -587,6 +587,44 @@ class TestRegistryV2Client(base.IsolatedUnitTest,
'from': from_state})
self.assertEqual(str(exc), msg)
def test_image_update_with_invalid_min_disk(self):
"""Tests that the registry API updates the image"""
next_state = 'saving'
fixture = {'name': 'fake image',
'disk_format': 'vmdk',
'min_disk': str(2 ** 31 + 1),
'status': next_state}
image = self.client.image_get(image_id=UUID2)
current = image['status']
self.assertEqual('active', current)
# image is in 'active' state so this should cause a failure.
from_state = 'saving'
self.assertRaises(exception.Invalid, self.client.image_update,
image_id=UUID2, values=fixture,
from_state=from_state)
def test_image_update_with_invalid_min_ram(self):
"""Tests that the registry API updates the image"""
next_state = 'saving'
fixture = {'name': 'fake image',
'disk_format': 'vmdk',
'min_ram': str(2 ** 31 + 1),
'status': next_state}
image = self.client.image_get(image_id=UUID2)
current = image['status']
self.assertEqual('active', current)
# image is in 'active' state so this should cause a failure.
from_state = 'saving'
self.assertRaises(exception.Invalid, self.client.image_update,
image_id=UUID2, values=fixture,
from_state=from_state)
def _test_image_update_not_existing(self):
"""Tests non existing image update doesn't work"""
fixture = self.get_fixture(status='bad status')