From 78bc42dc098da0068feb4f7c3ceea21be7192275 Mon Sep 17 00:00:00 2001 From: Abhishek Raut Date: Tue, 18 Aug 2015 06:50:15 -0700 Subject: [PATCH] NSXv3: Add more unit tests for layer 2 gateway support Change-Id: I7611117b39f68001526c3a1b18d0709f96a6ffa9 Partial-bug: #1481087 --- vmware_nsx/tests/unit/vmware/nsx_v3_mocks.py | 11 ++ .../services/l2gateway/test_nsxv3_driver.py | 173 +++++++++++++++++- 2 files changed, 175 insertions(+), 9 deletions(-) diff --git a/vmware_nsx/tests/unit/vmware/nsx_v3_mocks.py b/vmware_nsx/tests/unit/vmware/nsx_v3_mocks.py index a1fbc7b88f..b246caa5a5 100644 --- a/vmware_nsx/tests/unit/vmware/nsx_v3_mocks.py +++ b/vmware_nsx/tests/unit/vmware/nsx_v3_mocks.py @@ -205,6 +205,17 @@ def delete_resource(resource): pass +def create_bridge_endpoint(device_name, seg_id, tags): + FAKE_BE = { + "id": uuidutils.generate_uuid(), + "display_name": FAKE_NAME, + "resource_type": "BridgeEndpoint", + "bridge_endpoint_id": device_name, + "vlan": seg_id, + } + return FAKE_BE + + class NsxV3Mock(object): def __init__(self, default_tier0_router_uuid=DEFAULT_TIER0_ROUTER_UUID): self.logical_routers = {} diff --git a/vmware_nsx/tests/unit/vmware/services/l2gateway/test_nsxv3_driver.py b/vmware_nsx/tests/unit/vmware/services/l2gateway/test_nsxv3_driver.py index df29947a28..f3327dbd6d 100644 --- a/vmware_nsx/tests/unit/vmware/services/l2gateway/test_nsxv3_driver.py +++ b/vmware_nsx/tests/unit/vmware/services/l2gateway/test_nsxv3_driver.py @@ -14,32 +14,44 @@ # limitations under the License. import mock +from networking_l2gw.services.l2gateway.common import constants +from networking_l2gw.services.l2gateway import exceptions as l2gw_exc +from networking_l2gw.tests.unit.db import test_l2gw_db +from oslo_config import cfg +from oslo_utils import importutils +from oslo_utils import uuidutils + +from neutron.common import exceptions as n_exc from neutron import context from neutron.tests import base -from networking_l2gw.tests.unit.db import test_l2gw_db - +from vmware_nsx.common import nsx_constants +from vmware_nsx.nsxlib import v3 as nsxlib from vmware_nsx.services.l2gateway.common import plugin as l2gw_plugin from vmware_nsx.services.l2gateway.nsx_v3 import driver as nsx_v3_driver +from vmware_nsx.tests.unit.vmware import nsx_v3_mocks +from vmware_nsx.tests.unit.vmware import test_nsx_v3_plugin -from oslo_config import cfg -from oslo_utils import uuidutils - - -NSX_V3_DRIVER_CLASS_PATH = ( - 'vmware_nsx.services.l2gateway.nsx_v3.driver.NsxV3Driver') +NSX_V3_PLUGIN_CLASS = ('vmware_nsx.plugins.nsx_v3.plugin.NsxV3Plugin') +NSX_V3_L2GW_DRIVER_CLASS_PATH = ('vmware_nsx.services.l2gateway.' + 'nsx_v3.driver.NsxV3Driver') class TestNsxV3L2GatewayDriver(test_l2gw_db.L2GWTestCase, + test_nsx_v3_plugin.NsxPluginV3TestCase, base.BaseTestCase): def setUp(self): super(TestNsxV3L2GatewayDriver, self).setUp() cfg.CONF.set_override("nsx_l2gw_driver", - NSX_V3_DRIVER_CLASS_PATH, 'NSX') + NSX_V3_L2GW_DRIVER_CLASS_PATH, 'NSX') + self.core_plugin = importutils.import_object(NSX_V3_PLUGIN_CLASS) self.l2gw_plugin = l2gw_plugin.NsxL2GatewayPlugin() self.driver = nsx_v3_driver.NsxV3Driver() self.context = context.get_admin_context() + # Mock NSX lib backend calls + nsxlib.create_bridge_endpoint = nsx_v3_mocks.create_bridge_endpoint + nsxlib.delete_bridge_endpoint = mock.Mock() def test_nsxl2gw_driver_init(self): with mock.patch.object(nsx_v3_driver.NsxV3Driver, @@ -72,3 +84,146 @@ class TestNsxV3L2GatewayDriver(test_l2gw_db.L2GWTestCase, def_bridge_cluster_id) self.assertTrue(def_l2gw.devices[0].interfaces[0].interface_name, 'default-bridge-cluster') + + def test_create_duplicate_default_l2_gateway_noop(self): + def_bridge_cluster_id = uuidutils.generate_uuid() + with mock.patch.object(nsx_v3_driver.NsxV3Driver, + 'subscribe_callback_notifications'): + cfg.CONF.set_override("default_bridge_cluster_uuid", + def_bridge_cluster_id, + "nsx_v3") + l2gw_plugin.NsxL2GatewayPlugin() + l2gw_plugin.NsxL2GatewayPlugin() + l2gws = self.driver._get_l2_gateways(self.context) + # Verify whether only one default L2 gateway is created + self.assertEqual(1, len(l2gws)) + + def test_create_default_l2_gateway_no_bc_uuid_noop(self): + with mock.patch.object(nsx_v3_driver.NsxV3Driver, + 'subscribe_callback_notifications'): + l2gw_plugin.NsxL2GatewayPlugin() + l2gws = self.driver._get_l2_gateways(self.context) + # Verify no default L2 gateway is created if bridge cluster id is + # not configured in nsx.ini + self.assertEqual([], l2gws) + + def test_create_l2_gateway_multiple_devices_fail(self): + invalid_l2gw_dict = { + "l2_gateway": { + "tenant_id": "fake_tenant_id", + "name": "invalid_l2gw", + "devices": [{"interfaces": + [{"name": "interface1"}], + "device_name": "device1"}, + {"interfaces": + [{"name": "interface_2"}], + "device_name": "device2"}]}} + self.assertRaises(n_exc.InvalidInput, + self.driver.create_l2_gateway, + self.context, invalid_l2gw_dict) + + def test_create_l2_gateway_invalid_device_name_fail(self): + invalid_l2gw_dict = { + "l2_gateway": { + "tenant_id": "fake_tenant_id", + "name": "invalid_l2gw", + "devices": [{"interfaces": + [{"name": "interface_1"}], + "device_name": "device-1"}]}} + self.assertRaises(n_exc.InvalidInput, + self.driver.create_l2_gateway, + self.context, invalid_l2gw_dict) + + def test_create_l2_gateway_valid(self): + bc_uuid = uuidutils.generate_uuid() + l2gw_data = self._get_l2_gateway_data(name='gw1', + device_name=bc_uuid) + l2gw = self.driver.create_l2_gateway(self.context, l2gw_data) + self.assertIsNotNone(l2gw) + self.assertEqual("gw1", l2gw["name"]) + self.assertEqual("port1", + l2gw["devices"][0]["interfaces"][0]["name"]) + self.assertEqual(bc_uuid, l2gw["devices"][0]["device_name"]) + + def test_create_l2_gateway_connection(self): + type(self.driver)._core_plugin = self.core_plugin + bc_uuid = uuidutils.generate_uuid() + l2gw_data = self._get_l2_gateway_data(name='def-l2gw', + device_name=bc_uuid) + l2gw = self._create_l2gateway(l2gw_data) + net_data = self._get_nw_data() + net = self.core_plugin.create_network(self.context, net_data) + l2gw_conn_data = {constants.CONNECTION_RESOURCE_NAME: { + 'l2_gateway_id': l2gw['id'], + 'tenant_id': 'fake_tenant_id', + 'network_id': net['id']}} + l2gw_conn = self.driver.create_l2_gateway_connection(self.context, + l2gw_conn_data) + self.assertIsNotNone(l2gw_conn) + self.assertEqual(net['id'], l2gw_conn['network_id']) + self.assertEqual(l2gw['id'], l2gw_conn['l2_gateway_id']) + + def test_delete_l2_gateway_connection(self): + type(self.driver)._core_plugin = self.core_plugin + bc_uuid = uuidutils.generate_uuid() + l2gw_data = self._get_l2_gateway_data(name='def-l2gw', + device_name=bc_uuid) + l2gw = self._create_l2gateway(l2gw_data) + net_data = self._get_nw_data() + net = self.core_plugin.create_network(self.context, net_data) + l2gw_conn_data = {constants.CONNECTION_RESOURCE_NAME: { + 'l2_gateway_id': l2gw['id'], + 'tenant_id': 'fake_tenant_id', + 'network_id': net['id']}} + l2gw_conn = self.driver.create_l2_gateway_connection(self.context, + l2gw_conn_data) + self.driver.delete_l2_gateway_connection(self.context, + l2gw_conn['id']) + # Verify that the L2 gateway connection was deleted + self.assertRaises(l2gw_exc.L2GatewayConnectionNotFound, + self.driver.get_l2_gateway_connection, + self.context, l2gw_conn['id']) + ports = self.core_plugin.get_ports(self.context) + # Verify that the L2 gateway connection port was cleaned up + self.assertEqual(0, len(ports)) + + def test_create_l2_gateway_connection_creates_port(self): + type(self.driver)._core_plugin = self.core_plugin + bc_uuid = uuidutils.generate_uuid() + l2gw_data = self._get_l2_gateway_data(name='def-l2gw', + device_name=bc_uuid) + l2gw = self._create_l2gateway(l2gw_data) + net_data = self._get_nw_data() + net = self.core_plugin.create_network(self.context, net_data) + l2gw_conn_data = {constants.CONNECTION_RESOURCE_NAME: { + 'l2_gateway_id': l2gw['id'], + 'tenant_id': 'fake_tenant_id', + 'network_id': net['id']}} + self.driver.create_l2_gateway_connection(self.context, l2gw_conn_data) + ports = self.core_plugin.get_ports(self.context) + # Verify that the L2 gateway connection port was created with device + # owner BRIDGEENDPOINT + self.assertEqual(1, len(ports)) + port = ports[0] + self.assertEqual(nsx_constants.BRIDGE_ENDPOINT, port['device_owner']) + # Verify that the L2 gateway connection port was created with no + # fixed ips + self.assertEqual(0, len(port.get('fixed_ips'))) + + def test_core_plugin_delete_l2_gateway_connection_port_fail(self): + type(self.driver)._core_plugin = self.core_plugin + bc_uuid = uuidutils.generate_uuid() + l2gw_data = self._get_l2_gateway_data(name='def-l2gw', + device_name=bc_uuid) + l2gw = self._create_l2gateway(l2gw_data) + net_data = self._get_nw_data() + net = self.core_plugin.create_network(self.context, net_data) + l2gw_conn_data = {constants.CONNECTION_RESOURCE_NAME: { + 'l2_gateway_id': l2gw['id'], + 'tenant_id': 'fake_tenant_id', + 'network_id': net['id']}} + self.driver.create_l2_gateway_connection(self.context, l2gw_conn_data) + port = self.core_plugin.get_ports(self.context)[0] + self.assertRaises(n_exc.ServicePortInUse, + self.core_plugin.delete_port, + self.context, port['id'])