"""Base Test Case for all Unit Tests"""
import contextlib
-import gc
import logging as std_logging
import os
import os.path
import sys
-import weakref
import eventlet.timeout
import fixtures
from neutron.common import config
from neutron.common import rpc as n_rpc
-from neutron.db import agentschedulers_db
-from neutron import manager
from neutron.tests import fake_notifier
from neutron.tests import post_mortem_debug
class BaseTestCase(testtools.TestCase):
- def cleanup_core_plugin(self):
- """Ensure that the core plugin is deallocated."""
- nm = manager.NeutronManager
- if not nm.has_instance():
- return
-
- #TODO(marun) Fix plugins that do not properly initialize notifiers
- agentschedulers_db.AgentSchedulerDbMixin.agent_notifiers = {}
-
- # Perform a check for deallocation only if explicitly
- # configured to do so since calling gc.collect() after every
- # test increases test suite execution time by ~50%.
- check_plugin_deallocation = (
- os.environ.get('OS_CHECK_PLUGIN_DEALLOCATION') in TRUE_STRING)
- if check_plugin_deallocation:
- plugin = weakref.ref(nm._instance.plugin)
-
- nm.clear_instance()
-
- if check_plugin_deallocation:
- gc.collect()
-
- #TODO(marun) Ensure that mocks are deallocated?
- if plugin() and not isinstance(plugin(), mock.Base):
- self.fail('The plugin for this test was not deallocated.')
-
- def setup_coreplugin(self, core_plugin=None):
- if core_plugin is not None:
- cfg.CONF.set_override('core_plugin', core_plugin)
-
- def setup_notification_driver(self, notification_driver=None):
- self.addCleanup(fake_notifier.reset)
- if notification_driver is None:
- notification_driver = [fake_notifier.__name__]
- cfg.CONF.set_override("notification_driver", notification_driver)
-
@staticmethod
def config_parse(conf=None, args=None):
"""Create the default configurations."""
def setUp(self):
super(BaseTestCase, self).setUp()
- # Ensure plugin cleanup is triggered last so that
- # test-specific cleanup has a chance to release references.
- self.addCleanup(self.cleanup_core_plugin)
# Configure this first to ensure pm debugging support for setUp()
if os.environ.get('OS_POST_MORTEM_DEBUG') in TRUE_STRING:
from neutron.openstack.common import uuidutils
from neutron.plugins.metaplugin import meta_neutron_plugin
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
CONF_FILE = ""
META_PATH = "neutron.plugins.metaplugin"
models_v2.Port, 'metaplugin_port', None, None, None)
-class MetaNeutronPluginV2Test(testlib_api.SqlTestCase):
+class MetaNeutronPluginV2Test(testlib_api.SqlTestCase,
+ testlib_plugin.PluginSetupHelper):
"""Class conisting of MetaNeutronPluginV2 unit tests."""
has_l3 = True
from neutron.services.metering.agents import metering_agent
from neutron.tests import base
from neutron.tests import fake_notifier
+from neutron.tests.unit import testlib_plugin
_uuid = uuidutils.generate_uuid
'id': _uuid()}]
-class TestMeteringOperations(base.BaseTestCase):
+class TestMeteringOperations(base.BaseTestCase,
+ testlib_plugin.NotificationSetupHelper):
def setUp(self):
super(TestMeteringOperations, self).setUp()
from neutron.tests import base
from neutron.tests import fake_notifier
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
ROOTDIR = os.path.dirname(os.path.dirname(__file__))
self.assertEqual(link['rel'], 'self')
-class APIv2TestBase(base.BaseTestCase):
+class APIv2TestBase(base.BaseTestCase, testlib_plugin.PluginSetupHelper):
def setUp(self):
super(APIv2TestBase, self).setUp()
self.assertEqual(res.status_int, 400)
-class SubresourceTest(base.BaseTestCase):
+class SubresourceTest(base.BaseTestCase, testlib_plugin.PluginSetupHelper):
def setUp(self):
super(SubresourceTest, self).setUp()
self.assertEqual(res.status_int, exc.HTTPCreated.code)
-class ExtensionTestCase(base.BaseTestCase):
+class ExtensionTestCase(base.BaseTestCase, testlib_plugin.PluginSetupHelper):
def setUp(self):
super(ExtensionTestCase, self).setUp()
plugin = 'neutron.neutron_plugin_base_v2.NeutronPluginBaseV2'
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
-class ExtensionTestCase(testlib_api.WebTestCase):
+class ExtensionTestCase(testlib_api.WebTestCase,
+ testlib_plugin.PluginSetupHelper):
def _resotre_attr_map(self):
attributes.RESOURCE_ATTRIBUTE_MAP = self._saved_attr_map
from neutron.tests import base
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
return api_common.SortingEmulatedHelper(request, self._attr_info)
-class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
+class NeutronDbPluginV2TestCase(testlib_api.WebTestCase,
+ testlib_plugin.PluginSetupHelper):
fmt = 'json'
resource_prefix_map = {}
from neutron import manager
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
-class TestNetworks(testlib_api.SqlTestCase):
+class TestNetworks(testlib_api.SqlTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(TestNetworks, self).setUp()
self._tenant_id = 'test-tenant'
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_l3_plugin
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
_uuid = uuidutils.generate_uuid
FAKE_GW_PORT_ID = _uuid()
supported_extension_aliases = ["router", "ext-gw-mode"]
-class TestL3GwModeMixin(testlib_api.SqlTestCase):
+class TestL3GwModeMixin(testlib_api.SqlTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(TestL3GwModeMixin, self).setUp()
from neutron.tests.unit.extensions import extendedattribute as extattr
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
from neutron import wsgi
_uuid = test_api_v2._uuid
return self.objh[id]
-class ExtensionExtendedAttributeTestCase(base.BaseTestCase):
+class ExtensionExtendedAttributeTestCase(base.BaseTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(ExtensionExtendedAttributeTestCase, self).setUp()
plugin = (
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
class ProviderExtensionManager(object):
return pnet.get_extended_resources(version)
-class ProvidernetExtensionTestCase(testlib_api.WebTestCase):
+class ProvidernetExtensionTestCase(testlib_api.WebTestCase,
+ testlib_plugin.PluginSetupHelper):
fmt = 'json'
def setUp(self):
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_api_v2_extension
from neutron.tests.unit import test_db_plugin
-
+from neutron.tests.unit import testlib_plugin
LOG = logging.getLogger(__name__)
self._test_notify_op_agent(self._test_floatingips_op_agent)
-class L3BaseForIntTests(test_db_plugin.NeutronDbPluginV2TestCase):
+class L3BaseForIntTests(test_db_plugin.NeutronDbPluginV2TestCase,
+ testlib_plugin.NotificationSetupHelper):
mock_rescheduling = True
self.setup_notification_driver()
-class L3BaseForSepTests(test_db_plugin.NeutronDbPluginV2TestCase):
+class L3BaseForSepTests(test_db_plugin.NeutronDbPluginV2TestCase,
+ testlib_plugin.NotificationSetupHelper):
def setUp(self, plugin=None, ext_mgr=None):
# the plugin without L3 support
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_l3_plugin
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
HOST = 'my_l3_host'
FIRST_L3_AGENT = {
pass
-class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
+class L3DvrSchedulerTestCase(testlib_api.SqlTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
plugin = 'neutron.plugins.ml2.plugin.Ml2Plugin'
from neutron.plugins.common import constants
from neutron.tests import base
from neutron.tests.unit import dummy_plugin
+from neutron.tests.unit import testlib_plugin
LOG = logging.getLogger(__name__)
'dhcp': 'dhcp_agent_notifier'}
-class NeutronManagerTestCase(base.BaseTestCase):
+class NeutronManagerTestCase(base.BaseTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(NeutronManagerTestCase, self).setUp()
from neutron.tests import base
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
TARGET_PLUGIN = ('neutron.plugins.linuxbridge.lb_neutron_plugin'
'.LinuxBridgePluginV2')
_get_path = test_api_v2._get_path
-class QuotaExtensionTestCase(testlib_api.WebTestCase):
+class QuotaExtensionTestCase(testlib_api.WebTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(QuotaExtensionTestCase, self).setUp()
from neutron.plugins.common import constants
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
from neutron import wsgi
_uuid = test_api_v2._uuid
pass
-class RouterServiceInsertionTestCase(testlib_api.SqlTestCase):
+class RouterServiceInsertionTestCase(testlib_api.SqlTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(RouterServiceInsertionTestCase, self).setUp()
plugin = (
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
+from neutron.tests.unit import testlib_plugin
DEFAULT_SERVICE_DEFS = [{'service_class': constants.DUMMY,
return []
-class ServiceTypeExtensionTestCaseBase(testlib_api.WebTestCase):
+class ServiceTypeExtensionTestCaseBase(testlib_api.WebTestCase,
+ testlib_plugin.PluginSetupHelper):
fmt = 'json'
def setUp(self):
--- /dev/null
+# Copyright 2014 OpenStack Foundation.
+# All Rights Reserved.
+#
+# 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 gc
+import os
+import weakref
+
+import mock
+from oslo.config import cfg
+
+from neutron.db import agentschedulers_db
+from neutron import manager
+from neutron.tests import base
+from neutron.tests import fake_notifier
+
+
+class PluginSetupHelper(object):
+ """Mixin for use with testtools.TestCase."""
+
+ def cleanup_core_plugin(self):
+ """Ensure that the core plugin is deallocated."""
+ nm = manager.NeutronManager
+ if not nm.has_instance():
+ return
+
+ # TODO(marun) Fix plugins that do not properly initialize notifiers
+ agentschedulers_db.AgentSchedulerDbMixin.agent_notifiers = {}
+
+ # Perform a check for deallocation only if explicitly
+ # configured to do so since calling gc.collect() after every
+ # test increases test suite execution time by ~50%.
+ check_plugin_deallocation = (
+ os.environ.get('OS_CHECK_PLUGIN_DEALLOCATION') in base.TRUE_STRING)
+ if check_plugin_deallocation:
+ plugin = weakref.ref(nm._instance.plugin)
+
+ nm.clear_instance()
+
+ if check_plugin_deallocation:
+ gc.collect()
+
+ # TODO(marun) Ensure that mocks are deallocated?
+ if plugin() and not isinstance(plugin(), mock.Base):
+ self.fail('The plugin for this test was not deallocated.')
+
+ def setup_coreplugin(self, core_plugin=None):
+ # Plugin cleanup should be triggered last so that
+ # test-specific cleanup has a chance to release references.
+ self.addCleanup(self.cleanup_core_plugin)
+ if core_plugin is not None:
+ cfg.CONF.set_override('core_plugin', core_plugin)
+
+
+class NotificationSetupHelper(object):
+ """Mixin for use with testtools.TestCase."""
+
+ def setup_notification_driver(self, notification_driver=None):
+ self.addCleanup(fake_notifier.reset)
+ if notification_driver is None:
+ notification_driver = [fake_notifier.__name__]
+ cfg.CONF.set_override("notification_driver", notification_driver)
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_extensions
+from neutron.tests.unit import testlib_plugin
from neutron.tests.unit import vmware
from neutron.tests.unit.vmware import test_nsx_plugin
return []
-class NetworkGatewayExtensionTestCase(base.BaseTestCase):
+class NetworkGatewayExtensionTestCase(base.BaseTestCase,
+ testlib_plugin.PluginSetupHelper):
def setUp(self):
super(NetworkGatewayExtensionTestCase, self).setUp()
Code Review / openstack-build / neutron-build.git / commitdiff