fixed None for type/sender

IPython/kernel/core/notification.py

@@ -1,85 +1,110 b''
 # encoding: utf-8
 
 """The IPython Core Notification Center.
 
 See docs/blueprints/notification_blueprint.txt for an overview of the
 notification module.
 """
 
 __docformat__ = "restructuredtext en"
                                                                               
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team                           
 #                                                                             
 #  Distributed under the terms of the BSD License.  The full license is in    
 #  the file COPYING, distributed as part of this software.                    
 #-----------------------------------------------------------------------------
 
 
 class NotificationCenter(object):
     """Synchronous notification center"""
     def __init__(self):
         super(NotificationCenter, self).__init__()
+        self._init_observers()
+    
+    
+    def _init_observers(self):
+        """Initialize observer storage"""
+        
         self.registeredTypes = set() #set of types that are observed
         self.registeredSenders = set() #set of senders that are observed
         self.observers = {} #map (type,sender) => callback (callable)
     
     
     def post_notification(self, theType, sender, **kwargs):
         """Post notification (type,sender,**kwargs) to all registered
         observers. 
         
         Implementation
         --------------
         * If no registered observers, performance is O(1).
         * Notificaiton order is undefined.
         * Notifications are posted synchronously.
         """
         
         if(theType==None or sender==None):
             raise Exception("NotificationCenter.post_notification requires \
                 type and sender.")
         
         # If there are no registered observers for the type/sender pair
-        if((theType not in self.registeredTypes) or
-            (sender not in self.registeredSenders)):
+        if((theType not in self.registeredTypes and None not in self.registeredTypes) or
+            (sender not in self.registeredSenders and None not in self.registeredSenders)):
+            print theType,sender,self.registeredTypes,self.registeredSenders
             return
         
-        keys = ((theType,sender), (None, sender), (theType, None), (None,None))
-        observers = set()
-        for k in keys:
-            observers.update(self.observers.get(k, set()))
-        
-        for o in observers:
+        for o in self._observers_for_notification(theType, sender):
             o(theType, sender, args=kwargs)
+    
+        
+    def _observers_for_notification(self, theType, sender):
+        """Find all registered observers that should recieve notification"""
         
+        keys = (
+                    (theType,sender),
+                    (theType, None),
+                    (None, sender),  
+                    (None,None)
+                )
+        
+        
+        obs = set()
+        for k in keys:
+            obs.update(self.observers.get(k, set()))
+        
+        return obs
     
-    def add_observer(self, observerCallback, theType, sender):
+    
+    def add_observer(self, callback, theType, sender):
         """Add an observer callback to this notification center.
         
         The given callback will be called upon posting of notifications of
         the given type/sender and will receive any additional kwargs passed
         to post_notification.
         
         Parameters
         ----------
         observerCallback : callable
             Callable. Must take at least two arguments::
                 observerCallback(type, sender, args={})
         
         theType : hashable
             The notification type. If None, all notifications from sender
             will be posted.
         
         sender : hashable
             The notification sender. If None, all notifications of theType
             will be posted.
         """
-        assert(observerCallback != None)
+        assert(callback != None)
         self.registeredTypes.add(theType)
         self.registeredSenders.add(sender)
-        self.observers.setdefault((theType,sender), set()).add(observerCallback)
+        self.observers.setdefault((theType,sender), set()).add(callback)
+    
+    def remove_all_observers(self):
+        """Removes all observers from this notification center"""
+        
+        self._init_observers()
     
 
 
 sharedCenter = NotificationCenter()
  No newline at end of file

IPython/kernel/core/tests/test_notification.py

@@ -1,118 +1,179 b''
 # encoding: utf-8
 
 """This file contains unittests for the notification.py module."""
 
 __docformat__ = "restructuredtext en"
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team                           
 #                                                                             
 #  Distributed under the terms of the BSD License.  The full license is in    
 #  the file COPYING, distributed as part of this software.                    
 #-----------------------------------------------------------------------------
                                                                               
 #-----------------------------------------------------------------------------
 # Imports                                                                     
 #-----------------------------------------------------------------------------
 
-from IPython.kernel.core.notification import NotificationCenter,\
-                                            sharedCenter
+import unittest
+import IPython.kernel.core.notification as notification
+from nose.tools import timed
 
 #
 # Supporting test classes
 #
 
 class Observer(object):
     """docstring for Observer"""
-    def __init__(self, expectedType, expectedSender, **kwargs):
+    def __init__(self, expectedType, expectedSender, 
+                    center=notification.sharedCenter, **kwargs):
         super(Observer, self).__init__()
         self.expectedType = expectedType
         self.expectedSender = expectedSender
         self.expectedKwArgs = kwargs
         self.recieved = False
-        sharedCenter.add_observer(self.callback, 
-                                    self.expectedType, 
-                                    self.expectedSender)
+        center.add_observer(self.callback, 
+                            self.expectedType, 
+                            self.expectedSender)
     
     
     def callback(self, theType, sender, args={}):
         """callback"""
         
-        assert(theType == self.expectedType)
-        assert(sender == self.expectedSender)
+        assert(theType == self.expectedType or
+                self.expectedType == None)
+        assert(sender == self.expectedSender or
+                self.expectedSender == None)
         assert(args == self.expectedKwArgs)
         self.recieved = True
     
     
     def verify(self):
         """verify"""
         
         assert(self.recieved)
     
+    def reset(self):
+        """reset"""
+        
+        self.recieved = False
+    
 
 
 class Notifier(object):
     """docstring for Notifier"""
     def __init__(self, theType, **kwargs):
         super(Notifier, self).__init__()
         self.theType = theType
         self.kwargs = kwargs
     
-    def post(self, center=sharedCenter):
+    def post(self, center=notification.sharedCenter):
         """fire"""
         
         center.post_notification(self.theType, self,
             **self.kwargs)
     
 
 #
 # Test Cases
 #
 
-
-def test_notification_delivered():
-    """Test that notifications are delivered"""
-    expectedType = 'EXPECTED_TYPE'
-    sender = Notifier(expectedType)
-    observer = Observer(expectedType, sender)
+class NotificationTests(unittest.TestCase):
+    """docstring for NotificationTests"""
     
-    sender.post()
+    def tearDown(self):
+        notification.sharedCenter.remove_all_observers()
     
-    observer.verify()
-
-
-def test_type_specificity():
-    """Test that observers are registered by type"""
+    def test_notification_delivered(self):
+        """Test that notifications are delivered"""
+        expectedType = 'EXPECTED_TYPE'
+        sender = Notifier(expectedType)
+        observer = Observer(expectedType, sender)
+        
+        sender.post()
+        
+        observer.verify()
     
-    expectedType = 1
-    unexpectedType = "UNEXPECTED_TYPE"
-    sender = Notifier(expectedType)
-    unexpectedSender = Notifier(unexpectedType)
-    observer = Observer(expectedType, sender)
     
-    sender.post()
-    unexpectedSender.post()
+    def test_type_specificity(self):
+        """Test that observers are registered by type"""
+        
+        expectedType = 1
+        unexpectedType = "UNEXPECTED_TYPE"
+        sender = Notifier(expectedType)
+        unexpectedSender = Notifier(unexpectedType)
+        observer = Observer(expectedType, sender)
     
-    observer.verify()
-
-
-def test_sender_specificity():
-    """Test that observers are registered by sender"""
+        sender.post()
+        unexpectedSender.post()
     
-    expectedType = "EXPECTED_TYPE"
-    sender1 = Notifier(expectedType)
-    sender2 = Notifier(expectedType)
-    observer = Observer(expectedType, sender1)
+        observer.verify()
     
-    sender1.post()
-    sender2.post()
     
-    observer.verify()
-
-
-def test_complexity_with_no_observers():
-    """Test that the notification center's algorithmic complexity is O(1)
-    with no registered observers (for the given notification type)
-    """
+    def test_sender_specificity(self):
+        """Test that observers are registered by sender"""
+        
+        expectedType = "EXPECTED_TYPE"
+        sender1 = Notifier(expectedType)
+        sender2 = Notifier(expectedType)
+        observer = Observer(expectedType, sender1)
+        
+        sender1.post()
+        sender2.post()
+        
+        observer.verify()
+    
+    
+    def test_remove_all_observers(self):
+        """White-box test for remove_all_observers"""
+        
+        for i in xrange(10):
+            Observer('TYPE', None, center=notification.sharedCenter)
+        
+        self.assert_(len(notification.sharedCenter.observers[('TYPE',None)]) >= 10, 
+            "observers registered")
+        
+        notification.sharedCenter.remove_all_observers()
+        
+        self.assert_(len(notification.sharedCenter.observers) == 0, "observers removed")
+    
+    
+    def test_any_sender(self):
+        """test_any_sender"""
+        
+        expectedType = "EXPECTED_TYPE"
+        sender1 = Notifier(expectedType)
+        sender2 = Notifier(expectedType)
+        observer = Observer(expectedType, None)
+        
+        
+        sender1.post()
+        observer.verify()
+        
+        observer.reset()
+        sender2.post()
+        observer.verify()
+        
+    
+    @timed(.01)
+    def test_post_performance(self):
+        """Test that post_notification, even with many registered irrelevant
+        observers is fast"""
+        
+        for i in xrange(10):
+            Observer("UNRELATED_TYPE", None)
+        
+        o = Observer('EXPECTED_TYPE', None)
+        
+        notification.sharedCenter.post_notification('EXPECTED_TYPE', self)
+        
+        o.verify()
+    
+    
+    def test_complexity_with_no_observers(self):
+        """Test that the notification center's algorithmic complexity is O(1)
+        with no registered observers (for the given notification type)
+        """
+        
+        self.fail("I'm not sure how to test this.")
     
-    assert(False) #I'm not sure how to test this yet