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

@@ -20,9 +20,24 b' 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

IPython/frontend/tests/test_frontendbase.py

@@ -16,10 +16,11 b' __docformat__ = "restructuredtext en"'
 #---------------------------------------------------------------------------
 
 import unittest
-from IPython.frontend import frontendbase
+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, 
@@ -53,7 +54,7 b' class TestAsyncFrontendBase(unittest.TestCase):'
     
     def test_implements_IFrontEnd(self):
         assert(frontendbase.IFrontEnd.implementedBy(
-                                    frontendbase.AsyncFrontEndBase))
+                                    AsyncFrontEndBase))
     
     
     def test_is_complete_returns_False_for_incomplete_block(self):

IPython/kernel/core/tests/test_redirectors.py

@@ -1,31 +1,25 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.                  
-#---------------------------------------------------------------------------
- 
+#-------------------------------------------------------------------------------
+#  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.
+#-------------------------------------------------------------------------------
+
 
 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()
@@ -40,19 +34,13 b' def test_redirector():'
         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)
@@ -66,16 +54,8 b' def test_redirector_output_trap():'
         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."
+