From f504f142bb98b20ac022ca8e91e54206ee4d7db2 2008-08-10 22:11:34 From: Gael Varoquaux Date: 2008-08-10 22:11:34 Subject: [PATCH] More tests for frontends. --- diff --git a/IPython/frontend/_process/pipedprocess.py b/IPython/frontend/_process/pipedprocess.py index ee60ab4..0d83995 100644 --- a/IPython/frontend/_process/pipedprocess.py +++ b/IPython/frontend/_process/pipedprocess.py @@ -20,9 +20,24 @@ 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 diff --git a/IPython/frontend/tests/test_frontendbase.py b/IPython/frontend/tests/test_frontendbase.py index f74c43e..0b448e5 100644 --- a/IPython/frontend/tests/test_frontendbase.py +++ b/IPython/frontend/tests/test_frontendbase.py @@ -16,10 +16,11 @@ __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 @@ 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): diff --git a/IPython/frontend/tests/test_process.py b/IPython/frontend/tests/test_process.py new file mode 100644 index 0000000..a879029 --- /dev/null +++ b/IPython/frontend/tests/test_process.py @@ -0,0 +1,66 @@ +# 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() + diff --git a/IPython/kernel/core/tests/test_redirectors.py b/IPython/kernel/core/tests/test_redirectors.py index e5bc24e..2fe3d08 100644 --- a/IPython/kernel/core/tests/test_redirectors.py +++ b/IPython/kernel/core/tests/test_redirectors.py @@ -1,31 +1,25 @@ # 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 @@ 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 @@ 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." +