diff --git a/neutron/extensions/subnet_service_types.py b/neutron/extensions/subnet_service_types.py index 23611463d42..94000767c45 100644 --- a/neutron/extensions/subnet_service_types.py +++ b/neutron/extensions/subnet_service_types.py @@ -14,6 +14,7 @@ from neutron_lib.api import extensions from neutron_lib.api import validators from neutron_lib import constants from neutron_lib import exceptions +import six import webob.exc from neutron._i18n import _ @@ -29,6 +30,10 @@ class InvalidSubnetServiceType(exceptions.InvalidInput): "to a valid device owner.") +class InvalidInputSubnetServiceType(exceptions.InvalidInput): + message = _("Subnet service type %(service_type)s is not a string.") + + def _validate_subnet_service_types(service_types, valid_values=None): if service_types: if not isinstance(service_types, list): @@ -41,7 +46,9 @@ def _validate_subnet_service_types(service_types, valid_values=None): prefixes += constants.DEVICE_OWNER_COMPUTE_PREFIX for service_type in service_types: - if not service_type.startswith(tuple(prefixes)): + if not isinstance(service_type, six.text_type): + raise InvalidInputSubnetServiceType(service_type=service_type) + elif not service_type.startswith(tuple(prefixes)): raise InvalidSubnetServiceType(service_type=service_type) diff --git a/neutron/tests/unit/extensions/test_subnet_service_types.py b/neutron/tests/unit/extensions/test_subnet_service_types.py index 030ec3c0203..bd0d7d60084 100644 --- a/neutron/tests/unit/extensions/test_subnet_service_types.py +++ b/neutron/tests/unit/extensions/test_subnet_service_types.py @@ -98,6 +98,7 @@ class SubnetServiceTypesExtensionTestCase( def test_create_subnet_invalid_type(self): self._test_create_subnet(['foo'], expect_fail=True) + self._test_create_subnet([1], expect_fail=True) def test_create_subnet_no_type(self): res = self._create_service_subnet() @@ -105,12 +106,12 @@ class SubnetServiceTypesExtensionTestCase( subnet = subnet['subnet'] self.assertFalse(subnet['service_types']) - def _test_update_subnet(self, subnet, service_types, expect_fail=False): + def _test_update_subnet(self, subnet, service_types, fail_code=None): data = {'subnet': {'service_types': service_types}} req = self.new_update_request('subnets', data, subnet['id']) res = self.deserialize(self.fmt, req.get_response(self.api)) - if expect_fail: - self.assertEqual('InvalidSubnetServiceType', + if fail_code is not None: + self.assertEqual(fail_code, res['NeutronError']['type']) else: subnet = res['subnet'] @@ -155,12 +156,14 @@ class SubnetServiceTypesExtensionTestCase( self._test_update_subnet(subnet, service_types) def test_update_subnet_invalid_type(self): - service_types = ['foo'] # Create a subnet with no service type res = self._create_service_subnet() subnet = self.deserialize('json', res)['subnet'] - # Update it with an invalid service type - self._test_update_subnet(subnet, service_types, expect_fail=True) + # Update it with invalid service type(s) + self._test_update_subnet(subnet, ['foo'], + fail_code='InvalidSubnetServiceType') + self._test_update_subnet(subnet, [2], + fail_code='InvalidInputSubnetServiceType') def _assert_port_res(self, port, service_type, subnet, fallback, error='IpAddressGenerationFailureNoMatchingSubnet'):