import unittest from cloudbridge.cloud.base.events import SimpleEventDispatcher from cloudbridge.cloud.base.middleware import BaseMiddleware from cloudbridge.cloud.base.middleware import EventDebugLoggingMiddleware from cloudbridge.cloud.base.middleware import ExceptionWrappingMiddleware from cloudbridge.cloud.base.middleware import SimpleMiddlewareManager from cloudbridge.cloud.base.middleware import dispatch from cloudbridge.cloud.base.middleware import implement from cloudbridge.cloud.base.middleware import intercept from cloudbridge.cloud.base.middleware import observe from cloudbridge.cloud.interfaces.exceptions import CloudBridgeBaseException from cloudbridge.cloud.interfaces.exceptions import HandlerException from cloudbridge.cloud.interfaces.exceptions import \ InvalidConfigurationException from cloudbridge.cloud.interfaces.middleware import Middleware from .helpers import skipIfPython class MiddlewareSystemTestCase(unittest.TestCase): def test_basic_middleware(self): class DummyMiddleWare(Middleware): def __init__(self): self.invocation_order = "" def install(self, event_manager): self.event_manager = event_manager self.invocation_order += "install_" def uninstall(self): self.invocation_order += "uninstall" dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) middleware = DummyMiddleWare() manager.add(middleware) self.assertEqual(middleware.invocation_order, "install_", "install should be called when adding new middleware") manager.remove(middleware) self.assertEqual(middleware.invocation_order, "install_uninstall", "uninstall should be called when removing middleware") def test_base_middleware(self): EVENT_NAME = "some.event.occurred" class DummyMiddleWare(BaseMiddleware): def __init__(self): self.invocation_order = "" @intercept(event_pattern="some.event.*", priority=900) def my_callback_intcpt(self, event_args, *args, **kwargs): self.invocation_order += "intcpt_" assert 'first_pos_arg' in args assert kwargs.get('a_keyword_arg') == "something" next_handler = event_args.get('next_handler') return next_handler.invoke(event_args, *args, **kwargs) @implement(event_pattern="some.event.*", priority=950) def my_callback_impl(self, *args, **kwargs): self.invocation_order += "impl_" assert 'first_pos_arg' in args assert kwargs.get('a_keyword_arg') == "something" return "hello" @observe(event_pattern="some.event.*", priority=1000) def my_callback_obs(self, event_args, *args, **kwargs): self.invocation_order += "obs" assert 'first_pos_arg' in args assert event_args['result'] == "hello" assert kwargs.get('a_keyword_arg') == "something" dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) middleware = DummyMiddleWare() manager.add(middleware) dispatcher.dispatch(self, EVENT_NAME, 'first_pos_arg', a_keyword_arg='something') self.assertEqual(middleware.invocation_order, "intcpt_impl_obs") self.assertListEqual( [middleware.my_callback_intcpt, middleware.my_callback_impl, middleware.my_callback_obs], [handler.callback for handler in dispatcher.get_handlers_for_event(EVENT_NAME)]) manager.remove(middleware) self.assertListEqual([], dispatcher.get_handlers_for_event(EVENT_NAME)) def test_multiple_middleware(self): EVENT_NAME = "some.really.interesting.event.occurred" class DummyMiddleWare1(BaseMiddleware): @observe(event_pattern="some.really.*", priority=1000) def my_obs1_3(self, *args, **kwargs): pass @implement(event_pattern="some.*", priority=970) def my_impl1_2(self, *args, **kwargs): return "hello" @intercept(event_pattern="some.*", priority=900) def my_intcpt1_1(self, event_args, *args, **kwargs): next_handler = event_args.get('next_handler') return next_handler.invoke(event_args, *args, **kwargs) class DummyMiddleWare2(BaseMiddleware): @observe(event_pattern="some.really.*", priority=1050) def my_obs2_3(self, *args, **kwargs): pass @intercept(event_pattern="*", priority=950) def my_intcpt2_2(self, event_args, *args, **kwargs): next_handler = event_args.get('next_handler') return next_handler.invoke(event_args, *args, **kwargs) @implement(event_pattern="some.really.*", priority=920) def my_impl2_1(self, *args, **kwargs): pass dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) middleware1 = DummyMiddleWare1() middleware2 = DummyMiddleWare2() manager.add(middleware1) manager.add(middleware2) dispatcher.dispatch(self, EVENT_NAME) # Callbacks in both middleware classes should be registered self.assertListEqual( [middleware1.my_intcpt1_1, middleware2.my_impl2_1, middleware2.my_intcpt2_2, middleware1.my_impl1_2, middleware1.my_obs1_3, middleware2.my_obs2_3], [handler.callback for handler in dispatcher.get_handlers_for_event(EVENT_NAME)]) manager.remove(middleware1) # Only middleware2 callbacks should be registered self.assertListEqual( [middleware2.my_impl2_1, middleware2.my_intcpt2_2, middleware2.my_obs2_3], [handler.callback for handler in dispatcher.get_handlers_for_event(EVENT_NAME)]) # add middleware back to check that internal state is properly handled manager.add(middleware1) # should one again equal original list self.assertListEqual( [middleware1.my_intcpt1_1, middleware2.my_impl2_1, middleware2.my_intcpt2_2, middleware1.my_impl1_2, middleware1.my_obs1_3, middleware2.my_obs2_3], [handler.callback for handler in dispatcher.get_handlers_for_event(EVENT_NAME)]) def test_automatic_middleware(self): EVENT_NAME = "another.interesting.event.occurred" class SomeDummyClass(object): @observe(event_pattern="another.really.*", priority=1000) def not_a_match(self, *args, **kwargs): pass @intercept(event_pattern="another.*", priority=900) def my_callback_intcpt2(self, *args, **kwargs): pass def not_an_event_handler(self, *args, **kwargs): pass @observe(event_pattern="another.interesting.*", priority=1000) def my_callback_obs1(self, *args, **kwargs): pass @implement(event_pattern="another.interesting.*", priority=1050) def my_callback_impl(self, *args, **kwargs): pass dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) some_obj = SomeDummyClass() middleware = manager.add(some_obj) dispatcher.dispatch(self, EVENT_NAME) # Middleware should be discovered even if class containing interceptors # doesn't inherit from Middleware self.assertListEqual( [some_obj.my_callback_intcpt2, some_obj.my_callback_obs1, some_obj.my_callback_impl], [handler.callback for handler in dispatcher.get_handlers_for_event(EVENT_NAME)]) manager.remove(middleware) # Callbacks should be correctly removed self.assertListEqual( [], [handler.callback for handler in dispatcher.get_handlers_for_event(EVENT_NAME)]) def test_event_decorator(self): EVENT_NAME = "some.event.occurred" class SomeDummyClass(object): def __init__(self): self.invocation_order = "" self.events = SimpleEventDispatcher() @intercept(event_pattern="some.event.*", priority=900) def my_callback_intcpt(self, event_args, *args, **kwargs): self.invocation_order += "intcpt_" assert 'first_pos_arg' in args assert kwargs.get('a_keyword_arg') == "something" next_handler = event_args.get('next_handler') return next_handler.invoke(event_args, *args, **kwargs) @observe(event_pattern="some.event.*", priority=3000) def my_callback_obs(self, event_args, *args, **kwargs): self.invocation_order += "obs" assert 'first_pos_arg' in args assert event_args['result'] == "hello" assert kwargs.get('a_keyword_arg') == "something" @dispatch(event=EVENT_NAME, priority=2500) def my_callback_impl(self, *args, **kwargs): self.invocation_order += "impl_" assert 'first_pos_arg' in args assert kwargs.get('a_keyword_arg') == "something" return "hello" obj = SomeDummyClass() manager = SimpleMiddlewareManager(obj.events) middleware = manager.add(obj) # calling my_implementation should trigger all events result = obj.my_callback_impl( 'first_pos_arg', a_keyword_arg='something') self.assertEqual(result, "hello") self.assertEqual(obj.invocation_order, "intcpt_impl_obs") callbacks = [handler.callback for handler in middleware.events.get_handlers_for_event(EVENT_NAME)] self.assertNotIn( obj.my_callback_impl, callbacks, "The event impl callback should not be directly contained" " in callbacks to avoid a circular dispatch") self.assertEqual( len(set(callbacks).difference( set([obj.my_callback_intcpt, obj.my_callback_obs]))), 1, "The event impl callback should be included in the list of" " callbacks indirectly") manager.remove(middleware) # calling my_implementation again should trigger a None response result = obj.my_callback_impl( 'first_pos_arg', a_keyword_arg='something') self.assertEqual(result, None) def test_event_decorator_no_event_property(self): EVENT_NAME = "some.event.occurred" class SomeDummyClass(object): @dispatch(event=EVENT_NAME, priority=2500) def my_callback_impl(self, *args, **kwargs): assert 'first_pos_arg' in args assert kwargs.get('a_keyword_arg') == "something" return "hello" obj = SomeDummyClass() events = SimpleEventDispatcher() manager = SimpleMiddlewareManager(events) manager.add(obj) # calling my_implementation should raise an exception with self.assertRaises(HandlerException): obj.my_callback_impl('first_pos_arg', a_keyword_arg='something') obj.events = events result = obj.my_callback_impl('first_pos_arg', a_keyword_arg='something') self.assertEqual(result, "hello") class ExceptionWrappingMiddlewareTestCase(unittest.TestCase): def test_unknown_exception_is_wrapped(self): EVENT_NAME = "an.exceptional.event" class SomeDummyClass(object): @implement(event_pattern=EVENT_NAME, priority=2500) def raise_a_non_cloudbridge_exception(self, *args, **kwargs): raise Exception("Some unhandled exception") dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) middleware = ExceptionWrappingMiddleware() manager.add(middleware) # no exception should be raised when there's no next handler dispatcher.dispatch(self, EVENT_NAME) some_obj = SomeDummyClass() manager.add(some_obj) with self.assertRaises(CloudBridgeBaseException): dispatcher.dispatch(self, EVENT_NAME) def test_cloudbridge_exception_is_passed_through(self): EVENT_NAME = "an.exceptional.event" class SomeDummyClass(object): @implement(event_pattern=EVENT_NAME, priority=2500) def raise_a_cloudbridge_exception(self, *args, **kwargs): raise InvalidConfigurationException() dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) some_obj = SomeDummyClass() manager.add(some_obj) middleware = ExceptionWrappingMiddleware() manager.add(middleware) with self.assertRaises(InvalidConfigurationException): dispatcher.dispatch(self, EVENT_NAME) class EventDebugLoggingMiddlewareTestCase(unittest.TestCase): # Only python 3 has assertLogs support @skipIfPython("<", 3, 0) def test_messages_logged(self): EVENT_NAME = "an.exceptional.event" class SomeDummyClass(object): @implement(event_pattern=EVENT_NAME, priority=2500) def return_some_value(self, *args, **kwargs): return "hello world" dispatcher = SimpleEventDispatcher() manager = SimpleMiddlewareManager(dispatcher) middleware = EventDebugLoggingMiddleware() manager.add(middleware) some_obj = SomeDummyClass() manager.add(some_obj) with self.assertLogs('cloudbridge.cloud.base.middleware', level='DEBUG') as cm: dispatcher.dispatch(self, EVENT_NAME, "named_param", keyword_param="hello") self.assertTrue( "named_param" in cm.output[0] and "keyword_param" in cm.output[0] and "hello" in cm.output[0], "Log output {0} not as expected".format(cm.output[0])) self.assertTrue( "hello world" in cm.output[1], "Log output {0} does not contain result".format(cm.output[1]))