mirror of
https://github.com/python/cpython
synced 2024-10-08 21:20:05 +00:00
fc7bb8c786
svn+ssh://pythondev@svn.python.org/python/trunk ........ r53304 | vinay.sajip | 2007-01-09 15:50:28 +0100 (Tue, 09 Jan 2007) | 1 line Bug #1627575: Added _open() method to FileHandler which can be used to reopen files. The FileHandler instance now saves the encoding (which can be None) in an attribute called "encoding". ........ r53305 | vinay.sajip | 2007-01-09 15:51:36 +0100 (Tue, 09 Jan 2007) | 1 line Added entry about addition of _open() method to logging.FileHandler. ........ r53306 | vinay.sajip | 2007-01-09 15:54:56 +0100 (Tue, 09 Jan 2007) | 1 line Added a docstring ........ r53316 | thomas.heller | 2007-01-09 20:19:33 +0100 (Tue, 09 Jan 2007) | 4 lines Verify the sizes of the basic ctypes data types against the struct module. Will backport to release25-maint. ........ r53340 | gustavo.niemeyer | 2007-01-10 17:13:40 +0100 (Wed, 10 Jan 2007) | 3 lines Mention in the int() docstring that a base zero has meaning, as stated in http://docs.python.org/lib/built-in-funcs.html as well. ........ r53341 | gustavo.niemeyer | 2007-01-10 17:15:48 +0100 (Wed, 10 Jan 2007) | 2 lines Minor change in int() docstring for proper spacing. ........ r53358 | thomas.heller | 2007-01-10 21:12:13 +0100 (Wed, 10 Jan 2007) | 1 line Change the ctypes version number to "1.1.0". ........ r53361 | thomas.heller | 2007-01-10 21:51:19 +0100 (Wed, 10 Jan 2007) | 1 line Must change the version number in the _ctypes extension as well. ........ r53362 | guido.van.rossum | 2007-01-11 00:12:56 +0100 (Thu, 11 Jan 2007) | 3 lines Fix the signature of log_error(). (A subclass that did the right thing was getting complaints from pychecker.) ........ r53370 | matthias.klose | 2007-01-11 11:26:31 +0100 (Thu, 11 Jan 2007) | 2 lines - Make the documentation match the code and the docstring ........ r53375 | matthias.klose | 2007-01-11 12:44:04 +0100 (Thu, 11 Jan 2007) | 2 lines - idle: Honor the "Cancel" action in the save dialog (Debian bug #299092). ........ r53381 | raymond.hettinger | 2007-01-11 19:22:55 +0100 (Thu, 11 Jan 2007) | 1 line SF #1486663 -- Allow keyword args in subclasses of set() and frozenset(). ........ r53388 | thomas.heller | 2007-01-11 22:18:56 +0100 (Thu, 11 Jan 2007) | 4 lines Fixes for 64-bit Windows: In ctypes.wintypes, correct the definitions of HANDLE, WPARAM, LPARAM data types. Make parameterless foreign function calls work. ........ r53390 | thomas.heller | 2007-01-11 22:23:12 +0100 (Thu, 11 Jan 2007) | 2 lines Correct the comments: the code is right. ........ r53393 | brett.cannon | 2007-01-12 08:27:52 +0100 (Fri, 12 Jan 2007) | 3 lines Fix error where the end of a funcdesc environment was accidentally moved too far down. ........ r53397 | anthony.baxter | 2007-01-12 10:35:56 +0100 (Fri, 12 Jan 2007) | 3 lines add parsetok.h as a dependency - previously, changing this file doesn't cause the right files to be rebuilt. ........ r53401 | thomas.heller | 2007-01-12 21:08:19 +0100 (Fri, 12 Jan 2007) | 3 lines Avoid warnings in the test suite because ctypes.wintypes cannot be imported on non-windows systems. ........ r53402 | thomas.heller | 2007-01-12 21:17:34 +0100 (Fri, 12 Jan 2007) | 6 lines patch #1610795: BSD version of ctypes.util.find_library, by Martin Kammerhofer. release25-maint backport candidate, but the release manager has to decide. ........ r53403 | thomas.heller | 2007-01-12 21:21:53 +0100 (Fri, 12 Jan 2007) | 3 lines patch #1610795: BSD version of ctypes.util.find_library, by Martin Kammerhofer. ........ r53406 | brett.cannon | 2007-01-13 01:29:49 +0100 (Sat, 13 Jan 2007) | 2 lines Deprecate the sets module. ........ r53407 | georg.brandl | 2007-01-13 13:31:51 +0100 (Sat, 13 Jan 2007) | 3 lines Fix typo. ........ r53409 | marc-andre.lemburg | 2007-01-13 22:00:08 +0100 (Sat, 13 Jan 2007) | 16 lines Bump version number and change copyright year. Add new API linux_distribution() which supports reading the full distribution name and also knows how to parse LSB-style release files. Redirect the old dist() API to the new API (using the short distribution name taken from the release file filename). Add branch and revision to _sys_version(). Add work-around for Cygwin to libc_ver(). Add support for IronPython (thanks for Anthony Baxter) and make Jython support more robust. ........ r53410 | neal.norwitz | 2007-01-13 22:22:37 +0100 (Sat, 13 Jan 2007) | 1 line Fix grammar in docstrings ........ r53411 | marc-andre.lemburg | 2007-01-13 23:32:21 +0100 (Sat, 13 Jan 2007) | 9 lines Add parameter sys_version to _sys_version(). Change the cache for _sys_version() to take the parameter into account. Add support for parsing the IronPython 1.0.1 sys.version value - even though it still returns '1.0.0'; the version string no longer includes the patch level. ........ r53412 | peter.astrand | 2007-01-13 23:35:35 +0100 (Sat, 13 Jan 2007) | 1 line Fix for bug #1634343: allow specifying empty arguments on Windows ........ r53414 | marc-andre.lemburg | 2007-01-13 23:59:36 +0100 (Sat, 13 Jan 2007) | 14 lines Add Python implementation to the machine details. Pretty-print the Python version used for running PyBench. Let the user know when calibration has finished. [ 1563844 ] pybench support for IronPython: Simplify Unicode version detection. Make garbage collection and check interval settings optional if the Python implementation doesn't support thess (e.g. IronPython). ........ r53415 | marc-andre.lemburg | 2007-01-14 00:13:54 +0100 (Sun, 14 Jan 2007) | 5 lines Use defaults if sys.executable isn't set (e.g. on Jython). This change allows running PyBench under Jython. ........ r53416 | marc-andre.lemburg | 2007-01-14 00:15:33 +0100 (Sun, 14 Jan 2007) | 3 lines Jython doesn't have sys.setcheckinterval() - ignore it in that case. ........ r53420 | gerhard.haering | 2007-01-14 02:43:50 +0100 (Sun, 14 Jan 2007) | 29 lines Merged changes from standalone version 2.3.3. This should probably all be merged into the 2.5 maintenance branch: - self->statement was not checked while fetching data, which could lead to crashes if you used the pysqlite API in unusual ways. Closing the cursor and continuing to fetch data was enough. - Converters are stored in a converters dictionary. The converter name is uppercased first. The old upper-casing algorithm was wrong and was replaced by a simple call to the Python string's upper() method instead. -Applied patch by Glyph Lefkowitz that fixes the problem with subsequent SQLITE_SCHEMA errors. - Improvement to the row type: rows can now be iterated over and have a keys() method. This improves compatibility with both tuple and dict a lot. - A bugfix for the subsecond resolution in timestamps. - Corrected the way the flags PARSE_DECLTYPES and PARSE_COLNAMES are checked for. Now they work as documented. - gcc on Linux sucks. It exports all symbols by default in shared libraries, so if symbols are not unique it can lead to problems with symbol lookup. pysqlite used to crash under Apache when mod_cache was enabled because both modules had the symbol cache_init. I fixed this by applying the prefix pysqlite_ almost everywhere. Sigh. ........ r53423 | guido.van.rossum | 2007-01-14 04:46:33 +0100 (Sun, 14 Jan 2007) | 2 lines Remove a dependency of this test on $COLUMNS. ........ r53425 | ka-ping.yee | 2007-01-14 05:25:15 +0100 (Sun, 14 Jan 2007) | 3 lines Handle old-style instances more gracefully (display documentation on the relevant class instead of documentation on <type 'instance'>). ........ r53440 | vinay.sajip | 2007-01-14 22:49:59 +0100 (Sun, 14 Jan 2007) | 1 line Added WatchedFileHandler (based on SF patch #1598415) ........ r53441 | vinay.sajip | 2007-01-14 22:50:50 +0100 (Sun, 14 Jan 2007) | 1 line Added documentation for WatchedFileHandler (based on SF patch #1598415) ........ r53442 | guido.van.rossum | 2007-01-15 01:02:35 +0100 (Mon, 15 Jan 2007) | 2 lines Doc patch matching r53434 (htonl etc. now always take/return positive ints). ........
654 lines
26 KiB
Python
654 lines
26 KiB
Python
import unittest
|
|
from test import test_support
|
|
import subprocess
|
|
import sys
|
|
import signal
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import re
|
|
|
|
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 = ''
|
|
|
|
# In a debug build, stuff like "[6580 refs]" is printed to stderr at
|
|
# shutdown time. That frustrates tests trying to check stderr produced
|
|
# from a spawned Python process.
|
|
def remove_stderr_debug_decorations(stderr):
|
|
return re.sub(r"\[\d+ refs\]\r?\n?$", "", stderr)
|
|
|
|
class ProcessTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
# Try to minimize the number of children we have so this test
|
|
# doesn't crash on some buildbots (Alphas in particular).
|
|
if hasattr(test_support, "reap_children"):
|
|
test_support.reap_children()
|
|
|
|
def tearDown(self):
|
|
# Try to minimize the number of children we have so this test
|
|
# doesn't crash on some buildbots (Alphas in particular).
|
|
if hasattr(test_support, "reap_children"):
|
|
test_support.reap_children()
|
|
|
|
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_check_call_zero(self):
|
|
# check_call() function with zero return code
|
|
rc = subprocess.check_call([sys.executable, "-c",
|
|
"import sys; sys.exit(0)"])
|
|
self.assertEqual(rc, 0)
|
|
|
|
def test_check_call_nonzero(self):
|
|
# check_call() function with non-zero return code
|
|
try:
|
|
subprocess.check_call([sys.executable, "-c",
|
|
"import sys; sys.exit(47)"])
|
|
except subprocess.CalledProcessError as e:
|
|
self.assertEqual(e.returncode, 47)
|
|
else:
|
|
self.fail("Expected CalledProcessError")
|
|
|
|
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 " this bit of output is from a '
|
|
'test of stdout in a different '
|
|
'process ..."'],
|
|
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):
|
|
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(remove_stderr_debug_decorations(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(remove_stderr_debug_decorations(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(remove_stderr_debug_decorations(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)
|
|
output = p.stdout.read()
|
|
stripped = remove_stderr_debug_decorations(output)
|
|
self.assertEqual(stripped, "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)
|
|
output = tf.read()
|
|
stripped = remove_stderr_debug_decorations(output)
|
|
self.assertEqual(stripped, "appleorange")
|
|
|
|
def test_stdout_filedes_of_stdout(self):
|
|
# stdout is set to 1 (#1531862).
|
|
cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), '.\n'))"
|
|
rc = subprocess.call([sys.executable, "-c", cmd], stdout=1)
|
|
self.assertEquals(rc, 2)
|
|
|
|
def test_cwd(self):
|
|
tmpdir = os.getenv("TEMP", "/tmp")
|
|
# We cannot use os.path.realpath to canonicalize the path,
|
|
# since it doesn't expand Tru64 {memb} strings. See bug 1063571.
|
|
cwd = os.getcwd()
|
|
os.chdir(tmpdir)
|
|
tmpdir = os.getcwd()
|
|
os.chdir(cwd)
|
|
p = subprocess.Popen([sys.executable, "-c",
|
|
'import sys,os;' \
|
|
'sys.stdout.write(os.getcwd())'],
|
|
stdout=subprocess.PIPE,
|
|
cwd=tmpdir)
|
|
normcase = os.path.normcase
|
|
self.assertEqual(normcase(p.stdout.read()), normcase(tmpdir))
|
|
|
|
def test_env(self):
|
|
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_stdin(self):
|
|
p = subprocess.Popen([sys.executable, "-c",
|
|
'import sys; sys.exit(sys.stdin.read() == "pear")'],
|
|
stdin=subprocess.PIPE)
|
|
p.communicate("pear")
|
|
self.assertEqual(p.returncode, 1)
|
|
|
|
def test_communicate_stdout(self):
|
|
p = subprocess.Popen([sys.executable, "-c",
|
|
'import sys; sys.stdout.write("pineapple")'],
|
|
stdout=subprocess.PIPE)
|
|
(stdout, stderr) = p.communicate()
|
|
self.assertEqual(stdout, "pineapple")
|
|
self.assertEqual(stderr, None)
|
|
|
|
def test_communicate_stderr(self):
|
|
p = subprocess.Popen([sys.executable, "-c",
|
|
'import sys; sys.stderr.write("pineapple")'],
|
|
stderr=subprocess.PIPE)
|
|
(stdout, stderr) = p.communicate()
|
|
self.assertEqual(stdout, None)
|
|
# When running with a pydebug build, the # of references is outputted
|
|
# to stderr, so just check if stderr at least started with "pinapple"
|
|
self.assert_(stderr.startswith("pineapple"))
|
|
|
|
def test_communicate(self):
|
|
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(remove_stderr_debug_decorations(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(remove_stderr_debug_decorations(stderr), "")
|
|
|
|
def test_universal_newlines(self):
|
|
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(file, '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(file, '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
|
|
if not hasattr(test_support, "is_resource_enabled") \
|
|
or test_support.is_resource_enabled("subprocess") and not mswindows:
|
|
max_handles = 1026 # too much for most UNIX systems
|
|
else:
|
|
max_handles = 65
|
|
for i in range(max_handles):
|
|
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):
|
|
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')
|
|
self.assertEqual(subprocess.list2cmdline(['ab', '']),
|
|
'ab ""')
|
|
|
|
|
|
def test_poll(self):
|
|
p = subprocess.Popen([sys.executable,
|
|
"-c", "import time; time.sleep(1)"])
|
|
count = 0
|
|
while p.poll() is None:
|
|
time.sleep(0.1)
|
|
count += 1
|
|
# We expect that the poll loop probably went around about 10 times,
|
|
# but, based on system scheduling we can't control, it's possible
|
|
# poll() never returned None. It "should be" very rare that it
|
|
# didn't go around at least twice.
|
|
self.assert_(count >= 2)
|
|
# Subsequent invocations should just return the returncode
|
|
self.assertEqual(p.poll(), 0)
|
|
|
|
|
|
def test_wait(self):
|
|
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)
|
|
|
|
|
|
def test_invalid_bufsize(self):
|
|
# an invalid type of the bufsize argument should raise
|
|
# TypeError.
|
|
try:
|
|
subprocess.Popen([sys.executable, "-c", "pass"], "orange")
|
|
except TypeError:
|
|
pass
|
|
else:
|
|
self.fail("Expected TypeError")
|
|
|
|
#
|
|
# 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 as 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 _suppress_core_files(self):
|
|
"""Try to prevent core files from being created.
|
|
Returns previous ulimit if successful, else None.
|
|
"""
|
|
try:
|
|
import resource
|
|
old_limit = resource.getrlimit(resource.RLIMIT_CORE)
|
|
resource.setrlimit(resource.RLIMIT_CORE, (0,0))
|
|
return old_limit
|
|
except (ImportError, ValueError, resource.error):
|
|
return None
|
|
|
|
def _unsuppress_core_files(self, old_limit):
|
|
"""Return core file behavior to default."""
|
|
if old_limit is None:
|
|
return
|
|
try:
|
|
import resource
|
|
resource.setrlimit(resource.RLIMIT_CORE, old_limit)
|
|
except (ImportError, ValueError, resource.error):
|
|
return
|
|
|
|
def test_run_abort(self):
|
|
# returncode handles signal termination
|
|
old_limit = self._suppress_core_files()
|
|
try:
|
|
p = subprocess.Popen([sys.executable,
|
|
"-c", "import os; os.abort()"])
|
|
finally:
|
|
self._unsuppress_core_files(old_limit)
|
|
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_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()
|
|
os.remove(fname)
|
|
self.assertEqual(p.returncode, 47)
|
|
|
|
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)
|
|
os.remove(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
|
|
sys.stderr.write(" a DOS box should flash briefly ...\n")
|
|
subprocess.call(sys.executable +
|
|
' -c "import time; time.sleep(0.25)"',
|
|
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 hasattr(test_support, "reap_children"):
|
|
test_support.reap_children()
|
|
|
|
if __name__ == "__main__":
|
|
test_main()
|