openstack/oslo.messaging
Code Issues Proposed changes

Merge "Allow checking if notifier is enabled"

Browse Source
This commit is contained in:
Jenkins 2017-03-06 18:30:02 +00:00 committed by Gerrit Code Review
commit 4f7198597e
2 changed files with 27 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
oslo_messaging/notify/notifier.py
View File

@ -415,6 +415,14 @@ class Notifier(object):
""" """
self._notify(ctxt, event_type, payload, 'SAMPLE') self._notify(ctxt, event_type, payload, 'SAMPLE')
def is_enabled(self):
"""Check if the notifier will emit notifications anywhere.
:return: false if the driver of the notifier is set only to noop, true
otherwise
"""
return self._driver_mgr.names() != ['noop']
class _SubNotifier(Notifier): class _SubNotifier(Notifier):

19
oslo_messaging/tests/notify/test_notifier.py
View File

@ -226,6 +226,8 @@ class TestMessagingNotifier(test_utils.BaseTestCase):
uuid.uuid4.assert_called_once_with() uuid.uuid4.assert_called_once_with()
transport._send_notification.assert_has_calls(calls, any_order=True) transport._send_notification.assert_has_calls(calls, any_order=True)
self.assertTrue(notifier.is_enabled())
TestMessagingNotifier.generate_scenarios() TestMessagingNotifier.generate_scenarios()
@ -337,6 +339,8 @@ class TestLogNotifier(test_utils.BaseTestCase):
'notification.test.notify') 'notification.test.notify')
logger.info.assert_called_once_with(JsonMessageMatcher(message)) logger.info.assert_called_once_with(JsonMessageMatcher(message))
self.assertTrue(notifier.is_enabled())
def test_sample_priority(self): def test_sample_priority(self):
# Ensure logger drops sample-level notifications. # Ensure logger drops sample-level notifications.
driver = _impl_log.LogDriver(None, None, None) driver = _impl_log.LogDriver(None, None, None)
@ -379,6 +383,8 @@ class TestRoutingNotifier(test_utils.BaseTestCase):
self.notifier = oslo_messaging.Notifier(transport) self.notifier = oslo_messaging.Notifier(transport)
self.router = self.notifier._driver_mgr['routing'].obj self.router = self.notifier._driver_mgr['routing'].obj
self.assertTrue(self.notifier.is_enabled())
def _fake_extension_manager(self, ext): def _fake_extension_manager(self, ext):
return extension.ExtensionManager.make_test_instance( return extension.ExtensionManager.make_test_instance(
[extension.Extension('test', None, None, ext), ]) [extension.Extension('test', None, None, ext), ])
@ -593,3 +599,16 @@ group_1:
{}, mock.ANY, 'INFO', None) {}, mock.ANY, 'INFO', None)
rpc2_driver.notify.assert_called_once_with( rpc2_driver.notify.assert_called_once_with(
{}, mock.ANY, 'INFO', None) {}, mock.ANY, 'INFO', None)
class TestNoOpNotifier(test_utils.BaseTestCase):
def test_notifier(self):
self.config(driver=['noop'],
group='oslo_messaging_notifications')
transport = _FakeTransport(self.conf)
notifier = oslo_messaging.Notifier(transport, 'test.localhost')
self.assertFalse(notifier.is_enabled())