Ensure that HTTP 400 codes are returned for invalid input
Fixes bug 1062046 A number of additional validation functions were added. They do basic validations. Change-Id: I0fc207e996f11b50fdaa4d80075ae5704cba7003
This commit is contained in:
Gary Kotton
parent
20a63af370
commit
6a0db7180b
@ -36,7 +36,7 @@ def _validate_boolean(data, valid_values=None):
|
||||
if data in [True, False]:
|
||||
return
|
||||
else:
|
||||
msg = _("%s is not boolean") % data
|
||||
msg = _("'%s' is not boolean") % data
|
||||
LOG.debug("validate_boolean: %s", msg)
|
||||
return msg
|
||||
|
||||
@ -51,6 +51,19 @@ def _validate_values(data, valid_values=None):
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_string(data, max_len=None):
|
||||
if not isinstance(data, basestring):
|
||||
msg = _("'%s' is not a valid string") % data
|
||||
LOG.debug("validate_string: %s", msg)
|
||||
return msg
|
||||
if max_len is not None:
|
||||
if len(data) > max_len:
|
||||
msg = _("'%(data)s' exceeds maximum length of "
|
||||
"%(max_len)s.") % locals()
|
||||
LOG.debug("validate_string: %s", msg)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_range(data, valid_values=None):
|
||||
min_value = valid_values[0]
|
||||
max_value = valid_values[1]
|
||||
@ -69,7 +82,7 @@ def _validate_mac_address(data, valid_values=None):
|
||||
netaddr.EUI(data)
|
||||
return
|
||||
except Exception:
|
||||
msg = _("%s is not a valid MAC address") % data
|
||||
msg = _("'%s' is not a valid MAC address") % data
|
||||
LOG.debug("validate_mac_address: %s", msg)
|
||||
return msg
|
||||
|
||||
@ -79,11 +92,131 @@ def _validate_ip_address(data, valid_values=None):
|
||||
netaddr.IPAddress(data)
|
||||
return
|
||||
except Exception:
|
||||
msg = _("%s is not a valid IP address") % data
|
||||
msg = _("'%s' is not a valid IP address") % data
|
||||
LOG.debug("validate_ip_address: %s", msg)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_ip_pools(data, valid_values=None):
|
||||
"""Validate that start and end IP addresses are present
|
||||
|
||||
In addition to this the IP addresses will also be validated"""
|
||||
if not isinstance(data, list):
|
||||
msg = _("'%s' in not a valid IP pool") % data
|
||||
LOG.debug("validate_ip_pools: %s", msg)
|
||||
return msg
|
||||
|
||||
expected_keys = set(['start', 'end'])
|
||||
try:
|
||||
for ip_pool in data:
|
||||
if set(ip_pool.keys()) != expected_keys:
|
||||
msg = _("Expected keys not found. Expected: %s "
|
||||
"Provided: %s") % (expected_keys, ip_pool.keys())
|
||||
LOG.debug("validate_ip_pools: %s", msg)
|
||||
return msg
|
||||
for k in expected_keys:
|
||||
msg = _validate_ip_address(ip_pool[k])
|
||||
if msg:
|
||||
LOG.debug("validate_ip_pools: %s", msg)
|
||||
return msg
|
||||
except KeyError, e:
|
||||
args = {'key_name': e.message, 'ip_pool': ip_pool}
|
||||
msg = _("Invalid input. Required key: '%(key_name)s' "
|
||||
"missing from %(ip_pool)s.") % args
|
||||
LOG.debug("validate_ip_pools: %s", msg)
|
||||
return msg
|
||||
except TypeError, e:
|
||||
msg = _("Invalid input. Pool %s must be a dictionary.") % ip_pool
|
||||
LOG.debug("validate_ip_pools: %s", msg)
|
||||
return msg
|
||||
except Exception:
|
||||
msg = _("'%s' in not a valid IP pool") % data
|
||||
LOG.debug("validate_ip_pools: %s", msg)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_fixed_ips(data, valid_values=None):
|
||||
if not isinstance(data, list):
|
||||
msg = _("'%s' in not a valid fixed IP") % data
|
||||
LOG.debug("validate_fixed_ips: %s", msg)
|
||||
return msg
|
||||
ips = []
|
||||
try:
|
||||
for fixed_ip in data:
|
||||
if 'ip_address' in fixed_ip:
|
||||
msg = _validate_ip_address(fixed_ip['ip_address'])
|
||||
if msg:
|
||||
LOG.debug("validate_fixed_ips: %s", msg)
|
||||
return msg
|
||||
if 'subnet_id' in fixed_ip:
|
||||
msg = _validate_regex(fixed_ip['subnet_id'], UUID_PATTERN)
|
||||
if msg:
|
||||
LOG.debug("validate_fixed_ips: %s", msg)
|
||||
return msg
|
||||
# Ensure that duplicate entries are not set - just checking IP
|
||||
# suffices. Duplicate subnet_id's are legitimate.
|
||||
if 'ip_address' in fixed_ip:
|
||||
if fixed_ip['ip_address'] in ips:
|
||||
msg = _("Duplicate entry %s") % fixed_ip
|
||||
LOG.debug("validate_fixed_ips: %s", msg)
|
||||
return msg
|
||||
ips.append(fixed_ip['ip_address'])
|
||||
except Exception:
|
||||
msg = _("'%s' in not a valid fixed IP") % data
|
||||
LOG.debug("validate_fixed_ips: %s", msg)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_nameservers(data, valid_values=None):
|
||||
if not hasattr(data, '__iter__'):
|
||||
msg = _("'%s' in not a valid nameserver") % data
|
||||
LOG.debug("validate_nameservers: %s", msg)
|
||||
return msg
|
||||
ips = set()
|
||||
for ip in data:
|
||||
msg = _validate_ip_address(ip)
|
||||
if msg:
|
||||
# This may be a hostname
|
||||
msg = _validate_regex(ip, HOSTNAME_PATTERN)
|
||||
if msg:
|
||||
msg = _("'%s' in not a valid nameserver") % ip
|
||||
LOG.debug("validate_nameservers: %s", msg)
|
||||
return msg
|
||||
if ip in ips:
|
||||
msg = _("Duplicate nameserver %s") % ip
|
||||
LOG.debug("validate_nameservers: %s", msg)
|
||||
return msg
|
||||
ips.add(ip)
|
||||
|
||||
|
||||
def _validate_hostroutes(data, valid_values=None):
|
||||
if not isinstance(data, list):
|
||||
msg = _("'%s' in not a valid hostroute") % data
|
||||
LOG.debug("validate_hostroutes: %s", msg)
|
||||
return msg
|
||||
hostroutes = []
|
||||
try:
|
||||
for hostroute in data:
|
||||
msg = _validate_subnet(hostroute['destination'])
|
||||
if msg:
|
||||
LOG.debug("validate_hostroutes: %s", msg)
|
||||
return msg
|
||||
msg = _validate_ip_address(hostroute['nexthop'])
|
||||
if msg:
|
||||
LOG.debug("validate_hostroutes: %s", msg)
|
||||
return msg
|
||||
if hostroute in hostroutes:
|
||||
msg = _("Duplicate hostroute %s") % hostroute
|
||||
LOG.debug("validate_hostroutes: %s", msg)
|
||||
if msg:
|
||||
return msg
|
||||
hostroutes.append(hostroute)
|
||||
except:
|
||||
msg = _("'%s' in not a valid hostroute") % data
|
||||
LOG.debug("validate_hostroutes: %s", msg)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_ip_address_or_none(data, valid_values=None):
|
||||
if data is None:
|
||||
return None
|
||||
@ -98,17 +231,19 @@ def _validate_subnet(data, valid_values=None):
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
msg = _("%s is not a valid IP subnet") % data
|
||||
msg = _("'%s' is not a valid IP subnet") % data
|
||||
LOG.debug("validate_subnet: %s", msg)
|
||||
return msg
|
||||
|
||||
|
||||
def _validate_regex(data, valid_values=None):
|
||||
match = re.match(valid_values, data)
|
||||
if match:
|
||||
try:
|
||||
if re.match(valid_values, data):
|
||||
return
|
||||
else:
|
||||
msg = _("%s is not valid") % data
|
||||
except TypeError:
|
||||
pass
|
||||
|
||||
msg = _("'%s' is not valid input") % data
|
||||
LOG.debug("validate_regex: %s", msg)
|
||||
return msg
|
||||
|
||||
@ -131,12 +266,20 @@ def convert_to_boolean(data):
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
except ValueError, TypeError:
|
||||
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
|
||||
msg = _("'%s' is not boolean") % data
|
||||
raise q_exc.InvalidInput(error_message=msg)
|
||||
|
||||
|
||||
def convert_to_int(data):
|
||||
try:
|
||||
return int(data)
|
||||
except (ValueError, TypeError):
|
||||
msg = _("'%s' is not a integer") % data
|
||||
raise q_exc.InvalidInput(error_message=msg)
|
||||
|
||||
|
||||
@ -170,6 +313,8 @@ def convert_kvp_list_to_dict(kvp_list):
|
||||
kvp_map[key].add(value)
|
||||
return dict((x, list(y)) for x, y in kvp_map.iteritems())
|
||||
|
||||
HOSTNAME_PATTERN = ("(?=^.{1,254}$)(^(?:(?!\d+\.|-)[a-zA-Z0-9_\-]"
|
||||
"{1,63}(?<!-)\.?)+(?:[a-zA-Z]{2,})$)")
|
||||
|
||||
HEX_ELEM = '[0-9A-Fa-f]'
|
||||
UUID_PATTERN = '-'.join([HEX_ELEM + '{8}', HEX_ELEM + '{4}',
|
||||
@ -181,6 +326,11 @@ MAC_PATTERN = "^%s[aceACE02468](:%s{2}){5}$" % (HEX_ELEM, HEX_ELEM)
|
||||
|
||||
# Dictionary that maintains a list of validation functions
|
||||
validators = {'type:boolean': _validate_boolean,
|
||||
'type:values': _validate_values,
|
||||
'type:string': _validate_string,
|
||||
'type:range': _validate_range,
|
||||
'type:mac_address': _validate_mac_address,
|
||||
'type:fixed_ips': _validate_fixed_ips,
|
||||
'type:ip_address': _validate_ip_address,
|
||||
'type:ip_address_or_none': _validate_ip_address_or_none,
|
||||
'type:mac_address': _validate_mac_address,
|
||||
@ -188,7 +338,10 @@ validators = {'type:boolean': _validate_boolean,
|
||||
'type:regex': _validate_regex,
|
||||
'type:subnet': _validate_subnet,
|
||||
'type:uuid': _validate_uuid,
|
||||
'type:values': _validate_values}
|
||||
'type:regex': _validate_regex,
|
||||
'type:ip_pools': _validate_ip_pools,
|
||||
'type:hostroutes': _validate_hostroutes,
|
||||
'type:nameservers': _validate_nameservers}
|
||||
|
||||
# Note: a default of ATTR_NOT_SPECIFIED indicates that an
|
||||
# attribute is not required, but will be generated by the plugin
|
||||
@ -219,6 +372,7 @@ RESOURCE_ATTRIBUTE_MAP = {
|
||||
'validate': {'type:uuid': None},
|
||||
'is_visible': True},
|
||||
'name': {'allow_post': True, 'allow_put': True,
|
||||
'validate': {'type:string': None},
|
||||
'default': '', 'is_visible': True},
|
||||
'subnets': {'allow_post': False, 'allow_put': False,
|
||||
'default': [],
|
||||
@ -231,6 +385,7 @@ RESOURCE_ATTRIBUTE_MAP = {
|
||||
'status': {'allow_post': False, 'allow_put': False,
|
||||
'is_visible': True},
|
||||
'tenant_id': {'allow_post': True, 'allow_put': False,
|
||||
'validate': {'type:string': None},
|
||||
'required_by_policy': True,
|
||||
'is_visible': True},
|
||||
SHARED: {'allow_post': True,
|
||||
@ -247,6 +402,7 @@ RESOURCE_ATTRIBUTE_MAP = {
|
||||
'validate': {'type:uuid': None},
|
||||
'is_visible': True},
|
||||
'name': {'allow_post': True, 'allow_put': True, 'default': '',
|
||||
'validate': {'type:string': None},
|
||||
'is_visible': True},
|
||||
'network_id': {'allow_post': True, 'allow_put': False,
|
||||
'required_by_policy': True,
|
||||
@ -265,15 +421,19 @@ RESOURCE_ATTRIBUTE_MAP = {
|
||||
'fixed_ips': {'allow_post': True, 'allow_put': True,
|
||||
'default': ATTR_NOT_SPECIFIED,
|
||||
'convert_list_to': convert_kvp_list_to_dict,
|
||||
'validate': {'type:fixed_ips': None},
|
||||
'enforce_policy': True,
|
||||
'is_visible': True},
|
||||
'device_id': {'allow_post': True, 'allow_put': True,
|
||||
'validate': {'type:string': None},
|
||||
'default': '',
|
||||
'is_visible': True},
|
||||
'device_owner': {'allow_post': True, 'allow_put': True,
|
||||
'validate': {'type:string': None},
|
||||
'default': '',
|
||||
'is_visible': True},
|
||||
'tenant_id': {'allow_post': True, 'allow_put': False,
|
||||
'validate': {'type:string': None},
|
||||
'required_by_policy': True,
|
||||
'is_visible': True},
|
||||
'status': {'allow_post': False, 'allow_put': False,
|
||||
@ -284,9 +444,10 @@ RESOURCE_ATTRIBUTE_MAP = {
|
||||
'validate': {'type:uuid': None},
|
||||
'is_visible': True},
|
||||
'name': {'allow_post': True, 'allow_put': True, 'default': '',
|
||||
'validate': {'type:string': None},
|
||||
'is_visible': True},
|
||||
'ip_version': {'allow_post': True, 'allow_put': False,
|
||||
'convert_to': int,
|
||||
'convert_to': convert_to_int,
|
||||
'validate': {'type:values': [4, 6]},
|
||||
'is_visible': True},
|
||||
'network_id': {'allow_post': True, 'allow_put': False,
|
||||
@ -303,14 +464,18 @@ RESOURCE_ATTRIBUTE_MAP = {
|
||||
#TODO(salvatore-orlando): Enable PUT on allocation_pools
|
||||
'allocation_pools': {'allow_post': True, 'allow_put': False,
|
||||
'default': ATTR_NOT_SPECIFIED,
|
||||
'validate': {'type:ip_pools': None},
|
||||
'is_visible': True},
|
||||
'dns_nameservers': {'allow_post': True, 'allow_put': True,
|
||||
'default': ATTR_NOT_SPECIFIED,
|
||||
'validate': {'type:nameservers': None},
|
||||
'is_visible': True},
|
||||
'host_routes': {'allow_post': True, 'allow_put': True,
|
||||
'default': ATTR_NOT_SPECIFIED,
|
||||
'validate': {'type:hostroutes': None},
|
||||
'is_visible': True},
|
||||
'tenant_id': {'allow_post': True, 'allow_put': False,
|
||||
'validate': {'type:string': None},
|
||||
'required_by_policy': True,
|
||||
'is_visible': True},
|
||||
'enable_dhcp': {'allow_post': True, 'allow_put': True,
|
||||
|
||||
@ -15,6 +15,7 @@
|
||||
|
||||
import socket
|
||||
|
||||
import netaddr
|
||||
import webob.exc
|
||||
|
||||
from quantum.api.v2 import attributes
|
||||
@ -44,7 +45,10 @@ FAULT_MAP = {exceptions.NotFound: webob.exc.HTTPNotFound,
|
||||
exceptions.HostRoutesExhausted: webob.exc.HTTPBadRequest,
|
||||
exceptions.DNSNameServersExhausted: webob.exc.HTTPBadRequest,
|
||||
# Some plugins enforce policies as well
|
||||
exceptions.PolicyNotAuthorized: webob.exc.HTTPForbidden
|
||||
exceptions.PolicyNotAuthorized: webob.exc.HTTPForbidden,
|
||||
netaddr.AddrFormatError: webob.exc.HTTPBadRequest,
|
||||
AttributeError: webob.exc.HTTPBadRequest,
|
||||
ValueError: webob.exc.HTTPBadRequest,
|
||||
}
|
||||
|
||||
QUOTAS = quota.QUOTAS
|
||||
|
||||
@ -17,6 +17,7 @@
|
||||
Utility methods for working with WSGI servers redux
|
||||
"""
|
||||
|
||||
import netaddr
|
||||
import webob
|
||||
import webob.dec
|
||||
import webob.exc
|
||||
@ -93,7 +94,9 @@ def Resource(controller, faults=None, deserializers=None, serializers=None):
|
||||
method = getattr(controller, action)
|
||||
|
||||
result = method(request=request, **args)
|
||||
except exceptions.QuantumException as e:
|
||||
except (ValueError, AttributeError,
|
||||
exceptions.QuantumException,
|
||||
netaddr.AddrFormatError) as e:
|
||||
LOG.exception('%s failed' % action)
|
||||
body = serializer({'QuantumError': str(e)})
|
||||
kwargs = {'body': body, 'content_type': content_type}
|
||||
|
||||
@ -955,8 +955,8 @@ class QuantumDbPluginV2(quantum_plugin_base_v2.QuantumPluginBaseV2):
|
||||
s['gateway_ip'] != attributes.ATTR_NOT_SPECIFIED):
|
||||
self._validate_ip_version(ip_ver, s['gateway_ip'], 'gateway_ip')
|
||||
|
||||
if 'dns_nameservers' in s and \
|
||||
s['dns_nameservers'] != attributes.ATTR_NOT_SPECIFIED:
|
||||
if ('dns_nameservers' in s and
|
||||
s['dns_nameservers'] != attributes.ATTR_NOT_SPECIFIED):
|
||||
if len(s['dns_nameservers']) > cfg.CONF.max_dns_nameservers:
|
||||
raise q_exc.DNSNameServersExhausted(
|
||||
subnet_id=id,
|
||||
@ -969,8 +969,8 @@ class QuantumDbPluginV2(quantum_plugin_base_v2.QuantumPluginBaseV2):
|
||||
error_message=("error parsing dns address %s" % dns))
|
||||
self._validate_ip_version(ip_ver, dns, 'dns_nameserver')
|
||||
|
||||
if 'host_routes' in s and \
|
||||
s['host_routes'] != attributes.ATTR_NOT_SPECIFIED:
|
||||
if ('host_routes' in s and
|
||||
s['host_routes'] != attributes.ATTR_NOT_SPECIFIED):
|
||||
if len(s['host_routes']) > cfg.CONF.max_subnet_host_routes:
|
||||
raise q_exc.HostRoutesExhausted(
|
||||
subnet_id=id,
|
||||
|
||||
@ -23,6 +23,90 @@ from quantum.common import exceptions as q_exc
|
||||
|
||||
class TestAttributes(unittest2.TestCase):
|
||||
|
||||
def test_strings(self):
|
||||
msg = attributes._validate_string(None, None)
|
||||
self.assertEquals(msg, "'None' is not a valid string")
|
||||
|
||||
msg = attributes._validate_string("OK", None)
|
||||
self.assertEquals(msg, None)
|
||||
|
||||
msg = attributes._validate_string("123456789", 9)
|
||||
self.assertIsNone(msg)
|
||||
|
||||
msg = attributes._validate_string("1234567890", 9)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
def test_ip_pools(self):
|
||||
pools = [[{'end': '10.0.0.254'}],
|
||||
[{'start': '10.0.0.254'}],
|
||||
[{'start': '1000.0.0.254',
|
||||
'end': '1.1.1.1'}],
|
||||
[{'start': '10.0.0.2', 'end': '10.0.0.254',
|
||||
'forza': 'juve'}],
|
||||
[{'start': '10.0.0.2', 'end': '10.0.0.254'},
|
||||
{'end': '10.0.0.254'}],
|
||||
None]
|
||||
for pool in pools:
|
||||
msg = attributes._validate_ip_pools(pool, None)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
pools = [[{'end': '10.0.0.254', 'start': '10.0.0.2'},
|
||||
{'start': '11.0.0.2', 'end': '11.1.1.1'}],
|
||||
[{'start': '11.0.0.2', 'end': '11.0.0.100'}]]
|
||||
for pool in pools:
|
||||
msg = attributes._validate_ip_pools(pool, None)
|
||||
self.assertIsNone(msg)
|
||||
|
||||
def test_fixed_ips(self):
|
||||
fixed_ips = [[{'subnet_id': '00000000-ffff-ffff-ffff-000000000000',
|
||||
'ip_address': '1111.1.1.1'}],
|
||||
[{'subnet_id': 'invalid'}],
|
||||
None,
|
||||
[{'subnet_id': '00000000-0fff-ffff-ffff-000000000000',
|
||||
'ip_address': '1.1.1.1'},
|
||||
{'subnet_id': '00000000-ffff-ffff-ffff-000000000000',
|
||||
'ip_address': '1.1.1.1'}],
|
||||
[{'subnet_id': '00000000-ffff-ffff-ffff-000000000000',
|
||||
'ip_address': '1.1.1.1'},
|
||||
{'subnet_id': '00000000-ffff-ffff-ffff-000000000000',
|
||||
'ip_address': '1.1.1.1'}]]
|
||||
for fixed in fixed_ips:
|
||||
msg = attributes._validate_fixed_ips(fixed, None)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
def test_nameservers(self):
|
||||
ns_pools = [['1.1.1.2', '1.1.1.2'],
|
||||
['www.hostname.com', 'www.hostname.com'],
|
||||
['77.hostname.com'],
|
||||
['1000.0.0.1'],
|
||||
None]
|
||||
|
||||
for ns in ns_pools:
|
||||
msg = attributes._validate_nameservers(ns, None)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
ns_pools = [['100.0.0.2'],
|
||||
['www.hostname.com'],
|
||||
['www.great.marathons.to.travel'],
|
||||
['valid'],
|
||||
['www.internal.hostname.com']]
|
||||
|
||||
for ns in ns_pools:
|
||||
msg = attributes._validate_nameservers(ns, None)
|
||||
self.assertIsNone(msg)
|
||||
|
||||
def test_hostroutes(self):
|
||||
hostroute_pools = [[{'destination': '100.0.0.0/24'}],
|
||||
[{'nexthop': '10.0.2.20'}],
|
||||
[{'nexthop': '10.0.2.20',
|
||||
'destination': '100.0.0.0/8'},
|
||||
{'nexthop': '10.0.2.20',
|
||||
'destination': '100.0.0.0/8'}],
|
||||
None]
|
||||
for host_routes in hostroute_pools:
|
||||
msg = attributes._validate_hostroutes(host_routes, None)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
def test_mac_addresses(self):
|
||||
# Valid - 3 octets
|
||||
base_mac = "fa:16:3e:00:00:00"
|
||||
@ -40,50 +124,43 @@ class TestAttributes(unittest2.TestCase):
|
||||
base_mac = "01:16:3e:4f:00:00"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
# Invalid - invalid format
|
||||
base_mac = "a:16:3e:4f:00:00"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
# Invalid - invalid format
|
||||
base_mac = "ffa:16:3e:4f:00:00"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
# Invalid - invalid format
|
||||
base_mac = "01163e4f0000"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
# Invalid - invalid format
|
||||
base_mac = "01-16-3e-4f-00-00"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
# Invalid - invalid format
|
||||
base_mac = "00:16:3:f:00:00"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
# Invalid - invalid format
|
||||
base_mac = "12:3:4:5:67:89ab"
|
||||
msg = attributes._validate_regex(base_mac,
|
||||
attributes.MAC_PATTERN)
|
||||
error = '%s is not valid' % base_mac
|
||||
self.assertEquals(msg, error)
|
||||
self.assertIsNotNone(msg)
|
||||
|
||||
def test_cidr(self):
|
||||
# Valid - IPv4
|
||||
@ -108,21 +185,21 @@ class TestAttributes(unittest2.TestCase):
|
||||
cidr = "10.0.2.0"
|
||||
msg = attributes._validate_subnet(cidr,
|
||||
None)
|
||||
error = "%s is not a valid IP subnet" % cidr
|
||||
error = "'%s' is not a valid IP subnet" % cidr
|
||||
self.assertEquals(msg, error)
|
||||
|
||||
# Invalid - IPv6 without final octets, missing mask
|
||||
cidr = "fe80::"
|
||||
msg = attributes._validate_subnet(cidr,
|
||||
None)
|
||||
error = "%s is not a valid IP subnet" % cidr
|
||||
error = "'%s' is not a valid IP subnet" % cidr
|
||||
self.assertEquals(msg, error)
|
||||
|
||||
# Invalid - IPv6 with final octets, missing mask
|
||||
cidr = "fe80::0"
|
||||
msg = attributes._validate_subnet(cidr,
|
||||
None)
|
||||
error = "%s is not a valid IP subnet" % cidr
|
||||
error = "'%s' is not a valid IP subnet" % cidr
|
||||
self.assertEquals(msg, error)
|
||||
|
||||
|
||||
|
||||
@ -352,12 +352,12 @@ class QuantumDbPluginV2TestCase(unittest2.TestCase):
|
||||
def _do_side_effect(self, patched_plugin, orig, *args, **kwargs):
|
||||
""" Invoked by test cases for injecting failures in plugin """
|
||||
def second_call(*args, **kwargs):
|
||||
raise Exception('boom')
|
||||
raise AttributeError
|
||||
patched_plugin.side_effect = second_call
|
||||
return orig(*args, **kwargs)
|
||||
|
||||
def _validate_behavior_on_bulk_failure(self, res, collection):
|
||||
self.assertEqual(res.status_int, 500)
|
||||
self.assertEqual(res.status_int, 400)
|
||||
req = self.new_list_request(collection)
|
||||
res = req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 200)
|
||||
@ -559,6 +559,14 @@ class TestV2HTTPResponse(QuantumDbPluginV2TestCase):
|
||||
res = req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 200)
|
||||
|
||||
def test_update_invalid_json_400(self):
|
||||
with self.network() as net:
|
||||
req = self.new_update_request('networks',
|
||||
'{{"name": "aaa"}}',
|
||||
net['network']['id'])
|
||||
res = req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_bad_route_404(self):
|
||||
req = self.new_list_request('doohickeys')
|
||||
res = req.get_response(self.api)
|
||||
@ -808,6 +816,13 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
|
||||
self.assertEqual(res['port']['admin_state_up'],
|
||||
data['port']['admin_state_up'])
|
||||
|
||||
def test_update_device_id_null(self):
|
||||
with self.port() as port:
|
||||
data = {'port': {'device_id': None}}
|
||||
req = self.new_update_request('ports', data, port['port']['id'])
|
||||
res = req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_delete_network_if_port_exists(self):
|
||||
fmt = 'json'
|
||||
with self.port() as port:
|
||||
@ -1028,6 +1043,19 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
|
||||
res = self._create_port(fmt, net_id=net_id, **kwargs)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_overlapping_subnets(self):
|
||||
fmt = 'json'
|
||||
with self.subnet() as subnet:
|
||||
tenant_id = subnet['subnet']['tenant_id']
|
||||
net_id = subnet['subnet']['network_id']
|
||||
res = self._create_subnet(fmt,
|
||||
tenant_id=tenant_id,
|
||||
net_id=net_id,
|
||||
cidr='10.0.0.225/28',
|
||||
ip_version=4,
|
||||
gateway_ip=ATTR_NOT_SPECIFIED)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_requested_subnet_id_v4_and_v6(self):
|
||||
fmt = 'json'
|
||||
with self.subnet() as subnet:
|
||||
@ -1158,6 +1186,17 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
|
||||
self.assertEquals(ips[0]['subnet_id'], subnet['subnet']['id'])
|
||||
self._delete('ports', port2['port']['id'])
|
||||
|
||||
def test_invalid_ip(self):
|
||||
fmt = 'json'
|
||||
with self.subnet() as subnet:
|
||||
# Allocate specific IP
|
||||
kwargs = {"fixed_ips": [{'subnet_id': subnet['subnet']['id'],
|
||||
'ip_address': '1011.0.0.5'}]}
|
||||
net_id = subnet['subnet']['network_id']
|
||||
res = self._create_port(fmt, net_id=net_id, **kwargs)
|
||||
port = self.deserialize(fmt, res)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_requested_split(self):
|
||||
fmt = 'json'
|
||||
with self.subnet() as subnet:
|
||||
@ -1205,7 +1244,29 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
|
||||
net_id = subnet['subnet']['network_id']
|
||||
res = self._create_port(fmt, net_id=net_id, **kwargs)
|
||||
port2 = self.deserialize(fmt, res)
|
||||
self.assertEquals(res.status_int, 500)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_fixed_ip_invalid_subnet_id(self):
|
||||
fmt = 'json'
|
||||
with self.subnet() as subnet:
|
||||
# Allocate specific IP
|
||||
kwargs = {"fixed_ips": [{'subnet_id': 'i am invalid',
|
||||
'ip_address': '10.0.0.5'}]}
|
||||
net_id = subnet['subnet']['network_id']
|
||||
res = self._create_port(fmt, net_id=net_id, **kwargs)
|
||||
port2 = self.deserialize(fmt, res)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_fixed_ip_invalid_ip(self):
|
||||
fmt = 'json'
|
||||
with self.subnet() as subnet:
|
||||
# Allocate specific IP
|
||||
kwargs = {"fixed_ips": [{'subnet_id': subnet['subnet']['id'],
|
||||
'ip_address': '10.0.0.55555'}]}
|
||||
net_id = subnet['subnet']['network_id']
|
||||
res = self._create_port(fmt, net_id=net_id, **kwargs)
|
||||
port2 = self.deserialize(fmt, res)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_requested_ips_only(self):
|
||||
fmt = 'json'
|
||||
@ -1962,6 +2023,119 @@ class TestSubnetsV2(QuantumDbPluginV2TestCase):
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 403)
|
||||
|
||||
def test_create_subnet_bad_ip_version(self):
|
||||
with self.network() as network:
|
||||
# Check bad IP version
|
||||
data = {'subnet': {'network_id': network['network']['id'],
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': 'abc',
|
||||
'tenant_id': network['network']['tenant_id'],
|
||||
'gateway_ip': '10.0.2.1'}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_bad_ip_version_null(self):
|
||||
with self.network() as network:
|
||||
# Check bad IP version
|
||||
data = {'subnet': {'network_id': network['network']['id'],
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': None,
|
||||
'tenant_id': network['network']['tenant_id'],
|
||||
'gateway_ip': '10.0.2.1'}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_bad_uuid(self):
|
||||
with self.network() as network:
|
||||
# Check invalid UUID
|
||||
data = {'subnet': {'network_id': None,
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': 4,
|
||||
'tenant_id': network['network']['tenant_id'],
|
||||
'gateway_ip': '10.0.2.1'}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_bad_boolean(self):
|
||||
with self.network() as network:
|
||||
# Check invalid boolean
|
||||
data = {'subnet': {'network_id': network['network']['id'],
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': '4',
|
||||
'enable_dhcp': None,
|
||||
'tenant_id': network['network']['tenant_id'],
|
||||
'gateway_ip': '10.0.2.1'}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_bad_pools(self):
|
||||
with self.network() as network:
|
||||
# Check allocation pools
|
||||
allocation_pools = [[{'end': '10.0.0.254'}],
|
||||
[{'start': '10.0.0.254'}],
|
||||
[{'start': '1000.0.0.254'}],
|
||||
[{'start': '10.0.0.2', 'end': '10.0.0.254'},
|
||||
{'end': '10.0.0.254'}],
|
||||
None,
|
||||
[{'start': '10.0.0.2', 'end': '10.0.0.3'},
|
||||
{'start': '10.0.0.2', 'end': '10.0.0.3'}]]
|
||||
tenant_id = network['network']['tenant_id']
|
||||
for pool in allocation_pools:
|
||||
data = {'subnet': {'network_id': network['network']['id'],
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': '4',
|
||||
'tenant_id': tenant_id,
|
||||
'gateway_ip': '10.0.2.1',
|
||||
'allocation_pools': pool}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_bad_nameserver(self):
|
||||
with self.network() as network:
|
||||
# Check nameservers
|
||||
nameserver_pools = [['1100.0.0.2'],
|
||||
['1.1.1.2', '1.1000.1.3'],
|
||||
['1.1.1.2', '1.1.1.2'],
|
||||
None]
|
||||
tenant_id = network['network']['tenant_id']
|
||||
for nameservers in nameserver_pools:
|
||||
data = {'subnet': {'network_id': network['network']['id'],
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': '4',
|
||||
'tenant_id': tenant_id,
|
||||
'gateway_ip': '10.0.2.1',
|
||||
'dns_nameservers': nameservers}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_bad_hostroutes(self):
|
||||
with self.network() as network:
|
||||
# Check hostroutes
|
||||
hostroute_pools = [[{'destination': '100.0.0.0/24'}],
|
||||
[{'nexthop': '10.0.2.20'}],
|
||||
[{'nexthop': '10.0.2.20',
|
||||
'destination': '100.0.0.0/8'},
|
||||
{'nexthop': '10.0.2.20',
|
||||
'destination': '100.0.0.0/8'}],
|
||||
None]
|
||||
tenant_id = network['network']['tenant_id']
|
||||
for hostroutes in hostroute_pools:
|
||||
data = {'subnet': {'network_id': network['network']['id'],
|
||||
'cidr': '10.0.2.0/24',
|
||||
'ip_version': '4',
|
||||
'tenant_id': tenant_id,
|
||||
'gateway_ip': '10.0.2.1',
|
||||
'host_routes': hostroutes}}
|
||||
subnet_req = self.new_create_request('subnets', data)
|
||||
res = subnet_req.get_response(self.api)
|
||||
self.assertEquals(res.status_int, 400)
|
||||
|
||||
def test_create_subnet_defaults(self):
|
||||
gateway = '10.0.0.1'
|
||||
cidr = '10.0.0.0/24'
|
||||
|
||||
Loading…
Reference in New Issue
Block a user