Commit 22d0a57a authored by Jason Madden's avatar Jason Madden

Update tests

parent a8bde021
...@@ -500,7 +500,7 @@ class BasicSocketTests(unittest.TestCase): ...@@ -500,7 +500,7 @@ class BasicSocketTests(unittest.TestCase):
('email', 'null@python.org\x00user@example.org'), ('email', 'null@python.org\x00user@example.org'),
('URI', 'http://null.python.org\x00http://example.org'), ('URI', 'http://null.python.org\x00http://example.org'),
('IP Address', '192.0.2.1'), ('IP Address', '192.0.2.1'),
('IP Address', '2001:DB8:0:0:0:0:0:1\n')) ('IP Address', '2001:DB8:0:0:0:0:0:1'))
else: else:
# OpenSSL 0.9.7 doesn't support IPv6 addresses in subjectAltName # OpenSSL 0.9.7 doesn't support IPv6 addresses in subjectAltName
san = (('DNS', 'altnull.python.org\x00example.com'), san = (('DNS', 'altnull.python.org\x00example.com'),
...@@ -527,7 +527,7 @@ class BasicSocketTests(unittest.TestCase): ...@@ -527,7 +527,7 @@ class BasicSocketTests(unittest.TestCase):
(('commonName', 'dirname example'),))), (('commonName', 'dirname example'),))),
('URI', 'https://www.python.org/'), ('URI', 'https://www.python.org/'),
('IP Address', '127.0.0.1'), ('IP Address', '127.0.0.1'),
('IP Address', '0:0:0:0:0:0:0:1\n'), ('IP Address', '0:0:0:0:0:0:0:1'),
('Registered ID', '1.2.3.4.5') ('Registered ID', '1.2.3.4.5')
) )
) )
...@@ -554,11 +554,11 @@ class BasicSocketTests(unittest.TestCase): ...@@ -554,11 +554,11 @@ class BasicSocketTests(unittest.TestCase):
# Some sanity checks follow # Some sanity checks follow
# >= 0.9 # >= 0.9
self.assertGreaterEqual(n, 0x900000) self.assertGreaterEqual(n, 0x900000)
# < 3.0 # < 4.0
self.assertLess(n, 0x30000000) self.assertLess(n, 0x40000000)
major, minor, fix, patch, status = t major, minor, fix, patch, status = t
self.assertGreaterEqual(major, 0) self.assertGreaterEqual(major, 1)
self.assertLess(major, 3) self.assertLess(major, 4)
self.assertGreaterEqual(minor, 0) self.assertGreaterEqual(minor, 0)
self.assertLess(minor, 256) self.assertLess(minor, 256)
self.assertGreaterEqual(fix, 0) self.assertGreaterEqual(fix, 0)
...@@ -1220,12 +1220,18 @@ class ContextTests(unittest.TestCase): ...@@ -1220,12 +1220,18 @@ class ContextTests(unittest.TestCase):
# RHEL 8 uses TLS 1.2 by default # RHEL 8 uses TLS 1.2 by default
ssl.TLSVersion.TLSv1_2 ssl.TLSVersion.TLSv1_2
} }
maximum_range = {
# stock OpenSSL
ssl.TLSVersion.MAXIMUM_SUPPORTED,
# Fedora 32 uses TLS 1.3 by default
ssl.TLSVersion.TLSv1_3
}
self.assertIn( self.assertIn(
ctx.minimum_version, minimum_range ctx.minimum_version, minimum_range
) )
self.assertEqual( self.assertIn(
ctx.maximum_version, ssl.TLSVersion.MAXIMUM_SUPPORTED ctx.maximum_version, maximum_range
) )
ctx.minimum_version = ssl.TLSVersion.TLSv1_1 ctx.minimum_version = ssl.TLSVersion.TLSv1_1
......
...@@ -485,7 +485,7 @@ class BasicSocketTests(unittest.TestCase): ...@@ -485,7 +485,7 @@ class BasicSocketTests(unittest.TestCase):
('email', 'null@python.org\x00user@example.org'), ('email', 'null@python.org\x00user@example.org'),
('URI', 'http://null.python.org\x00http://example.org'), ('URI', 'http://null.python.org\x00http://example.org'),
('IP Address', '192.0.2.1'), ('IP Address', '192.0.2.1'),
('IP Address', '2001:DB8:0:0:0:0:0:1\n')) ('IP Address', '2001:DB8:0:0:0:0:0:1'))
else: else:
# OpenSSL 0.9.7 doesn't support IPv6 addresses in subjectAltName # OpenSSL 0.9.7 doesn't support IPv6 addresses in subjectAltName
san = (('DNS', 'altnull.python.org\x00example.com'), san = (('DNS', 'altnull.python.org\x00example.com'),
...@@ -512,7 +512,7 @@ class BasicSocketTests(unittest.TestCase): ...@@ -512,7 +512,7 @@ class BasicSocketTests(unittest.TestCase):
(('commonName', 'dirname example'),))), (('commonName', 'dirname example'),))),
('URI', 'https://www.python.org/'), ('URI', 'https://www.python.org/'),
('IP Address', '127.0.0.1'), ('IP Address', '127.0.0.1'),
('IP Address', '0:0:0:0:0:0:0:1\n'), ('IP Address', '0:0:0:0:0:0:0:1'),
('Registered ID', '1.2.3.4.5') ('Registered ID', '1.2.3.4.5')
) )
) )
...@@ -539,11 +539,11 @@ class BasicSocketTests(unittest.TestCase): ...@@ -539,11 +539,11 @@ class BasicSocketTests(unittest.TestCase):
# Some sanity checks follow # Some sanity checks follow
# >= 0.9 # >= 0.9
self.assertGreaterEqual(n, 0x900000) self.assertGreaterEqual(n, 0x900000)
# < 3.0 # < 4.0
self.assertLess(n, 0x30000000) self.assertLess(n, 0x40000000)
major, minor, fix, patch, status = t major, minor, fix, patch, status = t
self.assertGreaterEqual(major, 0) self.assertGreaterEqual(major, 1)
self.assertLess(major, 3) self.assertLess(major, 4)
self.assertGreaterEqual(minor, 0) self.assertGreaterEqual(minor, 0)
self.assertLess(minor, 256) self.assertLess(minor, 256)
self.assertGreaterEqual(fix, 0) self.assertGreaterEqual(fix, 0)
...@@ -1200,12 +1200,18 @@ class ContextTests(unittest.TestCase): ...@@ -1200,12 +1200,18 @@ class ContextTests(unittest.TestCase):
# RHEL 8 uses TLS 1.2 by default # RHEL 8 uses TLS 1.2 by default
ssl.TLSVersion.TLSv1_2 ssl.TLSVersion.TLSv1_2
} }
maximum_range = {
# stock OpenSSL
ssl.TLSVersion.MAXIMUM_SUPPORTED,
# Fedora 32 uses TLS 1.3 by default
ssl.TLSVersion.TLSv1_3
}
self.assertIn( self.assertIn(
ctx.minimum_version, minimum_range ctx.minimum_version, minimum_range
) )
self.assertEqual( self.assertIn(
ctx.maximum_version, ssl.TLSVersion.MAXIMUM_SUPPORTED ctx.maximum_version, maximum_range
) )
ctx.minimum_version = ssl.TLSVersion.TLSv1_1 ctx.minimum_version = ssl.TLSVersion.TLSv1_1
......
...@@ -45,6 +45,18 @@ NONEXISTING_CMD = ('nonexisting_i_hope',) ...@@ -45,6 +45,18 @@ NONEXISTING_CMD = ('nonexisting_i_hope',)
# Ignore errors that indicate the command was not found # Ignore errors that indicate the command was not found
NONEXISTING_ERRORS = (FileNotFoundError, NotADirectoryError, PermissionError) NONEXISTING_ERRORS = (FileNotFoundError, NotADirectoryError, PermissionError)
ZERO_RETURN_CMD = (sys.executable, '-c', 'pass')
def setUpModule():
shell_true = shutil.which('true')
if shell_true is None:
return
if (os.access(shell_true, os.X_OK) and
subprocess.run([shell_true]).returncode == 0):
global ZERO_RETURN_CMD
ZERO_RETURN_CMD = (shell_true,) # Faster than Python startup.
class BaseTestCase(unittest.TestCase): class BaseTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
...@@ -89,7 +101,7 @@ class PopenExecuteChildRaises(subprocess.Popen): ...@@ -89,7 +101,7 @@ class PopenExecuteChildRaises(subprocess.Popen):
class ProcessTestCase(BaseTestCase): class ProcessTestCase(BaseTestCase):
def test_io_buffered_by_default(self): def test_io_buffered_by_default(self):
p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], p = subprocess.Popen(ZERO_RETURN_CMD,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
try: try:
...@@ -103,7 +115,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -103,7 +115,7 @@ class ProcessTestCase(BaseTestCase):
p.wait() p.wait()
def test_io_unbuffered_works(self): def test_io_unbuffered_works(self):
p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], p = subprocess.Popen(ZERO_RETURN_CMD,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, bufsize=0) stderr=subprocess.PIPE, bufsize=0)
try: try:
...@@ -133,8 +145,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -133,8 +145,7 @@ class ProcessTestCase(BaseTestCase):
def test_check_call_zero(self): def test_check_call_zero(self):
# check_call() function with zero return code # check_call() function with zero return code
rc = subprocess.check_call([sys.executable, "-c", rc = subprocess.check_call(ZERO_RETURN_CMD)
"import sys; sys.exit(0)"])
self.assertEqual(rc, 0) self.assertEqual(rc, 0)
def test_check_call_nonzero(self): def test_check_call_nonzero(self):
...@@ -700,19 +711,19 @@ class ProcessTestCase(BaseTestCase): ...@@ -700,19 +711,19 @@ class ProcessTestCase(BaseTestCase):
newenv = os.environ.copy() newenv = os.environ.copy()
newenv["FRUIT\0VEGETABLE"] = "cabbage" newenv["FRUIT\0VEGETABLE"] = "cabbage"
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
subprocess.Popen([sys.executable, "-c", "pass"], env=newenv) subprocess.Popen(ZERO_RETURN_CMD, env=newenv)
# null character in the environment variable value # null character in the environment variable value
newenv = os.environ.copy() newenv = os.environ.copy()
newenv["FRUIT"] = "orange\0VEGETABLE=cabbage" newenv["FRUIT"] = "orange\0VEGETABLE=cabbage"
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
subprocess.Popen([sys.executable, "-c", "pass"], env=newenv) subprocess.Popen(ZERO_RETURN_CMD, env=newenv)
# equal character in the environment variable name # equal character in the environment variable name
newenv = os.environ.copy() newenv = os.environ.copy()
newenv["FRUIT=ORANGE"] = "lemon" newenv["FRUIT=ORANGE"] = "lemon"
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
subprocess.Popen([sys.executable, "-c", "pass"], env=newenv) subprocess.Popen(ZERO_RETURN_CMD, env=newenv)
# equal character in the environment variable value # equal character in the environment variable value
newenv = os.environ.copy() newenv = os.environ.copy()
...@@ -813,7 +824,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -813,7 +824,7 @@ class ProcessTestCase(BaseTestCase):
options['stderr'] = subprocess.PIPE options['stderr'] = subprocess.PIPE
if not options: if not options:
continue continue
p = subprocess.Popen((sys.executable, "-c", "pass"), **options) p = subprocess.Popen(ZERO_RETURN_CMD, **options)
p.communicate() p.communicate()
if p.stdin is not None: if p.stdin is not None:
self.assertTrue(p.stdin.closed) self.assertTrue(p.stdin.closed)
...@@ -952,7 +963,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -952,7 +963,7 @@ class ProcessTestCase(BaseTestCase):
# #
# We set stdout to PIPE because, as of this writing, a different # We set stdout to PIPE because, as of this writing, a different
# code path is tested when the number of pipes is zero or one. # code path is tested when the number of pipes is zero or one.
p = subprocess.Popen([sys.executable, "-c", "pass"], p = subprocess.Popen(ZERO_RETURN_CMD,
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True) universal_newlines=True)
...@@ -1100,7 +1111,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -1100,7 +1111,7 @@ class ProcessTestCase(BaseTestCase):
self.assertEqual(p.poll(), 0) self.assertEqual(p.poll(), 0)
def test_wait(self): def test_wait(self):
p = subprocess.Popen([sys.executable, "-c", "pass"]) p = subprocess.Popen(ZERO_RETURN_CMD)
self.assertEqual(p.wait(), 0) self.assertEqual(p.wait(), 0)
# Subsequent invocations should just return the returncode # Subsequent invocations should just return the returncode
self.assertEqual(p.wait(), 0) self.assertEqual(p.wait(), 0)
...@@ -1119,14 +1130,14 @@ class ProcessTestCase(BaseTestCase): ...@@ -1119,14 +1130,14 @@ class ProcessTestCase(BaseTestCase):
# an invalid type of the bufsize argument should raise # an invalid type of the bufsize argument should raise
# TypeError. # TypeError.
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
subprocess.Popen([sys.executable, "-c", "pass"], "orange") subprocess.Popen(ZERO_RETURN_CMD, "orange")
def test_bufsize_is_none(self): def test_bufsize_is_none(self):
# bufsize=None should be the same as bufsize=0. # bufsize=None should be the same as bufsize=0.
p = subprocess.Popen([sys.executable, "-c", "pass"], None) p = subprocess.Popen(ZERO_RETURN_CMD, None)
self.assertEqual(p.wait(), 0) self.assertEqual(p.wait(), 0)
# Again with keyword arg # Again with keyword arg
p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None) p = subprocess.Popen(ZERO_RETURN_CMD, bufsize=None)
self.assertEqual(p.wait(), 0) self.assertEqual(p.wait(), 0)
def _test_bufsize_equal_one(self, line, expected, universal_newlines): def _test_bufsize_equal_one(self, line, expected, universal_newlines):
...@@ -1331,7 +1342,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -1331,7 +1342,7 @@ class ProcessTestCase(BaseTestCase):
def test_communicate_epipe(self): def test_communicate_epipe(self):
# Issue 10963: communicate() should hide EPIPE # Issue 10963: communicate() should hide EPIPE
p = subprocess.Popen([sys.executable, "-c", 'pass'], p = subprocess.Popen(ZERO_RETURN_CMD,
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
...@@ -1342,7 +1353,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -1342,7 +1353,7 @@ class ProcessTestCase(BaseTestCase):
def test_communicate_epipe_only_stdin(self): def test_communicate_epipe_only_stdin(self):
# Issue 10963: communicate() should hide EPIPE # Issue 10963: communicate() should hide EPIPE
p = subprocess.Popen([sys.executable, "-c", 'pass'], p = subprocess.Popen(ZERO_RETURN_CMD,
stdin=subprocess.PIPE) stdin=subprocess.PIPE)
self.addCleanup(p.stdin.close) self.addCleanup(p.stdin.close)
p.wait() p.wait()
...@@ -1381,7 +1392,7 @@ class ProcessTestCase(BaseTestCase): ...@@ -1381,7 +1392,7 @@ class ProcessTestCase(BaseTestCase):
fds_before_popen = os.listdir(fd_directory) fds_before_popen = os.listdir(fd_directory)
with self.assertRaises(PopenTestException): with self.assertRaises(PopenTestException):
PopenExecuteChildRaises( PopenExecuteChildRaises(
[sys.executable, '-c', 'pass'], stdin=subprocess.PIPE, ZERO_RETURN_CMD, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# NOTE: This test doesn't verify that the real _execute_child # NOTE: This test doesn't verify that the real _execute_child
...@@ -1424,7 +1435,7 @@ class RunFuncTestCase(BaseTestCase): ...@@ -1424,7 +1435,7 @@ class RunFuncTestCase(BaseTestCase):
def test_check_zero(self): def test_check_zero(self):
# check_returncode shouldn't raise when returncode is zero # check_returncode shouldn't raise when returncode is zero
cp = self.run_python("import sys; sys.exit(0)", check=True) cp = subprocess.run(ZERO_RETURN_CMD, check=True)
self.assertEqual(cp.returncode, 0) self.assertEqual(cp.returncode, 0)
def test_timeout(self): def test_timeout(self):
...@@ -1824,7 +1835,7 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -1824,7 +1835,7 @@ class POSIXProcessTestCase(BaseTestCase):
with self.assertRaises(subprocess.SubprocessError): with self.assertRaises(subprocess.SubprocessError):
self._TestExecuteChildPopen( self._TestExecuteChildPopen(
self, [sys.executable, "-c", "pass"], self, ZERO_RETURN_CMD,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, preexec_fn=raise_it) stderr=subprocess.PIPE, preexec_fn=raise_it)
...@@ -2281,7 +2292,7 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2281,7 +2292,7 @@ class POSIXProcessTestCase(BaseTestCase):
try: try:
subprocess.call( subprocess.call(
[sys.executable, "-c", "pass"], ZERO_RETURN_CMD,
preexec_fn=prepare) preexec_fn=prepare)
except ValueError as err: except ValueError as err:
# Pure Python implementations keeps the message # Pure Python implementations keeps the message
...@@ -2324,29 +2335,30 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2324,29 +2335,30 @@ class POSIXProcessTestCase(BaseTestCase):
self.assertEqual(stdout.decode('ascii'), ascii(encoded_value)) self.assertEqual(stdout.decode('ascii'), ascii(encoded_value))
def test_bytes_program(self): def test_bytes_program(self):
abs_program = os.fsencode(sys.executable) abs_program = os.fsencode(ZERO_RETURN_CMD[0])
path, program = os.path.split(sys.executable) args = list(ZERO_RETURN_CMD[1:])
path, program = os.path.split(ZERO_RETURN_CMD[0])
program = os.fsencode(program) program = os.fsencode(program)
# absolute bytes path # absolute bytes path
exitcode = subprocess.call([abs_program, "-c", "pass"]) exitcode = subprocess.call([abs_program]+args)
self.assertEqual(exitcode, 0) self.assertEqual(exitcode, 0)
# absolute bytes path as a string # absolute bytes path as a string
cmd = b"'" + abs_program + b"' -c pass" cmd = b"'%s' %s" % (abs_program, " ".join(args).encode("utf-8"))
exitcode = subprocess.call(cmd, shell=True) exitcode = subprocess.call(cmd, shell=True)
self.assertEqual(exitcode, 0) self.assertEqual(exitcode, 0)
# bytes program, unicode PATH # bytes program, unicode PATH
env = os.environ.copy() env = os.environ.copy()
env["PATH"] = path env["PATH"] = path
exitcode = subprocess.call([program, "-c", "pass"], env=env) exitcode = subprocess.call([program]+args, env=env)
self.assertEqual(exitcode, 0) self.assertEqual(exitcode, 0)
# bytes program, bytes PATH # bytes program, bytes PATH
envb = os.environb.copy() envb = os.environb.copy()
envb[b"PATH"] = os.fsencode(path) envb[b"PATH"] = os.fsencode(path)
exitcode = subprocess.call([program, "-c", "pass"], env=envb) exitcode = subprocess.call([program]+args, env=envb)
self.assertEqual(exitcode, 0) self.assertEqual(exitcode, 0)
def test_pipe_cloexec(self): def test_pipe_cloexec(self):
...@@ -2574,7 +2586,7 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2574,7 +2586,7 @@ class POSIXProcessTestCase(BaseTestCase):
# pass_fds overrides close_fds with a warning. # pass_fds overrides close_fds with a warning.
with self.assertWarns(RuntimeWarning) as context: with self.assertWarns(RuntimeWarning) as context:
self.assertFalse(subprocess.call( self.assertFalse(subprocess.call(
[sys.executable, "-c", "import sys; sys.exit(0)"], ZERO_RETURN_CMD,
close_fds=False, pass_fds=(fd, ))) close_fds=False, pass_fds=(fd, )))
self.assertIn('overriding close_fds', str(context.warning)) self.assertIn('overriding close_fds', str(context.warning))
...@@ -2636,19 +2648,19 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2636,19 +2648,19 @@ class POSIXProcessTestCase(BaseTestCase):
def test_stdout_stdin_are_single_inout_fd(self): def test_stdout_stdin_are_single_inout_fd(self):
with io.open(os.devnull, "r+") as inout: with io.open(os.devnull, "r+") as inout:
p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], p = subprocess.Popen(ZERO_RETURN_CMD,
stdout=inout, stdin=inout) stdout=inout, stdin=inout)
p.wait() p.wait()
def test_stdout_stderr_are_single_inout_fd(self): def test_stdout_stderr_are_single_inout_fd(self):
with io.open(os.devnull, "r+") as inout: with io.open(os.devnull, "r+") as inout:
p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], p = subprocess.Popen(ZERO_RETURN_CMD,
stdout=inout, stderr=inout) stdout=inout, stderr=inout)
p.wait() p.wait()
def test_stderr_stdin_are_single_inout_fd(self): def test_stderr_stdin_are_single_inout_fd(self):
with io.open(os.devnull, "r+") as inout: with io.open(os.devnull, "r+") as inout:
p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], p = subprocess.Popen(ZERO_RETURN_CMD,
stderr=inout, stdin=inout) stderr=inout, stdin=inout)
p.wait() p.wait()
...@@ -2836,7 +2848,7 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2836,7 +2848,7 @@ class POSIXProcessTestCase(BaseTestCase):
def test_communicate_BrokenPipeError_stdin_close(self): def test_communicate_BrokenPipeError_stdin_close(self):
# By not setting stdout or stderr or a timeout we force the fast path # By not setting stdout or stderr or a timeout we force the fast path
# that just calls _stdin_write() internally due to our mock. # that just calls _stdin_write() internally due to our mock.
proc = subprocess.Popen([sys.executable, '-c', 'pass']) proc = subprocess.Popen(ZERO_RETURN_CMD)
with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin:
mock_proc_stdin.close.side_effect = BrokenPipeError mock_proc_stdin.close.side_effect = BrokenPipeError
proc.communicate() # Should swallow BrokenPipeError from close. proc.communicate() # Should swallow BrokenPipeError from close.
...@@ -2845,7 +2857,7 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2845,7 +2857,7 @@ class POSIXProcessTestCase(BaseTestCase):
def test_communicate_BrokenPipeError_stdin_write(self): def test_communicate_BrokenPipeError_stdin_write(self):
# By not setting stdout or stderr or a timeout we force the fast path # By not setting stdout or stderr or a timeout we force the fast path
# that just calls _stdin_write() internally due to our mock. # that just calls _stdin_write() internally due to our mock.
proc = subprocess.Popen([sys.executable, '-c', 'pass']) proc = subprocess.Popen(ZERO_RETURN_CMD)
with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin: with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin:
mock_proc_stdin.write.side_effect = BrokenPipeError mock_proc_stdin.write.side_effect = BrokenPipeError
proc.communicate(b'stuff') # Should swallow the BrokenPipeError. proc.communicate(b'stuff') # Should swallow the BrokenPipeError.
...@@ -2884,7 +2896,7 @@ class POSIXProcessTestCase(BaseTestCase): ...@@ -2884,7 +2896,7 @@ class POSIXProcessTestCase(BaseTestCase):
'need _testcapi.W_STOPCODE') 'need _testcapi.W_STOPCODE')
def test_stopped(self): def test_stopped(self):
"""Test wait() behavior when waitpid returns WIFSTOPPED; issue29335.""" """Test wait() behavior when waitpid returns WIFSTOPPED; issue29335."""
args = [sys.executable, '-c', 'pass'] args = ZERO_RETURN_CMD
proc = subprocess.Popen(args) proc = subprocess.Popen(args)
# Wait until the real process completes to avoid zombie process # Wait until the real process completes to avoid zombie process
...@@ -2914,7 +2926,7 @@ class Win32ProcessTestCase(BaseTestCase): ...@@ -2914,7 +2926,7 @@ class Win32ProcessTestCase(BaseTestCase):
# Since Python is a console process, it won't be affected # Since Python is a console process, it won't be affected
# by wShowWindow, but the argument should be silently # by wShowWindow, but the argument should be silently
# ignored # ignored
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], subprocess.call(ZERO_RETURN_CMD,
startupinfo=startupinfo) startupinfo=startupinfo)
def test_startupinfo_keywords(self): def test_startupinfo_keywords(self):
...@@ -2930,7 +2942,7 @@ class Win32ProcessTestCase(BaseTestCase): ...@@ -2930,7 +2942,7 @@ class Win32ProcessTestCase(BaseTestCase):
# Since Python is a console process, it won't be affected # Since Python is a console process, it won't be affected
# by wShowWindow, but the argument should be silently # by wShowWindow, but the argument should be silently
# ignored # ignored
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], subprocess.call(ZERO_RETURN_CMD,
startupinfo=startupinfo) startupinfo=startupinfo)
def test_startupinfo_copy(self): def test_startupinfo_copy(self):
...@@ -2942,7 +2954,7 @@ class Win32ProcessTestCase(BaseTestCase): ...@@ -2942,7 +2954,7 @@ class Win32ProcessTestCase(BaseTestCase):
# Call Popen() twice with the same startupinfo object to make sure # Call Popen() twice with the same startupinfo object to make sure
# that it's not modified # that it's not modified
for _ in range(2): for _ in range(2):
cmd = [sys.executable, "-c", "pass"] cmd = ZERO_RETURN_CMD
with open(os.devnull, 'w') as null: with open(os.devnull, 'w') as null:
proc = subprocess.Popen(cmd, proc = subprocess.Popen(cmd,
stdout=null, stdout=null,
...@@ -2982,7 +2994,7 @@ class Win32ProcessTestCase(BaseTestCase): ...@@ -2982,7 +2994,7 @@ class Win32ProcessTestCase(BaseTestCase):
class BadEnv(dict): class BadEnv(dict):
keys = None keys = None
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
subprocess.Popen([sys.executable, "-c", "pass"], env=BadEnv()) subprocess.Popen(ZERO_RETURN_CMD, env=BadEnv())
def test_close_fds(self): def test_close_fds(self):
# close file descriptors # close file descriptors
...@@ -3043,13 +3055,13 @@ class Win32ProcessTestCase(BaseTestCase): ...@@ -3043,13 +3055,13 @@ class Win32ProcessTestCase(BaseTestCase):
def test_empty_attribute_list(self): def test_empty_attribute_list(self):
startupinfo = subprocess.STARTUPINFO() startupinfo = subprocess.STARTUPINFO()
startupinfo.lpAttributeList = {} startupinfo.lpAttributeList = {}
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], subprocess.call(ZERO_RETURN_CMD,
startupinfo=startupinfo) startupinfo=startupinfo)
def test_empty_handle_list(self): def test_empty_handle_list(self):
startupinfo = subprocess.STARTUPINFO() startupinfo = subprocess.STARTUPINFO()
startupinfo.lpAttributeList = {"handle_list": []} startupinfo.lpAttributeList = {"handle_list": []}
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], subprocess.call(ZERO_RETURN_CMD,
startupinfo=startupinfo) startupinfo=startupinfo)
def test_shell_sequence(self): def test_shell_sequence(self):
...@@ -3348,7 +3360,7 @@ class ContextManagerTests(BaseTestCase): ...@@ -3348,7 +3360,7 @@ class ContextManagerTests(BaseTestCase):
def test_broken_pipe_cleanup(self): def test_broken_pipe_cleanup(self):
"""Broken pipe error should not prevent wait() (Issue 21619)""" """Broken pipe error should not prevent wait() (Issue 21619)"""
proc = subprocess.Popen([sys.executable, '-c', 'pass'], proc = subprocess.Popen(ZERO_RETURN_CMD,
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
bufsize=support.PIPE_MAX_SIZE*2) bufsize=support.PIPE_MAX_SIZE*2)
proc = proc.__enter__() proc = proc.__enter__()
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment