openstack/neutron-lib
Code Issues Proposed changes

Add subport validator for vlan-aware-vms

Browse Source 
This patch introduces a new validator method for a list of trunk
subports.

A subport input is defined  by a ternary tuple of 'port_id' and
optionally 'segmentation_id' and 'segmentation_type'. If the
segmentation information is specified, both type and id must be
present; port_id must be a valid UUID and unique within the list,
and the same goes for segmentation id. More implementation specific
validations are deferred to the plugin layer.

The validator may find use in other projects besides 'neutron' alone
and thus is contributed in neutron-lib sooner rather than later.
Alternatively local validators should be supported to avoid modifying
the module variable as reported in bug 1584237.

Partially-implements: blueprint vlan-aware-vms

Change-Id: Ic04c547a75a7851d268ec6121eceaf0be9ca3586
This commit is contained in:
Armando Migliaccio
2016-05-20 10:38:11 -07:00
parent 45aea71d7d
commit f54a138fbd
2 changed files with 112 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
neutron_lib/api/validators.py
View File

@@ -488,6 +488,54 @@ def validate_non_negative(data, valid_values=None):
return msg
def validate_subports(data, valid_values=None):
if not isinstance(data, list):
msg = _("Invalid data format for subports: '%s'") % data
LOG.debug(msg)
return msg
subport_ids = set()
segmentation_ids = set()
for subport in data:
if not isinstance(subport, dict):
msg = _("Invalid data format for subport: '%s'") % subport
LOG.debug(msg)
return msg
# Expect a non duplicated and valid port_id for the subport
if 'port_id' not in subport:
msg = _("A valid port UUID must be specified")
LOG.debug(msg)
return msg
elif validate_uuid(subport["port_id"]):
msg = _("Invalid UUID for subport: '%s'") % subport["port_id"]
return msg
elif subport["port_id"] in subport_ids:
msg = _("Non unique UUID for subport: '%s'") % subport["port_id"]
return msg
subport_ids.add(subport["port_id"])
# Validate that both segmentation id and segmentation type are
# specified, and that the client does not duplicate segmentation
# ids
segmentation_id = subport.get("segmentation_id")
segmentation_type = subport.get("segmentation_type")
if (not segmentation_id or not segmentation_type) and len(subport) > 1:
msg = _("Invalid subport details '%s': missing segmentation "
"information. Must specify both segmentation_id and "
"segmentation_type") % subport
LOG.debug(msg)
return msg
if segmentation_id in segmentation_ids:
msg = _("Segmentation ID '%(seg_id)s' for '%(subport)s' is not "
"unique") % {"seg_id": segmentation_id,
"subport": subport["port_id"]}
LOG.debug(msg)
return msg
if segmentation_id:
segmentation_ids.add(segmentation_id)
# Dictionary that maintains a list of validation functions
validators = {'type:dict': validate_dict,
'type:dict_or_none': validate_dict_or_none,
@@ -515,6 +563,7 @@ validators = {'type:dict': validate_dict,
'type:subnet_or_none': validate_subnet_or_none,
'type:subnetpool_id': validate_subnetpool_id,
'type:subnetpool_id_or_none': validate_subnetpool_id_or_none,
'type:subports': validate_subports,
'type:uuid': validate_uuid,
'type:uuid_or_none': validate_uuid_or_none,
'type:uuid_list': validate_uuid_list,

63
neutron_lib/tests/unit/api/test_validators.py
View File

@@ -838,3 +838,66 @@ class TestAttributeValidation(base.BaseTestCase):
for value in (0, 1, '2', True, False):
msg = validators.validate_non_negative(value)
self.assertIsNone(msg)
def test_validate_subports_invalid_body(self):
self.assertIsNotNone(validators.validate_subports(None))
def test_validate_subports_invalid_subport_object(self):
self.assertIsNotNone(validators.validate_subports(['foo_port']))
def test_validate_subports_invalid_port_uuid(self):
body = [{'port_id': 'foo_port'}]
self.assertIsNotNone(validators.validate_subports(body))
def test_validate_subports_invalid_missing_port_id(self):
body = [{'poort_id': 'foo_port'}]
self.assertIsNotNone(validators.validate_subports(body))
def test_validate_subports_invalid_duplicate_port_ids(self):
body = [
{'port_id': '00000000-ffff-ffff-ffff-000000000000'},
{'port_id': '00000000-ffff-ffff-ffff-000000000000'}
]
self.assertIsNotNone(validators.validate_subports(body))
def test_validate_subports_invalid_incomplete_segmentation_details(self):
body = [
{'port_id': '00000000-ffff-ffff-ffff-000000000000',
'segmentation_id': '3'}
]
self.assertIsNotNone(validators.validate_subports(body))
def test_validate_subports_invalid_unknown_paramenter(self):
body = [
{'port_id': '00000000-ffff-ffff-ffff-000000000000',
'segmentation_id': '3', 'segmeNAtion_type': 'vlan'}
]
self.assertIsNotNone(validators.validate_subports(body))
def test_validate_subports_invalid_duplicate_segmentation_id(self):
body = [
{'port_id': '00000000-ffff-ffff-ffff-000000000000',
'segmentation_id': '3', 'segmentation_type': 'vlan'},
{'port_id': '11111111-ffff-ffff-ffff-000000000000',
'segmentation_id': '3', 'segmentation_type': 'vlan'}
]
self.assertIsNotNone(validators.validate_subports(body))
def test_validate_subports_valid_empty_body(self):
self.assertIsNone(validators.validate_subports([]))
def test_validate_subports_valid_suports_with_segmentation_details(self):
body = [
{'port_id': '00000000-ffff-ffff-ffff-000000000000',
'segmentation_id': '3', 'segmentation_type': 'vlan'},
{'port_id': '11111111-ffff-ffff-ffff-000000000000',
'segmentation_id': '5', 'segmentation_type': 'vlan'}
]
self.assertIsNone(validators.validate_subports(body))
def test_validate_subports_valid_subports(self):
body = [
{'port_id': '00000000-ffff-ffff-ffff-000000000000'},
{'port_id': '11111111-ffff-ffff-ffff-000000000000'},
]
self.assertIsNone(validators.validate_subports(body))