blob: 4b904710931a285c9f9ea5d10338a2dba3d81f36 [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. Smith112bb3a2011-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
Andrew Svetlov47ec25d2012-08-19 16:25:37 +030019import textwrap
Benjamin Peterson964561b2011-12-10 12:31:42 -050020
21try:
22 import resource
23except ImportError:
24 resource = None
25
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000026mswindows = (sys.platform == "win32")
27
28#
29# Depends on the following external programs: Python
30#
31
32if mswindows:
Tim Peters3b01a702004-10-12 22:19:32 +000033 SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), '
34 'os.O_BINARY);')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +000035else:
36 SETBINARY = ''
37
Florent Xiclunab1e94e82010-02-27 22:12:37 +000038
39try:
40 mkstemp = tempfile.mkstemp
41except AttributeError:
42 # tempfile.mkstemp is not available
43 def mkstemp():
44 """Replacement for mkstemp, calling mktemp."""
45 fname = tempfile.mktemp()
46 return os.open(fname, os.O_RDWR|os.O_CREAT), fname
47
Tim Peters3761e8d2004-10-13 04:07:12 +000048
Florent Xiclunac049d872010-03-27 22:47:23 +000049class BaseTestCase(unittest.TestCase):
Thomas Wouters0e3f5912006-08-11 14:57:12 +000050 def setUp(self):
51 # Try to minimize the number of children we have so this test
52 # doesn't crash on some buildbots (Alphas in particular).
Florent Xiclunab1e94e82010-02-27 22:12:37 +000053 support.reap_children()
Thomas Wouters0e3f5912006-08-11 14:57:12 +000054
Florent Xiclunaf0cbd822010-03-04 21:50:56 +000055 def tearDown(self):
56 for inst in subprocess._active:
57 inst.wait()
58 subprocess._cleanup()
59 self.assertFalse(subprocess._active, "subprocess._active not empty")
60
Florent Xiclunab1e94e82010-02-27 22:12:37 +000061 def assertStderrEqual(self, stderr, expected, msg=None):
62 # In a debug build, stuff like "[6580 refs]" is printed to stderr at
63 # shutdown time. That frustrates tests trying to check stderr produced
64 # from a spawned Python process.
Antoine Pitrou62f68ed2010-08-04 11:48:56 +000065 actual = support.strip_python_stderr(stderr)
Reid Kleckner31aa7dd2011-03-14 12:02:10 -040066 # strip_python_stderr also strips whitespace, so we do too.
67 expected = expected.strip()
Florent Xiclunab1e94e82010-02-27 22:12:37 +000068 self.assertEqual(actual, expected, msg)
Thomas Wouters0e3f5912006-08-11 14:57:12 +000069
Florent Xiclunac049d872010-03-27 22:47:23 +000070
Gregory P. Smith3d8e7762012-11-10 22:32:22 -080071class PopenTestException(Exception):
72 pass
73
74
75class PopenExecuteChildRaises(subprocess.Popen):
76 """Popen subclass for testing cleanup of subprocess.PIPE filehandles when
77 _execute_child fails.
78 """
79 def _execute_child(self, *args, **kwargs):
80 raise PopenTestException("Forced Exception for Test")
81
82
Florent Xiclunac049d872010-03-27 22:47:23 +000083class ProcessTestCase(BaseTestCase):
84
Gregory P. Smitha1ed5392013-03-23 11:44:25 -070085 def test_io_buffered_by_default(self):
86 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
87 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
88 stderr=subprocess.PIPE)
89 try:
90 self.assertIsInstance(p.stdin, io.BufferedIOBase)
91 self.assertIsInstance(p.stdout, io.BufferedIOBase)
92 self.assertIsInstance(p.stderr, io.BufferedIOBase)
93 finally:
Gregory P. Smitha1b9ed32013-03-23 11:54:22 -070094 p.stdin.close()
95 p.stdout.close()
96 p.stderr.close()
Gregory P. Smitha1ed5392013-03-23 11:44:25 -070097 p.wait()
98
99 def test_io_unbuffered_works(self):
100 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
101 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
102 stderr=subprocess.PIPE, bufsize=0)
103 try:
104 self.assertIsInstance(p.stdin, io.RawIOBase)
105 self.assertIsInstance(p.stdout, io.RawIOBase)
106 self.assertIsInstance(p.stderr, io.RawIOBase)
107 finally:
Gregory P. Smitha1b9ed32013-03-23 11:54:22 -0700108 p.stdin.close()
109 p.stdout.close()
110 p.stderr.close()
Gregory P. Smitha1ed5392013-03-23 11:44:25 -0700111 p.wait()
112
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000113 def test_call_seq(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000114 # call() function with sequence argument
Tim Peters3b01a702004-10-12 22:19:32 +0000115 rc = subprocess.call([sys.executable, "-c",
116 "import sys; sys.exit(47)"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000117 self.assertEqual(rc, 47)
118
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400119 def test_call_timeout(self):
120 # call() function with timeout argument; we want to test that the child
121 # process gets killed when the timeout expires. If the child isn't
122 # killed, this call will deadlock since subprocess.call waits for the
123 # child.
124 self.assertRaises(subprocess.TimeoutExpired, subprocess.call,
125 [sys.executable, "-c", "while True: pass"],
126 timeout=0.1)
127
Peter Astrand454f7672005-01-01 09:36:35 +0000128 def test_check_call_zero(self):
129 # check_call() function with zero return code
130 rc = subprocess.check_call([sys.executable, "-c",
131 "import sys; sys.exit(0)"])
132 self.assertEqual(rc, 0)
133
134 def test_check_call_nonzero(self):
135 # check_call() function with non-zero return code
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000136 with self.assertRaises(subprocess.CalledProcessError) as c:
Peter Astrand454f7672005-01-01 09:36:35 +0000137 subprocess.check_call([sys.executable, "-c",
138 "import sys; sys.exit(47)"])
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000139 self.assertEqual(c.exception.returncode, 47)
Peter Astrand454f7672005-01-01 09:36:35 +0000140
Georg Brandlf9734072008-12-07 15:30:06 +0000141 def test_check_output(self):
142 # check_output() function with zero return code
143 output = subprocess.check_output(
144 [sys.executable, "-c", "print('BDFL')"])
Benjamin Peterson577473f2010-01-19 00:09:57 +0000145 self.assertIn(b'BDFL', output)
Georg Brandlf9734072008-12-07 15:30:06 +0000146
147 def test_check_output_nonzero(self):
148 # check_call() function with non-zero return code
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000149 with self.assertRaises(subprocess.CalledProcessError) as c:
Georg Brandlf9734072008-12-07 15:30:06 +0000150 subprocess.check_output(
151 [sys.executable, "-c", "import sys; sys.exit(5)"])
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000152 self.assertEqual(c.exception.returncode, 5)
Georg Brandlf9734072008-12-07 15:30:06 +0000153
154 def test_check_output_stderr(self):
155 # check_output() function stderr redirected to stdout
156 output = subprocess.check_output(
157 [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"],
158 stderr=subprocess.STDOUT)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000159 self.assertIn(b'BDFL', output)
Georg Brandlf9734072008-12-07 15:30:06 +0000160
161 def test_check_output_stdout_arg(self):
162 # check_output() function stderr redirected to stdout
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000163 with self.assertRaises(ValueError) as c:
Georg Brandlf9734072008-12-07 15:30:06 +0000164 output = subprocess.check_output(
165 [sys.executable, "-c", "print('will not be run')"],
166 stdout=sys.stdout)
Georg Brandlf9734072008-12-07 15:30:06 +0000167 self.fail("Expected ValueError when stdout arg supplied.")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000168 self.assertIn('stdout', c.exception.args[0])
Georg Brandlf9734072008-12-07 15:30:06 +0000169
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400170 def test_check_output_timeout(self):
171 # check_output() function with timeout arg
172 with self.assertRaises(subprocess.TimeoutExpired) as c:
173 output = subprocess.check_output(
174 [sys.executable, "-c",
Victor Stinner149b1c72011-06-06 23:43:02 +0200175 "import sys, time\n"
176 "sys.stdout.write('BDFL')\n"
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400177 "sys.stdout.flush()\n"
Victor Stinner149b1c72011-06-06 23:43:02 +0200178 "time.sleep(3600)"],
Reid Klecknerda9ac722011-03-16 17:08:21 -0400179 # Some heavily loaded buildbots (sparc Debian 3.x) require
180 # this much time to start and print.
181 timeout=3)
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400182 self.fail("Expected TimeoutExpired.")
183 self.assertEqual(c.exception.output, b'BDFL')
184
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000185 def test_call_kwargs(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000186 # call() function with keyword args
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000187 newenv = os.environ.copy()
188 newenv["FRUIT"] = "banana"
189 rc = subprocess.call([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000190 'import sys, os;'
191 'sys.exit(os.getenv("FRUIT")=="banana")'],
192 env=newenv)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000193 self.assertEqual(rc, 1)
194
Victor Stinner87b9bc32011-06-01 00:57:47 +0200195 def test_invalid_args(self):
196 # Popen() called with invalid arguments should raise TypeError
197 # but Popen.__del__ should not complain (issue #12085)
198 with support.captured_stderr() as s:
199 self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1)
200 argcount = subprocess.Popen.__init__.__code__.co_argcount
201 too_many_args = [0] * (argcount + 1)
202 self.assertRaises(TypeError, subprocess.Popen, *too_many_args)
203 self.assertEqual(s.getvalue(), '')
204
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000205 def test_stdin_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000206 # .stdin is None when not redirected
Georg Brandl88fc6642007-02-09 21:28:07 +0000207 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000208 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000209 self.addCleanup(p.stdout.close)
210 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000211 p.wait()
212 self.assertEqual(p.stdin, None)
213
214 def test_stdout_none(self):
Ezio Melotti42a541b2013-03-11 05:53:34 +0200215 # .stdout is None when not redirected, and the child's stdout will
216 # be inherited from the parent. In order to test this we run a
217 # subprocess in a subprocess:
218 # this_test
219 # \-- subprocess created by this test (parent)
220 # \-- subprocess created by the parent subprocess (child)
221 # The parent doesn't specify stdout, so the child will use the
222 # parent's stdout. This test checks that the message printed by the
223 # child goes to the parent stdout. The parent also checks that the
224 # child's stdout is None. See #11963.
225 code = ('import sys; from subprocess import Popen, PIPE;'
226 'p = Popen([sys.executable, "-c", "print(\'test_stdout_none\')"],'
227 ' stdin=PIPE, stderr=PIPE);'
228 'p.wait(); assert p.stdout is None;')
229 p = subprocess.Popen([sys.executable, "-c", code],
230 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
231 self.addCleanup(p.stdout.close)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000232 self.addCleanup(p.stderr.close)
Ezio Melotti42a541b2013-03-11 05:53:34 +0200233 out, err = p.communicate()
234 self.assertEqual(p.returncode, 0, err)
235 self.assertEqual(out.rstrip(), b'test_stdout_none')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000236
237 def test_stderr_none(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000238 # .stderr is None when not redirected
Georg Brandl88fc6642007-02-09 21:28:07 +0000239 p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000240 stdin=subprocess.PIPE, stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000241 self.addCleanup(p.stdout.close)
242 self.addCleanup(p.stdin.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000243 p.wait()
244 self.assertEqual(p.stderr, None)
245
Chris Jerdonek776cb192012-10-08 15:56:43 -0700246 def _assert_python(self, pre_args, **kwargs):
247 # We include sys.exit() to prevent the test runner from hanging
248 # whenever python is found.
249 args = pre_args + ["import sys; sys.exit(47)"]
250 p = subprocess.Popen(args, **kwargs)
251 p.wait()
252 self.assertEqual(47, p.returncode)
253
254 def test_executable(self):
255 # Check that the executable argument works.
Chris Jerdonek86b0fb22012-10-09 13:17:49 -0700256 #
257 # On Unix (non-Mac and non-Windows), Python looks at args[0] to
258 # determine where its standard library is, so we need the directory
259 # of args[0] to be valid for the Popen() call to Python to succeed.
260 # See also issue #16170 and issue #7774.
261 doesnotexist = os.path.join(os.path.dirname(sys.executable),
262 "doesnotexist")
263 self._assert_python([doesnotexist, "-c"], executable=sys.executable)
Chris Jerdonek776cb192012-10-08 15:56:43 -0700264
265 def test_executable_takes_precedence(self):
266 # Check that the executable argument takes precedence over args[0].
267 #
268 # Verify first that the call succeeds without the executable arg.
269 pre_args = [sys.executable, "-c"]
270 self._assert_python(pre_args)
271 self.assertRaises(FileNotFoundError, self._assert_python, pre_args,
272 executable="doesnotexist")
273
274 @unittest.skipIf(mswindows, "executable argument replaces shell")
275 def test_executable_replaces_shell(self):
276 # Check that the executable argument replaces the default shell
277 # when shell=True.
278 self._assert_python([], executable=sys.executable, shell=True)
279
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700280 # For use in the test_cwd* tests below.
281 def _normalize_cwd(self, cwd):
282 # Normalize an expected cwd (for Tru64 support).
283 # We can't use os.path.realpath since it doesn't expand Tru64 {memb}
284 # strings. See bug #1063571.
285 original_cwd = os.getcwd()
286 os.chdir(cwd)
287 cwd = os.getcwd()
288 os.chdir(original_cwd)
289 return cwd
290
291 # For use in the test_cwd* tests below.
292 def _split_python_path(self):
293 # Return normalized (python_dir, python_base).
294 python_path = os.path.realpath(sys.executable)
295 return os.path.split(python_path)
296
297 # For use in the test_cwd* tests below.
298 def _assert_cwd(self, expected_cwd, python_arg, **kwargs):
299 # Invoke Python via Popen, and assert that (1) the call succeeds,
300 # and that (2) the current working directory of the child process
301 # matches *expected_cwd*.
302 p = subprocess.Popen([python_arg, "-c",
303 "import os, sys; "
304 "sys.stdout.write(os.getcwd()); "
305 "sys.exit(47)"],
306 stdout=subprocess.PIPE,
307 **kwargs)
308 self.addCleanup(p.stdout.close)
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000309 p.wait()
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700310 self.assertEqual(47, p.returncode)
311 normcase = os.path.normcase
312 self.assertEqual(normcase(expected_cwd),
313 normcase(p.stdout.read().decode("utf-8")))
314
315 def test_cwd(self):
316 # Check that cwd changes the cwd for the child process.
317 temp_dir = tempfile.gettempdir()
318 temp_dir = self._normalize_cwd(temp_dir)
319 self._assert_cwd(temp_dir, sys.executable, cwd=temp_dir)
320
Chris Jerdonekc2cd6262012-09-30 09:45:00 -0700321 @unittest.skipIf(mswindows, "pending resolution of issue #15533")
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700322 def test_cwd_with_relative_arg(self):
323 # Check that Popen looks for args[0] relative to cwd if args[0]
324 # is relative.
325 python_dir, python_base = self._split_python_path()
326 rel_python = os.path.join(os.curdir, python_base)
327 with support.temp_cwd() as wrong_dir:
328 # Before calling with the correct cwd, confirm that the call fails
329 # without cwd and with the wrong cwd.
Chris Jerdonek28714c82012-09-30 02:15:37 -0700330 self.assertRaises(FileNotFoundError, subprocess.Popen,
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700331 [rel_python])
Chris Jerdonek28714c82012-09-30 02:15:37 -0700332 self.assertRaises(FileNotFoundError, subprocess.Popen,
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700333 [rel_python], cwd=wrong_dir)
334 python_dir = self._normalize_cwd(python_dir)
335 self._assert_cwd(python_dir, rel_python, cwd=python_dir)
336
Chris Jerdonekc2cd6262012-09-30 09:45:00 -0700337 @unittest.skipIf(mswindows, "pending resolution of issue #15533")
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700338 def test_cwd_with_relative_executable(self):
339 # Check that Popen looks for executable relative to cwd if executable
340 # is relative (and that executable takes precedence over args[0]).
341 python_dir, python_base = self._split_python_path()
342 rel_python = os.path.join(os.curdir, python_base)
343 doesntexist = "somethingyoudonthave"
344 with support.temp_cwd() as wrong_dir:
345 # Before calling with the correct cwd, confirm that the call fails
346 # without cwd and with the wrong cwd.
Chris Jerdonek28714c82012-09-30 02:15:37 -0700347 self.assertRaises(FileNotFoundError, subprocess.Popen,
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700348 [doesntexist], executable=rel_python)
Chris Jerdonek28714c82012-09-30 02:15:37 -0700349 self.assertRaises(FileNotFoundError, subprocess.Popen,
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700350 [doesntexist], executable=rel_python,
351 cwd=wrong_dir)
352 python_dir = self._normalize_cwd(python_dir)
353 self._assert_cwd(python_dir, doesntexist, executable=rel_python,
354 cwd=python_dir)
355
356 def test_cwd_with_absolute_arg(self):
357 # Check that Popen can find the executable when the cwd is wrong
358 # if args[0] is an absolute path.
359 python_dir, python_base = self._split_python_path()
360 abs_python = os.path.join(python_dir, python_base)
361 rel_python = os.path.join(os.curdir, python_base)
362 with script_helper.temp_dir() as wrong_dir:
363 # Before calling with an absolute path, confirm that using a
364 # relative path fails.
Chris Jerdonek28714c82012-09-30 02:15:37 -0700365 self.assertRaises(FileNotFoundError, subprocess.Popen,
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700366 [rel_python], cwd=wrong_dir)
367 wrong_dir = self._normalize_cwd(wrong_dir)
368 self._assert_cwd(wrong_dir, abs_python, cwd=wrong_dir)
369
Vinay Sajip7ded1f02012-05-26 03:45:29 +0100370 @unittest.skipIf(sys.base_prefix != sys.prefix,
371 'Test is not venv-compatible')
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000372 def test_executable_with_cwd(self):
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700373 python_dir, python_base = self._split_python_path()
374 python_dir = self._normalize_cwd(python_dir)
375 self._assert_cwd(python_dir, "somethingyoudonthave",
376 executable=sys.executable, cwd=python_dir)
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000377
Vinay Sajip7ded1f02012-05-26 03:45:29 +0100378 @unittest.skipIf(sys.base_prefix != sys.prefix,
379 'Test is not venv-compatible')
Ezio Melotti184bdfb2010-02-18 09:37:05 +0000380 @unittest.skipIf(sysconfig.is_python_build(),
381 "need an installed Python. See #7774")
382 def test_executable_without_cwd(self):
383 # For a normal installation, it should work without 'cwd'
384 # argument. For test runs in the build directory, see #7774.
Chris Jerdonekec3ea942012-09-30 00:10:28 -0700385 self._assert_cwd('', "somethingyoudonthave", executable=sys.executable)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000386
387 def test_stdin_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000388 # stdin redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000389 p = subprocess.Popen([sys.executable, "-c",
390 'import sys; sys.exit(sys.stdin.read() == "pear")'],
391 stdin=subprocess.PIPE)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000392 p.stdin.write(b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000393 p.stdin.close()
394 p.wait()
395 self.assertEqual(p.returncode, 1)
396
397 def test_stdin_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000398 # stdin is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000399 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000400 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000401 d = tf.fileno()
Antoine Pitrou9cadb1b2008-09-15 23:02:56 +0000402 os.write(d, b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000403 os.lseek(d, 0, 0)
404 p = subprocess.Popen([sys.executable, "-c",
405 'import sys; sys.exit(sys.stdin.read() == "pear")'],
406 stdin=d)
407 p.wait()
408 self.assertEqual(p.returncode, 1)
409
410 def test_stdin_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000411 # stdin is set to open file object
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000412 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000413 self.addCleanup(tf.close)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000414 tf.write(b"pear")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000415 tf.seek(0)
416 p = subprocess.Popen([sys.executable, "-c",
417 'import sys; sys.exit(sys.stdin.read() == "pear")'],
418 stdin=tf)
419 p.wait()
420 self.assertEqual(p.returncode, 1)
421
422 def test_stdout_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000423 # stdout redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000424 p = subprocess.Popen([sys.executable, "-c",
425 'import sys; sys.stdout.write("orange")'],
426 stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000427 self.addCleanup(p.stdout.close)
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000428 self.assertEqual(p.stdout.read(), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000429
430 def test_stdout_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000431 # stdout is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000432 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000433 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000434 d = tf.fileno()
435 p = subprocess.Popen([sys.executable, "-c",
436 'import sys; sys.stdout.write("orange")'],
437 stdout=d)
438 p.wait()
439 os.lseek(d, 0, 0)
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000440 self.assertEqual(os.read(d, 1024), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000441
442 def test_stdout_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000443 # stdout is set to open file object
Tim Peterse718f612004-10-12 21:51:32 +0000444 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000445 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000446 p = subprocess.Popen([sys.executable, "-c",
447 'import sys; sys.stdout.write("orange")'],
448 stdout=tf)
449 p.wait()
450 tf.seek(0)
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000451 self.assertEqual(tf.read(), b"orange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000452
453 def test_stderr_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000454 # stderr redirection
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000455 p = subprocess.Popen([sys.executable, "-c",
456 'import sys; sys.stderr.write("strawberry")'],
457 stderr=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000458 self.addCleanup(p.stderr.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000459 self.assertStderrEqual(p.stderr.read(), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000460
461 def test_stderr_filedes(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000462 # stderr is set to open file descriptor
Tim Peterse718f612004-10-12 21:51:32 +0000463 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000464 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000465 d = tf.fileno()
466 p = subprocess.Popen([sys.executable, "-c",
467 'import sys; sys.stderr.write("strawberry")'],
468 stderr=d)
469 p.wait()
470 os.lseek(d, 0, 0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000471 self.assertStderrEqual(os.read(d, 1024), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000472
473 def test_stderr_fileobj(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000474 # stderr is set to open file object
Tim Peterse718f612004-10-12 21:51:32 +0000475 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000476 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000477 p = subprocess.Popen([sys.executable, "-c",
478 'import sys; sys.stderr.write("strawberry")'],
479 stderr=tf)
480 p.wait()
481 tf.seek(0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000482 self.assertStderrEqual(tf.read(), b"strawberry")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000483
484 def test_stdout_stderr_pipe(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000485 # capture stdout and stderr to the same pipe
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000486 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000487 'import sys;'
488 'sys.stdout.write("apple");'
489 'sys.stdout.flush();'
490 'sys.stderr.write("orange")'],
491 stdout=subprocess.PIPE,
492 stderr=subprocess.STDOUT)
Brian Curtin3c6a9512010-11-05 03:58:52 +0000493 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000494 self.assertStderrEqual(p.stdout.read(), b"appleorange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000495
496 def test_stdout_stderr_file(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000497 # capture stdout and stderr to the same open file
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000498 tf = tempfile.TemporaryFile()
Benjamin Petersoncc221b22010-10-31 02:06:21 +0000499 self.addCleanup(tf.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000500 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000501 'import sys;'
502 'sys.stdout.write("apple");'
503 'sys.stdout.flush();'
504 'sys.stderr.write("orange")'],
505 stdout=tf,
506 stderr=tf)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000507 p.wait()
508 tf.seek(0)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000509 self.assertStderrEqual(tf.read(), b"appleorange")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000510
Thomas Wouters89f507f2006-12-13 04:49:30 +0000511 def test_stdout_filedes_of_stdout(self):
512 # stdout is set to 1 (#1531862).
Ezio Melotti42a541b2013-03-11 05:53:34 +0200513 # To avoid printing the text on stdout, we do something similar to
514 # test_stdout_none (see above). The parent subprocess calls the child
515 # subprocess passing stdout=1, and this test uses stdout=PIPE in
516 # order to capture and check the output of the parent. See #11963.
517 code = ('import sys, subprocess; '
518 'rc = subprocess.call([sys.executable, "-c", '
519 ' "import os, sys; sys.exit(os.write(sys.stdout.fileno(), '
520 'b\'test with stdout=1\'))"], stdout=1); '
521 'assert rc == 18')
522 p = subprocess.Popen([sys.executable, "-c", code],
523 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
524 self.addCleanup(p.stdout.close)
525 self.addCleanup(p.stderr.close)
526 out, err = p.communicate()
527 self.assertEqual(p.returncode, 0, err)
528 self.assertEqual(out.rstrip(), b'test with stdout=1')
Thomas Wouters89f507f2006-12-13 04:49:30 +0000529
Ross Lagerwallba102ec2011-03-16 18:40:25 +0200530 def test_stdout_devnull(self):
531 p = subprocess.Popen([sys.executable, "-c",
532 'for i in range(10240):'
533 'print("x" * 1024)'],
534 stdout=subprocess.DEVNULL)
535 p.wait()
536 self.assertEqual(p.stdout, None)
537
538 def test_stderr_devnull(self):
539 p = subprocess.Popen([sys.executable, "-c",
540 'import sys\n'
541 'for i in range(10240):'
542 'sys.stderr.write("x" * 1024)'],
543 stderr=subprocess.DEVNULL)
544 p.wait()
545 self.assertEqual(p.stderr, None)
546
547 def test_stdin_devnull(self):
548 p = subprocess.Popen([sys.executable, "-c",
549 'import sys;'
550 'sys.stdin.read(1)'],
551 stdin=subprocess.DEVNULL)
552 p.wait()
553 self.assertEqual(p.stdin, None)
554
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000555 def test_env(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000556 newenv = os.environ.copy()
557 newenv["FRUIT"] = "orange"
Victor Stinnerf1512a22011-06-21 17:18:38 +0200558 with subprocess.Popen([sys.executable, "-c",
559 'import sys,os;'
560 'sys.stdout.write(os.getenv("FRUIT"))'],
561 stdout=subprocess.PIPE,
562 env=newenv) as p:
563 stdout, stderr = p.communicate()
564 self.assertEqual(stdout, b"orange")
565
Victor Stinner62d51182011-06-23 01:02:25 +0200566 # Windows requires at least the SYSTEMROOT environment variable to start
567 # Python
568 @unittest.skipIf(sys.platform == 'win32',
569 'cannot test an empty env on Windows')
Victor Stinner237e5cb2011-06-22 21:28:43 +0200570 @unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') is not None,
Victor Stinner372309a2011-06-21 21:59:06 +0200571 'the python library cannot be loaded '
572 'with an empty environment')
Victor Stinnerf1512a22011-06-21 17:18:38 +0200573 def test_empty_env(self):
574 with subprocess.Popen([sys.executable, "-c",
575 'import os; '
Victor Stinner372309a2011-06-21 21:59:06 +0200576 'print(list(os.environ.keys()))'],
Victor Stinnerf1512a22011-06-21 17:18:38 +0200577 stdout=subprocess.PIPE,
578 env={}) as p:
579 stdout, stderr = p.communicate()
Victor Stinner237e5cb2011-06-22 21:28:43 +0200580 self.assertIn(stdout.strip(),
581 (b"[]",
582 # Mac OS X adds __CF_USER_TEXT_ENCODING variable to an empty
583 # environment
584 b"['__CF_USER_TEXT_ENCODING']"))
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000585
Peter Astrandcbac93c2005-03-03 20:24:28 +0000586 def test_communicate_stdin(self):
587 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000588 'import sys;'
589 'sys.exit(sys.stdin.read() == "pear")'],
Peter Astrandcbac93c2005-03-03 20:24:28 +0000590 stdin=subprocess.PIPE)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000591 p.communicate(b"pear")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000592 self.assertEqual(p.returncode, 1)
593
594 def test_communicate_stdout(self):
595 p = subprocess.Popen([sys.executable, "-c",
596 'import sys; sys.stdout.write("pineapple")'],
597 stdout=subprocess.PIPE)
598 (stdout, stderr) = p.communicate()
Guido van Rossumfa0054a2007-05-24 04:05:35 +0000599 self.assertEqual(stdout, b"pineapple")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000600 self.assertEqual(stderr, None)
601
602 def test_communicate_stderr(self):
603 p = subprocess.Popen([sys.executable, "-c",
604 'import sys; sys.stderr.write("pineapple")'],
605 stderr=subprocess.PIPE)
606 (stdout, stderr) = p.communicate()
607 self.assertEqual(stdout, None)
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000608 self.assertStderrEqual(stderr, b"pineapple")
Peter Astrandcbac93c2005-03-03 20:24:28 +0000609
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000610 def test_communicate(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000611 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000612 'import sys,os;'
613 'sys.stderr.write("pineapple");'
614 'sys.stdout.write(sys.stdin.read())'],
615 stdin=subprocess.PIPE,
616 stdout=subprocess.PIPE,
617 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000618 self.addCleanup(p.stdout.close)
619 self.addCleanup(p.stderr.close)
620 self.addCleanup(p.stdin.close)
Georg Brandl1abcbf82008-07-01 19:28:43 +0000621 (stdout, stderr) = p.communicate(b"banana")
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000622 self.assertEqual(stdout, b"banana")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000623 self.assertStderrEqual(stderr, b"pineapple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000624
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400625 def test_communicate_timeout(self):
626 p = subprocess.Popen([sys.executable, "-c",
627 'import sys,os,time;'
628 'sys.stderr.write("pineapple\\n");'
629 'time.sleep(1);'
630 'sys.stderr.write("pear\\n");'
631 'sys.stdout.write(sys.stdin.read())'],
632 universal_newlines=True,
633 stdin=subprocess.PIPE,
634 stdout=subprocess.PIPE,
635 stderr=subprocess.PIPE)
636 self.assertRaises(subprocess.TimeoutExpired, p.communicate, "banana",
637 timeout=0.3)
638 # Make sure we can keep waiting for it, and that we get the whole output
639 # after it completes.
640 (stdout, stderr) = p.communicate()
641 self.assertEqual(stdout, "banana")
642 self.assertStderrEqual(stderr.encode(), b"pineapple\npear\n")
643
644 def test_communicate_timeout_large_ouput(self):
Ross Lagerwall003c7a32012-02-12 09:02:01 +0200645 # Test an expiring timeout while the child is outputting lots of data.
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400646 p = subprocess.Popen([sys.executable, "-c",
647 'import sys,os,time;'
648 'sys.stdout.write("a" * (64 * 1024));'
649 'time.sleep(0.2);'
650 'sys.stdout.write("a" * (64 * 1024));'
651 'time.sleep(0.2);'
652 'sys.stdout.write("a" * (64 * 1024));'
653 'time.sleep(0.2);'
654 'sys.stdout.write("a" * (64 * 1024));'],
655 stdout=subprocess.PIPE)
656 self.assertRaises(subprocess.TimeoutExpired, p.communicate, timeout=0.4)
657 (stdout, _) = p.communicate()
658 self.assertEqual(len(stdout), 4 * 64 * 1024)
659
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000660 # Test for the fd leak reported in http://bugs.python.org/issue2791.
661 def test_communicate_pipe_fd_leak(self):
Victor Stinner667d4b52010-12-25 22:40:32 +0000662 for stdin_pipe in (False, True):
663 for stdout_pipe in (False, True):
664 for stderr_pipe in (False, True):
665 options = {}
666 if stdin_pipe:
667 options['stdin'] = subprocess.PIPE
668 if stdout_pipe:
669 options['stdout'] = subprocess.PIPE
670 if stderr_pipe:
671 options['stderr'] = subprocess.PIPE
672 if not options:
673 continue
674 p = subprocess.Popen((sys.executable, "-c", "pass"), **options)
675 p.communicate()
676 if p.stdin is not None:
677 self.assertTrue(p.stdin.closed)
678 if p.stdout is not None:
679 self.assertTrue(p.stdout.closed)
680 if p.stderr is not None:
681 self.assertTrue(p.stderr.closed)
Georg Brandlf08a9dd2008-06-10 16:57:31 +0000682
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000683 def test_communicate_returns(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000684 # communicate() should return None if no redirection is active
Tim Peters3b01a702004-10-12 22:19:32 +0000685 p = subprocess.Popen([sys.executable, "-c",
686 "import sys; sys.exit(47)"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000687 (stdout, stderr) = p.communicate()
688 self.assertEqual(stdout, None)
689 self.assertEqual(stderr, None)
690
691 def test_communicate_pipe_buf(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000692 # communicate() with writes larger than pipe_buf
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000693 # This test will probably deadlock rather than fail, if
Tim Peterse718f612004-10-12 21:51:32 +0000694 # communicate() does not work properly.
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000695 x, y = os.pipe()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000696 os.close(x)
697 os.close(y)
698 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000699 'import sys,os;'
700 'sys.stdout.write(sys.stdin.read(47));'
Charles-François Natali2d517212011-05-29 16:36:44 +0200701 'sys.stderr.write("x" * %d);'
702 'sys.stdout.write(sys.stdin.read())' %
703 support.PIPE_MAX_SIZE],
Guido van Rossum98297ee2007-11-06 21:34:58 +0000704 stdin=subprocess.PIPE,
705 stdout=subprocess.PIPE,
706 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000707 self.addCleanup(p.stdout.close)
708 self.addCleanup(p.stderr.close)
709 self.addCleanup(p.stdin.close)
Charles-François Natali2d517212011-05-29 16:36:44 +0200710 string_to_write = b"a" * support.PIPE_MAX_SIZE
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000711 (stdout, stderr) = p.communicate(string_to_write)
712 self.assertEqual(stdout, string_to_write)
713
714 def test_writes_before_communicate(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000715 # stdin.write before communicate()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000716 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000717 'import sys,os;'
718 'sys.stdout.write(sys.stdin.read())'],
719 stdin=subprocess.PIPE,
720 stdout=subprocess.PIPE,
721 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +0000722 self.addCleanup(p.stdout.close)
723 self.addCleanup(p.stderr.close)
724 self.addCleanup(p.stdin.close)
Guido van Rossumbb839ef2007-08-27 23:58:21 +0000725 p.stdin.write(b"banana")
726 (stdout, stderr) = p.communicate(b"split")
Guido van Rossumc9e363c2007-05-15 23:18:55 +0000727 self.assertEqual(stdout, b"bananasplit")
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000728 self.assertStderrEqual(stderr, b"")
Tim Peterse718f612004-10-12 21:51:32 +0000729
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000730 def test_universal_newlines(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000731 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000732 'import sys,os;' + SETBINARY +
Antoine Pitrouec2d2692012-08-05 00:23:40 +0200733 'buf = sys.stdout.buffer;'
734 'buf.write(sys.stdin.readline().encode());'
735 'buf.flush();'
736 'buf.write(b"line2\\n");'
737 'buf.flush();'
738 'buf.write(sys.stdin.read().encode());'
739 'buf.flush();'
740 'buf.write(b"line4\\n");'
741 'buf.flush();'
742 'buf.write(b"line5\\r\\n");'
743 'buf.flush();'
744 'buf.write(b"line6\\r");'
745 'buf.flush();'
746 'buf.write(b"\\nline7");'
747 'buf.flush();'
748 'buf.write(b"\\nline8");'],
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200749 stdin=subprocess.PIPE,
Guido van Rossum98297ee2007-11-06 21:34:58 +0000750 stdout=subprocess.PIPE,
751 universal_newlines=1)
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200752 p.stdin.write("line1\n")
753 self.assertEqual(p.stdout.readline(), "line1\n")
754 p.stdin.write("line3\n")
755 p.stdin.close()
Brian Curtin3c6a9512010-11-05 03:58:52 +0000756 self.addCleanup(p.stdout.close)
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200757 self.assertEqual(p.stdout.readline(),
758 "line2\n")
759 self.assertEqual(p.stdout.read(6),
760 "line3\n")
761 self.assertEqual(p.stdout.read(),
762 "line4\nline5\nline6\nline7\nline8")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000763
764 def test_universal_newlines_communicate(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000765 # universal newlines through communicate()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000766 p = subprocess.Popen([sys.executable, "-c",
Guido van Rossum98297ee2007-11-06 21:34:58 +0000767 'import sys,os;' + SETBINARY +
Antoine Pitrouec2d2692012-08-05 00:23:40 +0200768 'buf = sys.stdout.buffer;'
769 'buf.write(b"line2\\n");'
770 'buf.flush();'
771 'buf.write(b"line4\\n");'
772 'buf.flush();'
773 'buf.write(b"line5\\r\\n");'
774 'buf.flush();'
775 'buf.write(b"line6\\r");'
776 'buf.flush();'
777 'buf.write(b"\\nline7");'
778 'buf.flush();'
779 'buf.write(b"\\nline8");'],
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200780 stderr=subprocess.PIPE,
781 stdout=subprocess.PIPE,
Guido van Rossum98297ee2007-11-06 21:34:58 +0000782 universal_newlines=1)
Brian Curtin19a53792010-11-05 17:09:05 +0000783 self.addCleanup(p.stdout.close)
784 self.addCleanup(p.stderr.close)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000785 (stdout, stderr) = p.communicate()
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200786 self.assertEqual(stdout,
787 "line2\nline4\nline5\nline6\nline7\nline8")
788
789 def test_universal_newlines_communicate_stdin(self):
790 # universal newlines through communicate(), with only stdin
791 p = subprocess.Popen([sys.executable, "-c",
Andrew Svetlov47ec25d2012-08-19 16:25:37 +0300792 'import sys,os;' + SETBINARY + textwrap.dedent('''
793 s = sys.stdin.readline()
794 assert s == "line1\\n", repr(s)
795 s = sys.stdin.read()
796 assert s == "line3\\n", repr(s)
797 ''')],
Antoine Pitrouab85ff32011-07-23 22:03:45 +0200798 stdin=subprocess.PIPE,
799 universal_newlines=1)
800 (stdout, stderr) = p.communicate("line1\nline3\n")
801 self.assertEqual(p.returncode, 0)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000802
Andrew Svetlovf3765072012-08-14 18:35:17 +0300803 def test_universal_newlines_communicate_input_none(self):
804 # Test communicate(input=None) with universal newlines.
805 #
806 # We set stdout to PIPE because, as of this writing, a different
807 # code path is tested when the number of pipes is zero or one.
808 p = subprocess.Popen([sys.executable, "-c", "pass"],
809 stdin=subprocess.PIPE,
810 stdout=subprocess.PIPE,
811 universal_newlines=True)
812 p.communicate()
813 self.assertEqual(p.returncode, 0)
814
Andrew Svetlov5395d2f2012-08-15 22:46:43 +0300815 def test_universal_newlines_communicate_stdin_stdout_stderr(self):
Andrew Svetlov47ec25d2012-08-19 16:25:37 +0300816 # universal newlines through communicate(), with stdin, stdout, stderr
Andrew Svetlov5395d2f2012-08-15 22:46:43 +0300817 p = subprocess.Popen([sys.executable, "-c",
Andrew Svetlov47ec25d2012-08-19 16:25:37 +0300818 'import sys,os;' + SETBINARY + textwrap.dedent('''
819 s = sys.stdin.buffer.readline()
820 sys.stdout.buffer.write(s)
821 sys.stdout.buffer.write(b"line2\\r")
822 sys.stderr.buffer.write(b"eline2\\n")
823 s = sys.stdin.buffer.read()
824 sys.stdout.buffer.write(s)
825 sys.stdout.buffer.write(b"line4\\n")
826 sys.stdout.buffer.write(b"line5\\r\\n")
827 sys.stderr.buffer.write(b"eline6\\r")
828 sys.stderr.buffer.write(b"eline7\\r\\nz")
829 ''')],
Andrew Svetlov5395d2f2012-08-15 22:46:43 +0300830 stdin=subprocess.PIPE,
831 stderr=subprocess.PIPE,
832 stdout=subprocess.PIPE,
Andrew Svetlov47ec25d2012-08-19 16:25:37 +0300833 universal_newlines=True)
Andrew Svetlov5395d2f2012-08-15 22:46:43 +0300834 self.addCleanup(p.stdout.close)
835 self.addCleanup(p.stderr.close)
836 (stdout, stderr) = p.communicate("line1\nline3\n")
837 self.assertEqual(p.returncode, 0)
Andrew Svetlov943c5b32012-08-16 20:17:47 +0300838 self.assertEqual("line1\nline2\nline3\nline4\nline5\n", stdout)
Andrew Svetlov5395d2f2012-08-15 22:46:43 +0300839 # Python debug build push something like "[42442 refs]\n"
840 # to stderr at exit of subprocess.
Andrew Svetlov943c5b32012-08-16 20:17:47 +0300841 # Don't use assertStderrEqual because it strips CR and LF from output.
842 self.assertTrue(stderr.startswith("eline2\neline6\neline7\n"))
Andrew Svetlov5395d2f2012-08-15 22:46:43 +0300843
Andrew Svetlov82860712012-08-19 22:13:41 +0300844 def test_universal_newlines_communicate_encodings(self):
845 # Check that universal newlines mode works for various encodings,
846 # in particular for encodings in the UTF-16 and UTF-32 families.
847 # See issue #15595.
848 #
849 # UTF-16 and UTF-32-BE are sufficient to check both with BOM and
850 # without, and UTF-16 and UTF-32.
851 for encoding in ['utf-16', 'utf-32-be']:
852 old_getpreferredencoding = locale.getpreferredencoding
853 # Indirectly via io.TextIOWrapper, Popen() defaults to
854 # locale.getpreferredencoding(False) and earlier in Python 3.2 to
855 # locale.getpreferredencoding().
856 def getpreferredencoding(do_setlocale=True):
857 return encoding
858 code = ("import sys; "
859 r"sys.stdout.buffer.write('1\r\n2\r3\n4'.encode('%s'))" %
860 encoding)
861 args = [sys.executable, '-c', code]
862 try:
863 locale.getpreferredencoding = getpreferredencoding
864 # We set stdin to be non-None because, as of this writing,
865 # a different code path is used when the number of pipes is
866 # zero or one.
867 popen = subprocess.Popen(args, universal_newlines=True,
868 stdin=subprocess.PIPE,
869 stdout=subprocess.PIPE)
870 stdout, stderr = popen.communicate(input='')
871 finally:
872 locale.getpreferredencoding = old_getpreferredencoding
Andrew Svetlov82860712012-08-19 22:13:41 +0300873 self.assertEqual(stdout, '1\n2\n3\n4')
874
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000875 def test_no_leaking(self):
Tim Peters7b759da2004-10-12 22:29:54 +0000876 # Make sure we leak no resources
Antoine Pitrou8db30272010-09-18 22:38:48 +0000877 if not mswindows:
Peter Astrandf7f1bb72005-03-03 20:47:37 +0000878 max_handles = 1026 # too much for most UNIX systems
879 else:
Antoine Pitrou8db30272010-09-18 22:38:48 +0000880 max_handles = 2050 # too much for (at least some) Windows setups
881 handles = []
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400882 tmpdir = tempfile.mkdtemp()
Antoine Pitrou8db30272010-09-18 22:38:48 +0000883 try:
884 for i in range(max_handles):
885 try:
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400886 tmpfile = os.path.join(tmpdir, support.TESTFN)
887 handles.append(os.open(tmpfile, os.O_WRONLY|os.O_CREAT))
Antoine Pitrou8db30272010-09-18 22:38:48 +0000888 except OSError as e:
889 if e.errno != errno.EMFILE:
890 raise
891 break
892 else:
893 self.skipTest("failed to reach the file descriptor limit "
894 "(tried %d)" % max_handles)
895 # Close a couple of them (should be enough for a subprocess)
896 for i in range(10):
897 os.close(handles.pop())
898 # Loop creating some subprocesses. If one of them leaks some fds,
899 # the next loop iteration will fail by reaching the max fd limit.
900 for i in range(15):
901 p = subprocess.Popen([sys.executable, "-c",
902 "import sys;"
903 "sys.stdout.write(sys.stdin.read())"],
904 stdin=subprocess.PIPE,
905 stdout=subprocess.PIPE,
906 stderr=subprocess.PIPE)
907 data = p.communicate(b"lime")[0]
908 self.assertEqual(data, b"lime")
909 finally:
910 for h in handles:
911 os.close(h)
Gregory P. Smith81ce6852011-03-15 02:04:11 -0400912 shutil.rmtree(tmpdir)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000913
914 def test_list2cmdline(self):
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000915 self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),
916 '"a b c" d e')
917 self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),
918 'ab\\"c \\ d')
Christian Heimesfdab48e2008-01-20 09:06:41 +0000919 self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']),
920 'ab\\"c " \\\\" d')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000921 self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),
922 'a\\\\\\b "de fg" h')
923 self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),
924 'a\\\\\\"b c d')
925 self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),
926 '"a\\\\b c" d e')
927 self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),
928 '"a\\\\b\\ c" d e')
Thomas Woutersfc7bb8c2007-01-15 15:49:28 +0000929 self.assertEqual(subprocess.list2cmdline(['ab', '']),
930 'ab ""')
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000931
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000932 def test_poll(self):
Ross Lagerwallab66d2a2012-02-12 09:01:30 +0200933 p = subprocess.Popen([sys.executable, "-c",
Ross Lagerwalle7ad4192012-02-22 06:02:07 +0200934 "import os; os.read(0, 1)"],
935 stdin=subprocess.PIPE)
Ross Lagerwallab66d2a2012-02-12 09:01:30 +0200936 self.addCleanup(p.stdin.close)
937 self.assertIsNone(p.poll())
938 os.write(p.stdin.fileno(), b'A')
939 p.wait()
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000940 # Subsequent invocations should just return the returncode
941 self.assertEqual(p.poll(), 0)
942
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000943 def test_wait(self):
Ross Lagerwallab66d2a2012-02-12 09:01:30 +0200944 p = subprocess.Popen([sys.executable, "-c", "pass"])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +0000945 self.assertEqual(p.wait(), 0)
946 # Subsequent invocations should just return the returncode
947 self.assertEqual(p.wait(), 0)
Tim Peterse718f612004-10-12 21:51:32 +0000948
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400949 def test_wait_timeout(self):
950 p = subprocess.Popen([sys.executable,
Antoine Pitroudc49b2b2013-05-19 15:55:40 +0200951 "-c", "import time; time.sleep(0.3)"])
Reid Kleckner2b228f02011-03-16 16:57:54 -0400952 with self.assertRaises(subprocess.TimeoutExpired) as c:
Antoine Pitroudc49b2b2013-05-19 15:55:40 +0200953 p.wait(timeout=0.0001)
954 self.assertIn("0.0001", str(c.exception)) # For coverage of __str__.
Reid Klecknerda9ac722011-03-16 17:08:21 -0400955 # Some heavily loaded buildbots (sparc Debian 3.x) require this much
956 # time to start.
957 self.assertEqual(p.wait(timeout=3), 0)
Reid Kleckner31aa7dd2011-03-14 12:02:10 -0400958
Peter Astrand738131d2004-11-30 21:04:45 +0000959 def test_invalid_bufsize(self):
960 # an invalid type of the bufsize argument should raise
961 # TypeError.
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000962 with self.assertRaises(TypeError):
Peter Astrand738131d2004-11-30 21:04:45 +0000963 subprocess.Popen([sys.executable, "-c", "pass"], "orange")
Peter Astrand738131d2004-11-30 21:04:45 +0000964
Guido van Rossum46a05a72007-06-07 21:56:45 +0000965 def test_bufsize_is_none(self):
966 # bufsize=None should be the same as bufsize=0.
967 p = subprocess.Popen([sys.executable, "-c", "pass"], None)
968 self.assertEqual(p.wait(), 0)
969 # Again with keyword arg
970 p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None)
971 self.assertEqual(p.wait(), 0)
972
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000973 def test_leaking_fds_on_error(self):
974 # see bug #5179: Popen leaks file descriptors to PIPEs if
975 # the child fails to execute; this will eventually exhaust
976 # the maximum number of open fds. 1024 seems a very common
977 # value for that limit, but Windows has 2048, so we loop
978 # 1024 times (each call leaked two fds).
979 for i in range(1024):
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000980 # Windows raises IOError. Others raise OSError.
981 with self.assertRaises(EnvironmentError) as c:
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000982 subprocess.Popen(['nonexisting_i_hope'],
983 stdout=subprocess.PIPE,
984 stderr=subprocess.PIPE)
R David Murray384069c2011-03-13 22:26:53 -0400985 # ignore errors that indicate the command was not found
R David Murray6924bd72011-03-13 22:48:55 -0400986 if c.exception.errno not in (errno.ENOENT, errno.EACCES):
Florent Xiclunab1e94e82010-02-27 22:12:37 +0000987 raise c.exception
Benjamin Petersond75fcb42009-02-19 04:22:03 +0000988
Victor Stinnerb3693582010-05-21 20:13:12 +0000989 def test_issue8780(self):
990 # Ensure that stdout is inherited from the parent
991 # if stdout=PIPE is not used
992 code = ';'.join((
993 'import subprocess, sys',
994 'retcode = subprocess.call('
995 "[sys.executable, '-c', 'print(\"Hello World!\")'])",
996 'assert retcode == 0'))
997 output = subprocess.check_output([sys.executable, '-c', code])
Ezio Melottib3aedd42010-11-20 19:04:17 +0000998 self.assertTrue(output.startswith(b'Hello World!'), ascii(output))
Victor Stinnerb3693582010-05-21 20:13:12 +0000999
Tim Goldenaf5ac392010-08-06 13:03:56 +00001000 def test_handles_closed_on_exception(self):
1001 # If CreateProcess exits with an error, ensure the
1002 # duplicate output handles are released
1003 ifhandle, ifname = mkstemp()
1004 ofhandle, ofname = mkstemp()
1005 efhandle, efname = mkstemp()
1006 try:
1007 subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle,
1008 stderr=efhandle)
1009 except OSError:
1010 os.close(ifhandle)
1011 os.remove(ifname)
1012 os.close(ofhandle)
1013 os.remove(ofname)
1014 os.close(efhandle)
1015 os.remove(efname)
1016 self.assertFalse(os.path.exists(ifname))
1017 self.assertFalse(os.path.exists(ofname))
1018 self.assertFalse(os.path.exists(efname))
1019
Ross Lagerwall4f61b022011-04-05 15:34:00 +02001020 def test_communicate_epipe(self):
1021 # Issue 10963: communicate() should hide EPIPE
1022 p = subprocess.Popen([sys.executable, "-c", 'pass'],
1023 stdin=subprocess.PIPE,
1024 stdout=subprocess.PIPE,
1025 stderr=subprocess.PIPE)
1026 self.addCleanup(p.stdout.close)
1027 self.addCleanup(p.stderr.close)
1028 self.addCleanup(p.stdin.close)
1029 p.communicate(b"x" * 2**20)
1030
1031 def test_communicate_epipe_only_stdin(self):
1032 # Issue 10963: communicate() should hide EPIPE
1033 p = subprocess.Popen([sys.executable, "-c", 'pass'],
1034 stdin=subprocess.PIPE)
1035 self.addCleanup(p.stdin.close)
Ross Lagerwallab66d2a2012-02-12 09:01:30 +02001036 p.wait()
Ross Lagerwall4f61b022011-04-05 15:34:00 +02001037 p.communicate(b"x" * 2**20)
1038
Ross Lagerwallab66d2a2012-02-12 09:01:30 +02001039 @unittest.skipUnless(hasattr(signal, 'SIGUSR1'),
1040 "Requires signal.SIGUSR1")
1041 @unittest.skipUnless(hasattr(os, 'kill'),
1042 "Requires os.kill")
1043 @unittest.skipUnless(hasattr(os, 'getppid'),
1044 "Requires os.getppid")
Victor Stinner2cfb6f32011-07-05 14:00:56 +02001045 def test_communicate_eintr(self):
1046 # Issue #12493: communicate() should handle EINTR
1047 def handler(signum, frame):
1048 pass
Ross Lagerwallab66d2a2012-02-12 09:01:30 +02001049 old_handler = signal.signal(signal.SIGUSR1, handler)
1050 self.addCleanup(signal.signal, signal.SIGUSR1, old_handler)
Victor Stinner2cfb6f32011-07-05 14:00:56 +02001051
Ross Lagerwallab66d2a2012-02-12 09:01:30 +02001052 args = [sys.executable, "-c",
1053 'import os, signal;'
1054 'os.kill(os.getppid(), signal.SIGUSR1)']
Victor Stinner2cfb6f32011-07-05 14:00:56 +02001055 for stream in ('stdout', 'stderr'):
1056 kw = {stream: subprocess.PIPE}
1057 with subprocess.Popen(args, **kw) as process:
Ross Lagerwallab66d2a2012-02-12 09:01:30 +02001058 # communicate() will be interrupted by SIGUSR1
Victor Stinner2cfb6f32011-07-05 14:00:56 +02001059 process.communicate()
1060
Tim Peterse718f612004-10-12 21:51:32 +00001061
Gregory P. Smith3d8e7762012-11-10 22:32:22 -08001062 # This test is Linux-ish specific for simplicity to at least have
1063 # some coverage. It is not a platform specific bug.
1064 @unittest.skipUnless(os.path.isdir('/proc/%d/fd' % os.getpid()),
1065 "Linux specific")
1066 def test_failed_child_execute_fd_leak(self):
1067 """Test for the fork() failure fd leak reported in issue16327."""
1068 fd_directory = '/proc/%d/fd' % os.getpid()
1069 fds_before_popen = os.listdir(fd_directory)
1070 with self.assertRaises(PopenTestException):
1071 PopenExecuteChildRaises(
1072 [sys.executable, '-c', 'pass'], stdin=subprocess.PIPE,
1073 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1074
1075 # NOTE: This test doesn't verify that the real _execute_child
1076 # does not close the file descriptors itself on the way out
1077 # during an exception. Code inspection has confirmed that.
1078
1079 fds_after_exception = os.listdir(fd_directory)
1080 self.assertEqual(fds_before_popen, fds_after_exception)
1081
1082
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001083# context manager
1084class _SuppressCoreFiles(object):
1085 """Try to prevent core files from being created."""
1086 old_limit = None
Thomas Wouters0e3f5912006-08-11 14:57:12 +00001087
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001088 def __enter__(self):
1089 """Try to save previous ulimit, then set it to (0, 0)."""
Benjamin Peterson964561b2011-12-10 12:31:42 -05001090 if resource is not None:
1091 try:
1092 self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
1093 resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
1094 except (ValueError, resource.error):
1095 pass
Thomas Wouters0e3f5912006-08-11 14:57:12 +00001096
Ronald Oussoren102d11a2010-07-23 09:50:05 +00001097 if sys.platform == 'darwin':
1098 # Check if the 'Crash Reporter' on OSX was configured
1099 # in 'Developer' mode and warn that it will get triggered
1100 # when it is.
1101 #
1102 # This assumes that this context manager is used in tests
1103 # that might trigger the next manager.
1104 value = subprocess.Popen(['/usr/bin/defaults', 'read',
1105 'com.apple.CrashReporter', 'DialogType'],
1106 stdout=subprocess.PIPE).communicate()[0]
1107 if value.strip() == b'developer':
1108 print("this tests triggers the Crash Reporter, "
1109 "that is intentional", end='')
1110 sys.stdout.flush()
1111
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001112 def __exit__(self, *args):
1113 """Return core file behavior to default."""
1114 if self.old_limit is None:
1115 return
Benjamin Peterson964561b2011-12-10 12:31:42 -05001116 if resource is not None:
1117 try:
1118 resource.setrlimit(resource.RLIMIT_CORE, self.old_limit)
1119 except (ValueError, resource.error):
1120 pass
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001121
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001122
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001123@unittest.skipIf(mswindows, "POSIX specific tests")
Florent Xiclunac049d872010-03-27 22:47:23 +00001124class POSIXProcessTestCase(BaseTestCase):
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001125
Gregory P. Smith5591b022012-10-10 03:34:47 -07001126 def setUp(self):
1127 super().setUp()
1128 self._nonexistent_dir = "/_this/pa.th/does/not/exist"
1129
1130 def _get_chdir_exception(self):
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001131 try:
Gregory P. Smith5591b022012-10-10 03:34:47 -07001132 os.chdir(self._nonexistent_dir)
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001133 except OSError as e:
1134 # This avoids hard coding the errno value or the OS perror()
1135 # string and instead capture the exception that we want to see
1136 # below for comparison.
1137 desired_exception = e
Gregory P. Smith5591b022012-10-10 03:34:47 -07001138 desired_exception.strerror += ': ' + repr(self._nonexistent_dir)
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001139 else:
1140 self.fail("chdir to nonexistant directory %s succeeded." %
Gregory P. Smith5591b022012-10-10 03:34:47 -07001141 self._nonexistent_dir)
1142 return desired_exception
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001143
Gregory P. Smith5591b022012-10-10 03:34:47 -07001144 def test_exception_cwd(self):
1145 """Test error in the child raised in the parent for a bad cwd."""
1146 desired_exception = self._get_chdir_exception()
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001147 try:
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001148 p = subprocess.Popen([sys.executable, "-c", ""],
Gregory P. Smith5591b022012-10-10 03:34:47 -07001149 cwd=self._nonexistent_dir)
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001150 except OSError as e:
1151 # Test that the child process chdir failure actually makes
1152 # it up to the parent process as the correct exception.
1153 self.assertEqual(desired_exception.errno, e.errno)
1154 self.assertEqual(desired_exception.strerror, e.strerror)
1155 else:
1156 self.fail("Expected OSError: %s" % desired_exception)
1157
Gregory P. Smith5591b022012-10-10 03:34:47 -07001158 def test_exception_bad_executable(self):
1159 """Test error in the child raised in the parent for a bad executable."""
1160 desired_exception = self._get_chdir_exception()
1161 try:
1162 p = subprocess.Popen([sys.executable, "-c", ""],
1163 executable=self._nonexistent_dir)
1164 except OSError as e:
1165 # Test that the child process exec failure actually makes
1166 # it up to the parent process as the correct exception.
1167 self.assertEqual(desired_exception.errno, e.errno)
1168 self.assertEqual(desired_exception.strerror, e.strerror)
1169 else:
1170 self.fail("Expected OSError: %s" % desired_exception)
1171
1172 def test_exception_bad_args_0(self):
1173 """Test error in the child raised in the parent for a bad args[0]."""
1174 desired_exception = self._get_chdir_exception()
1175 try:
1176 p = subprocess.Popen([self._nonexistent_dir, "-c", ""])
1177 except OSError as e:
1178 # Test that the child process exec failure actually makes
1179 # it up to the parent process as the correct exception.
1180 self.assertEqual(desired_exception.errno, e.errno)
1181 self.assertEqual(desired_exception.strerror, e.strerror)
1182 else:
1183 self.fail("Expected OSError: %s" % desired_exception)
1184
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001185 def test_restore_signals(self):
1186 # Code coverage for both values of restore_signals to make sure it
1187 # at least does not blow up.
1188 # A test for behavior would be complex. Contributions welcome.
1189 subprocess.call([sys.executable, "-c", ""], restore_signals=True)
1190 subprocess.call([sys.executable, "-c", ""], restore_signals=False)
1191
1192 def test_start_new_session(self):
1193 # For code coverage of calling setsid(). We don't care if we get an
1194 # EPERM error from it depending on the test execution environment, that
1195 # still indicates that it was called.
1196 try:
1197 output = subprocess.check_output(
1198 [sys.executable, "-c",
1199 "import os; print(os.getpgid(os.getpid()))"],
1200 start_new_session=True)
1201 except OSError as e:
1202 if e.errno != errno.EPERM:
1203 raise
1204 else:
1205 parent_pgid = os.getpgid(os.getpid())
1206 child_pgid = int(output)
1207 self.assertNotEqual(parent_pgid, child_pgid)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001208
1209 def test_run_abort(self):
1210 # returncode handles signal termination
1211 with _SuppressCoreFiles():
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001212 p = subprocess.Popen([sys.executable, "-c",
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001213 'import os; os.abort()'])
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001214 p.wait()
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001215 self.assertEqual(-p.returncode, signal.SIGABRT)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001216
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001217 def test_preexec(self):
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001218 # DISCLAIMER: Setting environment variables is *not* a good use
1219 # of a preexec_fn. This is merely a test.
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001220 p = subprocess.Popen([sys.executable, "-c",
1221 'import sys,os;'
1222 'sys.stdout.write(os.getenv("FRUIT"))'],
1223 stdout=subprocess.PIPE,
1224 preexec_fn=lambda: os.putenv("FRUIT", "apple"))
Brian Curtin3c6a9512010-11-05 03:58:52 +00001225 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001226 self.assertEqual(p.stdout.read(), b"apple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001227
Gregory P. Smithfb94c5f2010-03-14 06:49:55 +00001228 def test_preexec_exception(self):
1229 def raise_it():
1230 raise ValueError("What if two swallows carried a coconut?")
1231 try:
1232 p = subprocess.Popen([sys.executable, "-c", ""],
1233 preexec_fn=raise_it)
1234 except RuntimeError as e:
1235 self.assertTrue(
1236 subprocess._posixsubprocess,
1237 "Expected a ValueError from the preexec_fn")
1238 except ValueError as e:
1239 self.assertIn("coconut", e.args[0])
1240 else:
1241 self.fail("Exception raised by preexec_fn did not make it "
1242 "to the parent process.")
1243
Gregory P. Smithe27faac2012-11-11 09:59:27 -08001244 class _TestExecuteChildPopen(subprocess.Popen):
1245 """Used to test behavior at the end of _execute_child."""
1246 def __init__(self, testcase, *args, **kwargs):
1247 self._testcase = testcase
1248 subprocess.Popen.__init__(self, *args, **kwargs)
Gregory P. Smith12489d92012-11-11 01:37:02 -08001249
Gregory P. Smithe27faac2012-11-11 09:59:27 -08001250 def _execute_child(self, *args, **kwargs):
Gregory P. Smith12489d92012-11-11 01:37:02 -08001251 try:
Gregory P. Smithe27faac2012-11-11 09:59:27 -08001252 subprocess.Popen._execute_child(self, *args, **kwargs)
Gregory P. Smith12489d92012-11-11 01:37:02 -08001253 finally:
1254 # Open a bunch of file descriptors and verify that
1255 # none of them are the same as the ones the Popen
1256 # instance is using for stdin/stdout/stderr.
1257 devzero_fds = [os.open("/dev/zero", os.O_RDONLY)
1258 for _ in range(8)]
1259 try:
1260 for fd in devzero_fds:
Gregory P. Smithe27faac2012-11-11 09:59:27 -08001261 self._testcase.assertNotIn(
1262 fd, (self.stdin.fileno(), self.stdout.fileno(),
1263 self.stderr.fileno()),
Gregory P. Smith12489d92012-11-11 01:37:02 -08001264 msg="At least one fd was closed early.")
1265 finally:
Richard Oudkerk0e547b62013-06-10 16:29:19 +01001266 for fd in devzero_fds:
1267 os.close(fd)
Gregory P. Smith12489d92012-11-11 01:37:02 -08001268
Gregory P. Smithe27faac2012-11-11 09:59:27 -08001269 @unittest.skipIf(not os.path.exists("/dev/zero"), "/dev/zero required.")
1270 def test_preexec_errpipe_does_not_double_close_pipes(self):
1271 """Issue16140: Don't double close pipes on preexec error."""
1272
1273 def raise_it():
1274 raise RuntimeError("force the _execute_child() errpipe_data path.")
Gregory P. Smith12489d92012-11-11 01:37:02 -08001275
1276 with self.assertRaises(RuntimeError):
Gregory P. Smithe27faac2012-11-11 09:59:27 -08001277 self._TestExecuteChildPopen(
1278 self, [sys.executable, "-c", "pass"],
Gregory P. Smith12489d92012-11-11 01:37:02 -08001279 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1280 stderr=subprocess.PIPE, preexec_fn=raise_it)
1281
Gregory P. Smith32ec9da2010-03-19 16:53:08 +00001282 def test_preexec_gc_module_failure(self):
1283 # This tests the code that disables garbage collection if the child
1284 # process will execute any Python.
1285 def raise_runtime_error():
1286 raise RuntimeError("this shouldn't escape")
1287 enabled = gc.isenabled()
1288 orig_gc_disable = gc.disable
1289 orig_gc_isenabled = gc.isenabled
1290 try:
1291 gc.disable()
1292 self.assertFalse(gc.isenabled())
1293 subprocess.call([sys.executable, '-c', ''],
1294 preexec_fn=lambda: None)
1295 self.assertFalse(gc.isenabled(),
1296 "Popen enabled gc when it shouldn't.")
1297
1298 gc.enable()
1299 self.assertTrue(gc.isenabled())
1300 subprocess.call([sys.executable, '-c', ''],
1301 preexec_fn=lambda: None)
1302 self.assertTrue(gc.isenabled(), "Popen left gc disabled.")
1303
1304 gc.disable = raise_runtime_error
1305 self.assertRaises(RuntimeError, subprocess.Popen,
1306 [sys.executable, '-c', ''],
1307 preexec_fn=lambda: None)
1308
1309 del gc.isenabled # force an AttributeError
1310 self.assertRaises(AttributeError, subprocess.Popen,
1311 [sys.executable, '-c', ''],
1312 preexec_fn=lambda: None)
1313 finally:
1314 gc.disable = orig_gc_disable
1315 gc.isenabled = orig_gc_isenabled
1316 if not enabled:
1317 gc.disable()
1318
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001319 def test_args_string(self):
1320 # args is a string
1321 fd, fname = mkstemp()
1322 # reopen in text mode
Victor Stinnerf6782ac2010-10-16 23:46:43 +00001323 with open(fd, "w", errors="surrogateescape") as fobj:
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001324 fobj.write("#!/bin/sh\n")
1325 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
1326 sys.executable)
1327 os.chmod(fname, 0o700)
1328 p = subprocess.Popen(fname)
1329 p.wait()
1330 os.remove(fname)
1331 self.assertEqual(p.returncode, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001332
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001333 def test_invalid_args(self):
1334 # invalid arguments should raise ValueError
1335 self.assertRaises(ValueError, subprocess.call,
1336 [sys.executable, "-c",
1337 "import sys; sys.exit(47)"],
1338 startupinfo=47)
1339 self.assertRaises(ValueError, subprocess.call,
1340 [sys.executable, "-c",
1341 "import sys; sys.exit(47)"],
1342 creationflags=47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001343
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001344 def test_shell_sequence(self):
1345 # Run command through the shell (sequence)
1346 newenv = os.environ.copy()
1347 newenv["FRUIT"] = "apple"
1348 p = subprocess.Popen(["echo $FRUIT"], shell=1,
1349 stdout=subprocess.PIPE,
1350 env=newenv)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001351 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001352 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001353
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001354 def test_shell_string(self):
1355 # Run command through the shell (string)
1356 newenv = os.environ.copy()
1357 newenv["FRUIT"] = "apple"
1358 p = subprocess.Popen("echo $FRUIT", shell=1,
1359 stdout=subprocess.PIPE,
1360 env=newenv)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001361 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001362 self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
Christian Heimesa342c012008-04-20 21:01:16 +00001363
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001364 def test_call_string(self):
1365 # call() function with string argument on UNIX
1366 fd, fname = mkstemp()
1367 # reopen in text mode
Victor Stinnerf6782ac2010-10-16 23:46:43 +00001368 with open(fd, "w", errors="surrogateescape") as fobj:
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001369 fobj.write("#!/bin/sh\n")
1370 fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
1371 sys.executable)
1372 os.chmod(fname, 0o700)
1373 rc = subprocess.call(fname)
1374 os.remove(fname)
1375 self.assertEqual(rc, 47)
Christian Heimesa342c012008-04-20 21:01:16 +00001376
Stefan Krah9542cc62010-07-19 14:20:53 +00001377 def test_specific_shell(self):
1378 # Issue #9265: Incorrect name passed as arg[0].
1379 shells = []
1380 for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']:
1381 for name in ['bash', 'ksh']:
1382 sh = os.path.join(prefix, name)
1383 if os.path.isfile(sh):
1384 shells.append(sh)
1385 if not shells: # Will probably work for any shell but csh.
1386 self.skipTest("bash or ksh required for this test")
1387 sh = '/bin/sh'
1388 if os.path.isfile(sh) and not os.path.islink(sh):
1389 # Test will fail if /bin/sh is a symlink to csh.
1390 shells.append(sh)
1391 for sh in shells:
1392 p = subprocess.Popen("echo $0", executable=sh, shell=True,
1393 stdout=subprocess.PIPE)
Brian Curtin3c6a9512010-11-05 03:58:52 +00001394 self.addCleanup(p.stdout.close)
Stefan Krah9542cc62010-07-19 14:20:53 +00001395 self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii'))
1396
Florent Xicluna4886d242010-03-08 13:27:26 +00001397 def _kill_process(self, method, *args):
Florent Xicluna1d8ee3a2010-03-05 20:26:54 +00001398 # Do not inherit file handles from the parent.
1399 # It should fix failures on some platforms.
Antoine Pitrou3d8580f2010-09-20 01:33:21 +00001400 p = subprocess.Popen([sys.executable, "-c", """if 1:
1401 import sys, time
1402 sys.stdout.write('x\\n')
1403 sys.stdout.flush()
1404 time.sleep(30)
1405 """],
1406 close_fds=True,
1407 stdin=subprocess.PIPE,
1408 stdout=subprocess.PIPE,
1409 stderr=subprocess.PIPE)
1410 # Wait for the interpreter to be completely initialized before
1411 # sending any signal.
1412 p.stdout.read(1)
1413 getattr(p, method)(*args)
Florent Xicluna4886d242010-03-08 13:27:26 +00001414 return p
1415
Charles-François Natali53221e32013-01-12 16:52:20 +01001416 @unittest.skipIf(sys.platform.startswith(('netbsd', 'openbsd')),
1417 "Due to known OS bug (issue #16762)")
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01001418 def _kill_dead_process(self, method, *args):
1419 # Do not inherit file handles from the parent.
1420 # It should fix failures on some platforms.
1421 p = subprocess.Popen([sys.executable, "-c", """if 1:
1422 import sys, time
1423 sys.stdout.write('x\\n')
1424 sys.stdout.flush()
1425 """],
1426 close_fds=True,
1427 stdin=subprocess.PIPE,
1428 stdout=subprocess.PIPE,
1429 stderr=subprocess.PIPE)
1430 # Wait for the interpreter to be completely initialized before
1431 # sending any signal.
1432 p.stdout.read(1)
1433 # The process should end after this
1434 time.sleep(1)
1435 # This shouldn't raise even though the child is now dead
1436 getattr(p, method)(*args)
1437 p.communicate()
1438
Florent Xicluna4886d242010-03-08 13:27:26 +00001439 def test_send_signal(self):
1440 p = self._kill_process('send_signal', signal.SIGINT)
Florent Xiclunac049d872010-03-27 22:47:23 +00001441 _, stderr = p.communicate()
1442 self.assertIn(b'KeyboardInterrupt', stderr)
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001443 self.assertNotEqual(p.wait(), 0)
Christian Heimesa342c012008-04-20 21:01:16 +00001444
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001445 def test_kill(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001446 p = self._kill_process('kill')
Florent Xiclunac049d872010-03-27 22:47:23 +00001447 _, stderr = p.communicate()
1448 self.assertStderrEqual(stderr, b'')
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001449 self.assertEqual(p.wait(), -signal.SIGKILL)
Tim Peterse718f612004-10-12 21:51:32 +00001450
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001451 def test_terminate(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00001452 p = self._kill_process('terminate')
Florent Xiclunac049d872010-03-27 22:47:23 +00001453 _, stderr = p.communicate()
1454 self.assertStderrEqual(stderr, b'')
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001455 self.assertEqual(p.wait(), -signal.SIGTERM)
1456
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01001457 def test_send_signal_dead(self):
1458 # Sending a signal to a dead process
1459 self._kill_dead_process('send_signal', signal.SIGINT)
1460
1461 def test_kill_dead(self):
1462 # Killing a dead process
1463 self._kill_dead_process('kill')
1464
1465 def test_terminate_dead(self):
1466 # Terminating a dead process
1467 self._kill_dead_process('terminate')
1468
Antoine Pitrouc9c83ba2011-01-03 18:23:55 +00001469 def check_close_std_fds(self, fds):
1470 # Issue #9905: test that subprocess pipes still work properly with
1471 # some standard fds closed
1472 stdin = 0
1473 newfds = []
1474 for a in fds:
1475 b = os.dup(a)
1476 newfds.append(b)
1477 if a == 0:
1478 stdin = b
1479 try:
1480 for fd in fds:
1481 os.close(fd)
1482 out, err = subprocess.Popen([sys.executable, "-c",
1483 'import sys;'
1484 'sys.stdout.write("apple");'
1485 'sys.stdout.flush();'
1486 'sys.stderr.write("orange")'],
1487 stdin=stdin,
1488 stdout=subprocess.PIPE,
1489 stderr=subprocess.PIPE).communicate()
1490 err = support.strip_python_stderr(err)
1491 self.assertEqual((out, err), (b'apple', b'orange'))
1492 finally:
1493 for b, a in zip(newfds, fds):
1494 os.dup2(b, a)
1495 for b in newfds:
1496 os.close(b)
1497
1498 def test_close_fd_0(self):
1499 self.check_close_std_fds([0])
1500
1501 def test_close_fd_1(self):
1502 self.check_close_std_fds([1])
1503
1504 def test_close_fd_2(self):
1505 self.check_close_std_fds([2])
1506
1507 def test_close_fds_0_1(self):
1508 self.check_close_std_fds([0, 1])
1509
1510 def test_close_fds_0_2(self):
1511 self.check_close_std_fds([0, 2])
1512
1513 def test_close_fds_1_2(self):
1514 self.check_close_std_fds([1, 2])
1515
1516 def test_close_fds_0_1_2(self):
1517 # Issue #10806: test that subprocess pipes still work properly with
1518 # all standard fds closed.
1519 self.check_close_std_fds([0, 1, 2])
1520
Antoine Pitrou95aaeee2011-01-03 21:15:48 +00001521 def test_remapping_std_fds(self):
1522 # open up some temporary files
1523 temps = [mkstemp() for i in range(3)]
1524 try:
1525 temp_fds = [fd for fd, fname in temps]
1526
1527 # unlink the files -- we won't need to reopen them
1528 for fd, fname in temps:
1529 os.unlink(fname)
1530
1531 # write some data to what will become stdin, and rewind
1532 os.write(temp_fds[1], b"STDIN")
1533 os.lseek(temp_fds[1], 0, 0)
1534
1535 # move the standard file descriptors out of the way
1536 saved_fds = [os.dup(fd) for fd in range(3)]
1537 try:
1538 # duplicate the file objects over the standard fd's
1539 for fd, temp_fd in enumerate(temp_fds):
1540 os.dup2(temp_fd, fd)
1541
1542 # now use those files in the "wrong" order, so that subprocess
1543 # has to rearrange them in the child
1544 p = subprocess.Popen([sys.executable, "-c",
1545 'import sys; got = sys.stdin.read();'
1546 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1547 stdin=temp_fds[1],
1548 stdout=temp_fds[2],
1549 stderr=temp_fds[0])
1550 p.wait()
1551 finally:
1552 # restore the original fd's underneath sys.stdin, etc.
1553 for std, saved in enumerate(saved_fds):
1554 os.dup2(saved, std)
1555 os.close(saved)
1556
1557 for fd in temp_fds:
1558 os.lseek(fd, 0, 0)
1559
1560 out = os.read(temp_fds[2], 1024)
1561 err = support.strip_python_stderr(os.read(temp_fds[0], 1024))
1562 self.assertEqual(out, b"got STDIN")
1563 self.assertEqual(err, b"err")
1564
1565 finally:
1566 for fd in temp_fds:
1567 os.close(fd)
1568
Ross Lagerwalld98646e2011-07-27 07:16:31 +02001569 def check_swap_fds(self, stdin_no, stdout_no, stderr_no):
1570 # open up some temporary files
1571 temps = [mkstemp() for i in range(3)]
1572 temp_fds = [fd for fd, fname in temps]
1573 try:
1574 # unlink the files -- we won't need to reopen them
1575 for fd, fname in temps:
1576 os.unlink(fname)
1577
1578 # save a copy of the standard file descriptors
1579 saved_fds = [os.dup(fd) for fd in range(3)]
1580 try:
1581 # duplicate the temp files over the standard fd's 0, 1, 2
1582 for fd, temp_fd in enumerate(temp_fds):
1583 os.dup2(temp_fd, fd)
1584
1585 # write some data to what will become stdin, and rewind
1586 os.write(stdin_no, b"STDIN")
1587 os.lseek(stdin_no, 0, 0)
1588
1589 # now use those files in the given order, so that subprocess
1590 # has to rearrange them in the child
1591 p = subprocess.Popen([sys.executable, "-c",
1592 'import sys; got = sys.stdin.read();'
1593 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1594 stdin=stdin_no,
1595 stdout=stdout_no,
1596 stderr=stderr_no)
1597 p.wait()
1598
1599 for fd in temp_fds:
1600 os.lseek(fd, 0, 0)
1601
1602 out = os.read(stdout_no, 1024)
1603 err = support.strip_python_stderr(os.read(stderr_no, 1024))
1604 finally:
1605 for std, saved in enumerate(saved_fds):
1606 os.dup2(saved, std)
1607 os.close(saved)
1608
1609 self.assertEqual(out, b"got STDIN")
1610 self.assertEqual(err, b"err")
1611
1612 finally:
1613 for fd in temp_fds:
1614 os.close(fd)
1615
1616 # When duping fds, if there arises a situation where one of the fds is
1617 # either 0, 1 or 2, it is possible that it is overwritten (#12607).
1618 # This tests all combinations of this.
1619 def test_swap_fds(self):
1620 self.check_swap_fds(0, 1, 2)
1621 self.check_swap_fds(0, 2, 1)
1622 self.check_swap_fds(1, 0, 2)
1623 self.check_swap_fds(1, 2, 0)
1624 self.check_swap_fds(2, 0, 1)
1625 self.check_swap_fds(2, 1, 0)
1626
Victor Stinner13bb71c2010-04-23 21:41:56 +00001627 def test_surrogates_error_message(self):
Victor Stinner4d078042010-04-23 19:28:32 +00001628 def prepare():
1629 raise ValueError("surrogate:\uDCff")
1630
1631 try:
1632 subprocess.call(
1633 [sys.executable, "-c", "pass"],
1634 preexec_fn=prepare)
1635 except ValueError as err:
1636 # Pure Python implementations keeps the message
1637 self.assertIsNone(subprocess._posixsubprocess)
1638 self.assertEqual(str(err), "surrogate:\uDCff")
1639 except RuntimeError as err:
1640 # _posixsubprocess uses a default message
1641 self.assertIsNotNone(subprocess._posixsubprocess)
1642 self.assertEqual(str(err), "Exception occurred in preexec_fn.")
1643 else:
1644 self.fail("Expected ValueError or RuntimeError")
1645
Victor Stinner13bb71c2010-04-23 21:41:56 +00001646 def test_undecodable_env(self):
1647 for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')):
Victor Stinner13bb71c2010-04-23 21:41:56 +00001648 # test str with surrogates
Antoine Pitroufb8db8f2010-09-19 22:46:05 +00001649 script = "import os; print(ascii(os.getenv(%s)))" % repr(key)
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001650 env = os.environ.copy()
1651 env[key] = value
Victor Stinner89f3ad12010-10-14 10:43:31 +00001652 # Use C locale to get ascii for the locale encoding to force
1653 # surrogate-escaping of \xFF in the child process; otherwise it can
1654 # be decoded as-is if the default locale is latin-1.
Victor Stinnerebc78d22010-10-14 10:38:17 +00001655 env['LC_ALL'] = 'C'
Victor Stinner13bb71c2010-04-23 21:41:56 +00001656 stdout = subprocess.check_output(
1657 [sys.executable, "-c", script],
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001658 env=env)
Victor Stinner13bb71c2010-04-23 21:41:56 +00001659 stdout = stdout.rstrip(b'\n\r')
Ezio Melottib3aedd42010-11-20 19:04:17 +00001660 self.assertEqual(stdout.decode('ascii'), ascii(value))
Victor Stinner13bb71c2010-04-23 21:41:56 +00001661
1662 # test bytes
1663 key = key.encode("ascii", "surrogateescape")
1664 value = value.encode("ascii", "surrogateescape")
Antoine Pitroufb8db8f2010-09-19 22:46:05 +00001665 script = "import os; print(ascii(os.getenvb(%s)))" % repr(key)
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001666 env = os.environ.copy()
1667 env[key] = value
Victor Stinner13bb71c2010-04-23 21:41:56 +00001668 stdout = subprocess.check_output(
1669 [sys.executable, "-c", script],
Victor Stinnerce2d24d2010-04-23 22:55:39 +00001670 env=env)
Victor Stinner13bb71c2010-04-23 21:41:56 +00001671 stdout = stdout.rstrip(b'\n\r')
Ezio Melottib3aedd42010-11-20 19:04:17 +00001672 self.assertEqual(stdout.decode('ascii'), ascii(value))
Victor Stinner13bb71c2010-04-23 21:41:56 +00001673
Victor Stinnerb745a742010-05-18 17:17:23 +00001674 def test_bytes_program(self):
1675 abs_program = os.fsencode(sys.executable)
1676 path, program = os.path.split(sys.executable)
1677 program = os.fsencode(program)
1678
1679 # absolute bytes path
1680 exitcode = subprocess.call([abs_program, "-c", "pass"])
Ezio Melottib3aedd42010-11-20 19:04:17 +00001681 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001682
Victor Stinner7b3b20a2011-03-03 12:54:05 +00001683 # absolute bytes path as a string
1684 cmd = b"'" + abs_program + b"' -c pass"
1685 exitcode = subprocess.call(cmd, shell=True)
1686 self.assertEqual(exitcode, 0)
1687
Victor Stinnerb745a742010-05-18 17:17:23 +00001688 # bytes program, unicode PATH
1689 env = os.environ.copy()
1690 env["PATH"] = path
1691 exitcode = subprocess.call([program, "-c", "pass"], env=env)
Ezio Melottib3aedd42010-11-20 19:04:17 +00001692 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001693
1694 # bytes program, bytes PATH
1695 envb = os.environb.copy()
1696 envb[b"PATH"] = os.fsencode(path)
1697 exitcode = subprocess.call([program, "-c", "pass"], env=envb)
Ezio Melottib3aedd42010-11-20 19:04:17 +00001698 self.assertEqual(exitcode, 0)
Victor Stinnerb745a742010-05-18 17:17:23 +00001699
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001700 def test_pipe_cloexec(self):
1701 sleeper = support.findfile("input_reader.py", subdir="subprocessdata")
1702 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1703
1704 p1 = subprocess.Popen([sys.executable, sleeper],
1705 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1706 stderr=subprocess.PIPE, close_fds=False)
1707
1708 self.addCleanup(p1.communicate, b'')
1709
1710 p2 = subprocess.Popen([sys.executable, fd_status],
1711 stdout=subprocess.PIPE, close_fds=False)
1712
1713 output, error = p2.communicate()
1714 result_fds = set(map(int, output.split(b',')))
1715 unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(),
1716 p1.stderr.fileno()])
1717
1718 self.assertFalse(result_fds & unwanted_fds,
1719 "Expected no fds from %r to be open in child, "
1720 "found %r" %
1721 (unwanted_fds, result_fds & unwanted_fds))
1722
1723 def test_pipe_cloexec_real_tools(self):
1724 qcat = support.findfile("qcat.py", subdir="subprocessdata")
1725 qgrep = support.findfile("qgrep.py", subdir="subprocessdata")
1726
1727 subdata = b'zxcvbn'
1728 data = subdata * 4 + b'\n'
1729
1730 p1 = subprocess.Popen([sys.executable, qcat],
1731 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1732 close_fds=False)
1733
1734 p2 = subprocess.Popen([sys.executable, qgrep, subdata],
1735 stdin=p1.stdout, stdout=subprocess.PIPE,
1736 close_fds=False)
1737
1738 self.addCleanup(p1.wait)
1739 self.addCleanup(p2.wait)
Gregory P. Smith886455c2012-01-21 22:05:10 -08001740 def kill_p1():
1741 try:
1742 p1.terminate()
1743 except ProcessLookupError:
1744 pass
1745 def kill_p2():
1746 try:
1747 p2.terminate()
1748 except ProcessLookupError:
1749 pass
1750 self.addCleanup(kill_p1)
1751 self.addCleanup(kill_p2)
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001752
1753 p1.stdin.write(data)
1754 p1.stdin.close()
1755
1756 readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10)
1757
1758 self.assertTrue(readfiles, "The child hung")
1759 self.assertEqual(p2.stdout.read(), data)
1760
Victor Stinnerfaa8c132011-01-03 16:36:00 +00001761 p1.stdout.close()
1762 p2.stdout.close()
1763
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001764 def test_close_fds(self):
1765 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1766
1767 fds = os.pipe()
1768 self.addCleanup(os.close, fds[0])
1769 self.addCleanup(os.close, fds[1])
1770
1771 open_fds = set(fds)
Gregory P. Smith8facece2012-01-21 14:01:08 -08001772 # add a bunch more fds
1773 for _ in range(9):
1774 fd = os.open("/dev/null", os.O_RDONLY)
1775 self.addCleanup(os.close, fd)
1776 open_fds.add(fd)
Gregory P. Smith51ee2702010-12-13 07:59:39 +00001777
1778 p = subprocess.Popen([sys.executable, fd_status],
1779 stdout=subprocess.PIPE, close_fds=False)
1780 output, ignored = p.communicate()
1781 remaining_fds = set(map(int, output.split(b',')))
1782
1783 self.assertEqual(remaining_fds & open_fds, open_fds,
1784 "Some fds were closed")
1785
1786 p = subprocess.Popen([sys.executable, fd_status],
1787 stdout=subprocess.PIPE, close_fds=True)
1788 output, ignored = p.communicate()
1789 remaining_fds = set(map(int, output.split(b',')))
1790
1791 self.assertFalse(remaining_fds & open_fds,
1792 "Some fds were left open")
1793 self.assertIn(1, remaining_fds, "Subprocess failed")
1794
Gregory P. Smith8facece2012-01-21 14:01:08 -08001795 # Keep some of the fd's we opened open in the subprocess.
1796 # This tests _posixsubprocess.c's proper handling of fds_to_keep.
1797 fds_to_keep = set(open_fds.pop() for _ in range(8))
1798 p = subprocess.Popen([sys.executable, fd_status],
1799 stdout=subprocess.PIPE, close_fds=True,
1800 pass_fds=())
1801 output, ignored = p.communicate()
1802 remaining_fds = set(map(int, output.split(b',')))
1803
1804 self.assertFalse(remaining_fds & fds_to_keep & open_fds,
1805 "Some fds not in pass_fds were left open")
1806 self.assertIn(1, remaining_fds, "Subprocess failed")
1807
Victor Stinner88701e22011-06-01 13:13:04 +02001808 # Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file
1809 # descriptor of a pipe closed in the parent process is valid in the
1810 # child process according to fstat(), but the mode of the file
1811 # descriptor is invalid, and read or write raise an error.
1812 @support.requires_mac_ver(10, 5)
Gregory P. Smith8edd99d2010-12-14 13:43:30 +00001813 def test_pass_fds(self):
1814 fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
1815
1816 open_fds = set()
1817
1818 for x in range(5):
1819 fds = os.pipe()
1820 self.addCleanup(os.close, fds[0])
1821 self.addCleanup(os.close, fds[1])
1822 open_fds.update(fds)
1823
1824 for fd in open_fds:
1825 p = subprocess.Popen([sys.executable, fd_status],
1826 stdout=subprocess.PIPE, close_fds=True,
1827 pass_fds=(fd, ))
1828 output, ignored = p.communicate()
1829
1830 remaining_fds = set(map(int, output.split(b',')))
1831 to_be_closed = open_fds - {fd}
1832
1833 self.assertIn(fd, remaining_fds, "fd to be passed not passed")
1834 self.assertFalse(remaining_fds & to_be_closed,
1835 "fd to be closed passed")
1836
1837 # pass_fds overrides close_fds with a warning.
1838 with self.assertWarns(RuntimeWarning) as context:
1839 self.assertFalse(subprocess.call(
1840 [sys.executable, "-c", "import sys; sys.exit(0)"],
1841 close_fds=False, pass_fds=(fd, )))
1842 self.assertIn('overriding close_fds', str(context.warning))
1843
Gregory P. Smith112bb3a2011-03-15 14:55:17 -04001844 def test_stdout_stdin_are_single_inout_fd(self):
1845 with io.open(os.devnull, "r+") as inout:
1846 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1847 stdout=inout, stdin=inout)
1848 p.wait()
1849
1850 def test_stdout_stderr_are_single_inout_fd(self):
1851 with io.open(os.devnull, "r+") as inout:
1852 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1853 stdout=inout, stderr=inout)
1854 p.wait()
1855
1856 def test_stderr_stdin_are_single_inout_fd(self):
1857 with io.open(os.devnull, "r+") as inout:
1858 p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
1859 stderr=inout, stdin=inout)
1860 p.wait()
1861
Gregory P. Smithe85db2b2010-12-14 14:38:00 +00001862 def test_wait_when_sigchild_ignored(self):
1863 # NOTE: sigchild_ignore.py may not be an effective test on all OSes.
1864 sigchild_ignore = support.findfile("sigchild_ignore.py",
1865 subdir="subprocessdata")
1866 p = subprocess.Popen([sys.executable, sigchild_ignore],
1867 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1868 stdout, stderr = p.communicate()
1869 self.assertEqual(0, p.returncode, "sigchild_ignore.py exited"
Gregory P. Smitha80f4fb2010-12-14 15:23:02 +00001870 " non-zero with this error:\n%s" %
Marc-André Lemburg8f36af72011-02-25 15:42:01 +00001871 stderr.decode('utf-8'))
Gregory P. Smithe85db2b2010-12-14 14:38:00 +00001872
Antoine Pitrou7b98d022011-03-19 17:04:13 +01001873 def test_select_unbuffered(self):
1874 # Issue #11459: bufsize=0 should really set the pipes as
1875 # unbuffered (and therefore let select() work properly).
1876 select = support.import_module("select")
1877 p = subprocess.Popen([sys.executable, "-c",
1878 'import sys;'
1879 'sys.stdout.write("apple")'],
1880 stdout=subprocess.PIPE,
1881 bufsize=0)
1882 f = p.stdout
Ross Lagerwall17ace7a2011-03-26 21:21:46 +02001883 self.addCleanup(f.close)
Antoine Pitrou7b98d022011-03-19 17:04:13 +01001884 try:
1885 self.assertEqual(f.read(4), b"appl")
1886 self.assertIn(f, select.select([f], [], [], 0.0)[0])
1887 finally:
1888 p.wait()
1889
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001890 def test_zombie_fast_process_del(self):
1891 # Issue #12650: on Unix, if Popen.__del__() was called before the
1892 # process exited, it wouldn't be added to subprocess._active, and would
1893 # remain a zombie.
1894 # spawn a Popen, and delete its reference before it exits
1895 p = subprocess.Popen([sys.executable, "-c",
1896 'import sys, time;'
1897 'time.sleep(0.2)'],
1898 stdout=subprocess.PIPE,
1899 stderr=subprocess.PIPE)
Nadeem Vawda0d7cda32011-08-19 05:12:01 +02001900 self.addCleanup(p.stdout.close)
1901 self.addCleanup(p.stderr.close)
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001902 ident = id(p)
1903 pid = p.pid
1904 del p
1905 # check that p is in the active processes list
1906 self.assertIn(ident, [id(o) for o in subprocess._active])
1907
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001908 def test_leak_fast_process_del_killed(self):
1909 # Issue #12650: on Unix, if Popen.__del__() was called before the
1910 # process exited, and the process got killed by a signal, it would never
1911 # be removed from subprocess._active, which triggered a FD and memory
1912 # leak.
1913 # spawn a Popen, delete its reference and kill it
1914 p = subprocess.Popen([sys.executable, "-c",
1915 'import time;'
1916 'time.sleep(3)'],
1917 stdout=subprocess.PIPE,
1918 stderr=subprocess.PIPE)
Nadeem Vawda0d7cda32011-08-19 05:12:01 +02001919 self.addCleanup(p.stdout.close)
1920 self.addCleanup(p.stderr.close)
Charles-François Natali134a8ba2011-08-18 18:49:39 +02001921 ident = id(p)
1922 pid = p.pid
1923 del p
1924 os.kill(pid, signal.SIGKILL)
1925 # check that p is in the active processes list
1926 self.assertIn(ident, [id(o) for o in subprocess._active])
1927
1928 # let some time for the process to exit, and create a new Popen: this
1929 # should trigger the wait() of p
1930 time.sleep(0.2)
1931 with self.assertRaises(EnvironmentError) as c:
1932 with subprocess.Popen(['nonexisting_i_hope'],
1933 stdout=subprocess.PIPE,
1934 stderr=subprocess.PIPE) as proc:
1935 pass
1936 # p should have been wait()ed on, and removed from the _active list
1937 self.assertRaises(OSError, os.waitpid, pid, 0)
1938 self.assertNotIn(ident, [id(o) for o in subprocess._active])
1939
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001940
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001941@unittest.skipUnless(mswindows, "Windows specific tests")
Florent Xiclunac049d872010-03-27 22:47:23 +00001942class Win32ProcessTestCase(BaseTestCase):
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00001943
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001944 def test_startupinfo(self):
1945 # startupinfo argument
1946 # We uses hardcoded constants, because we do not want to
1947 # depend on win32all.
1948 STARTF_USESHOWWINDOW = 1
1949 SW_MAXIMIZE = 3
1950 startupinfo = subprocess.STARTUPINFO()
1951 startupinfo.dwFlags = STARTF_USESHOWWINDOW
1952 startupinfo.wShowWindow = SW_MAXIMIZE
1953 # Since Python is a console process, it won't be affected
1954 # by wShowWindow, but the argument should be silently
1955 # ignored
1956 subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001957 startupinfo=startupinfo)
1958
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001959 def test_creationflags(self):
1960 # creationflags argument
1961 CREATE_NEW_CONSOLE = 16
1962 sys.stderr.write(" a DOS box should flash briefly ...\n")
1963 subprocess.call(sys.executable +
1964 ' -c "import time; time.sleep(0.25)"',
1965 creationflags=CREATE_NEW_CONSOLE)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001966
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001967 def test_invalid_args(self):
1968 # invalid arguments should raise ValueError
1969 self.assertRaises(ValueError, subprocess.call,
1970 [sys.executable, "-c",
1971 "import sys; sys.exit(47)"],
1972 preexec_fn=lambda: 1)
1973 self.assertRaises(ValueError, subprocess.call,
1974 [sys.executable, "-c",
1975 "import sys; sys.exit(47)"],
1976 stdout=subprocess.PIPE,
1977 close_fds=True)
1978
1979 def test_close_fds(self):
1980 # close file descriptors
1981 rc = subprocess.call([sys.executable, "-c",
1982 "import sys; sys.exit(47)"],
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001983 close_fds=True)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001984 self.assertEqual(rc, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00001985
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001986 def test_shell_sequence(self):
1987 # Run command through the shell (sequence)
1988 newenv = os.environ.copy()
1989 newenv["FRUIT"] = "physalis"
1990 p = subprocess.Popen(["set"], shell=1,
1991 stdout=subprocess.PIPE,
1992 env=newenv)
Brian Curtin19a53792010-11-05 17:09:05 +00001993 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001994 self.assertIn(b"physalis", p.stdout.read())
Guido van Rossume7ba4952007-06-06 23:52:48 +00001995
Florent Xiclunab1e94e82010-02-27 22:12:37 +00001996 def test_shell_string(self):
1997 # Run command through the shell (string)
1998 newenv = os.environ.copy()
1999 newenv["FRUIT"] = "physalis"
2000 p = subprocess.Popen("set", shell=1,
2001 stdout=subprocess.PIPE,
2002 env=newenv)
Brian Curtin19a53792010-11-05 17:09:05 +00002003 self.addCleanup(p.stdout.close)
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002004 self.assertIn(b"physalis", p.stdout.read())
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00002005
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002006 def test_call_string(self):
2007 # call() function with string argument on Windows
2008 rc = subprocess.call(sys.executable +
2009 ' -c "import sys; sys.exit(47)"')
2010 self.assertEqual(rc, 47)
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00002011
Florent Xicluna4886d242010-03-08 13:27:26 +00002012 def _kill_process(self, method, *args):
2013 # Some win32 buildbot raises EOFError if stdin is inherited
Antoine Pitroua4024e22010-09-24 18:57:01 +00002014 p = subprocess.Popen([sys.executable, "-c", """if 1:
2015 import sys, time
2016 sys.stdout.write('x\\n')
2017 sys.stdout.flush()
2018 time.sleep(30)
2019 """],
2020 stdin=subprocess.PIPE,
2021 stdout=subprocess.PIPE,
2022 stderr=subprocess.PIPE)
Brian Curtin19a53792010-11-05 17:09:05 +00002023 self.addCleanup(p.stdout.close)
2024 self.addCleanup(p.stderr.close)
2025 self.addCleanup(p.stdin.close)
Antoine Pitroua4024e22010-09-24 18:57:01 +00002026 # Wait for the interpreter to be completely initialized before
2027 # sending any signal.
2028 p.stdout.read(1)
2029 getattr(p, method)(*args)
Florent Xiclunac049d872010-03-27 22:47:23 +00002030 _, stderr = p.communicate()
2031 self.assertStderrEqual(stderr, b'')
Antoine Pitroua4024e22010-09-24 18:57:01 +00002032 returncode = p.wait()
Florent Xicluna4886d242010-03-08 13:27:26 +00002033 self.assertNotEqual(returncode, 0)
2034
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01002035 def _kill_dead_process(self, method, *args):
2036 p = subprocess.Popen([sys.executable, "-c", """if 1:
2037 import sys, time
2038 sys.stdout.write('x\\n')
2039 sys.stdout.flush()
2040 sys.exit(42)
2041 """],
2042 stdin=subprocess.PIPE,
2043 stdout=subprocess.PIPE,
2044 stderr=subprocess.PIPE)
2045 self.addCleanup(p.stdout.close)
2046 self.addCleanup(p.stderr.close)
2047 self.addCleanup(p.stdin.close)
2048 # Wait for the interpreter to be completely initialized before
2049 # sending any signal.
2050 p.stdout.read(1)
2051 # The process should end after this
2052 time.sleep(1)
2053 # This shouldn't raise even though the child is now dead
2054 getattr(p, method)(*args)
2055 _, stderr = p.communicate()
2056 self.assertStderrEqual(stderr, b'')
2057 rc = p.wait()
2058 self.assertEqual(rc, 42)
2059
Florent Xicluna4886d242010-03-08 13:27:26 +00002060 def test_send_signal(self):
2061 self._kill_process('send_signal', signal.SIGTERM)
Christian Heimesa342c012008-04-20 21:01:16 +00002062
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002063 def test_kill(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00002064 self._kill_process('kill')
Christian Heimesa342c012008-04-20 21:01:16 +00002065
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002066 def test_terminate(self):
Florent Xicluna4886d242010-03-08 13:27:26 +00002067 self._kill_process('terminate')
Christian Heimesa342c012008-04-20 21:01:16 +00002068
Antoine Pitrou1f9a8352012-03-11 19:29:12 +01002069 def test_send_signal_dead(self):
2070 self._kill_dead_process('send_signal', signal.SIGTERM)
2071
2072 def test_kill_dead(self):
2073 self._kill_dead_process('kill')
2074
2075 def test_terminate_dead(self):
2076 self._kill_dead_process('terminate')
2077
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00002078
Brett Cannona23810f2008-05-26 19:04:21 +00002079# The module says:
2080# "NB This only works (and is only relevant) for UNIX."
2081#
2082# Actually, getoutput should work on any platform with an os.popen, but
2083# I'll take the comment as given, and skip this suite.
Florent Xiclunaf0cbd822010-03-04 21:50:56 +00002084@unittest.skipUnless(os.name == 'posix', "only relevant for UNIX")
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002085class CommandTests(unittest.TestCase):
2086 def test_getoutput(self):
2087 self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy')
2088 self.assertEqual(subprocess.getstatusoutput('echo xyzzy'),
2089 (0, 'xyzzy'))
Brett Cannona23810f2008-05-26 19:04:21 +00002090
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002091 # we use mkdtemp in the next line to create an empty directory
2092 # under our exclusive control; from that, we can invent a pathname
2093 # that we _know_ won't exist. This is guaranteed to fail.
2094 dir = None
2095 try:
2096 dir = tempfile.mkdtemp()
2097 name = os.path.join(dir, "foo")
Brett Cannona23810f2008-05-26 19:04:21 +00002098
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002099 status, output = subprocess.getstatusoutput('cat ' + name)
2100 self.assertNotEqual(status, 0)
2101 finally:
2102 if dir is not None:
2103 os.rmdir(dir)
Brett Cannona23810f2008-05-26 19:04:21 +00002104
Gregory P. Smithd06fa472009-07-04 02:46:54 +00002105
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002106@unittest.skipUnless(getattr(subprocess, '_has_poll', False),
2107 "poll system call not supported")
2108class ProcessTestCaseNoPoll(ProcessTestCase):
2109 def setUp(self):
2110 subprocess._has_poll = False
2111 ProcessTestCase.setUp(self)
Gregory P. Smithd06fa472009-07-04 02:46:54 +00002112
Florent Xiclunab1e94e82010-02-27 22:12:37 +00002113 def tearDown(self):
2114 subprocess._has_poll = True
2115 ProcessTestCase.tearDown(self)
Gregory P. Smithd06fa472009-07-04 02:46:54 +00002116
2117
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00002118class HelperFunctionTests(unittest.TestCase):
Gregory P. Smithaf6d3b82010-03-01 02:56:44 +00002119 @unittest.skipIf(mswindows, "errno and EINTR make no sense on windows")
Gregory P. Smitha59c59f2010-03-01 00:17:40 +00002120 def test_eintr_retry_call(self):
2121 record_calls = []
2122 def fake_os_func(*args):
2123 record_calls.append(args)
2124 if len(record_calls) == 2:
2125 raise OSError(errno.EINTR, "fake interrupted system call")
2126 return tuple(reversed(args))
2127
2128 self.assertEqual((999, 256),
2129 subprocess._eintr_retry_call(fake_os_func, 256, 999))
2130 self.assertEqual([(256, 999)], record_calls)
2131 # This time there will be an EINTR so it will loop once.
2132 self.assertEqual((666,),
2133 subprocess._eintr_retry_call(fake_os_func, 666))
2134 self.assertEqual([(256, 999), (666,), (666,)], record_calls)
2135
2136
Tim Golden126c2962010-08-11 14:20:40 +00002137@unittest.skipUnless(mswindows, "Windows-specific tests")
2138class CommandsWithSpaces (BaseTestCase):
2139
2140 def setUp(self):
2141 super().setUp()
2142 f, fname = mkstemp(".py", "te st")
2143 self.fname = fname.lower ()
2144 os.write(f, b"import sys;"
2145 b"sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))"
2146 )
2147 os.close(f)
2148
2149 def tearDown(self):
2150 os.remove(self.fname)
2151 super().tearDown()
2152
2153 def with_spaces(self, *args, **kwargs):
2154 kwargs['stdout'] = subprocess.PIPE
2155 p = subprocess.Popen(*args, **kwargs)
Brian Curtin19a53792010-11-05 17:09:05 +00002156 self.addCleanup(p.stdout.close)
Tim Golden126c2962010-08-11 14:20:40 +00002157 self.assertEqual(
2158 p.stdout.read ().decode("mbcs"),
2159 "2 [%r, 'ab cd']" % self.fname
2160 )
2161
2162 def test_shell_string_with_spaces(self):
2163 # call() function with string argument with spaces on Windows
Brian Curtind835cf12010-08-13 20:42:57 +00002164 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
2165 "ab cd"), shell=1)
Tim Golden126c2962010-08-11 14:20:40 +00002166
2167 def test_shell_sequence_with_spaces(self):
2168 # call() function with sequence argument with spaces on Windows
Brian Curtind835cf12010-08-13 20:42:57 +00002169 self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1)
Tim Golden126c2962010-08-11 14:20:40 +00002170
2171 def test_noshell_string_with_spaces(self):
2172 # call() function with string argument with spaces on Windows
2173 self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
2174 "ab cd"))
2175
2176 def test_noshell_sequence_with_spaces(self):
2177 # call() function with sequence argument with spaces on Windows
2178 self.with_spaces([sys.executable, self.fname, "ab cd"])
2179
Brian Curtin79cdb662010-12-03 02:46:02 +00002180
Georg Brandla86b2622012-02-20 21:34:57 +01002181class ContextManagerTests(BaseTestCase):
Brian Curtin79cdb662010-12-03 02:46:02 +00002182
2183 def test_pipe(self):
2184 with subprocess.Popen([sys.executable, "-c",
2185 "import sys;"
2186 "sys.stdout.write('stdout');"
2187 "sys.stderr.write('stderr');"],
2188 stdout=subprocess.PIPE,
2189 stderr=subprocess.PIPE) as proc:
2190 self.assertEqual(proc.stdout.read(), b"stdout")
2191 self.assertStderrEqual(proc.stderr.read(), b"stderr")
2192
2193 self.assertTrue(proc.stdout.closed)
2194 self.assertTrue(proc.stderr.closed)
2195
2196 def test_returncode(self):
2197 with subprocess.Popen([sys.executable, "-c",
2198 "import sys; sys.exit(100)"]) as proc:
Gregory P. Smith6b657452011-05-11 21:42:08 -07002199 pass
2200 # __exit__ calls wait(), so the returncode should be set
Brian Curtin79cdb662010-12-03 02:46:02 +00002201 self.assertEqual(proc.returncode, 100)
2202
2203 def test_communicate_stdin(self):
2204 with subprocess.Popen([sys.executable, "-c",
2205 "import sys;"
2206 "sys.exit(sys.stdin.read() == 'context')"],
2207 stdin=subprocess.PIPE) as proc:
2208 proc.communicate(b"context")
2209 self.assertEqual(proc.returncode, 1)
2210
2211 def test_invalid_args(self):
Andrew Svetlovb1726972012-12-26 23:34:54 +02002212 with self.assertRaises(FileNotFoundError) as c:
Brian Curtin79cdb662010-12-03 02:46:02 +00002213 with subprocess.Popen(['nonexisting_i_hope'],
2214 stdout=subprocess.PIPE,
2215 stderr=subprocess.PIPE) as proc:
2216 pass
2217
Brian Curtin79cdb662010-12-03 02:46:02 +00002218
Gregory P. Smith3b4652e2011-03-15 15:43:39 -04002219def test_main():
2220 unit_tests = (ProcessTestCase,
2221 POSIXProcessTestCase,
2222 Win32ProcessTestCase,
Gregory P. Smith3b4652e2011-03-15 15:43:39 -04002223 CommandTests,
2224 ProcessTestCaseNoPoll,
2225 HelperFunctionTests,
2226 CommandsWithSpaces,
Antoine Pitrouab85ff32011-07-23 22:03:45 +02002227 ContextManagerTests,
2228 )
Gregory P. Smith3b4652e2011-03-15 15:43:39 -04002229
2230 support.run_unittest(*unit_tests)
2231 support.reap_children()
2232
Fredrik Lundh5b3687d2004-10-12 15:26:28 +00002233if __name__ == "__main__":
Gregory P. Smith112bb3a2011-03-15 14:55:17 -04002234 unittest.main()