我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用os.openpty()。
def spawn(self, command, cwd, env): """Starts the process Spawn a new process and register the listeners on it Arguments: command {list} -- command list for the process (ex: ['ls', '-la']) """ child_env = os.environ.copy() child_env.update(env if env is not None else {}) child_env.update({ "TERM":"sublimeterm", "COLUMNS":"40", "INPUTRC":"$(pwd)/inputrc" }) self.master, self.slave = os.openpty() self.process = subprocess.Popen(command, stdin=self.slave, stdout=self.slave, stderr=self.slave, preexec_fn=os.setsid, cwd=cwd, env=child_env)
def _open_terminal(): """Open pty master and return (master_fd, tty_name). SGI and generic BSD version, for when openpty() fails.""" try: import sgi except ImportError: pass else: try: tty_name, master_fd = sgi._getpty(os.O_RDWR, 0666, 0) except IOError, msg: raise os.error, msg return master_fd, tty_name for x in 'pqrstuvwxyzPQRST': for y in '0123456789abcdef': pty_name = '/dev/pty' + x + y try: fd = os.open(pty_name, os.O_RDWR) except os.error: continue return (fd, '/dev/tty' + x + y) raise os.error, 'out of pty devices'
def slave_open(tty_name): """slave_open(tty_name) -> slave_fd Open the pty slave and acquire the controlling terminal, returning opened filedescriptor. Deprecated, use openpty() instead.""" result = os.open(tty_name, os.O_RDWR) try: from fcntl import ioctl, I_PUSH except ImportError: return result try: ioctl(result, I_PUSH, "ptem") ioctl(result, I_PUSH, "ldterm") except IOError: pass return result
def tty_create_child(*args): master_fd, slave_fd = os.openpty() disable_echo(master_fd) disable_echo(slave_fd) pid = os.fork() if not pid: mitogen.core.set_block(slave_fd) os.dup2(slave_fd, 0) os.dup2(slave_fd, 1) os.dup2(slave_fd, 2) close_nonstandard_fds() os.setsid() os.close(os.open(os.ttyname(1), os.O_RDWR)) os.execvp(args[0], args) os.close(slave_fd) LOG.debug('tty_create_child() child %d fd %d, parent %d, cmd: %s', pid, master_fd, os.getpid(), Argv(args)) return pid, master_fd
def _open_terminal(): """Open pty master and return (master_fd, tty_name). SGI and generic BSD version, for when openpty() fails.""" try: import sgi except ImportError: pass else: try: tty_name, master_fd = sgi._getpty(os.O_RDWR, 0o666, 0) except IOError as msg: raise os.error(msg) return master_fd, tty_name for x in 'pqrstuvwxyzPQRST': for y in '0123456789abcdef': pty_name = '/dev/pty' + x + y try: fd = os.open(pty_name, os.O_RDWR) except os.error: continue return (fd, '/dev/tty' + x + y) raise os.error('out of pty devices')
def slave_open(tty_name): """slave_open(tty_name) -> slave_fd Open the pty slave and acquire the controlling terminal, returning opened filedescriptor. Deprecated, use openpty() instead.""" result = os.open(tty_name, os.O_RDWR) try: from fcntl import ioctl, I_PUSH except ImportError: return result try: ioctl(result, I_PUSH, "ptem") ioctl(result, I_PUSH, "ldterm") except OSError: pass return result
def launch_proc_with_pty(args, stdin=None, stdout=None, stderr=None, echo=True): """Similar to pty.fork, but handle stdin/stdout according to parameters instead of connecting to the pty :return tuple (subprocess.Popen, pty_master) """ def set_ctty(ctty_fd, master_fd): '''Set controlling terminal''' os.setsid() os.close(master_fd) fcntl.ioctl(ctty_fd, termios.TIOCSCTTY, 0) if not echo: termios_p = termios.tcgetattr(ctty_fd) # termios_p.c_lflags termios_p[3] &= ~termios.ECHO termios.tcsetattr(ctty_fd, termios.TCSANOW, termios_p) (pty_master, pty_slave) = os.openpty() p = subprocess.Popen(args, stdin=stdin, stdout=stdout, stderr=stderr, preexec_fn=lambda: set_ctty(pty_slave, pty_master)) os.close(pty_slave) return p, open(pty_master, 'wb+', buffering=0)
def __init__(self, args, *, size=core.STANDARD_SIZE, cmd=None, env=None, callbacks_cls=PtyCallbacks, __os_close=os.close, **kwargs): self._master_fd = -1 super().__init__(size) master_fd, slave_fd = os.openpty() try: self._slave_name = os.ttyname(slave_fd) _spawn(cmd, args, env, fds={0: slave_fd, 1: slave_fd, 2: slave_fd}, tty=slave_fd) self._master_fd = master_fd master_fd = -1 finally: __os_close(slave_fd) if master_fd != -1: __os_close(master_fd) # configure some sensible defaults callbacks = callbacks_cls(self, **kwargs) self.set_callbacks(callbacks) self.callbacks = callbacks
def openpty(): """openpty() -> (master_fd, slave_fd) Open a pty master/slave pair, using os.openpty() if possible.""" try: return os.openpty() except (AttributeError, OSError): pass master_fd, slave_name = _open_terminal() slave_fd = slave_open(slave_name) return master_fd, slave_fd
def master_open(): """master_open() -> (master_fd, slave_name) Open a pty master and return the fd, and the filename of the slave end. Deprecated, use openpty() instead.""" try: master_fd, slave_fd = os.openpty() except (AttributeError, OSError): pass else: slave_name = os.ttyname(slave_fd) os.close(slave_fd) return master_fd, slave_name return _open_terminal()
def fork(): """fork() -> (pid, master_fd) Fork and make the child a session leader with a controlling terminal.""" try: pid, fd = os.forkpty() except (AttributeError, OSError): pass else: if pid == CHILD: try: os.setsid() except OSError: # os.forkpty() already set us session leader pass return pid, fd master_fd, slave_fd = openpty() pid = os.fork() if pid == CHILD: # Establish a new session. os.setsid() os.close(master_fd) # Slave becomes stdin/stdout/stderr of child. 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) # Explicitly open the tty to make it become a controlling tty. tmp_fd = os.open(os.ttyname(STDOUT_FILENO), os.O_RDWR) os.close(tmp_fd) else: os.close(slave_fd) # Parent and child process. return pid, master_fd
def test_read_pty_output(self): proto = MyReadPipeProto(loop=self.loop) master, slave = os.openpty() master_read_obj = io.open(master, 'rb', 0) @asyncio.coroutine def connect(): t, p = yield from self.loop.connect_read_pipe(lambda: proto, master_read_obj) self.assertIs(p, proto) self.assertIs(t, proto.transport) self.assertEqual(['INITIAL', 'CONNECTED'], proto.state) self.assertEqual(0, proto.nbytes) self.loop.run_until_complete(connect()) os.write(slave, b'1') test_utils.run_until(self.loop, lambda: proto.nbytes) self.assertEqual(1, proto.nbytes) os.write(slave, b'2345') test_utils.run_until(self.loop, lambda: proto.nbytes >= 5) self.assertEqual(['INITIAL', 'CONNECTED'], proto.state) self.assertEqual(5, proto.nbytes) os.close(slave) self.loop.run_until_complete(proto.done) self.assertEqual( ['INITIAL', 'CONNECTED', 'EOF', 'CLOSED'], proto.state) # extra info is available self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
def test_write_pty(self): master, slave = os.openpty() slave_write_obj = io.open(slave, 'wb', 0) proto = MyWritePipeProto(loop=self.loop) connect = self.loop.connect_write_pipe(lambda: proto, slave_write_obj) transport, p = self.loop.run_until_complete(connect) self.assertIs(p, proto) self.assertIs(transport, proto.transport) self.assertEqual('CONNECTED', proto.state) transport.write(b'1') data = bytearray() def reader(data): chunk = os.read(master, 1024) data += chunk return len(data) test_utils.run_until(self.loop, lambda: reader(data) >= 1, timeout=10) self.assertEqual(b'1', data) transport.write(b'2345') test_utils.run_until(self.loop, lambda: reader(data) >= 5, timeout=10) self.assertEqual(b'12345', data) self.assertEqual('CONNECTED', proto.state) os.close(master) # extra info is available self.assertIsNotNone(proto.transport.get_extra_info('pipe')) # close connection proto.transport.close() self.loop.run_until_complete(proto.done) self.assertEqual('CLOSED', proto.state)
def fork_pty(): '''This implements a substitute for the forkpty system call. This should be more portable than the pty.fork() function. Specifically, this should work on Solaris. Modified 10.06.05 by Geoff Marshall: Implemented __fork_pty() method to resolve the issue with Python's pty.fork() not supporting Solaris, particularly ssh. Based on patch to posixmodule.c authored by Noah Spurrier:: http://mail.python.org/pipermail/python-dev/2003-May/035281.html ''' parent_fd, child_fd = os.openpty() if parent_fd < 0 or child_fd < 0: raise OSError("os.openpty() failed") pid = os.fork() if pid == CHILD: # Child. os.close(parent_fd) pty_make_controlling_tty(child_fd) os.dup2(child_fd, STDIN_FILENO) os.dup2(child_fd, STDOUT_FILENO) os.dup2(child_fd, STDERR_FILENO) else: # Parent. os.close(child_fd) return pid, parent_fd
def test(self): master, slave = os.openpty() if not os.isatty(slave): self.fail("Slave-end of pty is not a terminal.") os.write(slave, b'Ping!') self.assertEqual(os.read(master, 1024), b'Ping!')
def __fork_pty(self): """This implements a substitute for the forkpty system call. This should be more portable than the pty.fork() function. Specifically, this should work on Solaris. Modified 10.06.05 by Geoff Marshall: Implemented __fork_pty() method to resolve the issue with Python's pty.fork() not supporting Solaris, particularly ssh. Based on patch to posixmodule.c authored by Noah Spurrier:: http://mail.python.org/pipermail/python-dev/2003-May/035281.html """ parent_fd, child_fd = os.openpty() if parent_fd < 0 or child_fd < 0: raise ExceptionPexpect, "Error! Could not open pty with os.openpty()." pid = os.fork() if pid < 0: raise ExceptionPexpect, "Error! Failed os.fork()." elif pid == 0: # Child. os.close(parent_fd) self.__pty_make_controlling_tty(child_fd) os.dup2(child_fd, 0) os.dup2(child_fd, 1) os.dup2(child_fd, 2) if child_fd > 2: os.close(child_fd) else: # Parent. os.close(child_fd) return pid, parent_fd
def test(self): master, slave = os.openpty() self.addCleanup(os.close, master) self.addCleanup(os.close, slave) if not os.isatty(slave): self.fail("Slave-end of pty is not a terminal.") os.write(slave, 'Ping!') self.assertEqual(os.read(master, 1024), 'Ping!')
def test(self): master, slave = os.openpty() self.addCleanup(os.close, master) self.addCleanup(os.close, slave) if not os.isatty(slave): self.fail("Slave-end of pty is not a terminal.") os.write(slave, b'Ping!') self.assertEqual(os.read(master, 1024), b'Ping!')
def __fork_pty(self): '''This implements a substitute for the forkpty system call. This should be more portable than the pty.fork() function. Specifically, this should work on Solaris. Modified 10.06.05 by Geoff Marshall: Implemented __fork_pty() method to resolve the issue with Python's pty.fork() not supporting Solaris, particularly ssh. Based on patch to posixmodule.c authored by Noah Spurrier:: http://mail.python.org/pipermail/python-dev/2003-May/035281.html ''' parent_fd, child_fd = os.openpty() if parent_fd < 0 or child_fd < 0: raise ExceptionPexpect("Could not open with os.openpty().") pid = os.fork() if pid == pty.CHILD: # Child. os.close(parent_fd) self.__pty_make_controlling_tty(child_fd) os.dup2(child_fd, self.STDIN_FILENO) os.dup2(child_fd, self.STDOUT_FILENO) os.dup2(child_fd, self.STDERR_FILENO) else: # Parent. os.close(child_fd) return pid, parent_fd
def _open_pty(self): """Create a PTY """ # get our terminal params self.tcattr = termios.tcgetattr(STDIN) winsize = fcntl.ioctl(STDIN, termios.TIOCGWINSZ, '0123') # open a pty self.master, self.slave = os.openpty() # set the slave's terminal params termios.tcsetattr(self.slave, termios.TCSANOW, self.tcattr) fcntl.ioctl(self.slave, termios.TIOCSWINSZ, winsize)
def test_openpty(self): master_fd, slave_fd = os.openpty() self.addCleanup(os.close, master_fd) self.addCleanup(os.close, slave_fd) self.assertEqual(os.get_inheritable(master_fd), False) self.assertEqual(os.get_inheritable(slave_fd), False)
