Enhance V2 validations to work better for integers and booleans

Fixes bug 1026598

The attributes are now able to cast a input value to a specific type.
At the moment boolean and int are supported.

Change-Id: I568a95bc60f91c3eeae03b305031502d50de9c44
This commit is contained in:
Gary Kotton 2012-07-19 08:51:38 -04:00
parent 71eac0df7b
commit 73ec08b7fc
3 changed files with 59 additions and 14 deletions

View File

@ -29,9 +29,21 @@ import logging
import netaddr
import re
from quantum.common import exceptions as q_exc
LOG = logging.getLogger(__name__)
def _validate_boolean(data, valid_values=None):
if data in [True, False]:
return
else:
msg = _("%s is not boolean") % data
LOG.debug("validate_boolean: %s", msg)
return msg
def _validate_values(data, valid_values=None):
if data in valid_values:
return
@ -82,13 +94,32 @@ def _validate_regex(data, valid_values=None):
return msg
def convert_to_boolean(data):
try:
i = int(data)
if i in [True, False]:
# Ensure that the value is True or False
if i:
return True
else:
return False
except ValueError, TypeError:
if (data == "True" or data == "true"):
return True
if (data == "False" or data == "false"):
return False
msg = _("%s is not boolean") % data
raise q_exc.InvalidInput(error_message=msg)
HEX_ELEM = '[0-9A-Fa-f]'
UUID_PATTERN = '-'.join([HEX_ELEM + '{8}', HEX_ELEM + '{4}',
HEX_ELEM + '{4}', HEX_ELEM + '{4}',
HEX_ELEM + '{12}'])
# Dictionary that maintains a list of validation functions
validators = {'type:values': _validate_values,
validators = {'type:boolean': _validate_boolean,
'type:values': _validate_values,
'type:mac_address': _validate_mac_address,
'type:ip_address': _validate_ip_address,
'type:subnet': _validate_subnet,
@ -114,8 +145,8 @@ RESOURCE_ATTRIBUTE_MAP = {
'name': {'allow_post': True, 'allow_put': True},
'subnets': {'allow_post': True, 'allow_put': True, 'default': []},
'admin_state_up': {'allow_post': True, 'allow_put': True,
'default': True,
'validate': {'type:values': [True, False]}},
'default': True, 'convert_to': convert_to_boolean,
'validate': {'type:boolean': None}},
'status': {'allow_post': False, 'allow_put': False},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True},
@ -126,8 +157,8 @@ RESOURCE_ATTRIBUTE_MAP = {
'network_id': {'allow_post': True, 'allow_put': False,
'validate': {'type:regex': UUID_PATTERN}},
'admin_state_up': {'allow_post': True, 'allow_put': True,
'default': True,
'validate': {'type:values': [True, False]}},
'default': True, 'convert_to': convert_to_boolean,
'validate': {'type:boolean': None}},
'mac_address': {'allow_post': True, 'allow_put': False,
'default': ATTR_NOT_SPECIFIED,
'validate': {'type:mac_address': None}},
@ -143,6 +174,7 @@ RESOURCE_ATTRIBUTE_MAP = {
'id': {'allow_post': False, 'allow_put': False,
'validate': {'type:regex': UUID_PATTERN}},
'ip_version': {'allow_post': True, 'allow_put': False,
'convert_to': int,
'validate': {'type:values': [4, 6]}},
'network_id': {'allow_post': True, 'allow_put': False,
'validate': {'type:regex': UUID_PATTERN}},

View File

@ -327,15 +327,19 @@ class Controller(object):
if attr_vals['allow_post']:
res_dict[attr] = res_dict.get(attr,
attr_vals.get('default'))
else: # PUT
for attr, attr_vals in self._attr_info.iteritems():
if attr in res_dict and not attr_vals['allow_put']:
msg = _("Cannot update read-only attribute %s") % attr
raise webob.exc.HTTPUnprocessableEntity(msg)
# Check that configured values are correct
for attr, attr_vals in self._attr_info.iteritems():
# Convert values if necessary
if ('convert_to' in attr_vals and
attr in res_dict):
res_dict[attr] = attr_vals['convert_to'](res_dict[attr])
# Check that configured values are correct
if not ('validate' in attr_vals and
attr in res_dict and
res_dict[attr] != attributes.ATTR_NOT_SPECIFIED):

View File

@ -777,7 +777,7 @@ class TestPortsV2(QuantumDbPluginV2TestCase):
'fixed_ips': []}}
port_req = self.new_create_request('ports', data)
res = port_req.get_response(self.api)
self.assertEquals(res.status_int, 422)
self.assertEquals(res.status_int, 400)
def test_invalid_mac_address(self):
with self.network() as network:
@ -821,12 +821,21 @@ class TestNetworksV2(QuantumDbPluginV2TestCase):
net['network']['name'])
def test_invalid_admin_status(self):
data = {'network': {'name': 'net',
'admin_state_up': 7,
'tenant_id': self._tenant_id}}
network_req = self.new_create_request('networks', data)
res = network_req.get_response(self.api)
self.assertEquals(res.status_int, 422)
fmt = 'json'
value = [[7, False, 400], [True, True, 201], ["True", True, 201],
["true", True, 201], [1, True, 201], ["False", False, 201],
[False, False, 201], ["false", False, 201],
["7", False, 400]]
for v in value:
data = {'network': {'name': 'net',
'admin_state_up': v[0],
'tenant_id': self._tenant_id}}
network_req = self.new_create_request('networks', data)
req = network_req.get_response(self.api)
self.assertEquals(req.status_int, v[2])
if v[2] == 201:
res = self.deserialize(fmt, req)
self.assertEquals(res['network']['admin_state_up'], v[1])
class TestSubnetsV2(QuantumDbPluginV2TestCase):