blob: ea0e3bb4266fcff57cfc8817038962b277dd00fe [file] [log] [blame]
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001import unittest
Chris Jerdonekec3ea942012-09-30 00:10:28 -07002from test import script_helper
Benjamin Petersonee8712c2008-05-20 21:35:26 +00003from test import support
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00004import subprocess
5import sys
6import signal
Gregory P. Smithe14e9c22011-03-15 14:55:17 -04007import io
Andrew Svetlov82860712012-08-19 22:13:41 +03008import locale
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00009import os
Gregory P. Smitha59c59f2010-03-01 00:17:40 +000010import errno
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000011import tempfile
12import time
Tim Peters3761e8d2004-10-13 04:07:12 +000013import re
Ezio Melotti184bdfb2010-02-18 09:37:05 +000014import sysconfig
Gregory P. Smithd23047b2010-12-04 09:10:44 +000015import warnings
Gregory P. Smith51ee2702010-12-13 07:59:39 +000016import select
Gregory P. Smith81ce6852011-03-15 02:04:11 -040017import shutil
Benjamin Petersonb870aa12011-12-10 12:44:25 -050018import gc
Benjamin Peterson964561b2011-12-10 12:31:42 -050019
20try:
21 import resource
22except ImportError:
23 resource = None
24
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000025mswindows = (sys.platform == "win32")
26
27#
28# Depends on the following external programs: Python
29#
30
31if mswindows:
Tim Peters3b01a702004-10-12 22:19:32 +000032 SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), '
33 'os.O_BINARY);')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000034else:
35 SETBINARY = ''
36
Florent Xiclunab1e94e82010-02-27 22:12:37 +000037
38try:
39 mkstemp = tempfile.mkstemp
40except AttributeError:
41 # tempfile.mkstemp is not available
42 def mkstemp():
43 """Replacement for mkstemp, calling mktemp."""
44 fname = tempfile.mktemp()
45 return os.open(fname, os.O_RDWR|os.O_CREAT), fname
46
Tim Peters3761e8d2004-10-13 04:07:12 +000047
Florent Xiclunac049d872010-03-27 22:47:23 +000048class BaseTestCase(unittest.TestCase):
Thomas Wouters0e3f5912006-08-11 14:57:12 +000049 def setUp(self):
50 # Try to minimize the number of children we have so this test
51 # doesn't crash on some buildbots (Alphas in particular).
Florent Xiclunab1e94e82010-02-27 22:12:37 +000052 support.reap_children()
Thomas Wouters0e3f5912006-08-11 14:57:12 +000053
Florent Xiclunaf0cbd822010-03-04 21:50:56 +000054 def tearDown(self):
55 for inst in subprocess._active:
56 inst.wait()
57 subprocess._cleanup()
58 self.assertFalse(subprocess._active, "subprocess._active not empty")
59
Florent Xiclunab1e94e82010-02-27 22:12:37 +000060 def assertStderrEqual(self, stderr, expected, msg=None):
61 # In a debug build, stuff like "[6580 refs]" is printed to stderr at
62 # shutdown time. That frustrates tests trying to check stderr produced
63 # from a spawned Python process.
Antoine Pitrou62f68ed2010-08-04 11:48:56 +000064 actual = support.strip_python_stderr(stderr)
Florent Xiclunab1e94e82010-02-27 22:12:37 +000065 self.assertEqual(actual, expected, msg)
Thomas Wouters0e3f5912006-08-11 14:57:12 +000066
Florent Xiclunac049d872010-03-27 22:47:23 +000067
68class ProcessTestCase(BaseTestCase):
69
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000070 def test_call_seq(self):
Tim Peters7b759da2004-10-12 22:29:54 +000071 # call() function with sequence argument
Tim Peters3b01a702004-10-12 22:19:32 +000072 rc = subprocess.call([sys.executable, "-c",
73 "import sys; sys.exit(47)"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000074 self.assertEqual(rc, 47)
75
Peter Astrand454f7672005-01-01 09:36:35 +000076 def test_check_call_zero(self):
77 # check_call() function with zero return code
78 rc = subprocess.check_call([sys.executable, "-c",
79 "import sys; sys.exit(0)"])
80 self.assertEqual(rc, 0)
81
82 def test_check_call_nonzero(self):
83 # check_call() function with non-zero return code
Florent Xiclunab1e94e82010-02-27 22:12:37 +000084 with self.assertRaises(subprocess.CalledProcessError) as c:
Peter Astrand454f7672005-01-01 09:36:35 +000085 subprocess.check_call([sys.executable, "-c",
86 "import sys; sys.exit(47)"])
Florent Xiclunab1e94e82010-02-27 22:12:37 +000087 self.assertEqual(c.exception.returncode, 47)
Peter Astrand454f7672005-01-01 09:36:35 +000088
Georg Brandlf9734072008-12-07 15:30:06 +000089 def test_check_output(self):
90 # check_output() function with zero return code
91 output = subprocess.check_output(
92 [sys.executable, "-c", "print('BDFL')"])
Benjamin Peterson577473f2010-01-19 00:09:57 +000093 self.assertIn(b'BDFL', output)
Georg Brandlf9734072008-12-07 15:30:06 +000094
95 def test_check_output_nonzero(self):
96 # check_call() function with non-zero return code
Florent Xiclunab1e94e82010-02-27 22:12:37 +000097 with self.assertRaises(subprocess.CalledProcessError) as c:
Georg Brandlf9734072008-12-07 15:30:06 +000098 subprocess.check_output(
99 [sys.executable, "-c", "import sys; sys.exit(5)"])
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000100 self.assertEqual(c.exception.returncode, 5)
Georg Brandlf9734072008-12-07 15:30:06 +0000101
102 def test_check_output_stderr(self):
103 # check_output() function stderr redirected to stdout
104 output = subprocess.check_output(
105 [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"],
106 stderr=subprocess.STDOUT)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000107 self.assertIn(b'BDFL', output)
Georg Brandlf9734072008-12-07 15:30:06 +0000108
109 def test_check_output_stdout_arg(self):
110 # check_output() function stderr redirected to stdout
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000111 with self.assertRaises(ValueError) as c:
Georg Brandlf9734072008-12-07 15:30:06 +0000112 output = subprocess.check_output(
113 [sys.executable, "-c", "print('will not be run')"],
114 stdout=sys.stdout)
Georg Brandlf9734072008-12-07 15:30:06 +0000115 self.fail("Expected ValueError when stdout arg supplied.")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000116 self.assertIn('stdout', c.exception.args[0])
Georg Brandlf9734072008-12-07 15:30:06 +0000117
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000118 def test_call_kwargs(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000119 # call() function with keyword args
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000120 newenv = os.environ.copy()
121 newenv["FRUIT"] = "banana"
122 rc = subprocess.call([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000123 'import sys, os;'
124 'sys.exit(os.getenv("FRUIT")=="banana")'],
125 env=newenv)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000126 self.assertEqual(rc, 1)
127
Victor Stinner87b9bc32011-06-01 00:57:47 +0200128 def test_invalid_args(self):
129 # Popen() called with invalid arguments should raise TypeError
130 # but Popen.__del__ should not complain (issue #12085)
131 with support.captured_stderr() as s:
132 self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1)
133 argcount = subprocess.Popen.__init__.__code__.co_argcount
134 too_many_args = [0] * (argcount + 1)
135 self.assertRaises(TypeError, subprocess.Popen, *too_many_args)
136 self.assertEqual(s.getvalue(), '')
137
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000138 def test_stdin_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000139 # .stdin is None when not redirected
Georg Brandl88fc6642007-02-09 21:28:07 +0000140 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000141 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000142 self.addCleanup(p.stdout.close)
143 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000144 p.wait()
145 self.assertEqual(p.stdin, None)
146
147 def test_stdout_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000148 # .stdout is None when not redirected
Tim Peters29b6b4f2004-10-13 03:43:40 +0000149 p = subprocess.Popen([sys.executable, "-c",
Georg Brandl88fc6642007-02-09 21:28:07 +0000150 'print(" this bit of output is from a '
Tim Peters4052fe52004-10-13 03:29:54 +0000151 'test of stdout in a different '
Georg Brandl88fc6642007-02-09 21:28:07 +0000152 'process ...")'],
Tim Peters4052fe52004-10-13 03:29:54 +0000153 stdin=subprocess.PIPE, stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000154 self.addCleanup(p.stdin.close)
155 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000156 p.wait()
157 self.assertEqual(p.stdout, None)
158
159 def test_stderr_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000160 # .stderr is None when not redirected
Georg Brandl88fc6642007-02-09 21:28:07 +0000161 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000162 stdin=subprocess.PIPE, stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000163 self.addCleanup(p.stdout.close)
164 self.addCleanup(p.stdin.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000165 p.wait()
166 self.assertEqual(p.stderr, None)
167
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700168 # For use in the test_cwd* tests below.
169 def _normalize_cwd(self, cwd):
170 # Normalize an expected cwd (for Tru64 support).
171 # We can't use os.path.realpath since it doesn't expand Tru64 {memb}
172 # strings. See bug #1063571.
173 original_cwd = os.getcwd()
174 os.chdir(cwd)
175 cwd = os.getcwd()
176 os.chdir(original_cwd)
177 return cwd
178
179 # For use in the test_cwd* tests below.
180 def _split_python_path(self):
181 # Return normalized (python_dir, python_base).
182 python_path = os.path.realpath(sys.executable)
183 return os.path.split(python_path)
184
185 # For use in the test_cwd* tests below.
186 def _assert_cwd(self, expected_cwd, python_arg, **kwargs):
187 # Invoke Python via Popen, and assert that (1) the call succeeds,
188 # and that (2) the current working directory of the child process
189 # matches *expected_cwd*.
190 p = subprocess.Popen([python_arg, "-c",
191 "import os, sys; "
192 "sys.stdout.write(os.getcwd()); "
193 "sys.exit(47)"],
194 stdout=subprocess.PIPE,
195 **kwargs)
196 self.addCleanup(p.stdout.close)
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000197 p.wait()
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700198 self.assertEqual(47, p.returncode)
199 normcase = os.path.normcase
200 self.assertEqual(normcase(expected_cwd),
201 normcase(p.stdout.read().decode("utf-8")))
202
203 def test_cwd(self):
204 # Check that cwd changes the cwd for the child process.
205 temp_dir = tempfile.gettempdir()
206 temp_dir = self._normalize_cwd(temp_dir)
207 self._assert_cwd(temp_dir, sys.executable, cwd=temp_dir)
208
209 def test_cwd_with_relative_arg(self):
210 # Check that Popen looks for args[0] relative to cwd if args[0]
211 # is relative.
212 python_dir, python_base = self._split_python_path()
213 rel_python = os.path.join(os.curdir, python_base)
214 with support.temp_cwd() as wrong_dir:
215 # Before calling with the correct cwd, confirm that the call fails
216 # without cwd and with the wrong cwd.
217 self.assertRaises(OSError, subprocess.Popen,
218 [rel_python])
219 self.assertRaises(OSError, subprocess.Popen,
220 [rel_python], cwd=wrong_dir)
221 python_dir = self._normalize_cwd(python_dir)
222 self._assert_cwd(python_dir, rel_python, cwd=python_dir)
223
224 def test_cwd_with_relative_executable(self):
225 # Check that Popen looks for executable relative to cwd if executable
226 # is relative (and that executable takes precedence over args[0]).
227 python_dir, python_base = self._split_python_path()
228 rel_python = os.path.join(os.curdir, python_base)
229 doesntexist = "somethingyoudonthave"
230 with support.temp_cwd() as wrong_dir:
231 # Before calling with the correct cwd, confirm that the call fails
232 # without cwd and with the wrong cwd.
233 self.assertRaises(OSError, subprocess.Popen,
234 [doesntexist], executable=rel_python)
235 self.assertRaises(OSError, subprocess.Popen,
236 [doesntexist], executable=rel_python,
237 cwd=wrong_dir)
238 python_dir = self._normalize_cwd(python_dir)
239 self._assert_cwd(python_dir, doesntexist, executable=rel_python,
240 cwd=python_dir)
241
242 def test_cwd_with_absolute_arg(self):
243 # Check that Popen can find the executable when the cwd is wrong
244 # if args[0] is an absolute path.
245 python_dir, python_base = self._split_python_path()
246 abs_python = os.path.join(python_dir, python_base)
247 rel_python = os.path.join(os.curdir, python_base)
248 with script_helper.temp_dir() as wrong_dir:
249 # Before calling with an absolute path, confirm that using a
250 # relative path fails.
251 self.assertRaises(OSError, subprocess.Popen,
252 [rel_python], cwd=wrong_dir)
253 wrong_dir = self._normalize_cwd(wrong_dir)
254 self._assert_cwd(wrong_dir, abs_python, cwd=wrong_dir)
255
256 def test_executable_with_cwd(self):
257 python_dir, python_base = self._split_python_path()
258 python_dir = self._normalize_cwd(python_dir)
259 self._assert_cwd(python_dir, "somethingyoudonthave",
260 executable=sys.executable, cwd=python_dir)
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000261
262 @unittest.skipIf(sysconfig.is_python_build(),
263 "need an installed Python. See #7774")
264 def test_executable_without_cwd(self):
265 # For a normal installation, it should work without 'cwd'
266 # argument. For test runs in the build directory, see #7774.
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700267 self._assert_cwd('', "somethingyoudonthave", executable=sys.executable)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000268
269 def test_stdin_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000270 # stdin redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000271 p = subprocess.Popen([sys.executable, "-c",
272 'import sys; sys.exit(sys.stdin.read() == "pear")'],
273 stdin=subprocess.PIPE)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000274 p.stdin.write(b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000275 p.stdin.close()
276 p.wait()
277 self.assertEqual(p.returncode, 1)
278
279 def test_stdin_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000280 # stdin is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000281 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000282 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000283 d = tf.fileno()
Antoine Pitrou9cadb1b2008-09-15 23:02:56 +0000284 os.write(d, b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000285 os.lseek(d, 0, 0)
286 p = subprocess.Popen([sys.executable, "-c",
287 'import sys; sys.exit(sys.stdin.read() == "pear")'],
288 stdin=d)
289 p.wait()
290 self.assertEqual(p.returncode, 1)
291
292 def test_stdin_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000293 # stdin is set to open file object
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000294 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000295 self.addCleanup(tf.close)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000296 tf.write(b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000297 tf.seek(0)
298 p = subprocess.Popen([sys.executable, "-c",
299 'import sys; sys.exit(sys.stdin.read() == "pear")'],
300 stdin=tf)
301 p.wait()
302 self.assertEqual(p.returncode, 1)
303
304 def test_stdout_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000305 # stdout redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000306 p = subprocess.Popen([sys.executable, "-c",
307 'import sys; sys.stdout.write("orange")'],
308 stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000309 self.addCleanup(p.stdout.close)
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000310 self.assertEqual(p.stdout.read(), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000311
312 def test_stdout_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000313 # stdout is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000314 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000315 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000316 d = tf.fileno()
317 p = subprocess.Popen([sys.executable, "-c",
318 'import sys; sys.stdout.write("orange")'],
319 stdout=d)
320 p.wait()
321 os.lseek(d, 0, 0)
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000322 self.assertEqual(os.read(d, 1024), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000323
324 def test_stdout_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000325 # stdout is set to open file object
Tim Peterse718f612004-10-12 21:51:32 +0000326 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000327 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000328 p = subprocess.Popen([sys.executable, "-c",
329 'import sys; sys.stdout.write("orange")'],
330 stdout=tf)
331 p.wait()
332 tf.seek(0)
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000333 self.assertEqual(tf.read(), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000334
335 def test_stderr_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000336 # stderr redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000337 p = subprocess.Popen([sys.executable, "-c",
338 'import sys; sys.stderr.write("strawberry")'],
339 stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000340 self.addCleanup(p.stderr.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000341 self.assertStderrEqual(p.stderr.read(), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000342
343 def test_stderr_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000344 # stderr is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000345 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000346 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000347 d = tf.fileno()
348 p = subprocess.Popen([sys.executable, "-c",
349 'import sys; sys.stderr.write("strawberry")'],
350 stderr=d)
351 p.wait()
352 os.lseek(d, 0, 0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000353 self.assertStderrEqual(os.read(d, 1024), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000354
355 def test_stderr_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000356 # stderr is set to open file object
Tim Peterse718f612004-10-12 21:51:32 +0000357 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000358 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000359 p = subprocess.Popen([sys.executable, "-c",
360 'import sys; sys.stderr.write("strawberry")'],
361 stderr=tf)
362 p.wait()
363 tf.seek(0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000364 self.assertStderrEqual(tf.read(), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000365
366 def test_stdout_stderr_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000367 # capture stdout and stderr to the same pipe
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000368 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000369 'import sys;'
370 'sys.stdout.write("apple");'
371 'sys.stdout.flush();'
372 'sys.stderr.write("orange")'],
373 stdout=subprocess.PIPE,
374 stderr=subprocess.STDOUT)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000375 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000376 self.assertStderrEqual(p.stdout.read(), b"appleorange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000377
378 def test_stdout_stderr_file(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000379 # capture stdout and stderr to the same open file
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000380 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000381 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000382 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000383 'import sys;'
384 'sys.stdout.write("apple");'
385 'sys.stdout.flush();'
386 'sys.stderr.write("orange")'],
387 stdout=tf,
388 stderr=tf)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000389 p.wait()
390 tf.seek(0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000391 self.assertStderrEqual(tf.read(), b"appleorange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000392
Thomas Wouters89f507f2006-12-13 04:49:30 +0000393 def test_stdout_filedes_of_stdout(self):
394 # stdout is set to 1 (#1531862).
Antoine Pitrou9cadb1b2008-09-15 23:02:56 +0000395 cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), b'.\n'))"
Thomas Wouters89f507f2006-12-13 04:49:30 +0000396 rc = subprocess.call([sys.executable, "-c", cmd], stdout=1)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000397 self.assertEqual(rc, 2)
Thomas Wouters89f507f2006-12-13 04:49:30 +0000398
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000399 def test_env(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000400 newenv = os.environ.copy()
401 newenv["FRUIT"] = "orange"
Victor Stinnerf1512a22011-06-21 17:18:38 +0200402 with subprocess.Popen([sys.executable, "-c",
403 'import sys,os;'
404 'sys.stdout.write(os.getenv("FRUIT"))'],
405 stdout=subprocess.PIPE,
406 env=newenv) as p:
407 stdout, stderr = p.communicate()
408 self.assertEqual(stdout, b"orange")
409
Victor Stinner62d51182011-06-23 01:02:25 +0200410 # Windows requires at least the SYSTEMROOT environment variable to start
411 # Python
412 @unittest.skipIf(sys.platform == 'win32',
413 'cannot test an empty env on Windows')
Victor Stinner237e5cb2011-06-22 21:28:43 +0200414 @unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') is not None,
Victor Stinner372309a2011-06-21 21:59:06 +0200415 'the python library cannot be loaded '
416 'with an empty environment')
Victor Stinnerf1512a22011-06-21 17:18:38 +0200417 def test_empty_env(self):
418 with subprocess.Popen([sys.executable, "-c",
419 'import os; '
Victor Stinner372309a2011-06-21 21:59:06 +0200420 'print(list(os.environ.keys()))'],
Victor Stinnerf1512a22011-06-21 17:18:38 +0200421 stdout=subprocess.PIPE,
422 env={}) as p:
423 stdout, stderr = p.communicate()
Victor Stinner237e5cb2011-06-22 21:28:43 +0200424 self.assertIn(stdout.strip(),
425 (b"[]",
426 # Mac OS X adds __CF_USER_TEXT_ENCODING variable to an empty
427 # environment
428 b"['__CF_USER_TEXT_ENCODING']"))
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000429
Peter Astrandcbac93c2005-03-03 20:24:28 +0000430 def test_communicate_stdin(self):
431 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000432 'import sys;'
433 'sys.exit(sys.stdin.read() == "pear")'],
Peter Astrandcbac93c2005-03-03 20:24:28 +0000434 stdin=subprocess.PIPE)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000435 p.communicate(b"pear")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000436 self.assertEqual(p.returncode, 1)
437
438 def test_communicate_stdout(self):
439 p = subprocess.Popen([sys.executable, "-c",
440 'import sys; sys.stdout.write("pineapple")'],
441 stdout=subprocess.PIPE)
442 (stdout, stderr) = p.communicate()
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000443 self.assertEqual(stdout, b"pineapple")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000444 self.assertEqual(stderr, None)
445
446 def test_communicate_stderr(self):
447 p = subprocess.Popen([sys.executable, "-c",
448 'import sys; sys.stderr.write("pineapple")'],
449 stderr=subprocess.PIPE)
450 (stdout, stderr) = p.communicate()
451 self.assertEqual(stdout, None)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000452 self.assertStderrEqual(stderr, b"pineapple")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000453
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000454 def test_communicate(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000455 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000456 'import sys,os;'
457 'sys.stderr.write("pineapple");'
458 'sys.stdout.write(sys.stdin.read())'],
459 stdin=subprocess.PIPE,
460 stdout=subprocess.PIPE,
461 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000462 self.addCleanup(p.stdout.close)
463 self.addCleanup(p.stderr.close)
464 self.addCleanup(p.stdin.close)
Georg Brandl1abcbf82008-07-01 19:28:43 +0000465 (stdout, stderr) = p.communicate(b"banana")
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000466 self.assertEqual(stdout, b"banana")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000467 self.assertStderrEqual(stderr, b"pineapple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000468
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000469 # Test for the fd leak reported in http://bugs.python.org/issue2791.
470 def test_communicate_pipe_fd_leak(self):
Victor Stinner667d4b52010-12-25 22:40:32 +0000471 for stdin_pipe in (False, True):
472 for stdout_pipe in (False, True):
473 for stderr_pipe in (False, True):
474 options = {}
475 if stdin_pipe:
476 options['stdin'] = subprocess.PIPE
477 if stdout_pipe:
478 options['stdout'] = subprocess.PIPE
479 if stderr_pipe:
480 options['stderr'] = subprocess.PIPE
481 if not options:
482 continue
483 p = subprocess.Popen((sys.executable, "-c", "pass"), **options)
484 p.communicate()
485 if p.stdin is not None:
486 self.assertTrue(p.stdin.closed)
487 if p.stdout is not None:
488 self.assertTrue(p.stdout.closed)
489 if p.stderr is not None:
490 self.assertTrue(p.stderr.closed)
Georg Brandlf08a9dd2008-06-10 16:57:31 +0000491
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000492 def test_communicate_returns(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000493 # communicate() should return None if no redirection is active
Tim Peters3b01a702004-10-12 22:19:32 +0000494 p = subprocess.Popen([sys.executable, "-c",
495 "import sys; sys.exit(47)"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000496 (stdout, stderr) = p.communicate()
497 self.assertEqual(stdout, None)
498 self.assertEqual(stderr, None)
499
500 def test_communicate_pipe_buf(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000501 # communicate() with writes larger than pipe_buf
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000502 # This test will probably deadlock rather than fail, if
Tim Peterse718f612004-10-12 21:51:32 +0000503 # communicate() does not work properly.
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000504 x, y = os.pipe()
505 if mswindows:
506 pipe_buf = 512
507 else:
508 pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")
509 os.close(x)
510 os.close(y)
511 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000512 'import sys,os;'
513 'sys.stdout.write(sys.stdin.read(47));'
514 'sys.stderr.write("xyz"*%d);'
515 'sys.stdout.write(sys.stdin.read())' % pipe_buf],
516 stdin=subprocess.PIPE,
517 stdout=subprocess.PIPE,
518 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000519 self.addCleanup(p.stdout.close)
520 self.addCleanup(p.stderr.close)
521 self.addCleanup(p.stdin.close)
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000522 string_to_write = b"abc"*pipe_buf
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000523 (stdout, stderr) = p.communicate(string_to_write)
524 self.assertEqual(stdout, string_to_write)
525
526 def test_writes_before_communicate(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000527 # stdin.write before communicate()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000528 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000529 'import sys,os;'
530 'sys.stdout.write(sys.stdin.read())'],
531 stdin=subprocess.PIPE,
532 stdout=subprocess.PIPE,
533 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000534 self.addCleanup(p.stdout.close)
535 self.addCleanup(p.stderr.close)
536 self.addCleanup(p.stdin.close)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000537 p.stdin.write(b"banana")
538 (stdout, stderr) = p.communicate(b"split")
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000539 self.assertEqual(stdout, b"bananasplit")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000540 self.assertStderrEqual(stderr, b"")
Tim Peterse718f612004-10-12 21:51:32 +0000541
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000542 def test_universal_newlines(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000543 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000544 'import sys,os;' + SETBINARY +
Antoine Pitrouec2d2692012-08-05 00:23:40 +0200545 'buf = sys.stdout.buffer;'
546 'buf.write(sys.stdin.readline().encode());'
547 'buf.flush();'
548 'buf.write(b"line2\\n");'
549 'buf.flush();'
550 'buf.write(sys.stdin.read().encode());'
551 'buf.flush();'
552 'buf.write(b"line4\\n");'
553 'buf.flush();'
554 'buf.write(b"line5\\r\\n");'
555 'buf.flush();'
556 'buf.write(b"line6\\r");'
557 'buf.flush();'
558 'buf.write(b"\\nline7");'
559 'buf.flush();'
560 'buf.write(b"\\nline8");'],
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200561 stdin=subprocess.PIPE,
Guido van Rossum98297ee2007-11-06 21:34:58 +0000562 stdout=subprocess.PIPE,
563 universal_newlines=1)
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200564 p.stdin.write("line1\n")
565 self.assertEqual(p.stdout.readline(), "line1\n")
566 p.stdin.write("line3\n")
567 p.stdin.close()
Brian Curtin3c6a9512010-11-05 03:58:52 +0000568 self.addCleanup(p.stdout.close)
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200569 self.assertEqual(p.stdout.readline(),
570 "line2\n")
571 self.assertEqual(p.stdout.read(6),
572 "line3\n")
573 self.assertEqual(p.stdout.read(),
574 "line4\nline5\nline6\nline7\nline8")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000575
576 def test_universal_newlines_communicate(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000577 # universal newlines through communicate()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000578 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000579 'import sys,os;' + SETBINARY +
Antoine Pitrouec2d2692012-08-05 00:23:40 +0200580 'buf = sys.stdout.buffer;'
581 'buf.write(b"line2\\n");'
582 'buf.flush();'
583 'buf.write(b"line4\\n");'
584 'buf.flush();'
585 'buf.write(b"line5\\r\\n");'
586 'buf.flush();'
587 'buf.write(b"line6\\r");'
588 'buf.flush();'
589 'buf.write(b"\\nline7");'
590 'buf.flush();'
591 'buf.write(b"\\nline8");'],
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200592 stderr=subprocess.PIPE,
593 stdout=subprocess.PIPE,
Guido van Rossum98297ee2007-11-06 21:34:58 +0000594 universal_newlines=1)
Brian Curtin19a53792010-11-05 17:09:05 +0000595 self.addCleanup(p.stdout.close)
596 self.addCleanup(p.stderr.close)
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200597 # BUG: can't give a non-empty stdin because it breaks both the
598 # select- and poll-based communicate() implementations.
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000599 (stdout, stderr) = p.communicate()
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200600 self.assertEqual(stdout,
601 "line2\nline4\nline5\nline6\nline7\nline8")
602
603 def test_universal_newlines_communicate_stdin(self):
604 # universal newlines through communicate(), with only stdin
605 p = subprocess.Popen([sys.executable, "-c",
606 'import sys,os;' + SETBINARY + '''\nif True:
607 s = sys.stdin.readline()
608 assert s == "line1\\n", repr(s)
609 s = sys.stdin.read()
610 assert s == "line3\\n", repr(s)
611 '''],
612 stdin=subprocess.PIPE,
613 universal_newlines=1)
614 (stdout, stderr) = p.communicate("line1\nline3\n")
615 self.assertEqual(p.returncode, 0)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000616
Andrew Svetlovf3765072012-08-14 18:35:17 +0300617 def test_universal_newlines_communicate_input_none(self):
618 # Test communicate(input=None) with universal newlines.
619 #
620 # We set stdout to PIPE because, as of this writing, a different
621 # code path is tested when the number of pipes is zero or one.
622 p = subprocess.Popen([sys.executable, "-c", "pass"],
623 stdin=subprocess.PIPE,
624 stdout=subprocess.PIPE,
625 universal_newlines=True)
626 p.communicate()
627 self.assertEqual(p.returncode, 0)
628
Andrew Svetlov82860712012-08-19 22:13:41 +0300629 def test_universal_newlines_communicate_encodings(self):
630 # Check that universal newlines mode works for various encodings,
631 # in particular for encodings in the UTF-16 and UTF-32 families.
632 # See issue #15595.
633 #
634 # UTF-16 and UTF-32-BE are sufficient to check both with BOM and
635 # without, and UTF-16 and UTF-32.
636 for encoding in ['utf-16', 'utf-32-be']:
637 old_getpreferredencoding = locale.getpreferredencoding
638 # Indirectly via io.TextIOWrapper, Popen() defaults to
639 # locale.getpreferredencoding(False) and earlier in Python 3.2 to
640 # locale.getpreferredencoding().
641 def getpreferredencoding(do_setlocale=True):
642 return encoding
643 code = ("import sys; "
644 r"sys.stdout.buffer.write('1\r\n2\r3\n4'.encode('%s'))" %
645 encoding)
646 args = [sys.executable, '-c', code]
647 try:
648 locale.getpreferredencoding = getpreferredencoding
649 # We set stdin to be non-None because, as of this writing,
650 # a different code path is used when the number of pipes is
651 # zero or one.
652 popen = subprocess.Popen(args, universal_newlines=True,
653 stdin=subprocess.PIPE,
654 stdout=subprocess.PIPE)
655 stdout, stderr = popen.communicate(input='')
656 finally:
657 locale.getpreferredencoding = old_getpreferredencoding
658
659 self.assertEqual(stdout, '1\n2\n3\n4')
660
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000661 def test_no_leaking(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000662 # Make sure we leak no resources
Antoine Pitrou8db30272010-09-18 22:38:48 +0000663 if not mswindows:
Peter Astrandf7f1bb72005-03-03 20:47:37 +0000664 max_handles = 1026 # too much for most UNIX systems
665 else:
Antoine Pitrou8db30272010-09-18 22:38:48 +0000666 max_handles = 2050 # too much for (at least some) Windows setups
667 handles = []
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400668 tmpdir = tempfile.mkdtemp()
Antoine Pitrou8db30272010-09-18 22:38:48 +0000669 try:
670 for i in range(max_handles):
671 try:
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400672 tmpfile = os.path.join(tmpdir, support.TESTFN)
673 handles.append(os.open(tmpfile, os.O_WRONLY|os.O_CREAT))
Antoine Pitrou8db30272010-09-18 22:38:48 +0000674 except OSError as e:
675 if e.errno != errno.EMFILE:
676 raise
677 break
678 else:
679 self.skipTest("failed to reach the file descriptor limit "
680 "(tried %d)" % max_handles)
681 # Close a couple of them (should be enough for a subprocess)
682 for i in range(10):
683 os.close(handles.pop())
684 # Loop creating some subprocesses. If one of them leaks some fds,
685 # the next loop iteration will fail by reaching the max fd limit.
686 for i in range(15):
687 p = subprocess.Popen([sys.executable, "-c",
688 "import sys;"
689 "sys.stdout.write(sys.stdin.read())"],
690 stdin=subprocess.PIPE,
691 stdout=subprocess.PIPE,
692 stderr=subprocess.PIPE)
693 data = p.communicate(b"lime")[0]
694 self.assertEqual(data, b"lime")
695 finally:
696 for h in handles:
697 os.close(h)
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400698 shutil.rmtree(tmpdir)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000699
700 def test_list2cmdline(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000701 self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),
702 '"a b c" d e')
703 self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),
704 'ab\\"c \\ d')
Christian Heimesfdab48e2008-01-20 09:06:41 +0000705 self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']),
706 'ab\\"c " \\\\" d')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000707 self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),
708 'a\\\\\\b "de fg" h')
709 self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),
710 'a\\\\\\"b c d')
711 self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),
712 '"a\\\\b c" d e')
713 self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),
714 '"a\\\\b\\ c" d e')
Thomas Woutersfc7bb8c2007-01-15 15:49:28 +0000715 self.assertEqual(subprocess.list2cmdline(['ab', '']),
716 'ab ""')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000717
718
719 def test_poll(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000720 p = subprocess.Popen([sys.executable,
Tim Peters29b6b4f2004-10-13 03:43:40 +0000721 "-c", "import time; time.sleep(1)"])
722 count = 0
723 while p.poll() is None:
724 time.sleep(0.1)
725 count += 1
726 # We expect that the poll loop probably went around about 10 times,
727 # but, based on system scheduling we can't control, it's possible
728 # poll() never returned None. It "should be" very rare that it
729 # didn't go around at least twice.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000730 self.assertGreaterEqual(count, 2)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000731 # Subsequent invocations should just return the returncode
732 self.assertEqual(p.poll(), 0)
733
734
735 def test_wait(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000736 p = subprocess.Popen([sys.executable,
737 "-c", "import time; time.sleep(2)"])
738 self.assertEqual(p.wait(), 0)
739 # Subsequent invocations should just return the returncode
740 self.assertEqual(p.wait(), 0)
Tim Peterse718f612004-10-12 21:51:32 +0000741
Peter Astrand738131d2004-11-30 21:04:45 +0000742
743 def test_invalid_bufsize(self):
744 # an invalid type of the bufsize argument should raise
745 # TypeError.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000746 with self.assertRaises(TypeError):
Peter Astrand738131d2004-11-30 21:04:45 +0000747 subprocess.Popen([sys.executable, "-c", "pass"], "orange")
Peter Astrand738131d2004-11-30 21:04:45 +0000748
Guido van Rossum46a05a72007-06-07 21:56:45 +0000749 def test_bufsize_is_none(self):
750 # bufsize=None should be the same as bufsize=0.
751 p = subprocess.Popen([sys.executable, "-c", "pass"], None)
752 self.assertEqual(p.wait(), 0)
753 # Again with keyword arg
754 p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None)
755 self.assertEqual(p.wait(), 0)
756
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000757 def test_leaking_fds_on_error(self):
758 # see bug #5179: Popen leaks file descriptors to PIPEs if
759 # the child fails to execute; this will eventually exhaust
760 # the maximum number of open fds. 1024 seems a very common
761 # value for that limit, but Windows has 2048, so we loop
762 # 1024 times (each call leaked two fds).
763 for i in range(1024):
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000764 # Windows raises IOError. Others raise OSError.
765 with self.assertRaises(EnvironmentError) as c:
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000766 subprocess.Popen(['nonexisting_i_hope'],
767 stdout=subprocess.PIPE,
768 stderr=subprocess.PIPE)
R David Murray384069c2011-03-13 22:26:53 -0400769 # ignore errors that indicate the command was not found
R David Murray6924bd72011-03-13 22:48:55 -0400770 if c.exception.errno not in (errno.ENOENT, errno.EACCES):
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000771 raise c.exception
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000772
Victor Stinnerb3693582010-05-21 20:13:12 +0000773 def test_issue8780(self):
774 # Ensure that stdout is inherited from the parent
775 # if stdout=PIPE is not used
776 code = ';'.join((
777 'import subprocess, sys',
778 'retcode = subprocess.call('
779 "[sys.executable, '-c', 'print(\"Hello World!\")'])",
780 'assert retcode == 0'))
781 output = subprocess.check_output([sys.executable, '-c', code])
Ezio Melottib3aedd42010-11-20 19:04:17 +0000782 self.assertTrue(output.startswith(b'Hello World!'), ascii(output))
Victor Stinnerb3693582010-05-21 20:13:12 +0000783
Tim Goldenaf5ac392010-08-06 13:03:56 +0000784 def test_handles_closed_on_exception(self):
785 # If CreateProcess exits with an error, ensure the
786 # duplicate output handles are released
787 ifhandle, ifname = mkstemp()
788 ofhandle, ofname = mkstemp()
789 efhandle, efname = mkstemp()
790 try:
791 subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle,
792 stderr=efhandle)
793 except OSError:
794 os.close(ifhandle)
795 os.remove(ifname)
796 os.close(ofhandle)
797 os.remove(ofname)
798 os.close(efhandle)
799 os.remove(efname)
800 self.assertFalse(os.path.exists(ifname))
801 self.assertFalse(os.path.exists(ofname))
802 self.assertFalse(os.path.exists(efname))
803
Ross Lagerwall4f61b022011-04-05 15:34:00 +0200804 def test_communicate_epipe(self):
805 # Issue 10963: communicate() should hide EPIPE
806 p = subprocess.Popen([sys.executable, "-c", 'pass'],
807 stdin=subprocess.PIPE,
808 stdout=subprocess.PIPE,
809 stderr=subprocess.PIPE)
810 self.addCleanup(p.stdout.close)
811 self.addCleanup(p.stderr.close)
812 self.addCleanup(p.stdin.close)
813 p.communicate(b"x" * 2**20)
814
815 def test_communicate_epipe_only_stdin(self):
816 # Issue 10963: communicate() should hide EPIPE
817 p = subprocess.Popen([sys.executable, "-c", 'pass'],
818 stdin=subprocess.PIPE)
819 self.addCleanup(p.stdin.close)
820 time.sleep(2)
821 p.communicate(b"x" * 2**20)
822
Victor Stinner1848db82011-07-05 14:49:46 +0200823 @unittest.skipUnless(hasattr(signal, 'SIGALRM'),
824 "Requires signal.SIGALRM")
Victor Stinner2cfb6f32011-07-05 14:00:56 +0200825 def test_communicate_eintr(self):
826 # Issue #12493: communicate() should handle EINTR
827 def handler(signum, frame):
828 pass
829 old_handler = signal.signal(signal.SIGALRM, handler)
830 self.addCleanup(signal.signal, signal.SIGALRM, old_handler)
831
832 # the process is running for 2 seconds
833 args = [sys.executable, "-c", 'import time; time.sleep(2)']
834 for stream in ('stdout', 'stderr'):
835 kw = {stream: subprocess.PIPE}
836 with subprocess.Popen(args, **kw) as process:
837 signal.alarm(1)
838 # communicate() will be interrupted by SIGALRM
839 process.communicate()
840
Tim Peterse718f612004-10-12 21:51:32 +0000841
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000842# context manager
843class _SuppressCoreFiles(object):
844 """Try to prevent core files from being created."""
845 old_limit = None
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000846
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000847 def __enter__(self):
848 """Try to save previous ulimit, then set it to (0, 0)."""
Benjamin Peterson964561b2011-12-10 12:31:42 -0500849 if resource is not None:
850 try:
851 self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
852 resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
853 except (ValueError, resource.error):
854 pass
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000855
Ronald Oussoren102d11a2010-07-23 09:50:05 +0000856 if sys.platform == 'darwin':
857 # Check if the 'Crash Reporter' on OSX was configured
858 # in 'Developer' mode and warn that it will get triggered
859 # when it is.
860 #
861 # This assumes that this context manager is used in tests
862 # that might trigger the next manager.
863 value = subprocess.Popen(['/usr/bin/defaults', 'read',
864 'com.apple.CrashReporter', 'DialogType'],
865 stdout=subprocess.PIPE).communicate()[0]
866 if value.strip() == b'developer':
867 print("this tests triggers the Crash Reporter, "
868 "that is intentional", end='')
869 sys.stdout.flush()
870
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000871 def __exit__(self, *args):
872 """Return core file behavior to default."""
873 if self.old_limit is None:
874 return
Benjamin Peterson964561b2011-12-10 12:31:42 -0500875 if resource is not None:
876 try:
877 resource.setrlimit(resource.RLIMIT_CORE, self.old_limit)
878 except (ValueError, resource.error):
879 pass
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000880
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000881
Florent Xiclunaf0cbd822010-03-04 21:50:56 +0000882@unittest.skipIf(mswindows, "POSIX specific tests")
Florent Xiclunac049d872010-03-27 22:47:23 +0000883class POSIXProcessTestCase(BaseTestCase):
Florent Xiclunaf0cbd822010-03-04 21:50:56 +0000884
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000885 def test_exceptions(self):
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000886 nonexistent_dir = "/_this/pa.th/does/not/exist"
887 try:
888 os.chdir(nonexistent_dir)
889 except OSError as e:
890 # This avoids hard coding the errno value or the OS perror()
891 # string and instead capture the exception that we want to see
892 # below for comparison.
893 desired_exception = e
Benjamin Peterson5f780402010-11-20 18:07:52 +0000894 desired_exception.strerror += ': ' + repr(sys.executable)
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000895 else:
896 self.fail("chdir to nonexistant directory %s succeeded." %
897 nonexistent_dir)
898
899 # Error in the child re-raised in the parent.
900 try:
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000901 p = subprocess.Popen([sys.executable, "-c", ""],
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000902 cwd=nonexistent_dir)
903 except OSError as e:
904 # Test that the child process chdir failure actually makes
905 # it up to the parent process as the correct exception.
906 self.assertEqual(desired_exception.errno, e.errno)
907 self.assertEqual(desired_exception.strerror, e.strerror)
908 else:
909 self.fail("Expected OSError: %s" % desired_exception)
910
911 def test_restore_signals(self):
912 # Code coverage for both values of restore_signals to make sure it
913 # at least does not blow up.
914 # A test for behavior would be complex. Contributions welcome.
915 subprocess.call([sys.executable, "-c", ""], restore_signals=True)
916 subprocess.call([sys.executable, "-c", ""], restore_signals=False)
917
918 def test_start_new_session(self):
919 # For code coverage of calling setsid(). We don't care if we get an
920 # EPERM error from it depending on the test execution environment, that
921 # still indicates that it was called.
922 try:
923 output = subprocess.check_output(
924 [sys.executable, "-c",
925 "import os; print(os.getpgid(os.getpid()))"],
926 start_new_session=True)
927 except OSError as e:
928 if e.errno != errno.EPERM:
929 raise
930 else:
931 parent_pgid = os.getpgid(os.getpid())
932 child_pgid = int(output)
933 self.assertNotEqual(parent_pgid, child_pgid)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000934
935 def test_run_abort(self):
936 # returncode handles signal termination
937 with _SuppressCoreFiles():
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000938 p = subprocess.Popen([sys.executable, "-c",
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000939 'import os; os.abort()'])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000940 p.wait()
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000941 self.assertEqual(-p.returncode, signal.SIGABRT)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000942
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000943 def test_preexec(self):
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000944 # DISCLAIMER: Setting environment variables is *not* a good use
945 # of a preexec_fn. This is merely a test.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000946 p = subprocess.Popen([sys.executable, "-c",
947 'import sys,os;'
948 'sys.stdout.write(os.getenv("FRUIT"))'],
949 stdout=subprocess.PIPE,
950 preexec_fn=lambda: os.putenv("FRUIT", "apple"))
Brian Curtin3c6a9512010-11-05 03:58:52 +0000951 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000952 self.assertEqual(p.stdout.read(), b"apple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000953
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000954 def test_preexec_exception(self):
955 def raise_it():
956 raise ValueError("What if two swallows carried a coconut?")
957 try:
958 p = subprocess.Popen([sys.executable, "-c", ""],
959 preexec_fn=raise_it)
960 except RuntimeError as e:
961 self.assertTrue(
962 subprocess._posixsubprocess,
963 "Expected a ValueError from the preexec_fn")
964 except ValueError as e:
965 self.assertIn("coconut", e.args[0])
966 else:
967 self.fail("Exception raised by preexec_fn did not make it "
968 "to the parent process.")
969
Gregory P. Smith32ec9da2010-03-19 16:53:08 +0000970 def test_preexec_gc_module_failure(self):
971 # This tests the code that disables garbage collection if the child
972 # process will execute any Python.
973 def raise_runtime_error():
974 raise RuntimeError("this shouldn't escape")
975 enabled = gc.isenabled()
976 orig_gc_disable = gc.disable
977 orig_gc_isenabled = gc.isenabled
978 try:
979 gc.disable()
980 self.assertFalse(gc.isenabled())
981 subprocess.call([sys.executable, '-c', ''],
982 preexec_fn=lambda: None)
983 self.assertFalse(gc.isenabled(),
984 "Popen enabled gc when it shouldn't.")
985
986 gc.enable()
987 self.assertTrue(gc.isenabled())
988 subprocess.call([sys.executable, '-c', ''],
989 preexec_fn=lambda: None)
990 self.assertTrue(gc.isenabled(), "Popen left gc disabled.")
991
992 gc.disable = raise_runtime_error
993 self.assertRaises(RuntimeError, subprocess.Popen,
994 [sys.executable, '-c', ''],
995 preexec_fn=lambda: None)
996
997 del gc.isenabled # force an AttributeError
998 self.assertRaises(AttributeError, subprocess.Popen,
999 [sys.executable, '-c', ''],
1000 preexec_fn=lambda: None)
1001 finally:
1002 gc.disable = orig_gc_disable
1003 gc.isenabled = orig_gc_isenabled
1004 if not enabled:
1005 gc.disable()
1006
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001007 def test_args_string(self):
1008 # args is a string
1009 fd, fname = mkstemp()
1010 # reopen in text mode
Victor Stinnerf6782ac2010-10-16 23:46:43 +00001011 with open(fd, "w", errors="surrogateescape") as fobj:
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001012 fobj.write("#!/bin/sh\n")
1013 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
1014 sys.executable)
1015 os.chmod(fname, 0o700)
1016 p = subprocess.Popen(fname)
1017 p.wait()
1018 os.remove(fname)
1019 self.assertEqual(p.returncode, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001020
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001021 def test_invalid_args(self):
1022 # invalid arguments should raise ValueError
1023 self.assertRaises(ValueError, subprocess.call,
1024 [sys.executable, "-c",
1025 "import sys; sys.exit(47)"],
1026 startupinfo=47)
1027 self.assertRaises(ValueError, subprocess.call,
1028 [sys.executable, "-c",
1029 "import sys; sys.exit(47)"],
1030 creationflags=47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001031
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001032 def test_shell_sequence(self):
1033 # Run command through the shell (sequence)
1034 newenv = os.environ.copy()
1035 newenv["FRUIT"] = "apple"
1036 p = subprocess.Popen(["echo $FRUIT"], shell=1,
1037 stdout=subprocess.PIPE,
1038 env=newenv)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001039 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001040 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001041
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001042 def test_shell_string(self):
1043 # Run command through the shell (string)
1044 newenv = os.environ.copy()
1045 newenv["FRUIT"] = "apple"
1046 p = subprocess.Popen("echo $FRUIT", shell=1,
1047 stdout=subprocess.PIPE,
1048 env=newenv)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001049 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001050 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
Christian Heimesa342c012008-04-20 21:01:16 +00001051
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001052 def test_call_string(self):
1053 # call() function with string argument on UNIX
1054 fd, fname = mkstemp()
1055 # reopen in text mode
Victor Stinnerf6782ac2010-10-16 23:46:43 +00001056 with open(fd, "w", errors="surrogateescape") as fobj:
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001057 fobj.write("#!/bin/sh\n")
1058 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
1059 sys.executable)
1060 os.chmod(fname, 0o700)
1061 rc = subprocess.call(fname)
1062 os.remove(fname)
1063 self.assertEqual(rc, 47)
Christian Heimesa342c012008-04-20 21:01:16 +00001064
Stefan Krah9542cc62010-07-19 14:20:53 +00001065 def test_specific_shell(self):
1066 # Issue #9265: Incorrect name passed as arg[0].
1067 shells = []
1068 for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']:
1069 for name in ['bash', 'ksh']:
1070 sh = os.path.join(prefix, name)
1071 if os.path.isfile(sh):
1072 shells.append(sh)
1073 if not shells: # Will probably work for any shell but csh.
1074 self.skipTest("bash or ksh required for this test")
1075 sh = '/bin/sh'
1076 if os.path.isfile(sh) and not os.path.islink(sh):
1077 # Test will fail if /bin/sh is a symlink to csh.
1078 shells.append(sh)
1079 for sh in shells:
1080 p = subprocess.Popen("echo $0", executable=sh, shell=True,
1081 stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001082 self.addCleanup(p.stdout.close)
Stefan Krah9542cc62010-07-19 14:20:53 +00001083 self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii'))
1084
Florent Xicluna4886d242010-03-08 13:27:26 +00001085 def _kill_process(self, method, *args):
Florent Xicluna1d8ee3a2010-03-05 20:26:54 +00001086 # Do not inherit file handles from the parent.
1087 # It should fix failures on some platforms.
Antoine Pitrou3d8580f2010-09-20 01:33:21 +00001088 p = subprocess.Popen([sys.executable, "-c", """if 1:
1089 import sys, time
1090 sys.stdout.write('x\\n')
1091 sys.stdout.flush()
1092 time.sleep(30)
1093 """],
1094 close_fds=True,
1095 stdin=subprocess.PIPE,
1096 stdout=subprocess.PIPE,
1097 stderr=subprocess.PIPE)
1098 # Wait for the interpreter to be completely initialized before
1099 # sending any signal.
1100 p.stdout.read(1)
1101 getattr(p, method)(*args)
Florent Xicluna4886d242010-03-08 13:27:26 +00001102 return p
1103
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01001104 def _kill_dead_process(self, method, *args):
1105 # Do not inherit file handles from the parent.
1106 # It should fix failures on some platforms.
1107 p = subprocess.Popen([sys.executable, "-c", """if 1:
1108 import sys, time
1109 sys.stdout.write('x\\n')
1110 sys.stdout.flush()
1111 """],
1112 close_fds=True,
1113 stdin=subprocess.PIPE,
1114 stdout=subprocess.PIPE,
1115 stderr=subprocess.PIPE)
1116 # Wait for the interpreter to be completely initialized before
1117 # sending any signal.
1118 p.stdout.read(1)
1119 # The process should end after this
1120 time.sleep(1)
1121 # This shouldn't raise even though the child is now dead
1122 getattr(p, method)(*args)
1123 p.communicate()
1124
Florent Xicluna4886d242010-03-08 13:27:26 +00001125 def test_send_signal(self):
1126 p = self._kill_process('send_signal', signal.SIGINT)
Florent Xiclunac049d872010-03-27 22:47:23 +00001127 _, stderr = p.communicate()
1128 self.assertIn(b'KeyboardInterrupt', stderr)
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001129 self.assertNotEqual(p.wait(), 0)
Christian Heimesa342c012008-04-20 21:01:16 +00001130
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001131 def test_kill(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001132 p = self._kill_process('kill')
Florent Xiclunac049d872010-03-27 22:47:23 +00001133 _, stderr = p.communicate()
1134 self.assertStderrEqual(stderr, b'')
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001135 self.assertEqual(p.wait(), -signal.SIGKILL)
Tim Peterse718f612004-10-12 21:51:32 +00001136
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001137 def test_terminate(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001138 p = self._kill_process('terminate')
Florent Xiclunac049d872010-03-27 22:47:23 +00001139 _, stderr = p.communicate()
1140 self.assertStderrEqual(stderr, b'')
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001141 self.assertEqual(p.wait(), -signal.SIGTERM)
1142
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01001143 def test_send_signal_dead(self):
1144 # Sending a signal to a dead process
1145 self._kill_dead_process('send_signal', signal.SIGINT)
1146
1147 def test_kill_dead(self):
1148 # Killing a dead process
1149 self._kill_dead_process('kill')
1150
1151 def test_terminate_dead(self):
1152 # Terminating a dead process
1153 self._kill_dead_process('terminate')
1154
Antoine Pitrouc9c83ba2011-01-03 18:23:55 +00001155 def check_close_std_fds(self, fds):
1156 # Issue #9905: test that subprocess pipes still work properly with
1157 # some standard fds closed
1158 stdin = 0
1159 newfds = []
1160 for a in fds:
1161 b = os.dup(a)
1162 newfds.append(b)
1163 if a == 0:
1164 stdin = b
1165 try:
1166 for fd in fds:
1167 os.close(fd)
1168 out, err = subprocess.Popen([sys.executable, "-c",
1169 'import sys;'
1170 'sys.stdout.write("apple");'
1171 'sys.stdout.flush();'
1172 'sys.stderr.write("orange")'],
1173 stdin=stdin,
1174 stdout=subprocess.PIPE,
1175 stderr=subprocess.PIPE).communicate()
1176 err = support.strip_python_stderr(err)
1177 self.assertEqual((out, err), (b'apple', b'orange'))
1178 finally:
1179 for b, a in zip(newfds, fds):
1180 os.dup2(b, a)
1181 for b in newfds:
1182 os.close(b)
1183
1184 def test_close_fd_0(self):
1185 self.check_close_std_fds([0])
1186
1187 def test_close_fd_1(self):
1188 self.check_close_std_fds([1])
1189
1190 def test_close_fd_2(self):
1191 self.check_close_std_fds([2])
1192
1193 def test_close_fds_0_1(self):
1194 self.check_close_std_fds([0, 1])
1195
1196 def test_close_fds_0_2(self):
1197 self.check_close_std_fds([0, 2])
1198
1199 def test_close_fds_1_2(self):
1200 self.check_close_std_fds([1, 2])
1201
1202 def test_close_fds_0_1_2(self):
1203 # Issue #10806: test that subprocess pipes still work properly with
1204 # all standard fds closed.
1205 self.check_close_std_fds([0, 1, 2])
1206
Antoine Pitrou95aaeee2011-01-03 21:15:48 +00001207 def test_remapping_std_fds(self):
1208 # open up some temporary files
1209 temps = [mkstemp() for i in range(3)]
1210 try:
1211 temp_fds = [fd for fd, fname in temps]
1212
1213 # unlink the files -- we won't need to reopen them
1214 for fd, fname in temps:
1215 os.unlink(fname)
1216
1217 # write some data to what will become stdin, and rewind
1218 os.write(temp_fds[1], b"STDIN")
1219 os.lseek(temp_fds[1], 0, 0)
1220
1221 # move the standard file descriptors out of the way
1222 saved_fds = [os.dup(fd) for fd in range(3)]
1223 try:
1224 # duplicate the file objects over the standard fd's
1225 for fd, temp_fd in enumerate(temp_fds):
1226 os.dup2(temp_fd, fd)
1227
1228 # now use those files in the "wrong" order, so that subprocess
1229 # has to rearrange them in the child
1230 p = subprocess.Popen([sys.executable, "-c",
1231 'import sys; got = sys.stdin.read();'
1232 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1233 stdin=temp_fds[1],
1234 stdout=temp_fds[2],
1235 stderr=temp_fds[0])
1236 p.wait()
1237 finally:
1238 # restore the original fd's underneath sys.stdin, etc.
1239 for std, saved in enumerate(saved_fds):
1240 os.dup2(saved, std)
1241 os.close(saved)
1242
1243 for fd in temp_fds:
1244 os.lseek(fd, 0, 0)
1245
1246 out = os.read(temp_fds[2], 1024)
1247 err = support.strip_python_stderr(os.read(temp_fds[0], 1024))
1248 self.assertEqual(out, b"got STDIN")
1249 self.assertEqual(err, b"err")
1250
1251 finally:
1252 for fd in temp_fds:
1253 os.close(fd)
1254
Ross Lagerwalld98646e2011-07-27 07:16:31 +02001255 def check_swap_fds(self, stdin_no, stdout_no, stderr_no):
1256 # open up some temporary files
1257 temps = [mkstemp() for i in range(3)]
1258 temp_fds = [fd for fd, fname in temps]
1259 try:
1260 # unlink the files -- we won't need to reopen them
1261 for fd, fname in temps:
1262 os.unlink(fname)
1263
1264 # save a copy of the standard file descriptors
1265 saved_fds = [os.dup(fd) for fd in range(3)]
1266 try:
1267 # duplicate the temp files over the standard fd's 0, 1, 2
1268 for fd, temp_fd in enumerate(temp_fds):
1269 os.dup2(temp_fd, fd)
1270
1271 # write some data to what will become stdin, and rewind
1272 os.write(stdin_no, b"STDIN")
1273 os.lseek(stdin_no, 0, 0)
1274
1275 # now use those files in the given order, so that subprocess
1276 # has to rearrange them in the child
1277 p = subprocess.Popen([sys.executable, "-c",
1278 'import sys; got = sys.stdin.read();'
1279 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1280 stdin=stdin_no,
1281 stdout=stdout_no,
1282 stderr=stderr_no)
1283 p.wait()
1284
1285 for fd in temp_fds:
1286 os.lseek(fd, 0, 0)
1287
1288 out = os.read(stdout_no, 1024)
1289 err = support.strip_python_stderr(os.read(stderr_no, 1024))
1290 finally:
1291 for std, saved in enumerate(saved_fds):
1292 os.dup2(saved, std)
1293 os.close(saved)
1294
1295 self.assertEqual(out, b"got STDIN")
1296 self.assertEqual(err, b"err")
1297
1298 finally:
1299 for fd in temp_fds:
1300 os.close(fd)
1301
1302 # When duping fds, if there arises a situation where one of the fds is
1303 # either 0, 1 or 2, it is possible that it is overwritten (#12607).
1304 # This tests all combinations of this.
1305 def test_swap_fds(self):
1306 self.check_swap_fds(0, 1, 2)
1307 self.check_swap_fds(0, 2, 1)
1308 self.check_swap_fds(1, 0, 2)
1309 self.check_swap_fds(1, 2, 0)
1310 self.check_swap_fds(2, 0, 1)
1311 self.check_swap_fds(2, 1, 0)
1312
Victor Stinner13bb71c2010-04-23 21:41:56 +00001313 def test_surrogates_error_message(self):
Victor Stinner4d078042010-04-23 19:28:32 +00001314 def prepare():
1315 raise ValueError("surrogate:\uDCff")
1316
1317 try:
1318 subprocess.call(
1319 [sys.executable, "-c", "pass"],
1320 preexec_fn=prepare)
1321 except ValueError as err:
1322 # Pure Python implementations keeps the message
1323 self.assertIsNone(subprocess._posixsubprocess)
1324 self.assertEqual(str(err), "surrogate:\uDCff")
1325 except RuntimeError as err:
1326 # _posixsubprocess uses a default message
1327 self.assertIsNotNone(subprocess._posixsubprocess)
1328 self.assertEqual(str(err), "Exception occurred in preexec_fn.")
1329 else:
1330 self.fail("Expected ValueError or RuntimeError")
1331
Victor Stinner13bb71c2010-04-23 21:41:56 +00001332 def test_undecodable_env(self):
1333 for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')):
Victor Stinner13bb71c2010-04-23 21:41:56 +00001334 # test str with surrogates
Antoine Pitroufb8db8f2010-09-19 22:46:05 +00001335 script = "import os; print(ascii(os.getenv(%s)))" % repr(key)
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001336 env = os.environ.copy()
1337 env[key] = value
Victor Stinner89f3ad12010-10-14 10:43:31 +00001338 # Use C locale to get ascii for the locale encoding to force
1339 # surrogate-escaping of \xFF in the child process; otherwise it can
1340 # be decoded as-is if the default locale is latin-1.
Victor Stinnerebc78d22010-10-14 10:38:17 +00001341 env['LC_ALL'] = 'C'
Victor Stinner13bb71c2010-04-23 21:41:56 +00001342 stdout = subprocess.check_output(
1343 [sys.executable, "-c", script],
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001344 env=env)
Victor Stinner13bb71c2010-04-23 21:41:56 +00001345 stdout = stdout.rstrip(b'\n\r')
Ezio Melottib3aedd42010-11-20 19:04:17 +00001346 self.assertEqual(stdout.decode('ascii'), ascii(value))
Victor Stinner13bb71c2010-04-23 21:41:56 +00001347
1348 # test bytes
1349 key = key.encode("ascii", "surrogateescape")
1350 value = value.encode("ascii", "surrogateescape")
Antoine Pitroufb8db8f2010-09-19 22:46:05 +00001351 script = "import os; print(ascii(os.getenvb(%s)))" % repr(key)
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001352 env = os.environ.copy()
1353 env[key] = value
Victor Stinner13bb71c2010-04-23 21:41:56 +00001354 stdout = subprocess.check_output(
1355 [sys.executable, "-c", script],
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001356 env=env)
Victor Stinner13bb71c2010-04-23 21:41:56 +00001357 stdout = stdout.rstrip(b'\n\r')
Ezio Melottib3aedd42010-11-20 19:04:17 +00001358 self.assertEqual(stdout.decode('ascii'), ascii(value))
Victor Stinner13bb71c2010-04-23 21:41:56 +00001359
Victor Stinnerb745a742010-05-18 17:17:23 +00001360 def test_bytes_program(self):
1361 abs_program = os.fsencode(sys.executable)
1362 path, program = os.path.split(sys.executable)
1363 program = os.fsencode(program)
1364
1365 # absolute bytes path
1366 exitcode = subprocess.call([abs_program, "-c", "pass"])
Ezio Melottib3aedd42010-11-20 19:04:17 +00001367 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001368
1369 # bytes program, unicode PATH
1370 env = os.environ.copy()
1371 env["PATH"] = path
1372 exitcode = subprocess.call([program, "-c", "pass"], env=env)
Ezio Melottib3aedd42010-11-20 19:04:17 +00001373 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001374
1375 # bytes program, bytes PATH
1376 envb = os.environb.copy()
1377 envb[b"PATH"] = os.fsencode(path)
1378 exitcode = subprocess.call([program, "-c", "pass"], env=envb)
Ezio Melottib3aedd42010-11-20 19:04:17 +00001379 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001380
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001381 def test_pipe_cloexec(self):
1382 sleeper = support.findfile("input_reader.py", subdir="subprocessdata")
1383 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1384
1385 p1 = subprocess.Popen([sys.executable, sleeper],
1386 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1387 stderr=subprocess.PIPE, close_fds=False)
1388
1389 self.addCleanup(p1.communicate, b'')
1390
1391 p2 = subprocess.Popen([sys.executable, fd_status],
1392 stdout=subprocess.PIPE, close_fds=False)
1393
1394 output, error = p2.communicate()
1395 result_fds = set(map(int, output.split(b',')))
1396 unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(),
1397 p1.stderr.fileno()])
1398
1399 self.assertFalse(result_fds & unwanted_fds,
1400 "Expected no fds from %r to be open in child, "
1401 "found %r" %
1402 (unwanted_fds, result_fds & unwanted_fds))
1403
1404 def test_pipe_cloexec_real_tools(self):
1405 qcat = support.findfile("qcat.py", subdir="subprocessdata")
1406 qgrep = support.findfile("qgrep.py", subdir="subprocessdata")
1407
1408 subdata = b'zxcvbn'
1409 data = subdata * 4 + b'\n'
1410
1411 p1 = subprocess.Popen([sys.executable, qcat],
1412 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1413 close_fds=False)
1414
1415 p2 = subprocess.Popen([sys.executable, qgrep, subdata],
1416 stdin=p1.stdout, stdout=subprocess.PIPE,
1417 close_fds=False)
1418
1419 self.addCleanup(p1.wait)
1420 self.addCleanup(p2.wait)
Gregory P. Smith886455c2012-01-21 22:05:10 -08001421 def kill_p1():
1422 try:
1423 p1.terminate()
1424 except ProcessLookupError:
1425 pass
1426 def kill_p2():
1427 try:
1428 p2.terminate()
1429 except ProcessLookupError:
1430 pass
1431 self.addCleanup(kill_p1)
1432 self.addCleanup(kill_p2)
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001433
1434 p1.stdin.write(data)
1435 p1.stdin.close()
1436
1437 readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10)
1438
1439 self.assertTrue(readfiles, "The child hung")
1440 self.assertEqual(p2.stdout.read(), data)
1441
Victor Stinnerfaa8c132011-01-03 16:36:00 +00001442 p1.stdout.close()
1443 p2.stdout.close()
1444
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001445 def test_close_fds(self):
1446 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1447
1448 fds = os.pipe()
1449 self.addCleanup(os.close, fds[0])
1450 self.addCleanup(os.close, fds[1])
1451
1452 open_fds = set(fds)
Gregory P. Smith8facece2012-01-21 14:01:08 -08001453 # add a bunch more fds
1454 for _ in range(9):
1455 fd = os.open("/dev/null", os.O_RDONLY)
1456 self.addCleanup(os.close, fd)
1457 open_fds.add(fd)
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001458
1459 p = subprocess.Popen([sys.executable, fd_status],
1460 stdout=subprocess.PIPE, close_fds=False)
1461 output, ignored = p.communicate()
1462 remaining_fds = set(map(int, output.split(b',')))
1463
1464 self.assertEqual(remaining_fds & open_fds, open_fds,
1465 "Some fds were closed")
1466
1467 p = subprocess.Popen([sys.executable, fd_status],
1468 stdout=subprocess.PIPE, close_fds=True)
1469 output, ignored = p.communicate()
1470 remaining_fds = set(map(int, output.split(b',')))
1471
1472 self.assertFalse(remaining_fds & open_fds,
1473 "Some fds were left open")
1474 self.assertIn(1, remaining_fds, "Subprocess failed")
1475
Gregory P. Smith8facece2012-01-21 14:01:08 -08001476 # Keep some of the fd's we opened open in the subprocess.
1477 # This tests _posixsubprocess.c's proper handling of fds_to_keep.
1478 fds_to_keep = set(open_fds.pop() for _ in range(8))
1479 p = subprocess.Popen([sys.executable, fd_status],
1480 stdout=subprocess.PIPE, close_fds=True,
1481 pass_fds=())
1482 output, ignored = p.communicate()
1483 remaining_fds = set(map(int, output.split(b',')))
1484
1485 self.assertFalse(remaining_fds & fds_to_keep & open_fds,
1486 "Some fds not in pass_fds were left open")
1487 self.assertIn(1, remaining_fds, "Subprocess failed")
1488
Victor Stinner88701e22011-06-01 13:13:04 +02001489 # Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file
1490 # descriptor of a pipe closed in the parent process is valid in the
1491 # child process according to fstat(), but the mode of the file
1492 # descriptor is invalid, and read or write raise an error.
1493 @support.requires_mac_ver(10, 5)
Gregory P. Smith8edd99d2010-12-14 13:43:30 +00001494 def test_pass_fds(self):
1495 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1496
1497 open_fds = set()
1498
1499 for x in range(5):
1500 fds = os.pipe()
1501 self.addCleanup(os.close, fds[0])
1502 self.addCleanup(os.close, fds[1])
1503 open_fds.update(fds)
1504
1505 for fd in open_fds:
1506 p = subprocess.Popen([sys.executable, fd_status],
1507 stdout=subprocess.PIPE, close_fds=True,
1508 pass_fds=(fd, ))
1509 output, ignored = p.communicate()
1510
1511 remaining_fds = set(map(int, output.split(b',')))
1512 to_be_closed = open_fds - {fd}
1513
1514 self.assertIn(fd, remaining_fds, "fd to be passed not passed")
1515 self.assertFalse(remaining_fds & to_be_closed,
1516 "fd to be closed passed")
1517
1518 # pass_fds overrides close_fds with a warning.
1519 with self.assertWarns(RuntimeWarning) as context:
1520 self.assertFalse(subprocess.call(
1521 [sys.executable, "-c", "import sys; sys.exit(0)"],
1522 close_fds=False, pass_fds=(fd, )))
1523 self.assertIn('overriding close_fds', str(context.warning))
1524
Gregory P. Smithe14e9c22011-03-15 14:55:17 -04001525 def test_stdout_stdin_are_single_inout_fd(self):
1526 with io.open(os.devnull, "r+") as inout:
1527 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1528 stdout=inout, stdin=inout)
1529 p.wait()
1530
1531 def test_stdout_stderr_are_single_inout_fd(self):
1532 with io.open(os.devnull, "r+") as inout:
1533 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1534 stdout=inout, stderr=inout)
1535 p.wait()
1536
1537 def test_stderr_stdin_are_single_inout_fd(self):
1538 with io.open(os.devnull, "r+") as inout:
1539 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1540 stderr=inout, stdin=inout)
1541 p.wait()
1542
Gregory P. Smithe85db2b2010-12-14 14:38:00 +00001543 def test_wait_when_sigchild_ignored(self):
1544 # NOTE: sigchild_ignore.py may not be an effective test on all OSes.
1545 sigchild_ignore = support.findfile("sigchild_ignore.py",
1546 subdir="subprocessdata")
1547 p = subprocess.Popen([sys.executable, sigchild_ignore],
1548 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1549 stdout, stderr = p.communicate()
1550 self.assertEqual(0, p.returncode, "sigchild_ignore.py exited"
Gregory P. Smitha80f4fb2010-12-14 15:23:02 +00001551 " non-zero with this error:\n%s" %
1552 stderr.decode('utf8'))
Gregory P. Smithe85db2b2010-12-14 14:38:00 +00001553
Antoine Pitrou7b98d022011-03-19 17:04:13 +01001554 def test_select_unbuffered(self):
1555 # Issue #11459: bufsize=0 should really set the pipes as
1556 # unbuffered (and therefore let select() work properly).
1557 select = support.import_module("select")
1558 p = subprocess.Popen([sys.executable, "-c",
1559 'import sys;'
1560 'sys.stdout.write("apple")'],
1561 stdout=subprocess.PIPE,
1562 bufsize=0)
1563 f = p.stdout
Ross Lagerwall17ace7a2011-03-26 21:21:46 +02001564 self.addCleanup(f.close)
Antoine Pitrou7b98d022011-03-19 17:04:13 +01001565 try:
1566 self.assertEqual(f.read(4), b"appl")
1567 self.assertIn(f, select.select([f], [], [], 0.0)[0])
1568 finally:
1569 p.wait()
1570
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001571 def test_zombie_fast_process_del(self):
1572 # Issue #12650: on Unix, if Popen.__del__() was called before the
1573 # process exited, it wouldn't be added to subprocess._active, and would
1574 # remain a zombie.
1575 # spawn a Popen, and delete its reference before it exits
1576 p = subprocess.Popen([sys.executable, "-c",
1577 'import sys, time;'
1578 'time.sleep(0.2)'],
1579 stdout=subprocess.PIPE,
1580 stderr=subprocess.PIPE)
Nadeem Vawda0d7cda32011-08-19 05:12:01 +02001581 self.addCleanup(p.stdout.close)
1582 self.addCleanup(p.stderr.close)
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001583 ident = id(p)
1584 pid = p.pid
1585 del p
1586 # check that p is in the active processes list
1587 self.assertIn(ident, [id(o) for o in subprocess._active])
1588
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001589 def test_leak_fast_process_del_killed(self):
1590 # Issue #12650: on Unix, if Popen.__del__() was called before the
1591 # process exited, and the process got killed by a signal, it would never
1592 # be removed from subprocess._active, which triggered a FD and memory
1593 # leak.
1594 # spawn a Popen, delete its reference and kill it
1595 p = subprocess.Popen([sys.executable, "-c",
1596 'import time;'
1597 'time.sleep(3)'],
1598 stdout=subprocess.PIPE,
1599 stderr=subprocess.PIPE)
Nadeem Vawda0d7cda32011-08-19 05:12:01 +02001600 self.addCleanup(p.stdout.close)
1601 self.addCleanup(p.stderr.close)
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001602 ident = id(p)
1603 pid = p.pid
1604 del p
1605 os.kill(pid, signal.SIGKILL)
1606 # check that p is in the active processes list
1607 self.assertIn(ident, [id(o) for o in subprocess._active])
1608
1609 # let some time for the process to exit, and create a new Popen: this
1610 # should trigger the wait() of p
1611 time.sleep(0.2)
1612 with self.assertRaises(EnvironmentError) as c:
1613 with subprocess.Popen(['nonexisting_i_hope'],
1614 stdout=subprocess.PIPE,
1615 stderr=subprocess.PIPE) as proc:
1616 pass
1617 # p should have been wait()ed on, and removed from the _active list
1618 self.assertRaises(OSError, os.waitpid, pid, 0)
1619 self.assertNotIn(ident, [id(o) for o in subprocess._active])
1620
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001621
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001622@unittest.skipUnless(mswindows, "Windows specific tests")
Florent Xiclunac049d872010-03-27 22:47:23 +00001623class Win32ProcessTestCase(BaseTestCase):
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001624
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001625 def test_startupinfo(self):
1626 # startupinfo argument
1627 # We uses hardcoded constants, because we do not want to
1628 # depend on win32all.
1629 STARTF_USESHOWWINDOW = 1
1630 SW_MAXIMIZE = 3
1631 startupinfo = subprocess.STARTUPINFO()
1632 startupinfo.dwFlags = STARTF_USESHOWWINDOW
1633 startupinfo.wShowWindow = SW_MAXIMIZE
1634 # Since Python is a console process, it won't be affected
1635 # by wShowWindow, but the argument should be silently
1636 # ignored
1637 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001638 startupinfo=startupinfo)
1639
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001640 def test_creationflags(self):
1641 # creationflags argument
1642 CREATE_NEW_CONSOLE = 16
1643 sys.stderr.write(" a DOS box should flash briefly ...\n")
1644 subprocess.call(sys.executable +
1645 ' -c "import time; time.sleep(0.25)"',
1646 creationflags=CREATE_NEW_CONSOLE)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001647
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001648 def test_invalid_args(self):
1649 # invalid arguments should raise ValueError
1650 self.assertRaises(ValueError, subprocess.call,
1651 [sys.executable, "-c",
1652 "import sys; sys.exit(47)"],
1653 preexec_fn=lambda: 1)
1654 self.assertRaises(ValueError, subprocess.call,
1655 [sys.executable, "-c",
1656 "import sys; sys.exit(47)"],
1657 stdout=subprocess.PIPE,
1658 close_fds=True)
1659
1660 def test_close_fds(self):
1661 # close file descriptors
1662 rc = subprocess.call([sys.executable, "-c",
1663 "import sys; sys.exit(47)"],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001664 close_fds=True)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001665 self.assertEqual(rc, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001666
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001667 def test_shell_sequence(self):
1668 # Run command through the shell (sequence)
1669 newenv = os.environ.copy()
1670 newenv["FRUIT"] = "physalis"
1671 p = subprocess.Popen(["set"], shell=1,
1672 stdout=subprocess.PIPE,
1673 env=newenv)
Brian Curtin19a53792010-11-05 17:09:05 +00001674 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001675 self.assertIn(b"physalis", p.stdout.read())
Guido van Rossume7ba4952007-06-06 23:52:48 +00001676
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001677 def test_shell_string(self):
1678 # Run command through the shell (string)
1679 newenv = os.environ.copy()
1680 newenv["FRUIT"] = "physalis"
1681 p = subprocess.Popen("set", shell=1,
1682 stdout=subprocess.PIPE,
1683 env=newenv)
Brian Curtin19a53792010-11-05 17:09:05 +00001684 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001685 self.assertIn(b"physalis", p.stdout.read())
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001686
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001687 def test_call_string(self):
1688 # call() function with string argument on Windows
1689 rc = subprocess.call(sys.executable +
1690 ' -c "import sys; sys.exit(47)"')
1691 self.assertEqual(rc, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001692
Florent Xicluna4886d242010-03-08 13:27:26 +00001693 def _kill_process(self, method, *args):
1694 # Some win32 buildbot raises EOFError if stdin is inherited
Antoine Pitroua4024e22010-09-24 18:57:01 +00001695 p = subprocess.Popen([sys.executable, "-c", """if 1:
1696 import sys, time
1697 sys.stdout.write('x\\n')
1698 sys.stdout.flush()
1699 time.sleep(30)
1700 """],
1701 stdin=subprocess.PIPE,
1702 stdout=subprocess.PIPE,
1703 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +00001704 self.addCleanup(p.stdout.close)
1705 self.addCleanup(p.stderr.close)
1706 self.addCleanup(p.stdin.close)
Antoine Pitroua4024e22010-09-24 18:57:01 +00001707 # Wait for the interpreter to be completely initialized before
1708 # sending any signal.
1709 p.stdout.read(1)
1710 getattr(p, method)(*args)
Florent Xiclunac049d872010-03-27 22:47:23 +00001711 _, stderr = p.communicate()
1712 self.assertStderrEqual(stderr, b'')
Antoine Pitroua4024e22010-09-24 18:57:01 +00001713 returncode = p.wait()
Florent Xicluna4886d242010-03-08 13:27:26 +00001714 self.assertNotEqual(returncode, 0)
1715
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01001716 def _kill_dead_process(self, method, *args):
1717 p = subprocess.Popen([sys.executable, "-c", """if 1:
1718 import sys, time
1719 sys.stdout.write('x\\n')
1720 sys.stdout.flush()
1721 sys.exit(42)
1722 """],
1723 stdin=subprocess.PIPE,
1724 stdout=subprocess.PIPE,
1725 stderr=subprocess.PIPE)
1726 self.addCleanup(p.stdout.close)
1727 self.addCleanup(p.stderr.close)
1728 self.addCleanup(p.stdin.close)
1729 # Wait for the interpreter to be completely initialized before
1730 # sending any signal.
1731 p.stdout.read(1)
1732 # The process should end after this
1733 time.sleep(1)
1734 # This shouldn't raise even though the child is now dead
1735 getattr(p, method)(*args)
1736 _, stderr = p.communicate()
1737 self.assertStderrEqual(stderr, b'')
1738 rc = p.wait()
1739 self.assertEqual(rc, 42)
1740
Florent Xicluna4886d242010-03-08 13:27:26 +00001741 def test_send_signal(self):
1742 self._kill_process('send_signal', signal.SIGTERM)
Christian Heimesa342c012008-04-20 21:01:16 +00001743
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001744 def test_kill(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001745 self._kill_process('kill')
Christian Heimesa342c012008-04-20 21:01:16 +00001746
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001747 def test_terminate(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001748 self._kill_process('terminate')
Christian Heimesa342c012008-04-20 21:01:16 +00001749
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01001750 def test_send_signal_dead(self):
1751 self._kill_dead_process('send_signal', signal.SIGTERM)
1752
1753 def test_kill_dead(self):
1754 self._kill_dead_process('kill')
1755
1756 def test_terminate_dead(self):
1757 self._kill_dead_process('terminate')
1758
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001759
Brett Cannona23810f2008-05-26 19:04:21 +00001760# The module says:
1761# "NB This only works (and is only relevant) for UNIX."
1762#
1763# Actually, getoutput should work on any platform with an os.popen, but
1764# I'll take the comment as given, and skip this suite.
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001765@unittest.skipUnless(os.name == 'posix', "only relevant for UNIX")
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001766class CommandTests(unittest.TestCase):
1767 def test_getoutput(self):
1768 self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy')
1769 self.assertEqual(subprocess.getstatusoutput('echo xyzzy'),
1770 (0, 'xyzzy'))
Brett Cannona23810f2008-05-26 19:04:21 +00001771
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001772 # we use mkdtemp in the next line to create an empty directory
1773 # under our exclusive control; from that, we can invent a pathname
1774 # that we _know_ won't exist. This is guaranteed to fail.
1775 dir = None
1776 try:
1777 dir = tempfile.mkdtemp()
1778 name = os.path.join(dir, "foo")
Brett Cannona23810f2008-05-26 19:04:21 +00001779
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001780 status, output = subprocess.getstatusoutput('cat ' + name)
1781 self.assertNotEqual(status, 0)
1782 finally:
1783 if dir is not None:
1784 os.rmdir(dir)
Brett Cannona23810f2008-05-26 19:04:21 +00001785
Gregory P. Smithd06fa472009-07-04 02:46:54 +00001786
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001787@unittest.skipUnless(getattr(subprocess, '_has_poll', False),
1788 "poll system call not supported")
1789class ProcessTestCaseNoPoll(ProcessTestCase):
1790 def setUp(self):
1791 subprocess._has_poll = False
1792 ProcessTestCase.setUp(self)
Gregory P. Smithd06fa472009-07-04 02:46:54 +00001793
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001794 def tearDown(self):
1795 subprocess._has_poll = True
1796 ProcessTestCase.tearDown(self)
Gregory P. Smithd06fa472009-07-04 02:46:54 +00001797
1798
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001799@unittest.skipUnless(getattr(subprocess, '_posixsubprocess', False),
1800 "_posixsubprocess extension module not found.")
1801class ProcessTestCasePOSIXPurePython(ProcessTestCase, POSIXProcessTestCase):
Gregory P. Smith7439e7b2011-05-28 09:06:02 -07001802 @classmethod
1803 def setUpClass(cls):
1804 global subprocess
1805 assert subprocess._posixsubprocess
1806 # Reimport subprocess while forcing _posixsubprocess to not exist.
1807 with support.check_warnings(('.*_posixsubprocess .* not being used.*',
1808 RuntimeWarning)):
1809 subprocess = support.import_fresh_module(
1810 'subprocess', blocked=['_posixsubprocess'])
1811 assert not subprocess._posixsubprocess
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001812
Gregory P. Smith7439e7b2011-05-28 09:06:02 -07001813 @classmethod
1814 def tearDownClass(cls):
1815 global subprocess
1816 # Reimport subprocess as it should be, restoring order to the universe.
1817 subprocess = support.import_fresh_module('subprocess')
1818 assert subprocess._posixsubprocess
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001819
1820
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00001821class HelperFunctionTests(unittest.TestCase):
Gregory P. Smithaf6d3b82010-03-01 02:56:44 +00001822 @unittest.skipIf(mswindows, "errno and EINTR make no sense on windows")
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00001823 def test_eintr_retry_call(self):
1824 record_calls = []
1825 def fake_os_func(*args):
1826 record_calls.append(args)
1827 if len(record_calls) == 2:
1828 raise OSError(errno.EINTR, "fake interrupted system call")
1829 return tuple(reversed(args))
1830
1831 self.assertEqual((999, 256),
1832 subprocess._eintr_retry_call(fake_os_func, 256, 999))
1833 self.assertEqual([(256, 999)], record_calls)
1834 # This time there will be an EINTR so it will loop once.
1835 self.assertEqual((666,),
1836 subprocess._eintr_retry_call(fake_os_func, 666))
1837 self.assertEqual([(256, 999), (666,), (666,)], record_calls)
1838
1839
Tim Golden126c2962010-08-11 14:20:40 +00001840@unittest.skipUnless(mswindows, "Windows-specific tests")
1841class CommandsWithSpaces (BaseTestCase):
1842
1843 def setUp(self):
1844 super().setUp()
1845 f, fname = mkstemp(".py", "te st")
1846 self.fname = fname.lower ()
1847 os.write(f, b"import sys;"
1848 b"sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))"
1849 )
1850 os.close(f)
1851
1852 def tearDown(self):
1853 os.remove(self.fname)
1854 super().tearDown()
1855
1856 def with_spaces(self, *args, **kwargs):
1857 kwargs['stdout'] = subprocess.PIPE
1858 p = subprocess.Popen(*args, **kwargs)
Brian Curtin19a53792010-11-05 17:09:05 +00001859 self.addCleanup(p.stdout.close)
Tim Golden126c2962010-08-11 14:20:40 +00001860 self.assertEqual(
1861 p.stdout.read ().decode("mbcs"),
1862 "2 [%r, 'ab cd']" % self.fname
1863 )
1864
1865 def test_shell_string_with_spaces(self):
1866 # call() function with string argument with spaces on Windows
Brian Curtind835cf12010-08-13 20:42:57 +00001867 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
1868 "ab cd"), shell=1)
Tim Golden126c2962010-08-11 14:20:40 +00001869
1870 def test_shell_sequence_with_spaces(self):
1871 # call() function with sequence argument with spaces on Windows
Brian Curtind835cf12010-08-13 20:42:57 +00001872 self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1)
Tim Golden126c2962010-08-11 14:20:40 +00001873
1874 def test_noshell_string_with_spaces(self):
1875 # call() function with string argument with spaces on Windows
1876 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
1877 "ab cd"))
1878
1879 def test_noshell_sequence_with_spaces(self):
1880 # call() function with sequence argument with spaces on Windows
1881 self.with_spaces([sys.executable, self.fname, "ab cd"])
1882
Brian Curtin79cdb662010-12-03 02:46:02 +00001883
Georg Brandla86b2622012-02-20 21:34:57 +01001884class ContextManagerTests(BaseTestCase):
Brian Curtin79cdb662010-12-03 02:46:02 +00001885
1886 def test_pipe(self):
1887 with subprocess.Popen([sys.executable, "-c",
1888 "import sys;"
1889 "sys.stdout.write('stdout');"
1890 "sys.stderr.write('stderr');"],
1891 stdout=subprocess.PIPE,
1892 stderr=subprocess.PIPE) as proc:
1893 self.assertEqual(proc.stdout.read(), b"stdout")
1894 self.assertStderrEqual(proc.stderr.read(), b"stderr")
1895
1896 self.assertTrue(proc.stdout.closed)
1897 self.assertTrue(proc.stderr.closed)
1898
1899 def test_returncode(self):
1900 with subprocess.Popen([sys.executable, "-c",
1901 "import sys; sys.exit(100)"]) as proc:
Gregory P. Smithc9557af2011-05-11 22:18:23 -07001902 pass
1903 # __exit__ calls wait(), so the returncode should be set
Brian Curtin79cdb662010-12-03 02:46:02 +00001904 self.assertEqual(proc.returncode, 100)
1905
1906 def test_communicate_stdin(self):
1907 with subprocess.Popen([sys.executable, "-c",
1908 "import sys;"
1909 "sys.exit(sys.stdin.read() == 'context')"],
1910 stdin=subprocess.PIPE) as proc:
1911 proc.communicate(b"context")
1912 self.assertEqual(proc.returncode, 1)
1913
1914 def test_invalid_args(self):
1915 with self.assertRaises(EnvironmentError) as c:
1916 with subprocess.Popen(['nonexisting_i_hope'],
1917 stdout=subprocess.PIPE,
1918 stderr=subprocess.PIPE) as proc:
1919 pass
1920
1921 if c.exception.errno != errno.ENOENT: # ignore "no such file"
1922 raise c.exception
1923
1924
Gregory P. Smith961e0e82011-03-15 15:43:39 -04001925def test_main():
1926 unit_tests = (ProcessTestCase,
1927 POSIXProcessTestCase,
1928 Win32ProcessTestCase,
1929 ProcessTestCasePOSIXPurePython,
1930 CommandTests,
1931 ProcessTestCaseNoPoll,
1932 HelperFunctionTests,
1933 CommandsWithSpaces,
Antoine Pitrouab85ff32011-07-23 22:03:45 +02001934 ContextManagerTests,
1935 )
Gregory P. Smith961e0e82011-03-15 15:43:39 -04001936
1937 support.run_unittest(*unit_tests)
1938 support.reap_children()
1939
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001940if __name__ == "__main__":
Gregory P. Smithe14e9c22011-03-15 14:55:17 -04001941 unittest.main()