diff --git a/neutron_lib/api/validators.py b/neutron_lib/api/validators.py index 722c005..b40ef50 100644 --- a/neutron_lib/api/validators.py +++ b/neutron_lib/api/validators.py @@ -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, diff --git a/neutron_lib/tests/unit/api/test_validators.py b/neutron_lib/tests/unit/api/test_validators.py index fececbf..70ba5b5 100644 --- a/neutron_lib/tests/unit/api/test_validators.py +++ b/neutron_lib/tests/unit/api/test_validators.py @@ -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))