diff --git a/neutron/agent/firewall.py b/neutron/agent/firewall.py index caeecb018b4..6000521a9ff 100644 --- a/neutron/agent/firewall.py +++ b/neutron/agent/firewall.py @@ -18,11 +18,12 @@ import contextlib import six +import neutron.common.constants as const from neutron.common import utils from neutron.extensions import portsecurity as psec -INGRESS_DIRECTION = 'ingress' -EGRESS_DIRECTION = 'egress' +INGRESS_DIRECTION = const.INGRESS_DIRECTION +EGRESS_DIRECTION = const.EGRESS_DIRECTION DIRECTION_IP_PREFIX = {INGRESS_DIRECTION: 'source_ip_prefix', EGRESS_DIRECTION: 'dest_ip_prefix'} diff --git a/neutron/common/constants.py b/neutron/common/constants.py index 410012ca360..1a2fe03cfff 100644 --- a/neutron/common/constants.py +++ b/neutron/common/constants.py @@ -215,6 +215,11 @@ AGENT_ALIVE = 'alive' # agent has just returned to alive after being dead AGENT_REVIVED = 'revived' +INGRESS_DIRECTION = 'ingress' +EGRESS_DIRECTION = 'egress' + +VALID_DIRECTIONS = (INGRESS_DIRECTION, EGRESS_DIRECTION) +VALID_ETHERTYPES = (lib_constants.IPv4, lib_constants.IPv6) # Neutron-lib migration shim. This will wrap any constants that are moved # to that library in a deprecation warning, until they can be updated to diff --git a/neutron/objects/common_types.py b/neutron/objects/common_types.py index 76404f08f37..f77a1f50cb5 100644 --- a/neutron/objects/common_types.py +++ b/neutron/objects/common_types.py @@ -72,3 +72,16 @@ class DscpMark(IntegerEnum): class DscpMarkField(obj_fields.AutoTypedField): AUTO_TYPE = DscpMark() + + +class FlowDirectionEnumField(obj_fields.AutoTypedField): + AUTO_TYPE = obj_fields.Enum(valid_values=constants.VALID_DIRECTIONS) + + +class EtherTypeEnumField(obj_fields.AutoTypedField): + AUTO_TYPE = obj_fields.Enum(valid_values=constants.VALID_ETHERTYPES) + + +class IpProtocolEnumField(obj_fields.AutoTypedField): + AUTO_TYPE = obj_fields.Enum( + valid_values=list(constants.IP_PROTOCOL_MAP.keys())) diff --git a/neutron/tests/unit/objects/test_common_types.py b/neutron/tests/unit/objects/test_common_types.py index 45d5b8d06bd..bdae27dce3d 100644 --- a/neutron/tests/unit/objects/test_common_types.py +++ b/neutron/tests/unit/objects/test_common_types.py @@ -79,3 +79,49 @@ class DscpMarkFieldTest(test_base.BaseTestCase, TestField): def test_stringify(self): for in_val, out_val in self.coerce_good_values: self.assertEqual("%s" % in_val, self.field.stringify(in_val)) + + +class FlowDirectionEnumFieldTest(test_base.BaseTestCase, TestField): + def setUp(self): + super(FlowDirectionEnumFieldTest, self).setUp() + self.field = common_types.FlowDirectionEnumField() + self.coerce_good_values = [(val, val) + for val in constants.VALID_DIRECTIONS] + self.coerce_bad_values = ['test', '8', 10, []] + self.to_primitive_values = self.coerce_good_values + self.from_primitive_values = self.coerce_good_values + + def test_stringify(self): + for in_val, out_val in self.coerce_good_values: + self.assertEqual("'%s'" % in_val, self.field.stringify(in_val)) + + +class EtherTypeEnumFieldTest(test_base.BaseTestCase, TestField): + def setUp(self): + super(EtherTypeEnumFieldTest, self).setUp() + self.field = common_types.EtherTypeEnumField() + self.coerce_good_values = [(val, val) + for val in constants.VALID_ETHERTYPES] + self.coerce_bad_values = ['IpV4', 8, 'str', 'ipv6'] + self.to_primitive_values = self.coerce_good_values + self.from_primitive_values = self.coerce_good_values + + def test_stringify(self): + for in_val, out_val in self.coerce_good_values: + self.assertEqual("'%s'" % in_val, self.field.stringify(in_val)) + + +class IpProtocolEnumFieldTest(test_base.BaseTestCase, TestField): + def setUp(self): + super(IpProtocolEnumFieldTest, self).setUp() + self.field = common_types.IpProtocolEnumField() + self.coerce_good_values = [(val, val) + for val in + list(constants.IP_PROTOCOL_MAP.keys())] + self.coerce_bad_values = ['test', '8', 10, 'Udp'] + self.to_primitive_values = self.coerce_good_values + self.from_primitive_values = self.coerce_good_values + + def test_stringify(self): + for in_val, out_val in self.coerce_good_values: + self.assertEqual("'%s'" % in_val, self.field.stringify(in_val))