blob: b871359d7b2e4f35e11ff850d5161e73d516d2d2 [file] [log] [blame]
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001import unittest
Benjamin Petersonee8712c2008-05-20 21:35:26 +00002from test import support
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00003import subprocess
4import sys
5import signal
Gregory P. Smith112bb3a2011-03-15 14:55:17 -04006import io
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00007import os
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00008import errno
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00009import tempfile
10import time
Tim Peters3761e8d2004-10-13 04:07:12 +000011import re
Ezio Melotti184bdfb2010-02-18 09:37:05 +000012import sysconfig
Gregory P. Smithd23047b2010-12-04 09:10:44 +000013import warnings
Gregory P. Smith51ee2702010-12-13 07:59:39 +000014import select
Gregory P. Smith81ce6852011-03-15 02:04:11 -040015import shutil
Gregory P. Smith32ec9da2010-03-19 16:53:08 +000016try:
17 import gc
18except ImportError:
19 gc = None
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000020
21mswindows = (sys.platform == "win32")
22
23#
24# Depends on the following external programs: Python
25#
26
27if mswindows:
Tim Peters3b01a702004-10-12 22:19:32 +000028 SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), '
29 'os.O_BINARY);')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000030else:
31 SETBINARY = ''
32
Florent Xiclunab1e94e82010-02-27 22:12:37 +000033
34try:
35 mkstemp = tempfile.mkstemp
36except AttributeError:
37 # tempfile.mkstemp is not available
38 def mkstemp():
39 """Replacement for mkstemp, calling mktemp."""
40 fname = tempfile.mktemp()
41 return os.open(fname, os.O_RDWR|os.O_CREAT), fname
42
Tim Peters3761e8d2004-10-13 04:07:12 +000043
Florent Xiclunac049d872010-03-27 22:47:23 +000044class BaseTestCase(unittest.TestCase):
Thomas Wouters0e3f5912006-08-11 14:57:12 +000045 def setUp(self):
46 # Try to minimize the number of children we have so this test
47 # doesn't crash on some buildbots (Alphas in particular).
Florent Xiclunab1e94e82010-02-27 22:12:37 +000048 support.reap_children()
Thomas Wouters0e3f5912006-08-11 14:57:12 +000049
Florent Xiclunaf0cbd822010-03-04 21:50:56 +000050 def tearDown(self):
51 for inst in subprocess._active:
52 inst.wait()
53 subprocess._cleanup()
54 self.assertFalse(subprocess._active, "subprocess._active not empty")
55
Florent Xiclunab1e94e82010-02-27 22:12:37 +000056 def assertStderrEqual(self, stderr, expected, msg=None):
57 # In a debug build, stuff like "[6580 refs]" is printed to stderr at
58 # shutdown time. That frustrates tests trying to check stderr produced
59 # from a spawned Python process.
Antoine Pitrou62f68ed2010-08-04 11:48:56 +000060 actual = support.strip_python_stderr(stderr)
Reid Kleckner31aa7dd2011-03-14 12:02:10 -040061 # strip_python_stderr also strips whitespace, so we do too.
62 expected = expected.strip()
Florent Xiclunab1e94e82010-02-27 22:12:37 +000063 self.assertEqual(actual, expected, msg)
Thomas Wouters0e3f5912006-08-11 14:57:12 +000064
Florent Xiclunac049d872010-03-27 22:47:23 +000065
66class ProcessTestCase(BaseTestCase):
67
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000068 def test_call_seq(self):
Tim Peters7b759da2004-10-12 22:29:54 +000069 # call() function with sequence argument
Tim Peters3b01a702004-10-12 22:19:32 +000070 rc = subprocess.call([sys.executable, "-c",
71 "import sys; sys.exit(47)"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000072 self.assertEqual(rc, 47)
73
Reid Kleckner31aa7dd2011-03-14 12:02:10 -040074 def test_call_timeout(self):
75 # call() function with timeout argument; we want to test that the child
76 # process gets killed when the timeout expires. If the child isn't
77 # killed, this call will deadlock since subprocess.call waits for the
78 # child.
79 self.assertRaises(subprocess.TimeoutExpired, subprocess.call,
80 [sys.executable, "-c", "while True: pass"],
81 timeout=0.1)
82
Peter Astrand454f7672005-01-01 09:36:35 +000083 def test_check_call_zero(self):
84 # check_call() function with zero return code
85 rc = subprocess.check_call([sys.executable, "-c",
86 "import sys; sys.exit(0)"])
87 self.assertEqual(rc, 0)
88
89 def test_check_call_nonzero(self):
90 # check_call() function with non-zero return code
Florent Xiclunab1e94e82010-02-27 22:12:37 +000091 with self.assertRaises(subprocess.CalledProcessError) as c:
Peter Astrand454f7672005-01-01 09:36:35 +000092 subprocess.check_call([sys.executable, "-c",
93 "import sys; sys.exit(47)"])
Florent Xiclunab1e94e82010-02-27 22:12:37 +000094 self.assertEqual(c.exception.returncode, 47)
Peter Astrand454f7672005-01-01 09:36:35 +000095
Georg Brandlf9734072008-12-07 15:30:06 +000096 def test_check_output(self):
97 # check_output() function with zero return code
98 output = subprocess.check_output(
99 [sys.executable, "-c", "print('BDFL')"])
Benjamin Peterson577473f2010-01-19 00:09:57 +0000100 self.assertIn(b'BDFL', output)
Georg Brandlf9734072008-12-07 15:30:06 +0000101
102 def test_check_output_nonzero(self):
103 # check_call() function with non-zero return code
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000104 with self.assertRaises(subprocess.CalledProcessError) as c:
Georg Brandlf9734072008-12-07 15:30:06 +0000105 subprocess.check_output(
106 [sys.executable, "-c", "import sys; sys.exit(5)"])
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000107 self.assertEqual(c.exception.returncode, 5)
Georg Brandlf9734072008-12-07 15:30:06 +0000108
109 def test_check_output_stderr(self):
110 # check_output() function stderr redirected to stdout
111 output = subprocess.check_output(
112 [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"],
113 stderr=subprocess.STDOUT)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000114 self.assertIn(b'BDFL', output)
Georg Brandlf9734072008-12-07 15:30:06 +0000115
116 def test_check_output_stdout_arg(self):
117 # check_output() function stderr redirected to stdout
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000118 with self.assertRaises(ValueError) as c:
Georg Brandlf9734072008-12-07 15:30:06 +0000119 output = subprocess.check_output(
120 [sys.executable, "-c", "print('will not be run')"],
121 stdout=sys.stdout)
Georg Brandlf9734072008-12-07 15:30:06 +0000122 self.fail("Expected ValueError when stdout arg supplied.")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000123 self.assertIn('stdout', c.exception.args[0])
Georg Brandlf9734072008-12-07 15:30:06 +0000124
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400125 def test_check_output_timeout(self):
126 # check_output() function with timeout arg
127 with self.assertRaises(subprocess.TimeoutExpired) as c:
128 output = subprocess.check_output(
129 [sys.executable, "-c",
Victor Stinner149b1c72011-06-06 23:43:02 +0200130 "import sys, time\n"
131 "sys.stdout.write('BDFL')\n"
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400132 "sys.stdout.flush()\n"
Victor Stinner149b1c72011-06-06 23:43:02 +0200133 "time.sleep(3600)"],
Reid Klecknerda9ac722011-03-16 17:08:21 -0400134 # Some heavily loaded buildbots (sparc Debian 3.x) require
135 # this much time to start and print.
136 timeout=3)
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400137 self.fail("Expected TimeoutExpired.")
138 self.assertEqual(c.exception.output, b'BDFL')
139
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000140 def test_call_kwargs(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000141 # call() function with keyword args
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000142 newenv = os.environ.copy()
143 newenv["FRUIT"] = "banana"
144 rc = subprocess.call([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000145 'import sys, os;'
146 'sys.exit(os.getenv("FRUIT")=="banana")'],
147 env=newenv)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000148 self.assertEqual(rc, 1)
149
Victor Stinner87b9bc32011-06-01 00:57:47 +0200150 def test_invalid_args(self):
151 # Popen() called with invalid arguments should raise TypeError
152 # but Popen.__del__ should not complain (issue #12085)
153 with support.captured_stderr() as s:
154 self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1)
155 argcount = subprocess.Popen.__init__.__code__.co_argcount
156 too_many_args = [0] * (argcount + 1)
157 self.assertRaises(TypeError, subprocess.Popen, *too_many_args)
158 self.assertEqual(s.getvalue(), '')
159
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000160 def test_stdin_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000161 # .stdin is None when not redirected
Georg Brandl88fc6642007-02-09 21:28:07 +0000162 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000163 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000164 self.addCleanup(p.stdout.close)
165 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000166 p.wait()
167 self.assertEqual(p.stdin, None)
168
169 def test_stdout_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000170 # .stdout is None when not redirected
Tim Peters29b6b4f2004-10-13 03:43:40 +0000171 p = subprocess.Popen([sys.executable, "-c",
Georg Brandl88fc6642007-02-09 21:28:07 +0000172 'print(" this bit of output is from a '
Tim Peters4052fe52004-10-13 03:29:54 +0000173 'test of stdout in a different '
Georg Brandl88fc6642007-02-09 21:28:07 +0000174 'process ...")'],
Tim Peters4052fe52004-10-13 03:29:54 +0000175 stdin=subprocess.PIPE, stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000176 self.addCleanup(p.stdin.close)
177 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000178 p.wait()
179 self.assertEqual(p.stdout, None)
180
181 def test_stderr_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000182 # .stderr is None when not redirected
Georg Brandl88fc6642007-02-09 21:28:07 +0000183 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000184 stdin=subprocess.PIPE, stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000185 self.addCleanup(p.stdout.close)
186 self.addCleanup(p.stdin.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000187 p.wait()
188 self.assertEqual(p.stderr, None)
189
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000190 def test_executable_with_cwd(self):
Florent Xicluna1d1ab972010-03-11 01:53:10 +0000191 python_dir = os.path.dirname(os.path.realpath(sys.executable))
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000192 p = subprocess.Popen(["somethingyoudonthave", "-c",
193 "import sys; sys.exit(47)"],
194 executable=sys.executable, cwd=python_dir)
195 p.wait()
196 self.assertEqual(p.returncode, 47)
197
198 @unittest.skipIf(sysconfig.is_python_build(),
199 "need an installed Python. See #7774")
200 def test_executable_without_cwd(self):
201 # For a normal installation, it should work without 'cwd'
202 # argument. For test runs in the build directory, see #7774.
203 p = subprocess.Popen(["somethingyoudonthave", "-c",
204 "import sys; sys.exit(47)"],
Tim Peters3b01a702004-10-12 22:19:32 +0000205 executable=sys.executable)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000206 p.wait()
207 self.assertEqual(p.returncode, 47)
208
209 def test_stdin_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000210 # stdin redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000211 p = subprocess.Popen([sys.executable, "-c",
212 'import sys; sys.exit(sys.stdin.read() == "pear")'],
213 stdin=subprocess.PIPE)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000214 p.stdin.write(b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000215 p.stdin.close()
216 p.wait()
217 self.assertEqual(p.returncode, 1)
218
219 def test_stdin_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000220 # stdin is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000221 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000222 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000223 d = tf.fileno()
Antoine Pitrou9cadb1b2008-09-15 23:02:56 +0000224 os.write(d, b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000225 os.lseek(d, 0, 0)
226 p = subprocess.Popen([sys.executable, "-c",
227 'import sys; sys.exit(sys.stdin.read() == "pear")'],
228 stdin=d)
229 p.wait()
230 self.assertEqual(p.returncode, 1)
231
232 def test_stdin_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000233 # stdin is set to open file object
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000234 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000235 self.addCleanup(tf.close)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000236 tf.write(b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000237 tf.seek(0)
238 p = subprocess.Popen([sys.executable, "-c",
239 'import sys; sys.exit(sys.stdin.read() == "pear")'],
240 stdin=tf)
241 p.wait()
242 self.assertEqual(p.returncode, 1)
243
244 def test_stdout_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000245 # stdout redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000246 p = subprocess.Popen([sys.executable, "-c",
247 'import sys; sys.stdout.write("orange")'],
248 stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000249 self.addCleanup(p.stdout.close)
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000250 self.assertEqual(p.stdout.read(), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000251
252 def test_stdout_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000253 # stdout is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000254 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000255 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000256 d = tf.fileno()
257 p = subprocess.Popen([sys.executable, "-c",
258 'import sys; sys.stdout.write("orange")'],
259 stdout=d)
260 p.wait()
261 os.lseek(d, 0, 0)
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000262 self.assertEqual(os.read(d, 1024), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000263
264 def test_stdout_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000265 # stdout is set to open file object
Tim Peterse718f612004-10-12 21:51:32 +0000266 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000267 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000268 p = subprocess.Popen([sys.executable, "-c",
269 'import sys; sys.stdout.write("orange")'],
270 stdout=tf)
271 p.wait()
272 tf.seek(0)
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000273 self.assertEqual(tf.read(), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000274
275 def test_stderr_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000276 # stderr redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000277 p = subprocess.Popen([sys.executable, "-c",
278 'import sys; sys.stderr.write("strawberry")'],
279 stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000280 self.addCleanup(p.stderr.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000281 self.assertStderrEqual(p.stderr.read(), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000282
283 def test_stderr_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000284 # stderr is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000285 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000286 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000287 d = tf.fileno()
288 p = subprocess.Popen([sys.executable, "-c",
289 'import sys; sys.stderr.write("strawberry")'],
290 stderr=d)
291 p.wait()
292 os.lseek(d, 0, 0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000293 self.assertStderrEqual(os.read(d, 1024), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000294
295 def test_stderr_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000296 # stderr is set to open file object
Tim Peterse718f612004-10-12 21:51:32 +0000297 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000298 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000299 p = subprocess.Popen([sys.executable, "-c",
300 'import sys; sys.stderr.write("strawberry")'],
301 stderr=tf)
302 p.wait()
303 tf.seek(0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000304 self.assertStderrEqual(tf.read(), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000305
306 def test_stdout_stderr_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000307 # capture stdout and stderr to the same pipe
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000308 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000309 'import sys;'
310 'sys.stdout.write("apple");'
311 'sys.stdout.flush();'
312 'sys.stderr.write("orange")'],
313 stdout=subprocess.PIPE,
314 stderr=subprocess.STDOUT)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000315 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000316 self.assertStderrEqual(p.stdout.read(), b"appleorange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000317
318 def test_stdout_stderr_file(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000319 # capture stdout and stderr to the same open file
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000320 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000321 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000322 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000323 'import sys;'
324 'sys.stdout.write("apple");'
325 'sys.stdout.flush();'
326 'sys.stderr.write("orange")'],
327 stdout=tf,
328 stderr=tf)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000329 p.wait()
330 tf.seek(0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000331 self.assertStderrEqual(tf.read(), b"appleorange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000332
Thomas Wouters89f507f2006-12-13 04:49:30 +0000333 def test_stdout_filedes_of_stdout(self):
334 # stdout is set to 1 (#1531862).
Antoine Pitrou9cadb1b2008-09-15 23:02:56 +0000335 cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), b'.\n'))"
Thomas Wouters89f507f2006-12-13 04:49:30 +0000336 rc = subprocess.call([sys.executable, "-c", cmd], stdout=1)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000337 self.assertEqual(rc, 2)
Thomas Wouters89f507f2006-12-13 04:49:30 +0000338
Ross Lagerwallba102ec2011-03-16 18:40:25 +0200339 def test_stdout_devnull(self):
340 p = subprocess.Popen([sys.executable, "-c",
341 'for i in range(10240):'
342 'print("x" * 1024)'],
343 stdout=subprocess.DEVNULL)
344 p.wait()
345 self.assertEqual(p.stdout, None)
346
347 def test_stderr_devnull(self):
348 p = subprocess.Popen([sys.executable, "-c",
349 'import sys\n'
350 'for i in range(10240):'
351 'sys.stderr.write("x" * 1024)'],
352 stderr=subprocess.DEVNULL)
353 p.wait()
354 self.assertEqual(p.stderr, None)
355
356 def test_stdin_devnull(self):
357 p = subprocess.Popen([sys.executable, "-c",
358 'import sys;'
359 'sys.stdin.read(1)'],
360 stdin=subprocess.DEVNULL)
361 p.wait()
362 self.assertEqual(p.stdin, None)
363
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000364 def test_cwd(self):
Christian Heimes5fb7c2a2007-12-24 08:52:31 +0000365 tmpdir = tempfile.gettempdir()
Peter Astrand195404f2004-11-12 15:51:48 +0000366 # We cannot use os.path.realpath to canonicalize the path,
367 # since it doesn't expand Tru64 {memb} strings. See bug 1063571.
368 cwd = os.getcwd()
369 os.chdir(tmpdir)
370 tmpdir = os.getcwd()
371 os.chdir(cwd)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000372 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000373 'import sys,os;'
374 'sys.stdout.write(os.getcwd())'],
375 stdout=subprocess.PIPE,
376 cwd=tmpdir)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000377 self.addCleanup(p.stdout.close)
Fredrik Lundh59c05592004-10-13 06:55:40 +0000378 normcase = os.path.normcase
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000379 self.assertEqual(normcase(p.stdout.read().decode("utf-8")),
380 normcase(tmpdir))
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000381
382 def test_env(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000383 newenv = os.environ.copy()
384 newenv["FRUIT"] = "orange"
Victor Stinnerf1512a22011-06-21 17:18:38 +0200385 with subprocess.Popen([sys.executable, "-c",
386 'import sys,os;'
387 'sys.stdout.write(os.getenv("FRUIT"))'],
388 stdout=subprocess.PIPE,
389 env=newenv) as p:
390 stdout, stderr = p.communicate()
391 self.assertEqual(stdout, b"orange")
392
393 def test_empty_env(self):
394 with subprocess.Popen([sys.executable, "-c",
395 'import os; '
396 'print(len(os.environ))'],
397 stdout=subprocess.PIPE,
398 env={}) as p:
399 stdout, stderr = p.communicate()
400 self.assertEqual(stdout.strip(), b"0")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000401
Peter Astrandcbac93c2005-03-03 20:24:28 +0000402 def test_communicate_stdin(self):
403 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000404 'import sys;'
405 'sys.exit(sys.stdin.read() == "pear")'],
Peter Astrandcbac93c2005-03-03 20:24:28 +0000406 stdin=subprocess.PIPE)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000407 p.communicate(b"pear")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000408 self.assertEqual(p.returncode, 1)
409
410 def test_communicate_stdout(self):
411 p = subprocess.Popen([sys.executable, "-c",
412 'import sys; sys.stdout.write("pineapple")'],
413 stdout=subprocess.PIPE)
414 (stdout, stderr) = p.communicate()
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000415 self.assertEqual(stdout, b"pineapple")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000416 self.assertEqual(stderr, None)
417
418 def test_communicate_stderr(self):
419 p = subprocess.Popen([sys.executable, "-c",
420 'import sys; sys.stderr.write("pineapple")'],
421 stderr=subprocess.PIPE)
422 (stdout, stderr) = p.communicate()
423 self.assertEqual(stdout, None)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000424 self.assertStderrEqual(stderr, b"pineapple")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000425
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000426 def test_communicate(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000427 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000428 'import sys,os;'
429 'sys.stderr.write("pineapple");'
430 'sys.stdout.write(sys.stdin.read())'],
431 stdin=subprocess.PIPE,
432 stdout=subprocess.PIPE,
433 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000434 self.addCleanup(p.stdout.close)
435 self.addCleanup(p.stderr.close)
436 self.addCleanup(p.stdin.close)
Georg Brandl1abcbf82008-07-01 19:28:43 +0000437 (stdout, stderr) = p.communicate(b"banana")
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000438 self.assertEqual(stdout, b"banana")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000439 self.assertStderrEqual(stderr, b"pineapple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000440
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400441 def test_communicate_timeout(self):
442 p = subprocess.Popen([sys.executable, "-c",
443 'import sys,os,time;'
444 'sys.stderr.write("pineapple\\n");'
445 'time.sleep(1);'
446 'sys.stderr.write("pear\\n");'
447 'sys.stdout.write(sys.stdin.read())'],
448 universal_newlines=True,
449 stdin=subprocess.PIPE,
450 stdout=subprocess.PIPE,
451 stderr=subprocess.PIPE)
452 self.assertRaises(subprocess.TimeoutExpired, p.communicate, "banana",
453 timeout=0.3)
454 # Make sure we can keep waiting for it, and that we get the whole output
455 # after it completes.
456 (stdout, stderr) = p.communicate()
457 self.assertEqual(stdout, "banana")
458 self.assertStderrEqual(stderr.encode(), b"pineapple\npear\n")
459
460 def test_communicate_timeout_large_ouput(self):
461 # Test a expring timeout while the child is outputting lots of data.
462 p = subprocess.Popen([sys.executable, "-c",
463 'import sys,os,time;'
464 'sys.stdout.write("a" * (64 * 1024));'
465 'time.sleep(0.2);'
466 'sys.stdout.write("a" * (64 * 1024));'
467 'time.sleep(0.2);'
468 'sys.stdout.write("a" * (64 * 1024));'
469 'time.sleep(0.2);'
470 'sys.stdout.write("a" * (64 * 1024));'],
471 stdout=subprocess.PIPE)
472 self.assertRaises(subprocess.TimeoutExpired, p.communicate, timeout=0.4)
473 (stdout, _) = p.communicate()
474 self.assertEqual(len(stdout), 4 * 64 * 1024)
475
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000476 # Test for the fd leak reported in http://bugs.python.org/issue2791.
477 def test_communicate_pipe_fd_leak(self):
Victor Stinner667d4b52010-12-25 22:40:32 +0000478 for stdin_pipe in (False, True):
479 for stdout_pipe in (False, True):
480 for stderr_pipe in (False, True):
481 options = {}
482 if stdin_pipe:
483 options['stdin'] = subprocess.PIPE
484 if stdout_pipe:
485 options['stdout'] = subprocess.PIPE
486 if stderr_pipe:
487 options['stderr'] = subprocess.PIPE
488 if not options:
489 continue
490 p = subprocess.Popen((sys.executable, "-c", "pass"), **options)
491 p.communicate()
492 if p.stdin is not None:
493 self.assertTrue(p.stdin.closed)
494 if p.stdout is not None:
495 self.assertTrue(p.stdout.closed)
496 if p.stderr is not None:
497 self.assertTrue(p.stderr.closed)
Georg Brandlf08a9dd2008-06-10 16:57:31 +0000498
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000499 def test_communicate_returns(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000500 # communicate() should return None if no redirection is active
Tim Peters3b01a702004-10-12 22:19:32 +0000501 p = subprocess.Popen([sys.executable, "-c",
502 "import sys; sys.exit(47)"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000503 (stdout, stderr) = p.communicate()
504 self.assertEqual(stdout, None)
505 self.assertEqual(stderr, None)
506
507 def test_communicate_pipe_buf(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000508 # communicate() with writes larger than pipe_buf
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000509 # This test will probably deadlock rather than fail, if
Tim Peterse718f612004-10-12 21:51:32 +0000510 # communicate() does not work properly.
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000511 x, y = os.pipe()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000512 os.close(x)
513 os.close(y)
514 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000515 'import sys,os;'
516 'sys.stdout.write(sys.stdin.read(47));'
Charles-François Natali2d517212011-05-29 16:36:44 +0200517 'sys.stderr.write("x" * %d);'
518 'sys.stdout.write(sys.stdin.read())' %
519 support.PIPE_MAX_SIZE],
Guido van Rossum98297ee2007-11-06 21:34:58 +0000520 stdin=subprocess.PIPE,
521 stdout=subprocess.PIPE,
522 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000523 self.addCleanup(p.stdout.close)
524 self.addCleanup(p.stderr.close)
525 self.addCleanup(p.stdin.close)
Charles-François Natali2d517212011-05-29 16:36:44 +0200526 string_to_write = b"a" * support.PIPE_MAX_SIZE
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000527 (stdout, stderr) = p.communicate(string_to_write)
528 self.assertEqual(stdout, string_to_write)
529
530 def test_writes_before_communicate(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000531 # stdin.write before communicate()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000532 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000533 'import sys,os;'
534 'sys.stdout.write(sys.stdin.read())'],
535 stdin=subprocess.PIPE,
536 stdout=subprocess.PIPE,
537 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000538 self.addCleanup(p.stdout.close)
539 self.addCleanup(p.stderr.close)
540 self.addCleanup(p.stdin.close)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000541 p.stdin.write(b"banana")
542 (stdout, stderr) = p.communicate(b"split")
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000543 self.assertEqual(stdout, b"bananasplit")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000544 self.assertStderrEqual(stderr, b"")
Tim Peterse718f612004-10-12 21:51:32 +0000545
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000546 def test_universal_newlines(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000547 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000548 'import sys,os;' + SETBINARY +
549 'sys.stdout.write("line1\\n");'
550 'sys.stdout.flush();'
551 'sys.stdout.write("line2\\n");'
552 'sys.stdout.flush();'
553 'sys.stdout.write("line3\\r\\n");'
554 'sys.stdout.flush();'
555 'sys.stdout.write("line4\\r");'
556 'sys.stdout.flush();'
557 'sys.stdout.write("\\nline5");'
558 'sys.stdout.flush();'
559 'sys.stdout.write("\\nline6");'],
560 stdout=subprocess.PIPE,
561 universal_newlines=1)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000562 self.addCleanup(p.stdout.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000563 stdout = p.stdout.read()
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000564 self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000565
566 def test_universal_newlines_communicate(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000567 # universal newlines through communicate()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000568 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000569 'import sys,os;' + SETBINARY +
570 'sys.stdout.write("line1\\n");'
571 'sys.stdout.flush();'
572 'sys.stdout.write("line2\\n");'
573 'sys.stdout.flush();'
574 'sys.stdout.write("line3\\r\\n");'
575 'sys.stdout.flush();'
576 'sys.stdout.write("line4\\r");'
577 'sys.stdout.flush();'
578 'sys.stdout.write("\\nline5");'
579 'sys.stdout.flush();'
580 'sys.stdout.write("\\nline6");'],
581 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
582 universal_newlines=1)
Brian Curtin19a53792010-11-05 17:09:05 +0000583 self.addCleanup(p.stdout.close)
584 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000585 (stdout, stderr) = p.communicate()
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000586 self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000587
588 def test_no_leaking(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000589 # Make sure we leak no resources
Antoine Pitrou8db30272010-09-18 22:38:48 +0000590 if not mswindows:
Peter Astrandf7f1bb72005-03-03 20:47:37 +0000591 max_handles = 1026 # too much for most UNIX systems
592 else:
Antoine Pitrou8db30272010-09-18 22:38:48 +0000593 max_handles = 2050 # too much for (at least some) Windows setups
594 handles = []
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400595 tmpdir = tempfile.mkdtemp()
Antoine Pitrou8db30272010-09-18 22:38:48 +0000596 try:
597 for i in range(max_handles):
598 try:
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400599 tmpfile = os.path.join(tmpdir, support.TESTFN)
600 handles.append(os.open(tmpfile, os.O_WRONLY|os.O_CREAT))
Antoine Pitrou8db30272010-09-18 22:38:48 +0000601 except OSError as e:
602 if e.errno != errno.EMFILE:
603 raise
604 break
605 else:
606 self.skipTest("failed to reach the file descriptor limit "
607 "(tried %d)" % max_handles)
608 # Close a couple of them (should be enough for a subprocess)
609 for i in range(10):
610 os.close(handles.pop())
611 # Loop creating some subprocesses. If one of them leaks some fds,
612 # the next loop iteration will fail by reaching the max fd limit.
613 for i in range(15):
614 p = subprocess.Popen([sys.executable, "-c",
615 "import sys;"
616 "sys.stdout.write(sys.stdin.read())"],
617 stdin=subprocess.PIPE,
618 stdout=subprocess.PIPE,
619 stderr=subprocess.PIPE)
620 data = p.communicate(b"lime")[0]
621 self.assertEqual(data, b"lime")
622 finally:
623 for h in handles:
624 os.close(h)
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400625 shutil.rmtree(tmpdir)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000626
627 def test_list2cmdline(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000628 self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),
629 '"a b c" d e')
630 self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),
631 'ab\\"c \\ d')
Christian Heimesfdab48e2008-01-20 09:06:41 +0000632 self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']),
633 'ab\\"c " \\\\" d')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000634 self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),
635 'a\\\\\\b "de fg" h')
636 self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),
637 'a\\\\\\"b c d')
638 self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),
639 '"a\\\\b c" d e')
640 self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),
641 '"a\\\\b\\ c" d e')
Thomas Woutersfc7bb8c2007-01-15 15:49:28 +0000642 self.assertEqual(subprocess.list2cmdline(['ab', '']),
643 'ab ""')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000644
645
646 def test_poll(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000647 p = subprocess.Popen([sys.executable,
Tim Peters29b6b4f2004-10-13 03:43:40 +0000648 "-c", "import time; time.sleep(1)"])
649 count = 0
650 while p.poll() is None:
651 time.sleep(0.1)
652 count += 1
653 # We expect that the poll loop probably went around about 10 times,
654 # but, based on system scheduling we can't control, it's possible
655 # poll() never returned None. It "should be" very rare that it
656 # didn't go around at least twice.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000657 self.assertGreaterEqual(count, 2)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000658 # Subsequent invocations should just return the returncode
659 self.assertEqual(p.poll(), 0)
660
661
662 def test_wait(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000663 p = subprocess.Popen([sys.executable,
664 "-c", "import time; time.sleep(2)"])
665 self.assertEqual(p.wait(), 0)
666 # Subsequent invocations should just return the returncode
667 self.assertEqual(p.wait(), 0)
Tim Peterse718f612004-10-12 21:51:32 +0000668
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400669 def test_wait_timeout(self):
670 p = subprocess.Popen([sys.executable,
Reid Kleckner93479cc2011-03-14 19:32:41 -0400671 "-c", "import time; time.sleep(0.1)"])
Reid Kleckner2b228f02011-03-16 16:57:54 -0400672 with self.assertRaises(subprocess.TimeoutExpired) as c:
673 p.wait(timeout=0.01)
674 self.assertIn("0.01", str(c.exception)) # For coverage of __str__.
Reid Klecknerda9ac722011-03-16 17:08:21 -0400675 # Some heavily loaded buildbots (sparc Debian 3.x) require this much
676 # time to start.
677 self.assertEqual(p.wait(timeout=3), 0)
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400678
Peter Astrand738131d2004-11-30 21:04:45 +0000679 def test_invalid_bufsize(self):
680 # an invalid type of the bufsize argument should raise
681 # TypeError.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000682 with self.assertRaises(TypeError):
Peter Astrand738131d2004-11-30 21:04:45 +0000683 subprocess.Popen([sys.executable, "-c", "pass"], "orange")
Peter Astrand738131d2004-11-30 21:04:45 +0000684
Guido van Rossum46a05a72007-06-07 21:56:45 +0000685 def test_bufsize_is_none(self):
686 # bufsize=None should be the same as bufsize=0.
687 p = subprocess.Popen([sys.executable, "-c", "pass"], None)
688 self.assertEqual(p.wait(), 0)
689 # Again with keyword arg
690 p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None)
691 self.assertEqual(p.wait(), 0)
692
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000693 def test_leaking_fds_on_error(self):
694 # see bug #5179: Popen leaks file descriptors to PIPEs if
695 # the child fails to execute; this will eventually exhaust
696 # the maximum number of open fds. 1024 seems a very common
697 # value for that limit, but Windows has 2048, so we loop
698 # 1024 times (each call leaked two fds).
699 for i in range(1024):
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000700 # Windows raises IOError. Others raise OSError.
701 with self.assertRaises(EnvironmentError) as c:
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000702 subprocess.Popen(['nonexisting_i_hope'],
703 stdout=subprocess.PIPE,
704 stderr=subprocess.PIPE)
R David Murray384069c2011-03-13 22:26:53 -0400705 # ignore errors that indicate the command was not found
R David Murray6924bd72011-03-13 22:48:55 -0400706 if c.exception.errno not in (errno.ENOENT, errno.EACCES):
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000707 raise c.exception
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000708
Victor Stinnerb3693582010-05-21 20:13:12 +0000709 def test_issue8780(self):
710 # Ensure that stdout is inherited from the parent
711 # if stdout=PIPE is not used
712 code = ';'.join((
713 'import subprocess, sys',
714 'retcode = subprocess.call('
715 "[sys.executable, '-c', 'print(\"Hello World!\")'])",
716 'assert retcode == 0'))
717 output = subprocess.check_output([sys.executable, '-c', code])
Ezio Melottib3aedd42010-11-20 19:04:17 +0000718 self.assertTrue(output.startswith(b'Hello World!'), ascii(output))
Victor Stinnerb3693582010-05-21 20:13:12 +0000719
Tim Goldenaf5ac392010-08-06 13:03:56 +0000720 def test_handles_closed_on_exception(self):
721 # If CreateProcess exits with an error, ensure the
722 # duplicate output handles are released
723 ifhandle, ifname = mkstemp()
724 ofhandle, ofname = mkstemp()
725 efhandle, efname = mkstemp()
726 try:
727 subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle,
728 stderr=efhandle)
729 except OSError:
730 os.close(ifhandle)
731 os.remove(ifname)
732 os.close(ofhandle)
733 os.remove(ofname)
734 os.close(efhandle)
735 os.remove(efname)
736 self.assertFalse(os.path.exists(ifname))
737 self.assertFalse(os.path.exists(ofname))
738 self.assertFalse(os.path.exists(efname))
739
Ross Lagerwall4f61b022011-04-05 15:34:00 +0200740 def test_communicate_epipe(self):
741 # Issue 10963: communicate() should hide EPIPE
742 p = subprocess.Popen([sys.executable, "-c", 'pass'],
743 stdin=subprocess.PIPE,
744 stdout=subprocess.PIPE,
745 stderr=subprocess.PIPE)
746 self.addCleanup(p.stdout.close)
747 self.addCleanup(p.stderr.close)
748 self.addCleanup(p.stdin.close)
749 p.communicate(b"x" * 2**20)
750
751 def test_communicate_epipe_only_stdin(self):
752 # Issue 10963: communicate() should hide EPIPE
753 p = subprocess.Popen([sys.executable, "-c", 'pass'],
754 stdin=subprocess.PIPE)
755 self.addCleanup(p.stdin.close)
756 time.sleep(2)
757 p.communicate(b"x" * 2**20)
758
Tim Peterse718f612004-10-12 21:51:32 +0000759
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000760# context manager
761class _SuppressCoreFiles(object):
762 """Try to prevent core files from being created."""
763 old_limit = None
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000764
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000765 def __enter__(self):
766 """Try to save previous ulimit, then set it to (0, 0)."""
767 try:
768 import resource
769 self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
770 resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
771 except (ImportError, ValueError, resource.error):
772 pass
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000773
Ronald Oussoren102d11a2010-07-23 09:50:05 +0000774 if sys.platform == 'darwin':
775 # Check if the 'Crash Reporter' on OSX was configured
776 # in 'Developer' mode and warn that it will get triggered
777 # when it is.
778 #
779 # This assumes that this context manager is used in tests
780 # that might trigger the next manager.
781 value = subprocess.Popen(['/usr/bin/defaults', 'read',
782 'com.apple.CrashReporter', 'DialogType'],
783 stdout=subprocess.PIPE).communicate()[0]
784 if value.strip() == b'developer':
785 print("this tests triggers the Crash Reporter, "
786 "that is intentional", end='')
787 sys.stdout.flush()
788
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000789 def __exit__(self, *args):
790 """Return core file behavior to default."""
791 if self.old_limit is None:
792 return
793 try:
794 import resource
795 resource.setrlimit(resource.RLIMIT_CORE, self.old_limit)
796 except (ImportError, ValueError, resource.error):
797 pass
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000798
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000799
Florent Xiclunaf0cbd822010-03-04 21:50:56 +0000800@unittest.skipIf(mswindows, "POSIX specific tests")
Florent Xiclunac049d872010-03-27 22:47:23 +0000801class POSIXProcessTestCase(BaseTestCase):
Florent Xiclunaf0cbd822010-03-04 21:50:56 +0000802
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000803 def test_exceptions(self):
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000804 nonexistent_dir = "/_this/pa.th/does/not/exist"
805 try:
806 os.chdir(nonexistent_dir)
807 except OSError as e:
808 # This avoids hard coding the errno value or the OS perror()
809 # string and instead capture the exception that we want to see
810 # below for comparison.
811 desired_exception = e
Benjamin Peterson5f780402010-11-20 18:07:52 +0000812 desired_exception.strerror += ': ' + repr(sys.executable)
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000813 else:
814 self.fail("chdir to nonexistant directory %s succeeded." %
815 nonexistent_dir)
816
817 # Error in the child re-raised in the parent.
818 try:
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000819 p = subprocess.Popen([sys.executable, "-c", ""],
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000820 cwd=nonexistent_dir)
821 except OSError as e:
822 # Test that the child process chdir failure actually makes
823 # it up to the parent process as the correct exception.
824 self.assertEqual(desired_exception.errno, e.errno)
825 self.assertEqual(desired_exception.strerror, e.strerror)
826 else:
827 self.fail("Expected OSError: %s" % desired_exception)
828
829 def test_restore_signals(self):
830 # Code coverage for both values of restore_signals to make sure it
831 # at least does not blow up.
832 # A test for behavior would be complex. Contributions welcome.
833 subprocess.call([sys.executable, "-c", ""], restore_signals=True)
834 subprocess.call([sys.executable, "-c", ""], restore_signals=False)
835
836 def test_start_new_session(self):
837 # For code coverage of calling setsid(). We don't care if we get an
838 # EPERM error from it depending on the test execution environment, that
839 # still indicates that it was called.
840 try:
841 output = subprocess.check_output(
842 [sys.executable, "-c",
843 "import os; print(os.getpgid(os.getpid()))"],
844 start_new_session=True)
845 except OSError as e:
846 if e.errno != errno.EPERM:
847 raise
848 else:
849 parent_pgid = os.getpgid(os.getpid())
850 child_pgid = int(output)
851 self.assertNotEqual(parent_pgid, child_pgid)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000852
853 def test_run_abort(self):
854 # returncode handles signal termination
855 with _SuppressCoreFiles():
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000856 p = subprocess.Popen([sys.executable, "-c",
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000857 'import os; os.abort()'])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000858 p.wait()
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000859 self.assertEqual(-p.returncode, signal.SIGABRT)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000860
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000861 def test_preexec(self):
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000862 # DISCLAIMER: Setting environment variables is *not* a good use
863 # of a preexec_fn. This is merely a test.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000864 p = subprocess.Popen([sys.executable, "-c",
865 'import sys,os;'
866 'sys.stdout.write(os.getenv("FRUIT"))'],
867 stdout=subprocess.PIPE,
868 preexec_fn=lambda: os.putenv("FRUIT", "apple"))
Brian Curtin3c6a9512010-11-05 03:58:52 +0000869 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000870 self.assertEqual(p.stdout.read(), b"apple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000871
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +0000872 def test_preexec_exception(self):
873 def raise_it():
874 raise ValueError("What if two swallows carried a coconut?")
875 try:
876 p = subprocess.Popen([sys.executable, "-c", ""],
877 preexec_fn=raise_it)
878 except RuntimeError as e:
879 self.assertTrue(
880 subprocess._posixsubprocess,
881 "Expected a ValueError from the preexec_fn")
882 except ValueError as e:
883 self.assertIn("coconut", e.args[0])
884 else:
885 self.fail("Exception raised by preexec_fn did not make it "
886 "to the parent process.")
887
Gregory P. Smith32ec9da2010-03-19 16:53:08 +0000888 @unittest.skipUnless(gc, "Requires a gc module.")
889 def test_preexec_gc_module_failure(self):
890 # This tests the code that disables garbage collection if the child
891 # process will execute any Python.
892 def raise_runtime_error():
893 raise RuntimeError("this shouldn't escape")
894 enabled = gc.isenabled()
895 orig_gc_disable = gc.disable
896 orig_gc_isenabled = gc.isenabled
897 try:
898 gc.disable()
899 self.assertFalse(gc.isenabled())
900 subprocess.call([sys.executable, '-c', ''],
901 preexec_fn=lambda: None)
902 self.assertFalse(gc.isenabled(),
903 "Popen enabled gc when it shouldn't.")
904
905 gc.enable()
906 self.assertTrue(gc.isenabled())
907 subprocess.call([sys.executable, '-c', ''],
908 preexec_fn=lambda: None)
909 self.assertTrue(gc.isenabled(), "Popen left gc disabled.")
910
911 gc.disable = raise_runtime_error
912 self.assertRaises(RuntimeError, subprocess.Popen,
913 [sys.executable, '-c', ''],
914 preexec_fn=lambda: None)
915
916 del gc.isenabled # force an AttributeError
917 self.assertRaises(AttributeError, subprocess.Popen,
918 [sys.executable, '-c', ''],
919 preexec_fn=lambda: None)
920 finally:
921 gc.disable = orig_gc_disable
922 gc.isenabled = orig_gc_isenabled
923 if not enabled:
924 gc.disable()
925
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000926 def test_args_string(self):
927 # args is a string
928 fd, fname = mkstemp()
929 # reopen in text mode
Victor Stinnerf6782ac2010-10-16 23:46:43 +0000930 with open(fd, "w", errors="surrogateescape") as fobj:
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000931 fobj.write("#!/bin/sh\n")
932 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
933 sys.executable)
934 os.chmod(fname, 0o700)
935 p = subprocess.Popen(fname)
936 p.wait()
937 os.remove(fname)
938 self.assertEqual(p.returncode, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000939
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000940 def test_invalid_args(self):
941 # invalid arguments should raise ValueError
942 self.assertRaises(ValueError, subprocess.call,
943 [sys.executable, "-c",
944 "import sys; sys.exit(47)"],
945 startupinfo=47)
946 self.assertRaises(ValueError, subprocess.call,
947 [sys.executable, "-c",
948 "import sys; sys.exit(47)"],
949 creationflags=47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000950
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000951 def test_shell_sequence(self):
952 # Run command through the shell (sequence)
953 newenv = os.environ.copy()
954 newenv["FRUIT"] = "apple"
955 p = subprocess.Popen(["echo $FRUIT"], shell=1,
956 stdout=subprocess.PIPE,
957 env=newenv)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000958 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000959 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000960
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000961 def test_shell_string(self):
962 # Run command through the shell (string)
963 newenv = os.environ.copy()
964 newenv["FRUIT"] = "apple"
965 p = subprocess.Popen("echo $FRUIT", shell=1,
966 stdout=subprocess.PIPE,
967 env=newenv)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000968 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000969 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
Christian Heimesa342c012008-04-20 21:01:16 +0000970
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000971 def test_call_string(self):
972 # call() function with string argument on UNIX
973 fd, fname = mkstemp()
974 # reopen in text mode
Victor Stinnerf6782ac2010-10-16 23:46:43 +0000975 with open(fd, "w", errors="surrogateescape") as fobj:
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000976 fobj.write("#!/bin/sh\n")
977 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
978 sys.executable)
979 os.chmod(fname, 0o700)
980 rc = subprocess.call(fname)
981 os.remove(fname)
982 self.assertEqual(rc, 47)
Christian Heimesa342c012008-04-20 21:01:16 +0000983
Stefan Krah9542cc62010-07-19 14:20:53 +0000984 def test_specific_shell(self):
985 # Issue #9265: Incorrect name passed as arg[0].
986 shells = []
987 for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']:
988 for name in ['bash', 'ksh']:
989 sh = os.path.join(prefix, name)
990 if os.path.isfile(sh):
991 shells.append(sh)
992 if not shells: # Will probably work for any shell but csh.
993 self.skipTest("bash or ksh required for this test")
994 sh = '/bin/sh'
995 if os.path.isfile(sh) and not os.path.islink(sh):
996 # Test will fail if /bin/sh is a symlink to csh.
997 shells.append(sh)
998 for sh in shells:
999 p = subprocess.Popen("echo $0", executable=sh, shell=True,
1000 stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001001 self.addCleanup(p.stdout.close)
Stefan Krah9542cc62010-07-19 14:20:53 +00001002 self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii'))
1003
Florent Xicluna4886d242010-03-08 13:27:26 +00001004 def _kill_process(self, method, *args):
Florent Xicluna1d8ee3a2010-03-05 20:26:54 +00001005 # Do not inherit file handles from the parent.
1006 # It should fix failures on some platforms.
Antoine Pitrou3d8580f2010-09-20 01:33:21 +00001007 p = subprocess.Popen([sys.executable, "-c", """if 1:
1008 import sys, time
1009 sys.stdout.write('x\\n')
1010 sys.stdout.flush()
1011 time.sleep(30)
1012 """],
1013 close_fds=True,
1014 stdin=subprocess.PIPE,
1015 stdout=subprocess.PIPE,
1016 stderr=subprocess.PIPE)
1017 # Wait for the interpreter to be completely initialized before
1018 # sending any signal.
1019 p.stdout.read(1)
1020 getattr(p, method)(*args)
Florent Xicluna4886d242010-03-08 13:27:26 +00001021 return p
1022
1023 def test_send_signal(self):
1024 p = self._kill_process('send_signal', signal.SIGINT)
Florent Xiclunac049d872010-03-27 22:47:23 +00001025 _, stderr = p.communicate()
1026 self.assertIn(b'KeyboardInterrupt', stderr)
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001027 self.assertNotEqual(p.wait(), 0)
Christian Heimesa342c012008-04-20 21:01:16 +00001028
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001029 def test_kill(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001030 p = self._kill_process('kill')
Florent Xiclunac049d872010-03-27 22:47:23 +00001031 _, stderr = p.communicate()
1032 self.assertStderrEqual(stderr, b'')
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001033 self.assertEqual(p.wait(), -signal.SIGKILL)
Tim Peterse718f612004-10-12 21:51:32 +00001034
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001035 def test_terminate(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001036 p = self._kill_process('terminate')
Florent Xiclunac049d872010-03-27 22:47:23 +00001037 _, stderr = p.communicate()
1038 self.assertStderrEqual(stderr, b'')
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001039 self.assertEqual(p.wait(), -signal.SIGTERM)
1040
Antoine Pitrouc9c83ba2011-01-03 18:23:55 +00001041 def check_close_std_fds(self, fds):
1042 # Issue #9905: test that subprocess pipes still work properly with
1043 # some standard fds closed
1044 stdin = 0
1045 newfds = []
1046 for a in fds:
1047 b = os.dup(a)
1048 newfds.append(b)
1049 if a == 0:
1050 stdin = b
1051 try:
1052 for fd in fds:
1053 os.close(fd)
1054 out, err = subprocess.Popen([sys.executable, "-c",
1055 'import sys;'
1056 'sys.stdout.write("apple");'
1057 'sys.stdout.flush();'
1058 'sys.stderr.write("orange")'],
1059 stdin=stdin,
1060 stdout=subprocess.PIPE,
1061 stderr=subprocess.PIPE).communicate()
1062 err = support.strip_python_stderr(err)
1063 self.assertEqual((out, err), (b'apple', b'orange'))
1064 finally:
1065 for b, a in zip(newfds, fds):
1066 os.dup2(b, a)
1067 for b in newfds:
1068 os.close(b)
1069
1070 def test_close_fd_0(self):
1071 self.check_close_std_fds([0])
1072
1073 def test_close_fd_1(self):
1074 self.check_close_std_fds([1])
1075
1076 def test_close_fd_2(self):
1077 self.check_close_std_fds([2])
1078
1079 def test_close_fds_0_1(self):
1080 self.check_close_std_fds([0, 1])
1081
1082 def test_close_fds_0_2(self):
1083 self.check_close_std_fds([0, 2])
1084
1085 def test_close_fds_1_2(self):
1086 self.check_close_std_fds([1, 2])
1087
1088 def test_close_fds_0_1_2(self):
1089 # Issue #10806: test that subprocess pipes still work properly with
1090 # all standard fds closed.
1091 self.check_close_std_fds([0, 1, 2])
1092
Antoine Pitrou95aaeee2011-01-03 21:15:48 +00001093 def test_remapping_std_fds(self):
1094 # open up some temporary files
1095 temps = [mkstemp() for i in range(3)]
1096 try:
1097 temp_fds = [fd for fd, fname in temps]
1098
1099 # unlink the files -- we won't need to reopen them
1100 for fd, fname in temps:
1101 os.unlink(fname)
1102
1103 # write some data to what will become stdin, and rewind
1104 os.write(temp_fds[1], b"STDIN")
1105 os.lseek(temp_fds[1], 0, 0)
1106
1107 # move the standard file descriptors out of the way
1108 saved_fds = [os.dup(fd) for fd in range(3)]
1109 try:
1110 # duplicate the file objects over the standard fd's
1111 for fd, temp_fd in enumerate(temp_fds):
1112 os.dup2(temp_fd, fd)
1113
1114 # now use those files in the "wrong" order, so that subprocess
1115 # has to rearrange them in the child
1116 p = subprocess.Popen([sys.executable, "-c",
1117 'import sys; got = sys.stdin.read();'
1118 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1119 stdin=temp_fds[1],
1120 stdout=temp_fds[2],
1121 stderr=temp_fds[0])
1122 p.wait()
1123 finally:
1124 # restore the original fd's underneath sys.stdin, etc.
1125 for std, saved in enumerate(saved_fds):
1126 os.dup2(saved, std)
1127 os.close(saved)
1128
1129 for fd in temp_fds:
1130 os.lseek(fd, 0, 0)
1131
1132 out = os.read(temp_fds[2], 1024)
1133 err = support.strip_python_stderr(os.read(temp_fds[0], 1024))
1134 self.assertEqual(out, b"got STDIN")
1135 self.assertEqual(err, b"err")
1136
1137 finally:
1138 for fd in temp_fds:
1139 os.close(fd)
1140
Victor Stinner13bb71c2010-04-23 21:41:56 +00001141 def test_surrogates_error_message(self):
Victor Stinner4d078042010-04-23 19:28:32 +00001142 def prepare():
1143 raise ValueError("surrogate:\uDCff")
1144
1145 try:
1146 subprocess.call(
1147 [sys.executable, "-c", "pass"],
1148 preexec_fn=prepare)
1149 except ValueError as err:
1150 # Pure Python implementations keeps the message
1151 self.assertIsNone(subprocess._posixsubprocess)
1152 self.assertEqual(str(err), "surrogate:\uDCff")
1153 except RuntimeError as err:
1154 # _posixsubprocess uses a default message
1155 self.assertIsNotNone(subprocess._posixsubprocess)
1156 self.assertEqual(str(err), "Exception occurred in preexec_fn.")
1157 else:
1158 self.fail("Expected ValueError or RuntimeError")
1159
Victor Stinner13bb71c2010-04-23 21:41:56 +00001160 def test_undecodable_env(self):
1161 for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')):
Victor Stinner13bb71c2010-04-23 21:41:56 +00001162 # test str with surrogates
Antoine Pitroufb8db8f2010-09-19 22:46:05 +00001163 script = "import os; print(ascii(os.getenv(%s)))" % repr(key)
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001164 env = os.environ.copy()
1165 env[key] = value
Victor Stinner89f3ad12010-10-14 10:43:31 +00001166 # Use C locale to get ascii for the locale encoding to force
1167 # surrogate-escaping of \xFF in the child process; otherwise it can
1168 # be decoded as-is if the default locale is latin-1.
Victor Stinnerebc78d22010-10-14 10:38:17 +00001169 env['LC_ALL'] = 'C'
Victor Stinner13bb71c2010-04-23 21:41:56 +00001170 stdout = subprocess.check_output(
1171 [sys.executable, "-c", script],
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001172 env=env)
Victor Stinner13bb71c2010-04-23 21:41:56 +00001173 stdout = stdout.rstrip(b'\n\r')
Ezio Melottib3aedd42010-11-20 19:04:17 +00001174 self.assertEqual(stdout.decode('ascii'), ascii(value))
Victor Stinner13bb71c2010-04-23 21:41:56 +00001175
1176 # test bytes
1177 key = key.encode("ascii", "surrogateescape")
1178 value = value.encode("ascii", "surrogateescape")
Antoine Pitroufb8db8f2010-09-19 22:46:05 +00001179 script = "import os; print(ascii(os.getenvb(%s)))" % repr(key)
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001180 env = os.environ.copy()
1181 env[key] = value
Victor Stinner13bb71c2010-04-23 21:41:56 +00001182 stdout = subprocess.check_output(
1183 [sys.executable, "-c", script],
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001184 env=env)
Victor Stinner13bb71c2010-04-23 21:41:56 +00001185 stdout = stdout.rstrip(b'\n\r')
Ezio Melottib3aedd42010-11-20 19:04:17 +00001186 self.assertEqual(stdout.decode('ascii'), ascii(value))
Victor Stinner13bb71c2010-04-23 21:41:56 +00001187
Victor Stinnerb745a742010-05-18 17:17:23 +00001188 def test_bytes_program(self):
1189 abs_program = os.fsencode(sys.executable)
1190 path, program = os.path.split(sys.executable)
1191 program = os.fsencode(program)
1192
1193 # absolute bytes path
1194 exitcode = subprocess.call([abs_program, "-c", "pass"])
Ezio Melottib3aedd42010-11-20 19:04:17 +00001195 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001196
Victor Stinner7b3b20a2011-03-03 12:54:05 +00001197 # absolute bytes path as a string
1198 cmd = b"'" + abs_program + b"' -c pass"
1199 exitcode = subprocess.call(cmd, shell=True)
1200 self.assertEqual(exitcode, 0)
1201
Victor Stinnerb745a742010-05-18 17:17:23 +00001202 # bytes program, unicode PATH
1203 env = os.environ.copy()
1204 env["PATH"] = path
1205 exitcode = subprocess.call([program, "-c", "pass"], env=env)
Ezio Melottib3aedd42010-11-20 19:04:17 +00001206 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001207
1208 # bytes program, bytes PATH
1209 envb = os.environb.copy()
1210 envb[b"PATH"] = os.fsencode(path)
1211 exitcode = subprocess.call([program, "-c", "pass"], env=envb)
Ezio Melottib3aedd42010-11-20 19:04:17 +00001212 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001213
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001214 def test_pipe_cloexec(self):
1215 sleeper = support.findfile("input_reader.py", subdir="subprocessdata")
1216 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1217
1218 p1 = subprocess.Popen([sys.executable, sleeper],
1219 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1220 stderr=subprocess.PIPE, close_fds=False)
1221
1222 self.addCleanup(p1.communicate, b'')
1223
1224 p2 = subprocess.Popen([sys.executable, fd_status],
1225 stdout=subprocess.PIPE, close_fds=False)
1226
1227 output, error = p2.communicate()
1228 result_fds = set(map(int, output.split(b',')))
1229 unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(),
1230 p1.stderr.fileno()])
1231
1232 self.assertFalse(result_fds & unwanted_fds,
1233 "Expected no fds from %r to be open in child, "
1234 "found %r" %
1235 (unwanted_fds, result_fds & unwanted_fds))
1236
1237 def test_pipe_cloexec_real_tools(self):
1238 qcat = support.findfile("qcat.py", subdir="subprocessdata")
1239 qgrep = support.findfile("qgrep.py", subdir="subprocessdata")
1240
1241 subdata = b'zxcvbn'
1242 data = subdata * 4 + b'\n'
1243
1244 p1 = subprocess.Popen([sys.executable, qcat],
1245 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1246 close_fds=False)
1247
1248 p2 = subprocess.Popen([sys.executable, qgrep, subdata],
1249 stdin=p1.stdout, stdout=subprocess.PIPE,
1250 close_fds=False)
1251
1252 self.addCleanup(p1.wait)
1253 self.addCleanup(p2.wait)
1254 self.addCleanup(p1.terminate)
1255 self.addCleanup(p2.terminate)
1256
1257 p1.stdin.write(data)
1258 p1.stdin.close()
1259
1260 readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10)
1261
1262 self.assertTrue(readfiles, "The child hung")
1263 self.assertEqual(p2.stdout.read(), data)
1264
Victor Stinnerfaa8c132011-01-03 16:36:00 +00001265 p1.stdout.close()
1266 p2.stdout.close()
1267
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001268 def test_close_fds(self):
1269 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1270
1271 fds = os.pipe()
1272 self.addCleanup(os.close, fds[0])
1273 self.addCleanup(os.close, fds[1])
1274
1275 open_fds = set(fds)
1276
1277 p = subprocess.Popen([sys.executable, fd_status],
1278 stdout=subprocess.PIPE, close_fds=False)
1279 output, ignored = p.communicate()
1280 remaining_fds = set(map(int, output.split(b',')))
1281
1282 self.assertEqual(remaining_fds & open_fds, open_fds,
1283 "Some fds were closed")
1284
1285 p = subprocess.Popen([sys.executable, fd_status],
1286 stdout=subprocess.PIPE, close_fds=True)
1287 output, ignored = p.communicate()
1288 remaining_fds = set(map(int, output.split(b',')))
1289
1290 self.assertFalse(remaining_fds & open_fds,
1291 "Some fds were left open")
1292 self.assertIn(1, remaining_fds, "Subprocess failed")
1293
Victor Stinner88701e22011-06-01 13:13:04 +02001294 # Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file
1295 # descriptor of a pipe closed in the parent process is valid in the
1296 # child process according to fstat(), but the mode of the file
1297 # descriptor is invalid, and read or write raise an error.
1298 @support.requires_mac_ver(10, 5)
Gregory P. Smith8edd99d2010-12-14 13:43:30 +00001299 def test_pass_fds(self):
1300 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1301
1302 open_fds = set()
1303
1304 for x in range(5):
1305 fds = os.pipe()
1306 self.addCleanup(os.close, fds[0])
1307 self.addCleanup(os.close, fds[1])
1308 open_fds.update(fds)
1309
1310 for fd in open_fds:
1311 p = subprocess.Popen([sys.executable, fd_status],
1312 stdout=subprocess.PIPE, close_fds=True,
1313 pass_fds=(fd, ))
1314 output, ignored = p.communicate()
1315
1316 remaining_fds = set(map(int, output.split(b',')))
1317 to_be_closed = open_fds - {fd}
1318
1319 self.assertIn(fd, remaining_fds, "fd to be passed not passed")
1320 self.assertFalse(remaining_fds & to_be_closed,
1321 "fd to be closed passed")
1322
1323 # pass_fds overrides close_fds with a warning.
1324 with self.assertWarns(RuntimeWarning) as context:
1325 self.assertFalse(subprocess.call(
1326 [sys.executable, "-c", "import sys; sys.exit(0)"],
1327 close_fds=False, pass_fds=(fd, )))
1328 self.assertIn('overriding close_fds', str(context.warning))
1329
Gregory P. Smith112bb3a2011-03-15 14:55:17 -04001330 def test_stdout_stdin_are_single_inout_fd(self):
1331 with io.open(os.devnull, "r+") as inout:
1332 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1333 stdout=inout, stdin=inout)
1334 p.wait()
1335
1336 def test_stdout_stderr_are_single_inout_fd(self):
1337 with io.open(os.devnull, "r+") as inout:
1338 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1339 stdout=inout, stderr=inout)
1340 p.wait()
1341
1342 def test_stderr_stdin_are_single_inout_fd(self):
1343 with io.open(os.devnull, "r+") as inout:
1344 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1345 stderr=inout, stdin=inout)
1346 p.wait()
1347
Gregory P. Smithe85db2b2010-12-14 14:38:00 +00001348 def test_wait_when_sigchild_ignored(self):
1349 # NOTE: sigchild_ignore.py may not be an effective test on all OSes.
1350 sigchild_ignore = support.findfile("sigchild_ignore.py",
1351 subdir="subprocessdata")
1352 p = subprocess.Popen([sys.executable, sigchild_ignore],
1353 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1354 stdout, stderr = p.communicate()
1355 self.assertEqual(0, p.returncode, "sigchild_ignore.py exited"
Gregory P. Smitha80f4fb2010-12-14 15:23:02 +00001356 " non-zero with this error:\n%s" %
Marc-André Lemburg8f36af72011-02-25 15:42:01 +00001357 stderr.decode('utf-8'))
Gregory P. Smithe85db2b2010-12-14 14:38:00 +00001358
Antoine Pitrou7b98d022011-03-19 17:04:13 +01001359 def test_select_unbuffered(self):
1360 # Issue #11459: bufsize=0 should really set the pipes as
1361 # unbuffered (and therefore let select() work properly).
1362 select = support.import_module("select")
1363 p = subprocess.Popen([sys.executable, "-c",
1364 'import sys;'
1365 'sys.stdout.write("apple")'],
1366 stdout=subprocess.PIPE,
1367 bufsize=0)
1368 f = p.stdout
Ross Lagerwall17ace7a2011-03-26 21:21:46 +02001369 self.addCleanup(f.close)
Antoine Pitrou7b98d022011-03-19 17:04:13 +01001370 try:
1371 self.assertEqual(f.read(4), b"appl")
1372 self.assertIn(f, select.select([f], [], [], 0.0)[0])
1373 finally:
1374 p.wait()
1375
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001376
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001377@unittest.skipUnless(mswindows, "Windows specific tests")
Florent Xiclunac049d872010-03-27 22:47:23 +00001378class Win32ProcessTestCase(BaseTestCase):
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001379
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001380 def test_startupinfo(self):
1381 # startupinfo argument
1382 # We uses hardcoded constants, because we do not want to
1383 # depend on win32all.
1384 STARTF_USESHOWWINDOW = 1
1385 SW_MAXIMIZE = 3
1386 startupinfo = subprocess.STARTUPINFO()
1387 startupinfo.dwFlags = STARTF_USESHOWWINDOW
1388 startupinfo.wShowWindow = SW_MAXIMIZE
1389 # Since Python is a console process, it won't be affected
1390 # by wShowWindow, but the argument should be silently
1391 # ignored
1392 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001393 startupinfo=startupinfo)
1394
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001395 def test_creationflags(self):
1396 # creationflags argument
1397 CREATE_NEW_CONSOLE = 16
1398 sys.stderr.write(" a DOS box should flash briefly ...\n")
1399 subprocess.call(sys.executable +
1400 ' -c "import time; time.sleep(0.25)"',
1401 creationflags=CREATE_NEW_CONSOLE)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001402
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001403 def test_invalid_args(self):
1404 # invalid arguments should raise ValueError
1405 self.assertRaises(ValueError, subprocess.call,
1406 [sys.executable, "-c",
1407 "import sys; sys.exit(47)"],
1408 preexec_fn=lambda: 1)
1409 self.assertRaises(ValueError, subprocess.call,
1410 [sys.executable, "-c",
1411 "import sys; sys.exit(47)"],
1412 stdout=subprocess.PIPE,
1413 close_fds=True)
1414
1415 def test_close_fds(self):
1416 # close file descriptors
1417 rc = subprocess.call([sys.executable, "-c",
1418 "import sys; sys.exit(47)"],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001419 close_fds=True)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001420 self.assertEqual(rc, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001421
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001422 def test_shell_sequence(self):
1423 # Run command through the shell (sequence)
1424 newenv = os.environ.copy()
1425 newenv["FRUIT"] = "physalis"
1426 p = subprocess.Popen(["set"], shell=1,
1427 stdout=subprocess.PIPE,
1428 env=newenv)
Brian Curtin19a53792010-11-05 17:09:05 +00001429 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001430 self.assertIn(b"physalis", p.stdout.read())
Guido van Rossume7ba4952007-06-06 23:52:48 +00001431
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001432 def test_shell_string(self):
1433 # Run command through the shell (string)
1434 newenv = os.environ.copy()
1435 newenv["FRUIT"] = "physalis"
1436 p = subprocess.Popen("set", shell=1,
1437 stdout=subprocess.PIPE,
1438 env=newenv)
Brian Curtin19a53792010-11-05 17:09:05 +00001439 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001440 self.assertIn(b"physalis", p.stdout.read())
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001441
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001442 def test_call_string(self):
1443 # call() function with string argument on Windows
1444 rc = subprocess.call(sys.executable +
1445 ' -c "import sys; sys.exit(47)"')
1446 self.assertEqual(rc, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001447
Florent Xicluna4886d242010-03-08 13:27:26 +00001448 def _kill_process(self, method, *args):
1449 # Some win32 buildbot raises EOFError if stdin is inherited
Antoine Pitroua4024e22010-09-24 18:57:01 +00001450 p = subprocess.Popen([sys.executable, "-c", """if 1:
1451 import sys, time
1452 sys.stdout.write('x\\n')
1453 sys.stdout.flush()
1454 time.sleep(30)
1455 """],
1456 stdin=subprocess.PIPE,
1457 stdout=subprocess.PIPE,
1458 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +00001459 self.addCleanup(p.stdout.close)
1460 self.addCleanup(p.stderr.close)
1461 self.addCleanup(p.stdin.close)
Antoine Pitroua4024e22010-09-24 18:57:01 +00001462 # Wait for the interpreter to be completely initialized before
1463 # sending any signal.
1464 p.stdout.read(1)
1465 getattr(p, method)(*args)
Florent Xiclunac049d872010-03-27 22:47:23 +00001466 _, stderr = p.communicate()
1467 self.assertStderrEqual(stderr, b'')
Antoine Pitroua4024e22010-09-24 18:57:01 +00001468 returncode = p.wait()
Florent Xicluna4886d242010-03-08 13:27:26 +00001469 self.assertNotEqual(returncode, 0)
1470
1471 def test_send_signal(self):
1472 self._kill_process('send_signal', signal.SIGTERM)
Christian Heimesa342c012008-04-20 21:01:16 +00001473
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001474 def test_kill(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001475 self._kill_process('kill')
Christian Heimesa342c012008-04-20 21:01:16 +00001476
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001477 def test_terminate(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001478 self._kill_process('terminate')
Christian Heimesa342c012008-04-20 21:01:16 +00001479
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001480
Brett Cannona23810f2008-05-26 19:04:21 +00001481# The module says:
1482# "NB This only works (and is only relevant) for UNIX."
1483#
1484# Actually, getoutput should work on any platform with an os.popen, but
1485# I'll take the comment as given, and skip this suite.
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001486@unittest.skipUnless(os.name == 'posix', "only relevant for UNIX")
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001487class CommandTests(unittest.TestCase):
1488 def test_getoutput(self):
1489 self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy')
1490 self.assertEqual(subprocess.getstatusoutput('echo xyzzy'),
1491 (0, 'xyzzy'))
Brett Cannona23810f2008-05-26 19:04:21 +00001492
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001493 # we use mkdtemp in the next line to create an empty directory
1494 # under our exclusive control; from that, we can invent a pathname
1495 # that we _know_ won't exist. This is guaranteed to fail.
1496 dir = None
1497 try:
1498 dir = tempfile.mkdtemp()
1499 name = os.path.join(dir, "foo")
Brett Cannona23810f2008-05-26 19:04:21 +00001500
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001501 status, output = subprocess.getstatusoutput('cat ' + name)
1502 self.assertNotEqual(status, 0)
1503 finally:
1504 if dir is not None:
1505 os.rmdir(dir)
Brett Cannona23810f2008-05-26 19:04:21 +00001506
Gregory P. Smithd06fa472009-07-04 02:46:54 +00001507
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001508@unittest.skipUnless(getattr(subprocess, '_has_poll', False),
1509 "poll system call not supported")
1510class ProcessTestCaseNoPoll(ProcessTestCase):
1511 def setUp(self):
1512 subprocess._has_poll = False
1513 ProcessTestCase.setUp(self)
Gregory P. Smithd06fa472009-07-04 02:46:54 +00001514
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001515 def tearDown(self):
1516 subprocess._has_poll = True
1517 ProcessTestCase.tearDown(self)
Gregory P. Smithd06fa472009-07-04 02:46:54 +00001518
1519
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00001520class HelperFunctionTests(unittest.TestCase):
Gregory P. Smithaf6d3b82010-03-01 02:56:44 +00001521 @unittest.skipIf(mswindows, "errno and EINTR make no sense on windows")
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00001522 def test_eintr_retry_call(self):
1523 record_calls = []
1524 def fake_os_func(*args):
1525 record_calls.append(args)
1526 if len(record_calls) == 2:
1527 raise OSError(errno.EINTR, "fake interrupted system call")
1528 return tuple(reversed(args))
1529
1530 self.assertEqual((999, 256),
1531 subprocess._eintr_retry_call(fake_os_func, 256, 999))
1532 self.assertEqual([(256, 999)], record_calls)
1533 # This time there will be an EINTR so it will loop once.
1534 self.assertEqual((666,),
1535 subprocess._eintr_retry_call(fake_os_func, 666))
1536 self.assertEqual([(256, 999), (666,), (666,)], record_calls)
1537
1538
Tim Golden126c2962010-08-11 14:20:40 +00001539@unittest.skipUnless(mswindows, "Windows-specific tests")
1540class CommandsWithSpaces (BaseTestCase):
1541
1542 def setUp(self):
1543 super().setUp()
1544 f, fname = mkstemp(".py", "te st")
1545 self.fname = fname.lower ()
1546 os.write(f, b"import sys;"
1547 b"sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))"
1548 )
1549 os.close(f)
1550
1551 def tearDown(self):
1552 os.remove(self.fname)
1553 super().tearDown()
1554
1555 def with_spaces(self, *args, **kwargs):
1556 kwargs['stdout'] = subprocess.PIPE
1557 p = subprocess.Popen(*args, **kwargs)
Brian Curtin19a53792010-11-05 17:09:05 +00001558 self.addCleanup(p.stdout.close)
Tim Golden126c2962010-08-11 14:20:40 +00001559 self.assertEqual(
1560 p.stdout.read ().decode("mbcs"),
1561 "2 [%r, 'ab cd']" % self.fname
1562 )
1563
1564 def test_shell_string_with_spaces(self):
1565 # call() function with string argument with spaces on Windows
Brian Curtind835cf12010-08-13 20:42:57 +00001566 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
1567 "ab cd"), shell=1)
Tim Golden126c2962010-08-11 14:20:40 +00001568
1569 def test_shell_sequence_with_spaces(self):
1570 # call() function with sequence argument with spaces on Windows
Brian Curtind835cf12010-08-13 20:42:57 +00001571 self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1)
Tim Golden126c2962010-08-11 14:20:40 +00001572
1573 def test_noshell_string_with_spaces(self):
1574 # call() function with string argument with spaces on Windows
1575 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
1576 "ab cd"))
1577
1578 def test_noshell_sequence_with_spaces(self):
1579 # call() function with sequence argument with spaces on Windows
1580 self.with_spaces([sys.executable, self.fname, "ab cd"])
1581
Brian Curtin79cdb662010-12-03 02:46:02 +00001582
1583class ContextManagerTests(ProcessTestCase):
1584
1585 def test_pipe(self):
1586 with subprocess.Popen([sys.executable, "-c",
1587 "import sys;"
1588 "sys.stdout.write('stdout');"
1589 "sys.stderr.write('stderr');"],
1590 stdout=subprocess.PIPE,
1591 stderr=subprocess.PIPE) as proc:
1592 self.assertEqual(proc.stdout.read(), b"stdout")
1593 self.assertStderrEqual(proc.stderr.read(), b"stderr")
1594
1595 self.assertTrue(proc.stdout.closed)
1596 self.assertTrue(proc.stderr.closed)
1597
1598 def test_returncode(self):
1599 with subprocess.Popen([sys.executable, "-c",
1600 "import sys; sys.exit(100)"]) as proc:
Gregory P. Smith6b657452011-05-11 21:42:08 -07001601 pass
1602 # __exit__ calls wait(), so the returncode should be set
Brian Curtin79cdb662010-12-03 02:46:02 +00001603 self.assertEqual(proc.returncode, 100)
1604
1605 def test_communicate_stdin(self):
1606 with subprocess.Popen([sys.executable, "-c",
1607 "import sys;"
1608 "sys.exit(sys.stdin.read() == 'context')"],
1609 stdin=subprocess.PIPE) as proc:
1610 proc.communicate(b"context")
1611 self.assertEqual(proc.returncode, 1)
1612
1613 def test_invalid_args(self):
1614 with self.assertRaises(EnvironmentError) as c:
1615 with subprocess.Popen(['nonexisting_i_hope'],
1616 stdout=subprocess.PIPE,
1617 stderr=subprocess.PIPE) as proc:
1618 pass
1619
1620 if c.exception.errno != errno.ENOENT: # ignore "no such file"
1621 raise c.exception
1622
1623
Gregory P. Smith3b4652e2011-03-15 15:43:39 -04001624def test_main():
1625 unit_tests = (ProcessTestCase,
1626 POSIXProcessTestCase,
1627 Win32ProcessTestCase,
Gregory P. Smith3b4652e2011-03-15 15:43:39 -04001628 CommandTests,
1629 ProcessTestCaseNoPoll,
1630 HelperFunctionTests,
1631 CommandsWithSpaces,
1632 ContextManagerTests)
1633
1634 support.run_unittest(*unit_tests)
1635 support.reap_children()
1636
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001637if __name__ == "__main__":
Gregory P. Smith112bb3a2011-03-15 14:55:17 -04001638 unittest.main()