raise test.TestingException()
-class WrapExceptionTestCase(test.TestCase):
- def test_wrap_exception_good_return(self):
- wrapped = utils.wrap_exception()
- self.assertEquals(99, wrapped(good_function)())
-
- def test_wrap_exception_throws_error(self):
- wrapped = utils.wrap_exception()
- self.assertRaises(exception.Error, wrapped(bad_function_error))
-
- def test_wrap_exception_throws_exception(self):
- wrapped = utils.wrap_exception()
- self.assertRaises(test.TestingException,
- wrapped(bad_function_exception))
-
- def test_wrap_exception_with_notifier(self):
- notifier = FakeNotifier()
- wrapped = utils.wrap_exception(notifier, "publisher", "event",
- "level")
- self.assertRaises(test.TestingException,
- wrapped(bad_function_exception))
- self.assertEquals(notifier.provided_publisher, "publisher")
- self.assertEquals(notifier.provided_event, "event")
- self.assertEquals(notifier.provided_priority, "level")
- for key in ['exception', 'args']:
- self.assertTrue(key in notifier.provided_payload.keys())
-
- def test_wrap_exception_with_notifier_defaults(self):
- notifier = FakeNotifier()
- wrapped = utils.wrap_exception(notifier)
- self.assertRaises(test.TestingException,
- wrapped(bad_function_exception))
- self.assertEquals(notifier.provided_publisher, None)
- self.assertEquals(notifier.provided_event, "bad_function_exception")
- self.assertEquals(notifier.provided_priority, notifier.ERROR)
-
-
class CinderExceptionTestCase(test.TestCase):
def test_default_error_msg(self):
class FakeCinderException(exception.CinderException):
LOG.exception(msg, **kwargs)
self._rollback()
-
-
-def wrap_exception(notifier=None, publisher_id=None, event_type=None,
- level=None):
- """This decorator wraps a method to catch any exceptions that may
- get thrown. It logs the exception as well as optionally sending
- it to the notification system.
- """
- # TODO(sandy): Find a way to import cinder.notifier.api so we don't have
- # to pass it in as a parameter. Otherwise we get a cyclic import of
- # cinder.notifier.api -> cinder.utils -> cinder.exception :(
- # TODO(johannes): Also, it would be nice to use
- # utils.save_and_reraise_exception() without an import loop
- def inner(f):
- def wrapped(*args, **kw):
- try:
- return f(*args, **kw)
- except Exception, e:
- # Save exception since it can be clobbered during processing
- # below before we can re-raise
- exc_info = sys.exc_info()
-
- if notifier:
- payload = dict(args=args, exception=e)
- payload.update(kw)
-
- # Use a temp vars so we don't shadow
- # our outer definitions.
- temp_level = level
- if not temp_level:
- temp_level = notifier.ERROR
-
- temp_type = event_type
- if not temp_type:
- # If f has multiple decorators, they must use
- # functools.wraps to ensure the name is
- # propagated.
- temp_type = f.__name__
-
- context = get_context_from_function_and_args(f,
- args,
- kw)
-
- notifier.notify(context, publisher_id, temp_type,
- temp_level, payload)
-
- # re-raise original exception since it may have been clobbered
- raise exc_info[0], exc_info[1], exc_info[2]
-
- return functools.wraps(f)(wrapped)
- return inner
-
-
-def get_context_from_function_and_args(function, args, kwargs):
- """Find an arg of type RequestContext and return it.
-
- This is useful in a couple of decorators where we don't
- know much about the function we're wrapping.
- """
-
- # import here to avoid circularity:
- from cinder import context
-
- for arg in itertools.chain(kwargs.values(), args):
- if isinstance(arg, context.RequestContext):
- return arg
-
- return None