NSX-TVD basic unittests
Adding unitests the to NSX-TVD plugin, to make sure the correct sub plugin is being called, all plugins are initialized, and plugins seperation is enforced. Also fixing some problems discovered during those tests. Change-Id: I76392451b3542f6682dfd594992f0d1c0e550ac0
This commit is contained in:
Adit Sarfaty
parent
1f8492abd9
commit
a4e40114dd
|
|
@ -357,7 +357,8 @@ class ExtendedSecurityGroupPropertiesMixin(object):
|
|||
# later we will remove those from the regular sg list
|
||||
provider_groups = []
|
||||
for sec_group_mapping in port_db.security_groups:
|
||||
if sec_group_mapping.extended_grp.provider is True:
|
||||
if (sec_group_mapping.extended_grp and
|
||||
sec_group_mapping.extended_grp.provider is True):
|
||||
provider_groups.append(sec_group_mapping['security_group_id'])
|
||||
port_res[provider_sg.PROVIDER_SECURITYGROUPS] = provider_groups
|
||||
return port_res
|
||||
|
|
|
|||
|
|
@ -169,15 +169,17 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
self._unsupported_fields[plugin_type] = {'router': [],
|
||||
'port': []}
|
||||
|
||||
# router size and type are supported only by the V plugin
|
||||
self._unsupported_fields[t.NsxV3Plugin.plugin_type()]['router'] = [
|
||||
'router_size', 'router_type']
|
||||
self._unsupported_fields[dvs.NsxDvsV2.plugin_type()]['router'] = [
|
||||
'router_size', 'router_type']
|
||||
# router size and type are supported only by the V plugin
|
||||
if plugin_type in [t.NsxV3Plugin.plugin_type(),
|
||||
dvs.NsxDvsV2.plugin_type()]:
|
||||
self._unsupported_fields[plugin_type]['router'] = [
|
||||
'router_size', 'router_type']
|
||||
|
||||
# port mac learning is not supported by the dvs plugin
|
||||
self._unsupported_fields[dvs.NsxDvsV2.plugin_type()]['port'] = [
|
||||
'mac_learning_enabled']
|
||||
# port mac learning, and provider sg are not supported by
|
||||
# the dvs plugin
|
||||
if plugin_type in [dvs.NsxDvsV2.plugin_type()]:
|
||||
self._unsupported_fields[plugin_type]['port'] = [
|
||||
'mac_learning_enabled', 'provider_security_groups']
|
||||
|
||||
def _validate_obj_extensions(self, data, plugin_type, obj_type):
|
||||
"""prevent configuration of unsupported extensions"""
|
||||
|
|
@ -209,7 +211,7 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
def _get_plugin_from_net_id(self, context, net_id):
|
||||
# get the network using the super plugin - here we use the
|
||||
# _get_network (so as not to call the make dict method)
|
||||
network = super(NsxTVDPlugin, self)._get_network(context, net_id)
|
||||
network = self._get_network(context, net_id)
|
||||
return self._get_plugin_from_project(context, network['tenant_id'])
|
||||
|
||||
def get_network_availability_zones(self, net_db):
|
||||
|
|
@ -260,7 +262,6 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
new_port = p.create_port(context, port)
|
||||
self._cleanup_obj_fields(
|
||||
new_port, p.plugin_type(), 'port')
|
||||
LOG.error("DEBUG ADIT created new_port %s", new_port)
|
||||
return new_port
|
||||
|
||||
def update_port(self, context, id, port):
|
||||
|
|
@ -365,10 +366,10 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
interface_info):
|
||||
is_port, is_sub = self._validate_interface_info(interface_info)
|
||||
if is_port:
|
||||
net_id = self.get_port(
|
||||
net_id = self._get_port(
|
||||
context, interface_info['port_id'])['network_id']
|
||||
elif is_sub:
|
||||
net_id = self.get_subnet(
|
||||
net_id = self._get_subnet(
|
||||
context, interface_info['subnet_id'])['network_id']
|
||||
net_plugin = self._get_plugin_from_net_id(context, net_id)
|
||||
if net_plugin.plugin_type() != router_plugin.plugin_type():
|
||||
|
|
@ -379,7 +380,7 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
def _get_plugin_from_router_id(self, context, router_id):
|
||||
# get the router using the super plugin - here we use the
|
||||
# _get_router (so as not to call the make dict method)
|
||||
router = super(NsxTVDPlugin, self)._get_router(context, router_id)
|
||||
router = self._get_router(context, router_id)
|
||||
return self._get_plugin_from_project(context, router['tenant_id'])
|
||||
|
||||
def create_router(self, context, router):
|
||||
|
|
@ -409,6 +410,10 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
self._cleanup_obj_fields(router, p.plugin_type(), 'router')
|
||||
return router
|
||||
|
||||
def delete_router(self, context, id):
|
||||
p = self._get_plugin_from_router_id(context, id)
|
||||
p.delete_router(context, id)
|
||||
|
||||
def add_router_interface(self, context, router_id, interface_info):
|
||||
p = self._get_plugin_from_router_id(context, router_id)
|
||||
self._validate_router_interface_plugin(context, p, interface_info)
|
||||
|
|
@ -447,6 +452,12 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
p = self._get_plugin_from_net_id(context, net_id)
|
||||
return p.delete_floatingip(context, id)
|
||||
|
||||
def get_floatingip(self, context, id):
|
||||
fip = self._get_floatingip(context, id)
|
||||
net_id = fip['floating_network_id']
|
||||
p = self._get_plugin_from_net_id(context, net_id)
|
||||
return p.get_floatingip(context, id)
|
||||
|
||||
def disassociate_floatingips(self, context, port_id):
|
||||
db_port = self._get_port(context, port_id)
|
||||
p = self._get_plugin_from_net_id(context, db_port['network_id'])
|
||||
|
|
@ -455,10 +466,9 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
def _get_plugin_from_sg_id(self, context, sg_id):
|
||||
# get the router using the super plugin - here we use the
|
||||
# _get_router (so as not to call the make dict method)
|
||||
sg = super(NsxTVDPlugin, self)._get_security_group(context, sg_id)
|
||||
sg = self._get_security_group(context, sg_id)
|
||||
return self._get_plugin_from_project(context, sg['tenant_id'])
|
||||
|
||||
# TODO(asarfaty): no need to create on both any more?
|
||||
def create_security_group(self, context, security_group,
|
||||
default_sg=False):
|
||||
if not default_sg:
|
||||
|
|
@ -478,6 +488,10 @@ class NsxTVDPlugin(addr_pair_db.AllowedAddressPairsMixin,
|
|||
p = self._get_plugin_from_sg_id(context, id)
|
||||
return p.update_security_group(context, id, security_group)
|
||||
|
||||
def get_security_group(self, context, id):
|
||||
p = self._get_plugin_from_sg_id(context, id)
|
||||
return p.get_security_group(context, id)
|
||||
|
||||
def create_security_group_rule_bulk(self, context, security_group_rules):
|
||||
p = self._get_plugin_from_project(context, context.project_id)
|
||||
return p.create_security_group_rule_bulk(context,
|
||||
|
|
|
|||
|
|
@ -4096,15 +4096,6 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
|
|||
return super(NsxV3Plugin, self)._ensure_default_security_group(
|
||||
context, tenant_id)
|
||||
|
||||
def get_security_groups(self, context, filters=None, fields=None,
|
||||
sorts=None, limit=None,
|
||||
marker=None, page_reverse=False, default_sg=False):
|
||||
return super(NsxV3Plugin, self).get_security_groups(
|
||||
context, filters=filters, fields=fields,
|
||||
sorts=sorts, limit=limit,
|
||||
marker=marker, page_reverse=page_reverse,
|
||||
default_sg=default_sg)
|
||||
|
||||
def _create_fw_section_for_secgroup(self, nsgroup, is_provider):
|
||||
# NOTE(arosen): if a security group is provider we want to
|
||||
# insert our rules at the top.
|
||||
|
|
|
|||
|
|
@ -132,7 +132,7 @@ class DvsTestCase(base.BaseTestCase):
|
|||
fake_get_spec.assert_called_once_with(net_id, vlan, trunk_mode=False)
|
||||
|
||||
|
||||
class NeutronSimpleDvsTest(test_plugin.NeutronDbPluginV2TestCase):
|
||||
class NeutronSimpleDvsTestCase(test_plugin.NeutronDbPluginV2TestCase):
|
||||
|
||||
@mock.patch.object(dvs_utils, 'dvs_create_session',
|
||||
return_value=fake_session())
|
||||
|
|
@ -147,9 +147,12 @@ class NeutronSimpleDvsTest(test_plugin.NeutronDbPluginV2TestCase):
|
|||
cfg.CONF.set_override('host_username', 'fake_user', group='dvs')
|
||||
cfg.CONF.set_override('host_password', 'fake_password', group='dvs')
|
||||
cfg.CONF.set_override('dvs_name', 'fake_dvs', group='dvs')
|
||||
super(NeutronSimpleDvsTest, self).setUp(plugin=PLUGIN_NAME)
|
||||
super(NeutronSimpleDvsTestCase, self).setUp(plugin=plugin)
|
||||
self._plugin = directory.get_plugin()
|
||||
|
||||
|
||||
class NeutronSimpleDvsTest(NeutronSimpleDvsTestCase):
|
||||
|
||||
def _create_and_delete_dvs_network(self, network_type='flat', vlan_tag=0,
|
||||
trunk_mode=False):
|
||||
params = {'provider:network_type': network_type,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,384 @@
|
|||
# Copyright (c) 2017 OpenStack Foundation.
|
||||
#
|
||||
# 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 oslo_utils import uuidutils
|
||||
|
||||
from neutron_lib import context
|
||||
from neutron_lib import exceptions as n_exc
|
||||
from neutron_lib.plugins import directory
|
||||
|
||||
from vmware_nsx.tests.unit.dvs import test_plugin as dvs_tests
|
||||
from vmware_nsx.tests.unit.nsx_v import test_plugin as v_tests
|
||||
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as t_tests
|
||||
|
||||
PLUGIN_NAME = 'vmware_nsx.plugin.NsxTVDPlugin'
|
||||
_uuid = uuidutils.generate_uuid
|
||||
|
||||
|
||||
class NsxTVDPluginTestCase(v_tests.NsxVPluginV2TestCase,
|
||||
t_tests.NsxV3PluginTestCaseMixin,
|
||||
dvs_tests.NeutronSimpleDvsTestCase):
|
||||
|
||||
def setUp(self,
|
||||
plugin=PLUGIN_NAME,
|
||||
ext_mgr=None,
|
||||
service_plugins=None):
|
||||
super(NsxTVDPluginTestCase, self).setUp(
|
||||
plugin=plugin,
|
||||
ext_mgr=ext_mgr)
|
||||
|
||||
self.core_plugin = directory.get_plugin()
|
||||
|
||||
# create a context with this tenant
|
||||
self.context = context.get_admin_context()
|
||||
self.context.tenant_id = self.project_id
|
||||
|
||||
# create a default user for this plugin
|
||||
self.core_plugin.create_project_plugin_map(self.context,
|
||||
{'project_plugin_map': {'plugin': self.plugin_type,
|
||||
'project': self.project_id}})
|
||||
self.sub_plugin = self.core_plugin.get_plugin_by_type(self.plugin_type)
|
||||
|
||||
@property
|
||||
def project_id(self):
|
||||
pass
|
||||
|
||||
@property
|
||||
def plugin_type(self):
|
||||
pass
|
||||
|
||||
def _test_plugin_initialized(self):
|
||||
self.assertTrue(self.core_plugin.is_tvd_plugin())
|
||||
self.assertIsNotNone(self.sub_plugin)
|
||||
|
||||
def _test_call_create(self, obj_name, calls_count=1):
|
||||
method_name = 'create_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func:
|
||||
func_to_call(self.context,
|
||||
{obj_name: {'tenant_id': self.project_id}})
|
||||
self.assertEqual(calls_count, sub_func.call_count)
|
||||
|
||||
def _test_call_create_with_net_id(self, obj_name, field_name='network_id',
|
||||
calls_count=1):
|
||||
method_name = 'create_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
net_id = _uuid()
|
||||
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context,
|
||||
{obj_name: {'tenant_id': self.project_id,
|
||||
field_name: net_id}})
|
||||
self.assertEqual(calls_count, sub_func.call_count)
|
||||
|
||||
def _test_call_delete(self, obj_name):
|
||||
method_name = 'delete_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
obj_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_%s' % obj_name,
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context, obj_id)
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def _test_call_delete_with_net(self, obj_name, field_name='network_id'):
|
||||
method_name = 'delete_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
obj_id = _uuid()
|
||||
net_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_%s' % obj_name,
|
||||
return_value={field_name: net_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context, obj_id)
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def _test_call_update(self, obj_name):
|
||||
method_name = 'update_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
obj_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_%s' % obj_name,
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context, obj_id, {obj_name: {}})
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def _test_call_update_with_net(self, obj_name, field_name='network_id'):
|
||||
method_name = 'update_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
obj_id = _uuid()
|
||||
net_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_%s' % obj_name,
|
||||
return_value={field_name: net_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context, obj_id, {obj_name: {}})
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def _test_call_get(self, obj_name):
|
||||
method_name = 'get_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
obj_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_%s' % obj_name,
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context, obj_id)
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def _test_call_get_with_net(self, obj_name, field_name='network_id'):
|
||||
method_name = 'get_%s' % obj_name
|
||||
func_to_call = getattr(self.core_plugin, method_name)
|
||||
obj_id = _uuid()
|
||||
net_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin, method_name) as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_%s' % obj_name,
|
||||
return_value={field_name: net_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
func_to_call(self.context, obj_id)
|
||||
sub_func.assert_called_once()
|
||||
|
||||
|
||||
class TestPluginWithDefaultPlugin(NsxTVDPluginTestCase):
|
||||
"""Test TVD plugin with the NSX-T (default) sub plugin"""
|
||||
@property
|
||||
def project_id(self):
|
||||
return 'project_t'
|
||||
|
||||
@property
|
||||
def plugin_type(self):
|
||||
return 'nsx-t'
|
||||
|
||||
def test_plugin_initialized(self):
|
||||
self._test_plugin_initialized()
|
||||
|
||||
# no unsupported extensions for the nsx_t plugin
|
||||
self.assertItemsEqual(
|
||||
['router_type', 'router_size'],
|
||||
self.core_plugin._unsupported_fields[self.plugin_type]['router'])
|
||||
self.assertEqual(
|
||||
[],
|
||||
self.core_plugin._unsupported_fields[self.plugin_type]['port'])
|
||||
|
||||
def test_create_network(self):
|
||||
self._test_call_create('network')
|
||||
|
||||
def test_create_subnet(self):
|
||||
self._test_call_create_with_net_id('subnet')
|
||||
|
||||
def test_create_port(self):
|
||||
self._test_call_create_with_net_id('port')
|
||||
|
||||
def test_create_router(self):
|
||||
self._test_call_create('router')
|
||||
|
||||
def test_create_floatingip(self):
|
||||
self._test_call_create_with_net_id(
|
||||
'floatingip', field_name='floating_network_id')
|
||||
|
||||
def test_create_security_group(self):
|
||||
# plugin will be called twice because of the default sg
|
||||
self._test_call_create('security_group', calls_count=2)
|
||||
|
||||
def test_delete_network(self):
|
||||
self._test_call_delete('network')
|
||||
|
||||
def test_delete_subnet(self):
|
||||
self._test_call_delete_with_net('subnet')
|
||||
|
||||
def test_delete_port(self):
|
||||
self._test_call_delete_with_net('port')
|
||||
|
||||
def test_delete_router(self):
|
||||
self._test_call_delete('router')
|
||||
|
||||
def test_delete_floatingip(self):
|
||||
self._test_call_delete_with_net(
|
||||
'floatingip', field_name='floating_network_id')
|
||||
|
||||
def test_delete_security_group(self):
|
||||
self._test_call_delete('security_group')
|
||||
|
||||
def test_update_network(self):
|
||||
self._test_call_update('network')
|
||||
|
||||
def test_update_subnet(self):
|
||||
self._test_call_update_with_net('subnet')
|
||||
|
||||
def test_update_port(self):
|
||||
self._test_call_update_with_net('port')
|
||||
|
||||
def test_update_router(self):
|
||||
self._test_call_update('router')
|
||||
|
||||
def test_update_floatingip(self):
|
||||
self._test_call_update_with_net(
|
||||
'floatingip', field_name='floating_network_id')
|
||||
|
||||
def test_update_security_group(self):
|
||||
self._test_call_update('security_group')
|
||||
|
||||
def test_unsupported_extensions(self):
|
||||
self.assertRaises(n_exc.InvalidInput,
|
||||
self.core_plugin.create_router,
|
||||
self.context,
|
||||
{'router': {'tenant_id': self.project_id,
|
||||
'router_type': 'exclusive'}})
|
||||
|
||||
def test_get_network(self):
|
||||
self._test_call_get('network')
|
||||
|
||||
def test_get_subnet(self):
|
||||
self._test_call_get_with_net('subnet')
|
||||
|
||||
def test_get_port(self):
|
||||
self._test_call_get_with_net('port')
|
||||
|
||||
def test_get_router(self):
|
||||
self._test_call_get('router')
|
||||
|
||||
def test_get_floatingip(self):
|
||||
self._test_call_get_with_net(
|
||||
'floatingip', field_name='floating_network_id')
|
||||
|
||||
def test_get_security_group(self):
|
||||
self._test_call_get('security_group')
|
||||
|
||||
def test_add_router_interface(self):
|
||||
rtr_id = _uuid()
|
||||
port_id = _uuid()
|
||||
net_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin,
|
||||
'add_router_interface') as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_router',
|
||||
return_value={'tenant_id': self.project_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_port',
|
||||
return_value={'network_id': net_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}),\
|
||||
mock.patch.object(self.core_plugin, '_validate_interface_info',
|
||||
return_value=(True, False)):
|
||||
self.core_plugin.add_router_interface(self.context, rtr_id,
|
||||
{'port_id': port_id})
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def test_add_invalid_router_interface(self):
|
||||
# Test that the plugin prevents adding interface from one plugin
|
||||
# to a router of another plugin
|
||||
rtr_id = _uuid()
|
||||
port_id = _uuid()
|
||||
net_id = _uuid()
|
||||
another_tenant_id = _uuid()
|
||||
another_plugin = 'nsx-v' if self.plugin_type == 'nsx-t' else 'nsx-t'
|
||||
self.core_plugin.create_project_plugin_map(self.context,
|
||||
{'project_plugin_map': {'plugin': another_plugin,
|
||||
'project': another_tenant_id}})
|
||||
|
||||
with mock.patch.object(self.core_plugin, '_get_router',
|
||||
return_value={'tenant_id': self.project_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_port',
|
||||
return_value={'network_id': net_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': another_tenant_id}),\
|
||||
mock.patch.object(self.core_plugin, '_validate_interface_info',
|
||||
return_value=(True, False)):
|
||||
self.assertRaises(n_exc.InvalidInput,
|
||||
self.core_plugin.add_router_interface,
|
||||
self.context, rtr_id, {'port_id': port_id})
|
||||
|
||||
def test_remove_router_interface(self):
|
||||
rtr_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin,
|
||||
'remove_router_interface') as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_router',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
self.core_plugin.remove_router_interface(self.context, rtr_id, {})
|
||||
sub_func.assert_called_once()
|
||||
|
||||
def test_disassociate_floatingips(self):
|
||||
port_id = _uuid()
|
||||
net_id = _uuid()
|
||||
with mock.patch.object(self.sub_plugin,
|
||||
'disassociate_floatingips') as sub_func,\
|
||||
mock.patch.object(self.core_plugin, '_get_port',
|
||||
return_value={'network_id': net_id}),\
|
||||
mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
self.core_plugin.disassociate_floatingips(self.context, port_id)
|
||||
sub_func.assert_called_once()
|
||||
|
||||
|
||||
class TestPluginWithNsxv(TestPluginWithDefaultPlugin):
|
||||
"""Test TVD plugin with the NSX-V sub plugin"""
|
||||
|
||||
@property
|
||||
def project_id(self):
|
||||
return 'project_v'
|
||||
|
||||
@property
|
||||
def plugin_type(self):
|
||||
return 'nsx-v'
|
||||
|
||||
def test_plugin_initialized(self):
|
||||
self._test_plugin_initialized()
|
||||
|
||||
# no unsupported extensions for the nsx_v plugin
|
||||
self.assertEqual(
|
||||
[],
|
||||
self.core_plugin._unsupported_fields[self.plugin_type]['router'])
|
||||
self.assertEqual(
|
||||
[],
|
||||
self.core_plugin._unsupported_fields[self.plugin_type]['port'])
|
||||
|
||||
def test_unsupported_extensions(self):
|
||||
self.skipTest('No unsupported extensions in this plugin')
|
||||
|
||||
|
||||
class TestPluginWithDvs(TestPluginWithDefaultPlugin):
|
||||
"""Test TVD plugin with the DVS sub plugin"""
|
||||
|
||||
@property
|
||||
def project_id(self):
|
||||
return 'project_dvs'
|
||||
|
||||
@property
|
||||
def plugin_type(self):
|
||||
return 'dvs'
|
||||
|
||||
def test_plugin_initialized(self):
|
||||
self._test_plugin_initialized()
|
||||
|
||||
# no unsupported extensions for the dvs plugin
|
||||
self.assertItemsEqual(
|
||||
['mac_learning_enabled', 'provider_security_groups'],
|
||||
self.core_plugin._unsupported_fields[self.plugin_type]['port'])
|
||||
|
||||
def test_unsupported_extensions(self):
|
||||
net_id = _uuid()
|
||||
with mock.patch.object(self.core_plugin, '_get_network',
|
||||
return_value={'tenant_id': self.project_id}):
|
||||
self.assertRaises(n_exc.InvalidInput,
|
||||
self.core_plugin.create_port,
|
||||
self.context,
|
||||
{'port': {'tenant_id': self.project_id,
|
||||
'network_id': net_id,
|
||||
'mac_learning_enabled': True}})
|
||||
|
|
@ -67,6 +67,7 @@ from vmware_nsx.common import utils as c_utils
|
|||
from vmware_nsx.db import nsxv_db
|
||||
from vmware_nsx.dvs import dvs
|
||||
from vmware_nsx.dvs import dvs_utils
|
||||
from vmware_nsx.extensions import projectpluginmap
|
||||
from vmware_nsx.extensions import routersize as router_size
|
||||
from vmware_nsx.extensions import routertype as router_type
|
||||
from vmware_nsx.extensions import vnicindex as ext_vnic_idx
|
||||
|
|
@ -229,6 +230,10 @@ class NsxVPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
|
|||
ext_mgr=ext_mgr)
|
||||
self.addCleanup(self.fc2.reset_all)
|
||||
plugin_instance = directory.get_plugin()
|
||||
# handle TVD plugin case
|
||||
if plugin_instance.is_tvd_plugin():
|
||||
plugin_instance = plugin_instance.get_plugin_by_type(
|
||||
projectpluginmap.NsxPlugins.NSX_V)
|
||||
plugin_instance.real_get_edge = plugin_instance._get_edge_id_by_rtr_id
|
||||
plugin_instance._get_edge_id_by_rtr_id = mock.Mock()
|
||||
plugin_instance._get_edge_id_by_rtr_id.return_value = False
|
||||
|
|
@ -246,6 +251,35 @@ class NsxVPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
|
|||
plugin._vcm = dvs.VCManager()
|
||||
return plugin
|
||||
|
||||
|
||||
class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
|
||||
|
||||
def _test_create_bridge_network(self, vlan_id=0):
|
||||
net_type = vlan_id and 'vlan' or 'flat'
|
||||
name = 'bridge_net'
|
||||
expected = [('subnets', []), ('name', name), ('admin_state_up', True),
|
||||
('status', 'ACTIVE'), ('shared', False),
|
||||
(pnet.NETWORK_TYPE, net_type),
|
||||
(pnet.PHYSICAL_NETWORK, 'tzuuid'),
|
||||
(pnet.SEGMENTATION_ID, vlan_id)]
|
||||
providernet_args = {pnet.NETWORK_TYPE: net_type,
|
||||
pnet.PHYSICAL_NETWORK: 'tzuuid'}
|
||||
if vlan_id:
|
||||
providernet_args[pnet.SEGMENTATION_ID] = vlan_id
|
||||
with self.network(name=name,
|
||||
providernet_args=providernet_args,
|
||||
arg_list=(pnet.NETWORK_TYPE,
|
||||
pnet.PHYSICAL_NETWORK,
|
||||
pnet.SEGMENTATION_ID)) as net:
|
||||
for k, v in expected:
|
||||
self.assertEqual(net['network'][k], v)
|
||||
|
||||
def test_create_bridge_network(self):
|
||||
self._test_create_bridge_network()
|
||||
|
||||
def test_create_bridge_vlan_network(self):
|
||||
self._test_create_bridge_network(vlan_id=123)
|
||||
|
||||
def test_get_vlan_network_name(self):
|
||||
p = directory.get_plugin()
|
||||
net_id = uuidutils.generate_uuid()
|
||||
|
|
@ -282,35 +316,6 @@ class NsxVPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
|
|||
self.assertEqual(expected,
|
||||
p._get_vlan_network_name(net, dvs_id))
|
||||
|
||||
|
||||
class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
|
||||
|
||||
def _test_create_bridge_network(self, vlan_id=0):
|
||||
net_type = vlan_id and 'vlan' or 'flat'
|
||||
name = 'bridge_net'
|
||||
expected = [('subnets', []), ('name', name), ('admin_state_up', True),
|
||||
('status', 'ACTIVE'), ('shared', False),
|
||||
(pnet.NETWORK_TYPE, net_type),
|
||||
(pnet.PHYSICAL_NETWORK, 'tzuuid'),
|
||||
(pnet.SEGMENTATION_ID, vlan_id)]
|
||||
providernet_args = {pnet.NETWORK_TYPE: net_type,
|
||||
pnet.PHYSICAL_NETWORK: 'tzuuid'}
|
||||
if vlan_id:
|
||||
providernet_args[pnet.SEGMENTATION_ID] = vlan_id
|
||||
with self.network(name=name,
|
||||
providernet_args=providernet_args,
|
||||
arg_list=(pnet.NETWORK_TYPE,
|
||||
pnet.PHYSICAL_NETWORK,
|
||||
pnet.SEGMENTATION_ID)) as net:
|
||||
for k, v in expected:
|
||||
self.assertEqual(net['network'][k], v)
|
||||
|
||||
def test_create_bridge_network(self):
|
||||
self._test_create_bridge_network()
|
||||
|
||||
def test_create_bridge_vlan_network(self):
|
||||
self._test_create_bridge_network(vlan_id=123)
|
||||
|
||||
def _test_generate_tag(self, vlan_id):
|
||||
net_type = 'vlan'
|
||||
name = 'bridge_net'
|
||||
|
|
@ -4349,9 +4354,6 @@ class TestNSXvAllowedAddressPairs(NsxVPluginV2TestCase,
|
|||
def test_create_overlap_with_fixed_ip(self):
|
||||
pass
|
||||
|
||||
def test_get_vlan_network_name(self):
|
||||
pass
|
||||
|
||||
def test_create_port_with_cidr_address_pair(self):
|
||||
with self.network() as net:
|
||||
address_pairs = [{'mac_address': '00:00:00:00:00:01',
|
||||
|
|
|
|||
Loading…
Reference in New Issue