From b7ffbf96ff414ce615c10c83fea85d7f1fdcce70 Mon Sep 17 00:00:00 2001 From: Cedric Brandily Date: Sat, 30 May 2015 22:04:59 +0200 Subject: [PATCH] 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 --- .../plugins/ml2/drivers/base_type_tunnel.py | 48 +++++++++++++ .../unit/plugins/ml2/drivers/test_type_gre.py | 54 +++----------- .../plugins/ml2/drivers/test_type_vxlan.py | 71 ++++++------------- 3 files changed, 77 insertions(+), 96 deletions(-) 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 cd5469fd0..f239f9eaf 100644 --- 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 01f36ab46..c61f07b0e 100644 --- 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 b59f01180..8827fedb4 100644 --- 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') -- 2.45.2