test_process.py
68 lines
| 1.8 KiB
| text/x-python
|
PythonLexer
|
r1457 | # encoding: utf-8 | ||
""" | ||||
Test process execution and IO redirection. | ||||
""" | ||||
__docformat__ = "restructuredtext en" | ||||
|
r1960 | #----------------------------------------------------------------------------- | ||
# Copyright (C) 2008-2009 The IPython Development Team | ||||
|
r1457 | # | ||
# Distributed under the terms of the BSD License. The full license is | ||||
# in the file COPYING, distributed as part of this software. | ||||
|
r1960 | #----------------------------------------------------------------------------- | ||
|
r1457 | |||
from cStringIO import StringIO | ||||
from time import sleep | ||||
import sys | ||||
|
r1947 | from IPython.frontend.process import PipedProcess | ||
|
r1561 | from IPython.testing import decorators as testdec | ||
|
r1457 | |||
|
r1616 | |||
|
r1457 | 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() | ||||
|
r1561 | result = s.getvalue().rstrip() | ||
assert result == '1' | ||||
|
r1457 | |||
|
r1616 | |||
|
r1457 | def test_io(): | ||
""" Checks that we can send characters on stdin to the process. | ||||
""" | ||||
s = StringIO() | ||||
|
r4872 | p = PipedProcess(sys.executable + ' -c "a = raw_input(); print a"', | ||
|
r1457 | 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() | ||||
|
r1561 | result = s.getvalue() | ||
assert result == test_string | ||||
|
r1457 | |||
|
r2513 | @testdec.skip_win32 | ||
|
r1457 | def test_kill(): | ||
""" Check that we can kill a process, and its subprocess. | ||||
""" | ||||
s = StringIO() | ||||
|
r4872 | p = PipedProcess(sys.executable + ' -c "a = raw_input();"', | ||
|
r1457 | out_callback=s.write, ) | ||
p.start() | ||||
while not hasattr(p, 'process'): | ||||
sleep(0.1) | ||||
|
r4872 | p.process.kill() | ||
|
r1457 | assert p.process.poll() is not None | ||
if __name__ == '__main__': | ||||
test_capture_out() | ||||
test_io() | ||||
test_kill() | ||||