helper to run a script in a pseudoterminal workaround for opennet bsd bug issue 16762 avoid using defaultselector and pollselector kqueue does not work with pseudoterminals on os x 10 9 issue 20365 and open bsd issue 20667 poll does not work with os x 10 6 or 10 4 either issue 20472 hopefully the file descriptor is low enough to use with select linux raises eio when slave is closed issue 5380 apparently eio means the slave was closed workaround for open net bsd bug issue 16762 avoid using defaultselector and pollselector kqueue does not work with pseudo terminals on os x 10 9 issue 20365 and open bsd issue 20667 poll does not work with os x 10 6 or 10 4 either issue 20472 hopefully the file descriptor is low enough to use with select linux raises eio when slave is closed issue 5380 apparently eio means the slave was closed stop writing
import os
import selectors
import subprocess
import sys
from contextlib import ExitStack
from errno import EIO

from test.support.import_helper import import_module

def run_pty(script, input=b"dummy input\r", env=None):
    pty = import_module('pty')
    output = bytearray()
    [master, slave] = pty.openpty()
    args = (sys.executable, '-c', script)
    proc = subprocess.Popen(args, stdin=slave, stdout=slave, stderr=slave, env=env)
    os.close(slave)
    with ExitStack() as cleanup:
        cleanup.enter_context(proc)
        def terminate(proc):
            try:
                proc.terminate()
            except ProcessLookupError:
                
                pass
        cleanup.callback(terminate, proc)
        cleanup.callback(os.close, master)
        
        
        
        
        
        sel = cleanup.enter_context(selectors.SelectSelector())
        sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE)
        os.set_blocking(master, False)
        while True:
            for [_, events] in sel.select():
                if events & selectors.EVENT_READ:
                    try:
                        chunk = os.read(master, 0x10000)
                    except OSError as err:
                        
                        if err.errno != EIO:
                            raise
                        chunk = b""
                    if not chunk:
                        return output
                    output.extend(chunk)
                if events & selectors.EVENT_WRITE:
                    try:
                        input = input[os.write(master, input):]
                    except OSError as err:
                        
                        if err.errno != EIO:
                            raise
                        input = b""  
                    if not input:
                        sel.modify(master, selectors.EVENT_READ)