mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Recorded merge of revisions 78508-78509 via svnmerge from

Browse source 
svn+ssh://pythondev@svn.python.org/python/trunk

........
  r78508 | florent.xicluna | 2010-02-27 20:20:50 +0100 (sam, 27 fév 2010) | 2 lines

  Clean test_subprocess: use assertRaises, skipIf, skipUnless helpers and a custom helper assertStderrEqual.
........
  r78509 | florent.xicluna | 2010-02-27 22:15:27 +0100 (sam, 27 fév 2010) | 2 lines

  Fix an oversight in r78508: p.wait() should be compared to 0
........
This commit is contained in:
Florent Xicluna 2010-02-27 22:12:37 +00:00
parent 09259e2043
commit b1e94e8af5
1 changed files with 290 additions and 313 deletions
Download patch file Download diff file Expand all files Collapse all files

603
Lib/test/test_subprocess.py
View file

@ -21,37 +21,30 @@ if mswindows:
else: else:
SETBINARY = '' SETBINARY = ''
# In a debug build, stuff like "[6580 refs]" is printed to stderr at
# shutdown time. That frustrates tests trying to check stderr produced try:
# from a spawned Python process. mkstemp = tempfile.mkstemp
def remove_stderr_debug_decorations(stderr): except AttributeError:
return re.sub("\[\d+ refs\]\r?\n?$", "", stderr.decode()).encode() # tempfile.mkstemp is not available
#return re.sub(r"\[\d+ refs\]\r?\n?$", "", stderr) def mkstemp():
"""Replacement for mkstemp, calling mktemp."""
fname = tempfile.mktemp()
return os.open(fname, os.O_RDWR|os.O_CREAT), fname
class ProcessTestCase(unittest.TestCase): class ProcessTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
# Try to minimize the number of children we have so this test # Try to minimize the number of children we have so this test
# doesn't crash on some buildbots (Alphas in particular). # doesn't crash on some buildbots (Alphas in particular).
if hasattr(support, "reap_children"): support.reap_children()
support.reap_children()
def tearDown(self): def assertStderrEqual(self, stderr, expected, msg=None):
# Try to minimize the number of children we have so this test # In a debug build, stuff like "[6580 refs]" is printed to stderr at
# doesn't crash on some buildbots (Alphas in particular). # shutdown time. That frustrates tests trying to check stderr produced
if hasattr(support, "reap_children"): # from a spawned Python process.
support.reap_children() actual = re.sub("\[\d+ refs\]\r?\n?$", "", stderr.decode()).encode()
self.assertEqual(actual, expected, msg)
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): def test_call_seq(self):
# call() function with sequence argument # call() function with sequence argument
rc = subprocess.call([sys.executable, "-c", rc = subprocess.call([sys.executable, "-c",
@ -66,13 +59,10 @@ class ProcessTestCase(unittest.TestCase):
def test_check_call_nonzero(self): def test_check_call_nonzero(self):
# check_call() function with non-zero return code # check_call() function with non-zero return code
try: with self.assertRaises(subprocess.CalledProcessError) as c:
subprocess.check_call([sys.executable, "-c", subprocess.check_call([sys.executable, "-c",
"import sys; sys.exit(47)"]) "import sys; sys.exit(47)"])
except subprocess.CalledProcessError as e: self.assertEqual(c.exception.returncode, 47)
self.assertEqual(e.returncode, 47)
else:
self.fail("Expected CalledProcessError")
def test_check_output(self): def test_check_output(self):
# check_output() function with zero return code # check_output() function with zero return code
@ -82,13 +72,10 @@ class ProcessTestCase(unittest.TestCase):
def test_check_output_nonzero(self): def test_check_output_nonzero(self):
# check_call() function with non-zero return code # check_call() function with non-zero return code
try: with self.assertRaises(subprocess.CalledProcessError) as c:
subprocess.check_output( subprocess.check_output(
[sys.executable, "-c", "import sys; sys.exit(5)"]) [sys.executable, "-c", "import sys; sys.exit(5)"])
except subprocess.CalledProcessError as e: self.assertEqual(c.exception.returncode, 5)
self.assertEqual(e.returncode, 5)
else:
self.fail("Expected CalledProcessError")
def test_check_output_stderr(self): def test_check_output_stderr(self):
# check_output() function stderr redirected to stdout # check_output() function stderr redirected to stdout
@ -99,14 +86,12 @@ class ProcessTestCase(unittest.TestCase):
def test_check_output_stdout_arg(self): def test_check_output_stdout_arg(self):
# check_output() function stderr redirected to stdout # check_output() function stderr redirected to stdout
try: with self.assertRaises(ValueError) as c:
output = subprocess.check_output( output = subprocess.check_output(
[sys.executable, "-c", "print('will not be run')"], [sys.executable, "-c", "print('will not be run')"],
stdout=sys.stdout) stdout=sys.stdout)
except ValueError as e:
self.assertIn('stdout', e.args[0])
else:
self.fail("Expected ValueError when stdout arg supplied.") self.fail("Expected ValueError when stdout arg supplied.")
self.assertIn('stdout', c.exception.args[0])
def test_call_kwargs(self): def test_call_kwargs(self):
# call() function with keyword args # call() function with keyword args
@ -227,8 +212,7 @@ class ProcessTestCase(unittest.TestCase):
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
'import sys; sys.stderr.write("strawberry")'], 'import sys; sys.stderr.write("strawberry")'],
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
self.assertEqual(remove_stderr_debug_decorations(p.stderr.read()), self.assertStderrEqual(p.stderr.read(), b"strawberry")
b"strawberry")
def test_stderr_filedes(self): def test_stderr_filedes(self):
# stderr is set to open file descriptor # stderr is set to open file descriptor
@ -239,8 +223,7 @@ class ProcessTestCase(unittest.TestCase):
stderr=d) stderr=d)
p.wait() p.wait()
os.lseek(d, 0, 0) os.lseek(d, 0, 0)
self.assertEqual(remove_stderr_debug_decorations(os.read(d, 1024)), self.assertStderrEqual(os.read(d, 1024), b"strawberry")
b"strawberry")
def test_stderr_fileobj(self): def test_stderr_fileobj(self):
# stderr is set to open file object # stderr is set to open file object
@ -250,8 +233,7 @@ class ProcessTestCase(unittest.TestCase):
stderr=tf) stderr=tf)
p.wait() p.wait()
tf.seek(0) tf.seek(0)
self.assertEqual(remove_stderr_debug_decorations(tf.read()), self.assertStderrEqual(tf.read(), b"strawberry")
b"strawberry")
def test_stdout_stderr_pipe(self): def test_stdout_stderr_pipe(self):
# capture stdout and stderr to the same pipe # capture stdout and stderr to the same pipe
@ -262,9 +244,7 @@ class ProcessTestCase(unittest.TestCase):
'sys.stderr.write("orange")'], 'sys.stderr.write("orange")'],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT) stderr=subprocess.STDOUT)
output = p.stdout.read() self.assertStderrEqual(p.stdout.read(), b"appleorange")
stripped = remove_stderr_debug_decorations(output)
self.assertEqual(stripped, b"appleorange")
def test_stdout_stderr_file(self): def test_stdout_stderr_file(self):
# capture stdout and stderr to the same open file # capture stdout and stderr to the same open file
@ -278,15 +258,13 @@ class ProcessTestCase(unittest.TestCase):
stderr=tf) stderr=tf)
p.wait() p.wait()
tf.seek(0) tf.seek(0)
output = tf.read() self.assertStderrEqual(tf.read(), b"appleorange")
stripped = remove_stderr_debug_decorations(output)
self.assertEqual(stripped, b"appleorange")
def test_stdout_filedes_of_stdout(self): def test_stdout_filedes_of_stdout(self):
# stdout is set to 1 (#1531862). # stdout is set to 1 (#1531862).
cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), b'.\n'))" cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), b'.\n'))"
rc = subprocess.call([sys.executable, "-c", cmd], stdout=1) rc = subprocess.call([sys.executable, "-c", cmd], stdout=1)
self.assertEquals(rc, 2) self.assertEqual(rc, 2)
def test_cwd(self): def test_cwd(self):
tmpdir = tempfile.gettempdir() tmpdir = tempfile.gettempdir()
@ -337,9 +315,7 @@ class ProcessTestCase(unittest.TestCase):
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
(stdout, stderr) = p.communicate() (stdout, stderr) = p.communicate()
self.assertEqual(stdout, None) self.assertEqual(stdout, None)
# When running with a pydebug build, the # of references is outputted self.assertStderrEqual(stderr, b"pineapple")
# to stderr, so just check if stderr at least started with "pinapple"
self.assertEqual(remove_stderr_debug_decorations(stderr), b"pineapple")
def test_communicate(self): def test_communicate(self):
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
@ -351,24 +327,24 @@ class ProcessTestCase(unittest.TestCase):
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
(stdout, stderr) = p.communicate(b"banana") (stdout, stderr) = p.communicate(b"banana")
self.assertEqual(stdout, b"banana") self.assertEqual(stdout, b"banana")
self.assertEqual(remove_stderr_debug_decorations(stderr), self.assertStderrEqual(stderr, b"pineapple")
b"pineapple")
# This test is Linux specific for simplicity to at least have # This test is Linux specific for simplicity to at least have
# some coverage. It is not a platform specific bug. # some coverage. It is not a platform specific bug.
if os.path.isdir('/proc/%d/fd' % os.getpid()): @unittest.skipUnless(os.path.isdir('/proc/%d/fd' % os.getpid()),
# Test for the fd leak reported in http://bugs.python.org/issue2791. "Linux specific")
def test_communicate_pipe_fd_leak(self): # Test for the fd leak reported in http://bugs.python.org/issue2791.
fd_directory = '/proc/%d/fd' % os.getpid() def test_communicate_pipe_fd_leak(self):
num_fds_before_popen = len(os.listdir(fd_directory)) fd_directory = '/proc/%d/fd' % os.getpid()
p = subprocess.Popen([sys.executable, '-c', 'print()'], num_fds_before_popen = len(os.listdir(fd_directory))
stdout=subprocess.PIPE) p = subprocess.Popen([sys.executable, "-c", "print()"],
p.communicate() stdout=subprocess.PIPE)
num_fds_after_communicate = len(os.listdir(fd_directory)) p.communicate()
del p num_fds_after_communicate = len(os.listdir(fd_directory))
num_fds_after_destruction = len(os.listdir(fd_directory)) del p
self.assertEqual(num_fds_before_popen, num_fds_after_destruction) num_fds_after_destruction = len(os.listdir(fd_directory))
self.assertEqual(num_fds_before_popen, num_fds_after_communicate) self.assertEqual(num_fds_before_popen, num_fds_after_destruction)
self.assertEqual(num_fds_before_popen, num_fds_after_communicate)
def test_communicate_returns(self): def test_communicate_returns(self):
# communicate() should return None if no redirection is active # communicate() should return None if no redirection is active
@ -412,7 +388,7 @@ class ProcessTestCase(unittest.TestCase):
p.stdin.write(b"banana") p.stdin.write(b"banana")
(stdout, stderr) = p.communicate(b"split") (stdout, stderr) = p.communicate(b"split")
self.assertEqual(stdout, b"bananasplit") self.assertEqual(stdout, b"bananasplit")
self.assertEqual(remove_stderr_debug_decorations(stderr), b"") self.assertStderrEqual(stderr, b"")
def test_universal_newlines(self): def test_universal_newlines(self):
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
@ -503,7 +479,7 @@ class ProcessTestCase(unittest.TestCase):
# but, based on system scheduling we can't control, it's possible # but, based on system scheduling we can't control, it's possible
# poll() never returned None. It "should be" very rare that it # poll() never returned None. It "should be" very rare that it
# didn't go around at least twice. # didn't go around at least twice.
self.assertTrue(count >= 2) self.assertGreaterEqual(count, 2)
# Subsequent invocations should just return the returncode # Subsequent invocations should just return the returncode
self.assertEqual(p.poll(), 0) self.assertEqual(p.poll(), 0)
@ -519,12 +495,8 @@ class ProcessTestCase(unittest.TestCase):
def test_invalid_bufsize(self): def test_invalid_bufsize(self):
# an invalid type of the bufsize argument should raise # an invalid type of the bufsize argument should raise
# TypeError. # TypeError.
try: with self.assertRaises(TypeError):
subprocess.Popen([sys.executable, "-c", "pass"], "orange") subprocess.Popen([sys.executable, "-c", "pass"], "orange")
except TypeError:
pass
else:
self.fail("Expected TypeError")
def test_bufsize_is_none(self): def test_bufsize_is_none(self):
# bufsize=None should be the same as bufsize=0. # bufsize=None should be the same as bufsize=0.
@ -541,292 +513,297 @@ class ProcessTestCase(unittest.TestCase):
# value for that limit, but Windows has 2048, so we loop # value for that limit, but Windows has 2048, so we loop
# 1024 times (each call leaked two fds). # 1024 times (each call leaked two fds).
for i in range(1024): for i in range(1024):
try: # Windows raises IOError. Others raise OSError.
with self.assertRaises(EnvironmentError) as c:
subprocess.Popen(['nonexisting_i_hope'], subprocess.Popen(['nonexisting_i_hope'],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
# Windows raises IOError if c.exception.errno != 2: # ignore "no such file"
except (IOError, OSError) as err: raise c.exception
if err.errno != 2: # ignore "no such file"
raise
#
# POSIX tests
#
if not mswindows:
def test_exceptions(self):
# caught & 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): # context manager
"""Try to prevent core files from being created. class _SuppressCoreFiles(object):
Returns previous ulimit if successful, else None. """Try to prevent core files from being created."""
""" old_limit = 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): def __enter__(self):
"""Return core file behavior to default.""" """Try to save previous ulimit, then set it to (0, 0)."""
if old_limit is None: try:
return import resource
try: self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
import resource resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
resource.setrlimit(resource.RLIMIT_CORE, old_limit) except (ImportError, ValueError, resource.error):
except (ImportError, ValueError, resource.error): pass
return
def test_run_abort(self): def __exit__(self, *args):
# returncode handles signal termination """Return core file behavior to default."""
old_limit = self._suppress_core_files() if self.old_limit is None:
try: return
p = subprocess.Popen([sys.executable, try:
"-c", "import os; os.abort()"]) import resource
finally: resource.setrlimit(resource.RLIMIT_CORE, self.old_limit)
self._unsuppress_core_files(old_limit) except (ImportError, ValueError, resource.error):
p.wait() pass
self.assertEqual(-p.returncode, signal.SIGABRT)
def test_preexec(self):
# preexec function @unittest.skipIf(sys.platform == "win32", "POSIX specific tests")
class POSIXProcessTestCase(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).
support.reap_children()
def test_exceptions(self):
# caught & re-raised exceptions
with self.assertRaises(OSError) as c:
p = subprocess.Popen([sys.executable, "-c", ""],
cwd="/this/path/does/not/exist")
# The attribute child_traceback should contain "os.chdir" somewhere.
self.assertIn("os.chdir", c.exception.child_traceback)
def test_run_abort(self):
# returncode handles signal termination
with _SuppressCoreFiles():
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
'import sys,os;' 'import os; os.abort()'])
'sys.stdout.write(os.getenv("FRUIT"))'],
stdout=subprocess.PIPE,
preexec_fn=lambda: os.putenv("FRUIT",
"apple"))
self.assertEqual(p.stdout.read(), b"apple")
def test_args_string(self):
# args is a string
fd, fname = self.mkstemp()
# reopen in text mode
with open(fd, "w") as fobj:
fobj.write("#!/bin/sh\n")
fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
sys.executable)
os.chmod(fname, 0o700)
p = subprocess.Popen(fname)
p.wait() p.wait()
os.remove(fname) self.assertEqual(-p.returncode, signal.SIGABRT)
self.assertEqual(p.returncode, 47)
def test_invalid_args(self): def test_preexec(self):
# invalid arguments should raise ValueError # preexec function
self.assertRaises(ValueError, subprocess.call, p = subprocess.Popen([sys.executable, "-c",
[sys.executable, 'import sys,os;'
"-c", "import sys; sys.exit(47)"], 'sys.stdout.write(os.getenv("FRUIT"))'],
startupinfo=47) stdout=subprocess.PIPE,
self.assertRaises(ValueError, subprocess.call, preexec_fn=lambda: os.putenv("FRUIT", "apple"))
[sys.executable, self.assertEqual(p.stdout.read(), b"apple")
"-c", "import sys; sys.exit(47)"],
creationflags=47)
def test_shell_sequence(self): def test_args_string(self):
# Run command through the shell (sequence) # args is a string
newenv = os.environ.copy() fd, fname = mkstemp()
newenv["FRUIT"] = "apple" # reopen in text mode
p = subprocess.Popen(["echo $FRUIT"], shell=1, with open(fd, "w") as fobj:
stdout=subprocess.PIPE, fobj.write("#!/bin/sh\n")
env=newenv) fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple") sys.executable)
os.chmod(fname, 0o700)
p = subprocess.Popen(fname)
p.wait()
os.remove(fname)
self.assertEqual(p.returncode, 47)
def test_shell_string(self): def test_invalid_args(self):
# Run command through the shell (string) # invalid arguments should raise ValueError
newenv = os.environ.copy() self.assertRaises(ValueError, subprocess.call,
newenv["FRUIT"] = "apple" [sys.executable, "-c",
p = subprocess.Popen("echo $FRUIT", shell=1, "import sys; sys.exit(47)"],
stdout=subprocess.PIPE, startupinfo=47)
env=newenv) self.assertRaises(ValueError, subprocess.call,
self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple") [sys.executable, "-c",
"import sys; sys.exit(47)"],
creationflags=47)
def test_call_string(self): def test_shell_sequence(self):
# call() function with string argument on UNIX # Run command through the shell (sequence)
fd, fname = self.mkstemp() newenv = os.environ.copy()
# reopen in text mode newenv["FRUIT"] = "apple"
with open(fd, "w") as fobj: p = subprocess.Popen(["echo $FRUIT"], shell=1,
fobj.write("#!/bin/sh\n") stdout=subprocess.PIPE,
fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" % env=newenv)
sys.executable) self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
os.chmod(fname, 0o700)
rc = subprocess.call(fname)
os.remove(fname)
self.assertEqual(rc, 47)
def DISABLED_test_send_signal(self): def test_shell_string(self):
p = subprocess.Popen([sys.executable, # Run command through the shell (string)
"-c", "input()"]) 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(b" \t\r\n\f"), b"apple")
self.assertTrue(p.poll() is None, p.poll()) def test_call_string(self):
p.send_signal(signal.SIGINT) # call() function with string argument on UNIX
self.assertNotEqual(p.wait(), 0) fd, fname = mkstemp()
# reopen in text mode
with open(fd, "w") as fobj:
fobj.write("#!/bin/sh\n")
fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
sys.executable)
os.chmod(fname, 0o700)
rc = subprocess.call(fname)
os.remove(fname)
self.assertEqual(rc, 47)
def DISABLED_test_kill(self): @unittest.skip("See issue #2777")
p = subprocess.Popen([sys.executable, def test_send_signal(self):
"-c", "input()"]) p = subprocess.Popen([sys.executable, "-c", "input()"])
self.assertTrue(p.poll() is None, p.poll()) self.assertIs(p.poll(), None)
p.kill() p.send_signal(signal.SIGINT)
self.assertEqual(p.wait(), -signal.SIGKILL) self.assertIsNot(p.wait(), None)
def DISABLED_test_terminate(self): @unittest.skip("See issue #2777")
p = subprocess.Popen([sys.executable, def test_kill(self):
"-c", "input()"]) p = subprocess.Popen([sys.executable, "-c", "input()"])
self.assertTrue(p.poll() is None, p.poll()) self.assertIs(p.poll(), None)
p.terminate() p.kill()
self.assertEqual(p.wait(), -signal.SIGTERM) self.assertEqual(p.wait(), -signal.SIGKILL)
# @unittest.skip("See issue #2777")
# Windows tests def test_terminate(self):
# p = subprocess.Popen([sys.executable, "-c", "input()"])
if mswindows:
def test_startupinfo(self): self.assertIs(p.poll(), None)
# startupinfo argument p.terminate()
# We uses hardcoded constants, because we do not want to self.assertEqual(p.wait(), -signal.SIGTERM)
# depend on win32all.
STARTF_USESHOWWINDOW = 1
SW_MAXIMIZE = 3 @unittest.skipUnless(sys.platform == "win32", "Windows specific tests")
startupinfo = subprocess.STARTUPINFO() class Win32ProcessTestCase(unittest.TestCase):
startupinfo.dwFlags = STARTF_USESHOWWINDOW def setUp(self):
startupinfo.wShowWindow = SW_MAXIMIZE # Try to minimize the number of children we have so this test
# Since Python is a console process, it won't be affected # doesn't crash on some buildbots (Alphas in particular).
# by wShowWindow, but the argument should be silently support.reap_children()
# ignored
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], 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) startupinfo=startupinfo)
def test_creationflags(self): def test_creationflags(self):
# creationflags argument # creationflags argument
CREATE_NEW_CONSOLE = 16 CREATE_NEW_CONSOLE = 16
sys.stderr.write(" a DOS box should flash briefly ...\n") sys.stderr.write(" a DOS box should flash briefly ...\n")
subprocess.call(sys.executable + subprocess.call(sys.executable +
' -c "import time; time.sleep(0.25)"', ' -c "import time; time.sleep(0.25)"',
creationflags=CREATE_NEW_CONSOLE) creationflags=CREATE_NEW_CONSOLE)
def test_invalid_args(self): def test_invalid_args(self):
# invalid arguments should raise ValueError # invalid arguments should raise ValueError
self.assertRaises(ValueError, subprocess.call, self.assertRaises(ValueError, subprocess.call,
[sys.executable, [sys.executable, "-c",
"-c", "import sys; sys.exit(47)"], "import sys; sys.exit(47)"],
preexec_fn=lambda: 1) preexec_fn=lambda: 1)
self.assertRaises(ValueError, subprocess.call, self.assertRaises(ValueError, subprocess.call,
[sys.executable, [sys.executable, "-c",
"-c", "import sys; sys.exit(47)"], "import sys; sys.exit(47)"],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
close_fds=True)
def test_close_fds(self):
# close file descriptors
rc = subprocess.call([sys.executable, "-c",
"import sys; sys.exit(47)"],
close_fds=True) close_fds=True)
self.assertEqual(rc, 47)
def test_close_fds(self): def test_shell_sequence(self):
# close file descriptors # Run command through the shell (sequence)
rc = subprocess.call([sys.executable, "-c", newenv = os.environ.copy()
"import sys; sys.exit(47)"], newenv["FRUIT"] = "physalis"
close_fds=True) p = subprocess.Popen(["set"], shell=1,
self.assertEqual(rc, 47) stdout=subprocess.PIPE,
env=newenv)
self.assertIn(b"physalis", p.stdout.read())
def test_shell_sequence(self): def test_shell_string(self):
# Run command through the shell (sequence) # Run command through the shell (string)
newenv = os.environ.copy() newenv = os.environ.copy()
newenv["FRUIT"] = "physalis" newenv["FRUIT"] = "physalis"
p = subprocess.Popen(["set"], shell=1, p = subprocess.Popen("set", shell=1,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
env=newenv) env=newenv)
self.assertNotEqual(p.stdout.read().find(b"physalis"), -1) self.assertIn(b"physalis", p.stdout.read())
def test_shell_string(self): def test_call_string(self):
# Run command through the shell (string) # call() function with string argument on Windows
newenv = os.environ.copy() rc = subprocess.call(sys.executable +
newenv["FRUIT"] = "physalis" ' -c "import sys; sys.exit(47)"')
p = subprocess.Popen("set", shell=1, self.assertEqual(rc, 47)
stdout=subprocess.PIPE,
env=newenv)
self.assertNotEqual(p.stdout.read().find(b"physalis"), -1)
def test_call_string(self): @unittest.skip("See issue #2777")
# call() function with string argument on Windows def test_send_signal(self):
rc = subprocess.call(sys.executable + p = subprocess.Popen([sys.executable, "-c", "input()"])
' -c "import sys; sys.exit(47)"')
self.assertEqual(rc, 47)
def DISABLED_test_send_signal(self): self.assertIs(p.poll(), None)
p = subprocess.Popen([sys.executable, p.send_signal(signal.SIGTERM)
"-c", "input()"]) self.assertNotEqual(p.wait(), 0)
self.assertTrue(p.poll() is None, p.poll()) @unittest.skip("See issue #2777")
p.send_signal(signal.SIGTERM) def test_kill(self):
self.assertNotEqual(p.wait(), 0) p = subprocess.Popen([sys.executable, "-c", "input()"])
def DISABLED_test_kill(self): self.assertIs(p.poll(), None)
p = subprocess.Popen([sys.executable, p.kill()
"-c", "input()"]) self.assertNotEqual(p.wait(), 0)
self.assertTrue(p.poll() is None, p.poll()) @unittest.skip("See issue #2777")
p.kill() def test_terminate(self):
self.assertNotEqual(p.wait(), 0) p = subprocess.Popen([sys.executable, "-c", "input()"])
def DISABLED_test_terminate(self): self.assertIs(p.poll(), None)
p = subprocess.Popen([sys.executable, p.terminate()
"-c", "input()"]) self.assertNotEqual(p.wait(), 0)
self.assertTrue(p.poll() is None, p.poll())
p.terminate()
self.assertNotEqual(p.wait(), 0)
class CommandTests(unittest.TestCase):
# The module says: # The module says:
# "NB This only works (and is only relevant) for UNIX." # "NB This only works (and is only relevant) for UNIX."
# #
# Actually, getoutput should work on any platform with an os.popen, but # Actually, getoutput should work on any platform with an os.popen, but
# I'll take the comment as given, and skip this suite. # I'll take the comment as given, and skip this suite.
if os.name == 'posix': @unittest.skipUnless(os.name != 'posix', "only relevant for UNIX")
class CommandTests(unittest.TestCase):
def test_getoutput(self):
self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy')
self.assertEqual(subprocess.getstatusoutput('echo xyzzy'),
(0, 'xyzzy'))
def test_getoutput(self): # we use mkdtemp in the next line to create an empty directory
self.assertEquals(subprocess.getoutput('echo xyzzy'), 'xyzzy') # under our exclusive control; from that, we can invent a pathname
self.assertEquals(subprocess.getstatusoutput('echo xyzzy'), # that we _know_ won't exist. This is guaranteed to fail.
(0, 'xyzzy')) dir = None
try:
dir = tempfile.mkdtemp()
name = os.path.join(dir, "foo")
# we use mkdtemp in the next line to create an empty directory status, output = subprocess.getstatusoutput('cat ' + name)
# under our exclusive control; from that, we can invent a pathname self.assertNotEqual(status, 0)
# that we _know_ won't exist. This is guaranteed to fail. finally:
dir = None if dir is not None:
try: os.rmdir(dir)
dir = tempfile.mkdtemp()
name = os.path.join(dir, "foo")
status, output = subprocess.getstatusoutput('cat ' + name)
self.assertNotEquals(status, 0)
finally:
if dir is not None:
os.rmdir(dir)
unit_tests = [ProcessTestCase, CommandTests] @unittest.skipUnless(getattr(subprocess, '_has_poll', False),
"poll system call not supported")
class ProcessTestCaseNoPoll(ProcessTestCase):
def setUp(self):
subprocess._has_poll = False
ProcessTestCase.setUp(self)
if getattr(subprocess, '_has_poll', False): def tearDown(self):
class ProcessTestCaseNoPoll(ProcessTestCase): subprocess._has_poll = True
def setUp(self): ProcessTestCase.tearDown(self)
subprocess._has_poll = False
ProcessTestCase.setUp(self)
def tearDown(self):
subprocess._has_poll = True
ProcessTestCase.tearDown(self)
unit_tests.append(ProcessTestCaseNoPoll)
def test_main(): def test_main():
unit_tests = (ProcessTestCase,
POSIXProcessTestCase,
Win32ProcessTestCase,
CommandTests,
ProcessTestCaseNoPoll)
support.run_unittest(*unit_tests) support.run_unittest(*unit_tests)
support.reap_children() support.reap_children()