def posix_shell()

in pipe-cli/src/utilities/pipe_shell.py [0:0]

• 30 lines of code
• 24 McCabe index (conditional complexity)

def posix_shell(channel, is_interactive=True):
    # get the current TTY attributes to reapply after
    # the remote shell is closed
    oldtty_attrs = termios.tcgetattr(sys.stdin) if is_interactive else None
    stdout_encoding = sys.stdout.encoding if sys.stdout.encoding else "UTF-8"

    # wrap the whole thing in a try/finally construct to ensure
    # that exiting code for TTY handling runs
    try:
        if is_interactive:
            tty.setraw(sys.stdin.fileno())
            tty.setcbreak(sys.stdin.fileno())

        channel.settimeout(0.0)

        is_alive = True

        while is_alive:
            if is_interactive:
                # resize on every iteration of the main loop
                resize_pty(channel)

            # use a unix select call to wait until the remote shell
            # and stdin are ready for reading
            # this is the block until data is ready
            select_targets = [channel, sys.stdin] if is_interactive else [channel]
            read_ready, write_ready, exception_list = \
                    select.select(select_targets, [], [])

            # if the channel is one of the ready objects, print
            # it out 1024 chars at a time
            if channel in read_ready:
                # try to do a read from the remote end and print to screen
                try:
                    is_alive = transmit_to_std_out(channel, encoding=stdout_encoding)

                # do nothing on a timeout, as this is an ordinary condition
                except socket.timeout:
                    pass
            
            # if stdin is ready for reading
            if is_interactive and sys.stdin in read_ready and is_alive:
                # send a single character out at a time
                # this is typically human input, so sending it one character at
                # a time is the only correct action we can take

                # use an os.read to prevent nasty buffering problem with shell
                # history
                char = os.read(sys.stdin.fileno(), 1)

                # if this side of the connection closes, shut down gracefully
                if len(char) == 0:
                    is_alive = False
                else:
                    channel.send(char)

        # close down the channel for send/recv
        # this is an explicit call most likely redundant with the operations
        # that caused an exit from the REPL, but unusual exit conditions can
        # cause this to be reached uncalled
        channel.shutdown(2)

    # regardless of errors, restore the TTY to working order
    # upon exit and print that connection is closed
    finally:
        if is_interactive:
            termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, oldtty_attrs)