Added tests for event system and refactored

cloudbridge/cloud/base/events.py

@@ -1,169 +1,137 @@
+import bisect
+import collections
 import fnmatch
+import logging
 import re
-from copy import deepcopy
-from enum import Enum
 
 from ..interfaces.events import EventDispatcher
 from ..interfaces.exceptions import HandlerException
 
+log = logging.getLogger(__name__)
 
-class HandlerType(Enum):
-    """
-    Handler Types.
-    """
-    SUBSCRIPTION = 'subscription'
-    INTERCEPTION = 'intercept'
 
+class InterceptingEventHandler(object):
 
-class SimpleEventDispatcher(EventDispatcher):
+    def __init__(self, event_name, priority, callback):
+        self.dispatcher = None
+        self.event_name = event_name
+        self.priority = priority
+        self.callback = callback
 
-    def __init__(self):
-        self.__events = {}
-        self.__initialized = {}
+    def __lt__(self, other):
+        # This is required for the bisect module to insert
+        # event handlers sorted by priority
+        return self.priority < other.priority
+
+    def get_next_handler(self, full_event_name):
+        handler_list = self.dispatcher.handler_cache.get(full_event_name, [])
+        # find position of this handler
+        pos = bisect.bisect_left(handler_list, self)
+        assert handler_list[pos] == self
+        if pos < len(handler_list)-1:
+            return handler_list[pos+1]
+        else:
+            return None
 
-    def get_handlers(self, event_name):
-        return self.__events.get(event_name)
+    def invoke(self, **kwargs):
+        kwargs.pop('next_handler', None)
+        next_handler = self.get_next_handler(kwargs.get('event_name', None))
+        # callback is responsible for invoking the next_handler and
+        # controlling the result value
+        return self.callback(next_handler=next_handler, **kwargs)
 
-    def check_initialized(self, event_name):
-        return self.__initialized.get(event_name) or False
 
-    def mark_initialized(self, event_name):
-        self.__initialized[event_name] = True
+class ObservingEventHandler(InterceptingEventHandler):
 
-    def observe(self, event_name, priority, callback):
-        handler = EventHandler(HandlerType.SUBSCRIPTION, callback, priority)
-        if not self.__events.get(event_name):
-            self.__events[event_name] = list()
-        self.__events[event_name].append((priority, handler))
+    def __init__(self, event_name, priority, callback):
+        super(ObservingEventHandler, self).__init__(event_name, priority,
+                                                    callback)
 
-    def intercept(self, event_name, priority, callback):
-        handler = EventHandler(HandlerType.INTERCEPTION, callback, priority)
-        if not self.__events.get(event_name):
-            self.__events[event_name] = list()
-        self.__events[event_name].append((priority, handler))
-
-    def call(self, event_name, priority, callback, **kwargs):
-        handler = EventHandler(HandlerType.SUBSCRIPTION, callback, priority)
-        if not self.__events.get(event_name):
-            self.__events[event_name] = list()
-        # Although handler object has priority property, keep it a pair to not
-        # access each handler when sorting
-        self.__events[event_name].append((priority, handler))
-        try:
-            ret_obj = self._emit(event_name, **kwargs)
-        finally:
-            self.__events[event_name].remove((priority, handler))
-        return ret_obj
-
-    def _emit(self, event_name, **kwargs):
-
-        def _match_and_sort(event_name):
-            new_list = []
-            for key in self.__events.keys():
-                if re.search(fnmatch.translate(key), event_name):
-                    new_list.extend(deepcopy(self.__events[key]))
-            new_list.sort(key=lambda x: x[0])
-            # Make sure all priorities are unique
-            priority_list = [x[0] for x in new_list]
-            if len(set(priority_list)) != len(priority_list):
-
-                guilty_prio = None
-                for prio in priority_list:
-                    if prio == guilty_prio:
-                        break
-                    guilty_prio = prio
-
-                # guilty_prio should never be none since we checked for
-                # duplicates before iterating
-                guilty_names = [x[1].callback.__name__
-                                for x in new_list
-                                if x[0] == guilty_prio]
-
-                message = "Event '{}' has multiple subscribed handlers " \
-                          "at priority '{}', with function names [{}]. " \
-                          "Each priority must only have a single " \
-                          "corresponding handler." \
-                    .format(event_name, priority, ", ".join(guilty_names))
-                raise HandlerException(message)
-
-            return new_list
-
-        if not self.__events.get(event_name):
-            message = "Event '{}' has no subscribed handlers.".\
-                format(event_name)
-            raise HandlerException(message)
+    def invoke(self, **kwargs):
+        # Notify listener. Ignore result from observable handler
+        kwargs.pop('next_handler', None)
+        self.callback(**kwargs)
+        # Kick off the handler chain
+        next_handler = self.get_next_handler(kwargs.get('event_name', None))
+        if next_handler:
+            return next_handler.invoke(**kwargs)
+        else:
+            return None
 
-        prev_handler = None
-        first_handler = None
-        for (priority, handler) in _match_and_sort(event_name):
-            if not first_handler:
-                first_handler = handler
-            if prev_handler:
-                prev_handler.next_handler = handler
-            prev_handler = handler
-        return first_handler.invoke(**kwargs)
 
+class SimpleEventDispatcher(EventDispatcher):
 
-class EventHandler(object):
-    def __init__(self, handler_type, callback, priority):
-        self.handler_type = handler_type
-        self.callback = callback
-        self._next_handler = None
-        self.priority = priority
+    def __init__(self):
+        # The dict key is event_name.
+        # The dict value is a list of handlers for the event, sorted by event
+        # priority
+        self.__events = collections.OrderedDict({})
+        self.__handler_cache = {}
 
     @property
-    def next_handler(self):
-        return self._next_handler
+    def handler_cache(self):
+        return self.__handler_cache
+
+    def _create_handler_cache(self, event_name):
+        cache_list = []
+        # sort from most specific to least specific
+        for key in self.__events.keys():
+            if re.search(fnmatch.translate(key), event_name):
+                cache_list.extend(self.__events[key])
+        cache_list.sort(key=lambda h: h.priority)
+
+        # Make sure all priorities are unique
+        priority_list = [h.priority for h in cache_list]
+        if len(set(priority_list)) != len(priority_list):
+            guilty_prio = None
+            for prio in priority_list:
+                if prio == guilty_prio:
+                    break
+                guilty_prio = prio
+
+            # guilty_prio should never be none since we checked for
+            # duplicates before iterating
+            guilty_names = [h.callback.__name__ for h in cache_list
+                            if h.priority == guilty_prio]
+
+            message = "Event '{}' has multiple subscribed handlers " \
+                      "at priority '{}', with function names [{}]. " \
+                      "Each priority must only have a single " \
+                      "corresponding handler." \
+                .format(event_name, guilty_prio, ", ".join(guilty_names))
+            raise HandlerException(message)
+        return cache_list
 
-    @next_handler.setter
-    def next_handler(self, new_handler):
-        self._next_handler = new_handler
+    def _subscribe(self, event_handler):
+        """
+        subscribe an event handler to this dispatcher
+        """
+        event_handler.dispatcher = self
+        handler_list = self.__events.get(event_handler.event_name, [])
+        handler_list.append(event_handler)
+        self.__events[event_handler.event_name] = handler_list
 
-    def invoke(self, **kwargs):
-        if self.handler_type == HandlerType.SUBSCRIPTION:
-            result = self.callback(**kwargs)
-
-            next = self.next_handler
-            if next:
-                if next.handler_type == HandlerType.SUBSCRIPTION:
-                    if result or not kwargs.get('callback_result', None):
-                        kwargs['callback_result'] = result
-                    new_result = next.invoke(**kwargs)
-                elif next.handler_type == HandlerType.INTERCEPTION:
-                    new_result = next.invoke(**kwargs)
-
-                if new_result:
-                    result = new_result
-
-            self.next_handler = None
-
-        elif self.handler_type == HandlerType.INTERCEPTION:
-            kwargs.pop('next_handler', None)
-            result = self.callback(next_handler=self.next_handler, **kwargs)
-            self.next_handler = None
-
-        return result
-
-    def skip(self, **kwargs):
-        if self.next_handler:
-            self.next_handler.invoke(**kwargs)
-            self.next_handler = None
-
-    def skip_to_name(self, function_name, **kwargs):
-        if self.callback.__name__ == function_name:
-            self.invoke(**kwargs)
-        elif self.next_handler:
-            self.next_handler.skip_to_name(function_name, **kwargs)
-            self.next_handler = None
-
-    def skip_to_priority(self, priority, **kwargs):
-        if self.priority == priority:
-            self.invoke(**kwargs)
-        elif self.next_handler:
-            self.next_handler.skip_to_priority(priority, **kwargs)
-            self.next_handler = None
-
-    def skip_rest(self):
-        if self.next_handler:
-            self.next_handler.skip_rest()
-            self.next_handler = None
+    def observe(self, event_name, priority, callback):
+        handler = ObservingEventHandler(event_name, priority, callback)
+        self._subscribe(handler)
+
+    def intercept(self, event_name, priority, callback):
+        handler = InterceptingEventHandler(event_name, priority, callback)
+        self._subscribe(handler)
+
+    def emit(self, sender, event_name, **kwargs):
+        handlers = self.handler_cache.get(event_name)
+        if handlers is None:
+            self.__handler_cache[event_name] = self._create_handler_cache(
+                event_name)
+            handlers = self.handler_cache.get(event_name)
+
+        if handlers:
+            # only kick off first handler in chain
+            return handlers[0].invoke(sender=sender, event_name=event_name,
+                                      **kwargs)
+        else:
+            message = "Event '{}' has no subscribed handlers.".\
+                format(event_name)
+            log.warning(message)
+            return None

cloudbridge/cloud/interfaces/events.py

@@ -55,21 +55,14 @@ class EventDispatcher(object):
         pass
 
     @abstractmethod
-    def call(self, event_name, priority, callback, **kwargs):
+    def emit(self, sender, event_name, **kwargs):
         """
         Raises an event while registering a given callback
 
         :type event_name: str
-        :param event_name: The name of the event to which you are subscribing
-            the callback function. Accepts wildcard parameters.
-
-        :type priority: int
-        :param priority: The priority that this handler should be given.
-            When the event is emitted, all handlers will be run in order of
-            priority.
+        :param event_name: The name of the event which is being raised.
 
-        :type callback: function
-        :param callback: The callback function that should be called with
-            the parameters given at when the even is emitted.
+        :type sender: object
+        :param sender: The object which is raising the event
         """
         pass

test/test_event_system.py

@@ -0,0 +1,275 @@
+import unittest
+
+from cloudbridge.cloud.base.events import SimpleEventDispatcher
+from cloudbridge.cloud.interfaces.exceptions import HandlerException
+
+
+class EventSystemTestCase(unittest.TestCase):
+
+    def test_emit_event_no_handlers(self):
+        dispatcher = SimpleEventDispatcher()
+        result = dispatcher.emit(self, "event.hello.world")
+        self.assertIsNone(result, "Result should be none as there are no"
+                          "registered handlers")
+
+    def test_emit_event_observing_handler(self):
+        EVENT_NAME = "event.hello.world"
+        callback_tracker = ['']
+
+        def my_callback(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': EVENT_NAME})
+            callback_tracker[0] += 'obs'
+            return "hello"
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.observe(EVENT_NAME, 1000, my_callback)
+        result = dispatcher.emit(self, EVENT_NAME)
+        self.assertEqual(
+            callback_tracker[0], "obs", "callback should have been invoked"
+            "once and contain value `obs` but tracker value is {0}".format(
+                callback_tracker[0]))
+        self.assertIsNone(result, "Result should be none as this is an"
+                          " observing handler")
+
+    def test_emit_event_intercepting_handler(self):
+        EVENT_NAME = "event.hello.world"
+        callback_tracker = ['']
+
+        def my_callback(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': EVENT_NAME,
+                                  'next_handler': None})
+            callback_tracker[0] += "intcpt"
+            return "world"
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.intercept(EVENT_NAME, 1000, my_callback)
+        result = dispatcher.emit(self, EVENT_NAME)
+        self.assertEqual(
+            callback_tracker[0], "intcpt", "callback should have been invoked"
+            "once and contain value `intcpt` but tracker value is {0}".format(
+                callback_tracker[0]))
+        self.assertEqual(result, "world", "Result should be `world` as this"
+                         " is an intercepting handler")
+
+    def test_emit_event_observe_precedes_intercept(self):
+        EVENT_NAME = "event.hello.world"
+        callback_tracker = ['']
+
+        def my_callback_obs(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': EVENT_NAME})
+            callback_tracker[0] += "obs_"
+            return "hello"
+
+        def my_callback_intcpt(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': EVENT_NAME,
+                                  'next_handler': None})
+            callback_tracker[0] += "intcpt_"
+            return "world"
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.observe(EVENT_NAME, 1000, my_callback_obs)
+        dispatcher.intercept(EVENT_NAME, 1001, my_callback_intcpt)
+        result = dispatcher.emit(self, EVENT_NAME)
+        self.assertEqual(
+            callback_tracker[0], "obs_intcpt_", "callback was not invoked in "
+            "expected order. Should have been obs_intcpt_ but is {0}".format(
+                callback_tracker[0]))
+        self.assertEqual(result, "world", "Result should be `world` as this"
+                         " is the return value of the intercepting handler")
+
+    def test_emit_event_observe_follows_intercept(self):
+        EVENT_NAME = "event.hello.world"
+        callback_tracker = ['']
+
+        def my_callback_intcpt(**kwargs):
+            self.assertEqual(kwargs.get('sender'), self)
+            self.assertEqual(kwargs.get('next_handler').priority, 1001)
+            self.assertEqual(kwargs.get('next_handler').callback.__name__,
+                             "my_callback_obs")
+            callback_tracker[0] += "intcpt_"
+            # invoke next handler
+            retval = kwargs.get('next_handler').invoke(**kwargs)
+            self.assertIsNone(retval, "Return values of observable handlers"
+                              " should not be propagated.")
+            return "world"
+
+        def my_callback_obs(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': EVENT_NAME})
+            callback_tracker[0] += "obs_"
+            return "hello"
+
+        dispatcher = SimpleEventDispatcher()
+        # register priorities out of order to test that too
+        dispatcher.observe(EVENT_NAME, 1001, my_callback_obs)
+        dispatcher.intercept(EVENT_NAME, 1000, my_callback_intcpt)
+        result = dispatcher.emit(self, EVENT_NAME)
+        self.assertEqual(
+            callback_tracker[0], "intcpt_obs_", "callback was not invoked in "
+            "expected order. Should have been intcpt_obs_ but is {0}".format(
+                callback_tracker[0]))
+        self.assertEqual(result, "world", "Result should be `world` as this"
+                         " is the return value of the intercepting handler")
+
+    def test_emit_event_intercept_follows_intercept(self):
+        EVENT_NAME = "event.hello.world"
+        callback_tracker = ['']
+
+        def my_callback_intcpt1(**kwargs):
+            self.assertEqual(kwargs.get('sender'), self)
+            self.assertEqual(kwargs.get('next_handler').priority, 2020)
+            self.assertEqual(kwargs.get('next_handler').callback.__name__,
+                             "my_callback_intcpt2")
+            callback_tracker[0] += "intcpt1_"
+            # invoke next handler but ignore return value
+            return "hello" + kwargs.get('next_handler').invoke(**kwargs)
+
+        def my_callback_intcpt2(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': EVENT_NAME,
+                                  'next_handler': None})
+            callback_tracker[0] += "intcpt2_"
+            return "world"
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.intercept(EVENT_NAME, 2000, my_callback_intcpt1)
+        dispatcher.intercept(EVENT_NAME, 2020, my_callback_intcpt2)
+        result = dispatcher.emit(self, EVENT_NAME)
+        self.assertEqual(
+            callback_tracker[0], "intcpt1_intcpt2_", "callback was not invoked"
+            " in expected order. Should have been intcpt1_intcpt2_ but is"
+            " {0}".format(callback_tracker[0]))
+        self.assertEqual(result, "helloworld", "Result should be `helloworld` "
+                         "as this is the expected return value from the chain")
+
+    def test_subscribe_event_duplicate_priority(self):
+
+        def my_callback(**kwargs):
+            pass
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.intercept("event.hello.world", 1000, my_callback)
+        dispatcher.intercept("event.hello.world", 1000, my_callback)
+        with self.assertRaises(HandlerException):
+            dispatcher.emit(self, "event.hello.world")
+
+    def test_subscribe_event_duplicate_wildcard_priority(self):
+
+        def my_callback(**kwargs):
+            pass
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.intercept("event.hello.world", 1000, my_callback)
+        dispatcher.intercept("event.hello.*", 1000, my_callback)
+        with self.assertRaises(HandlerException):
+            dispatcher.emit(self, "event.hello.world")
+
+    def test_subscribe_event_duplicate_wildcard_priority_allowed(self):
+        # duplicate priorities for different wildcard namespaces allowed
+        def my_callback(**kwargs):
+            pass
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.intercept("event.hello.world", 1000, my_callback)
+        dispatcher.intercept("someevent.hello.*", 1000, my_callback)
+        # emit should work fine in this case with no exceptions
+        dispatcher.emit(self, "event.hello.world")
+
+    def test_subscribe_multiple_events(self):
+        EVENT_NAME = "event.hello.world"
+        callback_tracker = ['']
+
+        def my_callback1(**kwargs):
+            self.assertDictEqual(kwargs, {'sender': self,
+                                          'event_name': EVENT_NAME})
+            callback_tracker[0] += "event1_"
+            return "hello"
+
+        def my_callback2(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': "event.hello.anotherworld"})
+            callback_tracker[0] += "event2_"
+            return "another"
+
+        def my_callback3(**kwargs):
+            self.assertDictEqual(kwargs,
+                                 {'sender': self,
+                                  'event_name': "event.hello.anotherworld",
+                                  'next_handler': None})
+            callback_tracker[0] += "event3_"
+            return "world"
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.observe(EVENT_NAME, 2000, my_callback1)
+        # register to a different event with the same priority
+        dispatcher.observe("event.hello.anotherworld", 2000, my_callback2)
+        dispatcher.intercept("event.hello.anotherworld", 2020, my_callback3)
+        result = dispatcher.emit(self, EVENT_NAME)
+        self.assertEqual(
+            callback_tracker[0], "event1_", "only `event.hello.world` handlers"
+            " should have been  triggered but received {0}".format(
+                callback_tracker[0]))
+        self.assertEqual(result, None, "Result should be `helloworld` "
+                         "as this is the expected return value from the chain")
+
+        result = dispatcher.emit(self, "event.hello.anotherworld")
+        self.assertEqual(
+            callback_tracker[0], "event1_event2_event3_", "only handlers for"
+            "  event `event.hello.anotherworld` should have been  triggered"
+            " but received {0}".format(callback_tracker[0]))
+        self.assertEqual(result, "world", "Result should be `world` "
+                         "as this is the expected return value from the chain")
+
+    def test_subscribe_wildcard(self):
+        callback_tracker = ['']
+
+        def my_callback1(**kwargs):
+            callback_tracker[0] += "event1_"
+            return "hello" + kwargs.get('next_handler').invoke(**kwargs)
+
+        def my_callback2(**kwargs):
+            callback_tracker[0] += "event2_"
+            return "some" + kwargs.get('next_handler').invoke(**kwargs)
+
+        def my_callback3(**kwargs):
+            callback_tracker[0] += "event3_"
+            return "other" + kwargs.get('next_handler').invoke(**kwargs)
+
+        def my_callback4(**kwargs):
+            callback_tracker[0] += "event4_"
+            return "world"
+
+        dispatcher = SimpleEventDispatcher()
+        dispatcher.intercept("event.*", 2000, my_callback1)
+        # register to a different event with the same priority
+        dispatcher.intercept("event.hello.*", 2010, my_callback2)
+        dispatcher.intercept("event.hello.there", 2030, my_callback4)
+        dispatcher.intercept("event.*.there", 2020, my_callback3)
+        dispatcher.intercept("event.*.world", 2020, my_callback4)
+        dispatcher.intercept("someevent.hello.there", 2030, my_callback3)
+        # emit a series of events
+        result = dispatcher.emit(self, "event.hello.there")
+
+        self.assertEqual(
+            callback_tracker[0], "event1_event2_event3_event4_",
+            "Event handlers executed in unexpected order {0}".format(
+                callback_tracker[0]))
+        self.assertEqual(result, "hellosomeotherworld")
+
+        result = dispatcher.emit(self, "event.test.hello.world")
+        self.assertEqual(
+            callback_tracker[0], "event1_event2_event3_event4_event1_event4_",
+            "Event handlers executed in unexpected order {0}".format(
+                callback_tracker[0]))
+        self.assertEqual(result, "helloworld")