[Python-checkins] python/dist/src/Lib/test test_subprocess.py, NONE, 1.1

Tue Oct 12 17:26:30 CEST 2004

Update of /cvsroot/python/python/dist/src/Lib/test
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv24287/Lib/test
Added Files:
	test_subprocess.py 
Log Message:
Added Peter Astrand's subprocess module.
--- NEW FILE: test_subprocess.py ---
import unittest
from test import test_support
import subprocess
import sys
import signal
import os
import tempfile
import time
mswindows = (sys.platform == "win32")
#
# Depends on the following external programs: Python
#
if mswindows:
 SETBINARY = 'import msvcrt; msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY);'
else:
 SETBINARY = ''
class ProcessTestCase(unittest.TestCase):
 def mkstemp(self):
 """wrapper for mkstemp, calling mktemp if mkstemp is not available"""
 if hasattr(tempfile, "mkstemp"):
 return tempfile.mkstemp()
 else:
 fname = tempfile.mktemp()
 return os.open(fname, os.O_RDWR|os.O_CREAT), fname

 #
 # Generic tests
 #
 def test_call_seq(self):
 """call() function with sequence argument"""
 rc = subprocess.call([sys.executable, "-c", "import sys; sys.exit(47)"])
 self.assertEqual(rc, 47)
 def test_call_kwargs(self):
 """call() function with keyword args"""
 newenv = os.environ.copy()
 newenv["FRUIT"] = "banana"
 rc = subprocess.call([sys.executable, "-c",
 'import sys, os;' \
 'sys.exit(os.getenv("FRUIT")=="banana")'],
 env=newenv)
 self.assertEqual(rc, 1)
 def test_stdin_none(self):
 """.stdin is None when not redirected"""
 p = subprocess.Popen([sys.executable, "-c", 'print "banana"'],
 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 p.wait()
 self.assertEqual(p.stdin, None)
 def test_stdout_none(self):
 """.stdout is None when not redirected"""
 p = subprocess.Popen([sys.executable, "-c", 'print "banana"'],
 stdin=subprocess.PIPE, stderr=subprocess.PIPE)
 p.wait()
 self.assertEqual(p.stdout, None)
 def test_stderr_none(self):
 """.stderr is None when not redirected"""
 p = subprocess.Popen([sys.executable, "-c", 'print "banana"'],
 stdin=subprocess.PIPE, stdout=subprocess.PIPE)
 p.wait()
 self.assertEqual(p.stderr, None)
 def test_executable(self):
 """executable"""
 p = subprocess.Popen(["somethingyoudonthave", "-c", "import sys; sys.exit(47)"],
 executable=sys.executable)
 p.wait()
 self.assertEqual(p.returncode, 47)
 def test_stdin_pipe(self):
 """stdin redirection"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.exit(sys.stdin.read() == "pear")'],
 stdin=subprocess.PIPE)
 p.stdin.write("pear")
 p.stdin.close()
 p.wait()
 self.assertEqual(p.returncode, 1)
 def test_stdin_filedes(self):
 """stdin is set to open file descriptor"""
 tf = tempfile.TemporaryFile() 
 d = tf.fileno()
 os.write(d, "pear")
 os.lseek(d, 0, 0)
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.exit(sys.stdin.read() == "pear")'],
 stdin=d)
 p.wait()
 self.assertEqual(p.returncode, 1)
 def test_stdin_fileobj(self):
 """stdin is set to open file object"""
 tf = tempfile.TemporaryFile()
 tf.write("pear")
 tf.seek(0)
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.exit(sys.stdin.read() == "pear")'],
 stdin=tf)
 p.wait()
 self.assertEqual(p.returncode, 1)
 def test_stdout_pipe(self):
 """stdout redirection"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.stdout.write("orange")'],
 stdout=subprocess.PIPE)
 self.assertEqual(p.stdout.read(), "orange")
 def test_stdout_filedes(self):
 """stdout is set to open file descriptor"""
 tf = tempfile.TemporaryFile() 
 d = tf.fileno()
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.stdout.write("orange")'],
 stdout=d)
 p.wait()
 os.lseek(d, 0, 0)
 self.assertEqual(os.read(d, 1024), "orange")
 def test_stdout_fileobj(self):
 """stdout is set to open file object"""
 tf = tempfile.TemporaryFile() 
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.stdout.write("orange")'],
 stdout=tf)
 p.wait()
 tf.seek(0)
 self.assertEqual(tf.read(), "orange")
 def test_stderr_pipe(self):
 """stderr redirection"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.stderr.write("strawberry")'],
 stderr=subprocess.PIPE)
 self.assertEqual(p.stderr.read(), "strawberry")
 def test_stderr_filedes(self):
 """stderr is set to open file descriptor"""
 tf = tempfile.TemporaryFile() 
 d = tf.fileno()
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.stderr.write("strawberry")'],
 stderr=d)
 p.wait()
 os.lseek(d, 0, 0)
 self.assertEqual(os.read(d, 1024), "strawberry")
 def test_stderr_fileobj(self):
 """stderr is set to open file object"""
 tf = tempfile.TemporaryFile() 
 p = subprocess.Popen([sys.executable, "-c",
 'import sys; sys.stderr.write("strawberry")'],
 stderr=tf)
 p.wait()
 tf.seek(0)
 self.assertEqual(tf.read(), "strawberry")
 def test_stdout_stderr_pipe(self):
 """capture stdout and stderr to the same pipe"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys;' \
 'sys.stdout.write("apple");' \
 'sys.stdout.flush();' \
 'sys.stderr.write("orange")'],
 stdout=subprocess.PIPE,
 stderr=subprocess.STDOUT)
 self.assertEqual(p.stdout.read(), "appleorange")
 def test_stdout_stderr_file(self):
 """capture stdout and stderr to the same open file"""
 tf = tempfile.TemporaryFile()
 p = subprocess.Popen([sys.executable, "-c",
 'import sys;' \
 'sys.stdout.write("apple");' \
 'sys.stdout.flush();' \
 'sys.stderr.write("orange")'],
 stdout=tf,
 stderr=tf)
 p.wait()
 tf.seek(0)
 self.assertEqual(tf.read(), "appleorange")
 def test_cwd(self):
 """cwd"""
 tmpdir = os.getenv("TEMP", "/tmp")
 tmpdir = os.path.realpath(tmpdir)
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' \
 'sys.stdout.write(os.getcwd())'],
 stdout=subprocess.PIPE,
 cwd=tmpdir)
 self.assertEqual(p.stdout.read(), tmpdir)
 def test_env(self):
 """env"""
 newenv = os.environ.copy()
 newenv["FRUIT"] = "orange"
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' \
 'sys.stdout.write(os.getenv("FRUIT"))'],
 stdout=subprocess.PIPE,
 env=newenv)
 self.assertEqual(p.stdout.read(), "orange")
 def test_communicate(self):
 """communicate()"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' \
 'sys.stderr.write("pineapple");' \
 'sys.stdout.write(sys.stdin.read())'],
 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 (stdout, stderr) = p.communicate("banana")
 self.assertEqual(stdout, "banana")
 self.assertEqual(stderr, "pineapple")
 def test_communicate_returns(self):
 """communicate() should return None if no redirection is active"""
 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(47)"])
 (stdout, stderr) = p.communicate()
 self.assertEqual(stdout, None)
 self.assertEqual(stderr, None)
 def test_communicate_pipe_buf(self):
 """communicate() with writes larger than pipe_buf"""
 # This test will probably deadlock rather than fail, if
 # communicate() does not work properly. 
 x, y = os.pipe()
 if mswindows:
 pipe_buf = 512
 else:
 pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")
 os.close(x)
 os.close(y)
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' 
 'sys.stdout.write(sys.stdin.read(47));' \
 'sys.stderr.write("xyz"*%d);' \
 'sys.stdout.write(sys.stdin.read())' % pipe_buf],
 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 string_to_write = "abc"*pipe_buf
 (stdout, stderr) = p.communicate(string_to_write)
 self.assertEqual(stdout, string_to_write)
 def test_writes_before_communicate(self):
 """stdin.write before communicate()"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' \
 'sys.stdout.write(sys.stdin.read())'],
 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 p.stdin.write("banana")
 (stdout, stderr) = p.communicate("split")
 self.assertEqual(stdout, "bananasplit")
 self.assertEqual(stderr, "")

 def test_universal_newlines(self):
 """universal newlines"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' + SETBINARY + \
 'sys.stdout.write("line1\\n");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("line2\\r");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("line3\\r\\n");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("line4\\r");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("\\nline5");'
 'sys.stdout.flush();' \
 'sys.stdout.write("\\nline6");'],
 stdout=subprocess.PIPE,
 universal_newlines=1)
 stdout = p.stdout.read()
 if hasattr(open, 'newlines'):
 # Interpreter with universal newline support
 self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
 else:
 # Interpreter without universal newline support
 self.assertEqual(stdout, "line1\nline2\rline3\r\nline4\r\nline5\nline6")
 def test_universal_newlines_communicate(self):
 """universal newlines through communicate()"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' + SETBINARY + \
 'sys.stdout.write("line1\\n");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("line2\\r");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("line3\\r\\n");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("line4\\r");' \
 'sys.stdout.flush();' \
 'sys.stdout.write("\\nline5");'
 'sys.stdout.flush();' \
 'sys.stdout.write("\\nline6");'],
 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
 universal_newlines=1)
 (stdout, stderr) = p.communicate()
 if hasattr(open, 'newlines'):
 # Interpreter with universal newline support
 self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
 else:
 # Interpreter without universal newline support
 self.assertEqual(stdout, "line1\nline2\rline3\r\nline4\r\nline5\nline6")
 def test_no_leaking(self):
 """Make sure we leak no resources"""
 for i in range(1026):
 p = subprocess.Popen([sys.executable, "-c", "import sys;sys.stdout.write(sys.stdin.read())"],
 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 data = p.communicate("lime")[0]
 self.assertEqual(data, "lime")
 def test_list2cmdline(self):
 """list2cmdline"""
 self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),
 '"a b c" d e')
 self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),
 'ab\\"c \\ d')
 self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),
 'a\\\\\\b "de fg" h')
 self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),
 'a\\\\\\"b c d')
 self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),
 '"a\\\\b c" d e')
 self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),
 '"a\\\\b\\ c" d e')
 def test_poll(self):
 """poll"""
 p = subprocess.Popen([sys.executable,
 "-c", "import time; time.sleep(4)"])
 while p.poll() == None:
 sys.stdout.write(".")
 sys.stdout.flush()
 time.sleep(0.5)
 # Subsequent invocations should just return the returncode
 self.assertEqual(p.poll(), 0)
 def test_wait(self):
 """wait"""
 p = subprocess.Popen([sys.executable,
 "-c", "import time; time.sleep(2)"])
 self.assertEqual(p.wait(), 0)
 # Subsequent invocations should just return the returncode
 self.assertEqual(p.wait(), 0)

 #
 # POSIX tests
 #
 if not mswindows:
 def test_exceptions(self):
 """catched & re-raised exceptions"""
 try:
 p = subprocess.Popen([sys.executable, "-c", ""],
 cwd="/this/path/does/not/exist")
 except OSError, e:
 # The attribute child_traceback should contain "os.chdir"
 # somewhere.
 self.assertNotEqual(e.child_traceback.find("os.chdir"), -1)
 else:
 self.fail("Expected OSError")

 def test_run_abort(self):
 """returncode handles signal termination"""
 p = subprocess.Popen([sys.executable, "-c", "import os; os.abort()"])
 p.wait()
 self.assertEqual(-p.returncode, signal.SIGABRT)
 def test_preexec(self):
 """preexec function"""
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' \
 'sys.stdout.write(os.getenv("FRUIT"))'],
 stdout=subprocess.PIPE,
 preexec_fn=lambda: os.putenv("FRUIT", "apple"))
 self.assertEqual(p.stdout.read(), "apple")
 def test_close_fds(self):
 """close_fds"""
 # Make sure we have some fds open
 os.pipe()
 p = subprocess.Popen([sys.executable, "-c",
 'import sys,os;' \
 'sys.stdout.write(str(os.dup(0)))'],
 stdout=subprocess.PIPE, close_fds=1)
 # When all fds are closed, the next free fd should be 3. 
 self.assertEqual(p.stdout.read(), "3")
 def test_args_string(self):
 """args is a string"""
 f, fname = self.mkstemp()
 os.write(f, "#!/bin/sh\n")
 os.write(f, "exec %s -c 'import sys; sys.exit(47)'\n" % sys.executable)
 os.close(f)
 os.chmod(fname, 0700)
 p = subprocess.Popen(fname)
 p.wait()
 self.assertEqual(p.returncode, 47)
 os.remove(fname)
 def test_invalid_args(self):
 """invalid arguments should raise ValueError"""
 self.assertRaises(ValueError, subprocess.call,
 [sys.executable, "-c", "import sys; sys.exit(47)"],
 startupinfo=47)
 self.assertRaises(ValueError, subprocess.call,
 [sys.executable, "-c", "import sys; sys.exit(47)"],
 creationflags=47)
 def test_shell_sequence(self):
 """Run command through the shell (sequence)"""
 newenv = os.environ.copy()
 newenv["FRUIT"] = "apple"
 p = subprocess.Popen(["echo $FRUIT"], shell=1,
 stdout=subprocess.PIPE,
 env=newenv)
 self.assertEqual(p.stdout.read().strip(), "apple")
 def test_shell_string(self):
 """Run command through the shell (string)"""
 newenv = os.environ.copy()
 newenv["FRUIT"] = "apple"
 p = subprocess.Popen("echo $FRUIT", shell=1,
 stdout=subprocess.PIPE,
 env=newenv)
 self.assertEqual(p.stdout.read().strip(), "apple")
 def test_call_string(self):
 """call() function with string argument on UNIX"""
 f, fname = self.mkstemp()
 os.write(f, "#!/bin/sh\n")
 os.write(f, "exec %s -c 'import sys; sys.exit(47)'\n" % sys.executable)
 os.close(f)
 os.chmod(fname, 0700)
 rc = subprocess.call(fname)
 self.assertEqual(rc, 47)

 #
 # Windows tests
 #
 if mswindows:
 def test_startupinfo(self):
 """startupinfo argument"""
 # We uses hardcoded constants, because we do not want to
 # depend on win32all. 
 STARTF_USESHOWWINDOW = 1
 SW_MAXIMIZE = 3
 startupinfo = subprocess.STARTUPINFO()
 startupinfo.dwFlags = STARTF_USESHOWWINDOW
 startupinfo.wShowWindow = SW_MAXIMIZE
 # Since Python is a console process, it won't be affected
 # by wShowWindow, but the argument should be silently
 # ignored
 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
 startupinfo=startupinfo)
 def test_creationflags(self):
 """creationflags argument"""
 CREATE_NEW_CONSOLE = 16
 subprocess.call(sys.executable + ' -c "import time; time.sleep(2)"',
 creationflags=CREATE_NEW_CONSOLE)
 def test_invalid_args(self):
 """invalid arguments should raise ValueError"""
 self.assertRaises(ValueError, subprocess.call,
 [sys.executable, "-c", "import sys; sys.exit(47)"],
 preexec_fn=lambda: 1)
 self.assertRaises(ValueError, subprocess.call,
 [sys.executable, "-c", "import sys; sys.exit(47)"],
 close_fds=True)
 def test_shell_sequence(self):
 """Run command through the shell (sequence)"""
 newenv = os.environ.copy()
 newenv["FRUIT"] = "physalis"
 p = subprocess.Popen(["set"], shell=1,
 stdout=subprocess.PIPE, 
 env=newenv)
 self.assertNotEqual(p.stdout.read().find("physalis"), -1)
 def test_shell_string(self):
 """Run command through the shell (string)"""
 newenv = os.environ.copy()
 newenv["FRUIT"] = "physalis"
 p = subprocess.Popen("set", shell=1,
 stdout=subprocess.PIPE, 
 env=newenv)
 self.assertNotEqual(p.stdout.read().find("physalis"), -1)
 def test_call_string(self):
 """call() function with string argument on Windows"""
 rc = subprocess.call(sys.executable + ' -c "import sys; sys.exit(47)"')
 self.assertEqual(rc, 47)
def test_main():
 test_support.run_unittest(ProcessTestCase)
if __name__ == "__main__":
 test_main()