skip these tests if termios or fcntl are not available fcntl is a proxy for not being one of the wasm32 platforms even though we don t use this module a proper check for what crashes those is needed note that os read is nondeterministic so we need to be very careful to make the test suite deterministic a normal call to os read may give us less than expected beware on my linux system if i put foon into a terminal fd i get back foorn at the other end the behavior depends on the termios setting the newline translation may be osspecific to make the test suite deterministic and osindependent the functions readline and normalizeoutput can be used some operating systems do conversions on newline we could possibly fix that by doing the appropriate termios tcsetattrs i couldn t figure out the right combo on tru64 so just normalize the output and doc the problem oses by allowing certain combinations for some platforms but avoid allowing other differences like extra whitespace trailing garbage etc this is about the best we can do without getting some feedback from someone more knowledgable osf1 tru64 apparently turns n into rrn read one line may block forever if no newline is read reader io fileiofd mode rb closefdfalse return reader readline def expectedfailureifstdinisttyfun avoid isatty try tty tcgetattrpty stdinfileno return unittest expectedfailurefun except tty error pass return fun def writeallfd data written os writefd data if written lendata gh73256 gh110673 it should never happen but check just in case raise exceptionfshort write os writefd lendata bytes fwrote written bytes marginal testing of pty suite cannot do extensive do or fail testing because pty code is not too portable class ptytestunittest testcase def setupself oldsighup signal signalsignal sighup self handlesighup self addcleanupsignal signal signal sighup oldsighup save original stdin window size self stdindim none if havewinsz try self stdindim tty tcgetwinsizepty stdinfileno self addcleanuptty tcsetwinsize pty stdinfileno self stdindim except tty error pass staticmethod def handlesighupsignum frame pass expectedfailureifstdinistty def testopenptyself try mode tty tcgetattrpty stdinfileno except tty error not a tty or badclosed fd debugtty tcgetattrpty stdinfileno failed mode none newdim none if self stdindim try modify pty stdinfileno window size we need to check if pty openpty is able to set pty slave window size accordingly debugsetting pty stdinfileno window size debugforiginal size row col self stdindim targetdim self stdindim0 1 self stdindim1 1 debugftarget size row col targetdim tty tcsetwinsizepty stdinfileno targetdim were we able to set the window size of pty stdinfileno successfully newdim tty tcgetwinsizepty stdinfileno self assertequalnewdim targetdim pty stdinfileno window size unchanged except oserror warnings warnfailed to set pty stdinfileno window size pass try debugcalling pty openpty try masterfd slavefd slavename pty openptymode newdim true except typeerror masterfd slavefd pty openpty slavename none debugfgot masterfd slavefd slavename except oserror an optional feature could not be imported raise unittest skiptestpseudoterminals seemingly not functional closing masterfd can raise a sighup if the process is the session leader we installed a sighup signal handler to ignore this signal self addcleanupos close masterfd self addcleanupos close slavefd self asserttrueos isattyslavefd slavefd is not a tty if mode self assertequaltty tcgetattrslavefd mode openpty failed to set slave termios if newdim self assertequaltty tcgetwinsizeslavefd newdim openpty failed to set slave window size ensure the fd is nonblocking in case there s nothing to read blocking os getblockingmasterfd try os setblockingmasterfd false try s1 os readmasterfd 1024 self assertequalb s1 except oserror as e if e errno errno eagain raise finally restore the original flags os setblockingmasterfd blocking debugwriting to slavefd writeallslavefd teststring1 s1 readlinemasterfd self assertequalb i wish to buy a fish license n normalizeoutputs1 debugwriting chunked output writeallslavefd teststring2 5 writeallslavefd teststring25 s2 readlinemasterfd self assertequalb for my pet fish eric n normalizeoutputs2 def testforkself debugcalling pty fork pid masterfd pty fork self addcleanupos close masterfd if pid pty child stdout should be connected to a tty if not os isatty1 debugchild s fd 1 is not a tty os exit3 after pty fork the child should already be a session leader on those systems that have that concept debugin child calling os setsid try os setsid except oserror good we already were session leader debuggood oserror was raised pass except attributeerror have pty but not setsid debugno setsid available pass except we don t want this error to propagate escaping the call to os exit and causing very peculiar behavior in the calling regrtest py note could add traceback printing here debugan unexpected error was raised os exit1 else debugos setsid succeeded bad os exit2 os exit4 else debugwaiting for child d to finish pid in verbose mode we have to consume the debug output from the child or the child will block causing this test to hang in the parent s waitpid call the child blocks after a platformdependent amount of data is written to its fd on linux 2 6 it s 4000 bytes and the child won t block but on os x even the small writes in the child above will block it also on linux the read will raise an oserror inputoutput error when it tries to read past the end of the buffer but the child s already exited so catch and discard those exceptions it s not worth checking for eio while true try data os readmasterfd 80 except oserror break if not data break sys stdout writestrdata replaceb rn b n encoding ascii line os readmasterfd 80 lines line replace rn n split n if false and lines in child calling os setsid good oserror was raised raise testfailedunexpected output from child r line pid status os waitpidpid 0 res os waitstatustoexitcodestatus debugchild d exited with code d status d pid res status if res 1 self failchild raised an unexpected exception in os setsid elif res 2 self failpty fork failed to make child a session leader elif res 3 self failchild spawned by pty fork did not have a tty as stdout elif res 4 self failpty fork failed for unknown reasons debugreading from masterfd now that the child has exited try s1 os readmasterfd 1024 except oserror pass else raise testfailedread from masterfd did not raise exception def testmasterreadself xxxnnorwitz this test leaks fds when there is an error debugcalling pty openpty masterfd slavefd pty openpty debugfgot masterfd masterfd slavefd slavefd self addcleanupos close masterfd debugclosing slavefd os closeslavefd debugreading from masterfd try data os readmasterfd 1 except oserror linux data b self assertequaldata b def testspawndoesnthangself pty spawnsys executable c printhi there class smallptytestsunittest testcase this will raise indexerror when no more expected calls exist test the normal data case on both masterfd and stdin readfromstdoutfd mockstdoutfd self pipe pty stdoutfileno mockstdoutfd mockstdinfd writetostdinfd self pipe pty stdinfileno mockstdinfd socketpair self socketpair masters s fileno for s in socketpair feed data smaller than pipebuf these writes will not block writeallmasters1 b from master writeallwritetostdinfd b from stdin expect three select calls the last one will cause indexerror pty select self mockselect self selectinput appendmockstdinfd masters0 self selectoutput appendmockstdinfd masters0 self selectinput appendmockstdinfd masters0 mockstdoutfd masters0 self selectoutput append mockstdoutfd masters0 self selectinput appendmockstdinfd masters0 with self assertraisesindexerror pty copymasters0 test that the right data went to the right places rfds select selectreadfromstdoutfd masters1 00 self assertequalreadfromstdoutfd masters1 rfds self assertequalos readreadfromstdoutfd 20 b from master self assertequalos readmasters1 20 b from stdin def testrestorettymodenormalreturnself pid 1 is returned from mocked fork to run the parent branch of code skip these tests if termios or fcntl are not available fcntl is a proxy for not being one of the wasm32 platforms even though we don t use this module a proper check for what crashes those is needed readline note that os read is nondeterministic so we need to be very careful to make the test suite deterministic a normal call to os read may give us less than expected beware on my linux system if i put foo n into a terminal fd i get back foo r n at the other end the behavior depends on the termios setting the newline translation may be os specific to make the test suite deterministic and os independent the functions _readline and normalize_output can be used some operating systems do conversions on newline we could possibly fix that by doing the appropriate termios tcsetattr s i couldn t figure out the right combo on tru64 so just normalize the output and doc the problem o ses by allowing certain combinations for some platforms but avoid allowing other differences like extra whitespace trailing garbage etc this is about the best we can do without getting some feedback from someone more knowledgable osf 1 tru64 apparently turns n into r r n read one line may block forever if no newline is read avoid isatty gh 73256 gh 110673 it should never happen but check just in case marginal testing of pty suite cannot do extensive do or fail testing because pty code is not too portable save original stdin window size not a tty or bad closed fd modify pty stdin_fileno window size we need to check if pty openpty is able to set pty slave window size accordingly were we able to set the window size of pty stdin_fileno successfully an optional feature could not be imported closing master_fd can raise a sighup if the process is the session leader we installed a sighup signal handler to ignore this signal ensure the fd is non blocking in case there s nothing to read restore the original flags stdout should be connected to a tty after pty fork the child should already be a session leader on those systems that have that concept good we already were session leader have pty but not setsid we don t want this error to propagate escaping the call to os _exit and causing very peculiar behavior in the calling regrtest py note could add traceback printing here in verbose mode we have to consume the debug output from the child or the child will block causing this test to hang in the parent s waitpid call the child blocks after a platform dependent amount of data is written to its fd on linux 2 6 it s 4000 bytes and the child won t block but on os x even the small writes in the child above will block it also on linux the read will raise an oserror input output error when it tries to read past the end of the buffer but the child s already exited so catch and discard those exceptions it s not worth checking for eio line os read master_fd 80 lines line replace r n n split n if false and lines in child calling os setsid good oserror was raised raise testfailed unexpected output from child r line debug reading from master_fd now that the child has exited try s1 os read master_fd 1024 except oserror pass else raise testfailed read from master_fd did not raise exception xxx nnorwitz this test leaks fds when there is an error linux these tests don t spawn children or hang a list of file descriptors to close this will raise indexerror when no more expected calls exist test the normal data case on both master_fd and stdin feed data smaller than pipebuf these writes will not block expect three select calls the last one will cause indexerror test that the right data went to the right places 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
from test.support import verbose, reap_children
from test.support.import_helper import import_module


import_module('termios')


import_module("fcntl")

import errno
import os
import pty
import tty
import sys
import select
import signal
import socket
import io 
import unittest
import warnings

TEST_STRING_1 = b"I wish to buy a fish license.\n"
TEST_STRING_2 = b"For my pet fish, Eric.\n"

_HAVE_WINSZ = hasattr(tty, "TIOCGWINSZ") and hasattr(tty, "TIOCSWINSZ")

if verbose:
    def debug(msg):
        print(msg)
else:
    def debug(msg):
        pass












def normalize_output(data):
    
    
    
    
    
    

    
    

    
    if data.endswith(b'\r\r\n'):
        return data.replace(b'\r\r\n', b'\n')

    if data.endswith(b'\r\n'):
        return data.replace(b'\r\n', b'\n')

    return data

def _readline(fd):
    
    reader = io.FileIO(fd, mode='rb', closefd=False)
    return reader.readline()

def expectedFailureIfStdinIsTTY(fun):
    
    try:
        tty.tcgetattr(pty.STDIN_FILENO)
        return unittest.expectedFailure(fun)
    except tty.error:
        pass
    return fun


def write_all(fd, data):
    written = os.write(fd, data)
    if written != len(data):
        
        raise Exception(f"short write: os.write({fd}, {len(data)} bytes) "
                        f"wrote {written} bytes")




class PtyTest(unittest.TestCase):
    def setUp(self):
        old_sighup = signal.signal(signal.SIGHUP, self.handle_sighup)
        self.addCleanup(signal.signal, signal.SIGHUP, old_sighup)

        
        self.stdin_dim = None
        if _HAVE_WINSZ:
            try:
                self.stdin_dim = tty.tcgetwinsize(pty.STDIN_FILENO)
                self.addCleanup(tty.tcsetwinsize, pty.STDIN_FILENO,
                                self.stdin_dim)
            except tty.error:
                pass

    @staticmethod
    def handle_sighup(signum, frame):
        pass

    @expectedFailureIfStdinIsTTY
    def test_openpty(self):
        try:
            mode = tty.tcgetattr(pty.STDIN_FILENO)
        except tty.error:
            
            debug("tty.tcgetattr(pty.STDIN_FILENO) failed")
            mode = None

        new_dim = None
        if self.stdin_dim:
            try:
                
                
                
                debug("Setting pty.STDIN_FILENO window size.")
                debug(f"original size: (row, col) = {self.stdin_dim}")
                target_dim = (self.stdin_dim[0] + 1, self.stdin_dim[1] + 1)
                debug(f"target size: (row, col) = {target_dim}")
                tty.tcsetwinsize(pty.STDIN_FILENO, target_dim)

                
                
                new_dim = tty.tcgetwinsize(pty.STDIN_FILENO)
                self.assertEqual(new_dim, target_dim,
                                 "pty.STDIN_FILENO window size unchanged")
            except OSError:
                warnings.warn("Failed to set pty.STDIN_FILENO window size.")
                pass

        try:
            debug("Calling pty.openpty()")
            try:
                master_fd, slave_fd, slave_name = pty.openpty(mode, new_dim,
                                                              True)
            except TypeError:
                master_fd, slave_fd = pty.openpty()
                slave_name = None
            debug(f"Got {master_fd=}, {slave_fd=}, {slave_name=}")
        except OSError:
            
            raise unittest.SkipTest("Pseudo-terminals (seemingly) not functional.")

        
        
        
        self.addCleanup(os.close, master_fd)
        self.addCleanup(os.close, slave_fd)

        self.assertTrue(os.isatty(slave_fd), "slave_fd is not a tty")

        if mode:
            self.assertEqual(tty.tcgetattr(slave_fd), mode,
                             "openpty() failed to set slave termios")
        if new_dim:
            self.assertEqual(tty.tcgetwinsize(slave_fd), new_dim,
                             "openpty() failed to set slave window size")

        
        blocking = os.get_blocking(master_fd)
        try:
            os.set_blocking(master_fd, False)
            try:
                s1 = os.read(master_fd, 1024)
                self.assertEqual(b'', s1)
            except OSError as e:
                if e.errno != errno.EAGAIN:
                    raise
        finally:
            
            os.set_blocking(master_fd, blocking)

        debug("Writing to slave_fd")
        write_all(slave_fd, TEST_STRING_1)
        s1 = _readline(master_fd)
        self.assertEqual(b'I wish to buy a fish license.\n',
                         normalize_output(s1))

        debug("Writing chunked output")
        write_all(slave_fd, TEST_STRING_2[:5])
        write_all(slave_fd, TEST_STRING_2[5:])
        s2 = _readline(master_fd)
        self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))

    def test_fork(self):
        debug("calling pty.fork()")
        pid, master_fd = pty.fork()
        self.addCleanup(os.close, master_fd)
        if pid == pty.CHILD:
            
            if not os.isatty(1):
                debug("Child's fd 1 is not a tty?!")
                os._exit(3)

            
            
            debug("In child, calling os.setsid()")
            try:
                os.setsid()
            except OSError:
                
                debug("Good: OSError was raised.")
                pass
            except AttributeError:
                
                debug("No setsid() available?")
                pass
            except:
                
                
                
                
                debug("An unexpected error was raised.")
                os._exit(1)
            else:
                debug("os.setsid() succeeded! (bad!)")
                os._exit(2)
            os._exit(4)
        else:
            debug("Waiting for child (%d) to finish." % pid)
            
            
            
            
            
            
            
            
            
            
            while True:
                try:
                    data = os.read(master_fd, 80)
                except OSError:
                    break
                if not data:
                    break
                sys.stdout.write(str(data.replace(b'\r\n', b'\n'),
                                     encoding='ascii'))

            
            
            
            
            

            (pid, status) = os.waitpid(pid, 0)
            res = os.waitstatus_to_exitcode(status)
            debug("Child (%d) exited with code %d (status %d)." % (pid, res, status))
            if res == 1:
                self.fail("Child raised an unexpected exception in os.setsid()")
            elif res == 2:
                self.fail("pty.fork() failed to make child a session leader.")
            elif res == 3:
                self.fail("Child spawned by pty.fork() did not have a tty as stdout")
            elif res != 4:
                self.fail("pty.fork() failed for unknown reasons.")

            
            
            
            
            
            
            

    def test_master_read(self):
        
        debug("Calling pty.openpty()")
        master_fd, slave_fd = pty.openpty()
        debug(f"Got master_fd '{master_fd}', slave_fd '{slave_fd}'")

        self.addCleanup(os.close, master_fd)

        debug("Closing slave_fd")
        os.close(slave_fd)

        debug("Reading from master_fd")
        try:
            data = os.read(master_fd, 1)
        except OSError: 
            data = b""

        self.assertEqual(data, b"")

    def test_spawn_doesnt_hang(self):
        pty.spawn([sys.executable, '-c', 'print("hi there")'])

class SmallPtyTests(unittest.TestCase):
    

    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 = []  
        self.files = []
        self.select_input = []
        self.select_output = []
        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()
            except OSError:
                pass
        for fd in self.fds:
            try:
                os.close(fd)
            except OSError:
                pass

    def _pipe(self):
        pipe_fds = os.pipe()
        self.fds.extend(pipe_fds)
        return pipe_fds

    def _socketpair(self):
        socketpair = socket.socketpair()
        self.files.extend(socketpair)
        return socketpair

    def _mock_select(self, rfds, wfds, xfds):
        
        self.assertEqual((rfds, wfds, xfds), self.select_input.pop(0))
        return self.select_output.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):
        
        read_from_stdout_fd, mock_stdout_fd = self._pipe()
        pty.STDOUT_FILENO = mock_stdout_fd
        mock_stdin_fd, write_to_stdin_fd = self._pipe()
        pty.STDIN_FILENO = mock_stdin_fd
        socketpair = self._socketpair()
        masters = [s.fileno() for s in socketpair]

        
        write_all(masters[1], b'from master')
        write_all(write_to_stdin_fd, b'from stdin')

        
        pty.select = self._mock_select
        self.select_input.append(([mock_stdin_fd, masters[0]], [], []))
        self.select_output.append(([mock_stdin_fd, masters[0]], [], []))
        self.select_input.append(([mock_stdin_fd, masters[0]], [mock_stdout_fd, masters[0]], []))
        self.select_output.append(([], [mock_stdout_fd, masters[0]], []))
        self.select_input.append(([mock_stdin_fd, masters[0]], [], []))

        with self.assertRaises(IndexError):
            pty._copy(masters[0])

        
        rfds = select.select([read_from_stdout_fd, masters[1]], [], [], 0)[0]
        self.assertEqual([read_from_stdout_fd, masters[1]], rfds)
        self.assertEqual(os.read(read_from_stdout_fd, 20), b'from master')
        self.assertEqual(os.read(masters[1], 20), b'from stdin')

    def test__restore_tty_mode_normal_return(self):
        

        
        
        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()