Merge "Adds Enum fields for different types"

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'}

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

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()))

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))