x/python-ganttclient
Code Issues Proposed changes

Added ability to detect import errors in list_notifier if one or more drivers could not be loaded.

Browse Source
This commit is contained in:
Tim Simpson
2011-08-22 14:06:59 -05:00
parent 828d0499ec
commit 4ba34dc826
3 changed files with 46 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Authors
View File

@@ -99,6 +99,7 @@ Scott Moser <smoser@ubuntu.com>
Soren Hansen <soren.hansen@rackspace.com>
Stephanie Reese <reese.sm@gmail.com>
Thierry Carrez <thierry@openstack.org>
Tim Simpson <tim.simpson@rackspace.com>
Todd Willey <todd@ansolabs.com>
Trey Morris <trey.morris@rackspace.com>
Tushar Patil <tushar.vitthal.patil@gmail.com>

23
nova/notifier/list_notifier.py
View File

@@ -16,6 +16,7 @@
from nova import flags
from nova import log as logging
from nova import utils
from nova.exception import ClassNotFound
flags.DEFINE_multistring('list_notifier_drivers',
['nova.notifier.no_op_notifier'],
@@ -27,17 +28,31 @@ LOG = logging.getLogger('nova.notifier.list_notifier')
drivers = None
def __get_drivers():
class ImportFailureNotifier(object):
def __init__(self, exception):
self.exception = exception
def notify(message):
raise self.exception
def _get_drivers():
"""Instantiates and returns drivers based on the flag values."""
global drivers
if not drivers:
drivers = [utils.import_object(notification_driver)
for notification_driver in FLAGS.list_notifier_drivers]
drivers = []
for notification_driver in FLAGS.list_notifier_drivers:
try:
drivers.append(utils.import_object(notification_driver))
except ClassNotFound as e:
drivers.append(ImportFailureNotifier(e))
return drivers
def notify(message):
"""Passes notification to mulitple notifiers in a list."""
for driver in __get_drivers():
for driver in _get_drivers():
try:
driver.notify(message)
except Exception as e:

52
nova/tests/notifier/test_list_notifier.py
View File

@@ -31,10 +31,26 @@ from nova import test
class NotifierListTestCase(test.TestCase):
"""Test case for notifications"""
def setUp(self):
super(NotifierListTestCase, self).setUp()
list_notifier._reset_drivers()
self.stubs = stubout.StubOutForTesting()
# Mock log to add one to exception_count when log.exception is called
def mock_exception(cls, *args):
self.exception_count += 1
self.exception_count = 0
list_notifier_log = logging.getLogger('nova.notifier.list_notifier')
self.stubs.Set(list_notifier_log, "exception", mock_exception)
# Mock no_op notifier to add one to notify_count when called.
def mock_notify(cls, *args):
self.notify_count += 1
self.notify_count = 0
self.stubs.Set(nova.notifier.no_op_notifier, 'notify', mock_notify)
# Mock log_notifier to raise RuntimeError when called.
def mock_notify2(cls, *args):
raise RuntimeError("Bad notifier.")
self.stubs.Set(nova.notifier.log_notifier, 'notify', mock_notify2)
def tearDown(self):
self.stubs.UnsetAll()
@@ -45,41 +61,25 @@ class NotifierListTestCase(test.TestCase):
self.flags(notification_driver='nova.notifier.list_notifier',
list_notifier_drivers=['nova.notifier.no_op_notifier',
'nova.notifier.no_op_notifier'])
self.notify_count = 0
def mock_notify(cls, *args):
self.notify_count += 1
self.stubs.Set(nova.notifier.no_op_notifier, 'notify',
mock_notify)
notify('publisher_id', 'event_type',
nova.notifier.api.WARN, dict(a=3))
self.assertEqual(self.notify_count, 2)
self.assertEqual(self.exception_count, 0)
def test_send_notifications_with_errors(self):
self.exception_count = 0
def mock_exception(cls, *args):
self.exception_count += 1
self.notify_count = 0
def mock_notify(cls, *args):
self.notify_count += 1
def mock_notify2(cls, *args):
raise RuntimeError("Bad notifier.")
self.flags(notification_driver='nova.notifier.list_notifier',
list_notifier_drivers=['nova.notifier.no_op_notifier',
'nova.notifier.log_notifier'])
list_notifier_log = logging.getLogger('nova.notifier.list_notifier')
list_notifier_log.exception
self.stubs.Set(list_notifier_log, "exception", mock_exception)
self.stubs.Set(nova.notifier.no_op_notifier, 'notify', mock_notify)
self.stubs.Set(nova.notifier.log_notifier, 'notify', mock_notify2)
notify('publisher_id', 'event_type', nova.notifier.api.WARN, dict(a=3))
self.assertEqual(self.notify_count, 1)
self.assertEqual(self.exception_count, 1)
def test_when_driver_fails_to_import(self):
self.flags(notification_driver='nova.notifier.list_notifier',
list_notifier_drivers=['nova.notifier.no_op_notifier',
'nova.notifier.logo_notifier',
'fdsjgsdfhjkhgsfkj'])
notify('publisher_id', 'event_type', nova.notifier.api.WARN, dict(a=3))
self.assertEqual(self.exception_count, 2)
self.assertEqual(self.notify_count, 1)