Refactor test_type_gre/vxlan to reduce duplicate code

gre and vxlan type drivers have similar implementation for multiple
methods and associated tests. This change abstracts these tests.

Change-Id: Ic0e2119a81540c509c214008cce9dc50f55dd6de

diff --git a/neutron/tests/unit/ml2/test_type_gre.py b/neutron/tests/unit/ml2/test_type_gre.py
index 98df5b98096c85b64a25bfcc10b57cdd64a4de3c..e753a16dc8722fd9334d880b2c2cd67da96b720e 100644 (file)
--- a/neutron/tests/unit/ml2/test_type_gre.py
+++ b/neutron/tests/unit/ml2/test_type_gre.py
@@ -14,183 +14,23 @@
 #    under the License.
 
 import mock
-from six import moves
-import testtools
-from testtools import matchers
 
-from neutron.common import exceptions as exc
-import neutron.db.api as db
-from neutron.plugins.ml2 import driver_api as api
+from neutron.plugins.common import constants as p_const
 from neutron.plugins.ml2.drivers import type_gre
+from neutron.tests.unit.ml2 import test_type_vxlan
 from neutron.tests.unit import testlib_api
 
+
 TUNNEL_IP_ONE = "10.10.10.10"
 TUNNEL_IP_TWO = "10.10.10.20"
-TUN_MIN = 100
-TUN_MAX = 109
-TUNNEL_RANGES = [(TUN_MIN, TUN_MAX)]
-UPDATED_TUNNEL_RANGES = [(TUN_MIN + 5, TUN_MAX + 5)]
-
-
-class GreTypeTest(testlib_api.SqlTestCase):
-
-    def setUp(self):
-        super(GreTypeTest, self).setUp()
-        self.driver = type_gre.GreTypeDriver()
-        self.driver.tunnel_ranges = TUNNEL_RANGES
-        self.driver.sync_allocations()
-        self.session = db.get_session()
-
-    def test_validate_provider_segment(self):
-        segment = {api.NETWORK_TYPE: 'gre',
-                   api.PHYSICAL_NETWORK: 'phys_net',
-                   api.SEGMENTATION_ID: None}
-
-        with testtools.ExpectedException(exc.InvalidInput):
-            self.driver.validate_provider_segment(segment)
-
-        segment[api.PHYSICAL_NETWORK] = None
-        self.driver.validate_provider_segment(segment)
-
-        segment[api.SEGMENTATION_ID] = 1
-        self.driver.validate_provider_segment(segment)
-
-    def test_sync_tunnel_allocations(self):
-        self.assertIsNone(
-            self.driver.get_allocation(self.session, (TUN_MIN - 1)))
-        self.assertFalse(
-            self.driver.get_allocation(self.session, (TUN_MIN)).allocated)
-        self.assertFalse(
-            self.driver.get_allocation(self.session, (TUN_MIN + 1)).allocated)
-        self.assertFalse(
-            self.driver.get_allocation(self.session, (TUN_MAX - 1)).allocated)
-        self.assertFalse(
-            self.driver.get_allocation(self.session, (TUN_MAX)).allocated)
-        self.assertIsNone(
-            self.driver.get_allocation(self.session, (TUN_MAX + 1)))
-
-        self.driver.tunnel_ranges = UPDATED_TUNNEL_RANGES
-        self.driver.sync_allocations()
-
-        self.assertIsNone(
-            self.driver.get_allocation(self.session, (TUN_MIN + 5 - 1)))
-        self.assertFalse(
-            self.driver.get_allocation(self.session, (TUN_MIN + 5)).allocated)
-        self.assertFalse(
-            self.driver.get_allocation(self.session,
-                                       (TUN_MIN + 5 + 1)).allocated)
-        self.assertFalse(
-            self.driver.get_allocation(self.session,
-                                       (TUN_MAX + 5 - 1)).allocated)
-        self.assertFalse(
-            self.driver.get_allocation(self.session, (TUN_MAX + 5)).allocated)
-        self.assertIsNone(
-            self.driver.get_allocation(self.session, (TUN_MAX + 5 + 1)))
-
-    def test_partial_segment_is_partial_segment(self):
-        segment = {api.NETWORK_TYPE: 'gre',
-                   api.PHYSICAL_NETWORK: None,
-                   api.SEGMENTATION_ID: None}
-        self.assertTrue(self.driver.is_partial_segment(segment))
-
-    def test_specific_segment_is_not_partial_segment(self):
-        segment = {api.NETWORK_TYPE: 'gre',
-                   api.PHYSICAL_NETWORK: None,
-                   api.SEGMENTATION_ID: 101}
-        self.assertFalse(self.driver.is_partial_segment(segment))
-
-    def test_reserve_provider_segment_full_specs(self):
-        segment = {api.NETWORK_TYPE: 'gre',
-                   api.PHYSICAL_NETWORK: None,
-                   api.SEGMENTATION_ID: 101}
-        observed = self.driver.reserve_provider_segment(self.session, segment)
-        alloc = self.driver.get_allocation(self.session,
-                                           observed[api.SEGMENTATION_ID])
-        self.assertTrue(alloc.allocated)
-
-        with testtools.ExpectedException(exc.TunnelIdInUse):
-            self.driver.reserve_provider_segment(self.session, segment)
 
-        self.driver.release_segment(self.session, segment)
-        alloc = self.driver.get_allocation(self.session,
-                                           observed[api.SEGMENTATION_ID])
-        self.assertFalse(alloc.allocated)
 
-        segment[api.SEGMENTATION_ID] = 1000
-        observed = self.driver.reserve_provider_segment(self.session, segment)
-        alloc = self.driver.get_allocation(self.session,
-                                           observed[api.SEGMENTATION_ID])
-        self.assertTrue(alloc.allocated)
+class GreTypeTest(test_type_vxlan.TunnelTypeTestMixin,
+                  testlib_api.SqlTestCase):
+    DRIVER_CLASS = type_gre.GreTypeDriver
+    TYPE = p_const.TYPE_GRE
 
-        self.driver.release_segment(self.session, segment)
-        alloc = self.driver.get_allocation(self.session,
-                                           observed[api.SEGMENTATION_ID])
-        self.assertIsNone(alloc)
-
-    def test_reserve_provider_segment(self):
-        tunnel_ids = set()
-        specs = {api.NETWORK_TYPE: 'gre',
-                 api.PHYSICAL_NETWORK: 'None',
-                 api.SEGMENTATION_ID: None}
-
-        for x in xrange(TUN_MIN, TUN_MAX + 1):
-            segment = self.driver.reserve_provider_segment(self.session,
-                                                           specs)
-            self.assertEqual('gre', segment[api.NETWORK_TYPE])
-            self.assertThat(segment[api.SEGMENTATION_ID],
-                            matchers.GreaterThan(TUN_MIN - 1))
-            self.assertThat(segment[api.SEGMENTATION_ID],
-                            matchers.LessThan(TUN_MAX + 1))
-            tunnel_ids.add(segment[api.SEGMENTATION_ID])
-
-        with testtools.ExpectedException(exc.NoNetworkAvailable):
-            segment = self.driver.reserve_provider_segment(self.session,
-                                                           specs)
-
-        segment = {api.NETWORK_TYPE: 'gre',
-                   api.PHYSICAL_NETWORK: 'None',
-                   api.SEGMENTATION_ID: tunnel_ids.pop()}
-        self.driver.release_segment(self.session, segment)
-        segment = self.driver.reserve_provider_segment(self.session, specs)
-        self.assertThat(segment[api.SEGMENTATION_ID],
-                        matchers.GreaterThan(TUN_MIN - 1))
-        self.assertThat(segment[api.SEGMENTATION_ID],
-                        matchers.LessThan(TUN_MAX + 1))
-        tunnel_ids.add(segment[api.SEGMENTATION_ID])
-
-        for tunnel_id in tunnel_ids:
-            segment[api.SEGMENTATION_ID] = tunnel_id
-            self.driver.release_segment(self.session, segment)
-
-    def test_allocate_tenant_segment(self):
-        tunnel_ids = set()
-        for x in moves.xrange(TUN_MIN, TUN_MAX + 1):
-            segment = self.driver.allocate_tenant_segment(self.session)
-            self.assertThat(segment[api.SEGMENTATION_ID],
-                            matchers.GreaterThan(TUN_MIN - 1))
-            self.assertThat(segment[api.SEGMENTATION_ID],
-                            matchers.LessThan(TUN_MAX + 1))
-            tunnel_ids.add(segment[api.SEGMENTATION_ID])
-
-        segment = self.driver.allocate_tenant_segment(self.session)
-        self.assertIsNone(segment)
-
-        segment = {api.NETWORK_TYPE: 'gre',
-                   api.PHYSICAL_NETWORK: 'None',
-                   api.SEGMENTATION_ID: tunnel_ids.pop()}
-        self.driver.release_segment(self.session, segment)
-        segment = self.driver.allocate_tenant_segment(self.session)
-        self.assertThat(segment[api.SEGMENTATION_ID],
-                        matchers.GreaterThan(TUN_MIN - 1))
-        self.assertThat(segment[api.SEGMENTATION_ID],
-                        matchers.LessThan(TUN_MAX + 1))
-        tunnel_ids.add(segment[api.SEGMENTATION_ID])
-
-        for tunnel_id in tunnel_ids:
-            segment[api.SEGMENTATION_ID] = tunnel_id
-            self.driver.release_segment(self.session, segment)
-
-    def test_gre_endpoints(self):
+    def test_endpoints(self):
         tun_1 = self.driver.add_endpoint(TUNNEL_IP_ONE)
         tun_2 = self.driver.add_endpoint(TUNNEL_IP_TWO)
         self.assertEqual(TUNNEL_IP_ONE, tun_1.ip_address)
@@ -202,37 +42,13 @@ class GreTypeTest(testlib_api.SqlTestCase):
             self.assertIn(endpoint['ip_address'],
                           [TUNNEL_IP_ONE, TUNNEL_IP_TWO])
 
-    def test_add_same_endpoints(self):
+    def test_add_same_ip_endpoints(self):
         self.driver.add_endpoint(TUNNEL_IP_ONE)
         with mock.patch.object(type_gre.LOG, 'warning') as log_warn:
             self.driver.add_endpoint(TUNNEL_IP_ONE)
         log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
 
 
-class GreTypeMultiRangeTest(testlib_api.SqlTestCase):
-
-    TUN_MIN0 = 100
-    TUN_MAX0 = 101
-    TUN_MIN1 = 200
-    TUN_MAX1 = 201
-    TUNNEL_MULTI_RANGES = [(TUN_MIN0, TUN_MAX0), (TUN_MIN1, TUN_MAX1)]
-
-    def setUp(self):
-        super(GreTypeMultiRangeTest, self).setUp()
-        self.driver = type_gre.GreTypeDriver()
-        self.driver.tunnel_ranges = self.TUNNEL_MULTI_RANGES
-        self.driver.sync_allocations()
-        self.session = db.get_session()
-
-    def test_release_segment(self):
-        segments = [self.driver.allocate_tenant_segment(self.session)
-                    for i in range(4)]
-
-        # Release them in random order. No special meaning.
-        for i in (0, 2, 1, 3):
-            self.driver.release_segment(self.session, segments[i])
-
-        for key in (self.TUN_MIN0, self.TUN_MAX0,
-                    self.TUN_MIN1, self.TUN_MAX1):
-            alloc = self.driver.get_allocation(self.session, key)
-            self.assertFalse(alloc.allocated)
+class GreTypeMultiRangeTest(test_type_vxlan.TunnelTypeMultiRangeTestMixin,
+                           testlib_api.SqlTestCase):
+    DRIVER_CLASS = type_gre.GreTypeDriver

diff --git a/neutron/tests/unit/ml2/test_type_vxlan.py b/neutron/tests/unit/ml2/test_type_vxlan.py
index ab8d4409c64a2c630d6ca4c3540b4f84f124401f..a2f1274576ae85e67521c0e835b87664c6763240 100644 (file)
--- a/neutron/tests/unit/ml2/test_type_vxlan.py
+++ b/neutron/tests/unit/ml2/test_type_vxlan.py
@@ -38,19 +38,22 @@ VXLAN_UDP_PORT_ONE = 9999
 VXLAN_UDP_PORT_TWO = 8888
 
 
-class VxlanTypeTest(testlib_api.SqlTestCase):
+class TunnelTypeTestMixin(object):
+    DRIVER_CLASS = None
+    TYPE = None
+
     def setUp(self):
-        super(VxlanTypeTest, self).setUp()
-        self.driver = type_vxlan.VxlanTypeDriver()
+        super(TunnelTypeTestMixin, self).setUp()
+        self.driver = self.DRIVER_CLASS()
         self.driver.tunnel_ranges = TUNNEL_RANGES
         self.driver.sync_allocations()
         self.session = db.get_session()
 
-    def test_vxlan_tunnel_type(self):
-        self.assertEqual(self.driver.get_type(), p_const.TYPE_VXLAN)
+    def test_tunnel_type(self):
+        self.assertEqual(self.TYPE, self.driver.get_type())
 
     def test_validate_provider_segment(self):
-        segment = {api.NETWORK_TYPE: 'vxlan',
+        segment = {api.NETWORK_TYPE: self.TYPE,
                    api.PHYSICAL_NETWORK: 'phys_net',
                    api.SEGMENTATION_ID: None}
 
@@ -96,19 +99,19 @@ class VxlanTypeTest(testlib_api.SqlTestCase):
             self.driver.get_allocation(self.session, (TUN_MAX + 5 + 1)))
 
     def test_partial_segment_is_partial_segment(self):
-        segment = {api.NETWORK_TYPE: 'vxlan',
+        segment = {api.NETWORK_TYPE: self.TYPE,
                    api.PHYSICAL_NETWORK: None,
                    api.SEGMENTATION_ID: None}
         self.assertTrue(self.driver.is_partial_segment(segment))
 
     def test_specific_segment_is_not_partial_segment(self):
-        segment = {api.NETWORK_TYPE: 'vxlan',
+        segment = {api.NETWORK_TYPE: self.TYPE,
                    api.PHYSICAL_NETWORK: None,
                    api.SEGMENTATION_ID: 101}
         self.assertFalse(self.driver.is_partial_segment(segment))
 
     def test_reserve_provider_segment_full_specs(self):
-        segment = {api.NETWORK_TYPE: 'vxlan',
+        segment = {api.NETWORK_TYPE: self.TYPE,
                    api.PHYSICAL_NETWORK: None,
                    api.SEGMENTATION_ID: 101}
         observed = self.driver.reserve_provider_segment(self.session, segment)
@@ -137,14 +140,14 @@ class VxlanTypeTest(testlib_api.SqlTestCase):
 
     def test_reserve_provider_segment(self):
         tunnel_ids = set()
-        specs = {api.NETWORK_TYPE: 'vxlan',
+        specs = {api.NETWORK_TYPE: self.TYPE,
                  api.PHYSICAL_NETWORK: 'None',
                  api.SEGMENTATION_ID: None}
 
-        for x in xrange(TUN_MIN, TUN_MAX + 1):
+        for x in moves.xrange(TUN_MIN, TUN_MAX + 1):
             segment = self.driver.reserve_provider_segment(self.session,
                                                            specs)
-            self.assertEqual('vxlan', segment[api.NETWORK_TYPE])
+            self.assertEqual(self.TYPE, segment[api.NETWORK_TYPE])
             self.assertThat(segment[api.SEGMENTATION_ID],
                             matchers.GreaterThan(TUN_MIN - 1))
             self.assertThat(segment[api.SEGMENTATION_ID],
@@ -155,7 +158,7 @@ class VxlanTypeTest(testlib_api.SqlTestCase):
             segment = self.driver.reserve_provider_segment(self.session,
                                                            specs)
 
-        segment = {api.NETWORK_TYPE: 'vxlan',
+        segment = {api.NETWORK_TYPE: self.TYPE,
                    api.PHYSICAL_NETWORK: 'None',
                    api.SEGMENTATION_ID: tunnel_ids.pop()}
         self.driver.release_segment(self.session, segment)
@@ -183,7 +186,7 @@ class VxlanTypeTest(testlib_api.SqlTestCase):
         segment = self.driver.allocate_tenant_segment(self.session)
         self.assertIsNone(segment)
 
-        segment = {api.NETWORK_TYPE: 'vxlan',
+        segment = {api.NETWORK_TYPE: self.TYPE,
                    api.PHYSICAL_NETWORK: 'None',
                    api.SEGMENTATION_ID: tunnel_ids.pop()}
         self.driver.release_segment(self.session, segment)
@@ -198,9 +201,12 @@ class VxlanTypeTest(testlib_api.SqlTestCase):
             segment[api.SEGMENTATION_ID] = tunnel_id
             self.driver.release_segment(self.session, segment)
 
-    def test_vxlan_endpoints(self):
-        """Test VXLAN allocation/de-allocation."""
 
+class VxlanTypeTest(TunnelTypeTestMixin, testlib_api.SqlTestCase):
+    DRIVER_CLASS = type_vxlan.VxlanTypeDriver
+    TYPE = p_const.TYPE_VXLAN
+
+    def test_endpoints(self):
         # Set first endpoint, verify it gets VXLAN VNI 1
         vxlan1_endpoint = self.driver.add_endpoint(TUNNEL_IP_ONE,
                                                    VXLAN_UDP_PORT_ONE)
@@ -230,7 +236,8 @@ class VxlanTypeTest(testlib_api.SqlTestCase):
             log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
 
 
-class VxlanTypeMultiRangeTest(testlib_api.SqlTestCase):
+class TunnelTypeMultiRangeTestMixin(object):
+    DRIVER_CLASS = None
 
     TUN_MIN0 = 100
     TUN_MAX0 = 101
@@ -239,8 +246,8 @@ class VxlanTypeMultiRangeTest(testlib_api.SqlTestCase):
     TUNNEL_MULTI_RANGES = [(TUN_MIN0, TUN_MAX0), (TUN_MIN1, TUN_MAX1)]
 
     def setUp(self):
-        super(VxlanTypeMultiRangeTest, self).setUp()
-        self.driver = type_vxlan.VxlanTypeDriver()
+        super(TunnelTypeMultiRangeTestMixin, self).setUp()
+        self.driver = self.DRIVER_CLASS()
         self.driver.tunnel_ranges = self.TUNNEL_MULTI_RANGES
         self.driver.sync_allocations()
         self.session = db.get_session()
@@ -257,3 +264,8 @@ class VxlanTypeMultiRangeTest(testlib_api.SqlTestCase):
                     self.TUN_MIN1, self.TUN_MAX1):
             alloc = self.driver.get_allocation(self.session, key)
             self.assertFalse(alloc.allocated)
+
+
+class VxlanTypeMultiRangeTest(TunnelTypeMultiRangeTestMixin,
+                              testlib_api.SqlTestCase):
+    DRIVER_CLASS = type_vxlan.VxlanTypeDriver