From a4e40114dddc47566d64920e96f3601989a32096 Mon Sep 17 00:00:00 2001 From: Adit Sarfaty Date: Tue, 19 Dec 2017 14:39:34 +0200 Subject: [PATCH] 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 --- vmware_nsx/db/extended_security_group.py | 3 +- vmware_nsx/plugins/nsx/plugin.py | 44 ++- vmware_nsx/plugins/nsx_v3/plugin.py | 9 - vmware_nsx/tests/unit/dvs/test_plugin.py | 7 +- vmware_nsx/tests/unit/nsx_tvd/__init__.py | 0 vmware_nsx/tests/unit/nsx_tvd/test_plugin.py | 384 +++++++++++++++++++ vmware_nsx/tests/unit/nsx_v/test_plugin.py | 66 ++-- 7 files changed, 454 insertions(+), 59 deletions(-) create mode 100644 vmware_nsx/tests/unit/nsx_tvd/__init__.py create mode 100644 vmware_nsx/tests/unit/nsx_tvd/test_plugin.py diff --git a/vmware_nsx/db/extended_security_group.py b/vmware_nsx/db/extended_security_group.py index 2a1a48a55c..2885863d08 100644 --- a/vmware_nsx/db/extended_security_group.py +++ b/vmware_nsx/db/extended_security_group.py @@ -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 diff --git a/vmware_nsx/plugins/nsx/plugin.py b/vmware_nsx/plugins/nsx/plugin.py index 9042fd5513..521f653cbc 100644 --- a/vmware_nsx/plugins/nsx/plugin.py +++ b/vmware_nsx/plugins/nsx/plugin.py @@ -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, diff --git a/vmware_nsx/plugins/nsx_v3/plugin.py b/vmware_nsx/plugins/nsx_v3/plugin.py index 7d562d8485..0cac68415c 100644 --- a/vmware_nsx/plugins/nsx_v3/plugin.py +++ b/vmware_nsx/plugins/nsx_v3/plugin.py @@ -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. diff --git a/vmware_nsx/tests/unit/dvs/test_plugin.py b/vmware_nsx/tests/unit/dvs/test_plugin.py index 98eb40c9d3..97c10c0712 100644 --- a/vmware_nsx/tests/unit/dvs/test_plugin.py +++ b/vmware_nsx/tests/unit/dvs/test_plugin.py @@ -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, diff --git a/vmware_nsx/tests/unit/nsx_tvd/__init__.py b/vmware_nsx/tests/unit/nsx_tvd/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/vmware_nsx/tests/unit/nsx_tvd/test_plugin.py b/vmware_nsx/tests/unit/nsx_tvd/test_plugin.py new file mode 100644 index 0000000000..2d1d22c9c7 --- /dev/null +++ b/vmware_nsx/tests/unit/nsx_tvd/test_plugin.py @@ -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}}) diff --git a/vmware_nsx/tests/unit/nsx_v/test_plugin.py b/vmware_nsx/tests/unit/nsx_v/test_plugin.py index 7df13916ef..1b7cd07618 100644 --- a/vmware_nsx/tests/unit/nsx_v/test_plugin.py +++ b/vmware_nsx/tests/unit/nsx_v/test_plugin.py @@ -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',