| 12 | |
| 13 | |
| 14 | def run_pty(script, input=b"dummy input\r", env=None): |
| 15 | pty = import_module('pty') |
| 16 | output = bytearray() |
| 17 | [master, slave] = pty.openpty() |
| 18 | args = (sys.executable, '-c', script) |
| 19 | proc = subprocess.Popen(args, stdin=slave, stdout=slave, stderr=slave, env=env) |
| 20 | os.close(slave) |
| 21 | with ExitStack() as cleanup: |
| 22 | cleanup.enter_context(proc) |
| 23 | def terminate(proc): |
| 24 | try: |
| 25 | proc.terminate() |
| 26 | except ProcessLookupError: |
| 27 | # Workaround for Open/Net BSD bug (Issue 16762) |
| 28 | pass |
| 29 | cleanup.callback(terminate, proc) |
| 30 | cleanup.callback(os.close, master) |
| 31 | # Avoid using DefaultSelector and PollSelector. Kqueue() does not |
| 32 | # work with pseudo-terminals on OS X < 10.9 (Issue 20365) and Open |
| 33 | # BSD (Issue 20667). Poll() does not work with OS X 10.6 or 10.4 |
| 34 | # either (Issue 20472). Hopefully the file descriptor is low enough |
| 35 | # to use with select(). |
| 36 | sel = cleanup.enter_context(selectors.SelectSelector()) |
| 37 | sel.register(master, selectors.EVENT_READ | selectors.EVENT_WRITE) |
| 38 | os.set_blocking(master, False) |
| 39 | while True: |
| 40 | for [_, events] in sel.select(): |
| 41 | if events & selectors.EVENT_READ: |
| 42 | try: |
| 43 | chunk = os.read(master, 0x10000) |
| 44 | except OSError as err: |
| 45 | # Linux raises EIO when slave is closed (Issue 5380) |
| 46 | if err.errno != EIO: |
| 47 | raise |
| 48 | chunk = b"" |
| 49 | if not chunk: |
| 50 | return output |
| 51 | output.extend(chunk) |
| 52 | if events & selectors.EVENT_WRITE: |
| 53 | try: |
| 54 | input = input[os.write(master, input):] |
| 55 | except OSError as err: |
| 56 | # Apparently EIO means the slave was closed |
| 57 | if err.errno != EIO: |
| 58 | raise |
| 59 | input = b"" # Stop writing |
| 60 | if not input: |
| 61 | sel.modify(master, selectors.EVENT_READ) |
| 62 | |
| 63 | |
| 64 | ###################################################################### |