The functions *master_read* and *stdin_read* are passed a file descriptor
which they should read from, and they should always return a byte string. In
order to force spawn to return before the child process exits an
- :exc:`OSError` should be thrown.
+ empty byte array should be returned to signal end of file.
The default implementation for both functions will read and return up to 1024
bytes each time the function is called. The *master_read* callback is passed
process will quit without any input, *spawn* will then loop forever. If
*master_read* signals EOF the same behavior results (on linux at least).
- If both callbacks signal EOF then *spawn* will probably never return, unless
- *select* throws an error on your platform when passed three empty lists. This
- is a bug, documented in `issue 26228 <https://bugs.python.org/issue26228>`_.
-
Return the exit status value from :func:`os.waitpid` on the child process.
:func:`waitstatus_to_exitcode` can be used to convert the exit status into
import sys
import tty
-__all__ = ["openpty","fork","spawn"]
+# names imported directly for test mocking purposes
+from os import close, waitpid
+from tty import setraw, tcgetattr, tcsetattr
+
+__all__ = ["openpty", "fork", "spawn"]
STDIN_FILENO = 0
STDOUT_FILENO = 1
os.dup2(slave_fd, STDIN_FILENO)
os.dup2(slave_fd, STDOUT_FILENO)
os.dup2(slave_fd, STDERR_FILENO)
- if (slave_fd > STDERR_FILENO):
- os.close (slave_fd)
+ if slave_fd > STDERR_FILENO:
+ os.close(slave_fd)
# Explicitly open the tty to make it become a controlling tty.
tmp_fd = os.open(os.ttyname(STDOUT_FILENO), os.O_RDWR)
pty master -> standard output (master_read)
standard input -> pty master (stdin_read)"""
fds = [master_fd, STDIN_FILENO]
- while True:
- rfds, wfds, xfds = select(fds, [], [])
+ while fds:
+ rfds, _wfds, _xfds = select(fds, [], [])
+
if master_fd in rfds:
- data = master_read(master_fd)
+ # Some OSes signal EOF by returning an empty byte string,
+ # some throw OSErrors.
+ try:
+ data = master_read(master_fd)
+ except OSError:
+ data = b""
if not data: # Reached EOF.
- fds.remove(master_fd)
+ return # Assume the child process has exited and is
+ # unreachable, so we clean up.
else:
os.write(STDOUT_FILENO, data)
+
if STDIN_FILENO in rfds:
data = stdin_read(STDIN_FILENO)
if not data:
if type(argv) == type(''):
argv = (argv,)
sys.audit('pty.spawn', argv)
+
pid, master_fd = fork()
if pid == CHILD:
os.execlp(argv[0], *argv)
+
try:
- mode = tty.tcgetattr(STDIN_FILENO)
- tty.setraw(STDIN_FILENO)
- restore = 1
+ mode = tcgetattr(STDIN_FILENO)
+ setraw(STDIN_FILENO)
+ restore = True
except tty.error: # This is the same as termios.error
- restore = 0
+ restore = False
+
try:
_copy(master_fd, master_read, stdin_read)
- except OSError:
+ finally:
if restore:
- tty.tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode)
+ tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode)
- os.close(master_fd)
- return os.waitpid(pid, 0)[1]
+ close(master_fd)
+ return waitpid(pid, 0)[1]
import_module('termios')
import errno
-import pty
import os
+import pty
+import tty
import sys
import select
import signal
@staticmethod
def handle_sighup(signum, frame):
- # bpo-38547: if the process is the session leader, os.close(master_fd)
- # of "master_fd, slave_name = pty.master_open()" raises SIGHUP
- # signal: just ignore the signal.
- #
- # NOTE: the above comment is from an older version of the test;
- # master_open() is not being used anymore.
pass
@expectedFailureIfStdinIsTTY
self.assertEqual(_get_term_winsz(slave_fd), new_stdin_winsz,
"openpty() failed to set slave window size")
- # Solaris requires reading the fd before anything is returned.
- # My guess is that since we open and close the slave fd
- # in master_open(), we need to read the EOF.
- #
- # NOTE: the above comment is from an older version of the test;
- # master_open() is not being used anymore.
-
# Ensure the fd is non-blocking in case there's nothing to read.
blocking = os.get_blocking(master_fd)
try:
self.assertEqual(data, b"")
+ def test_spawn_doesnt_hang(self):
+ pty.spawn([sys.executable, '-c', 'print("hi there")'])
+
class SmallPtyTests(unittest.TestCase):
"""These tests don't spawn children or hang."""
def setUp(self):
self.orig_stdin_fileno = pty.STDIN_FILENO
self.orig_stdout_fileno = pty.STDOUT_FILENO
+ self.orig_pty_close = pty.close
+ self.orig_pty__copy = pty._copy
+ self.orig_pty_fork = pty.fork
self.orig_pty_select = pty.select
+ self.orig_pty_setraw = pty.setraw
+ self.orig_pty_tcgetattr = pty.tcgetattr
+ self.orig_pty_tcsetattr = pty.tcsetattr
+ self.orig_pty_waitpid = pty.waitpid
self.fds = [] # A list of file descriptors to close.
self.files = []
self.select_rfds_lengths = []
self.select_rfds_results = []
+ self.tcsetattr_mode_setting = None
def tearDown(self):
pty.STDIN_FILENO = self.orig_stdin_fileno
pty.STDOUT_FILENO = self.orig_stdout_fileno
+ pty.close = self.orig_pty_close
+ pty._copy = self.orig_pty__copy
+ pty.fork = self.orig_pty_fork
pty.select = self.orig_pty_select
+ pty.setraw = self.orig_pty_setraw
+ pty.tcgetattr = self.orig_pty_tcgetattr
+ pty.tcsetattr = self.orig_pty_tcsetattr
+ pty.waitpid = self.orig_pty_waitpid
for file in self.files:
try:
file.close()
self.assertEqual(self.select_rfds_lengths.pop(0), len(rfds))
return self.select_rfds_results.pop(0), [], []
+ def _make_mock_fork(self, pid):
+ def mock_fork():
+ return (pid, 12)
+ return mock_fork
+
+ def _mock_tcsetattr(self, fileno, opt, mode):
+ self.tcsetattr_mode_setting = mode
+
def test__copy_to_each(self):
"""Test the normal data case on both master_fd and stdin."""
read_from_stdout_fd, mock_stdout_fd = self._pipe()
socketpair[1].close()
os.close(write_to_stdin_fd)
- # Expect two select calls, the last one will cause IndexError
pty.select = self._mock_select
self.select_rfds_lengths.append(2)
self.select_rfds_results.append([mock_stdin_fd, masters[0]])
# both encountered an EOF before the second select call.
self.select_rfds_lengths.append(0)
- with self.assertRaises(IndexError):
- pty._copy(masters[0])
+ # We expect the function to return without error.
+ self.assertEqual(pty._copy(masters[0]), None)
+
+ def test__restore_tty_mode_normal_return(self):
+ """Test that spawn resets the tty mode no when _copy returns normally."""
+
+ # PID 1 is returned from mocked fork to run the parent branch
+ # of code
+ pty.fork = self._make_mock_fork(1)
+
+ status_sentinel = object()
+ pty.waitpid = lambda _1, _2: [None, status_sentinel]
+ pty.close = lambda _: None
+
+ pty._copy = lambda _1, _2, _3: None
+
+ mode_sentinel = object()
+ pty.tcgetattr = lambda fd: mode_sentinel
+ pty.tcsetattr = self._mock_tcsetattr
+ pty.setraw = lambda _: None
+
+ self.assertEqual(pty.spawn([]), status_sentinel, "pty.waitpid process status not returned by pty.spawn")
+ self.assertEqual(self.tcsetattr_mode_setting, mode_sentinel, "pty.tcsetattr not called with original mode value")
def tearDownModule():
reap_children()
+
if __name__ == "__main__":
unittest.main()
projects / thirdparty / Python / cpython.git / commitdiff
