Python code coverage for Lib/test/test_subprocess.py

#	count	content
1	n/a	import unittest
2	n/a	from unittest import mock
3	n/a	from test import support
4	n/a	import subprocess
5	n/a	import sys
6	n/a	import platform
7	n/a	import signal
8	n/a	import io
9	n/a	import os
10	n/a	import errno
11	n/a	import tempfile
12	n/a	import time
13	n/a	import selectors
14	n/a	import sysconfig
15	n/a	import select
16	n/a	import shutil
17	n/a	import gc
18	n/a	import textwrap
19	n/a
20	n/a	try:
21	n/a	import ctypes
22	n/a	except ImportError:
23	n/a	ctypes = None
24	n/a
25	n/a	try:
26	n/a	import threading
27	n/a	except ImportError:
28	n/a	threading = None
29	n/a
30	n/a	if support.PGO:
31	n/a	raise unittest.SkipTest("test is not helpful for PGO")
32	n/a
33	n/a	mswindows = (sys.platform == "win32")
34	n/a
35	n/a	#
36	n/a	# Depends on the following external programs: Python
37	n/a	#
38	n/a
39	n/a	if mswindows:
40	n/a	SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), '
41	n/a	'os.O_BINARY);')
42	n/a	else:
43	n/a	SETBINARY = ''
44	n/a
45	n/a
46	n/a	class BaseTestCase(unittest.TestCase):
47	n/a	def setUp(self):
48	n/a	# Try to minimize the number of children we have so this test
49	n/a	# doesn't crash on some buildbots (Alphas in particular).
50	n/a	support.reap_children()
51	n/a
52	n/a	def tearDown(self):
53	n/a	for inst in subprocess._active:
54	n/a	inst.wait()
55	n/a	subprocess._cleanup()
56	n/a	self.assertFalse(subprocess._active, "subprocess._active not empty")
57	n/a
58	n/a	def assertStderrEqual(self, stderr, expected, msg=None):
59	n/a	# In a debug build, stuff like "[6580 refs]" is printed to stderr at
60	n/a	# shutdown time. That frustrates tests trying to check stderr produced
61	n/a	# from a spawned Python process.
62	n/a	actual = support.strip_python_stderr(stderr)
63	n/a	# strip_python_stderr also strips whitespace, so we do too.
64	n/a	expected = expected.strip()
65	n/a	self.assertEqual(actual, expected, msg)
66	n/a
67	n/a
68	n/a	class PopenTestException(Exception):
69	n/a	pass
70	n/a
71	n/a
72	n/a	class PopenExecuteChildRaises(subprocess.Popen):
73	n/a	"""Popen subclass for testing cleanup of subprocess.PIPE filehandles when
74	n/a	_execute_child fails.
75	n/a	"""
76	n/a	def _execute_child(self, args, *kwargs):
77	n/a	raise PopenTestException("Forced Exception for Test")
78	n/a
79	n/a
80	n/a	class ProcessTestCase(BaseTestCase):
81	n/a
82	n/a	def test_io_buffered_by_default(self):
83	n/a	p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
84	n/a	stdin=subprocess.PIPE, stdout=subprocess.PIPE,
85	n/a	stderr=subprocess.PIPE)
86	n/a	try:
87	n/a	self.assertIsInstance(p.stdin, io.BufferedIOBase)
88	n/a	self.assertIsInstance(p.stdout, io.BufferedIOBase)
89	n/a	self.assertIsInstance(p.stderr, io.BufferedIOBase)
90	n/a	finally:
91	n/a	p.stdin.close()
92	n/a	p.stdout.close()
93	n/a	p.stderr.close()
94	n/a	p.wait()
95	n/a
96	n/a	def test_io_unbuffered_works(self):
97	n/a	p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
98	n/a	stdin=subprocess.PIPE, stdout=subprocess.PIPE,
99	n/a	stderr=subprocess.PIPE, bufsize=0)
100	n/a	try:
101	n/a	self.assertIsInstance(p.stdin, io.RawIOBase)
102	n/a	self.assertIsInstance(p.stdout, io.RawIOBase)
103	n/a	self.assertIsInstance(p.stderr, io.RawIOBase)
104	n/a	finally:
105	n/a	p.stdin.close()
106	n/a	p.stdout.close()
107	n/a	p.stderr.close()
108	n/a	p.wait()
109	n/a
110	n/a	def test_call_seq(self):
111	n/a	# call() function with sequence argument
112	n/a	rc = subprocess.call([sys.executable, "-c",
113	n/a	"import sys; sys.exit(47)"])
114	n/a	self.assertEqual(rc, 47)
115	n/a
116	n/a	def test_call_timeout(self):
117	n/a	# call() function with timeout argument; we want to test that the child
118	n/a	# process gets killed when the timeout expires. If the child isn't
119	n/a	# killed, this call will deadlock since subprocess.call waits for the
120	n/a	# child.
121	n/a	self.assertRaises(subprocess.TimeoutExpired, subprocess.call,
122	n/a	[sys.executable, "-c", "while True: pass"],
123	n/a	timeout=0.1)
124	n/a
125	n/a	def test_check_call_zero(self):
126	n/a	# check_call() function with zero return code
127	n/a	rc = subprocess.check_call([sys.executable, "-c",
128	n/a	"import sys; sys.exit(0)"])
129	n/a	self.assertEqual(rc, 0)
130	n/a
131	n/a	def test_check_call_nonzero(self):
132	n/a	# check_call() function with non-zero return code
133	n/a	with self.assertRaises(subprocess.CalledProcessError) as c:
134	n/a	subprocess.check_call([sys.executable, "-c",
135	n/a	"import sys; sys.exit(47)"])
136	n/a	self.assertEqual(c.exception.returncode, 47)
137	n/a
138	n/a	def test_check_output(self):
139	n/a	# check_output() function with zero return code
140	n/a	output = subprocess.check_output(
141	n/a	[sys.executable, "-c", "print('BDFL')"])
142	n/a	self.assertIn(b'BDFL', output)
143	n/a
144	n/a	def test_check_output_nonzero(self):
145	n/a	# check_call() function with non-zero return code
146	n/a	with self.assertRaises(subprocess.CalledProcessError) as c:
147	n/a	subprocess.check_output(
148	n/a	[sys.executable, "-c", "import sys; sys.exit(5)"])
149	n/a	self.assertEqual(c.exception.returncode, 5)
150	n/a
151	n/a	def test_check_output_stderr(self):
152	n/a	# check_output() function stderr redirected to stdout
153	n/a	output = subprocess.check_output(
154	n/a	[sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"],
155	n/a	stderr=subprocess.STDOUT)
156	n/a	self.assertIn(b'BDFL', output)
157	n/a
158	n/a	def test_check_output_stdin_arg(self):
159	n/a	# check_output() can be called with stdin set to a file
160	n/a	tf = tempfile.TemporaryFile()
161	n/a	self.addCleanup(tf.close)
162	n/a	tf.write(b'pear')
163	n/a	tf.seek(0)
164	n/a	output = subprocess.check_output(
165	n/a	[sys.executable, "-c",
166	n/a	"import sys; sys.stdout.write(sys.stdin.read().upper())"],
167	n/a	stdin=tf)
168	n/a	self.assertIn(b'PEAR', output)
169	n/a
170	n/a	def test_check_output_input_arg(self):
171	n/a	# check_output() can be called with input set to a string
172	n/a	output = subprocess.check_output(
173	n/a	[sys.executable, "-c",
174	n/a	"import sys; sys.stdout.write(sys.stdin.read().upper())"],
175	n/a	input=b'pear')
176	n/a	self.assertIn(b'PEAR', output)
177	n/a
178	n/a	def test_check_output_stdout_arg(self):
179	n/a	# check_output() refuses to accept 'stdout' argument
180	n/a	with self.assertRaises(ValueError) as c:
181	n/a	output = subprocess.check_output(
182	n/a	[sys.executable, "-c", "print('will not be run')"],
183	n/a	stdout=sys.stdout)
184	n/a	self.fail("Expected ValueError when stdout arg supplied.")
185	n/a	self.assertIn('stdout', c.exception.args[0])
186	n/a
187	n/a	def test_check_output_stdin_with_input_arg(self):
188	n/a	# check_output() refuses to accept 'stdin' with 'input'
189	n/a	tf = tempfile.TemporaryFile()
190	n/a	self.addCleanup(tf.close)
191	n/a	tf.write(b'pear')
192	n/a	tf.seek(0)
193	n/a	with self.assertRaises(ValueError) as c:
194	n/a	output = subprocess.check_output(
195	n/a	[sys.executable, "-c", "print('will not be run')"],
196	n/a	stdin=tf, input=b'hare')
197	n/a	self.fail("Expected ValueError when stdin and input args supplied.")
198	n/a	self.assertIn('stdin', c.exception.args[0])
199	n/a	self.assertIn('input', c.exception.args[0])
200	n/a
201	n/a	def test_check_output_timeout(self):
202	n/a	# check_output() function with timeout arg
203	n/a	with self.assertRaises(subprocess.TimeoutExpired) as c:
204	n/a	output = subprocess.check_output(
205	n/a	[sys.executable, "-c",
206	n/a	"import sys, time\n"
207	n/a	"sys.stdout.write('BDFL')\n"
208	n/a	"sys.stdout.flush()\n"
209	n/a	"time.sleep(3600)"],
210	n/a	# Some heavily loaded buildbots (sparc Debian 3.x) require
211	n/a	# this much time to start and print.
212	n/a	timeout=3)
213	n/a	self.fail("Expected TimeoutExpired.")
214	n/a	self.assertEqual(c.exception.output, b'BDFL')
215	n/a
216	n/a	def test_call_kwargs(self):
217	n/a	# call() function with keyword args
218	n/a	newenv = os.environ.copy()
219	n/a	newenv["FRUIT"] = "banana"
220	n/a	rc = subprocess.call([sys.executable, "-c",
221	n/a	'import sys, os;'
222	n/a	'sys.exit(os.getenv("FRUIT")=="banana")'],
223	n/a	env=newenv)
224	n/a	self.assertEqual(rc, 1)
225	n/a
226	n/a	def test_invalid_args(self):
227	n/a	# Popen() called with invalid arguments should raise TypeError
228	n/a	# but Popen.__del__ should not complain (issue #12085)
229	n/a	with support.captured_stderr() as s:
230	n/a	self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1)
231	n/a	argcount = subprocess.Popen.__init__.__code__.co_argcount
232	n/a	too_many_args = [0] * (argcount + 1)
233	n/a	self.assertRaises(TypeError, subprocess.Popen, *too_many_args)
234	n/a	self.assertEqual(s.getvalue(), '')
235	n/a
236	n/a	def test_stdin_none(self):
237	n/a	# .stdin is None when not redirected
238	n/a	p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
239	n/a	stdout=subprocess.PIPE, stderr=subprocess.PIPE)
240	n/a	self.addCleanup(p.stdout.close)
241	n/a	self.addCleanup(p.stderr.close)
242	n/a	p.wait()
243	n/a	self.assertEqual(p.stdin, None)
244	n/a
245	n/a	def test_stdout_none(self):
246	n/a	# .stdout is None when not redirected, and the child's stdout will
247	n/a	# be inherited from the parent. In order to test this we run a
248	n/a	# subprocess in a subprocess:
249	n/a	# this_test
250	n/a	# \-- subprocess created by this test (parent)
251	n/a	# \-- subprocess created by the parent subprocess (child)
252	n/a	# The parent doesn't specify stdout, so the child will use the
253	n/a	# parent's stdout. This test checks that the message printed by the
254	n/a	# child goes to the parent stdout. The parent also checks that the
255	n/a	# child's stdout is None. See #11963.
256	n/a	code = ('import sys; from subprocess import Popen, PIPE;'
257	n/a	'p = Popen([sys.executable, "-c", "print(\'test_stdout_none\')"],'
258	n/a	' stdin=PIPE, stderr=PIPE);'
259	n/a	'p.wait(); assert p.stdout is None;')
260	n/a	p = subprocess.Popen([sys.executable, "-c", code],
261	n/a	stdout=subprocess.PIPE, stderr=subprocess.PIPE)
262	n/a	self.addCleanup(p.stdout.close)
263	n/a	self.addCleanup(p.stderr.close)
264	n/a	out, err = p.communicate()
265	n/a	self.assertEqual(p.returncode, 0, err)
266	n/a	self.assertEqual(out.rstrip(), b'test_stdout_none')
267	n/a
268	n/a	def test_stderr_none(self):
269	n/a	# .stderr is None when not redirected
270	n/a	p = subprocess.Popen([sys.executable, "-c", 'print("banana")'],
271	n/a	stdin=subprocess.PIPE, stdout=subprocess.PIPE)
272	n/a	self.addCleanup(p.stdout.close)
273	n/a	self.addCleanup(p.stdin.close)
274	n/a	p.wait()
275	n/a	self.assertEqual(p.stderr, None)
276	n/a
277	n/a	def _assert_python(self, pre_args, **kwargs):
278	n/a	# We include sys.exit() to prevent the test runner from hanging
279	n/a	# whenever python is found.
280	n/a	args = pre_args + ["import sys; sys.exit(47)"]
281	n/a	p = subprocess.Popen(args, **kwargs)
282	n/a	p.wait()
283	n/a	self.assertEqual(47, p.returncode)
284	n/a
285	n/a	def test_executable(self):
286	n/a	# Check that the executable argument works.
287	n/a	#
288	n/a	# On Unix (non-Mac and non-Windows), Python looks at args[0] to
289	n/a	# determine where its standard library is, so we need the directory
290	n/a	# of args[0] to be valid for the Popen() call to Python to succeed.
291	n/a	# See also issue #16170 and issue #7774.
292	n/a	doesnotexist = os.path.join(os.path.dirname(sys.executable),
293	n/a	"doesnotexist")
294	n/a	self._assert_python([doesnotexist, "-c"], executable=sys.executable)
295	n/a
296	n/a	def test_executable_takes_precedence(self):
297	n/a	# Check that the executable argument takes precedence over args[0].
298	n/a	#
299	n/a	# Verify first that the call succeeds without the executable arg.
300	n/a	pre_args = [sys.executable, "-c"]
301	n/a	self._assert_python(pre_args)
302	n/a	self.assertRaises((FileNotFoundError, PermissionError),
303	n/a	self._assert_python, pre_args,
304	n/a	executable="doesnotexist")
305	n/a
306	n/a	@unittest.skipIf(mswindows, "executable argument replaces shell")
307	n/a	def test_executable_replaces_shell(self):
308	n/a	# Check that the executable argument replaces the default shell
309	n/a	# when shell=True.
310	n/a	self._assert_python([], executable=sys.executable, shell=True)
311	n/a
312	n/a	# For use in the test_cwd* tests below.
313	n/a	def _normalize_cwd(self, cwd):
314	n/a	# Normalize an expected cwd (for Tru64 support).
315	n/a	# We can't use os.path.realpath since it doesn't expand Tru64 {memb}
316	n/a	# strings. See bug #1063571.
317	n/a	with support.change_cwd(cwd):
318	n/a	return os.getcwd()
319	n/a
320	n/a	# For use in the test_cwd* tests below.
321	n/a	def _split_python_path(self):
322	n/a	# Return normalized (python_dir, python_base).
323	n/a	python_path = os.path.realpath(sys.executable)
324	n/a	return os.path.split(python_path)
325	n/a
326	n/a	# For use in the test_cwd* tests below.
327	n/a	def _assert_cwd(self, expected_cwd, python_arg, **kwargs):
328	n/a	# Invoke Python via Popen, and assert that (1) the call succeeds,
329	n/a	# and that (2) the current working directory of the child process
330	n/a	# matches expected_cwd.
331	n/a	p = subprocess.Popen([python_arg, "-c",
332	n/a	"import os, sys; "
333	n/a	"sys.stdout.write(os.getcwd()); "
334	n/a	"sys.exit(47)"],
335	n/a	stdout=subprocess.PIPE,
336	n/a	**kwargs)
337	n/a	self.addCleanup(p.stdout.close)
338	n/a	p.wait()
339	n/a	self.assertEqual(47, p.returncode)
340	n/a	normcase = os.path.normcase
341	n/a	self.assertEqual(normcase(expected_cwd),
342	n/a	normcase(p.stdout.read().decode("utf-8")))
343	n/a
344	n/a	def test_cwd(self):
345	n/a	# Check that cwd changes the cwd for the child process.
346	n/a	temp_dir = tempfile.gettempdir()
347	n/a	temp_dir = self._normalize_cwd(temp_dir)
348	n/a	self._assert_cwd(temp_dir, sys.executable, cwd=temp_dir)
349	n/a
350	n/a	@unittest.skipIf(mswindows, "pending resolution of issue #15533")
351	n/a	def test_cwd_with_relative_arg(self):
352	n/a	# Check that Popen looks for args[0] relative to cwd if args[0]
353	n/a	# is relative.
354	n/a	python_dir, python_base = self._split_python_path()
355	n/a	rel_python = os.path.join(os.curdir, python_base)
356	n/a	with support.temp_cwd() as wrong_dir:
357	n/a	# Before calling with the correct cwd, confirm that the call fails
358	n/a	# without cwd and with the wrong cwd.
359	n/a	self.assertRaises(FileNotFoundError, subprocess.Popen,
360	n/a	[rel_python])
361	n/a	self.assertRaises(FileNotFoundError, subprocess.Popen,
362	n/a	[rel_python], cwd=wrong_dir)
363	n/a	python_dir = self._normalize_cwd(python_dir)
364	n/a	self._assert_cwd(python_dir, rel_python, cwd=python_dir)
365	n/a
366	n/a	@unittest.skipIf(mswindows, "pending resolution of issue #15533")
367	n/a	def test_cwd_with_relative_executable(self):
368	n/a	# Check that Popen looks for executable relative to cwd if executable
369	n/a	# is relative (and that executable takes precedence over args[0]).
370	n/a	python_dir, python_base = self._split_python_path()
371	n/a	rel_python = os.path.join(os.curdir, python_base)
372	n/a	doesntexist = "somethingyoudonthave"
373	n/a	with support.temp_cwd() as wrong_dir:
374	n/a	# Before calling with the correct cwd, confirm that the call fails
375	n/a	# without cwd and with the wrong cwd.
376	n/a	self.assertRaises(FileNotFoundError, subprocess.Popen,
377	n/a	[doesntexist], executable=rel_python)
378	n/a	self.assertRaises(FileNotFoundError, subprocess.Popen,
379	n/a	[doesntexist], executable=rel_python,
380	n/a	cwd=wrong_dir)
381	n/a	python_dir = self._normalize_cwd(python_dir)
382	n/a	self._assert_cwd(python_dir, doesntexist, executable=rel_python,
383	n/a	cwd=python_dir)
384	n/a
385	n/a	def test_cwd_with_absolute_arg(self):
386	n/a	# Check that Popen can find the executable when the cwd is wrong
387	n/a	# if args[0] is an absolute path.
388	n/a	python_dir, python_base = self._split_python_path()
389	n/a	abs_python = os.path.join(python_dir, python_base)
390	n/a	rel_python = os.path.join(os.curdir, python_base)
391	n/a	with support.temp_dir() as wrong_dir:
392	n/a	# Before calling with an absolute path, confirm that using a
393	n/a	# relative path fails.
394	n/a	self.assertRaises(FileNotFoundError, subprocess.Popen,
395	n/a	[rel_python], cwd=wrong_dir)
396	n/a	wrong_dir = self._normalize_cwd(wrong_dir)
397	n/a	self._assert_cwd(wrong_dir, abs_python, cwd=wrong_dir)
398	n/a
399	n/a	@unittest.skipIf(sys.base_prefix != sys.prefix,
400	n/a	'Test is not venv-compatible')
401	n/a	def test_executable_with_cwd(self):
402	n/a	python_dir, python_base = self._split_python_path()
403	n/a	python_dir = self._normalize_cwd(python_dir)
404	n/a	self._assert_cwd(python_dir, "somethingyoudonthave",
405	n/a	executable=sys.executable, cwd=python_dir)
406	n/a
407	n/a	@unittest.skipIf(sys.base_prefix != sys.prefix,
408	n/a	'Test is not venv-compatible')
409	n/a	@unittest.skipIf(sysconfig.is_python_build(),
410	n/a	"need an installed Python. See #7774")
411	n/a	def test_executable_without_cwd(self):
412	n/a	# For a normal installation, it should work without 'cwd'
413	n/a	# argument. For test runs in the build directory, see #7774.
414	n/a	self._assert_cwd(os.getcwd(), "somethingyoudonthave",
415	n/a	executable=sys.executable)
416	n/a
417	n/a	def test_stdin_pipe(self):
418	n/a	# stdin redirection
419	n/a	p = subprocess.Popen([sys.executable, "-c",
420	n/a	'import sys; sys.exit(sys.stdin.read() == "pear")'],
421	n/a	stdin=subprocess.PIPE)
422	n/a	p.stdin.write(b"pear")
423	n/a	p.stdin.close()
424	n/a	p.wait()
425	n/a	self.assertEqual(p.returncode, 1)
426	n/a
427	n/a	def test_stdin_filedes(self):
428	n/a	# stdin is set to open file descriptor
429	n/a	tf = tempfile.TemporaryFile()
430	n/a	self.addCleanup(tf.close)
431	n/a	d = tf.fileno()
432	n/a	os.write(d, b"pear")
433	n/a	os.lseek(d, 0, 0)
434	n/a	p = subprocess.Popen([sys.executable, "-c",
435	n/a	'import sys; sys.exit(sys.stdin.read() == "pear")'],
436	n/a	stdin=d)
437	n/a	p.wait()
438	n/a	self.assertEqual(p.returncode, 1)
439	n/a
440	n/a	def test_stdin_fileobj(self):
441	n/a	# stdin is set to open file object
442	n/a	tf = tempfile.TemporaryFile()
443	n/a	self.addCleanup(tf.close)
444	n/a	tf.write(b"pear")
445	n/a	tf.seek(0)
446	n/a	p = subprocess.Popen([sys.executable, "-c",
447	n/a	'import sys; sys.exit(sys.stdin.read() == "pear")'],
448	n/a	stdin=tf)
449	n/a	p.wait()
450	n/a	self.assertEqual(p.returncode, 1)
451	n/a
452	n/a	def test_stdout_pipe(self):
453	n/a	# stdout redirection
454	n/a	p = subprocess.Popen([sys.executable, "-c",
455	n/a	'import sys; sys.stdout.write("orange")'],
456	n/a	stdout=subprocess.PIPE)
457	n/a	with p:
458	n/a	self.assertEqual(p.stdout.read(), b"orange")
459	n/a
460	n/a	def test_stdout_filedes(self):
461	n/a	# stdout is set to open file descriptor
462	n/a	tf = tempfile.TemporaryFile()
463	n/a	self.addCleanup(tf.close)
464	n/a	d = tf.fileno()
465	n/a	p = subprocess.Popen([sys.executable, "-c",
466	n/a	'import sys; sys.stdout.write("orange")'],
467	n/a	stdout=d)
468	n/a	p.wait()
469	n/a	os.lseek(d, 0, 0)
470	n/a	self.assertEqual(os.read(d, 1024), b"orange")
471	n/a
472	n/a	def test_stdout_fileobj(self):
473	n/a	# stdout is set to open file object
474	n/a	tf = tempfile.TemporaryFile()
475	n/a	self.addCleanup(tf.close)
476	n/a	p = subprocess.Popen([sys.executable, "-c",
477	n/a	'import sys; sys.stdout.write("orange")'],
478	n/a	stdout=tf)
479	n/a	p.wait()
480	n/a	tf.seek(0)
481	n/a	self.assertEqual(tf.read(), b"orange")
482	n/a
483	n/a	def test_stderr_pipe(self):
484	n/a	# stderr redirection
485	n/a	p = subprocess.Popen([sys.executable, "-c",
486	n/a	'import sys; sys.stderr.write("strawberry")'],
487	n/a	stderr=subprocess.PIPE)
488	n/a	with p:
489	n/a	self.assertStderrEqual(p.stderr.read(), b"strawberry")
490	n/a
491	n/a	def test_stderr_filedes(self):
492	n/a	# stderr is set to open file descriptor
493	n/a	tf = tempfile.TemporaryFile()
494	n/a	self.addCleanup(tf.close)
495	n/a	d = tf.fileno()
496	n/a	p = subprocess.Popen([sys.executable, "-c",
497	n/a	'import sys; sys.stderr.write("strawberry")'],
498	n/a	stderr=d)
499	n/a	p.wait()
500	n/a	os.lseek(d, 0, 0)
501	n/a	self.assertStderrEqual(os.read(d, 1024), b"strawberry")
502	n/a
503	n/a	def test_stderr_fileobj(self):
504	n/a	# stderr is set to open file object
505	n/a	tf = tempfile.TemporaryFile()
506	n/a	self.addCleanup(tf.close)
507	n/a	p = subprocess.Popen([sys.executable, "-c",
508	n/a	'import sys; sys.stderr.write("strawberry")'],
509	n/a	stderr=tf)
510	n/a	p.wait()
511	n/a	tf.seek(0)
512	n/a	self.assertStderrEqual(tf.read(), b"strawberry")
513	n/a
514	n/a	def test_stderr_redirect_with_no_stdout_redirect(self):
515	n/a	# test stderr=STDOUT while stdout=None (not set)
516	n/a
517	n/a	# - grandchild prints to stderr
518	n/a	# - child redirects grandchild's stderr to its stdout
519	n/a	# - the parent should get grandchild's stderr in child's stdout
520	n/a	p = subprocess.Popen([sys.executable, "-c",
521	n/a	'import sys, subprocess;'
522	n/a	'rc = subprocess.call([sys.executable, "-c",'
523	n/a	' "import sys;"'
524	n/a	' "sys.stderr.write(\'42\')"],'
525	n/a	' stderr=subprocess.STDOUT);'
526	n/a	'sys.exit(rc)'],
527	n/a	stdout=subprocess.PIPE,
528	n/a	stderr=subprocess.PIPE)
529	n/a	stdout, stderr = p.communicate()
530	n/a	#NOTE: stdout should get stderr from grandchild
531	n/a	self.assertStderrEqual(stdout, b'42')
532	n/a	self.assertStderrEqual(stderr, b'') # should be empty
533	n/a	self.assertEqual(p.returncode, 0)
534	n/a
535	n/a	def test_stdout_stderr_pipe(self):
536	n/a	# capture stdout and stderr to the same pipe
537	n/a	p = subprocess.Popen([sys.executable, "-c",
538	n/a	'import sys;'
539	n/a	'sys.stdout.write("apple");'
540	n/a	'sys.stdout.flush();'
541	n/a	'sys.stderr.write("orange")'],
542	n/a	stdout=subprocess.PIPE,
543	n/a	stderr=subprocess.STDOUT)
544	n/a	with p:
545	n/a	self.assertStderrEqual(p.stdout.read(), b"appleorange")
546	n/a
547	n/a	def test_stdout_stderr_file(self):
548	n/a	# capture stdout and stderr to the same open file
549	n/a	tf = tempfile.TemporaryFile()
550	n/a	self.addCleanup(tf.close)
551	n/a	p = subprocess.Popen([sys.executable, "-c",
552	n/a	'import sys;'
553	n/a	'sys.stdout.write("apple");'
554	n/a	'sys.stdout.flush();'
555	n/a	'sys.stderr.write("orange")'],
556	n/a	stdout=tf,
557	n/a	stderr=tf)
558	n/a	p.wait()
559	n/a	tf.seek(0)
560	n/a	self.assertStderrEqual(tf.read(), b"appleorange")
561	n/a
562	n/a	def test_stdout_filedes_of_stdout(self):
563	n/a	# stdout is set to 1 (#1531862).
564	n/a	# To avoid printing the text on stdout, we do something similar to
565	n/a	# test_stdout_none (see above). The parent subprocess calls the child
566	n/a	# subprocess passing stdout=1, and this test uses stdout=PIPE in
567	n/a	# order to capture and check the output of the parent. See #11963.
568	n/a	code = ('import sys, subprocess; '
569	n/a	'rc = subprocess.call([sys.executable, "-c", '
570	n/a	' "import os, sys; sys.exit(os.write(sys.stdout.fileno(), '
571	n/a	'b\'test with stdout=1\'))"], stdout=1); '
572	n/a	'assert rc == 18')
573	n/a	p = subprocess.Popen([sys.executable, "-c", code],
574	n/a	stdout=subprocess.PIPE, stderr=subprocess.PIPE)
575	n/a	self.addCleanup(p.stdout.close)
576	n/a	self.addCleanup(p.stderr.close)
577	n/a	out, err = p.communicate()
578	n/a	self.assertEqual(p.returncode, 0, err)
579	n/a	self.assertEqual(out.rstrip(), b'test with stdout=1')
580	n/a
581	n/a	def test_stdout_devnull(self):
582	n/a	p = subprocess.Popen([sys.executable, "-c",
583	n/a	'for i in range(10240):'
584	n/a	'print("x" * 1024)'],
585	n/a	stdout=subprocess.DEVNULL)
586	n/a	p.wait()
587	n/a	self.assertEqual(p.stdout, None)
588	n/a
589	n/a	def test_stderr_devnull(self):
590	n/a	p = subprocess.Popen([sys.executable, "-c",
591	n/a	'import sys\n'
592	n/a	'for i in range(10240):'
593	n/a	'sys.stderr.write("x" * 1024)'],
594	n/a	stderr=subprocess.DEVNULL)
595	n/a	p.wait()
596	n/a	self.assertEqual(p.stderr, None)
597	n/a
598	n/a	def test_stdin_devnull(self):
599	n/a	p = subprocess.Popen([sys.executable, "-c",
600	n/a	'import sys;'
601	n/a	'sys.stdin.read(1)'],
602	n/a	stdin=subprocess.DEVNULL)
603	n/a	p.wait()
604	n/a	self.assertEqual(p.stdin, None)
605	n/a
606	n/a	def test_env(self):
607	n/a	newenv = os.environ.copy()
608	n/a	newenv["FRUIT"] = "orange"
609	n/a	with subprocess.Popen([sys.executable, "-c",
610	n/a	'import sys,os;'
611	n/a	'sys.stdout.write(os.getenv("FRUIT"))'],
612	n/a	stdout=subprocess.PIPE,
613	n/a	env=newenv) as p:
614	n/a	stdout, stderr = p.communicate()
615	n/a	self.assertEqual(stdout, b"orange")
616	n/a
617	n/a	# Windows requires at least the SYSTEMROOT environment variable to start
618	n/a	# Python
619	n/a	@unittest.skipIf(sys.platform == 'win32',
620	n/a	'cannot test an empty env on Windows')
621	n/a	@unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') is not None,
622	n/a	'the python library cannot be loaded '
623	n/a	'with an empty environment')
624	n/a	def test_empty_env(self):
625	n/a	with subprocess.Popen([sys.executable, "-c",
626	n/a	'import os; '
627	n/a	'print(list(os.environ.keys()))'],
628	n/a	stdout=subprocess.PIPE,
629	n/a	env={}) as p:
630	n/a	stdout, stderr = p.communicate()
631	n/a	self.assertIn(stdout.strip(),
632	n/a	(b"[]",
633	n/a	# Mac OS X adds __CF_USER_TEXT_ENCODING variable to an empty
634	n/a	# environment
635	n/a	b"['__CF_USER_TEXT_ENCODING']"))
636	n/a
637	n/a	def test_communicate_stdin(self):
638	n/a	p = subprocess.Popen([sys.executable, "-c",
639	n/a	'import sys;'
640	n/a	'sys.exit(sys.stdin.read() == "pear")'],
641	n/a	stdin=subprocess.PIPE)
642	n/a	p.communicate(b"pear")
643	n/a	self.assertEqual(p.returncode, 1)
644	n/a
645	n/a	def test_communicate_stdout(self):
646	n/a	p = subprocess.Popen([sys.executable, "-c",
647	n/a	'import sys; sys.stdout.write("pineapple")'],
648	n/a	stdout=subprocess.PIPE)
649	n/a	(stdout, stderr) = p.communicate()
650	n/a	self.assertEqual(stdout, b"pineapple")
651	n/a	self.assertEqual(stderr, None)
652	n/a
653	n/a	def test_communicate_stderr(self):
654	n/a	p = subprocess.Popen([sys.executable, "-c",
655	n/a	'import sys; sys.stderr.write("pineapple")'],
656	n/a	stderr=subprocess.PIPE)
657	n/a	(stdout, stderr) = p.communicate()
658	n/a	self.assertEqual(stdout, None)
659	n/a	self.assertStderrEqual(stderr, b"pineapple")
660	n/a
661	n/a	def test_communicate(self):
662	n/a	p = subprocess.Popen([sys.executable, "-c",
663	n/a	'import sys,os;'
664	n/a	'sys.stderr.write("pineapple");'
665	n/a	'sys.stdout.write(sys.stdin.read())'],
666	n/a	stdin=subprocess.PIPE,
667	n/a	stdout=subprocess.PIPE,
668	n/a	stderr=subprocess.PIPE)
669	n/a	self.addCleanup(p.stdout.close)
670	n/a	self.addCleanup(p.stderr.close)
671	n/a	self.addCleanup(p.stdin.close)
672	n/a	(stdout, stderr) = p.communicate(b"banana")
673	n/a	self.assertEqual(stdout, b"banana")
674	n/a	self.assertStderrEqual(stderr, b"pineapple")
675	n/a
676	n/a	def test_communicate_timeout(self):
677	n/a	p = subprocess.Popen([sys.executable, "-c",
678	n/a	'import sys,os,time;'
679	n/a	'sys.stderr.write("pineapple\\n");'
680	n/a	'time.sleep(1);'
681	n/a	'sys.stderr.write("pear\\n");'
682	n/a	'sys.stdout.write(sys.stdin.read())'],
683	n/a	universal_newlines=True,
684	n/a	stdin=subprocess.PIPE,
685	n/a	stdout=subprocess.PIPE,
686	n/a	stderr=subprocess.PIPE)
687	n/a	self.assertRaises(subprocess.TimeoutExpired, p.communicate, "banana",
688	n/a	timeout=0.3)
689	n/a	# Make sure we can keep waiting for it, and that we get the whole output
690	n/a	# after it completes.
691	n/a	(stdout, stderr) = p.communicate()
692	n/a	self.assertEqual(stdout, "banana")
693	n/a	self.assertStderrEqual(stderr.encode(), b"pineapple\npear\n")
694	n/a
695	n/a	def test_communicate_timeout_large_output(self):
696	n/a	# Test an expiring timeout while the child is outputting lots of data.
697	n/a	p = subprocess.Popen([sys.executable, "-c",
698	n/a	'import sys,os,time;'
699	n/a	'sys.stdout.write("a" * (64 * 1024));'
700	n/a	'time.sleep(0.2);'
701	n/a	'sys.stdout.write("a" * (64 * 1024));'
702	n/a	'time.sleep(0.2);'
703	n/a	'sys.stdout.write("a" * (64 * 1024));'
704	n/a	'time.sleep(0.2);'
705	n/a	'sys.stdout.write("a" * (64 * 1024));'],
706	n/a	stdout=subprocess.PIPE)
707	n/a	self.assertRaises(subprocess.TimeoutExpired, p.communicate, timeout=0.4)
708	n/a	(stdout, _) = p.communicate()
709	n/a	self.assertEqual(len(stdout), 4 * 64 * 1024)
710	n/a
711	n/a	# Test for the fd leak reported in http://bugs.python.org/issue2791.
712	n/a	def test_communicate_pipe_fd_leak(self):
713	n/a	for stdin_pipe in (False, True):
714	n/a	for stdout_pipe in (False, True):
715	n/a	for stderr_pipe in (False, True):
716	n/a	options = {}
717	n/a	if stdin_pipe:
718	n/a	options['stdin'] = subprocess.PIPE
719	n/a	if stdout_pipe:
720	n/a	options['stdout'] = subprocess.PIPE
721	n/a	if stderr_pipe:
722	n/a	options['stderr'] = subprocess.PIPE
723	n/a	if not options:
724	n/a	continue
725	n/a	p = subprocess.Popen((sys.executable, "-c", "pass"), **options)
726	n/a	p.communicate()
727	n/a	if p.stdin is not None:
728	n/a	self.assertTrue(p.stdin.closed)
729	n/a	if p.stdout is not None:
730	n/a	self.assertTrue(p.stdout.closed)
731	n/a	if p.stderr is not None:
732	n/a	self.assertTrue(p.stderr.closed)
733	n/a
734	n/a	def test_communicate_returns(self):
735	n/a	# communicate() should return None if no redirection is active
736	n/a	p = subprocess.Popen([sys.executable, "-c",
737	n/a	"import sys; sys.exit(47)"])
738	n/a	(stdout, stderr) = p.communicate()
739	n/a	self.assertEqual(stdout, None)
740	n/a	self.assertEqual(stderr, None)
741	n/a
742	n/a	def test_communicate_pipe_buf(self):
743	n/a	# communicate() with writes larger than pipe_buf
744	n/a	# This test will probably deadlock rather than fail, if
745	n/a	# communicate() does not work properly.
746	n/a	x, y = os.pipe()
747	n/a	os.close(x)
748	n/a	os.close(y)
749	n/a	p = subprocess.Popen([sys.executable, "-c",
750	n/a	'import sys,os;'
751	n/a	'sys.stdout.write(sys.stdin.read(47));'
752	n/a	'sys.stderr.write("x" * %d);'
753	n/a	'sys.stdout.write(sys.stdin.read())' %
754	n/a	support.PIPE_MAX_SIZE],
755	n/a	stdin=subprocess.PIPE,
756	n/a	stdout=subprocess.PIPE,
757	n/a	stderr=subprocess.PIPE)
758	n/a	self.addCleanup(p.stdout.close)
759	n/a	self.addCleanup(p.stderr.close)
760	n/a	self.addCleanup(p.stdin.close)
761	n/a	string_to_write = b"a" * support.PIPE_MAX_SIZE
762	n/a	(stdout, stderr) = p.communicate(string_to_write)
763	n/a	self.assertEqual(stdout, string_to_write)
764	n/a
765	n/a	def test_writes_before_communicate(self):
766	n/a	# stdin.write before communicate()
767	n/a	p = subprocess.Popen([sys.executable, "-c",
768	n/a	'import sys,os;'
769	n/a	'sys.stdout.write(sys.stdin.read())'],
770	n/a	stdin=subprocess.PIPE,
771	n/a	stdout=subprocess.PIPE,
772	n/a	stderr=subprocess.PIPE)
773	n/a	self.addCleanup(p.stdout.close)
774	n/a	self.addCleanup(p.stderr.close)
775	n/a	self.addCleanup(p.stdin.close)
776	n/a	p.stdin.write(b"banana")
777	n/a	(stdout, stderr) = p.communicate(b"split")
778	n/a	self.assertEqual(stdout, b"bananasplit")
779	n/a	self.assertStderrEqual(stderr, b"")
780	n/a
781	n/a	def test_universal_newlines(self):
782	n/a	p = subprocess.Popen([sys.executable, "-c",
783	n/a	'import sys,os;' + SETBINARY +
784	n/a	'buf = sys.stdout.buffer;'
785	n/a	'buf.write(sys.stdin.readline().encode());'
786	n/a	'buf.flush();'
787	n/a	'buf.write(b"line2\\n");'
788	n/a	'buf.flush();'
789	n/a	'buf.write(sys.stdin.read().encode());'
790	n/a	'buf.flush();'
791	n/a	'buf.write(b"line4\\n");'
792	n/a	'buf.flush();'
793	n/a	'buf.write(b"line5\\r\\n");'
794	n/a	'buf.flush();'
795	n/a	'buf.write(b"line6\\r");'
796	n/a	'buf.flush();'
797	n/a	'buf.write(b"\\nline7");'
798	n/a	'buf.flush();'
799	n/a	'buf.write(b"\\nline8");'],
800	n/a	stdin=subprocess.PIPE,
801	n/a	stdout=subprocess.PIPE,
802	n/a	universal_newlines=1)
803	n/a	with p:
804	n/a	p.stdin.write("line1\n")
805	n/a	p.stdin.flush()
806	n/a	self.assertEqual(p.stdout.readline(), "line1\n")
807	n/a	p.stdin.write("line3\n")
808	n/a	p.stdin.close()
809	n/a	self.addCleanup(p.stdout.close)
810	n/a	self.assertEqual(p.stdout.readline(),
811	n/a	"line2\n")
812	n/a	self.assertEqual(p.stdout.read(6),
813	n/a	"line3\n")
814	n/a	self.assertEqual(p.stdout.read(),
815	n/a	"line4\nline5\nline6\nline7\nline8")
816	n/a
817	n/a	def test_universal_newlines_communicate(self):
818	n/a	# universal newlines through communicate()
819	n/a	p = subprocess.Popen([sys.executable, "-c",
820	n/a	'import sys,os;' + SETBINARY +
821	n/a	'buf = sys.stdout.buffer;'
822	n/a	'buf.write(b"line2\\n");'
823	n/a	'buf.flush();'
824	n/a	'buf.write(b"line4\\n");'
825	n/a	'buf.flush();'
826	n/a	'buf.write(b"line5\\r\\n");'
827	n/a	'buf.flush();'
828	n/a	'buf.write(b"line6\\r");'
829	n/a	'buf.flush();'
830	n/a	'buf.write(b"\\nline7");'
831	n/a	'buf.flush();'
832	n/a	'buf.write(b"\\nline8");'],
833	n/a	stderr=subprocess.PIPE,
834	n/a	stdout=subprocess.PIPE,
835	n/a	universal_newlines=1)
836	n/a	self.addCleanup(p.stdout.close)
837	n/a	self.addCleanup(p.stderr.close)
838	n/a	(stdout, stderr) = p.communicate()
839	n/a	self.assertEqual(stdout,
840	n/a	"line2\nline4\nline5\nline6\nline7\nline8")
841	n/a
842	n/a	def test_universal_newlines_communicate_stdin(self):
843	n/a	# universal newlines through communicate(), with only stdin
844	n/a	p = subprocess.Popen([sys.executable, "-c",
845	n/a	'import sys,os;' + SETBINARY + textwrap.dedent('''
846	n/a	s = sys.stdin.readline()
847	n/a	assert s == "line1\\n", repr(s)
848	n/a	s = sys.stdin.read()
849	n/a	assert s == "line3\\n", repr(s)
850	n/a	''')],
851	n/a	stdin=subprocess.PIPE,
852	n/a	universal_newlines=1)
853	n/a	(stdout, stderr) = p.communicate("line1\nline3\n")
854	n/a	self.assertEqual(p.returncode, 0)
855	n/a
856	n/a	def test_universal_newlines_communicate_input_none(self):
857	n/a	# Test communicate(input=None) with universal newlines.
858	n/a	#
859	n/a	# We set stdout to PIPE because, as of this writing, a different
860	n/a	# code path is tested when the number of pipes is zero or one.
861	n/a	p = subprocess.Popen([sys.executable, "-c", "pass"],
862	n/a	stdin=subprocess.PIPE,
863	n/a	stdout=subprocess.PIPE,
864	n/a	universal_newlines=True)
865	n/a	p.communicate()
866	n/a	self.assertEqual(p.returncode, 0)
867	n/a
868	n/a	def test_universal_newlines_communicate_stdin_stdout_stderr(self):
869	n/a	# universal newlines through communicate(), with stdin, stdout, stderr
870	n/a	p = subprocess.Popen([sys.executable, "-c",
871	n/a	'import sys,os;' + SETBINARY + textwrap.dedent('''
872	n/a	s = sys.stdin.buffer.readline()
873	n/a	sys.stdout.buffer.write(s)
874	n/a	sys.stdout.buffer.write(b"line2\\r")
875	n/a	sys.stderr.buffer.write(b"eline2\\n")
876	n/a	s = sys.stdin.buffer.read()
877	n/a	sys.stdout.buffer.write(s)
878	n/a	sys.stdout.buffer.write(b"line4\\n")
879	n/a	sys.stdout.buffer.write(b"line5\\r\\n")
880	n/a	sys.stderr.buffer.write(b"eline6\\r")
881	n/a	sys.stderr.buffer.write(b"eline7\\r\\nz")
882	n/a	''')],
883	n/a	stdin=subprocess.PIPE,
884	n/a	stderr=subprocess.PIPE,
885	n/a	stdout=subprocess.PIPE,
886	n/a	universal_newlines=True)
887	n/a	self.addCleanup(p.stdout.close)
888	n/a	self.addCleanup(p.stderr.close)
889	n/a	(stdout, stderr) = p.communicate("line1\nline3\n")
890	n/a	self.assertEqual(p.returncode, 0)
891	n/a	self.assertEqual("line1\nline2\nline3\nline4\nline5\n", stdout)
892	n/a	# Python debug build push something like "[42442 refs]\n"
893	n/a	# to stderr at exit of subprocess.
894	n/a	# Don't use assertStderrEqual because it strips CR and LF from output.
895	n/a	self.assertTrue(stderr.startswith("eline2\neline6\neline7\n"))
896	n/a
897	n/a	def test_universal_newlines_communicate_encodings(self):
898	n/a	# Check that universal newlines mode works for various encodings,
899	n/a	# in particular for encodings in the UTF-16 and UTF-32 families.
900	n/a	# See issue #15595.
901	n/a	#
902	n/a	# UTF-16 and UTF-32-BE are sufficient to check both with BOM and
903	n/a	# without, and UTF-16 and UTF-32.
904	n/a	for encoding in ['utf-16', 'utf-32-be']:
905	n/a	code = ("import sys; "
906	n/a	r"sys.stdout.buffer.write('1\r\n2\r3\n4'.encode('%s'))" %
907	n/a	encoding)
908	n/a	args = [sys.executable, '-c', code]
909	n/a	# We set stdin to be non-None because, as of this writing,
910	n/a	# a different code path is used when the number of pipes is
911	n/a	# zero or one.
912	n/a	popen = subprocess.Popen(args,
913	n/a	stdin=subprocess.PIPE,
914	n/a	stdout=subprocess.PIPE,
915	n/a	encoding=encoding)
916	n/a	stdout, stderr = popen.communicate(input='')
917	n/a	self.assertEqual(stdout, '1\n2\n3\n4')
918	n/a
919	n/a	def test_communicate_errors(self):
920	n/a	for errors, expected in [
921	n/a	('ignore', ''),
922	n/a	('replace', '\ufffd\ufffd'),
923	n/a	('surrogateescape', '\udc80\udc80'),
924	n/a	('backslashreplace', '\\x80\\x80'),
925	n/a	]:
926	n/a	code = ("import sys; "
927	n/a	r"sys.stdout.buffer.write(b'[\x80\x80]')")
928	n/a	args = [sys.executable, '-c', code]
929	n/a	# We set stdin to be non-None because, as of this writing,
930	n/a	# a different code path is used when the number of pipes is
931	n/a	# zero or one.
932	n/a	popen = subprocess.Popen(args,
933	n/a	stdin=subprocess.PIPE,
934	n/a	stdout=subprocess.PIPE,
935	n/a	encoding='utf-8',
936	n/a	errors=errors)
937	n/a	stdout, stderr = popen.communicate(input='')
938	n/a	self.assertEqual(stdout, '[{}]'.format(expected))
939	n/a
940	n/a	def test_no_leaking(self):
941	n/a	# Make sure we leak no resources
942	n/a	if not mswindows:
943	n/a	max_handles = 1026 # too much for most UNIX systems
944	n/a	else:
945	n/a	max_handles = 2050 # too much for (at least some) Windows setups
946	n/a	handles = []
947	n/a	tmpdir = tempfile.mkdtemp()
948	n/a	try:
949	n/a	for i in range(max_handles):
950	n/a	try:
951	n/a	tmpfile = os.path.join(tmpdir, support.TESTFN)
952	n/a	handles.append(os.open(tmpfile, os.O_WRONLY\|os.O_CREAT))
953	n/a	except OSError as e:
954	n/a	if e.errno != errno.EMFILE:
955	n/a	raise
956	n/a	break
957	n/a	else:
958	n/a	self.skipTest("failed to reach the file descriptor limit "
959	n/a	"(tried %d)" % max_handles)
960	n/a	# Close a couple of them (should be enough for a subprocess)
961	n/a	for i in range(10):
962	n/a	os.close(handles.pop())
963	n/a	# Loop creating some subprocesses. If one of them leaks some fds,
964	n/a	# the next loop iteration will fail by reaching the max fd limit.
965	n/a	for i in range(15):
966	n/a	p = subprocess.Popen([sys.executable, "-c",
967	n/a	"import sys;"
968	n/a	"sys.stdout.write(sys.stdin.read())"],
969	n/a	stdin=subprocess.PIPE,
970	n/a	stdout=subprocess.PIPE,
971	n/a	stderr=subprocess.PIPE)
972	n/a	data = p.communicate(b"lime")[0]
973	n/a	self.assertEqual(data, b"lime")
974	n/a	finally:
975	n/a	for h in handles:
976	n/a	os.close(h)
977	n/a	shutil.rmtree(tmpdir)
978	n/a
979	n/a	def test_list2cmdline(self):
980	n/a	self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']),
981	n/a	'"a b c" d e')
982	n/a	self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']),
983	n/a	'ab\\"c \\ d')
984	n/a	self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']),
985	n/a	'ab\\"c " \\\\" d')
986	n/a	self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']),
987	n/a	'a\\\\\\b "de fg" h')
988	n/a	self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']),
989	n/a	'a\\\\\\"b c d')
990	n/a	self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']),
991	n/a	'"a\\\\b c" d e')
992	n/a	self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']),
993	n/a	'"a\\\\b\\ c" d e')
994	n/a	self.assertEqual(subprocess.list2cmdline(['ab', '']),
995	n/a	'ab ""')
996	n/a
997	n/a	def test_poll(self):
998	n/a	p = subprocess.Popen([sys.executable, "-c",
999	n/a	"import os; os.read(0, 1)"],
1000	n/a	stdin=subprocess.PIPE)
1001	n/a	self.addCleanup(p.stdin.close)
1002	n/a	self.assertIsNone(p.poll())
1003	n/a	os.write(p.stdin.fileno(), b'A')
1004	n/a	p.wait()
1005	n/a	# Subsequent invocations should just return the returncode
1006	n/a	self.assertEqual(p.poll(), 0)
1007	n/a
1008	n/a	def test_wait(self):
1009	n/a	p = subprocess.Popen([sys.executable, "-c", "pass"])
1010	n/a	self.assertEqual(p.wait(), 0)
1011	n/a	# Subsequent invocations should just return the returncode
1012	n/a	self.assertEqual(p.wait(), 0)
1013	n/a
1014	n/a	def test_wait_timeout(self):
1015	n/a	p = subprocess.Popen([sys.executable,
1016	n/a	"-c", "import time; time.sleep(0.3)"])
1017	n/a	with self.assertRaises(subprocess.TimeoutExpired) as c:
1018	n/a	p.wait(timeout=0.0001)
1019	n/a	self.assertIn("0.0001", str(c.exception)) # For coverage of __str__.
1020	n/a	# Some heavily loaded buildbots (sparc Debian 3.x) require this much
1021	n/a	# time to start.
1022	n/a	self.assertEqual(p.wait(timeout=3), 0)
1023	n/a
1024	n/a	def test_invalid_bufsize(self):
1025	n/a	# an invalid type of the bufsize argument should raise
1026	n/a	# TypeError.
1027	n/a	with self.assertRaises(TypeError):
1028	n/a	subprocess.Popen([sys.executable, "-c", "pass"], "orange")
1029	n/a
1030	n/a	def test_bufsize_is_none(self):
1031	n/a	# bufsize=None should be the same as bufsize=0.
1032	n/a	p = subprocess.Popen([sys.executable, "-c", "pass"], None)
1033	n/a	self.assertEqual(p.wait(), 0)
1034	n/a	# Again with keyword arg
1035	n/a	p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None)
1036	n/a	self.assertEqual(p.wait(), 0)
1037	n/a
1038	n/a	def _test_bufsize_equal_one(self, line, expected, universal_newlines):
1039	n/a	# subprocess may deadlock with bufsize=1, see issue #21332
1040	n/a	with subprocess.Popen([sys.executable, "-c", "import sys;"
1041	n/a	"sys.stdout.write(sys.stdin.readline());"
1042	n/a	"sys.stdout.flush()"],
1043	n/a	stdin=subprocess.PIPE,
1044	n/a	stdout=subprocess.PIPE,
1045	n/a	stderr=subprocess.DEVNULL,
1046	n/a	bufsize=1,
1047	n/a	universal_newlines=universal_newlines) as p:
1048	n/a	p.stdin.write(line) # expect that it flushes the line in text mode
1049	n/a	os.close(p.stdin.fileno()) # close it without flushing the buffer
1050	n/a	read_line = p.stdout.readline()
1051	n/a	try:
1052	n/a	p.stdin.close()
1053	n/a	except OSError:
1054	n/a	pass
1055	n/a	p.stdin = None
1056	n/a	self.assertEqual(p.returncode, 0)
1057	n/a	self.assertEqual(read_line, expected)
1058	n/a
1059	n/a	def test_bufsize_equal_one_text_mode(self):
1060	n/a	# line is flushed in text mode with bufsize=1.
1061	n/a	# we should get the full line in return
1062	n/a	line = "line\n"
1063	n/a	self._test_bufsize_equal_one(line, line, universal_newlines=True)
1064	n/a
1065	n/a	def test_bufsize_equal_one_binary_mode(self):
1066	n/a	# line is not flushed in binary mode with bufsize=1.
1067	n/a	# we should get empty response
1068	n/a	line = b'line' + os.linesep.encode() # assume ascii-based locale
1069	n/a	self._test_bufsize_equal_one(line, b'', universal_newlines=False)
1070	n/a
1071	n/a	def test_leaking_fds_on_error(self):
1072	n/a	# see bug #5179: Popen leaks file descriptors to PIPEs if
1073	n/a	# the child fails to execute; this will eventually exhaust
1074	n/a	# the maximum number of open fds. 1024 seems a very common
1075	n/a	# value for that limit, but Windows has 2048, so we loop
1076	n/a	# 1024 times (each call leaked two fds).
1077	n/a	for i in range(1024):
1078	n/a	with self.assertRaises(OSError) as c:
1079	n/a	subprocess.Popen(['nonexisting_i_hope'],
1080	n/a	stdout=subprocess.PIPE,
1081	n/a	stderr=subprocess.PIPE)
1082	n/a	# ignore errors that indicate the command was not found
1083	n/a	if c.exception.errno not in (errno.ENOENT, errno.EACCES):
1084	n/a	raise c.exception
1085	n/a
1086	n/a	@unittest.skipIf(threading is None, "threading required")
1087	n/a	def test_double_close_on_error(self):
1088	n/a	# Issue #18851
1089	n/a	fds = []
1090	n/a	def open_fds():
1091	n/a	for i in range(20):
1092	n/a	fds.extend(os.pipe())
1093	n/a	time.sleep(0.001)
1094	n/a	t = threading.Thread(target=open_fds)
1095	n/a	t.start()
1096	n/a	try:
1097	n/a	with self.assertRaises(EnvironmentError):
1098	n/a	subprocess.Popen(['nonexisting_i_hope'],
1099	n/a	stdin=subprocess.PIPE,
1100	n/a	stdout=subprocess.PIPE,
1101	n/a	stderr=subprocess.PIPE)
1102	n/a	finally:
1103	n/a	t.join()
1104	n/a	exc = None
1105	n/a	for fd in fds:
1106	n/a	# If a double close occurred, some of those fds will
1107	n/a	# already have been closed by mistake, and os.close()
1108	n/a	# here will raise.
1109	n/a	try:
1110	n/a	os.close(fd)
1111	n/a	except OSError as e:
1112	n/a	exc = e
1113	n/a	if exc is not None:
1114	n/a	raise exc
1115	n/a
1116	n/a	@unittest.skipIf(threading is None, "threading required")
1117	n/a	def test_threadsafe_wait(self):
1118	n/a	"""Issue21291: Popen.wait() needs to be threadsafe for returncode."""
1119	n/a	proc = subprocess.Popen([sys.executable, '-c',
1120	n/a	'import time; time.sleep(12)'])
1121	n/a	self.assertEqual(proc.returncode, None)
1122	n/a	results = []
1123	n/a
1124	n/a	def kill_proc_timer_thread():
1125	n/a	results.append(('thread-start-poll-result', proc.poll()))
1126	n/a	# terminate it from the thread and wait for the result.
1127	n/a	proc.kill()
1128	n/a	proc.wait()
1129	n/a	results.append(('thread-after-kill-and-wait', proc.returncode))
1130	n/a	# this wait should be a no-op given the above.
1131	n/a	proc.wait()
1132	n/a	results.append(('thread-after-second-wait', proc.returncode))
1133	n/a
1134	n/a	# This is a timing sensitive test, the failure mode is
1135	n/a	# triggered when both the main thread and this thread are in
1136	n/a	# the wait() call at once. The delay here is to allow the
1137	n/a	# main thread to most likely be blocked in its wait() call.
1138	n/a	t = threading.Timer(0.2, kill_proc_timer_thread)
1139	n/a	t.start()
1140	n/a
1141	n/a	if mswindows:
1142	n/a	expected_errorcode = 1
1143	n/a	else:
1144	n/a	# Should be -9 because of the proc.kill() from the thread.
1145	n/a	expected_errorcode = -9
1146	n/a
1147	n/a	# Wait for the process to finish; the thread should kill it
1148	n/a	# long before it finishes on its own. Supplying a timeout
1149	n/a	# triggers a different code path for better coverage.
1150	n/a	proc.wait(timeout=20)
1151	n/a	self.assertEqual(proc.returncode, expected_errorcode,
1152	n/a	msg="unexpected result in wait from main thread")
1153	n/a
1154	n/a	# This should be a no-op with no change in returncode.
1155	n/a	proc.wait()
1156	n/a	self.assertEqual(proc.returncode, expected_errorcode,
1157	n/a	msg="unexpected result in second main wait.")
1158	n/a
1159	n/a	t.join()
1160	n/a	# Ensure that all of the thread results are as expected.
1161	n/a	# When a race condition occurs in wait(), the returncode could
1162	n/a	# be set by the wrong thread that doesn't actually have it
1163	n/a	# leading to an incorrect value.
1164	n/a	self.assertEqual([('thread-start-poll-result', None),
1165	n/a	('thread-after-kill-and-wait', expected_errorcode),
1166	n/a	('thread-after-second-wait', expected_errorcode)],
1167	n/a	results)
1168	n/a
1169	n/a	def test_issue8780(self):
1170	n/a	# Ensure that stdout is inherited from the parent
1171	n/a	# if stdout=PIPE is not used
1172	n/a	code = ';'.join((
1173	n/a	'import subprocess, sys',
1174	n/a	'retcode = subprocess.call('
1175	n/a	"[sys.executable, '-c', 'print(\"Hello World!\")'])",
1176	n/a	'assert retcode == 0'))
1177	n/a	output = subprocess.check_output([sys.executable, '-c', code])
1178	n/a	self.assertTrue(output.startswith(b'Hello World!'), ascii(output))
1179	n/a
1180	n/a	def test_handles_closed_on_exception(self):
1181	n/a	# If CreateProcess exits with an error, ensure the
1182	n/a	# duplicate output handles are released
1183	n/a	ifhandle, ifname = tempfile.mkstemp()
1184	n/a	ofhandle, ofname = tempfile.mkstemp()
1185	n/a	efhandle, efname = tempfile.mkstemp()
1186	n/a	try:
1187	n/a	subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle,
1188	n/a	stderr=efhandle)
1189	n/a	except OSError:
1190	n/a	os.close(ifhandle)
1191	n/a	os.remove(ifname)
1192	n/a	os.close(ofhandle)
1193	n/a	os.remove(ofname)
1194	n/a	os.close(efhandle)
1195	n/a	os.remove(efname)
1196	n/a	self.assertFalse(os.path.exists(ifname))
1197	n/a	self.assertFalse(os.path.exists(ofname))
1198	n/a	self.assertFalse(os.path.exists(efname))
1199	n/a
1200	n/a	def test_communicate_epipe(self):
1201	n/a	# Issue 10963: communicate() should hide EPIPE
1202	n/a	p = subprocess.Popen([sys.executable, "-c", 'pass'],
1203	n/a	stdin=subprocess.PIPE,
1204	n/a	stdout=subprocess.PIPE,
1205	n/a	stderr=subprocess.PIPE)
1206	n/a	self.addCleanup(p.stdout.close)
1207	n/a	self.addCleanup(p.stderr.close)
1208	n/a	self.addCleanup(p.stdin.close)
1209	n/a	p.communicate(b"x" * 2**20)
1210	n/a
1211	n/a	def test_communicate_epipe_only_stdin(self):
1212	n/a	# Issue 10963: communicate() should hide EPIPE
1213	n/a	p = subprocess.Popen([sys.executable, "-c", 'pass'],
1214	n/a	stdin=subprocess.PIPE)
1215	n/a	self.addCleanup(p.stdin.close)
1216	n/a	p.wait()
1217	n/a	p.communicate(b"x" * 2**20)
1218	n/a
1219	n/a	@unittest.skipUnless(hasattr(signal, 'SIGUSR1'),
1220	n/a	"Requires signal.SIGUSR1")
1221	n/a	@unittest.skipUnless(hasattr(os, 'kill'),
1222	n/a	"Requires os.kill")
1223	n/a	@unittest.skipUnless(hasattr(os, 'getppid'),
1224	n/a	"Requires os.getppid")
1225	n/a	def test_communicate_eintr(self):
1226	n/a	# Issue #12493: communicate() should handle EINTR
1227	n/a	def handler(signum, frame):
1228	n/a	pass
1229	n/a	old_handler = signal.signal(signal.SIGUSR1, handler)
1230	n/a	self.addCleanup(signal.signal, signal.SIGUSR1, old_handler)
1231	n/a
1232	n/a	args = [sys.executable, "-c",
1233	n/a	'import os, signal;'
1234	n/a	'os.kill(os.getppid(), signal.SIGUSR1)']
1235	n/a	for stream in ('stdout', 'stderr'):
1236	n/a	kw = {stream: subprocess.PIPE}
1237	n/a	with subprocess.Popen(args, **kw) as process:
1238	n/a	# communicate() will be interrupted by SIGUSR1
1239	n/a	process.communicate()
1240	n/a
1241	n/a
1242	n/a	# This test is Linux-ish specific for simplicity to at least have
1243	n/a	# some coverage. It is not a platform specific bug.
1244	n/a	@unittest.skipUnless(os.path.isdir('/proc/%d/fd' % os.getpid()),
1245	n/a	"Linux specific")
1246	n/a	def test_failed_child_execute_fd_leak(self):
1247	n/a	"""Test for the fork() failure fd leak reported in issue16327."""
1248	n/a	fd_directory = '/proc/%d/fd' % os.getpid()
1249	n/a	fds_before_popen = os.listdir(fd_directory)
1250	n/a	with self.assertRaises(PopenTestException):
1251	n/a	PopenExecuteChildRaises(
1252	n/a	[sys.executable, '-c', 'pass'], stdin=subprocess.PIPE,
1253	n/a	stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1254	n/a
1255	n/a	# NOTE: This test doesn't verify that the real _execute_child
1256	n/a	# does not close the file descriptors itself on the way out
1257	n/a	# during an exception. Code inspection has confirmed that.
1258	n/a
1259	n/a	fds_after_exception = os.listdir(fd_directory)
1260	n/a	self.assertEqual(fds_before_popen, fds_after_exception)
1261	n/a
1262	n/a
1263	n/a	class RunFuncTestCase(BaseTestCase):
1264	n/a	def run_python(self, code, **kwargs):
1265	n/a	"""Run Python code in a subprocess using subprocess.run"""
1266	n/a	argv = [sys.executable, "-c", code]
1267	n/a	return subprocess.run(argv, **kwargs)
1268	n/a
1269	n/a	def test_returncode(self):
1270	n/a	# call() function with sequence argument
1271	n/a	cp = self.run_python("import sys; sys.exit(47)")
1272	n/a	self.assertEqual(cp.returncode, 47)
1273	n/a	with self.assertRaises(subprocess.CalledProcessError):
1274	n/a	cp.check_returncode()
1275	n/a
1276	n/a	def test_check(self):
1277	n/a	with self.assertRaises(subprocess.CalledProcessError) as c:
1278	n/a	self.run_python("import sys; sys.exit(47)", check=True)
1279	n/a	self.assertEqual(c.exception.returncode, 47)
1280	n/a
1281	n/a	def test_check_zero(self):
1282	n/a	# check_returncode shouldn't raise when returncode is zero
1283	n/a	cp = self.run_python("import sys; sys.exit(0)", check=True)
1284	n/a	self.assertEqual(cp.returncode, 0)
1285	n/a
1286	n/a	def test_timeout(self):
1287	n/a	# run() function with timeout argument; we want to test that the child
1288	n/a	# process gets killed when the timeout expires. If the child isn't
1289	n/a	# killed, this call will deadlock since subprocess.run waits for the
1290	n/a	# child.
1291	n/a	with self.assertRaises(subprocess.TimeoutExpired):
1292	n/a	self.run_python("while True: pass", timeout=0.0001)
1293	n/a
1294	n/a	def test_capture_stdout(self):
1295	n/a	# capture stdout with zero return code
1296	n/a	cp = self.run_python("print('BDFL')", stdout=subprocess.PIPE)
1297	n/a	self.assertIn(b'BDFL', cp.stdout)
1298	n/a
1299	n/a	def test_capture_stderr(self):
1300	n/a	cp = self.run_python("import sys; sys.stderr.write('BDFL')",
1301	n/a	stderr=subprocess.PIPE)
1302	n/a	self.assertIn(b'BDFL', cp.stderr)
1303	n/a
1304	n/a	def test_check_output_stdin_arg(self):
1305	n/a	# run() can be called with stdin set to a file
1306	n/a	tf = tempfile.TemporaryFile()
1307	n/a	self.addCleanup(tf.close)
1308	n/a	tf.write(b'pear')
1309	n/a	tf.seek(0)
1310	n/a	cp = self.run_python(
1311	n/a	"import sys; sys.stdout.write(sys.stdin.read().upper())",
1312	n/a	stdin=tf, stdout=subprocess.PIPE)
1313	n/a	self.assertIn(b'PEAR', cp.stdout)
1314	n/a
1315	n/a	def test_check_output_input_arg(self):
1316	n/a	# check_output() can be called with input set to a string
1317	n/a	cp = self.run_python(
1318	n/a	"import sys; sys.stdout.write(sys.stdin.read().upper())",
1319	n/a	input=b'pear', stdout=subprocess.PIPE)
1320	n/a	self.assertIn(b'PEAR', cp.stdout)
1321	n/a
1322	n/a	def test_check_output_stdin_with_input_arg(self):
1323	n/a	# run() refuses to accept 'stdin' with 'input'
1324	n/a	tf = tempfile.TemporaryFile()
1325	n/a	self.addCleanup(tf.close)
1326	n/a	tf.write(b'pear')
1327	n/a	tf.seek(0)
1328	n/a	with self.assertRaises(ValueError,
1329	n/a	msg="Expected ValueError when stdin and input args supplied.") as c:
1330	n/a	output = self.run_python("print('will not be run')",
1331	n/a	stdin=tf, input=b'hare')
1332	n/a	self.assertIn('stdin', c.exception.args[0])
1333	n/a	self.assertIn('input', c.exception.args[0])
1334	n/a
1335	n/a	def test_check_output_timeout(self):
1336	n/a	with self.assertRaises(subprocess.TimeoutExpired) as c:
1337	n/a	cp = self.run_python((
1338	n/a	"import sys, time\n"
1339	n/a	"sys.stdout.write('BDFL')\n"
1340	n/a	"sys.stdout.flush()\n"
1341	n/a	"time.sleep(3600)"),
1342	n/a	# Some heavily loaded buildbots (sparc Debian 3.x) require
1343	n/a	# this much time to start and print.
1344	n/a	timeout=3, stdout=subprocess.PIPE)
1345	n/a	self.assertEqual(c.exception.output, b'BDFL')
1346	n/a	# output is aliased to stdout
1347	n/a	self.assertEqual(c.exception.stdout, b'BDFL')
1348	n/a
1349	n/a	def test_run_kwargs(self):
1350	n/a	newenv = os.environ.copy()
1351	n/a	newenv["FRUIT"] = "banana"
1352	n/a	cp = self.run_python(('import sys, os;'
1353	n/a	'sys.exit(33 if os.getenv("FRUIT")=="banana" else 31)'),
1354	n/a	env=newenv)
1355	n/a	self.assertEqual(cp.returncode, 33)
1356	n/a
1357	n/a
1358	n/a	@unittest.skipIf(mswindows, "POSIX specific tests")
1359	n/a	class POSIXProcessTestCase(BaseTestCase):
1360	n/a
1361	n/a	def setUp(self):
1362	n/a	super().setUp()
1363	n/a	self._nonexistent_dir = "/_this/pa.th/does/not/exist"
1364	n/a
1365	n/a	def _get_chdir_exception(self):
1366	n/a	try:
1367	n/a	os.chdir(self._nonexistent_dir)
1368	n/a	except OSError as e:
1369	n/a	# This avoids hard coding the errno value or the OS perror()
1370	n/a	# string and instead capture the exception that we want to see
1371	n/a	# below for comparison.
1372	n/a	desired_exception = e
1373	n/a	desired_exception.strerror += ': ' + repr(self._nonexistent_dir)
1374	n/a	else:
1375	n/a	self.fail("chdir to nonexistent directory %s succeeded." %
1376	n/a	self._nonexistent_dir)
1377	n/a	return desired_exception
1378	n/a
1379	n/a	def test_exception_cwd(self):
1380	n/a	"""Test error in the child raised in the parent for a bad cwd."""
1381	n/a	desired_exception = self._get_chdir_exception()
1382	n/a	try:
1383	n/a	p = subprocess.Popen([sys.executable, "-c", ""],
1384	n/a	cwd=self._nonexistent_dir)
1385	n/a	except OSError as e:
1386	n/a	# Test that the child process chdir failure actually makes
1387	n/a	# it up to the parent process as the correct exception.
1388	n/a	self.assertEqual(desired_exception.errno, e.errno)
1389	n/a	self.assertEqual(desired_exception.strerror, e.strerror)
1390	n/a	else:
1391	n/a	self.fail("Expected OSError: %s" % desired_exception)
1392	n/a
1393	n/a	def test_exception_bad_executable(self):
1394	n/a	"""Test error in the child raised in the parent for a bad executable."""
1395	n/a	desired_exception = self._get_chdir_exception()
1396	n/a	try:
1397	n/a	p = subprocess.Popen([sys.executable, "-c", ""],
1398	n/a	executable=self._nonexistent_dir)
1399	n/a	except OSError as e:
1400	n/a	# Test that the child process exec failure actually makes
1401	n/a	# it up to the parent process as the correct exception.
1402	n/a	self.assertEqual(desired_exception.errno, e.errno)
1403	n/a	self.assertEqual(desired_exception.strerror, e.strerror)
1404	n/a	else:
1405	n/a	self.fail("Expected OSError: %s" % desired_exception)
1406	n/a
1407	n/a	def test_exception_bad_args_0(self):
1408	n/a	"""Test error in the child raised in the parent for a bad args[0]."""
1409	n/a	desired_exception = self._get_chdir_exception()
1410	n/a	try:
1411	n/a	p = subprocess.Popen([self._nonexistent_dir, "-c", ""])
1412	n/a	except OSError as e:
1413	n/a	# Test that the child process exec failure actually makes
1414	n/a	# it up to the parent process as the correct exception.
1415	n/a	self.assertEqual(desired_exception.errno, e.errno)
1416	n/a	self.assertEqual(desired_exception.strerror, e.strerror)
1417	n/a	else:
1418	n/a	self.fail("Expected OSError: %s" % desired_exception)
1419	n/a
1420	n/a	def test_restore_signals(self):
1421	n/a	# Code coverage for both values of restore_signals to make sure it
1422	n/a	# at least does not blow up.
1423	n/a	# A test for behavior would be complex. Contributions welcome.
1424	n/a	subprocess.call([sys.executable, "-c", ""], restore_signals=True)
1425	n/a	subprocess.call([sys.executable, "-c", ""], restore_signals=False)
1426	n/a
1427	n/a	def test_start_new_session(self):
1428	n/a	# For code coverage of calling setsid(). We don't care if we get an
1429	n/a	# EPERM error from it depending on the test execution environment, that
1430	n/a	# still indicates that it was called.
1431	n/a	try:
1432	n/a	output = subprocess.check_output(
1433	n/a	[sys.executable, "-c",
1434	n/a	"import os; print(os.getpgid(os.getpid()))"],
1435	n/a	start_new_session=True)
1436	n/a	except OSError as e:
1437	n/a	if e.errno != errno.EPERM:
1438	n/a	raise
1439	n/a	else:
1440	n/a	parent_pgid = os.getpgid(os.getpid())
1441	n/a	child_pgid = int(output)
1442	n/a	self.assertNotEqual(parent_pgid, child_pgid)
1443	n/a
1444	n/a	def test_run_abort(self):
1445	n/a	# returncode handles signal termination
1446	n/a	with support.SuppressCrashReport():
1447	n/a	p = subprocess.Popen([sys.executable, "-c",
1448	n/a	'import os; os.abort()'])
1449	n/a	p.wait()
1450	n/a	self.assertEqual(-p.returncode, signal.SIGABRT)
1451	n/a
1452	n/a	def test_CalledProcessError_str_signal(self):
1453	n/a	err = subprocess.CalledProcessError(-int(signal.SIGABRT), "fake cmd")
1454	n/a	error_string = str(err)
1455	n/a	# We're relying on the repr() of the signal.Signals intenum to provide
1456	n/a	# the word signal, the signal name and the numeric value.
1457	n/a	self.assertIn("signal", error_string.lower())
1458	n/a	# We're not being specific about the signal name as some signals have
1459	n/a	# multiple names and which name is revealed can vary.
1460	n/a	self.assertIn("SIG", error_string)
1461	n/a	self.assertIn(str(signal.SIGABRT), error_string)
1462	n/a
1463	n/a	def test_CalledProcessError_str_unknown_signal(self):
1464	n/a	err = subprocess.CalledProcessError(-9876543, "fake cmd")
1465	n/a	error_string = str(err)
1466	n/a	self.assertIn("unknown signal 9876543.", error_string)
1467	n/a
1468	n/a	def test_CalledProcessError_str_non_zero(self):
1469	n/a	err = subprocess.CalledProcessError(2, "fake cmd")
1470	n/a	error_string = str(err)
1471	n/a	self.assertIn("non-zero exit status 2.", error_string)
1472	n/a
1473	n/a	def test_preexec(self):
1474	n/a	# DISCLAIMER: Setting environment variables is not a good use
1475	n/a	# of a preexec_fn. This is merely a test.
1476	n/a	p = subprocess.Popen([sys.executable, "-c",
1477	n/a	'import sys,os;'
1478	n/a	'sys.stdout.write(os.getenv("FRUIT"))'],
1479	n/a	stdout=subprocess.PIPE,
1480	n/a	preexec_fn=lambda: os.putenv("FRUIT", "apple"))
1481	n/a	with p:
1482	n/a	self.assertEqual(p.stdout.read(), b"apple")
1483	n/a
1484	n/a	def test_preexec_exception(self):
1485	n/a	def raise_it():
1486	n/a	raise ValueError("What if two swallows carried a coconut?")
1487	n/a	try:
1488	n/a	p = subprocess.Popen([sys.executable, "-c", ""],
1489	n/a	preexec_fn=raise_it)
1490	n/a	except subprocess.SubprocessError as e:
1491	n/a	self.assertTrue(
1492	n/a	subprocess._posixsubprocess,
1493	n/a	"Expected a ValueError from the preexec_fn")
1494	n/a	except ValueError as e:
1495	n/a	self.assertIn("coconut", e.args[0])
1496	n/a	else:
1497	n/a	self.fail("Exception raised by preexec_fn did not make it "
1498	n/a	"to the parent process.")
1499	n/a
1500	n/a	class _TestExecuteChildPopen(subprocess.Popen):
1501	n/a	"""Used to test behavior at the end of _execute_child."""
1502	n/a	def __init__(self, testcase, args, *kwargs):
1503	n/a	self._testcase = testcase
1504	n/a	subprocess.Popen.__init__(self, args, *kwargs)
1505	n/a
1506	n/a	def _execute_child(self, args, *kwargs):
1507	n/a	try:
1508	n/a	subprocess.Popen._execute_child(self, args, *kwargs)
1509	n/a	finally:
1510	n/a	# Open a bunch of file descriptors and verify that
1511	n/a	# none of them are the same as the ones the Popen
1512	n/a	# instance is using for stdin/stdout/stderr.
1513	n/a	devzero_fds = [os.open("/dev/zero", os.O_RDONLY)
1514	n/a	for _ in range(8)]
1515	n/a	try:
1516	n/a	for fd in devzero_fds:
1517	n/a	self._testcase.assertNotIn(
1518	n/a	fd, (self.stdin.fileno(), self.stdout.fileno(),
1519	n/a	self.stderr.fileno()),
1520	n/a	msg="At least one fd was closed early.")
1521	n/a	finally:
1522	n/a	for fd in devzero_fds:
1523	n/a	os.close(fd)
1524	n/a
1525	n/a	@unittest.skipIf(not os.path.exists("/dev/zero"), "/dev/zero required.")
1526	n/a	def test_preexec_errpipe_does_not_double_close_pipes(self):
1527	n/a	"""Issue16140: Don't double close pipes on preexec error."""
1528	n/a
1529	n/a	def raise_it():
1530	n/a	raise subprocess.SubprocessError(
1531	n/a	"force the _execute_child() errpipe_data path.")
1532	n/a
1533	n/a	with self.assertRaises(subprocess.SubprocessError):
1534	n/a	self._TestExecuteChildPopen(
1535	n/a	self, [sys.executable, "-c", "pass"],
1536	n/a	stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1537	n/a	stderr=subprocess.PIPE, preexec_fn=raise_it)
1538	n/a
1539	n/a	def test_preexec_gc_module_failure(self):
1540	n/a	# This tests the code that disables garbage collection if the child
1541	n/a	# process will execute any Python.
1542	n/a	def raise_runtime_error():
1543	n/a	raise RuntimeError("this shouldn't escape")
1544	n/a	enabled = gc.isenabled()
1545	n/a	orig_gc_disable = gc.disable
1546	n/a	orig_gc_isenabled = gc.isenabled
1547	n/a	try:
1548	n/a	gc.disable()
1549	n/a	self.assertFalse(gc.isenabled())
1550	n/a	subprocess.call([sys.executable, '-c', ''],
1551	n/a	preexec_fn=lambda: None)
1552	n/a	self.assertFalse(gc.isenabled(),
1553	n/a	"Popen enabled gc when it shouldn't.")
1554	n/a
1555	n/a	gc.enable()
1556	n/a	self.assertTrue(gc.isenabled())
1557	n/a	subprocess.call([sys.executable, '-c', ''],
1558	n/a	preexec_fn=lambda: None)
1559	n/a	self.assertTrue(gc.isenabled(), "Popen left gc disabled.")
1560	n/a
1561	n/a	gc.disable = raise_runtime_error
1562	n/a	self.assertRaises(RuntimeError, subprocess.Popen,
1563	n/a	[sys.executable, '-c', ''],
1564	n/a	preexec_fn=lambda: None)
1565	n/a
1566	n/a	del gc.isenabled # force an AttributeError
1567	n/a	self.assertRaises(AttributeError, subprocess.Popen,
1568	n/a	[sys.executable, '-c', ''],
1569	n/a	preexec_fn=lambda: None)
1570	n/a	finally:
1571	n/a	gc.disable = orig_gc_disable
1572	n/a	gc.isenabled = orig_gc_isenabled
1573	n/a	if not enabled:
1574	n/a	gc.disable()
1575	n/a
1576	n/a	@unittest.skipIf(
1577	n/a	sys.platform == 'darwin', 'setrlimit() seems to fail on OS X')
1578	n/a	def test_preexec_fork_failure(self):
1579	n/a	# The internal code did not preserve the previous exception when
1580	n/a	# re-enabling garbage collection
1581	n/a	try:
1582	n/a	from resource import getrlimit, setrlimit, RLIMIT_NPROC
1583	n/a	except ImportError as err:
1584	n/a	self.skipTest(err) # RLIMIT_NPROC is specific to Linux and BSD
1585	n/a	limits = getrlimit(RLIMIT_NPROC)
1586	n/a	[_, hard] = limits
1587	n/a	setrlimit(RLIMIT_NPROC, (0, hard))
1588	n/a	self.addCleanup(setrlimit, RLIMIT_NPROC, limits)
1589	n/a	try:
1590	n/a	subprocess.call([sys.executable, '-c', ''],
1591	n/a	preexec_fn=lambda: None)
1592	n/a	except BlockingIOError:
1593	n/a	# Forking should raise EAGAIN, translated to BlockingIOError
1594	n/a	pass
1595	n/a	else:
1596	n/a	self.skipTest('RLIMIT_NPROC had no effect; probably superuser')
1597	n/a
1598	n/a	def test_args_string(self):
1599	n/a	# args is a string
1600	n/a	fd, fname = tempfile.mkstemp()
1601	n/a	# reopen in text mode
1602	n/a	with open(fd, "w", errors="surrogateescape") as fobj:
1603	n/a	fobj.write("#!%s\n" % support.unix_shell)
1604	n/a	fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
1605	n/a	sys.executable)
1606	n/a	os.chmod(fname, 0o700)
1607	n/a	p = subprocess.Popen(fname)
1608	n/a	p.wait()
1609	n/a	os.remove(fname)
1610	n/a	self.assertEqual(p.returncode, 47)
1611	n/a
1612	n/a	def test_invalid_args(self):
1613	n/a	# invalid arguments should raise ValueError
1614	n/a	self.assertRaises(ValueError, subprocess.call,
1615	n/a	[sys.executable, "-c",
1616	n/a	"import sys; sys.exit(47)"],
1617	n/a	startupinfo=47)
1618	n/a	self.assertRaises(ValueError, subprocess.call,
1619	n/a	[sys.executable, "-c",
1620	n/a	"import sys; sys.exit(47)"],
1621	n/a	creationflags=47)
1622	n/a
1623	n/a	def test_shell_sequence(self):
1624	n/a	# Run command through the shell (sequence)
1625	n/a	newenv = os.environ.copy()
1626	n/a	newenv["FRUIT"] = "apple"
1627	n/a	p = subprocess.Popen(["echo $FRUIT"], shell=1,
1628	n/a	stdout=subprocess.PIPE,
1629	n/a	env=newenv)
1630	n/a	with p:
1631	n/a	self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
1632	n/a
1633	n/a	def test_shell_string(self):
1634	n/a	# Run command through the shell (string)
1635	n/a	newenv = os.environ.copy()
1636	n/a	newenv["FRUIT"] = "apple"
1637	n/a	p = subprocess.Popen("echo $FRUIT", shell=1,
1638	n/a	stdout=subprocess.PIPE,
1639	n/a	env=newenv)
1640	n/a	with p:
1641	n/a	self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
1642	n/a
1643	n/a	def test_call_string(self):
1644	n/a	# call() function with string argument on UNIX
1645	n/a	fd, fname = tempfile.mkstemp()
1646	n/a	# reopen in text mode
1647	n/a	with open(fd, "w", errors="surrogateescape") as fobj:
1648	n/a	fobj.write("#!%s\n" % support.unix_shell)
1649	n/a	fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" %
1650	n/a	sys.executable)
1651	n/a	os.chmod(fname, 0o700)
1652	n/a	rc = subprocess.call(fname)
1653	n/a	os.remove(fname)
1654	n/a	self.assertEqual(rc, 47)
1655	n/a
1656	n/a	def test_specific_shell(self):
1657	n/a	# Issue #9265: Incorrect name passed as arg[0].
1658	n/a	shells = []
1659	n/a	for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']:
1660	n/a	for name in ['bash', 'ksh']:
1661	n/a	sh = os.path.join(prefix, name)
1662	n/a	if os.path.isfile(sh):
1663	n/a	shells.append(sh)
1664	n/a	if not shells: # Will probably work for any shell but csh.
1665	n/a	self.skipTest("bash or ksh required for this test")
1666	n/a	sh = '/bin/sh'
1667	n/a	if os.path.isfile(sh) and not os.path.islink(sh):
1668	n/a	# Test will fail if /bin/sh is a symlink to csh.
1669	n/a	shells.append(sh)
1670	n/a	for sh in shells:
1671	n/a	p = subprocess.Popen("echo $0", executable=sh, shell=True,
1672	n/a	stdout=subprocess.PIPE)
1673	n/a	with p:
1674	n/a	self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii'))
1675	n/a
1676	n/a	def _kill_process(self, method, *args):
1677	n/a	# Do not inherit file handles from the parent.
1678	n/a	# It should fix failures on some platforms.
1679	n/a	# Also set the SIGINT handler to the default to make sure it's not
1680	n/a	# being ignored (some tests rely on that.)
1681	n/a	old_handler = signal.signal(signal.SIGINT, signal.default_int_handler)
1682	n/a	try:
1683	n/a	p = subprocess.Popen([sys.executable, "-c", """if 1:
1684	n/a	import sys, time
1685	n/a	sys.stdout.write('x\\n')
1686	n/a	sys.stdout.flush()
1687	n/a	time.sleep(30)
1688	n/a	"""],
1689	n/a	close_fds=True,
1690	n/a	stdin=subprocess.PIPE,
1691	n/a	stdout=subprocess.PIPE,
1692	n/a	stderr=subprocess.PIPE)
1693	n/a	finally:
1694	n/a	signal.signal(signal.SIGINT, old_handler)
1695	n/a	# Wait for the interpreter to be completely initialized before
1696	n/a	# sending any signal.
1697	n/a	p.stdout.read(1)
1698	n/a	getattr(p, method)(*args)
1699	n/a	return p
1700	n/a
1701	n/a	@unittest.skipIf(sys.platform.startswith(('netbsd', 'openbsd')),
1702	n/a	"Due to known OS bug (issue #16762)")
1703	n/a	def _kill_dead_process(self, method, *args):
1704	n/a	# Do not inherit file handles from the parent.
1705	n/a	# It should fix failures on some platforms.
1706	n/a	p = subprocess.Popen([sys.executable, "-c", """if 1:
1707	n/a	import sys, time
1708	n/a	sys.stdout.write('x\\n')
1709	n/a	sys.stdout.flush()
1710	n/a	"""],
1711	n/a	close_fds=True,
1712	n/a	stdin=subprocess.PIPE,
1713	n/a	stdout=subprocess.PIPE,
1714	n/a	stderr=subprocess.PIPE)
1715	n/a	# Wait for the interpreter to be completely initialized before
1716	n/a	# sending any signal.
1717	n/a	p.stdout.read(1)
1718	n/a	# The process should end after this
1719	n/a	time.sleep(1)
1720	n/a	# This shouldn't raise even though the child is now dead
1721	n/a	getattr(p, method)(*args)
1722	n/a	p.communicate()
1723	n/a
1724	n/a	def test_send_signal(self):
1725	n/a	p = self._kill_process('send_signal', signal.SIGINT)
1726	n/a	_, stderr = p.communicate()
1727	n/a	self.assertIn(b'KeyboardInterrupt', stderr)
1728	n/a	self.assertNotEqual(p.wait(), 0)
1729	n/a
1730	n/a	def test_kill(self):
1731	n/a	p = self._kill_process('kill')
1732	n/a	_, stderr = p.communicate()
1733	n/a	self.assertStderrEqual(stderr, b'')
1734	n/a	self.assertEqual(p.wait(), -signal.SIGKILL)
1735	n/a
1736	n/a	def test_terminate(self):
1737	n/a	p = self._kill_process('terminate')
1738	n/a	_, stderr = p.communicate()
1739	n/a	self.assertStderrEqual(stderr, b'')
1740	n/a	self.assertEqual(p.wait(), -signal.SIGTERM)
1741	n/a
1742	n/a	def test_send_signal_dead(self):
1743	n/a	# Sending a signal to a dead process
1744	n/a	self._kill_dead_process('send_signal', signal.SIGINT)
1745	n/a
1746	n/a	def test_kill_dead(self):
1747	n/a	# Killing a dead process
1748	n/a	self._kill_dead_process('kill')
1749	n/a
1750	n/a	def test_terminate_dead(self):
1751	n/a	# Terminating a dead process
1752	n/a	self._kill_dead_process('terminate')
1753	n/a
1754	n/a	def _save_fds(self, save_fds):
1755	n/a	fds = []
1756	n/a	for fd in save_fds:
1757	n/a	inheritable = os.get_inheritable(fd)
1758	n/a	saved = os.dup(fd)
1759	n/a	fds.append((fd, saved, inheritable))
1760	n/a	return fds
1761	n/a
1762	n/a	def _restore_fds(self, fds):
1763	n/a	for fd, saved, inheritable in fds:
1764	n/a	os.dup2(saved, fd, inheritable=inheritable)
1765	n/a	os.close(saved)
1766	n/a
1767	n/a	def check_close_std_fds(self, fds):
1768	n/a	# Issue #9905: test that subprocess pipes still work properly with
1769	n/a	# some standard fds closed
1770	n/a	stdin = 0
1771	n/a	saved_fds = self._save_fds(fds)
1772	n/a	for fd, saved, inheritable in saved_fds:
1773	n/a	if fd == 0:
1774	n/a	stdin = saved
1775	n/a	break
1776	n/a	try:
1777	n/a	for fd in fds:
1778	n/a	os.close(fd)
1779	n/a	out, err = subprocess.Popen([sys.executable, "-c",
1780	n/a	'import sys;'
1781	n/a	'sys.stdout.write("apple");'
1782	n/a	'sys.stdout.flush();'
1783	n/a	'sys.stderr.write("orange")'],
1784	n/a	stdin=stdin,
1785	n/a	stdout=subprocess.PIPE,
1786	n/a	stderr=subprocess.PIPE).communicate()
1787	n/a	err = support.strip_python_stderr(err)
1788	n/a	self.assertEqual((out, err), (b'apple', b'orange'))
1789	n/a	finally:
1790	n/a	self._restore_fds(saved_fds)
1791	n/a
1792	n/a	def test_close_fd_0(self):
1793	n/a	self.check_close_std_fds([0])
1794	n/a
1795	n/a	def test_close_fd_1(self):
1796	n/a	self.check_close_std_fds([1])
1797	n/a
1798	n/a	def test_close_fd_2(self):
1799	n/a	self.check_close_std_fds([2])
1800	n/a
1801	n/a	def test_close_fds_0_1(self):
1802	n/a	self.check_close_std_fds([0, 1])
1803	n/a
1804	n/a	def test_close_fds_0_2(self):
1805	n/a	self.check_close_std_fds([0, 2])
1806	n/a
1807	n/a	def test_close_fds_1_2(self):
1808	n/a	self.check_close_std_fds([1, 2])
1809	n/a
1810	n/a	def test_close_fds_0_1_2(self):
1811	n/a	# Issue #10806: test that subprocess pipes still work properly with
1812	n/a	# all standard fds closed.
1813	n/a	self.check_close_std_fds([0, 1, 2])
1814	n/a
1815	n/a	def test_small_errpipe_write_fd(self):
1816	n/a	"""Issue #15798: Popen should work when stdio fds are available."""
1817	n/a	new_stdin = os.dup(0)
1818	n/a	new_stdout = os.dup(1)
1819	n/a	try:
1820	n/a	os.close(0)
1821	n/a	os.close(1)
1822	n/a
1823	n/a	# Side test: if errpipe_write fails to have its CLOEXEC
1824	n/a	# flag set this should cause the parent to think the exec
1825	n/a	# failed. Extremely unlikely: everyone supports CLOEXEC.
1826	n/a	subprocess.Popen([
1827	n/a	sys.executable, "-c",
1828	n/a	"print('AssertionError:0:CLOEXEC failure.')"]).wait()
1829	n/a	finally:
1830	n/a	# Restore original stdin and stdout
1831	n/a	os.dup2(new_stdin, 0)
1832	n/a	os.dup2(new_stdout, 1)
1833	n/a	os.close(new_stdin)
1834	n/a	os.close(new_stdout)
1835	n/a
1836	n/a	def test_remapping_std_fds(self):
1837	n/a	# open up some temporary files
1838	n/a	temps = [tempfile.mkstemp() for i in range(3)]
1839	n/a	try:
1840	n/a	temp_fds = [fd for fd, fname in temps]
1841	n/a
1842	n/a	# unlink the files -- we won't need to reopen them
1843	n/a	for fd, fname in temps:
1844	n/a	os.unlink(fname)
1845	n/a
1846	n/a	# write some data to what will become stdin, and rewind
1847	n/a	os.write(temp_fds[1], b"STDIN")
1848	n/a	os.lseek(temp_fds[1], 0, 0)
1849	n/a
1850	n/a	# move the standard file descriptors out of the way
1851	n/a	saved_fds = self._save_fds(range(3))
1852	n/a	try:
1853	n/a	# duplicate the file objects over the standard fd's
1854	n/a	for fd, temp_fd in enumerate(temp_fds):
1855	n/a	os.dup2(temp_fd, fd)
1856	n/a
1857	n/a	# now use those files in the "wrong" order, so that subprocess
1858	n/a	# has to rearrange them in the child
1859	n/a	p = subprocess.Popen([sys.executable, "-c",
1860	n/a	'import sys; got = sys.stdin.read();'
1861	n/a	'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1862	n/a	stdin=temp_fds[1],
1863	n/a	stdout=temp_fds[2],
1864	n/a	stderr=temp_fds[0])
1865	n/a	p.wait()
1866	n/a	finally:
1867	n/a	self._restore_fds(saved_fds)
1868	n/a
1869	n/a	for fd in temp_fds:
1870	n/a	os.lseek(fd, 0, 0)
1871	n/a
1872	n/a	out = os.read(temp_fds[2], 1024)
1873	n/a	err = support.strip_python_stderr(os.read(temp_fds[0], 1024))
1874	n/a	self.assertEqual(out, b"got STDIN")
1875	n/a	self.assertEqual(err, b"err")
1876	n/a
1877	n/a	finally:
1878	n/a	for fd in temp_fds:
1879	n/a	os.close(fd)
1880	n/a
1881	n/a	def check_swap_fds(self, stdin_no, stdout_no, stderr_no):
1882	n/a	# open up some temporary files
1883	n/a	temps = [tempfile.mkstemp() for i in range(3)]
1884	n/a	temp_fds = [fd for fd, fname in temps]
1885	n/a	try:
1886	n/a	# unlink the files -- we won't need to reopen them
1887	n/a	for fd, fname in temps:
1888	n/a	os.unlink(fname)
1889	n/a
1890	n/a	# save a copy of the standard file descriptors
1891	n/a	saved_fds = self._save_fds(range(3))
1892	n/a	try:
1893	n/a	# duplicate the temp files over the standard fd's 0, 1, 2
1894	n/a	for fd, temp_fd in enumerate(temp_fds):
1895	n/a	os.dup2(temp_fd, fd)
1896	n/a
1897	n/a	# write some data to what will become stdin, and rewind
1898	n/a	os.write(stdin_no, b"STDIN")
1899	n/a	os.lseek(stdin_no, 0, 0)
1900	n/a
1901	n/a	# now use those files in the given order, so that subprocess
1902	n/a	# has to rearrange them in the child
1903	n/a	p = subprocess.Popen([sys.executable, "-c",
1904	n/a	'import sys; got = sys.stdin.read();'
1905	n/a	'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
1906	n/a	stdin=stdin_no,
1907	n/a	stdout=stdout_no,
1908	n/a	stderr=stderr_no)
1909	n/a	p.wait()
1910	n/a
1911	n/a	for fd in temp_fds:
1912	n/a	os.lseek(fd, 0, 0)
1913	n/a
1914	n/a	out = os.read(stdout_no, 1024)
1915	n/a	err = support.strip_python_stderr(os.read(stderr_no, 1024))
1916	n/a	finally:
1917	n/a	self._restore_fds(saved_fds)
1918	n/a
1919	n/a	self.assertEqual(out, b"got STDIN")
1920	n/a	self.assertEqual(err, b"err")
1921	n/a
1922	n/a	finally:
1923	n/a	for fd in temp_fds:
1924	n/a	os.close(fd)
1925	n/a
1926	n/a	# When duping fds, if there arises a situation where one of the fds is
1927	n/a	# either 0, 1 or 2, it is possible that it is overwritten (#12607).
1928	n/a	# This tests all combinations of this.
1929	n/a	def test_swap_fds(self):
1930	n/a	self.check_swap_fds(0, 1, 2)
1931	n/a	self.check_swap_fds(0, 2, 1)
1932	n/a	self.check_swap_fds(1, 0, 2)
1933	n/a	self.check_swap_fds(1, 2, 0)
1934	n/a	self.check_swap_fds(2, 0, 1)
1935	n/a	self.check_swap_fds(2, 1, 0)
1936	n/a
1937	n/a	def test_surrogates_error_message(self):
1938	n/a	def prepare():
1939	n/a	raise ValueError("surrogate:\uDCff")
1940	n/a
1941	n/a	try:
1942	n/a	subprocess.call(
1943	n/a	[sys.executable, "-c", "pass"],
1944	n/a	preexec_fn=prepare)
1945	n/a	except ValueError as err:
1946	n/a	# Pure Python implementations keeps the message
1947	n/a	self.assertIsNone(subprocess._posixsubprocess)
1948	n/a	self.assertEqual(str(err), "surrogate:\uDCff")
1949	n/a	except subprocess.SubprocessError as err:
1950	n/a	# _posixsubprocess uses a default message
1951	n/a	self.assertIsNotNone(subprocess._posixsubprocess)
1952	n/a	self.assertEqual(str(err), "Exception occurred in preexec_fn.")
1953	n/a	else:
1954	n/a	self.fail("Expected ValueError or subprocess.SubprocessError")
1955	n/a
1956	n/a	def test_undecodable_env(self):
1957	n/a	for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')):
1958	n/a	encoded_value = value.encode("ascii", "surrogateescape")
1959	n/a
1960	n/a	# test str with surrogates
1961	n/a	script = "import os; print(ascii(os.getenv(%s)))" % repr(key)
1962	n/a	env = os.environ.copy()
1963	n/a	env[key] = value
1964	n/a	# Use C locale to get ASCII for the locale encoding to force
1965	n/a	# surrogate-escaping of \xFF in the child process; otherwise it can
1966	n/a	# be decoded as-is if the default locale is latin-1.
1967	n/a	env['LC_ALL'] = 'C'
1968	n/a	if sys.platform.startswith("aix"):
1969	n/a	# On AIX, the C locale uses the Latin1 encoding
1970	n/a	decoded_value = encoded_value.decode("latin1", "surrogateescape")
1971	n/a	else:
1972	n/a	# On other UNIXes, the C locale uses the ASCII encoding
1973	n/a	decoded_value = value
1974	n/a	stdout = subprocess.check_output(
1975	n/a	[sys.executable, "-c", script],
1976	n/a	env=env)
1977	n/a	stdout = stdout.rstrip(b'\n\r')
1978	n/a	self.assertEqual(stdout.decode('ascii'), ascii(decoded_value))
1979	n/a
1980	n/a	# test bytes
1981	n/a	key = key.encode("ascii", "surrogateescape")
1982	n/a	script = "import os; print(ascii(os.getenvb(%s)))" % repr(key)
1983	n/a	env = os.environ.copy()
1984	n/a	env[key] = encoded_value
1985	n/a	stdout = subprocess.check_output(
1986	n/a	[sys.executable, "-c", script],
1987	n/a	env=env)
1988	n/a	stdout = stdout.rstrip(b'\n\r')
1989	n/a	self.assertEqual(stdout.decode('ascii'), ascii(encoded_value))
1990	n/a
1991	n/a	def test_bytes_program(self):
1992	n/a	abs_program = os.fsencode(sys.executable)
1993	n/a	path, program = os.path.split(sys.executable)
1994	n/a	program = os.fsencode(program)
1995	n/a
1996	n/a	# absolute bytes path
1997	n/a	exitcode = subprocess.call([abs_program, "-c", "pass"])
1998	n/a	self.assertEqual(exitcode, 0)
1999	n/a
2000	n/a	# absolute bytes path as a string
2001	n/a	cmd = b"'" + abs_program + b"' -c pass"
2002	n/a	exitcode = subprocess.call(cmd, shell=True)
2003	n/a	self.assertEqual(exitcode, 0)
2004	n/a
2005	n/a	# bytes program, unicode PATH
2006	n/a	env = os.environ.copy()
2007	n/a	env["PATH"] = path
2008	n/a	exitcode = subprocess.call([program, "-c", "pass"], env=env)
2009	n/a	self.assertEqual(exitcode, 0)
2010	n/a
2011	n/a	# bytes program, bytes PATH
2012	n/a	envb = os.environb.copy()
2013	n/a	envb[b"PATH"] = os.fsencode(path)
2014	n/a	exitcode = subprocess.call([program, "-c", "pass"], env=envb)
2015	n/a	self.assertEqual(exitcode, 0)
2016	n/a
2017	n/a	def test_pipe_cloexec(self):
2018	n/a	sleeper = support.findfile("input_reader.py", subdir="subprocessdata")
2019	n/a	fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
2020	n/a
2021	n/a	p1 = subprocess.Popen([sys.executable, sleeper],
2022	n/a	stdin=subprocess.PIPE, stdout=subprocess.PIPE,
2023	n/a	stderr=subprocess.PIPE, close_fds=False)
2024	n/a
2025	n/a	self.addCleanup(p1.communicate, b'')
2026	n/a
2027	n/a	p2 = subprocess.Popen([sys.executable, fd_status],
2028	n/a	stdout=subprocess.PIPE, close_fds=False)
2029	n/a
2030	n/a	output, error = p2.communicate()
2031	n/a	result_fds = set(map(int, output.split(b',')))
2032	n/a	unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(),
2033	n/a	p1.stderr.fileno()])
2034	n/a
2035	n/a	self.assertFalse(result_fds & unwanted_fds,
2036	n/a	"Expected no fds from %r to be open in child, "
2037	n/a	"found %r" %
2038	n/a	(unwanted_fds, result_fds & unwanted_fds))
2039	n/a
2040	n/a	def test_pipe_cloexec_real_tools(self):
2041	n/a	qcat = support.findfile("qcat.py", subdir="subprocessdata")
2042	n/a	qgrep = support.findfile("qgrep.py", subdir="subprocessdata")
2043	n/a
2044	n/a	subdata = b'zxcvbn'
2045	n/a	data = subdata * 4 + b'\n'
2046	n/a
2047	n/a	p1 = subprocess.Popen([sys.executable, qcat],
2048	n/a	stdin=subprocess.PIPE, stdout=subprocess.PIPE,
2049	n/a	close_fds=False)
2050	n/a
2051	n/a	p2 = subprocess.Popen([sys.executable, qgrep, subdata],
2052	n/a	stdin=p1.stdout, stdout=subprocess.PIPE,
2053	n/a	close_fds=False)
2054	n/a
2055	n/a	self.addCleanup(p1.wait)
2056	n/a	self.addCleanup(p2.wait)
2057	n/a	def kill_p1():
2058	n/a	try:
2059	n/a	p1.terminate()
2060	n/a	except ProcessLookupError:
2061	n/a	pass
2062	n/a	def kill_p2():
2063	n/a	try:
2064	n/a	p2.terminate()
2065	n/a	except ProcessLookupError:
2066	n/a	pass
2067	n/a	self.addCleanup(kill_p1)
2068	n/a	self.addCleanup(kill_p2)
2069	n/a
2070	n/a	p1.stdin.write(data)
2071	n/a	p1.stdin.close()
2072	n/a
2073	n/a	readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10)
2074	n/a
2075	n/a	self.assertTrue(readfiles, "The child hung")
2076	n/a	self.assertEqual(p2.stdout.read(), data)
2077	n/a
2078	n/a	p1.stdout.close()
2079	n/a	p2.stdout.close()
2080	n/a
2081	n/a	def test_close_fds(self):
2082	n/a	fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
2083	n/a
2084	n/a	fds = os.pipe()
2085	n/a	self.addCleanup(os.close, fds[0])
2086	n/a	self.addCleanup(os.close, fds[1])
2087	n/a
2088	n/a	open_fds = set(fds)
2089	n/a	# add a bunch more fds
2090	n/a	for _ in range(9):
2091	n/a	fd = os.open(os.devnull, os.O_RDONLY)
2092	n/a	self.addCleanup(os.close, fd)
2093	n/a	open_fds.add(fd)
2094	n/a
2095	n/a	for fd in open_fds:
2096	n/a	os.set_inheritable(fd, True)
2097	n/a
2098	n/a	p = subprocess.Popen([sys.executable, fd_status],
2099	n/a	stdout=subprocess.PIPE, close_fds=False)
2100	n/a	output, ignored = p.communicate()
2101	n/a	remaining_fds = set(map(int, output.split(b',')))
2102	n/a
2103	n/a	self.assertEqual(remaining_fds & open_fds, open_fds,
2104	n/a	"Some fds were closed")
2105	n/a
2106	n/a	p = subprocess.Popen([sys.executable, fd_status],
2107	n/a	stdout=subprocess.PIPE, close_fds=True)
2108	n/a	output, ignored = p.communicate()
2109	n/a	remaining_fds = set(map(int, output.split(b',')))
2110	n/a
2111	n/a	self.assertFalse(remaining_fds & open_fds,
2112	n/a	"Some fds were left open")
2113	n/a	self.assertIn(1, remaining_fds, "Subprocess failed")
2114	n/a
2115	n/a	# Keep some of the fd's we opened open in the subprocess.
2116	n/a	# This tests _posixsubprocess.c's proper handling of fds_to_keep.
2117	n/a	fds_to_keep = set(open_fds.pop() for _ in range(8))
2118	n/a	p = subprocess.Popen([sys.executable, fd_status],
2119	n/a	stdout=subprocess.PIPE, close_fds=True,
2120	n/a	pass_fds=())
2121	n/a	output, ignored = p.communicate()
2122	n/a	remaining_fds = set(map(int, output.split(b',')))
2123	n/a
2124	n/a	self.assertFalse(remaining_fds & fds_to_keep & open_fds,
2125	n/a	"Some fds not in pass_fds were left open")
2126	n/a	self.assertIn(1, remaining_fds, "Subprocess failed")
2127	n/a
2128	n/a
2129	n/a	@unittest.skipIf(sys.platform.startswith("freebsd") and
2130	n/a	os.stat("/dev").st_dev == os.stat("/dev/fd").st_dev,
2131	n/a	"Requires fdescfs mounted on /dev/fd on FreeBSD.")
2132	n/a	def test_close_fds_when_max_fd_is_lowered(self):
2133	n/a	"""Confirm that issue21618 is fixed (may fail under valgrind)."""
2134	n/a	fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
2135	n/a
2136	n/a	# This launches the meat of the test in a child process to
2137	n/a	# avoid messing with the larger unittest processes maximum
2138	n/a	# number of file descriptors.
2139	n/a	# This process launches:
2140	n/a	# +--> Process that lowers its RLIMIT_NOFILE aftr setting up
2141	n/a	# a bunch of high open fds above the new lower rlimit.
2142	n/a	# Those are reported via stdout before launching a new
2143	n/a	# process with close_fds=False to run the actual test:
2144	n/a	# +--> The TEST: This one launches a fd_status.py
2145	n/a	# subprocess with close_fds=True so we can find out if
2146	n/a	# any of the fds above the lowered rlimit are still open.
2147	n/a	p = subprocess.Popen([sys.executable, '-c', textwrap.dedent(
2148	n/a	'''
2149	n/a	import os, resource, subprocess, sys, textwrap
2150	n/a	open_fds = set()
2151	n/a	# Add a bunch more fds to pass down.
2152	n/a	for _ in range(40):
2153	n/a	fd = os.open(os.devnull, os.O_RDONLY)
2154	n/a	open_fds.add(fd)
2155	n/a
2156	n/a	# Leave a two pairs of low ones available for use by the
2157	n/a	# internal child error pipe and the stdout pipe.
2158	n/a	# We also leave 10 more open as some Python buildbots run into
2159	n/a	# "too many open files" errors during the test if we do not.
2160	n/a	for fd in sorted(open_fds)[:14]:
2161	n/a	os.close(fd)
2162	n/a	open_fds.remove(fd)
2163	n/a
2164	n/a	for fd in open_fds:
2165	n/a	#self.addCleanup(os.close, fd)
2166	n/a	os.set_inheritable(fd, True)
2167	n/a
2168	n/a	max_fd_open = max(open_fds)
2169	n/a
2170	n/a	# Communicate the open_fds to the parent unittest.TestCase process.
2171	n/a	print(','.join(map(str, sorted(open_fds))))
2172	n/a	sys.stdout.flush()
2173	n/a
2174	n/a	rlim_cur, rlim_max = resource.getrlimit(resource.RLIMIT_NOFILE)
2175	n/a	try:
2176	n/a	# 29 is lower than the highest fds we are leaving open.
2177	n/a	resource.setrlimit(resource.RLIMIT_NOFILE, (29, rlim_max))
2178	n/a	# Launch a new Python interpreter with our low fd rlim_cur that
2179	n/a	# inherits open fds above that limit. It then uses subprocess
2180	n/a	# with close_fds=True to get a report of open fds in the child.
2181	n/a	# An explicit list of fds to check is passed to fd_status.py as
2182	n/a	# letting fd_status rely on its default logic would miss the
2183	n/a	# fds above rlim_cur as it normally only checks up to that limit.
2184	n/a	subprocess.Popen(
2185	n/a	[sys.executable, '-c',
2186	n/a	textwrap.dedent("""
2187	n/a	import subprocess, sys
2188	n/a	subprocess.Popen([sys.executable, %r] +
2189	n/a	[str(x) for x in range({max_fd})],
2190	n/a	close_fds=True).wait()
2191	n/a	""".format(max_fd=max_fd_open+1))],
2192	n/a	close_fds=False).wait()
2193	n/a	finally:
2194	n/a	resource.setrlimit(resource.RLIMIT_NOFILE, (rlim_cur, rlim_max))
2195	n/a	''' % fd_status)], stdout=subprocess.PIPE)
2196	n/a
2197	n/a	output, unused_stderr = p.communicate()
2198	n/a	output_lines = output.splitlines()
2199	n/a	self.assertEqual(len(output_lines), 2,
2200	n/a	msg="expected exactly two lines of output:\n%r" % output)
2201	n/a	opened_fds = set(map(int, output_lines[0].strip().split(b',')))
2202	n/a	remaining_fds = set(map(int, output_lines[1].strip().split(b',')))
2203	n/a
2204	n/a	self.assertFalse(remaining_fds & opened_fds,
2205	n/a	msg="Some fds were left open.")
2206	n/a
2207	n/a
2208	n/a	# Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file
2209	n/a	# descriptor of a pipe closed in the parent process is valid in the
2210	n/a	# child process according to fstat(), but the mode of the file
2211	n/a	# descriptor is invalid, and read or write raise an error.
2212	n/a	@support.requires_mac_ver(10, 5)
2213	n/a	def test_pass_fds(self):
2214	n/a	fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
2215	n/a
2216	n/a	open_fds = set()
2217	n/a
2218	n/a	for x in range(5):
2219	n/a	fds = os.pipe()
2220	n/a	self.addCleanup(os.close, fds[0])
2221	n/a	self.addCleanup(os.close, fds[1])
2222	n/a	os.set_inheritable(fds[0], True)
2223	n/a	os.set_inheritable(fds[1], True)
2224	n/a	open_fds.update(fds)
2225	n/a
2226	n/a	for fd in open_fds:
2227	n/a	p = subprocess.Popen([sys.executable, fd_status],
2228	n/a	stdout=subprocess.PIPE, close_fds=True,
2229	n/a	pass_fds=(fd, ))
2230	n/a	output, ignored = p.communicate()
2231	n/a
2232	n/a	remaining_fds = set(map(int, output.split(b',')))
2233	n/a	to_be_closed = open_fds - {fd}
2234	n/a
2235	n/a	self.assertIn(fd, remaining_fds, "fd to be passed not passed")
2236	n/a	self.assertFalse(remaining_fds & to_be_closed,
2237	n/a	"fd to be closed passed")
2238	n/a
2239	n/a	# pass_fds overrides close_fds with a warning.
2240	n/a	with self.assertWarns(RuntimeWarning) as context:
2241	n/a	self.assertFalse(subprocess.call(
2242	n/a	[sys.executable, "-c", "import sys; sys.exit(0)"],
2243	n/a	close_fds=False, pass_fds=(fd, )))
2244	n/a	self.assertIn('overriding close_fds', str(context.warning))
2245	n/a
2246	n/a	def test_pass_fds_inheritable(self):
2247	n/a	script = support.findfile("fd_status.py", subdir="subprocessdata")
2248	n/a
2249	n/a	inheritable, non_inheritable = os.pipe()
2250	n/a	self.addCleanup(os.close, inheritable)
2251	n/a	self.addCleanup(os.close, non_inheritable)
2252	n/a	os.set_inheritable(inheritable, True)
2253	n/a	os.set_inheritable(non_inheritable, False)
2254	n/a	pass_fds = (inheritable, non_inheritable)
2255	n/a	args = [sys.executable, script]
2256	n/a	args += list(map(str, pass_fds))
2257	n/a
2258	n/a	p = subprocess.Popen(args,
2259	n/a	stdout=subprocess.PIPE, close_fds=True,
2260	n/a	pass_fds=pass_fds)
2261	n/a	output, ignored = p.communicate()
2262	n/a	fds = set(map(int, output.split(b',')))
2263	n/a
2264	n/a	# the inheritable file descriptor must be inherited, so its inheritable
2265	n/a	# flag must be set in the child process after fork() and before exec()
2266	n/a	self.assertEqual(fds, set(pass_fds), "output=%a" % output)
2267	n/a
2268	n/a	# inheritable flag must not be changed in the parent process
2269	n/a	self.assertEqual(os.get_inheritable(inheritable), True)
2270	n/a	self.assertEqual(os.get_inheritable(non_inheritable), False)
2271	n/a
2272	n/a	def test_stdout_stdin_are_single_inout_fd(self):
2273	n/a	with io.open(os.devnull, "r+") as inout:
2274	n/a	p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
2275	n/a	stdout=inout, stdin=inout)
2276	n/a	p.wait()
2277	n/a
2278	n/a	def test_stdout_stderr_are_single_inout_fd(self):
2279	n/a	with io.open(os.devnull, "r+") as inout:
2280	n/a	p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
2281	n/a	stdout=inout, stderr=inout)
2282	n/a	p.wait()
2283	n/a
2284	n/a	def test_stderr_stdin_are_single_inout_fd(self):
2285	n/a	with io.open(os.devnull, "r+") as inout:
2286	n/a	p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],
2287	n/a	stderr=inout, stdin=inout)
2288	n/a	p.wait()
2289	n/a
2290	n/a	def test_wait_when_sigchild_ignored(self):
2291	n/a	# NOTE: sigchild_ignore.py may not be an effective test on all OSes.
2292	n/a	sigchild_ignore = support.findfile("sigchild_ignore.py",
2293	n/a	subdir="subprocessdata")
2294	n/a	p = subprocess.Popen([sys.executable, sigchild_ignore],
2295	n/a	stdout=subprocess.PIPE, stderr=subprocess.PIPE)
2296	n/a	stdout, stderr = p.communicate()
2297	n/a	self.assertEqual(0, p.returncode, "sigchild_ignore.py exited"
2298	n/a	" non-zero with this error:\n%s" %
2299	n/a	stderr.decode('utf-8'))
2300	n/a
2301	n/a	def test_select_unbuffered(self):
2302	n/a	# Issue #11459: bufsize=0 should really set the pipes as
2303	n/a	# unbuffered (and therefore let select() work properly).
2304	n/a	select = support.import_module("select")
2305	n/a	p = subprocess.Popen([sys.executable, "-c",
2306	n/a	'import sys;'
2307	n/a	'sys.stdout.write("apple")'],
2308	n/a	stdout=subprocess.PIPE,
2309	n/a	bufsize=0)
2310	n/a	f = p.stdout
2311	n/a	self.addCleanup(f.close)
2312	n/a	try:
2313	n/a	self.assertEqual(f.read(4), b"appl")
2314	n/a	self.assertIn(f, select.select([f], [], [], 0.0)[0])
2315	n/a	finally:
2316	n/a	p.wait()
2317	n/a
2318	n/a	def test_zombie_fast_process_del(self):
2319	n/a	# Issue #12650: on Unix, if Popen.__del__() was called before the
2320	n/a	# process exited, it wouldn't be added to subprocess._active, and would
2321	n/a	# remain a zombie.
2322	n/a	# spawn a Popen, and delete its reference before it exits
2323	n/a	p = subprocess.Popen([sys.executable, "-c",
2324	n/a	'import sys, time;'
2325	n/a	'time.sleep(0.2)'],
2326	n/a	stdout=subprocess.PIPE,
2327	n/a	stderr=subprocess.PIPE)
2328	n/a	self.addCleanup(p.stdout.close)
2329	n/a	self.addCleanup(p.stderr.close)
2330	n/a	ident = id(p)
2331	n/a	pid = p.pid
2332	n/a	with support.check_warnings(('', ResourceWarning)):
2333	n/a	p = None
2334	n/a
2335	n/a	# check that p is in the active processes list
2336	n/a	self.assertIn(ident, [id(o) for o in subprocess._active])
2337	n/a
2338	n/a	def test_leak_fast_process_del_killed(self):
2339	n/a	# Issue #12650: on Unix, if Popen.__del__() was called before the
2340	n/a	# process exited, and the process got killed by a signal, it would never
2341	n/a	# be removed from subprocess._active, which triggered a FD and memory
2342	n/a	# leak.
2343	n/a	# spawn a Popen, delete its reference and kill it
2344	n/a	p = subprocess.Popen([sys.executable, "-c",
2345	n/a	'import time;'
2346	n/a	'time.sleep(3)'],
2347	n/a	stdout=subprocess.PIPE,
2348	n/a	stderr=subprocess.PIPE)
2349	n/a	self.addCleanup(p.stdout.close)
2350	n/a	self.addCleanup(p.stderr.close)
2351	n/a	ident = id(p)
2352	n/a	pid = p.pid
2353	n/a	with support.check_warnings(('', ResourceWarning)):
2354	n/a	p = None
2355	n/a
2356	n/a	os.kill(pid, signal.SIGKILL)
2357	n/a	# check that p is in the active processes list
2358	n/a	self.assertIn(ident, [id(o) for o in subprocess._active])
2359	n/a
2360	n/a	# let some time for the process to exit, and create a new Popen: this
2361	n/a	# should trigger the wait() of p
2362	n/a	time.sleep(0.2)
2363	n/a	with self.assertRaises(OSError) as c:
2364	n/a	with subprocess.Popen(['nonexisting_i_hope'],
2365	n/a	stdout=subprocess.PIPE,
2366	n/a	stderr=subprocess.PIPE) as proc:
2367	n/a	pass
2368	n/a	# p should have been wait()ed on, and removed from the _active list
2369	n/a	self.assertRaises(OSError, os.waitpid, pid, 0)
2370	n/a	self.assertNotIn(ident, [id(o) for o in subprocess._active])
2371	n/a
2372	n/a	def test_close_fds_after_preexec(self):
2373	n/a	fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
2374	n/a
2375	n/a	# this FD is used as dup2() target by preexec_fn, and should be closed
2376	n/a	# in the child process
2377	n/a	fd = os.dup(1)
2378	n/a	self.addCleanup(os.close, fd)
2379	n/a
2380	n/a	p = subprocess.Popen([sys.executable, fd_status],
2381	n/a	stdout=subprocess.PIPE, close_fds=True,
2382	n/a	preexec_fn=lambda: os.dup2(1, fd))
2383	n/a	output, ignored = p.communicate()
2384	n/a
2385	n/a	remaining_fds = set(map(int, output.split(b',')))
2386	n/a
2387	n/a	self.assertNotIn(fd, remaining_fds)
2388	n/a
2389	n/a	@support.cpython_only
2390	n/a	def test_fork_exec(self):
2391	n/a	# Issue #22290: fork_exec() must not crash on memory allocation failure
2392	n/a	# or other errors
2393	n/a	import _posixsubprocess
2394	n/a	gc_enabled = gc.isenabled()
2395	n/a	try:
2396	n/a	# Use a preexec function and enable the garbage collector
2397	n/a	# to force fork_exec() to re-enable the garbage collector
2398	n/a	# on error.
2399	n/a	func = lambda: None
2400	n/a	gc.enable()
2401	n/a
2402	n/a	for args, exe_list, cwd, env_list in (
2403	n/a	(123, [b"exe"], None, [b"env"]),
2404	n/a	([b"arg"], 123, None, [b"env"]),
2405	n/a	([b"arg"], [b"exe"], 123, [b"env"]),
2406	n/a	([b"arg"], [b"exe"], None, 123),
2407	n/a	):
2408	n/a	with self.assertRaises(TypeError):
2409	n/a	_posixsubprocess.fork_exec(
2410	n/a	args, exe_list,
2411	n/a	True, [], cwd, env_list,
2412	n/a	-1, -1, -1, -1,
2413	n/a	1, 2, 3, 4,
2414	n/a	True, True, func)
2415	n/a	finally:
2416	n/a	if not gc_enabled:
2417	n/a	gc.disable()
2418	n/a
2419	n/a	@support.cpython_only
2420	n/a	def test_fork_exec_sorted_fd_sanity_check(self):
2421	n/a	# Issue #23564: sanity check the fork_exec() fds_to_keep sanity check.
2422	n/a	import _posixsubprocess
2423	n/a	gc_enabled = gc.isenabled()
2424	n/a	try:
2425	n/a	gc.enable()
2426	n/a
2427	n/a	for fds_to_keep in (
2428	n/a	(-1, 2, 3, 4, 5), # Negative number.
2429	n/a	('str', 4), # Not an int.
2430	n/a	(18, 23, 42, 2**63), # Out of range.
2431	n/a	(5, 4), # Not sorted.
2432	n/a	(6, 7, 7, 8), # Duplicate.
2433	n/a	):
2434	n/a	with self.assertRaises(
2435	n/a	ValueError,
2436	n/a	msg='fds_to_keep={}'.format(fds_to_keep)) as c:
2437	n/a	_posixsubprocess.fork_exec(
2438	n/a	[b"false"], [b"false"],
2439	n/a	True, fds_to_keep, None, [b"env"],
2440	n/a	-1, -1, -1, -1,
2441	n/a	1, 2, 3, 4,
2442	n/a	True, True, None)
2443	n/a	self.assertIn('fds_to_keep', str(c.exception))
2444	n/a	finally:
2445	n/a	if not gc_enabled:
2446	n/a	gc.disable()
2447	n/a
2448	n/a	def test_communicate_BrokenPipeError_stdin_close(self):
2449	n/a	# By not setting stdout or stderr or a timeout we force the fast path
2450	n/a	# that just calls _stdin_write() internally due to our mock.
2451	n/a	proc = subprocess.Popen([sys.executable, '-c', 'pass'])
2452	n/a	with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin:
2453	n/a	mock_proc_stdin.close.side_effect = BrokenPipeError
2454	n/a	proc.communicate() # Should swallow BrokenPipeError from close.
2455	n/a	mock_proc_stdin.close.assert_called_with()
2456	n/a
2457	n/a	def test_communicate_BrokenPipeError_stdin_write(self):
2458	n/a	# By not setting stdout or stderr or a timeout we force the fast path
2459	n/a	# that just calls _stdin_write() internally due to our mock.
2460	n/a	proc = subprocess.Popen([sys.executable, '-c', 'pass'])
2461	n/a	with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin:
2462	n/a	mock_proc_stdin.write.side_effect = BrokenPipeError
2463	n/a	proc.communicate(b'stuff') # Should swallow the BrokenPipeError.
2464	n/a	mock_proc_stdin.write.assert_called_once_with(b'stuff')
2465	n/a	mock_proc_stdin.close.assert_called_once_with()
2466	n/a
2467	n/a	def test_communicate_BrokenPipeError_stdin_flush(self):
2468	n/a	# Setting stdin and stdout forces the ._communicate() code path.
2469	n/a	# python -h exits faster than python -c pass (but spams stdout).
2470	n/a	proc = subprocess.Popen([sys.executable, '-h'],
2471	n/a	stdin=subprocess.PIPE,
2472	n/a	stdout=subprocess.PIPE)
2473	n/a	with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin, \
2474	n/a	open(os.devnull, 'wb') as dev_null:
2475	n/a	mock_proc_stdin.flush.side_effect = BrokenPipeError
2476	n/a	# because _communicate registers a selector using proc.stdin...
2477	n/a	mock_proc_stdin.fileno.return_value = dev_null.fileno()
2478	n/a	# _communicate() should swallow BrokenPipeError from flush.
2479	n/a	proc.communicate(b'stuff')
2480	n/a	mock_proc_stdin.flush.assert_called_once_with()
2481	n/a
2482	n/a	def test_communicate_BrokenPipeError_stdin_close_with_timeout(self):
2483	n/a	# Setting stdin and stdout forces the ._communicate() code path.
2484	n/a	# python -h exits faster than python -c pass (but spams stdout).
2485	n/a	proc = subprocess.Popen([sys.executable, '-h'],
2486	n/a	stdin=subprocess.PIPE,
2487	n/a	stdout=subprocess.PIPE)
2488	n/a	with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin:
2489	n/a	mock_proc_stdin.close.side_effect = BrokenPipeError
2490	n/a	# _communicate() should swallow BrokenPipeError from close.
2491	n/a	proc.communicate(timeout=999)
2492	n/a	mock_proc_stdin.close.assert_called_once_with()
2493	n/a
2494	n/a	_libc_file_extensions = {
2495	n/a	'Linux': 'so.6',
2496	n/a	'Darwin': 'dylib',
2497	n/a	}
2498	n/a	@unittest.skipIf(not ctypes, 'ctypes module required.')
2499	n/a	@unittest.skipIf(platform.uname()[0] not in _libc_file_extensions,
2500	n/a	'Test requires a libc this code can load with ctypes.')
2501	n/a	@unittest.skipIf(not sys.executable, 'Test requires sys.executable.')
2502	n/a	def test_child_terminated_in_stopped_state(self):
2503	n/a	"""Test wait() behavior when waitpid returns WIFSTOPPED; issue29335."""
2504	n/a	PTRACE_TRACEME = 0 # From glibc and MacOS (PT_TRACE_ME).
2505	n/a	libc_name = 'libc.' + self._libc_file_extensions[platform.uname()[0]]
2506	n/a	libc = ctypes.CDLL(libc_name)
2507	n/a	if not hasattr(libc, 'ptrace'):
2508	n/a	raise unittest.SkipTest('ptrace() required.')
2509	n/a	test_ptrace = subprocess.Popen(
2510	n/a	[sys.executable, '-c', """if True:
2511	n/a	import ctypes
2512	n/a	libc = ctypes.CDLL({libc_name!r})
2513	n/a	libc.ptrace({PTRACE_TRACEME}, 0, 0)
2514	n/a	""".format(libc_name=libc_name, PTRACE_TRACEME=PTRACE_TRACEME)
2515	n/a	])
2516	n/a	if test_ptrace.wait() != 0:
2517	n/a	raise unittest.SkipTest('ptrace() failed - unable to test.')
2518	n/a	child = subprocess.Popen(
2519	n/a	[sys.executable, '-c', """if True:
2520	n/a	import ctypes
2521	n/a	libc = ctypes.CDLL({libc_name!r})
2522	n/a	libc.ptrace({PTRACE_TRACEME}, 0, 0)
2523	n/a	libc.printf(ctypes.c_char_p(0xdeadbeef)) # Crash the process.
2524	n/a	""".format(libc_name=libc_name, PTRACE_TRACEME=PTRACE_TRACEME)
2525	n/a	])
2526	n/a	try:
2527	n/a	returncode = child.wait()
2528	n/a	except Exception as e:
2529	n/a	child.kill() # Clean up the hung stopped process.
2530	n/a	raise e
2531	n/a	self.assertNotEqual(0, returncode)
2532	n/a	self.assertLess(returncode, 0) # signal death, likely SIGSEGV.
2533	n/a
2534	n/a
2535	n/a	@unittest.skipUnless(mswindows, "Windows specific tests")
2536	n/a	class Win32ProcessTestCase(BaseTestCase):
2537	n/a
2538	n/a	def test_startupinfo(self):
2539	n/a	# startupinfo argument
2540	n/a	# We uses hardcoded constants, because we do not want to
2541	n/a	# depend on win32all.
2542	n/a	STARTF_USESHOWWINDOW = 1
2543	n/a	SW_MAXIMIZE = 3
2544	n/a	startupinfo = subprocess.STARTUPINFO()
2545	n/a	startupinfo.dwFlags = STARTF_USESHOWWINDOW
2546	n/a	startupinfo.wShowWindow = SW_MAXIMIZE
2547	n/a	# Since Python is a console process, it won't be affected
2548	n/a	# by wShowWindow, but the argument should be silently
2549	n/a	# ignored
2550	n/a	subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
2551	n/a	startupinfo=startupinfo)
2552	n/a
2553	n/a	def test_creationflags(self):
2554	n/a	# creationflags argument
2555	n/a	CREATE_NEW_CONSOLE = 16
2556	n/a	sys.stderr.write(" a DOS box should flash briefly ...\n")
2557	n/a	subprocess.call(sys.executable +
2558	n/a	' -c "import time; time.sleep(0.25)"',
2559	n/a	creationflags=CREATE_NEW_CONSOLE)
2560	n/a
2561	n/a	def test_invalid_args(self):
2562	n/a	# invalid arguments should raise ValueError
2563	n/a	self.assertRaises(ValueError, subprocess.call,
2564	n/a	[sys.executable, "-c",
2565	n/a	"import sys; sys.exit(47)"],
2566	n/a	preexec_fn=lambda: 1)
2567	n/a	self.assertRaises(ValueError, subprocess.call,
2568	n/a	[sys.executable, "-c",
2569	n/a	"import sys; sys.exit(47)"],
2570	n/a	stdout=subprocess.PIPE,
2571	n/a	close_fds=True)
2572	n/a
2573	n/a	def test_close_fds(self):
2574	n/a	# close file descriptors
2575	n/a	rc = subprocess.call([sys.executable, "-c",
2576	n/a	"import sys; sys.exit(47)"],
2577	n/a	close_fds=True)
2578	n/a	self.assertEqual(rc, 47)
2579	n/a
2580	n/a	def test_shell_sequence(self):
2581	n/a	# Run command through the shell (sequence)
2582	n/a	newenv = os.environ.copy()
2583	n/a	newenv["FRUIT"] = "physalis"
2584	n/a	p = subprocess.Popen(["set"], shell=1,
2585	n/a	stdout=subprocess.PIPE,
2586	n/a	env=newenv)
2587	n/a	with p:
2588	n/a	self.assertIn(b"physalis", p.stdout.read())
2589	n/a
2590	n/a	def test_shell_string(self):
2591	n/a	# Run command through the shell (string)
2592	n/a	newenv = os.environ.copy()
2593	n/a	newenv["FRUIT"] = "physalis"
2594	n/a	p = subprocess.Popen("set", shell=1,
2595	n/a	stdout=subprocess.PIPE,
2596	n/a	env=newenv)
2597	n/a	with p:
2598	n/a	self.assertIn(b"physalis", p.stdout.read())
2599	n/a
2600	n/a	def test_shell_encodings(self):
2601	n/a	# Run command through the shell (string)
2602	n/a	for enc in ['ansi', 'oem']:
2603	n/a	newenv = os.environ.copy()
2604	n/a	newenv["FRUIT"] = "physalis"
2605	n/a	p = subprocess.Popen("set", shell=1,
2606	n/a	stdout=subprocess.PIPE,
2607	n/a	env=newenv,
2608	n/a	encoding=enc)
2609	n/a	with p:
2610	n/a	self.assertIn("physalis", p.stdout.read(), enc)
2611	n/a
2612	n/a	def test_call_string(self):
2613	n/a	# call() function with string argument on Windows
2614	n/a	rc = subprocess.call(sys.executable +
2615	n/a	' -c "import sys; sys.exit(47)"')
2616	n/a	self.assertEqual(rc, 47)
2617	n/a
2618	n/a	def _kill_process(self, method, *args):
2619	n/a	# Some win32 buildbot raises EOFError if stdin is inherited
2620	n/a	p = subprocess.Popen([sys.executable, "-c", """if 1:
2621	n/a	import sys, time
2622	n/a	sys.stdout.write('x\\n')
2623	n/a	sys.stdout.flush()
2624	n/a	time.sleep(30)
2625	n/a	"""],
2626	n/a	stdin=subprocess.PIPE,
2627	n/a	stdout=subprocess.PIPE,
2628	n/a	stderr=subprocess.PIPE)
2629	n/a	with p:
2630	n/a	# Wait for the interpreter to be completely initialized before
2631	n/a	# sending any signal.
2632	n/a	p.stdout.read(1)
2633	n/a	getattr(p, method)(*args)
2634	n/a	_, stderr = p.communicate()
2635	n/a	self.assertStderrEqual(stderr, b'')
2636	n/a	returncode = p.wait()
2637	n/a	self.assertNotEqual(returncode, 0)
2638	n/a
2639	n/a	def _kill_dead_process(self, method, *args):
2640	n/a	p = subprocess.Popen([sys.executable, "-c", """if 1:
2641	n/a	import sys, time
2642	n/a	sys.stdout.write('x\\n')
2643	n/a	sys.stdout.flush()
2644	n/a	sys.exit(42)
2645	n/a	"""],
2646	n/a	stdin=subprocess.PIPE,
2647	n/a	stdout=subprocess.PIPE,
2648	n/a	stderr=subprocess.PIPE)
2649	n/a	with p:
2650	n/a	# Wait for the interpreter to be completely initialized before
2651	n/a	# sending any signal.
2652	n/a	p.stdout.read(1)
2653	n/a	# The process should end after this
2654	n/a	time.sleep(1)
2655	n/a	# This shouldn't raise even though the child is now dead
2656	n/a	getattr(p, method)(*args)
2657	n/a	_, stderr = p.communicate()
2658	n/a	self.assertStderrEqual(stderr, b'')
2659	n/a	rc = p.wait()
2660	n/a	self.assertEqual(rc, 42)
2661	n/a
2662	n/a	def test_send_signal(self):
2663	n/a	self._kill_process('send_signal', signal.SIGTERM)
2664	n/a
2665	n/a	def test_kill(self):
2666	n/a	self._kill_process('kill')
2667	n/a
2668	n/a	def test_terminate(self):
2669	n/a	self._kill_process('terminate')
2670	n/a
2671	n/a	def test_send_signal_dead(self):
2672	n/a	self._kill_dead_process('send_signal', signal.SIGTERM)
2673	n/a
2674	n/a	def test_kill_dead(self):
2675	n/a	self._kill_dead_process('kill')
2676	n/a
2677	n/a	def test_terminate_dead(self):
2678	n/a	self._kill_dead_process('terminate')
2679	n/a
2680	n/a	class MiscTests(unittest.TestCase):
2681	n/a	def test_getoutput(self):
2682	n/a	self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy')
2683	n/a	self.assertEqual(subprocess.getstatusoutput('echo xyzzy'),
2684	n/a	(0, 'xyzzy'))
2685	n/a
2686	n/a	# we use mkdtemp in the next line to create an empty directory
2687	n/a	# under our exclusive control; from that, we can invent a pathname
2688	n/a	# that we _know_ won't exist. This is guaranteed to fail.
2689	n/a	dir = None
2690	n/a	try:
2691	n/a	dir = tempfile.mkdtemp()
2692	n/a	name = os.path.join(dir, "foo")
2693	n/a	status, output = subprocess.getstatusoutput(
2694	n/a	("type " if mswindows else "cat ") + name)
2695	n/a	self.assertNotEqual(status, 0)
2696	n/a	finally:
2697	n/a	if dir is not None:
2698	n/a	os.rmdir(dir)
2699	n/a
2700	n/a	def test__all__(self):
2701	n/a	"""Ensure that __all__ is populated properly."""
2702	n/a	intentionally_excluded = {"list2cmdline", "Handle"}
2703	n/a	exported = set(subprocess.__all__)
2704	n/a	possible_exports = set()
2705	n/a	import types
2706	n/a	for name, value in subprocess.__dict__.items():
2707	n/a	if name.startswith('_'):
2708	n/a	continue
2709	n/a	if isinstance(value, (types.ModuleType,)):
2710	n/a	continue
2711	n/a	possible_exports.add(name)
2712	n/a	self.assertEqual(exported, possible_exports - intentionally_excluded)
2713	n/a
2714	n/a
2715	n/a	@unittest.skipUnless(hasattr(selectors, 'PollSelector'),
2716	n/a	"Test needs selectors.PollSelector")
2717	n/a	class ProcessTestCaseNoPoll(ProcessTestCase):
2718	n/a	def setUp(self):
2719	n/a	self.orig_selector = subprocess._PopenSelector
2720	n/a	subprocess._PopenSelector = selectors.SelectSelector
2721	n/a	ProcessTestCase.setUp(self)
2722	n/a
2723	n/a	def tearDown(self):
2724	n/a	subprocess._PopenSelector = self.orig_selector
2725	n/a	ProcessTestCase.tearDown(self)
2726	n/a
2727	n/a
2728	n/a	@unittest.skipUnless(mswindows, "Windows-specific tests")
2729	n/a	class CommandsWithSpaces (BaseTestCase):
2730	n/a
2731	n/a	def setUp(self):
2732	n/a	super().setUp()
2733	n/a	f, fname = tempfile.mkstemp(".py", "te st")
2734	n/a	self.fname = fname.lower ()
2735	n/a	os.write(f, b"import sys;"
2736	n/a	b"sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))"
2737	n/a	)
2738	n/a	os.close(f)
2739	n/a
2740	n/a	def tearDown(self):
2741	n/a	os.remove(self.fname)
2742	n/a	super().tearDown()
2743	n/a
2744	n/a	def with_spaces(self, args, *kwargs):
2745	n/a	kwargs['stdout'] = subprocess.PIPE
2746	n/a	p = subprocess.Popen(args, *kwargs)
2747	n/a	with p:
2748	n/a	self.assertEqual(
2749	n/a	p.stdout.read ().decode("mbcs"),
2750	n/a	"2 [%r, 'ab cd']" % self.fname
2751	n/a	)
2752	n/a
2753	n/a	def test_shell_string_with_spaces(self):
2754	n/a	# call() function with string argument with spaces on Windows
2755	n/a	self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
2756	n/a	"ab cd"), shell=1)
2757	n/a
2758	n/a	def test_shell_sequence_with_spaces(self):
2759	n/a	# call() function with sequence argument with spaces on Windows
2760	n/a	self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1)
2761	n/a
2762	n/a	def test_noshell_string_with_spaces(self):
2763	n/a	# call() function with string argument with spaces on Windows
2764	n/a	self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
2765	n/a	"ab cd"))
2766	n/a
2767	n/a	def test_noshell_sequence_with_spaces(self):
2768	n/a	# call() function with sequence argument with spaces on Windows
2769	n/a	self.with_spaces([sys.executable, self.fname, "ab cd"])
2770	n/a
2771	n/a
2772	n/a	class ContextManagerTests(BaseTestCase):
2773	n/a
2774	n/a	def test_pipe(self):
2775	n/a	with subprocess.Popen([sys.executable, "-c",
2776	n/a	"import sys;"
2777	n/a	"sys.stdout.write('stdout');"
2778	n/a	"sys.stderr.write('stderr');"],
2779	n/a	stdout=subprocess.PIPE,
2780	n/a	stderr=subprocess.PIPE) as proc:
2781	n/a	self.assertEqual(proc.stdout.read(), b"stdout")
2782	n/a	self.assertStderrEqual(proc.stderr.read(), b"stderr")
2783	n/a
2784	n/a	self.assertTrue(proc.stdout.closed)
2785	n/a	self.assertTrue(proc.stderr.closed)
2786	n/a
2787	n/a	def test_returncode(self):
2788	n/a	with subprocess.Popen([sys.executable, "-c",
2789	n/a	"import sys; sys.exit(100)"]) as proc:
2790	n/a	pass
2791	n/a	# __exit__ calls wait(), so the returncode should be set
2792	n/a	self.assertEqual(proc.returncode, 100)
2793	n/a
2794	n/a	def test_communicate_stdin(self):
2795	n/a	with subprocess.Popen([sys.executable, "-c",
2796	n/a	"import sys;"
2797	n/a	"sys.exit(sys.stdin.read() == 'context')"],
2798	n/a	stdin=subprocess.PIPE) as proc:
2799	n/a	proc.communicate(b"context")
2800	n/a	self.assertEqual(proc.returncode, 1)
2801	n/a
2802	n/a	def test_invalid_args(self):
2803	n/a	with self.assertRaises((FileNotFoundError, PermissionError)) as c:
2804	n/a	with subprocess.Popen(['nonexisting_i_hope'],
2805	n/a	stdout=subprocess.PIPE,
2806	n/a	stderr=subprocess.PIPE) as proc:
2807	n/a	pass
2808	n/a
2809	n/a	def test_broken_pipe_cleanup(self):
2810	n/a	"""Broken pipe error should not prevent wait() (Issue 21619)"""
2811	n/a	proc = subprocess.Popen([sys.executable, '-c', 'pass'],
2812	n/a	stdin=subprocess.PIPE,
2813	n/a	bufsize=support.PIPE_MAX_SIZE*2)
2814	n/a	proc = proc.__enter__()
2815	n/a	# Prepare to send enough data to overflow any OS pipe buffering and
2816	n/a	# guarantee a broken pipe error. Data is held in BufferedWriter
2817	n/a	# buffer until closed.
2818	n/a	proc.stdin.write(b'x' * support.PIPE_MAX_SIZE)
2819	n/a	self.assertIsNone(proc.returncode)
2820	n/a	# EPIPE expected under POSIX; EINVAL under Windows
2821	n/a	self.assertRaises(OSError, proc.__exit__, None, None, None)
2822	n/a	self.assertEqual(proc.returncode, 0)
2823	n/a	self.assertTrue(proc.stdin.closed)
2824	n/a
2825	n/a
2826	n/a	if __name__ == "__main__":
2827	n/a	unittest.main()