x
/
vmware-nsx
Code Issues Proposed changes

Integration with neutron-lib plugin directory 

Co-authored-by: Armando Migliaccio <armamig@gmail.com>
Change-Id: I7c7efbaa7a53f8d244f1b19ea3a7c8a8900602be
Depends-on: I7331e914234c5f0b7abe836604fdd7e4067551cf
Depends-on: Ia91dfbf9d93e19b43c0dd0b58b95fc0080b0ad7c
Depends-on: I48cd9257f419ad949ba0cecc9aca98a624ca4dcc
This commit is contained in:
Adit Sarfaty 2016-11-16 13:58:51 +02:00 committed by garyk
parent 854fb2098f
commit 3009f37757
25 changed files with 161 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
vmware_nsx/extensions/lsn.py
View File

@ -14,10 +14,9 @@
# License for the specific language governing permissions and limitations
# under the License.
#
from neutron.api import extensions
from neutron.api.v2 import base
from neutron import manager
from neutron_lib.plugins import directory
EXT_ALIAS = 'lsn'
@ -60,7 +59,7 @@ class Lsn(extensions.ExtensionDescriptor):
def get_resources(cls):
"""Returns Ext Resources."""
exts = []
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
resource_name = EXT_ALIAS
collection_name = resource_name.replace('_', '-') + "s"
params = RESOURCE_ATTRIBUTE_MAP.get(COLLECTION_NAME, dict())

4
vmware_nsx/extensions/qos_queue.py
View File

@ -19,10 +19,10 @@ import abc
from neutron.api import extensions
from neutron.api.v2 import attributes as attr
from neutron.api.v2 import base
from neutron import manager
from neutron_lib.api import converters
from neutron_lib import exceptions as nexception
from neutron_lib.plugins import directory
from vmware_nsx._i18n import _
@ -188,7 +188,7 @@ class Qos_queue(extensions.ExtensionDescriptor):
def get_resources(cls):
"""Returns Ext Resources."""
exts = []
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
resource_name = 'qos_queue'
collection_name = resource_name.replace('_', '-') + "s"
params = RESOURCE_ATTRIBUTE_MAP.get(resource_name + "s", dict())

5
vmware_nsx/plugins/nsx_v/plugin.py
View File

@ -21,6 +21,7 @@ import netaddr
from neutron_lib.api import validators
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
@ -66,7 +67,6 @@ from neutron.extensions import portbindings as pbin
from neutron.extensions import portsecurity as psec
from neutron.extensions import providernet as pnet
from neutron.extensions import securitygroup as ext_sg
from neutron import manager
from neutron.plugins.common import constants as plugin_const
from neutron.plugins.common import utils
from neutron.quota import resource_registry
@ -2244,8 +2244,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
r[ROUTER_SIZE] = cfg.CONF.nsxv.exclusive_router_appliance_size
def _get_router_flavor_profile(self, context, flavor_id):
flv_plugin = manager.NeutronManager.get_service_plugins().get(
plugin_const.FLAVORS)
flv_plugin = directory.get_plugin(plugin_const.FLAVORS)
if not flv_plugin:
msg = _("Flavors plugin not found")
raise n_exc.BadRequest(resource="router", msg=msg)

8
vmware_nsx/services/flowclassifier/nsx_v/driver.py
View File

@ -23,7 +23,7 @@ from neutron.callbacks import events
from neutron.callbacks import registry
from neutron.callbacks import resources
from neutron import context as n_context
from neutron import manager
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import helpers as log_helpers
from oslo_log import log as logging
@ -121,7 +121,7 @@ class NsxvFlowClassifierDriver(fc_driver.FlowClassifierDriverBase):
if self._is_new_security_group:
# add existing VMs to the new security group
# This code must run after init is done
core_plugin = manager.NeutronManager.get_plugin()
core_plugin = directory.get_plugin()
core_plugin.add_vms_to_service_insertion(
self._security_group_id)
@ -137,9 +137,7 @@ class NsxvFlowClassifierDriver(fc_driver.FlowClassifierDriverBase):
The user will be able to delete/change it later
"""
context = n_context.get_admin_context()
fc_plugin = manager.NeutronManager.get_service_plugins().get(
flowclassifier.FLOW_CLASSIFIER_EXT)
fc_plugin = directory.get_plugin(flowclassifier.FLOW_CLASSIFIER_EXT)
# first check that there is no other flow classifier entry defined:
fcs = fc_plugin.get_flow_classifiers(context)
if len(fcs) > 0:

5
vmware_nsx/services/flowclassifier/nsx_v/utils.py
View File

@ -14,7 +14,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron import manager
from neutron_lib.plugins import directory
from oslo_log import log as logging
LOG = logging.getLogger(__name__)
@ -44,8 +44,7 @@ class NsxvServiceInsertionHandler(object):
def is_service_insertion_enabled(self):
# Note - this cannot be called during init, since the manager is busy
if (manager.NeutronManager.get_service_plugins().get(
FLOW_CLASSIFIER_EXT)):
if (directory.get_plugin(FLOW_CLASSIFIER_EXT)):
return True
return False

4
vmware_nsx/services/ipam/nsx_v/driver.py
View File

@ -25,8 +25,8 @@ from neutron.ipam.drivers.neutrondb_ipam import driver as neutron_driver
from neutron.ipam import exceptions as ipam_exc
from neutron.ipam import requests as ipam_req
from neutron.ipam import subnet_alloc
from neutron import manager
from neutron_lib.api import validators
from neutron_lib.plugins import directory
from oslo_log import log as logging
from vmware_nsx._i18n import _, _LE
@ -41,7 +41,7 @@ LOG = logging.getLogger(__name__)
class NsxvIpamBase(object):
@classmethod
def get_core_plugin(cls):
return manager.NeutronManager.get_plugin()
return directory.get_plugin()
@classmethod
def _fetch_subnet(cls, context, id):

4
vmware_nsx/services/l2gateway/nsx_v/driver.py
View File

@ -17,8 +17,8 @@
from networking_l2gw.db.l2gateway import l2gateway_db
from networking_l2gw.services.l2gateway.common import constants as l2gw_const
from networking_l2gw.services.l2gateway import exceptions as l2gw_exc
from neutron import manager
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from oslo_log import log as logging
from oslo_utils import uuidutils
@ -43,7 +43,7 @@ class NsxvL2GatewayDriver(l2gateway_db.L2GatewayMixin):
@property
def _core_plugin(self):
return manager.NeutronManager.get_plugin()
return directory.get_plugin()
@property
def _nsxv(self):

4
vmware_nsx/services/l2gateway/nsx_v3/driver.py
View File

@ -28,10 +28,10 @@ from neutron.callbacks import registry
from neutron.callbacks import resources
from neutron import context
from neutron.extensions import providernet
from neutron import manager
from neutron.plugins.common import utils as n_utils
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from vmware_nsx._i18n import _, _LE, _LI
from vmware_nsx.common import utils as nsx_utils
@ -59,7 +59,7 @@ class NsxV3Driver(l2gateway_db.L2GatewayMixin):
@property
def _core_plugin(self):
return manager.NeutronManager.get_plugin()
return directory.get_plugin()
def subscribe_callback_notifications(self):
registry.subscribe(self._prevent_l2gw_port_delete, resources.PORT,

7
vmware_nsx/services/lbaas/nsx_v/v2/base_mgr.py
View File

@ -13,8 +13,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron import manager
from neutron.plugins.common import constants
from neutron_lib import constants
from neutron_lib.plugins import directory
class EdgeLoadbalancerBaseManager(object):
@ -26,8 +26,7 @@ class EdgeLoadbalancerBaseManager(object):
self.vcns_driver = vcns_driver
def _get_plugin(self, plugin_type):
loaded_plugins = manager.NeutronManager.get_service_plugins()
return loaded_plugins[plugin_type]
return directory.get_plugin(plugin_type)
@property
def lbv2_driver(self):

4
vmware_nsx/services/neutron_taas/nsx_v3/driver.py
View File

@ -16,7 +16,7 @@
import netaddr
from neutron import manager
from neutron_lib.plugins import directory
from neutron_taas.db import taas_db
from neutron_taas.services.taas import service_drivers as base_driver
@ -47,7 +47,7 @@ class NsxV3Driver(base_driver.TaasBaseDriver,
@property
def _nsx_plugin(self):
return manager.NeutronManager.get_plugin()
return directory.get_plugin()
def _validate_tap_flow(self, source_port, dest_port):
# Verify whether the source port is not same as the destination port

5
vmware_nsx/services/qos/nsx_v/utils.py
View File

@ -16,10 +16,10 @@
from neutron.api.rpc.callbacks import events as callbacks_events
from neutron import context as n_context
from neutron import manager
from neutron.objects.qos import policy as qos_policy
from neutron.plugins.common import constants
from neutron.services.qos import qos_consts
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging
@ -52,8 +52,7 @@ class NsxVQosRule(object):
def _get_qos_plugin(self):
if not self._qos_plugin:
loaded_plugins = manager.NeutronManager.get_service_plugins()
self._qos_plugin = loaded_plugins[constants.QOS]
self._qos_plugin = directory.get_plugin(constants.QOS)
return self._qos_plugin
# init the nsx_v qos data (outShapingPolicy) from a neutron qos policy

4
vmware_nsx/services/qos/nsx_v3/utils.py
View File

@ -16,10 +16,10 @@
from neutron.api.rpc.callbacks import events as callbacks_events
from neutron import context as n_context
from neutron import manager
from neutron.objects.qos import policy as qos_policy
from neutron.services.qos import qos_consts
from neutron_lib.api import validators
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_log import log as logging
@ -81,7 +81,7 @@ class QosNotificationsHandler(object):
@property
def _core_plugin(self):
return manager.NeutronManager.get_plugin()
return directory.get_plugin()
@property
def _nsxlib_qos(self):

4
vmware_nsx/tests/unit/dvs/test_plugin.py
View File

@ -19,10 +19,10 @@ from oslo_utils import uuidutils
from neutron import context
from neutron.extensions import portbindings
from neutron import manager
from neutron.tests import base
import neutron.tests.unit.db.test_db_base_plugin_v2 as test_plugin
from neutron_lib import exceptions as exp
from neutron_lib.plugins import directory
from vmware_nsx.common import exceptions as nsx_exc
from vmware_nsx.db import db as nsx_db
from vmware_nsx.dvs import dvs
@ -152,7 +152,7 @@ class NeutronSimpleDvsTest(test_plugin.NeutronDbPluginV2TestCase):
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)
self._plugin = manager.NeutronManager.get_plugin()
self._plugin = directory.get_plugin()
def _create_and_delete_dvs_network(self, network_type='flat', vlan_tag=0):
params = {'provider:network_type': network_type,

7
vmware_nsx/tests/unit/extensions/test_metadata.py
View File

@ -17,11 +17,10 @@ import mock
import netaddr
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from oslo_config import cfg
import webob.exc
from neutron import manager
from vmware_nsx.api_client import exception as api_exc
from vmware_nsx.common import config
@ -176,7 +175,7 @@ class MetaDataTestCase(object):
with self.subnet() as s:
# Save function being mocked.
real_func = self._plugin_class.add_router_interface
plugin_instance = manager.NeutronManager.get_plugin()
plugin_instance = directory.get_plugin()
# Raise a NeutronException when adding metadata subnet
# to router.
@ -236,7 +235,7 @@ class MetaDataTestCase(object):
meta_port_id = ports[0]['id']
# Save function being mocked.
real_func = self._plugin_class.remove_router_interface
plugin_instance = manager.NeutronManager.get_plugin()
plugin_instance = directory.get_plugin()
# Raise a NeutronException when removing metadata subnet
# from router.

6
vmware_nsx/tests/unit/extensions/test_networkgw.py
View File

@ -24,12 +24,13 @@ from neutron.api.v2 import attributes
from neutron import context
from neutron.db import api as db_api
from neutron.db import db_base_plugin_v2
from neutron import manager
from neutron import quota
from neutron.tests import base
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_plugin
from neutron_lib import constants
from neutron_lib.plugins import directory
from vmware_nsx.api_client import exception as api_exc
from vmware_nsx.common import exceptions as nsx_exc
@ -85,8 +86,9 @@ class NetworkGatewayExtensionTestCase(base.BaseTestCase):
self.plugin = _plugin_patcher.start()
# Instantiate mock plugin and enable extensions
manager.NeutronManager.get_plugin().supported_extension_aliases = (
self.plugin.return_value.supported_extension_aliases = (
[networkgw.EXT_ALIAS])
directory.add_plugin(constants.CORE, self.plugin.return_value)
ext_mgr = TestExtensionManager()
extensions.PluginAwareExtensionManager._instance = ext_mgr
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)

4
vmware_nsx/tests/unit/extensions/test_secgroup_rule_local_ip_prefix.py
View File

@ -21,9 +21,9 @@ from oslo_utils import uuidutils
from neutron.api.v2 import attributes
from neutron.db import db_base_plugin_v2
from neutron.db import securitygroups_db
from neutron import manager
from neutron.tests.unit.extensions import test_securitygroup
from neutron_lib import constants as const
from neutron_lib.plugins import directory
from vmware_nsx.db import extended_security_group_rule as ext_rule_db
from vmware_nsx.extensions import secgroup_rule_local_ip_prefix as ext_loip
@ -107,7 +107,7 @@ class TestNsxVExtendedSGRule(test_nsxv_plugin.NsxVSecurityGroupsTestCase,
def test_create_rule_with_local_ip_prefix(self):
sg_utils = securitygroup_utils.NsxSecurityGroupUtils(None)
local_ip_prefix = '239.255.0.0/16'
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
dest = {'type': 'Ipv4Address', 'value': local_ip_prefix}
plugin.nsx_sg_utils.get_rule_config = mock.Mock(

9
vmware_nsx/tests/unit/extensions/test_vnic_index.py
View File

@ -19,9 +19,10 @@ from oslo_utils import uuidutils
from neutron import context as neutron_context
from neutron.db import db_base_plugin_v2
from neutron import manager
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_plugin
from neutron_lib.api import validators
from neutron_lib.plugins import directory
from vmware_nsx.db import vnic_index_db
from vmware_nsx.extensions import vnicindex as vnicidx
from vmware_nsx.tests import unit as vmware
@ -80,7 +81,7 @@ class VnicIndexDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
return res
def test_vnic_index_db(self):
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
vnic_index = 2
device_id = _uuid()
context = neutron_context.get_admin_context()
@ -100,7 +101,7 @@ class VnicIndexDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
# self.assertIsNone(plugin._get_port_vnic_index(context, port_id))
def test_vnic_index_db_duplicate(self):
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
vnic_index = 2
device_id = _uuid()
context = neutron_context.get_admin_context()
@ -113,7 +114,7 @@ class VnicIndexDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
vnic_index)
def test_vnic_index_db_duplicate_new_port(self):
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
vnic_index = 2
device_id = _uuid()
context = neutron_context.get_admin_context()

56
vmware_nsx/tests/unit/nsx_mh/test_opts.py
View File

@ -16,7 +16,6 @@
import fixtures
import mock
from neutron import manager
from neutron.tests import base
from oslo_config import cfg
from oslo_utils import uuidutils
@ -29,6 +28,7 @@ from vmware_nsx.common import exceptions
from vmware_nsx.common import sync
from vmware_nsx import nsx_cluster
from vmware_nsx.nsxlib.mh import lsn as lsnlib
from vmware_nsx import plugin as mh_plugin
from vmware_nsx.tests import unit as vmware
BASE_CONF_PATH = vmware.get_fake_conf('neutron.conf.test')
@ -76,7 +76,7 @@ class ConfigurationTest(base.BaseTestCase):
def setUp(self):
super(ConfigurationTest, self).setUp()
self.useFixture(fixtures.MonkeyPatch(
'neutron.manager.NeutronManager._instance',
'neutron_lib.plugins.directory._instance',
None))
# Avoid runs of the synchronizer looping call
patch_sync = mock.patch.object(sync, '_start_loopingcall')
@ -96,11 +96,16 @@ class ConfigurationTest(base.BaseTestCase):
self.assertEqual('whatever', cluster.default_l3_gw_service_uuid)
self.assertEqual('whatever', cluster.nsx_default_interface_name)
def _get_mh_plugin(self):
with mock.patch("neutron.common.rpc.create_connection"):
plugin = mh_plugin.NsxPlugin()
return plugin
def test_load_plugin_with_full_options(self):
self.config_parse(args=['--config-file', BASE_CONF_PATH,
'--config-file', NSX_INI_FULL_PATH])
cfg.CONF.set_override('core_plugin', vmware.PLUGIN_NAME)
plugin = manager.NeutronManager().get_plugin()
plugin = self._get_mh_plugin()
cluster = plugin.cluster
self._assert_required_options(cluster)
self._assert_extra_options(cluster)
@ -109,7 +114,7 @@ class ConfigurationTest(base.BaseTestCase):
self.config_parse(args=['--config-file', BASE_CONF_PATH,
'--config-file', NSX_INI_PATH])
cfg.CONF.set_override('core_plugin', vmware.PLUGIN_NAME)
plugin = manager.NeutronManager().get_plugin()
plugin = self._get_mh_plugin()
self._assert_required_options(plugin.cluster)
def test_defaults(self):
@ -137,7 +142,7 @@ class ConfigurationTest(base.BaseTestCase):
'--config-file', NSX_INI_FULL_PATH])
cfg.CONF.set_override('core_plugin', vmware.PLUGIN_NAME)
# Load the configuration, and initialize the plugin
manager.NeutronManager().get_plugin()
self._get_mh_plugin()
self.assertIn('extensions', cfg.CONF.api_extensions_path)
def test_agentless_extensions(self):
@ -153,7 +158,7 @@ class ConfigurationTest(base.BaseTestCase):
with mock.patch.object(lsnlib,
'service_cluster_exists',
return_value=True):
plugin = manager.NeutronManager().get_plugin()
plugin = self._get_mh_plugin()
self.assertNotIn('agent',
plugin.supported_extension_aliases)
self.assertNotIn('dhcp_agent_scheduler',
@ -170,8 +175,13 @@ class ConfigurationTest(base.BaseTestCase):
with mock.patch.object(client.NsxApiClient,
'get_version',
return_value=version.Version("3.2")):
self.assertRaises(exceptions.NsxPluginException,
manager.NeutronManager)
try:
self._get_mh_plugin()
except exceptions.NsxPluginException:
# This is the correct result
pass
else:
self.fail('Expected NsxPluginException exception')
def test_agentless_extensions_unmet_deps_fail(self):
self.config_parse(args=['--config-file', BASE_CONF_PATH,
@ -185,8 +195,13 @@ class ConfigurationTest(base.BaseTestCase):
with mock.patch.object(lsnlib,
'service_cluster_exists',
return_value=False):
self.assertRaises(exceptions.NsxPluginException,
manager.NeutronManager)
try:
self._get_mh_plugin()
except exceptions.NsxPluginException:
# This is the correct result
pass
else:
self.fail('Expected NsxPluginException exception')
def test_agent_extensions(self):
self.config_parse(args=['--config-file', BASE_CONF_PATH,
@ -194,7 +209,7 @@ class ConfigurationTest(base.BaseTestCase):
cfg.CONF.set_override('core_plugin', vmware.PLUGIN_NAME)
self.assertEqual(config.AgentModes.AGENT,
cfg.CONF.NSX.agent_mode)
plugin = manager.NeutronManager().get_plugin()
plugin = self._get_mh_plugin()
self.assertIn('agent',
plugin.supported_extension_aliases)
self.assertIn('dhcp_agent_scheduler',
@ -212,7 +227,7 @@ class ConfigurationTest(base.BaseTestCase):
with mock.patch.object(lsnlib,
'service_cluster_exists',
return_value=True):
plugin = manager.NeutronManager().get_plugin()
plugin = self._get_mh_plugin()
self.assertIn('agent',
plugin.supported_extension_aliases)
self.assertIn('dhcp_agent_scheduler',
@ -226,7 +241,7 @@ class OldNVPConfigurationTest(base.BaseTestCase):
def setUp(self):
super(OldNVPConfigurationTest, self).setUp()
self.useFixture(fixtures.MonkeyPatch(
'neutron.manager.NeutronManager._instance',
'neutron_lib.plugins.directory._instance',
None))
# Avoid runs of the synchronizer looping call
patch_sync = mock.patch.object(sync, '_start_loopingcall')
@ -242,10 +257,11 @@ class OldNVPConfigurationTest(base.BaseTestCase):
self.config_parse(args=['--config-file', BASE_CONF_PATH,
'--config-file', NVP_INI_DEPR_PATH])
cfg.CONF.set_override('core_plugin', vmware.PLUGIN_NAME)
plugin = manager.NeutronManager().get_plugin()
cluster = plugin.cluster
# Verify old nvp_* params have been fully parsed
self._assert_required_options(cluster)
self.assertEqual(3, cluster.http_timeout)
self.assertEqual(2, cluster.retries)
self.assertEqual(2, cluster.redirects)
with mock.patch("neutron.common.rpc.create_connection"):
plugin = mh_plugin.NsxPlugin()
cluster = plugin.cluster
# Verify old nvp_* params have been fully parsed
self._assert_required_options(cluster)
self.assertEqual(3, cluster.http_timeout)
self.assertEqual(2, cluster.retries)
self.assertEqual(2, cluster.redirects)

16
vmware_nsx/tests/unit/nsx_mh/test_plugin.py
View File

@ -25,7 +25,6 @@ from neutron.extensions import l3_ext_gw_mode
from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet
from neutron.extensions import securitygroup as secgrp
from neutron import manager
from neutron.tests.unit import _test_extension_portbindings as test_bindings
import neutron.tests.unit.db.test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.extensions import test_extra_dhcp_opt as test_dhcpopts
@ -35,6 +34,7 @@ import neutron.tests.unit.extensions.test_securitygroup as ext_sg
from neutron.tests.unit import testlib_api
from neutron_lib import constants
from neutron_lib import exceptions as ntn_exc
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log
@ -225,7 +225,7 @@ class TestPortsV2(NsxPluginV2TestCase,
'admin_state_up': False,
'fixed_ips': [],
'tenant_id': self._tenant_id}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
with mock.patch.object(plugin, 'get_network',
return_value=net['network']):
port_req = self.new_create_request('ports', data, self.fmt)
@ -350,7 +350,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxPluginV2TestCase):
def test_update_network_with_admin_false(self):
data = {'network': {'admin_state_up': False}}
with self.network() as net:
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
self.assertRaises(NotImplementedError,
plugin.update_network,
context.get_admin_context(),
@ -365,7 +365,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxPluginV2TestCase):
# Because of commit 79c9712 a tenant must be specified otherwise
# the unit test will fail
ctx.tenant_id = 'whatever'
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
net = plugin.create_network(
ctx, {'network': {'name': 'xxx',
'admin_state_up': True,
@ -393,7 +393,7 @@ class SecurityGroupsTestCase(ext_sg.SecurityGroupDBTestCase):
instance.return_value.request.side_effect = self.fc.fake_request
super(SecurityGroupsTestCase, self).setUp(vmware.PLUGIN_NAME)
self.plugin = manager.NeutronManager.get_plugin()
self.plugin = directory.get_plugin()
class TestSecurityGroup(ext_sg.TestSecurityGroups, SecurityGroupsTestCase):
@ -516,7 +516,7 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxPluginV2TestCase):
ext_mgr = ext_mgr or TestL3ExtensionManager()
super(L3NatTest, self).setUp(
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
plugin_instance = manager.NeutronManager.get_plugin()
plugin_instance = directory.get_plugin()
self._plugin_name = "%s.%s" % (
plugin_instance.__module__,
plugin_instance.__class__.__name__)
@ -812,7 +812,7 @@ class TestL3NatTestCase(L3NatTest,
with self.port() as p:
private_sub = {'subnet': {'id':
p['port']['fixed_ips'][0]['subnet_id']}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
with mock.patch.object(plugin, 'notify_routers_updated') as notify:
with self.floatingip_no_assoc(private_sub) as fip:
port_id = p['port']['id']
@ -873,7 +873,7 @@ class TestL3NatTestCase(L3NatTest,
p['port']['id'])
def test_update_subnet_gateway_for_external_net(self):
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
port_mock = {'uuid': uuidutils.generate_uuid()}
with mock.patch.object(plugin, '_find_router_gw_port',
return_value=port_mock):

14
vmware_nsx/tests/unit/nsx_mh/test_sync.py
View File

@ -294,15 +294,15 @@ class SyncTestCase(testlib_api.SqlTestCase):
'--config-file', vmware.get_fake_conf('nsx.ini.test')]
self.config_parse(args=args)
cfg.CONF.set_override('allow_overlapping_ips', True)
self._plugin = plugin.NsxPlugin()
# Mock neutron manager plugin load functions to speed up tests
mock_nm_get_plugin = mock.patch('neutron.manager.NeutronManager.'
'get_plugin')
mock_nm_get_service_plugins = mock.patch(
'neutron.manager.NeutronManager.get_service_plugins')
with mock.patch("neutron.common.rpc.create_connection"):
self._plugin = plugin.NsxPlugin()
mock_nm_get_plugin = mock.patch(
"neutron_lib.plugins.directory.get_plugin")
self.mock_nm_get_plugin = mock_nm_get_plugin.start()
self.mock_nm_get_plugin.return_value = self._plugin
mock_nm_get_service_plugins.start()
super(SyncTestCase, self).setUp()
self.addCleanup(self.fc.reset_all)

90
vmware_nsx/tests/unit/nsx_v/test_plugin.py
View File

@ -30,7 +30,6 @@ from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet
from neutron.extensions import router_availability_zone
from neutron.extensions import securitygroup as secgrp
from neutron import manager
from neutron.objects.qos import policy as qos_pol
from neutron.plugins.common import constants as plugin_const
from neutron.services.qos import qos_consts
@ -46,6 +45,7 @@ from neutron.tests.unit import testlib_api
from neutron_lib.api import validators
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from neutron_lib.utils import helpers
from oslo_config import cfg
from oslo_utils import uuidutils
@ -151,7 +151,7 @@ class NsxVPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
plugin=plugin,
ext_mgr=ext_mgr)
self.addCleanup(self.fc2.reset_all)
plugin_instance = manager.NeutronManager.get_plugin()
plugin_instance = directory.get_plugin()
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
@ -160,14 +160,14 @@ class NsxVPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
def _get_core_plugin_with_dvs(self):
# enable dvs features to allow policy with QOS
cfg.CONF.set_default('use_dvs_features', True, 'nsxv')
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
with mock.patch.object(dvs_utils, 'dvs_create_session'):
with mock.patch.object(dvs.DvsManager, '_get_dvs_moref'):
plugin._dvs = dvs.DvsManager()
return plugin
def test_get_vlan_network_name(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
net_id = uuidutils.generate_uuid()
dvs_id = 'dvs-10'
net = {'name': '',
@ -192,7 +192,7 @@ class NsxVPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
p._get_vlan_network_name(net, dvs_id))
def test_get_vlan_network_name_with_net_name_missing(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
net_id = uuidutils.generate_uuid()
dvs_id = 'dvs-10'
net = {'id': net_id}
@ -297,7 +297,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
def test_update_network_with_admin_false(self):
data = {'network': {'admin_state_up': False}}
with self.network() as net:
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
self.assertRaises(NotImplementedError,
plugin.update_network,
context.get_admin_context(),
@ -355,7 +355,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
providernet_args = {pnet.NETWORK_TYPE: 'vlan',
pnet.SEGMENTATION_ID: 100,
pnet.PHYSICAL_NETWORK: 'dvs-1, dvs-2, dvs-3'}
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with mock.patch.object(
p, '_create_vlan_network_at_backend',
# Return three netmorefs as side effect
@ -377,7 +377,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
pnet.SEGMENTATION_ID: 100,
pnet.PHYSICAL_NETWORK: 'dvs-1, dvs-2, dvs-3'}
network = {'network': net_data}
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with mock.patch.object(
p, '_create_vlan_network_at_backend',
# Return two successful netmorefs and fail on the backend
@ -401,7 +401,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
pnet.SEGMENTATION_ID: 100,
pnet.PHYSICAL_NETWORK: 'dvs-1, dvs-2, dvs-3'}
network = {'network': net_data}
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with mock.patch.object(
p, '_validate_provider_create',
side_effect=[nsxv_exc.NsxResourceNotFound(res_id='dvs-2',
@ -420,7 +420,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
providernet_args = {pnet.NETWORK_TYPE: 'vlan',
pnet.SEGMENTATION_ID: 100,
pnet.PHYSICAL_NETWORK: 'dvs-1, dvs-2, dvs-1'}
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with mock.patch.object(
p, '_create_vlan_network_at_backend',
# Return two netmorefs as side effect
@ -436,7 +436,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
self.assertEqual(2, vlan_net_call.call_count)
def test_get_dvs_ids_for_multiple_dvs_vlan_network(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
# If no DVS-ID is provided as part of physical network, return
# global DVS-ID configured in nsx.ini
physical_network = constants.ATTR_NOT_SPECIFIED
@ -481,7 +481,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
pnet.SEGMENTATION_ID: constants.ATTR_NOT_SPECIFIED,
pnet.NETWORK_TYPE: 'vxlan',
pnet.PHYSICAL_NETWORK: 'vdnscope-2'}}
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with mock.patch.object(p.nsx_v.vcns, 'validate_vdn_scope',
side_effect=[False]):
self.assertRaises(nsxv_exc.NsxResourceNotFound,
@ -496,7 +496,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
'name': 'test-qos',
'tenant_id': self._tenant_id,
'qos_policy_id': _uuid()}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
self.assertRaises(n_exc.InvalidInput,
plugin.create_network,
context.get_admin_context(),
@ -507,7 +507,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
# and no use_dvs_features configured
data = {'network': {'qos_policy_id': _uuid()}}
with self.network() as net:
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
self.assertRaises(n_exc.InvalidInput,
plugin.update_network,
context.get_admin_context(),
@ -626,7 +626,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
self.assertTrue(fake_dvs_update.called)
def test_create_network_with_bad_az_hint(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
ctx = context.get_admin_context()
data = {'network': {
'name': 'test-qos',
@ -644,7 +644,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
az_name = 'az7'
az_config = az_name + ':respool-7:datastore-7:False'
cfg.CONF.set_override('availability_zones', [az_config], group="nsxv")
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
p._availability_zones_data = nsx_az.ConfiguredAvailabilityZones()
ctx = context.get_admin_context()
@ -708,7 +708,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
tenid = p['port']['tenant_id']
ctx = context.Context(user_id=None, tenant_id=tenid,
is_admin=False)
pl = manager.NeutronManager.get_plugin()
pl = directory.get_plugin()
count = pl.get_ports_count(ctx, filters={'tenant_id': [tenid]})
# Each port above has subnet => we have an additional port
# for DHCP
@ -1015,7 +1015,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
self.assertEqual('testhosttemp', port[portbindings.HOST_ID])
def test_ports_vif_details(self):
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
cfg.CONF.set_default('allow_overlapping_ips', True)
with self.subnet(enable_dhcp=False) as subnet,\
self.port(subnet), self.port(subnet):
@ -1217,7 +1217,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
data = {'port': {'fixed_ips': [{'subnet_id':
subnet['subnet']['id'],
'ip_address': "10.0.0.10"}]}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
ctx = context.get_admin_context()
with mock.patch.object(
plugin.edge_manager,
@ -1235,7 +1235,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
data = {'port': {'fixed_ips': [{'subnet_id':
subnet['subnet']['id'],
'ip_address': "10.0.0.10"}]}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
with mock.patch.object(
plugin.edge_manager,
'delete_dhcp_binding') as delete_dhcp:
@ -1261,7 +1261,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
'fixed_ips': [{'subnet_id':
subnet['subnet']['id'],
'ip_address': "10.0.0.10"}]}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
self.assertRaises(n_exc.BadRequest,
plugin.update_port,
context.get_admin_context(),
@ -1280,7 +1280,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
data = {'port': {'fixed_ips': [{'subnet_id':
subnet['subnet']['id'],
'ip_address': new_ip}]}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
ctx = context.get_admin_context()
router_obj = router_driver.RouterSharedDriver(plugin)
with mock.patch.object(plugin, '_find_router_driver',
@ -1313,7 +1313,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
data = {'port': {'fixed_ips': [{'subnet_id':
subnet['subnet']['id'],
'ip_address': new_ip}]}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
ctx = context.get_admin_context()
router_obj = router_driver.RouterSharedDriver(plugin)
with mock.patch.object(plugin, '_find_router_driver',
@ -1342,7 +1342,7 @@ class TestPortsV2(NsxVPluginV2TestCase,
device_owner=owner,
fixed_ips=[{'ip_address': old_ip}]) as port:
data = {'port': {'fixed_ips': []}}
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
ctx = context.get_admin_context()
router_obj = router_driver.RouterSharedDriver(plugin)
with mock.patch.object(plugin, '_find_router_driver',
@ -1823,14 +1823,14 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxVPluginV2TestCase):
ext_mgr = ext_mgr or TestL3ExtensionManager()
super(L3NatTest, self).setUp(
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
self.plugin_instance = manager.NeutronManager.get_plugin()
self.plugin_instance = directory.get_plugin()
self._plugin_name = "%s.%s" % (
self.plugin_instance.__module__,
self.plugin_instance.__class__.__name__)
self._plugin_class = self.plugin_instance.__class__
def tearDown(self):
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
_manager = plugin.nsx_v.task_manager
# wait max ~10 seconds for all tasks to be finished
for i in range(100):
@ -2499,7 +2499,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
self.assertEqual(net['network'][k], v)
def test_create_router_fail_at_the_backend(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
edge_manager = p.edge_manager
with mock.patch.object(edge_manager, 'create_lrouter',
side_effect=[n_exc.NeutronException]):
@ -2597,7 +2597,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
new_name = 'new_name'
router_id = r['router']['id']
# get the edge of this router
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
router_obj = ex_router_driver.RouterExclusiveDriver(plugin)
ctx = context.get_admin_context()
edge_id = router_obj._get_edge_id_or_raise(ctx, router_id)
@ -2618,7 +2618,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
new_size = 'large'
router_id = r['router']['id']
# get the edge of this router
plugin = manager.NeutronManager.get_plugin()
plugin = directory.get_plugin()
router_obj = ex_router_driver.RouterExclusiveDriver(plugin)
ctx = context.get_admin_context()
edge_id = router_obj._get_edge_id_or_raise(ctx, router_id)
@ -2737,7 +2737,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
constants.FLOATINGIP_STATUS_DOWN)
def test_update_floatingip_with_edge_router_update_failure(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with self.subnet() as subnet,\
self.port(subnet=subnet) as p1,\
self.port(subnet=subnet) as p2:
@ -2759,7 +2759,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
self.assertEqual(len(res['floatingips']), 0)
def test_create_floatingip_with_edge_router_update_failure(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
with self.subnet(cidr='200.0.0.0/24') as public_sub:
public_network_id = public_sub['subnet']['network_id']
self._set_net_external(public_network_id)
@ -3063,7 +3063,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
status=400, header={'status': 200}, uri='fake_url', response='')
def test_create_router_with_update_error(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
# make sure there is an available edge so we will use backend update
available_edge = {'edge_id': 'edge-11', 'router_id': 'fake_id'}
@ -3093,7 +3093,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
self.assertEqual(plugin_const.ERROR, new_router['status'])
def test_create_router_with_bad_az_hint(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
router = {'router': {'admin_state_up': True,
'name': 'e161be1d-0d0d-4046-9823-5a593d94f72c',
'tenant_id': context.get_admin_context().tenant_id,
@ -3108,7 +3108,7 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
az_name = 'az7'
az_config = az_name + ':respool-7:datastore-7:True'
cfg.CONF.set_override('availability_zones', [az_config], group="nsxv")
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
p._availability_zones_data = nsx_az.ConfiguredAvailabilityZones()
p._get_edge_id_by_rtr_id = p.real_get_edge
@ -3167,7 +3167,7 @@ class NsxVSecurityGroupsTestCase(ext_sg.SecurityGroupDBTestCase):
c_utils.spawn_n = mock.Mock(side_effect=lambda f: f())
super(NsxVSecurityGroupsTestCase, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
self.plugin = manager.NeutronManager.get_plugin()
self.plugin = directory.get_plugin()
self.addCleanup(self.fc2.reset_all)
@ -3182,7 +3182,7 @@ class NsxVTestSecurityGroup(ext_sg.TestSecurityGroups,
super(NsxVTestSecurityGroup, self).setUp(
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
plugin_instance = manager.NeutronManager.get_plugin()
plugin_instance = directory.get_plugin()
plugin_instance._get_edge_id_by_rtr_id = mock.Mock()
plugin_instance._get_edge_id_by_rtr_id.return_value = False
@ -3198,7 +3198,7 @@ class NsxVTestSecurityGroup(ext_sg.TestSecurityGroups,
self._delete('ports', port['id'])
def test_vnic_security_group_membership(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
self.fc2.add_member_to_security_group = (
mock.Mock().add_member_to_security_group)
self.fc2.remove_member_from_security_group = (
@ -3338,7 +3338,7 @@ class TestVdrTestCase(L3NatTest, L3NatTestCaseBase,
self.skipTest('skipped')
def test_create_router_fail_at_the_backend(self):
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
edge_manager = p.edge_manager
with mock.patch.object(edge_manager, 'create_lrouter',
side_effect=[n_exc.NeutronException]):
@ -3519,7 +3519,7 @@ class TestVdrTestCase(L3NatTest, L3NatTestCaseBase,
az_name = 'az7'
az_config = az_name + ':respool-7:datastore-7:False'
cfg.CONF.set_override('availability_zones', [az_config], group="nsxv")
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
p._availability_zones_data = nsx_az.ConfiguredAvailabilityZones()
# create a router with/without hints
@ -3980,7 +3980,7 @@ class TestNSXPortSecurity(test_psec.TestPortSecurity,
def test_service_insertion(self):
# init the plugin mocks
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
self.fc2.add_member_to_security_group = (
mock.Mock().add_member_to_security_group)
self.fc2.remove_member_from_security_group = (
@ -4024,7 +4024,7 @@ class TestNSXPortSecurity(test_psec.TestPortSecurity,
self._add_vnic_to_port(port3['port']['id'], True, 3)
# init the plugin mocks
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
self.fc2.add_member_to_security_group = (
mock.Mock().add_member_to_security_group)
@ -4727,7 +4727,7 @@ class TestSharedRouterTestCase(L3NatTest, L3NatTestCaseBase,
az_name = 'az7'
az_config = az_name + ':respool-7:datastore-7:True'
cfg.CONF.set_override('availability_zones', [az_config], group="nsxv")
p = manager.NeutronManager.get_plugin()
p = directory.get_plugin()
p._availability_zones_data = nsx_az.ConfiguredAvailabilityZones()
# create a router with/without hints
@ -4782,10 +4782,8 @@ class TestRouterFlavorTestCase(extension.ExtensionTestCase,
service_plugins = {plugin_const.FLAVORS: self.FLAVOR_PLUGIN}
super(TestRouterFlavorTestCase, self).setUp(
plugin=plugin, service_plugins=service_plugins)
self.plugin = manager.NeutronManager.get_plugin()
self.plugin._flv_plugin = (
manager.NeutronManager.get_service_plugins().
get(plugin_const.FLAVORS))
self.plugin = directory.get_plugin()
self.plugin._flv_plugin = directory.get_plugin(plugin_const.FLAVORS)
self.plugin._process_router_flavor_create = mock.Mock()
# init the availability zones

4
vmware_nsx/tests/unit/nsx_v3/test_api_replay.py
View File

@ -15,7 +15,7 @@
from vmware_nsx.tests.unit.nsx_v3 import test_plugin
from neutron import manager
from neutron_lib.plugins import directory
from oslo_config import cfg
@ -32,7 +32,7 @@ class TestApiReplay(test_plugin.NsxV3PluginTestCaseMixin):
cfg.CONF.set_override('api_replay_mode', False)
# remove the extension from the plugin
manager.NeutronManager.get_plugin().supported_extension_aliases.remove(
directory.get_plugin().supported_extension_aliases.remove(
'api-replay')
super(TestApiReplay, self).tearDown()

15
vmware_nsx/tests/unit/nsx_v3/test_plugin.py
View File

@ -27,7 +27,6 @@ from neutron.extensions import l3_ext_gw_mode
from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet
from neutron.extensions import securitygroup as secgrp
from neutron import manager
from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.extensions import test_extra_dhcp_opt as test_dhcpopts
@ -40,6 +39,7 @@ from neutron.tests.unit.scheduler \
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_utils import uuidutils
@ -258,7 +258,7 @@ class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin,
def setUp(self):
super(TestPortsV2, self).setUp()
self.plugin = manager.NeutronManager.get_plugin()
self.plugin = directory.get_plugin()
self.ctx = context.get_admin_context()
def test_update_port_delete_ip(self):
@ -408,7 +408,7 @@ class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin,
get_profile.assert_called_once_with(self.ctx, policy_id)
def _get_ports_with_fields(self, tenid, fields, expected_count):
pl = manager.NeutronManager.get_plugin()
pl = directory.get_plugin()
ctx = context.Context(user_id=None, tenant_id=tenid,
is_admin=False)
ports = pl.get_ports(ctx, filters={'tenant_id': [tenid]},
@ -440,12 +440,13 @@ class NSXv3DHCPAgentAZAwareWeightSchedulerTestCase(
def setUp(self):
super(NSXv3DHCPAgentAZAwareWeightSchedulerTestCase, self).setUp()
self.plugin = manager.NeutronManager.get_plugin()
self.plugin = directory.get_plugin()
self.ctx = context.get_admin_context()
def setup_coreplugin(self, core_plugin=None):
def setup_coreplugin(self, core_plugin=None, load_plugins=True):
super(NSXv3DHCPAgentAZAwareWeightSchedulerTestCase,
self).setup_coreplugin(core_plugin=PLUGIN_NAME)
self).setup_coreplugin(core_plugin=PLUGIN_NAME,
load_plugins=load_plugins)
class TestL3ExtensionManager(object):
@ -495,7 +496,7 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxV3PluginTestCaseMixin):
ext_mgr = ext_mgr or TestL3ExtensionManager()
super(L3NatTest, self).setUp(
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
self.plugin_instance = manager.NeutronManager.get_plugin()
self.plugin_instance = directory.get_plugin()
self._plugin_name = "%s.%s" % (
self.plugin_instance.__module__,
self.plugin_instance.__class__.__name__)

17
vmware_nsx/tests/unit/services/flowclassifier/test_nsxv_driver.py
View File

@ -24,8 +24,8 @@ from neutron.api import extensions as api_ext
from neutron.common import config
from neutron import context
from neutron.extensions import portbindings
from neutron import manager
from neutron.plugins.ml2.drivers import type_vxlan # noqa
from neutron_lib.plugins import directory
from networking_sfc.db import flowclassifier_db as fdb
from networking_sfc.extensions import flowclassifier
@ -118,11 +118,18 @@ class TestNsxvFlowClassifierDriver(
self.assertEqual(self._profile_id, self.driver._profile_id)
self.assertEqual(self.driver._security_group_id, '0')
orig_get_plugin = directory.get_plugin
def mocked_get_plugin(plugin=None):
# mock only the core plugin
if plugin:
return orig_get_plugin(plugin)
return mock_nsxv_plugin
mock_nsxv_plugin = mock.Mock()
fc_plugin = manager.NeutronManager.get_service_plugins().get(
flowclassifier.FLOW_CLASSIFIER_EXT)
with mock.patch.object(manager.NeutronManager, 'get_plugin',
return_value=mock_nsxv_plugin):
fc_plugin = directory.get_plugin(flowclassifier.FLOW_CLASSIFIER_EXT)
with mock.patch.object(directory, 'get_plugin',
new=mocked_get_plugin):
with mock.patch.object(
mock_nsxv_plugin,
'add_vms_to_service_insertion') as fake_add:

4
vmware_nsx/tests/unit/services/qos/test_nsxv_notification.py
View File

@ -18,12 +18,12 @@ from oslo_config import cfg
from oslo_utils import uuidutils
from neutron import context
from neutron import manager
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.services.qos import qos_consts
from neutron.services.qos import qos_plugin
from neutron.tests.unit.services.qos import base
from neutron_lib.plugins import directory
from vmware_nsx.dvs import dvs
from vmware_nsx.dvs import dvs_utils
@ -44,7 +44,7 @@ class TestQosNsxVNotification(test_plugin.NsxVPluginV2TestCase,
self._init_dvs_config()
super(TestQosNsxVNotification, self).setUp(plugin=CORE_PLUGIN,
ext_mgr=None)
plugin_instance = manager.NeutronManager.get_plugin()
plugin_instance = directory.get_plugin()
self._core_plugin = plugin_instance
self._core_plugin.init_is_complete = True