35bf6d16b8
this patch adds the port security and address pairs extensions to the nsx v3 plugin. the patch also includes the core neutron unit tests for those extensions. NB: I did a good bit of manual testing with this code and didn't find any problems. however I'm not fully vested on all the other impl details so would be good to get some burn-in on this code by others. Change-Id: Iaca19428515a25ac6cf433c65cc81968e3175768
237 lines
11 KiB
Python
237 lines
11 KiB
Python
# Copyright (c) 2015 VMware, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# 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 vmware_nsx.common import nsx_constants
|
|
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.nsx_v3 import test_plugin as test_nsx_v3_plugin
|
|
|
|
|
|
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.NsxV3PluginTestCaseMixin,
|
|
base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestNsxV3L2GatewayDriver, self).setUp()
|
|
cfg.CONF.set_override("nsx_l2gw_driver",
|
|
NSX_V3_L2GW_DRIVER_CLASS_PATH)
|
|
|
|
self.core_plugin = importutils.import_object(NSX_V3_PLUGIN_CLASS)
|
|
|
|
self.core_plugin._nsx_client = self.client
|
|
self.core_plugin._port_client._client._session = self.mock_api
|
|
|
|
self.maxDiff = None
|
|
self.driver = nsx_v3_driver.NsxV3Driver()
|
|
self.l2gw_plugin = l2gw_plugin.NsxL2GatewayPlugin()
|
|
self.context = context.get_admin_context()
|
|
|
|
def _get_nw_data(self):
|
|
net_data = super(TestNsxV3L2GatewayDriver, self)._get_nw_data()
|
|
net_data['network']['port_security_enabled'] = True
|
|
return net_data
|
|
|
|
def test_nsxl2gw_driver_init(self):
|
|
with mock.patch.object(nsx_v3_driver.NsxV3Driver,
|
|
'_ensure_default_l2_gateway') as def_gw:
|
|
with mock.patch.object(nsx_v3_driver.NsxV3Driver,
|
|
'subscribe_callback_notifications') as sub:
|
|
with mock.patch.object(nsx_v3_driver.LOG,
|
|
'debug') as debug:
|
|
l2gw_plugin.NsxL2GatewayPlugin()
|
|
self.assertTrue(def_gw.called)
|
|
self.assertTrue(sub.called)
|
|
self.assertTrue(debug.called)
|
|
|
|
def test_create_default_l2_gateway(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()
|
|
l2gws = self.driver._get_l2_gateways(self.context)
|
|
def_l2gw = None
|
|
for l2gw in l2gws:
|
|
for device in l2gw['devices']:
|
|
if device['device_name'] == def_bridge_cluster_id:
|
|
def_l2gw = l2gw
|
|
self.assertIsNotNone(def_l2gw)
|
|
self.assertTrue(def_l2gw.devices[0].device_name,
|
|
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'])
|