##// END OF EJS Templates
More tests for frontends.
Gael Varoquaux -
Show More
@@ -0,0 +1,66 b''
1 # encoding: utf-8
2 """
3 Test process execution and IO redirection.
4 """
5
6 __docformat__ = "restructuredtext en"
7
8 #-------------------------------------------------------------------------------
9 # Copyright (C) 2008 The IPython Development Team
10 #
11 # Distributed under the terms of the BSD License. The full license is
12 # in the file COPYING, distributed as part of this software.
13 #-------------------------------------------------------------------------------
14
15 from cStringIO import StringIO
16 from time import sleep
17 import sys
18
19 from IPython.frontend._process import PipedProcess
20
21 def test_capture_out():
22 """ A simple test to see if we can execute a process and get the output.
23 """
24 s = StringIO()
25 p = PipedProcess('echo 1', out_callback=s.write, )
26 p.start()
27 p.join()
28 assert s.getvalue() == '1\n'
29
30
31 def test_io():
32 """ Checks that we can send characters on stdin to the process.
33 """
34 s = StringIO()
35 p = PipedProcess(sys.executable + ' -c "a = raw_input(); print a"',
36 out_callback=s.write, )
37 p.start()
38 test_string = '12345\n'
39 while not hasattr(p, 'process'):
40 sleep(0.1)
41 p.process.stdin.write(test_string)
42 p.join()
43 assert s.getvalue() == test_string
44
45
46 def test_kill():
47 pass
48
49 if True:
50 """ Check that we can kill a process, and its subprocess.
51 """
52 s = StringIO()
53 p = PipedProcess(sys.executable + ' -c "a = raw_input();"',
54 out_callback=s.write, )
55 p.start()
56 while not hasattr(p, 'process'):
57 sleep(0.1)
58 p.process.kill()
59 assert p.process.poll() is not None
60
61
62 if __name__ == '__main__':
63 test_capture_out()
64 test_io()
65 test_kill()
66
@@ -20,9 +20,24 b' from threading import Thread'
20 from time import sleep
20 from time import sleep
21
21
22 class PipedProcess(Thread):
22 class PipedProcess(Thread):
23 """ Class that encapsulates process execution by using callbacks for
24 stdout, stderr and stdin, and providing a reliable way of
25 killing it.
26 """
23
27
24 def __init__(self, command_string, out_callback,
28 def __init__(self, command_string, out_callback,
25 end_callback=None,):
29 end_callback=None,):
30 """ command_string: the command line executed to start the
31 process.
32
33 out_callback: the python callable called on stdout/stderr.
34
35 end_callback: an optional callable called when the process
36 finishes.
37
38 These callbacks are called from a different thread as the
39 thread from which is started.
40 """
26 self.command_string = command_string
41 self.command_string = command_string
27 self.out_callback = out_callback
42 self.out_callback = out_callback
28 self.end_callback = end_callback
43 self.end_callback = end_callback
@@ -16,10 +16,11 b' __docformat__ = "restructuredtext en"'
16 #---------------------------------------------------------------------------
16 #---------------------------------------------------------------------------
17
17
18 import unittest
18 import unittest
19 from IPython.frontend.asyncfrontendbase import AsyncFrontEndBase
19 from IPython.frontend import frontendbase
20 from IPython.frontend import frontendbase
20 from IPython.kernel.engineservice import EngineService
21 from IPython.kernel.engineservice import EngineService
21
22
22 class FrontEndCallbackChecker(frontendbase.AsyncFrontEndBase):
23 class FrontEndCallbackChecker(AsyncFrontEndBase):
23 """FrontEndBase subclass for checking callbacks"""
24 """FrontEndBase subclass for checking callbacks"""
24 def __init__(self, engine=None, history=None):
25 def __init__(self, engine=None, history=None):
25 super(FrontEndCallbackChecker, self).__init__(engine=engine,
26 super(FrontEndCallbackChecker, self).__init__(engine=engine,
@@ -53,7 +54,7 b' class TestAsyncFrontendBase(unittest.TestCase):'
53
54
54 def test_implements_IFrontEnd(self):
55 def test_implements_IFrontEnd(self):
55 assert(frontendbase.IFrontEnd.implementedBy(
56 assert(frontendbase.IFrontEnd.implementedBy(
56 frontendbase.AsyncFrontEndBase))
57 AsyncFrontEndBase))
57
58
58
59
59 def test_is_complete_returns_False_for_incomplete_block(self):
60 def test_is_complete_returns_False_for_incomplete_block(self):
@@ -1,31 +1,25 b''
1 # encoding: utf-8
1 # encoding: utf-8
2
3 """
2 """
4 Test the output capture at the OS level, using file descriptors.
3 Test the output capture at the OS level, using file descriptors.
5 """
4 """
6
5
7 __docformat__ = "restructuredtext en"
6 __docformat__ = "restructuredtext en"
8
7
9 #---------------------------------------------------------------------------
8 #-------------------------------------------------------------------------------
10 # Copyright (C) 2008 The IPython Development Team
9 # Copyright (C) 2008 The IPython Development Team
11 #
10 #
12 # Distributed under the terms of the BSD License. The full license is in
11 # Distributed under the terms of the BSD License. The full license is
13 # the file COPYING, distributed as part of this software.
12 # in the file COPYING, distributed as part of this software.
14 #---------------------------------------------------------------------------
13 #-------------------------------------------------------------------------------
15
14
16
15
17 import os
16 import os
18 from cStringIO import StringIO
17 from cStringIO import StringIO
19 import sys
20
18
21
19
22 def test_redirector():
20 def test_redirector():
23 """ Checks that the redirector can be used to do synchronous capture.
21 """ Checks that the redirector can be used to do synchronous capture.
24 """
22 """
25 # Flush the stdout, so as not to have side effects between
26 # tests.
27 sys.stdout.flush()
28 sys.stderr.flush()
29 from IPython.kernel.core.fd_redirector import FDRedirector
23 from IPython.kernel.core.fd_redirector import FDRedirector
30 r = FDRedirector()
24 r = FDRedirector()
31 out = StringIO()
25 out = StringIO()
@@ -40,19 +34,13 b' def test_redirector():'
40 raise
34 raise
41 r.stop()
35 r.stop()
42 assert out.getvalue() == "".join("%ic\n%i\n" %(i, i) for i in range(10))
36 assert out.getvalue() == "".join("%ic\n%i\n" %(i, i) for i in range(10))
43 sys.stdout.flush()
44 sys.stderr.flush()
45
37
46
38
47 def test_redirector_output_trap():
39 def test_redirector_output_trap():
48 """ This test check not only that the redirector_output_trap does
40 """ This test check not only that the redirector_output_trap does
49 trap the output, but also that it does it in a gready way, that
41 trap the output, but also that it does it in a gready way, that
50 is by calling the callback ASAP.
42 is by calling the callabck ASAP.
51 """
43 """
52 # Flush the stdout, so as not to have side effects between
53 # tests.
54 sys.stdout.flush()
55 sys.stderr.flush()
56 from IPython.kernel.core.redirector_output_trap import RedirectorOutputTrap
44 from IPython.kernel.core.redirector_output_trap import RedirectorOutputTrap
57 out = StringIO()
45 out = StringIO()
58 trap = RedirectorOutputTrap(out.write, out.write)
46 trap = RedirectorOutputTrap(out.write, out.write)
@@ -66,16 +54,8 b' def test_redirector_output_trap():'
66 trap.unset()
54 trap.unset()
67 raise
55 raise
68 trap.unset()
56 trap.unset()
69 sys.stdout.flush()
70 sys.stderr.flush()
71 assert out.getvalue() == "".join("%ic\n%ip\n%i\n" %(i, i, i)
57 assert out.getvalue() == "".join("%ic\n%ip\n%i\n" %(i, i, i)
72 for i in range(10))
58 for i in range(10))
73
59
74
60
75
61
76 if __name__ == '__main__':
77 print "Testing redirector...",
78 test_redirector()
79 test_redirector_output_trap()
80 print "Done."
81
General Comments 0
You need to be logged in to leave comments. Login now