Skip to content

Update subprocess to CPython 3.11 #4981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
youknowone merged 3 commits into from
May 20, 2023
Merged

Update subprocess to CPython 3.11 #4981

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2b43d48
Add additional arguments to fork_exec.
DimitrisJim May 19, 2023
7366a41
Update subprocess from CPython 3.11
DimitrisJim May 19, 2023
283a804
Mark failing tests.
DimitrisJim May 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Mark failing tests.
  • Loading branch information
@DimitrisJim
DimitrisJim committed May 19, 2023
commit 283a8046aade9ed676e49bdd44a0587da6e1028a
70 changes: 70 additions & 0 deletions Lib/test/test_subprocess.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -811,6 +811,7 @@ def is_env_var_to_ignore(n):
if not is_env_var_to_ignore(k)]
self.assertEqual(child_env_names, [])

@unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON, null byte is not checked")
def test_invalid_cmd(self):
# null character in the command name
cmd = sys.executable + '\0'
Expand Down Expand Up @@ -956,6 +957,7 @@ def test_communicate_returns(self):
self.assertEqual(stdout, None)
self.assertEqual(stderr, None)

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_communicate_pipe_buf(self):
# communicate() with writes larger than pipe_buf
# This test will probably deadlock rather than fail, if
Expand Down Expand Up @@ -995,6 +997,8 @@ def test_writes_before_communicate(self):
self.assertEqual(stdout, b"bananasplit")
self.assertEqual(stderr, b"")

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_universal_newlines_and_text(self):
args = [
sys.executable, "-c",
Expand Down Expand Up @@ -1034,6 +1038,7 @@ def test_universal_newlines_and_text(self):
self.assertEqual(p.stdout.read(),
"line4\nline5\nline6\nline7\nline8")

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_universal_newlines_communicate(self):
# universal newlines through communicate()
p = subprocess.Popen([sys.executable, "-c",
Expand Down Expand Up @@ -1085,6 +1090,7 @@ def test_universal_newlines_communicate_input_none(self):
p.communicate()
self.assertEqual(p.returncode, 0)

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_universal_newlines_communicate_stdin_stdout_stderr(self):
# universal newlines through communicate(), with stdin, stdout, stderr
p = subprocess.Popen([sys.executable, "-c",
Expand Down Expand Up @@ -1113,6 +1119,8 @@ def test_universal_newlines_communicate_stdin_stdout_stderr(self):
# to stderr at exit of subprocess.
self.assertTrue(stderr.startswith("eline2\neline6\neline7\n"))

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_universal_newlines_communicate_encodings(self):
# Check that universal newlines mode works for various encodings,
# in particular for encodings in the UTF-16 and UTF-32 families.
Expand All @@ -1135,6 +1143,8 @@ def test_universal_newlines_communicate_encodings(self):
stdout, stderr = popen.communicate(input='')
self.assertEqual(stdout, '1\n2\n3\n4')

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_communicate_errors(self):
for errors, expected in [
('ignore', ''),
Expand Down Expand Up @@ -1274,12 +1284,15 @@ def _test_bufsize_equal_one(self, line, expected, universal_newlines):
self.assertEqual(p.returncode, 0)
self.assertEqual(read_line, expected)

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_bufsize_equal_one_text_mode(self):
# line is flushed in text mode with bufsize=1.
# we should get the full line in return
line = "line\n"
self._test_bufsize_equal_one(line, line, universal_newlines=True)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_bufsize_equal_one_binary_mode(self):
# line is not flushed in binary mode with bufsize=1.
# we should get empty response
Expand Down Expand Up @@ -1452,6 +1465,7 @@ def test_handles_closed_on_exception(self):
self.assertFalse(os.path.exists(ofname))
self.assertFalse(os.path.exists(efname))

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_communicate_epipe(self):
# Issue 10963: communicate() should hide EPIPE
p = subprocess.Popen(ZERO_RETURN_CMD,
Expand Down Expand Up @@ -1482,6 +1496,7 @@ def test_repr(self):
p.returncode = code
self.assertEqual(repr(p), sx)

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_communicate_epipe_only_stdin(self):
# Issue 10963: communicate() should hide EPIPE
p = subprocess.Popen(ZERO_RETURN_CMD,
Expand All @@ -1490,6 +1505,8 @@ def test_communicate_epipe_only_stdin(self):
p.wait()
p.communicate(b"x" * 2**20)

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipUnless(hasattr(signal, 'SIGUSR1'),
"Requires signal.SIGUSR1")
@unittest.skipUnless(hasattr(os, 'kill'),
Expand Down Expand Up @@ -1539,6 +1556,8 @@ def test_file_not_found_includes_filename(self):
subprocess.call(['/opt/nonexistent_binary', 'with', 'some', 'args'])
self.assertEqual(c.exception.filename, '/opt/nonexistent_binary')

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipIf(mswindows, "behavior currently not supported on Windows")
def test_file_not_found_with_bad_cwd(self):
with self.assertRaises(FileNotFoundError) as c:
Expand Down Expand Up @@ -1739,6 +1758,8 @@ def test_run_with_shell_timeout_and_capture_output(self):
msg="TimeoutExpired was delayed! Bad traceback:\n```\n"
f"{stacks}```")

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_encoding_warning(self):
code = textwrap.dedent("""\
from subprocess import *
Expand Down Expand Up @@ -1785,6 +1806,8 @@ def _get_chdir_exception(self):
self._nonexistent_dir)
return desired_exception

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_exception_cwd(self):
"""Test error in the child raised in the parent for a bad cwd."""
desired_exception = self._get_chdir_exception()
Expand All @@ -1800,6 +1823,8 @@ def test_exception_cwd(self):
else:
self.fail("Expected OSError: %s" % desired_exception)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_exception_bad_executable(self):
"""Test error in the child raised in the parent for a bad executable."""
desired_exception = self._get_chdir_exception()
Expand All @@ -1815,6 +1840,8 @@ def test_exception_bad_executable(self):
else:
self.fail("Expected OSError: %s" % desired_exception)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_exception_bad_args_0(self):
"""Test error in the child raised in the parent for a bad args[0]."""
desired_exception = self._get_chdir_exception()
Expand Down Expand Up @@ -1879,6 +1906,8 @@ def bad_error(*args):

self.assertIn(repr(error_data), str(e.exception))

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipIf(not os.path.exists('/proc/self/status'),
"need /proc/self/status")
def test_restore_signals(self):
Expand Down Expand Up @@ -1919,6 +1948,8 @@ def test_start_new_session(self):
child_sid = int(output)
self.assertNotEqual(parent_sid, child_sid)

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipUnless(hasattr(os, 'setpgid') and hasattr(os, 'getpgid'),
'no setpgid or getpgid on platform')
def test_process_group_0(self):
Expand All @@ -1937,6 +1968,8 @@ def test_process_group_0(self):
child_pgid = int(output)
self.assertNotEqual(parent_pgid, child_pgid)

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipUnless(hasattr(os, 'setreuid'), 'no setreuid on platform')
def test_user(self):
# For code coverage of the user parameter. We don't care if we get an
Expand Down Expand Up @@ -1994,6 +2027,8 @@ def test_user_error(self):
with self.assertRaises(ValueError):
subprocess.check_call(ZERO_RETURN_CMD, user=65535)

# TODO: RUSTPYTHON, observed gids do not match expected gids
@unittest.expectedFailure
@unittest.skipUnless(hasattr(os, 'setregid'), 'no setregid() on platform')
def test_group(self):
gid = os.getegid()
Expand Down Expand Up @@ -2041,6 +2076,8 @@ def test_group_error(self):
with self.assertRaises(ValueError):
subprocess.check_call(ZERO_RETURN_CMD, group=65535)

# TODO: RUSTPYTHON, observed gids do not match expected gids
@unittest.expectedFailure
@unittest.skipUnless(hasattr(os, 'setgroups'), 'no setgroups() on platform')
def test_extra_groups(self):
gid = os.getegid()
Expand Down Expand Up @@ -2095,6 +2132,8 @@ def test_extra_groups_error(self):
with self.assertRaises(ValueError):
subprocess.check_call(ZERO_RETURN_CMD, extra_groups=[])

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipIf(mswindows or not hasattr(os, 'umask'),
'POSIX umask() is not available.')
def test_umask(self):
Expand Down Expand Up @@ -2146,6 +2185,8 @@ def test_CalledProcessError_str_non_zero(self):
error_string = str(err)
self.assertIn("non-zero exit status 2.", error_string)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_preexec(self):
# DISCLAIMER: Setting environment variables is *not* a good use
# of a preexec_fn. This is merely a test.
Expand All @@ -2157,6 +2198,8 @@ def test_preexec(self):
with p:
self.assertEqual(p.stdout.read(), b"apple")

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_preexec_exception(self):
def raise_it():
raise ValueError("What if two swallows carried a coconut?")
Expand Down Expand Up @@ -2198,6 +2241,8 @@ def _execute_child(self, *args, **kwargs):
for fd in devzero_fds:
os.close(fd)

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipIf(not os.path.exists("/dev/zero"), "/dev/zero required.")
def test_preexec_errpipe_does_not_double_close_pipes(self):
"""Issue16140: Don't double close pipes on preexec error."""
Expand All @@ -2212,6 +2257,8 @@ def raise_it():
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, preexec_fn=raise_it)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_preexec_gc_module_failure(self):
# This tests the code that disables garbage collection if the child
# process will execute any Python.
Expand All @@ -2233,6 +2280,8 @@ def test_preexec_gc_module_failure(self):
if not enabled:
gc.disable()

# TODO: RUSTPYTHON
@unittest.expectedFailure
@unittest.skipIf(
sys.platform == 'darwin', 'setrlimit() seems to fail on OS X')
def test_preexec_fork_failure(self):
Expand Down Expand Up @@ -2643,6 +2692,8 @@ def test_swap_std_fds_with_one_closed(self):
for to_fds in itertools.permutations(range(3), 2):
self._check_swap_std_fds_with_one_closed(from_fds, to_fds)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_surrogates_error_message(self):
def prepare():
raise ValueError("surrogate:\uDCff")
Expand All @@ -2662,6 +2713,8 @@ def prepare():
else:
self.fail("Expected ValueError or subprocess.SubprocessError")

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_undecodable_env(self):
for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')):
encoded_value = value.encode("ascii", "surrogateescape")
Expand Down Expand Up @@ -2782,6 +2835,7 @@ def kill_p2():
p1.stdout.close()
p2.stdout.close()

@unittest.skip("TODO: RUSTPYTHON, flaky test")
def test_close_fds(self):
fd_status = support.findfile("fd_status.py", subdir="subprocessdata")

Expand Down Expand Up @@ -2909,6 +2963,7 @@ def test_close_fds_when_max_fd_is_lowered(self):
msg="Some fds were left open.")


@unittest.skip("TODO: RUSTPYTHON, flaky test")
# Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file
# descriptor of a pipe closed in the parent process is valid in the
# child process according to fstat(), but the mode of the file
Expand Down Expand Up @@ -3116,6 +3171,8 @@ def test_leak_fast_process_del_killed(self):
else:
self.assertNotIn(ident, [id(o) for o in subprocess._active])

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_close_fds_after_preexec(self):
fd_status = support.findfile("fd_status.py", subdir="subprocessdata")

Expand Down Expand Up @@ -3418,6 +3475,8 @@ def test_close_fds(self):
close_fds=True)
self.assertEqual(rc, 47)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_close_fds_with_stdio(self):
import msvcrt

Expand Down Expand Up @@ -3500,6 +3559,8 @@ def test_shell_string(self):
with p:
self.assertIn(b"physalis", p.stdout.read())

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_shell_encodings(self):
# Run command through the shell (string)
for enc in ['ansi', 'oem']:
Expand Down Expand Up @@ -3646,6 +3707,7 @@ def popen_via_context_manager(*args, **kwargs):
raise KeyboardInterrupt # Test how __exit__ handles ^C.
self._test_keyboardinterrupt_no_kill(popen_via_context_manager)

@unittest.expectedFailureIfWindows("TODO: RUSTPYTHON")
def test_getoutput(self):
self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy')
self.assertEqual(subprocess.getstatusoutput('echo xyzzy'),
Expand Down Expand Up @@ -3718,20 +3780,28 @@ def with_spaces(self, *args, **kwargs):
"2 [%r, 'ab cd']" % self.fname
)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_shell_string_with_spaces(self):
# call() function with string argument with spaces on Windows
self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
"ab cd"), shell=1)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_shell_sequence_with_spaces(self):
# call() function with sequence argument with spaces on Windows
self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1)

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_noshell_string_with_spaces(self):
# call() function with string argument with spaces on Windows
self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname,
"ab cd"))

# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_noshell_sequence_with_spaces(self):
# call() function with sequence argument with spaces on Windows
self.with_spaces([sys.executable, self.fname, "ab cd"])
Expand Down