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