import testtools
from neutron import manager
+from neutron.openstack.common.notifier import api as notifier_api
+from neutron.openstack.common.notifier import test_notifier
from neutron.tests import post_mortem_debug
if core_plugin is not None:
cfg.CONF.set_override('core_plugin', core_plugin)
+ def _cleanup_test_notifier(self):
+ test_notifier.NOTIFICATIONS = []
+
+ def setup_notification_driver(self, notification_driver=None):
+ # to reload the drivers
+ self.addCleanup(notifier_api._reset_drivers)
+ self.addCleanup(self._cleanup_test_notifier)
+ notifier_api._reset_drivers()
+ if notification_driver is None:
+ notification_driver = [test_notifier.__name__]
+ cfg.CONF.set_override("notification_driver", notification_driver)
+
def setUp(self):
super(BaseTestCase, self).setUp()
from oslo.config import cfg
from neutron.agent.common import config
-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.services.metering.agents import metering_agent
cfg.CONF.set_override('measure_interval', 0)
cfg.CONF.set_override('report_interval', 0)
- notifier_api._drivers = None
- cfg.CONF.set_override("notification_driver", [test_notifier.__name__])
+ self.setup_notification_driver()
metering_rpc = ('neutron.services.metering.agents.metering_agent.'
'MeteringPluginRpc._get_sync_data_metering')
self.agent = metering_agent.MeteringAgent('my agent', cfg.CONF)
self.driver = self.agent.metering_driver
- def tearDown(self):
- test_notifier.NOTIFICATIONS = []
- super(TestMeteringOperations, self).tearDown()
-
def test_add_metering_label(self):
self.agent.add_metering_label(None, ROUTERS)
self.assertEqual(self.driver.add_metering_label.call_count, 1)
from neutron.extensions import extraroute
from neutron.extensions import l3
from neutron.openstack.common import log as logging
-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.tests.unit import test_api_v2
from neutron.tests.unit import test_l3_plugin as test_l3
ext_mgr = ExtraRouteTestExtensionManager()
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__])
+ self.setup_notification_driver()
class ExtraRouteDBIntTestCaseXML(ExtraRouteDBIntTestCase):
plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
- # Set to None to reload the drivers
- notifier_api._drivers = None
- cfg.CONF.set_override("notification_driver", [test_notifier.__name__])
+ self.setup_notification_driver()
class ExtraRouteDBSepTestCaseXML(ExtraRouteDBSepTestCase):
from neutron.extensions import l3
from neutron.manager import NeutronManager
from neutron.openstack.common import log as logging
-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.common import constants as service_constants
super(L3BaseForIntTests, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
- # Set to None to reload the drivers
- notifier_api._drivers = None
- cfg.CONF.set_override("notification_driver", [test_notifier.__name__])
-
- def tearDown(self):
- test_notifier.NOTIFICATIONS = []
- super(L3BaseForIntTests, self).tearDown()
+ self.setup_notification_driver()
class L3BaseForSepTests(test_db_plugin.NeutronDbPluginV2TestCase):
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
- cfg.CONF.set_override("notification_driver", [test_notifier.__name__])
-
- def tearDown(self):
- test_notifier.NOTIFICATIONS = []
- super(L3BaseForSepTests, self).tearDown()
+ self.setup_notification_driver()
class L3AgentDbIntTestCase(L3BaseForIntTests, L3AgentDbTestCaseBase):