More tests for frontends.

IPython/frontend/tests/test_process.py

@@ -0,0 +1,66 b''
+# encoding: utf-8
+"""
+Test process execution and IO redirection.
+"""
+
+__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.
+#-------------------------------------------------------------------------------
+
+from cStringIO import StringIO
+from time import sleep
+import sys
+
+from IPython.frontend._process import PipedProcess
+
+def test_capture_out():
+    """ A simple test to see if we can execute a process and get the output.
+    """
+    s = StringIO()
+    p = PipedProcess('echo 1', out_callback=s.write, )
+    p.start()
+    p.join()
+    assert s.getvalue() == '1\n'
+
+
+def test_io():
+    """ Checks that we can send characters on stdin to the process.
+    """
+    s = StringIO()
+    p = PipedProcess(sys.executable + ' -c "a = raw_input(); print a"', 
+                            out_callback=s.write, )
+    p.start()
+    test_string = '12345\n'
+    while not hasattr(p, 'process'):
+        sleep(0.1)
+    p.process.stdin.write(test_string)
+    p.join()
+    assert s.getvalue() == test_string
+
+
+def test_kill():
+    pass
+
+if True:
+    """ Check that we can kill a process, and its subprocess.
+    """
+    s = StringIO()
+    p = PipedProcess(sys.executable + ' -c "a = raw_input();"', 
+                            out_callback=s.write, )
+    p.start()
+    while not hasattr(p, 'process'):
+        sleep(0.1)
+    p.process.kill() 
+    assert p.process.poll() is not None
+
+
+if __name__ == '__main__':
+    test_capture_out()
+    test_io()
+    test_kill()
+

IPython/frontend/_process/pipedprocess.py

@@ -1,55 +1,70 b''
 # encoding: utf-8
 """
 Object for encapsulating process execution by using callbacks for stdout, 
 stderr and stdin.
 """
 __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 killableprocess import Popen, PIPE
 from threading import Thread
 from time import sleep
 
 class PipedProcess(Thread):
+    """ Class that encapsulates process execution by using callbacks for
+        stdout, stderr and stdin, and providing a reliable way of
+        killing it.
+    """
 
     def __init__(self, command_string, out_callback, 
                         end_callback=None,):
+        """ command_string: the command line executed to start the
+        process. 
+
+        out_callback: the python callable called on stdout/stderr.
+
+        end_callback: an optional callable called when the process
+        finishes.
+
+        These callbacks are called from a different thread as the
+        thread from which is started.
+        """
         self.command_string = command_string
         self.out_callback = out_callback
         self.end_callback = end_callback
         Thread.__init__(self)
     
 
     def run(self):
         """ Start the process and hook up the callbacks.
         """
         process = Popen((self.command_string + ' 2>&1', ), shell=True,
                                 universal_newlines=True,
                                 stdout=PIPE, stdin=PIPE, )
         self.process = process
         while True:
             out_char = process.stdout.read(1)
             if out_char == '':
                 if process.poll() is not None:
                     # The process has finished
                     break
                 else:
                     # The process is not giving any interesting
                     # output. No use polling it immediatly.
                     sleep(0.1)
             else:
                 self.out_callback(out_char)
 
         if self.end_callback is not None:
             self.end_callback()
     
 

IPython/frontend/tests/test_frontendbase.py

@@ -1,151 +1,152 b''
 # encoding: utf-8
 
 """This file contains unittests for the frontendbase 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                                                                   
 #---------------------------------------------------------------------------
 
 import unittest
+from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
 from IPython.frontend import frontendbase
 from IPython.kernel.engineservice import EngineService
 
-class FrontEndCallbackChecker(frontendbase.AsyncFrontEndBase):
+class FrontEndCallbackChecker(AsyncFrontEndBase):
     """FrontEndBase subclass for checking callbacks"""
     def __init__(self, engine=None, history=None):
         super(FrontEndCallbackChecker, self).__init__(engine=engine, 
                                                     history=history)
         self.updateCalled = False
         self.renderResultCalled = False
         self.renderErrorCalled = False
     
     def update_cell_prompt(self, result, blockID=None):
         self.updateCalled = True
         return result
     
     def render_result(self, result):
         self.renderResultCalled = True
         return result
     
     
     def render_error(self, failure):
         self.renderErrorCalled = True
         return failure
     
 
 
     
 class TestAsyncFrontendBase(unittest.TestCase):
     def setUp(self):
         """Setup the EngineService and FrontEndBase"""
         
         self.fb = FrontEndCallbackChecker(engine=EngineService())
     
     
     def test_implements_IFrontEnd(self):
         assert(frontendbase.IFrontEnd.implementedBy(
-                                    frontendbase.AsyncFrontEndBase))
+                                    AsyncFrontEndBase))
     
     
     def test_is_complete_returns_False_for_incomplete_block(self):
         """"""
         
         block = """def test(a):"""
         
         assert(self.fb.is_complete(block) == False)
     
     def test_is_complete_returns_True_for_complete_block(self):
         """"""
         
         block = """def test(a): pass"""
             
         assert(self.fb.is_complete(block))
         
         block = """a=3"""
         
         assert(self.fb.is_complete(block))
     
     
     def test_blockID_added_to_result(self):
         block = """3+3"""
         
         d = self.fb.execute(block, blockID='TEST_ID')
         
         d.addCallback(self.checkBlockID, expected='TEST_ID')
     
     def test_blockID_added_to_failure(self):
         block = "raise  Exception()"
         
         d = self.fb.execute(block,blockID='TEST_ID')
         d.addErrback(self.checkFailureID, expected='TEST_ID')
     
     def checkBlockID(self, result, expected=""):
         assert(result['blockID'] == expected)
     
     
     def checkFailureID(self, failure, expected=""):
         assert(failure.blockID == expected)
     
     
     def test_callbacks_added_to_execute(self):
         """test that
         update_cell_prompt
         render_result
         
         are added to execute request
         """
         
         d = self.fb.execute("10+10")
         d.addCallback(self.checkCallbacks)
     
     
     def checkCallbacks(self, result):
         assert(self.fb.updateCalled)
         assert(self.fb.renderResultCalled)
     
     
     def test_error_callback_added_to_execute(self):
         """test that render_error called on execution error"""
         
         d = self.fb.execute("raise Exception()")
         d.addCallback(self.checkRenderError)
     
     def checkRenderError(self, result):
         assert(self.fb.renderErrorCalled)
     
     def test_history_returns_expected_block(self):
         """Make sure history browsing doesn't fail"""
         
         blocks = ["a=1","a=2","a=3"]
         for b in blocks:
             d = self.fb.execute(b)
         
         # d is now the deferred for the last executed block
         d.addCallback(self.historyTests, blocks)
         
     
     def historyTests(self, result, blocks):
         """historyTests"""
         
         assert(len(blocks) >= 3)
         assert(self.fb.get_history_previous("") == blocks[-2])
         assert(self.fb.get_history_previous("") == blocks[-3])
         assert(self.fb.get_history_next() == blocks[-2])
     
     
     def test_history_returns_none_at_startup(self):
         """test_history_returns_none_at_startup"""
         
         assert(self.fb.get_history_previous("")==None)
         assert(self.fb.get_history_next()==None)
     
     

IPython/kernel/core/tests/test_redirectors.py

@@ -1,81 +1,61 b''
 # encoding: utf-8
-
 """
 Test the output capture at the OS level, using file descriptors.
 """
 
 __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.                  
-#---------------------------------------------------------------------------
+#  Distributed under the terms of the BSD License.  The full license is
+#  in the file COPYING, distributed as part of this software.
+#-------------------------------------------------------------------------------
  
 
 import os
 from cStringIO import StringIO
-import sys
 
 
 def test_redirector():
     """ Checks that the redirector can be used to do synchronous capture.
     """
-    # Flush the stdout, so as not to have side effects between
-    # tests.
-    sys.stdout.flush()
-    sys.stderr.flush()
     from IPython.kernel.core.fd_redirector import FDRedirector
     r = FDRedirector()
     out = StringIO()
     try:
         r.start()
         for i in range(10):
             os.system('echo %ic' % i)
             print >>out, r.getvalue(),
             print >>out, i
     except:
         r.stop()
         raise
     r.stop()
     assert out.getvalue() == "".join("%ic\n%i\n" %(i, i) for i in range(10))
-    sys.stdout.flush()
-    sys.stderr.flush()
 
 
 def test_redirector_output_trap():
     """ This test check not only that the redirector_output_trap does
         trap the output, but also that it does it in a gready way, that
-        is by calling the callback ASAP.
+        is by calling the callabck ASAP.
     """
-    # Flush the stdout, so as not to have side effects between
-    # tests.
-    sys.stdout.flush()
-    sys.stderr.flush()
     from IPython.kernel.core.redirector_output_trap import RedirectorOutputTrap
     out = StringIO()
     trap = RedirectorOutputTrap(out.write, out.write)
     try:
         trap.set()
         for i in range(10):
             os.system('echo %ic' % i)
             print "%ip" % i
             print >>out, i
     except:
         trap.unset()
         raise
     trap.unset()
-    sys.stdout.flush()
-    sys.stderr.flush()
     assert out.getvalue() == "".join("%ic\n%ip\n%i\n" %(i, i, i) 
                                                     for i in range(10))
 
 
 
-if __name__ == '__main__':
-    print "Testing redirector...",
-    test_redirector()
-    test_redirector_output_trap()
-    print "Done."
-