Semi-working refactored ipcluster.

This version uses the NotificationCenter, which I am not sure I like.
But, it is important to see how all of this works.

IPython/config/loader.py

@@ -271,6 +271,7 b' class CommandLineConfigLoader(ConfigLoader):'
 class NoConfigDefault(object): pass
 NoConfigDefault = NoConfigDefault()
 
+
 class ArgParseConfigLoader(CommandLineConfigLoader):
     
     # arguments = [(('-f','--file'),dict(type=str,dest='file'))]

IPython/kernel/clusterdir.py

@@ -58,18 +58,29 b' class ClusterDir(Component):'
             os.chmod(new, 0777)
         self.security_dir = os.path.join(new, self.security_dir_name)
         self.log_dir = os.path.join(new, self.log_dir_name)
+        self.check_dirs()
 
     def _log_dir_changed(self, name, old, new):
-        if not os.path.isdir(new):
-            os.mkdir(new, 0777)
+        self.check_log_dir()
+
+    def check_log_dir(self):
+        if not os.path.isdir(self.log_dir):
+            os.mkdir(self.log_dir, 0777)
         else:
-            os.chmod(new, 0777)
+            os.chmod(self.log_dir, 0777)
 
     def _security_dir_changed(self, name, old, new):
-        if not os.path.isdir(new):
-            os.mkdir(new, 0700)
+        self.check_security_dir()
+
+    def check_security_dir(self):
+        if not os.path.isdir(self.security_dir):
+            os.mkdir(self.security_dir, 0700)
         else:
-            os.chmod(new, 0700)
+            os.chmod(self.security_dir, 0700)
+
+    def check_dirs(self):
+        self.check_security_dir()
+        self.check_log_dir()
 
     def load_config_file(self, filename):
         """Load a config file from the top level of the cluster dir.
@@ -83,7 +94,7 b' class ClusterDir(Component):'
         loader = PyFileConfigLoader(filename, self.location)
         return loader.load_config()
 
-    def copy_config_file(self, config_file, path=None):
+    def copy_config_file(self, config_file, path=None, overwrite=False):
         """Copy a default config file into the active cluster directory.
 
         Default configuration files are kept in :mod:`IPython.config.default`.
@@ -96,12 +107,14 b' class ClusterDir(Component):'
             path = os.path.sep.join(path)
         src = os.path.join(path, config_file)
         dst = os.path.join(self.location, config_file)
+        if not os.path.isfile(dst) or overwrite:
         shutil.copy(src, dst)
 
-    def copy_all_config_files(self):
+    def copy_all_config_files(self, path=None, overwrite=False):
         """Copy all config files into the active cluster directory."""
-        for f in ['ipcontroller_config.py', 'ipengine_config.py']:
-            self.copy_config_file(f)
+        for f in ['ipcontroller_config.py', 'ipengine_config.py',
+                  'ipcluster_config.py']:
+            self.copy_config_file(f, path=path, overwrite=overwrite)
 
     @classmethod
     def find_cluster_dir_by_profile(cls, path, profile='default'):
@@ -245,7 +258,6 b' class ApplicationWithClusterDir(Application):'
         # priority, this will always end up in the master_config.
         self.default_config.Global.cluster_dir = self.cluster_dir
         self.command_line_config.Global.cluster_dir = self.cluster_dir
-        self.log.info("Cluster directory set to: %s" % self.cluster_dir)
 
         # Set the search path to the cluster directory
         self.config_file_paths = (self.cluster_dir,)

IPython/kernel/engineconnector.py

@@ -20,14 +20,21 b' import cPickle as pickle'
 
 from twisted.python import log, failure
 from twisted.internet import defer
+from twisted.internet.defer import inlineCallbacks, returnValue
 
 from IPython.kernel.fcutil import find_furl
 from IPython.kernel.enginefc import IFCEngine
+from IPython.kernel.twistedutil import sleep_deferred
 
 #-------------------------------------------------------------------------------
 # The ClientConnector class
 #-------------------------------------------------------------------------------
 
+
+class EngineConnectorError(Exception):
+    pass
+
+
 class EngineConnector(object):
     """Manage an engines connection to a controller.
     
@@ -39,7 +46,8 b' class EngineConnector(object):'
     def __init__(self, tub):
         self.tub = tub
         
-    def connect_to_controller(self, engine_service, furl_or_file):
+    def connect_to_controller(self, engine_service, furl_or_file,
+                              delay=0.1, max_tries=10):
         """
         Make a connection to a controller specified by a furl.
         
@@ -53,29 +61,60 b' class EngineConnector(object):'
         `register_engine` method of the controller to actually register the 
         engine.
         
-        :Parameters:
+        This method will try to connect to the controller multiple times with
+        a delay in between.  Each time the FURL file is read anew.
+
+        Parameters
+        __________
             engine_service : IEngineBase
                 An instance of an `IEngineBase` implementer
             furl_or_file : str
                 A furl or a filename containing a furl
+        delay : float
+            The intial time to wait between connection attempts.  Subsequent
+            attempts have increasing delays.
+        max_tries : int
+            The maximum number of connection attempts.
         """
         if not self.tub.running:
             self.tub.startService()
         self.engine_service = engine_service
         self.engine_reference = IFCEngine(self.engine_service)
+
+        d = self._try_to_connect(furl_or_file, delay, max_tries, attempt=0)
+        return d
+
+    @inlineCallbacks
+    def _try_to_connect(self, furl_or_file, delay, max_tries, attempt):
+        """Try to connect to the controller with retry logic."""
+        if attempt < max_tries:
+            log.msg("Attempting to connect to controller [%r]: %s" % \
+                (attempt, furl_or_file))
         try:
             self.furl = find_furl(furl_or_file)
-        except ValueError:
-            return defer.fail(failure.Failure())
+                # Uncomment this to see the FURL being tried.
+                # log.msg("FURL: %s" % self.furl)
+                rr = yield self.tub.getReference(self.furl)
+            except:
+                if attempt==max_tries-1:
+                    # This will propagate the exception all the way to the top
+                    # where it can be handled.
+                    raise
         else:
-            d = self.tub.getReference(self.furl)
-            d.addCallbacks(self._register, self._log_failure)
-            return d
-    
-    def _log_failure(self, reason):
-        log.err('EngineConnector: engine registration failed:')
-        log.err(reason)
-        return reason
+                    yield sleep_deferred(delay)
+                    yield self._try_to_connect(
+                        furl_or_file, 1.5*delay, max_tries, attempt+1
+                    )
+            else:
+                result = yield self._register(rr)
+                returnValue(result)
+        else:
+            raise EngineConnectorError(
+                'Could not connect to controller, max_tries (%r) exceeded. '
+                'This usually means that i) the controller was not started, '
+                'or ii) a firewall was blocking the engine from connecting '
+                'to the controller.' % max_tries
+            )
 
     def _register(self, rr):
         self.remote_ref = rr
@@ -83,7 +122,7 b' class EngineConnector(object):'
         desired_id = self.engine_service.id
         d = self.remote_ref.callRemote('register_engine', self.engine_reference, 
             desired_id, os.getpid(), pickle.dumps(self.engine_service.properties,2))
-        return d.addCallbacks(self._reference_sent, self._log_failure)
+        return d.addCallback(self._reference_sent)
 
     def _reference_sent(self, registration_dict):
         self.engine_service.id = registration_dict['id']

IPython/kernel/fcutil.py

@@ -15,6 +15,8 b' Foolscap related utilities.'
 # Imports
 #-----------------------------------------------------------------------------
 
+from __future__ import with_statement
+
 import os
 import tempfile
 
@@ -85,10 +87,11 b' def find_furl(furl_or_file):'
         if is_valid(furl_or_file):
             return furl_or_file
     if os.path.isfile(furl_or_file):
-        furl = open(furl_or_file, 'r').read().strip()
+        with open(furl_or_file, 'r') as f:
+            furl = f.read().strip()
         if is_valid(furl):
             return furl
-    raise ValueError("not a FURL or a file containing a FURL: %s" % furl_or_file)
+    raise ValueError("Not a FURL or a file containing a FURL: %s" % furl_or_file)
 
 
 def get_temp_furlfile(filename):

IPython/kernel/ipcontrollerapp.py

@@ -213,6 +213,7 b' class IPControllerApp(ApplicationWithClusterDir):'
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
+        self.log.info("Cluster directory set to: %s" % self.cluster_dir)
         self.log.info("Log directory set to: %s" % self.log_dir)
         self.log.info("Security directory set to: %s" % self.security_dir)
 
@@ -266,3 +267,8 b' def launch_new_instance():'
     """Create and run the IPython controller"""
     app = IPControllerApp()
     app.start()
+
+
+if __name__ == '__main__':
+    launch_new_instance()
+

IPython/kernel/ipengineapp.py

@@ -133,29 +133,27 b' class IPEngineApp(ApplicationWithClusterDir):'
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
+        self.log.info("Cluster directory set to: %s" % self.cluster_dir)
         self.log.info("Log directory set to: %s" % self.log_dir)
         self.log.info("Security directory set to: %s" % self.security_dir)
 
         self.find_cont_furl_file()
 
     def find_cont_furl_file(self):
+        """Set the furl file.
+
+        Here we don't try to actually see if it exists for is valid as that
+        is hadled by the connection logic.
+        """
         config = self.master_config
         # Find the actual controller FURL file
-        if os.path.isfile(config.Global.furl_file):
-            return
-        else:
-            # We should know what the app dir is
+        if not config.Global.furl_file:
             try_this = os.path.join(
                 config.Global.cluster_dir, 
                 config.Global.security_dir,
                 config.Global.furl_file_name
             )
-            if os.path.isfile(try_this):
                 config.Global.furl_file = try_this
-                return
-            else:
-                self.log.critical("Could not find a valid controller FURL file.")
-                self.abort()
 
     def construct(self):
         # I am a little hesitant to put these into InteractiveShell itself.
@@ -194,11 +192,11 b' class IPEngineApp(ApplicationWithClusterDir):'
         )
 
         def handle_error(f):
-            # If this print statement is replaced by a log.err(f) I get
-            # an unhandled error, which makes no sense.  I shouldn't have
-            # to use a print statement here.  My only thought is that
-            # at the beginning of the process the logging is still starting up
-            print "Error connecting to controller:", f.getErrorMessage()
+            log.msg('Error connecting to controller. This usually means that '
+            'i) the controller was not started, ii) a firewall was blocking '
+            'the engine from connecting to the controller or iii) the engine '
+            ' was not pointed at the right FURL file:')
+            log.msg(f.getErrorMessage())
             reactor.callLater(0.1, reactor.stop)
 
         d.addErrback(handle_error)
@@ -243,3 +241,8 b' def launch_new_instance():'
     """Create and run the IPython controller"""
     app = IPEngineApp()
     app.start()
+
+
+if __name__ == '__main__':
+    launch_new_instance()
+

IPython/kernel/scripts/ipcluster

@@ -1,22 +1,18 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
-"""ipcluster script"""
-
-__docformat__ = "restructuredtext en"
-
-#-------------------------------------------------------------------------------
-#  Copyright (C) 2008  The IPython Development Team
+#-----------------------------------------------------------------------------
+#  Copyright (C) 2008-2009  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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
+
 
-if __name__ == '__main__':
-	from IPython.kernel.scripts import ipcluster
-	ipcluster.main()
+from IPython.kernel.ipclusterapp import launch_new_instance
 
+launch_new_instance()

IPython/kernel/scripts/ipcluster.py

@@ -4,7 +4,7 b''
 """Start an IPython cluster = (controller + engines)."""
 
 #-----------------------------------------------------------------------------
-#  Copyright (C) 2008  The IPython Development Team
+#  Copyright (C) 2008-2009  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.
@@ -25,7 +25,7 b' from twisted.internet import reactor, defer'
 from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.error import ProcessDone, ProcessTerminated
 from twisted.internet.utils import getProcessOutput
-from twisted.python import failure, log
+from twisted.python import log
 
 from IPython.external import argparse
 from IPython.external import Itpl
@@ -49,10 +49,8 b' get_log_dir()'
 get_security_dir()
 
 from IPython.kernel.config import config_manager as kernel_config_manager
-from IPython.kernel.error import SecurityError, FileTimeoutError
-from IPython.kernel.fcutil import have_crypto
 from IPython.kernel.twistedutil import gatherBoth, wait_for_file
-from IPython.kernel.util import printer
+
 
 #-----------------------------------------------------------------------------
 # General process handling code
@@ -140,7 +138,7 b' class ProcessLauncher(object):'
             )
             return self.start_deferred
         else:
-            s = 'the process has already been started and has state: %r' % \
+            s = 'The process has already been started and has state: %r' % \
                 self.state
             return defer.fail(ProcessStateError(s))
     

IPython/kernel/twistedutil.py

@@ -247,3 +247,10 b' def wait_for_file(filename, delay=0.1, max_tries=10):'
     
     _test_for_file(filename)
     return d
+
+
+def sleep_deferred(seconds):
+    """Sleep without blocking the event loop."""
+    d = defer.Deferred()
+    reactor.callLater(seconds, d.callback, seconds)
+    return d

IPython/utils/notification.py

@@ -1,42 +1,52 b''
+#!/usr/bin/env python
 # encoding: utf-8
-
-"""The IPython Core Notification Center.
+"""
+The IPython Core Notification Center.
 
 See docs/source/development/notification_blueprint.txt for an overview of the
 notification module.
-"""
 
-__docformat__ = "restructuredtext en"
+Authors:
+
+* Barry Wark
+* Brian Granger
+"""
                                                                               
 #-----------------------------------------------------------------------------
-#  Copyright (C) 2008  The IPython Development Team                           
+#  Copyright (C) 2008-2009  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.                    
 #-----------------------------------------------------------------------------
 
-# Tell nose to skip the testing of this module
-__test__ = {}
+#-----------------------------------------------------------------------------
+# Code
+#-----------------------------------------------------------------------------
+
+
+class NotificationError(Exception):
+    pass
+
 
 class NotificationCenter(object):
-    """Synchronous notification center
+    """Synchronous notification center.
     
     Examples
     --------
-    >>> import IPython.kernel.core.notification as notification
-    >>> def callback(theType, theSender, args={}):
-    ...     print theType,theSender,args
-    ...     
-    >>> notification.sharedCenter.add_observer(callback, 'NOTIFICATION_TYPE', None)
-    >>> notification.sharedCenter.post_notification('NOTIFICATION_TYPE', object()) # doctest:+ELLIPSIS
-    NOTIFICATION_TYPE ...
+    Here is a simple example of how to use this::
         
+        import IPython.kernel.core.notification as notification
+        def callback(ntype, theSender, args={}):
+            print ntype,theSender,args
+
+        notification.sharedCenter.add_observer(callback, 'NOTIFICATION_TYPE', None)
+        notification.sharedCenter.post_notification('NOTIFICATION_TYPE', object()) # doctest:+ELLIPSIS
+        NOTIFICATION_TYPE ...
     """
     def __init__(self):
         super(NotificationCenter, self).__init__()
         self._init_observers()
     
-    
     def _init_observers(self):
         """Initialize observer storage"""
         
@@ -44,76 +54,84 b' class NotificationCenter(object):'
         self.registered_senders = set() #set of senders that are observed
         self.observers = {} #map (type,sender) => callback (callable)
     
+    def post_notification(self, ntype, sender, *args, **kwargs):
+        """Post notification to all registered observers.
     
-    def post_notification(self, theType, sender, **kwargs):
-        """Post notification (type,sender,**kwargs) to all registered
-        observers.
+        The registered callback will be called as::
 
-        Implementation notes:
+            callback(ntype, sender, *args, **kwargs)
 
+        Parameters
+        ----------
+        ntype : hashable
+            The notification type.
+        sender : hashable
+            The object sending the notification.
+        *args : tuple
+            The positional arguments to be passed to the callback.
+        **kwargs : dict
+            The keyword argument to be passed to the callback.
+
+        Notes
+        -----
         * 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(ntype==None or sender==None):
+            raise NotificationError(
+                "Notification type and sender are required.")
         
         # If there are no registered observers for the type/sender pair
-        if((theType not in self.registered_types and 
+        if((ntype not in self.registered_types and 
                 None not in self.registered_types) or
             (sender not in self.registered_senders and 
                 None not in self.registered_senders)):
             return
         
-        for o in self._observers_for_notification(theType, sender):
-            o(theType, sender, args=kwargs)
+        for o in self._observers_for_notification(ntype, sender):
+            o(ntype, sender, *args, **kwargs)
     
-        
-    def _observers_for_notification(self, theType, sender):
+    def _observers_for_notification(self, ntype, sender):
         """Find all registered observers that should recieve notification"""
         
         keys = (
-                    (theType,sender),
-                    (theType, None),
+                   (ntype,sender),
+                   (ntype, None),
                     (None, sender),  
                     (None,None)
                 )
         
-        
         obs = set()
         for k in keys:
             obs.update(self.observers.get(k, set()))
         
         return obs
     
-    
-    def add_observer(self, callback, theType, sender):
+    def add_observer(self, callback, ntype, 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
+        the given type/sender and will receive any additional arguments passed
         to post_notification.
         
         Parameters
         ----------
-        observerCallback : callable
-            Callable. Must take at least two arguments::
-                observerCallback(type, sender, args={})
-        
-        theType : hashable
+        callback : callable
+            The callable that will be called by :meth:`post_notification`
+            as ``callback(ntype, sender, *args, **kwargs)
+        ntype : hashable
             The notification type. If None, all notifications from sender
             will be posted.
-        
         sender : hashable
-            The notification sender. If None, all notifications of theType
+            The notification sender. If None, all notifications of ntype
             will be posted.
         """
         assert(callback != None)
-        self.registered_types.add(theType)
+        self.registered_types.add(ntype)
         self.registered_senders.add(sender)
-        self.observers.setdefault((theType,sender), set()).add(callback)
+        self.observers.setdefault((ntype,sender), set()).add(callback)
     
     def remove_all_observers(self):
         """Removes all observers from this notification center"""
@@ -122,4 +140,4 b' class NotificationCenter(object):'
     
 
 
-sharedCenter = NotificationCenter()
+shared_center = NotificationCenter()

IPython/utils/tests/test_notification.py

@@ -13,104 +13,102 b''
 # Imports
 #-----------------------------------------------------------------------------
 
-# Tell nose to skip this module
-__test__ = {}
+import unittest
 
-from twisted.trial import unittest
-import IPython.kernel.core.notification as notification
+from IPython.utils.notification import (
+    NotificationCenter,
+    NotificationError,
+    shared_center
+)
 
 #-----------------------------------------------------------------------------
 # Support Classes
 #-----------------------------------------------------------------------------
 
+
 class Observer(object):
-    """docstring for Observer"""
-    def __init__(self, expectedType, expectedSender, 
-                    center=notification.sharedCenter, **kwargs):
+
+    def __init__(self, expected_ntype, expected_sender, 
+                    center=shared_center, *args, **kwargs):
         super(Observer, self).__init__()
-        self.expectedType = expectedType
-        self.expectedSender = expectedSender
-        self.expectedKwArgs = kwargs
+        self.expected_ntype = expected_ntype
+        self.expected_sender = expected_sender
+        self.expected_args = args
+        self.expected_kwargs = kwargs
         self.recieved = False
         center.add_observer(self.callback, 
-                            self.expectedType, 
-                            self.expectedSender)
-    
-    def callback(self, theType, sender, args={}):
-        """callback"""
-        
-        assert(theType == self.expectedType or
-                self.expectedType == None)
-        assert(sender == self.expectedSender or
-                self.expectedSender == None)
-        assert(args == self.expectedKwArgs)
+                            self.expected_ntype, 
+                            self.expected_sender)
+    
+    def callback(self, ntype, sender, *args, **kwargs):
+        assert(ntype == self.expected_ntype or
+                self.expected_ntype == None)
+        assert(sender == self.expected_sender or
+                self.expected_sender == None)
+        assert(args == self.expected_args)
+        assert(kwargs == self.expected_kwargs)
         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):
+
+    def __init__(self, ntype, **kwargs):
         super(Notifier, self).__init__()
-        self.theType = theType
+        self.ntype = ntype
         self.kwargs = kwargs
     
-    def post(self, center=notification.sharedCenter):
-        """fire"""
+    def post(self, center=shared_center):
         
-        center.post_notification(self.theType, self,
+        center.post_notification(self.ntype, self,
             **self.kwargs)
 
+
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
+
 class NotificationTests(unittest.TestCase):
-    """docstring for NotificationTests"""
     
     def tearDown(self):
-        notification.sharedCenter.remove_all_observers()
+        shared_center.remove_all_observers()
     
     def test_notification_delivered(self):
         """Test that notifications are delivered"""
-        expectedType = 'EXPECTED_TYPE'
-        sender = Notifier(expectedType)
-        observer = Observer(expectedType, sender)
         
-        sender.post()
+        expected_ntype = 'EXPECTED_TYPE'
+        sender = Notifier(expected_ntype)
+        observer = Observer(expected_ntype, sender)
         
+        sender.post()
         observer.verify()
     
     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)
+        expected_ntype = 1
+        unexpected_ntype = "UNEXPECTED_TYPE"
+        sender = Notifier(expected_ntype)
+        unexpected_sender = Notifier(unexpected_ntype)
+        observer = Observer(expected_ntype, sender)
     
         sender.post()
-        unexpectedSender.post()
-    
+        unexpected_sender.post()
         observer.verify()
     
     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)
+        expected_ntype = "EXPECTED_TYPE"
+        sender1 = Notifier(expected_ntype)
+        sender2 = Notifier(expected_ntype)
+        observer = Observer(expected_ntype, sender1)
         
         sender1.post()
         sender2.post()
@@ -121,23 +119,19 b' class NotificationTests(unittest.TestCase):'
         """White-box test for remove_all_observers"""
         
         for i in xrange(10):
-            Observer('TYPE', None, center=notification.sharedCenter)
+            Observer('TYPE', None, center=shared_center)
         
-        self.assert_(len(notification.sharedCenter.observers[('TYPE',None)]) >= 10, 
+        self.assert_(len(shared_center.observers[('TYPE',None)]) >= 10, 
             "observers registered")
         
-        notification.sharedCenter.remove_all_observers()
-        
-        self.assert_(len(notification.sharedCenter.observers) == 0, "observers removed")
+        shared_center.remove_all_observers()
+        self.assert_(len(shared_center.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)
-        
+        expected_ntype = "EXPECTED_TYPE"
+        sender1 = Notifier(expected_ntype)
+        sender2 = Notifier(expected_ntype)
+        observer = Observer(expected_ntype, None)
         
         sender1.post()
         observer.verify()
@@ -154,8 +148,7 b' class NotificationTests(unittest.TestCase):'
             Observer("UNRELATED_TYPE", None)
         
         o = Observer('EXPECTED_TYPE', None)
-        
-        notification.sharedCenter.post_notification('EXPECTED_TYPE', self)
-        
+        shared_center.post_notification('EXPECTED_TYPE', self)
         o.verify()
 
+

setup.py

@@ -171,7 +171,7 b" if 'setuptools' in sys.modules:"
             'pycolor = IPython.utils.PyColorize:main',
             'ipcontroller = IPython.kernel.ipcontrollerapp:launch_new_instance',
             'ipengine = IPython.kernel.ipengineapp:launch_new_instance',
-            'ipcluster = IPython.kernel.scripts.ipcluster:main',
+            'ipcluster = IPython.kernel.ipclusterapp:launch_new_instance',
             'ipythonx = IPython.frontend.wx.ipythonx:main',
             'iptest = IPython.testing.iptest:main',
             'irunner = IPython.lib.irunner:main'