test_direct_access.py 文件源码

def _run_ssh_pexpect(cmd, password, using_bashc=False):
    """
    Run a given command using pexpect.
    """
    logger.debug(u'{}SSH Command: {}{}'.format(Style.DIM, cmd,
                                               Style.RESET_ALL))
    if using_bashc:
        ssh_cli = pexpect.spawn('/bin/bash', ['-c', cmd])
    else:
        ssh_cli = pexpect.spawn(cmd)

    i = ssh_cli.expect(['[Pp]assword: ', '\(yes/no\)\? '])
    if i == 1:
        ssh_cli.sendline('yes')
        ssh_cli.expect('[Pp]assword: ')
    ssh_cli.sendline(password)
    time.sleep(1)

    ssh_cli.expect(['Connection to [0-9\.a-z]+ is closed.', pexpect.EOF,
                    pexpect.TIMEOUT], timeout=5)

    # Expected behavior is to get pexpect.EOF or closed connection, but due to
    # a bug in docker we have to send an additional new line or Ctrl^C
    out = str(ssh_cli.before) + str(ssh_cli.after)
    logger.debug(u'Output:\n {}{}{}'.format(Fore.YELLOW, out, Style.RESET_ALL))

    if ssh_cli.isalive():
        ssh_cli.close()

    return out