From 190d0691b5fca5a0666e46f6dad98d6386b7dd29 Mon Sep 17 00:00:00 2001 From: Cedric Brandily Date: Tue, 19 Aug 2014 19:15:12 +0200 Subject: [PATCH] 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 --- neutron/tests/unit/ml2/test_type_gre.py | 208 ++-------------------- neutron/tests/unit/ml2/test_type_vxlan.py | 50 ++++-- 2 files changed, 43 insertions(+), 215 deletions(-) diff --git a/neutron/tests/unit/ml2/test_type_gre.py b/neutron/tests/unit/ml2/test_type_gre.py index 98df5b98096..e753a16dc87 100644 --- 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): +class GreTypeTest(test_type_vxlan.TunnelTypeTestMixin, + testlib_api.SqlTestCase): + DRIVER_CLASS = type_gre.GreTypeDriver + TYPE = p_const.TYPE_GRE - 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) - - 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 ab8d4409c64..a2f1274576a 100644 --- 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