Remove plugin_name_v2 and extension_manager in test_config

There are two ways to specify a core plugin and an extension manager
in the unit tests: test_config and arguments of the constructor.
Both are used and it sometimes makes it a bit difficult to debug.
This patch removes the way via test_config and makes constructor
arguments the only way to do it.

Also removes the default entries in test_config because they are
not used anywhere.

Change-Id: I491003f4246e62001e9b74743c8fbfa922088383
Closes-Bug: #1245335
This commit is contained in:
Akihiro MOTOKI 2013-10-26 21:53:21 +09:00 committed by Akihiro Motoki
parent affb042639
commit 237b24b0c8
20 changed files with 76 additions and 169 deletions

View File

@ -37,15 +37,8 @@
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from neutron.common import constants
# describes parameters used by different unit/functional tests
# a plugin-specific testing mechanism should import this dictionary
# and override the values in it if needed (e.g., run_tests.py in
# neutron/plugins/openvswitch/ )
test_config = {
"plugin_name": "",
"default_net_op_status": constants.NET_STATUS_ACTIVE,
"default_port_op_status": constants.PORT_STATUS_ACTIVE,
}
test_config = {}

View File

@ -20,7 +20,6 @@
import contextlib
import copy
import os
from mock import patch
from oslo.config import cfg
@ -30,8 +29,6 @@ from neutron.common.test_lib import test_config
from neutron import context
from neutron.extensions import l3
from neutron.manager import NeutronManager
from neutron.openstack.common.notifier import api as notifier_api
from neutron.openstack.common.notifier import test_notifier
from neutron.openstack.common import uuidutils
from neutron.plugins.bigswitch.extensions import routerrule
from neutron.tests.unit.bigswitch import fake_server
@ -44,25 +41,6 @@ from neutron.tests.unit import test_l3_plugin
_uuid = uuidutils.generate_uuid
def new_L3_setUp(self):
test_config['plugin_name_v2'] = (
'neutron.plugins.bigswitch.plugin.NeutronRestProxyV2')
etc_path = os.path.join(os.path.dirname(__file__), 'etc')
rp_conf_file = os.path.join(etc_path, 'restproxy.ini.test')
test_config['config_files'] = [rp_conf_file]
cfg.CONF.set_default('allow_overlapping_ips', False)
ext_mgr = RouterRulesTestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(test_l3_plugin.L3BaseForIntTests, self).setUp()
# Set to None to reload the drivers
notifier_api._drivers = None
cfg.CONF.set_override("notification_driver", [test_notifier.__name__])
origSetUp = test_l3_plugin.L3NatDBIntTestCase.setUp
class RouterRulesTestExtensionManager(object):
def get_resources(self):
@ -92,15 +70,16 @@ class RouterDBTestCase(test_base.BigSwitchTestBase,
def setUp(self):
self.setup_patches()
test_l3_plugin.L3NatDBIntTestCase.setUp = new_L3_setUp
super(RouterDBTestCase, self).setUp()
self.setup_config_files()
ext_mgr = RouterRulesTestExtensionManager()
super(RouterDBTestCase, self).setUp(plugin=self._plugin_name,
ext_mgr=ext_mgr)
cfg.CONF.set_default('allow_overlapping_ips', False)
self.plugin_obj = NeutronManager.get_plugin()
def tearDown(self):
super(RouterDBTestCase, self).tearDown()
del test_config['plugin_name_v2']
del test_config['config_files']
test_l3_plugin.L3NatDBIntTestCase.setUp = origSetUp
def test_router_remove_router_interface_wrong_subnet_returns_400(self):
with self.router() as r:

View File

@ -22,7 +22,6 @@ from oslo.config import cfg
from neutron.api import extensions as neutron_extensions
from neutron.api.v2 import attributes
from neutron.common.test_lib import test_config
from neutron import context
import neutron.db.api as db
from neutron.plugins.cisco.db import n1kv_db_v2
@ -203,12 +202,9 @@ class N1kvPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
n1kv_neutron_plugin.N1kvNeutronPluginV2._setup_vsm = _fake_setup_vsm
test_config['plugin_name_v2'] = self._plugin_name
neutron_extensions.append_api_extensions_path(extensions.__path__)
self.addCleanup(cfg.CONF.reset)
ext_mgr = NetworkProfileTestExtensionManager()
test_config['extension_manager'] = ext_mgr
self.addCleanup(self.restore_test_config)
# Save the original RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {}
@ -221,7 +217,8 @@ class N1kvPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
n1kv_profile.EXTENDED_ATTRIBUTES_2_0["ports"])
self.addCleanup(self.restore_resource_attribute_map)
self.addCleanup(db.clear_db)
super(N1kvPluginTestCase, self).setUp(self._plugin_name)
super(N1kvPluginTestCase, self).setUp(self._plugin_name,
ext_mgr=ext_mgr)
# Create some of the database entries that we require.
self._make_test_profile()
self._make_test_policy_profile()
@ -230,10 +227,6 @@ class N1kvPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
# Restore the original RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
def restore_test_config(self):
# Restore the original test_config
del test_config['plugin_name_v2']
def test_plugin(self):
self._make_network('json',
'some_net',

View File

@ -100,12 +100,13 @@ class NecPluginV2TestCase(NecPluginV2TestCaseBase,
'port_added': added, 'port_removed': removed}
self.callback_nec.update_ports(self.context, **kwargs)
def setUp(self):
def setUp(self, plugin=None, ext_mgr=None):
self.addCleanup(mock.patch.stopall)
self._set_nec_ini()
self.addCleanup(self._clean_nec_ini)
super(NecPluginV2TestCase, self).setUp(self._plugin_name)
plugin = plugin or self._plugin_name
super(NecPluginV2TestCase, self).setUp(plugin, ext_mgr=ext_mgr)
self.plugin = manager.NeutronManager.get_plugin()
self.plugin.ofc = fake_ofc_manager.patch_ofc_manager()

View File

@ -19,7 +19,6 @@ import mock
import webob.exc
from neutron.api.v2 import attributes
from neutron.common.test_lib import test_config
from neutron import context
from neutron.plugins.nec.common import exceptions as nexc
from neutron.plugins.nec.extensions import packetfilter
@ -54,8 +53,8 @@ class TestNecPluginPacketFilter(test_nec_plugin.NecPluginV2TestCase):
_nec_ini = NEC_PLUGIN_PF_INI
def setUp(self):
test_config['extension_manager'] = PacketfilterExtensionManager()
super(TestNecPluginPacketFilter, self).setUp()
ext_mgr = PacketfilterExtensionManager()
super(TestNecPluginPacketFilter, self).setUp(ext_mgr=ext_mgr)
def _create_packet_filter(self, fmt, net_id, expected_res_status=None,
arg_list=None, **kwargs):

View File

@ -33,7 +33,6 @@ class NVPDhcpAgentNotifierTestCase(test_base.OvsDhcpAgentNotifierTestCase):
plugin_str = PLUGIN_NAME
def setUp(self):
test_config['plugin_name_v2'] = PLUGIN_NAME
test_config['config_files'] = [get_fake_conf('nsx.ini.full.test')]
# mock nvp api client

View File

@ -59,14 +59,12 @@ class MacLearningDBTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
self.adminContext = context.get_admin_context()
test_config['config_files'] = [get_fake_conf('nsx.ini.full.test')]
test_config['plugin_name_v2'] = PLUGIN_NAME
cfg.CONF.set_override('api_extensions_path', NVPEXT_PATH)
# Save the original RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {}
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
self.saved_attr_map[resource] = attrs.copy()
ext_mgr = MacLearningExtensionManager()
test_config['extension_manager'] = ext_mgr
# mock nvp api client
self.fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
@ -87,7 +85,8 @@ class MacLearningDBTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
self.addCleanup(patch_sync.stop)
self.addCleanup(self.restore_resource_attribute_map)
self.addCleanup(cfg.CONF.reset)
super(MacLearningDBTestCase, self).setUp()
super(MacLearningDBTestCase, self).setUp(plugin=PLUGIN_NAME,
ext_mgr=ext_mgr)
def restore_resource_attribute_map(self):
# Restore the original RESOURCE_ATTRIBUTE_MAP

View File

@ -25,7 +25,6 @@ from neutron.api import extensions
from neutron.api.extensions import PluginAwareExtensionManager
from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import api as db_api
from neutron.db import db_base_plugin_v2
@ -246,13 +245,14 @@ class NetworkGatewayExtensionTestCase(base.BaseTestCase):
class NetworkGatewayDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
"""Unit tests for Network Gateway DB support."""
def setUp(self):
test_config['plugin_name_v2'] = '%s.%s' % (
__name__, TestNetworkGatewayPlugin.__name__)
def setUp(self, plugin=None, ext_mgr=None):
if not plugin:
plugin = '%s.%s' % (__name__, TestNetworkGatewayPlugin.__name__)
if not ext_mgr:
ext_mgr = TestExtensionManager()
test_config['extension_manager'] = ext_mgr
self.resource = networkgw.RESOURCE_NAME.replace('-', '_')
super(NetworkGatewayDbTestCase, self).setUp()
super(NetworkGatewayDbTestCase, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
def _create_network_gateway(self, fmt, tenant_id, name=None,
devices=None, arg_list=None, **kwargs):

View File

@ -119,7 +119,6 @@ class NiciraPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
self.mock_instance.return_value.get_nvp_version.return_value = (
NVPVersion("2.9"))
self.mock_instance.return_value.request.side_effect = _fake_request
plugin = plugin or PLUGIN_NAME
super(NiciraPluginV2TestCase, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
cfg.CONF.set_override('metadata_mode', None, 'NSX')
@ -400,8 +399,8 @@ class TestNiciraPortSecurity(NiciraPortSecurityTestCase,
pass
class TestNiciraAllowedAddressPairs(test_addr_pair.TestAllowedAddressPairs,
NiciraPluginV2TestCase):
class TestNiciraAllowedAddressPairs(NiciraPluginV2TestCase,
test_addr_pair.TestAllowedAddressPairs):
pass
@ -480,7 +479,7 @@ class NiciraL3NatTest(test_l3_plugin.L3BaseForIntTests,
def _restore_l3_attribute_map(self):
l3.RESOURCE_ATTRIBUTE_MAP = self._l3_attribute_map_bk
def setUp(self, plugin=None, ext_mgr=None, service_plugins=None):
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None, service_plugins=None):
self._l3_attribute_map_bk = {}
for item in l3.RESOURCE_ATTRIBUTE_MAP:
self._l3_attribute_map_bk[item] = (
@ -1248,8 +1247,7 @@ class NiciraNeutronNVPOutOfSync(NiciraPluginV2TestCase,
def setUp(self):
ext_mgr = test_l3_plugin.L3TestExtensionManager()
test_lib.test_config['extension_manager'] = ext_mgr
super(NiciraNeutronNVPOutOfSync, self).setUp()
super(NiciraNeutronNVPOutOfSync, self).setUp(ext_mgr=ext_mgr)
def test_delete_network_not_in_nvp(self):
res = self._create_network('json', 'net1', True)
@ -1404,12 +1402,13 @@ class NiciraNeutronNVPOutOfSync(NiciraPluginV2TestCase,
constants.NET_STATUS_ERROR)
class TestNiciraNetworkGateway(test_l2_gw.NetworkGatewayDbTestCase,
NiciraPluginV2TestCase):
class TestNiciraNetworkGateway(NiciraPluginV2TestCase,
test_l2_gw.NetworkGatewayDbTestCase):
def setUp(self):
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None):
cfg.CONF.set_override('api_extensions_path', NVPEXT_PATH)
super(TestNiciraNetworkGateway, self).setUp()
super(TestNiciraNetworkGateway,
self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def test_create_network_gateway_name_exceeds_40_chars(self):
name = 'this_is_a_gateway_whose_name_is_longer_than_40_chars'

View File

@ -17,7 +17,6 @@
import mock
from neutron.api.v2 import attributes as attr
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import agents_db
from neutron.db import l3_agentschedulers_db
@ -67,11 +66,12 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
def setUp(self):
service_plugins = {'metering_plugin_name': DB_METERING_PLUGIN_KLASS}
test_config['plugin_name_v2'] = ('neutron.tests.unit.test_l3_plugin.'
'TestL3NatIntPlugin')
plugin = 'neutron.tests.unit.test_l3_plugin.TestL3NatIntPlugin'
ext_mgr = MeteringTestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(TestMeteringPlugin, self).setUp(service_plugins=service_plugins)
super(TestMeteringPlugin, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
self.addCleanup(mock.patch.stopall)
self.uuid = '654f6b9d-0f36-4ae5-bd1b-01616794ca60'
@ -92,14 +92,6 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
self.topic = 'metering_agent'
def tearDown(self):
self.uuid_patch.stop()
self.fanout_patch.stop()
self.context_patch.stop()
del test_config['extension_manager']
del test_config['plugin_name_v2']
super(TestMeteringPlugin, self).tearDown()
def test_add_metering_label_rpc_call(self):
second_uuid = 'e27fe2df-376e-4ac7-ae13-92f050a21f84'
expected = {'args': {'routers': [{'status': 'ACTIVE',
@ -278,18 +270,17 @@ class TestMeteringPluginL3AgentScheduler(
def setUp(self):
service_plugins = {'metering_plugin_name': DB_METERING_PLUGIN_KLASS}
plugin_str = ('neutron.tests.unit.services.metering.'
'test_metering_plugin.TestRouteIntPlugin')
test_config['plugin_name_v2'] = plugin_str
ext_mgr = MeteringTestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(TestMeteringPluginL3AgentScheduler,
self).setUp(service_plugins=service_plugins)
self).setUp(plugin=plugin_str, ext_mgr=ext_mgr,
service_plugins=service_plugins)
self.uuid = '654f6b9d-0f36-4ae5-bd1b-01616794ca60'
self.addCleanup(mock.patch.stopall)
uuid = 'neutron.openstack.common.uuidutils.generate_uuid'
self.uuid_patch = mock.patch(uuid, return_value=self.uuid)
self.mock_uuid = self.uuid_patch.start()
@ -310,15 +301,6 @@ class TestMeteringPluginL3AgentScheduler(
self.topic = 'metering_agent'
def tearDown(self):
self.uuid_patch.stop()
self.cast_patch.stop()
self.context_patch.stop()
self.l3routers_patch.stop()
del test_config['extension_manager']
del test_config['plugin_name_v2']
super(TestMeteringPluginL3AgentScheduler, self).tearDown()
def test_add_metering_label_rpc_call(self):
second_uuid = 'e27fe2df-376e-4ac7-ae13-92f050a21f84'
expected = {'args': {'routers': [{'status': 'ACTIVE',

View File

@ -23,7 +23,6 @@ from webob import exc
from neutron.api.v2 import attributes
from neutron.common import constants
from neutron.common.test_lib import test_config
from neutron.common import topics
from neutron import context
from neutron.db import agents_db
@ -169,8 +168,7 @@ class AgentDBTestCase(AgentDBTestMixIn,
def setUp(self):
self.adminContext = context.get_admin_context()
test_config['plugin_name_v2'] = (
'neutron.tests.unit.test_agent_ext_plugin.TestAgentPlugin')
plugin = 'neutron.tests.unit.test_agent_ext_plugin.TestAgentPlugin'
# for these tests we need to enable overlapping ips
cfg.CONF.set_default('allow_overlapping_ips', True)
# Save the original RESOURCE_ATTRIBUTE_MAP
@ -178,10 +176,9 @@ class AgentDBTestCase(AgentDBTestMixIn,
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
self.saved_attr_map[resource] = attrs.copy()
ext_mgr = AgentTestExtensionManager()
test_config['extension_manager'] = ext_mgr
self.addCleanup(self.restore_resource_attribute_map)
self.addCleanup(cfg.CONF.reset)
super(AgentDBTestCase, self).setUp()
super(AgentDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def restore_resource_attribute_map(self):
# Restore the originak RESOURCE_ATTRIBUTE_MAP

View File

@ -99,7 +99,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
self._tenant_id = 'test-tenant'
if not plugin:
plugin = test_config.get('plugin_name_v2', DB_PLUGIN_KLASS)
plugin = DB_PLUGIN_KLASS
# Create the default configurations
args = ['--config-file', etcdir('neutron.conf.test')]
@ -152,7 +152,6 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
native_sorting_attr_name, False))
self._skip_native_sorting = not _is_native_sorting_support()
ext_mgr = ext_mgr or test_config.get('extension_manager')
if ext_mgr:
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)

View File

@ -15,7 +15,6 @@
from neutron.api.v2 import attributes as attr
from neutron.common.test_lib import test_config
from neutron.db import allowedaddresspairs_db as addr_pair_db
from neutron.db import db_base_plugin_v2
from neutron.db import portsecurity_db
@ -29,8 +28,8 @@ DB_PLUGIN_KLASS = ('neutron.tests.unit.test_extension_allowedaddresspairs.'
class AllowedAddressPairTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
def setUp(self, plugin=None):
super(AllowedAddressPairTestCase, self).setUp()
def setUp(self, plugin=None, ext_mgr=None):
super(AllowedAddressPairTestCase, self).setUp(plugin)
# Check if a plugin supports security groups
plugin_obj = NeutronManager.get_plugin()
@ -90,13 +89,10 @@ class AllowedAddressPairTestPlugin(portsecurity_db.PortSecurityDbMixin,
class AllowedAddressPairDBTestCase(AllowedAddressPairTestCase):
def setUp(self, plugin=None):
test_config['plugin_name_v2'] = DB_PLUGIN_KLASS
super(AllowedAddressPairDBTestCase, self).setUp()
def tearDown(self):
del test_config['plugin_name_v2']
super(AllowedAddressPairDBTestCase, self).tearDown()
def setUp(self, plugin=None, ext_mgr=None):
plugin = plugin or DB_PLUGIN_KLASS
super(AllowedAddressPairDBTestCase,
self).setUp(plugin=plugin, ext_mgr=ext_mgr)
class TestAllowedAddressPairs(AllowedAddressPairDBTestCase):

View File

@ -21,7 +21,6 @@ import itertools
import testtools
from webob import exc
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import models_v2
from neutron.extensions import external_net as external_net
@ -64,11 +63,9 @@ class ExtNetDBTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
fmt, name, admin_state_up, arg_list=arg_list, **new_args)
def setUp(self):
test_config['plugin_name_v2'] = (
'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin')
plugin = 'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin'
ext_mgr = ExtNetTestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(ExtNetDBTestCase, self).setUp()
super(ExtNetDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def _set_net_external(self, net_id):
self._update('networks', net_id,

View File

@ -19,7 +19,6 @@ import contextlib
from oslo.config import cfg
from webob import exc
from neutron.common.test_lib import test_config
from neutron.db import extraroute_db
from neutron.extensions import extraroute
from neutron.extensions import l3
@ -460,20 +459,16 @@ class ExtraRouteDBTestCaseBase(object):
class ExtraRouteDBIntTestCase(test_l3.L3NatDBIntTestCase,
ExtraRouteDBTestCaseBase):
def setUp(self, plugin=None):
def setUp(self, plugin=None, ext_mgr=None):
if not plugin:
plugin = ('neutron.tests.unit.test_extension_extraroute.'
'TestExtraRouteIntPlugin')
test_config['plugin_name_v2'] = plugin
# for these tests we need to enable overlapping ips
cfg.CONF.set_default('allow_overlapping_ips', True)
cfg.CONF.set_default('max_routes', 3)
ext_mgr = ExtraRouteTestExtensionManager()
test_config['extension_manager'] = ext_mgr
# L3NatDBIntTestCase will overwrite plugin_name_v2,
# so we don't need to setUp on the class here
super(test_l3.L3BaseForIntTests, self).setUp()
super(test_l3.L3BaseForIntTests, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
# Set to None to reload the drivers
notifier_api._drivers = None
cfg.CONF.set_override("notification_driver", [test_notifier.__name__])
@ -487,8 +482,7 @@ class ExtraRouteDBSepTestCase(test_l3.L3NatDBSepTestCase,
ExtraRouteDBTestCaseBase):
def setUp(self):
# the plugin without L3 support
test_config['plugin_name_v2'] = (
'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin')
plugin = 'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin'
# the L3 service plugin
l3_plugin = ('neutron.tests.unit.test_extension_extraroute.'
'TestExtraRouteL3NatServicePlugin')
@ -498,10 +492,8 @@ class ExtraRouteDBSepTestCase(test_l3.L3NatDBSepTestCase,
cfg.CONF.set_default('allow_overlapping_ips', True)
cfg.CONF.set_default('max_routes', 3)
ext_mgr = ExtraRouteTestExtensionManager()
test_config['extension_manager'] = ext_mgr
# L3NatDBSepTestCase will overwrite plugin_name_v2,
# so we don't need to setUp on the class here
super(test_l3.L3BaseForSepTests, self).setUp(
plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
# Set to None to reload the drivers

View File

@ -15,7 +15,6 @@
from neutron.api.v2 import attributes as attr
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import db_base_plugin_v2
from neutron.db import portsecurity_db
@ -31,7 +30,7 @@ DB_PLUGIN_KLASS = ('neutron.tests.unit.test_extension_portsecurity.'
class PortSecurityTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
def setUp(self, plugin=None):
super(PortSecurityTestCase, self).setUp()
super(PortSecurityTestCase, self).setUp(plugin)
# Check if a plugin supports security groups
plugin_obj = NeutronManager.get_plugin()
@ -161,12 +160,8 @@ class PortSecurityTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
class PortSecurityDBTestCase(PortSecurityTestCase):
def setUp(self, plugin=None):
test_config['plugin_name_v2'] = plugin or DB_PLUGIN_KLASS
super(PortSecurityDBTestCase, self).setUp()
def tearDown(self):
del test_config['plugin_name_v2']
super(PortSecurityDBTestCase, self).tearDown()
plugin = plugin or DB_PLUGIN_KLASS
super(PortSecurityDBTestCase, self).setUp(plugin)
class TestPortSecurity(PortSecurityDBTestCase):

View File

@ -21,7 +21,6 @@ import webob.exc
from neutron.api.v2 import attributes as attr
from neutron.common import constants as const
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import db_base_plugin_v2
from neutron.db import securitygroups_db
@ -239,11 +238,11 @@ class SecurityGroupTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
class SecurityGroupDBTestCase(SecurityGroupsTestCase):
def setUp(self, plugin=None):
def setUp(self, plugin=None, ext_mgr=None):
plugin = plugin or DB_PLUGIN_KLASS
ext_mgr = SecurityGroupTestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(SecurityGroupDBTestCase, self).setUp(plugin)
ext_mgr = ext_mgr or SecurityGroupTestExtensionManager()
super(SecurityGroupDBTestCase,
self).setUp(plugin=plugin, ext_mgr=ext_mgr)
class TestSecurityGroups(SecurityGroupDBTestCase):

View File

@ -31,7 +31,6 @@ from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import constants as l3_constants
from neutron.common import exceptions as q_exc
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import api as qdbapi
from neutron.db import db_base_plugin_v2
@ -1728,12 +1727,11 @@ class L3AgentDbTestCaseBase(L3NatTestCaseMixin):
class L3BaseForIntTests(test_db_plugin.NeutronDbPluginV2TestCase):
def setUp(self, plugin=None, ext_mgr=None, service_plugins=None):
test_config['plugin_name_v2'] = (
'neutron.tests.unit.test_l3_plugin.TestL3NatIntPlugin')
if not plugin:
plugin = 'neutron.tests.unit.test_l3_plugin.TestL3NatIntPlugin'
# for these tests we need to enable overlapping ips
cfg.CONF.set_default('allow_overlapping_ips', True)
ext_mgr = ext_mgr or L3TestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(L3BaseForIntTests, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
@ -1743,16 +1741,15 @@ class L3BaseForIntTests(test_db_plugin.NeutronDbPluginV2TestCase):
def tearDown(self):
test_notifier.NOTIFICATIONS = []
del test_config['extension_manager']
super(L3BaseForIntTests, self).tearDown()
class L3BaseForSepTests(test_db_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
def setUp(self, plugin=None, ext_mgr=None):
# the plugin without L3 support
test_config['plugin_name_v2'] = (
'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin')
if not plugin:
plugin = 'neutron.tests.unit.test_l3_plugin.TestNoL3NatPlugin'
# the L3 service plugin
l3_plugin = ('neutron.tests.unit.test_l3_plugin.'
'TestL3NatServicePlugin')
@ -1760,9 +1757,10 @@ class L3BaseForSepTests(test_db_plugin.NeutronDbPluginV2TestCase):
# for these tests we need to enable overlapping ips
cfg.CONF.set_default('allow_overlapping_ips', True)
if not ext_mgr:
ext_mgr = L3TestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(L3BaseForSepTests, self).setUp(service_plugins=service_plugins)
super(L3BaseForSepTests, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
# Set to None to reload the drivers
notifier_api._drivers = None
@ -1770,7 +1768,6 @@ class L3BaseForSepTests(test_db_plugin.NeutronDbPluginV2TestCase):
def tearDown(self):
test_notifier.NOTIFICATIONS = []
del test_config['extension_manager']
super(L3BaseForSepTests, self).tearDown()

View File

@ -24,7 +24,6 @@ from oslo.config import cfg
from neutron.api.v2 import attributes as attr
from neutron.common import constants
from neutron.common.test_lib import test_config
from neutron.common import topics
from neutron import context as q_context
from neutron.db import agents_db
@ -78,12 +77,9 @@ class L3SchedulerTestCase(l3_agentschedulers_db.L3AgentSchedulerDbMixin,
test_l3_plugin.L3NatTestCaseMixin):
def setUp(self):
test_config['plugin_name_v2'] = DB_PLUGIN_KLASS
ext_mgr = L3SchedulerTestExtensionManager()
test_config['extension_manager'] = ext_mgr
super(L3SchedulerTestCase, self).setUp()
super(L3SchedulerTestCase, self).setUp(plugin=DB_PLUGIN_KLASS,
ext_mgr=ext_mgr)
self.adminContext = q_context.get_admin_context()
self.plugin = manager.NeutronManager.get_plugin()

View File

@ -23,7 +23,6 @@ import fixtures
from oslo.config import cfg
from neutron.common import config
from neutron.common.test_lib import test_config
from neutron.manager import NeutronManager
from neutron.manager import validate_post_plugin_load
from neutron.manager import validate_pre_plugin_load
@ -65,9 +64,7 @@ class NeutronManagerTestCase(base.BaseTestCase):
fixtures.MonkeyPatch('neutron.manager.NeutronManager._instance'))
def test_service_plugin_is_loaded(self):
cfg.CONF.set_override("core_plugin",
test_config.get('plugin_name_v2',
DB_PLUGIN_KLASS))
cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
cfg.CONF.set_override("service_plugins",
["neutron.tests.unit.dummy_plugin."
"DummyServicePlugin"])
@ -85,9 +82,7 @@ class NeutronManagerTestCase(base.BaseTestCase):
"DummyServicePlugin",
"neutron.tests.unit.dummy_plugin."
"DummyServicePlugin"])
cfg.CONF.set_override("core_plugin",
test_config.get('plugin_name_v2',
DB_PLUGIN_KLASS))
cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
self.assertRaises(ValueError, NeutronManager.get_instance)
def test_service_plugin_conflicts_with_core_plugin(self):