Refactor type_gre.vxlan tests to reduce duplicate code

gre and vxlan type drivers unittests have similar tests. This change
abstracts these tests and moves them to TunnelTypeTestMixin[1].

[1] neutron.tests.unit.plugins.ml2.drivers.base_type_tunnel

Change-Id: I13b0507991e840ff34de7ed7ffd31d359691b0ca

diff --git a/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py b/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py
index cd5469fd001281db80e115fb56824308b877a8f4..f239f9eafcb8f051ca87bd91419b1f0a65041319 100644 (file)
--- a/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py
+++ b/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py
@@ -34,6 +34,7 @@ UPDATED_TUNNEL_RANGES = [(TUN_MIN + 5, TUN_MAX + 5)]
 
 
 class TunnelTypeTestMixin(object):
+    DRIVER_MODULE = None
     DRIVER_CLASS = None
     TYPE = None
 
@@ -196,6 +197,53 @@ class TunnelTypeTestMixin(object):
             segment[api.SEGMENTATION_ID] = tunnel_id
             self.driver.release_segment(self.session, segment)
 
+    def add_endpoint(self, ip=TUNNEL_IP_ONE, host=HOST_ONE):
+        return self.driver.add_endpoint(ip, host)
+
+    def test_add_endpoint(self):
+        endpoint = self.add_endpoint()
+        self.assertEqual(TUNNEL_IP_ONE, endpoint.ip_address)
+        self.assertEqual(HOST_ONE, endpoint.host)
+        return endpoint
+
+    def test_add_endpoint_for_existing_tunnel_ip(self):
+        self.add_endpoint()
+
+        log = getattr(self.DRIVER_MODULE, 'LOG')
+        with mock.patch.object(log, 'warning') as log_warn:
+            self.add_endpoint()
+            log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
+
+    def test_get_endpoint_by_host(self):
+        self.add_endpoint()
+
+        host_endpoint = self.driver.get_endpoint_by_host(HOST_ONE)
+        self.assertEqual(TUNNEL_IP_ONE, host_endpoint.ip_address)
+        return host_endpoint
+
+    def test_get_endpoint_by_host_for_not_existing_host(self):
+        ip_endpoint = self.driver.get_endpoint_by_host(HOST_TWO)
+        self.assertIsNone(ip_endpoint)
+
+    def test_get_endpoint_by_ip(self):
+        self.add_endpoint()
+
+        ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_ONE)
+        self.assertEqual(HOST_ONE, ip_endpoint.host)
+        return ip_endpoint
+
+    def test_get_endpoint_by_ip_for_not_existing_tunnel_ip(self):
+        ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_TWO)
+        self.assertIsNone(ip_endpoint)
+
+    def test_delete_endpoint(self):
+        self.add_endpoint()
+
+        self.assertIsNone(self.driver.delete_endpoint(TUNNEL_IP_ONE))
+        # Get all the endpoints and verify its empty
+        endpoints = self.driver.get_endpoints()
+        self.assertNotIn(TUNNEL_IP_ONE, endpoints)
+
 
 class TunnelTypeMultiRangeTestMixin(object):
     DRIVER_CLASS = None

diff --git a/neutron/tests/unit/plugins/ml2/drivers/test_type_gre.py b/neutron/tests/unit/plugins/ml2/drivers/test_type_gre.py
index 01f36ab464e14bf667b7915e803346db11982614..c61f07b0e63ed690e20a50608e57ac948585b128 100644 (file)
--- a/neutron/tests/unit/plugins/ml2/drivers/test_type_gre.py
+++ b/neutron/tests/unit/plugins/ml2/drivers/test_type_gre.py
@@ -46,59 +46,21 @@ def _get_allocation(session, gre_id):
 
 class GreTypeTest(base_type_tunnel.TunnelTypeTestMixin,
                   testlib_api.SqlTestCase):
+    DRIVER_MODULE = type_gre
     DRIVER_CLASS = type_gre.GreTypeDriver
     TYPE = p_const.TYPE_GRE
 
-    def test_add_endpoint(self):
-        endpoint = self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-        self.assertEqual(TUNNEL_IP_ONE, endpoint.ip_address)
-        self.assertEqual(HOST_ONE, endpoint.host)
-
-    def test_add_endpoint_for_existing_tunnel_ip(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-
-        with mock.patch.object(type_gre.LOG, 'warning') as log_warn:
-            self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-            log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
-
-    def test_get_endpoint_by_host(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-
-        host_endpoint = self.driver.get_endpoint_by_host(HOST_ONE)
-        self.assertEqual(TUNNEL_IP_ONE, host_endpoint.ip_address)
-
-    def test_get_endpoint_by_host_for_not_existing_host(self):
-        ip_endpoint = self.driver.get_endpoint_by_host(HOST_TWO)
-        self.assertIsNone(ip_endpoint)
-
-    def test_get_endpoint_by_ip(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-
-        ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_ONE)
-        self.assertEqual(HOST_ONE, ip_endpoint.host)
-
-    def test_get_endpoint_by_ip_for_not_existing_tunnel_ip(self):
-        ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_TWO)
-        self.assertIsNone(ip_endpoint)
-
     def test_get_endpoints(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-        self.driver.add_endpoint(TUNNEL_IP_TWO, HOST_TWO)
+        self.add_endpoint()
+        self.add_endpoint(
+            base_type_tunnel.TUNNEL_IP_TWO, base_type_tunnel.HOST_TWO)
 
         endpoints = self.driver.get_endpoints()
         for endpoint in endpoints:
-            if endpoint['ip_address'] == TUNNEL_IP_ONE:
-                self.assertEqual(HOST_ONE, endpoint['host'])
-            elif endpoint['ip_address'] == TUNNEL_IP_TWO:
-                self.assertEqual(HOST_TWO, endpoint['host'])
-
-    def test_delete_endpoint(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE)
-
-        self.assertIsNone(self.driver.delete_endpoint(TUNNEL_IP_ONE))
-        # Get all the endpoints and verify its empty
-        endpoints = self.driver.get_endpoints()
-        self.assertNotIn(TUNNEL_IP_ONE, endpoints)
+            if endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_ONE:
+                self.assertEqual(base_type_tunnel.HOST_ONE, endpoint['host'])
+            elif endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_TWO:
+                self.assertEqual(base_type_tunnel.HOST_TWO, endpoint['host'])
 
     def test_sync_allocations_entry_added_during_session(self):
         with mock.patch.object(self.driver, '_add_allocation',

diff --git a/neutron/tests/unit/plugins/ml2/drivers/test_type_vxlan.py b/neutron/tests/unit/plugins/ml2/drivers/test_type_vxlan.py
index b59f0118080d471e3a836387f61e74628dd29802..8827fedb4d8c0c63e04e5c0fdfa9bd6297d6e2f5 100644 (file)
--- a/neutron/tests/unit/plugins/ml2/drivers/test_type_vxlan.py
+++ b/neutron/tests/unit/plugins/ml2/drivers/test_type_vxlan.py
@@ -13,8 +13,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import mock
-
 from neutron.plugins.common import constants as p_const
 from neutron.plugins.ml2 import config
 from neutron.plugins.ml2.drivers import type_vxlan
@@ -23,76 +21,49 @@ from neutron.tests.unit.plugins.ml2 import test_rpc
 from neutron.tests.unit import testlib_api
 
 
-TUNNEL_IP_ONE = "10.10.10.10"
-TUNNEL_IP_TWO = "10.10.10.20"
-HOST_ONE = 'fake_host_one'
-HOST_TWO = 'fake_host_two'
 VXLAN_UDP_PORT_ONE = 9999
 VXLAN_UDP_PORT_TWO = 8888
 
 
 class VxlanTypeTest(base_type_tunnel.TunnelTypeTestMixin,
                     testlib_api.SqlTestCase):
+    DRIVER_MODULE = type_vxlan
     DRIVER_CLASS = type_vxlan.VxlanTypeDriver
     TYPE = p_const.TYPE_VXLAN
 
+    def add_endpoint(self, ip=base_type_tunnel.TUNNEL_IP_ONE,
+                     host=base_type_tunnel.HOST_ONE):
+        if ip == base_type_tunnel.TUNNEL_IP_ONE:
+            port = VXLAN_UDP_PORT_ONE
+        else:
+            port = VXLAN_UDP_PORT_TWO
+        return self.driver.add_endpoint(ip, host, port)
+
     def test_add_endpoint(self):
-        endpoint = self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE,
-                                            VXLAN_UDP_PORT_ONE)
-        self.assertEqual(TUNNEL_IP_ONE, endpoint.ip_address)
+        endpoint = super(VxlanTypeTest, self).test_add_endpoint()
         self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint.udp_port)
-        self.assertEqual(HOST_ONE, endpoint.host)
-
-    def test_add_endpoint_for_existing_tunnel_ip(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
-
-        with mock.patch.object(type_vxlan.LOG, 'warning') as log_warn:
-            self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE,
-                                     VXLAN_UDP_PORT_ONE)
-            log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
 
     def test_get_endpoint_by_host(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
-
-        host_endpoint = self.driver.get_endpoint_by_host(HOST_ONE)
-        self.assertEqual(TUNNEL_IP_ONE, host_endpoint.ip_address)
-        self.assertEqual(VXLAN_UDP_PORT_ONE, host_endpoint.udp_port)
-
-    def test_get_endpoint_by_host_for_not_existing_host(self):
-        ip_endpoint = self.driver.get_endpoint_by_host(HOST_TWO)
-        self.assertIsNone(ip_endpoint)
+        endpoint = super(VxlanTypeTest, self).test_get_endpoint_by_host()
+        self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint.udp_port)
 
     def test_get_endpoint_by_ip(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
-
-        ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_ONE)
-        self.assertEqual(HOST_ONE, ip_endpoint.host)
-        self.assertEqual(VXLAN_UDP_PORT_ONE, ip_endpoint.udp_port)
-
-    def test_get_endpoint_by_ip_for_not_existing_tunnel_ip(self):
-        ip_endpoint = self.driver.get_endpoint_by_ip(TUNNEL_IP_TWO)
-        self.assertIsNone(ip_endpoint)
+        endpoint = super(VxlanTypeTest, self).test_get_endpoint_by_ip()
+        self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint.udp_port)
 
     def test_get_endpoints(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
-        self.driver.add_endpoint(TUNNEL_IP_TWO, HOST_TWO, VXLAN_UDP_PORT_TWO)
+        self.add_endpoint()
+        self.add_endpoint(base_type_tunnel.TUNNEL_IP_TWO,
+                          base_type_tunnel.HOST_TWO)
 
         endpoints = self.driver.get_endpoints()
         for endpoint in endpoints:
-            if endpoint['ip_address'] == TUNNEL_IP_ONE:
+            if endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_ONE:
                 self.assertEqual(VXLAN_UDP_PORT_ONE, endpoint['udp_port'])
-                self.assertEqual(HOST_ONE, endpoint['host'])
-            elif endpoint['ip_address'] == TUNNEL_IP_TWO:
+                self.assertEqual(base_type_tunnel.HOST_ONE, endpoint['host'])
+            elif endpoint['ip_address'] == base_type_tunnel.TUNNEL_IP_TWO:
                 self.assertEqual(VXLAN_UDP_PORT_TWO, endpoint['udp_port'])
-                self.assertEqual(HOST_TWO, endpoint['host'])
-
-    def test_delete_endpoint(self):
-        self.driver.add_endpoint(TUNNEL_IP_ONE, HOST_ONE, VXLAN_UDP_PORT_ONE)
-
-        self.assertIsNone(self.driver.delete_endpoint(TUNNEL_IP_ONE))
-        # Get all the endpoints and verify its empty
-        endpoints = self.driver.get_endpoints()
-        self.assertNotIn(TUNNEL_IP_ONE, endpoints)
+                self.assertEqual(base_type_tunnel.HOST_TWO, endpoint['host'])
 
     def test_get_mtu(self):
         config.cfg.CONF.set_override('segment_mtu', 1500, group='ml2')