python类EOF的实例源码

iosxr_test.py 文件源码 项目:iosxrv-x64-vbox 作者: ios-xr 项目源码 文件源码 阅读 16 收藏 0 点赞 0 评论 0
def check_result(result, success_message):
    '''
    Function to check result of a pexpect operation.
    Accepts a success message.
    '''
    if result == 0:
        logger.debug('Test passed: %s' % success_message)
        return True
    elif result == 1:
        logger.warning('EOF - Test failed')
        return False
    elif result == 2:
        logger.warning(' Timed out - Test failed')
        return False
    else:
        logger.warning(' Generic - Test failed')
        return False
__init__.py 文件源码 项目:ssh-tunnel 作者: aalku 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def readline(self, size=-1):
        '''This reads and returns one entire line. The newline at the end of
        line is returned as part of the string, unless the file ends without a
        newline. An empty string is returned if EOF is encountered immediately.
        This looks for a newline as a CR/LF pair (\\r\\n) even on UNIX because
        this is what the pseudotty device returns. So contrary to what you may
        expect you will receive newlines as \\r\\n.

        If the size argument is 0 then an empty string is returned. In all
        other cases the size argument is ignored, which is not standard
        behavior for a file-like object. '''

        if size == 0:
            return self.string_type()
        # delimiter default is EOF
        index = self.expect([self.crlf, self.delimiter])
        if index == 0:
            return self.before + self.crlf
        else:
            return self.before
__init__.py 文件源码 项目:ssh-tunnel 作者: aalku 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def __init__(self, strings):

        '''This creates an instance of searcher_string. This argument 'strings'
        may be a list; a sequence of strings; or the EOF or TIMEOUT types. '''

        self.eof_index = -1
        self.timeout_index = -1
        self._strings = []
        for n, s in enumerate(strings):
            if s is EOF:
                self.eof_index = n
                continue
            if s is TIMEOUT:
                self.timeout_index = n
                continue
            self._strings.append((n, s))
__init__.py 文件源码 项目:ssh-tunnel 作者: aalku 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def __init__(self, patterns):

        '''This creates an instance that searches for 'patterns' Where
        'patterns' may be a list or other sequence of compiled regular
        expressions, or the EOF or TIMEOUT types.'''

        self.eof_index = -1
        self.timeout_index = -1
        self._searches = []
        for n, s in zip(list(range(len(patterns))), patterns):
            if s is EOF:
                self.eof_index = n
                continue
            if s is TIMEOUT:
                self.timeout_index = n
                continue
            self._searches.append((n, s))
__init__.py 文件源码 项目:ssh-tunnel 作者: aalku 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def __str__(self):

        '''This returns a human-readable string that represents the state of
        the object.'''

        #ss = [(n, '    %d: re.compile("%s")' %
        #    (n, repr(s.pattern))) for n, s in self._searches]
        ss = list()
        for n, s in self._searches:
            try:
                ss.append((n, '    %d: re.compile("%s")' % (n, s.pattern)))
            except UnicodeEncodeError:
                # for test cases that display __str__ of searches, dont throw
                # another exception just because stdout is ascii-only, using
                # repr()
                ss.append((n, '    %d: re.compile(%r)' % (n, s.pattern)))
        ss.append((-1, 'searcher_re:'))
        if self.eof_index >= 0:
            ss.append((self.eof_index, '    %d: EOF' % self.eof_index))
        if self.timeout_index >= 0:
            ss.append((self.timeout_index, '    %d: TIMEOUT' %
                self.timeout_index))
        ss.sort()
        ss = list(zip(*ss))[1]
        return '\n'.join(ss)
git_downloader.py 文件源码 项目:GithubSpider 作者: tommiu 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def goToLine(self, fh, linenumber):
        """
        Go to 'linenumber' of a huge text file in an (memory-)efficient way.
        """
        if linenumber < 1:
            raise IOError(
                "Specified linenumber '%d' is smaller than 1." % linenumber
                )

        fh.seek(0, os.SEEK_SET)

        # Skip lines until desired line is reached.
        for _ in range(0, linenumber - 1):
            read = fh.readline()
            if read == "":
                # Empty string represents EOF.
                raise OutOfScopeException(msg="goToLine error: ", 
                                          line=linenumber)
client.py 文件源码 项目:PenguinDome 作者: quantopian 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
def poll(self, timeout=None):
        if self.pending_data:
            return True
        if self.done:
            raise EOFError('Child process is done')
        if timeout is None:
            timeout = 0
        while True:
            try:
                self.child.expect(r'.', timeout=timeout)
            except pexpect.TIMEOUT:
                return True if self.pending_data else False
            except pexpect.EOF:
                self.pending_data += self.child.before
                self.done = True
                if self.pending_data:
                    return True
                raise EOFError('Child process is done')
            else:
                self.pending_data += self.child.before + self.child.after
ssh.py 文件源码 项目:omniduct 作者: airbnb 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def _execute(self, cmd, **kwargs):
        """
        Execute a command on a remote host.

        Parameters
        ----------
        cmd : string
            Command to be executed on remote host.
        kwargs : keywords
            Options to pass to subprocess.Popen.

        Returns
        -------
        proc : Popen subprocess
            Subprocess used to run remote command.
        """
        template = 'ssh {login} -T -o ControlPath={socket} << EOF\n{cmd}\nEOF'
        config = dict(self._subprocess_config)
        config.update(kwargs)
        return run_in_subprocess(template.format(login=self._login_info,
                                                 socket=self._socket_path,
                                                 cmd=cmd),
                                 check_output=True,
                                 **config)
test_forge.py 文件源码 项目:forge 作者: datawire 项目源码 文件源码 阅读 31 收藏 0 点赞 0 评论 0
def test_deploy():
    directory = mktree(FORGE_YAML + APP, MANGLE=MANGLE)
    os.environ["FORGE_PROFILE"] = "dev"
    forge = launch(directory, "forge deploy")
    forge.expect('built')
    forge.expect('forgetest/Dockerfile')
    forge.expect('pushed')
    forge.expect('forgetest-[0-9-]+:')
    forge.expect('rendered')
    forge.expect('service/forgetest-[0-9-]+')
    forge.expect('deployment/forgetest-[0-9-]+')
    forge.expect('deployed')
    forge.expect('forgetest-[0-9-]+')
    forge.expect(pexpect.EOF)
    assert forge.wait() == 0

    for sub in ("forgetest", "forgetest/subdir"):
        forge = launch(os.path.join(directory, "forgetest/subdir"), "forge deploy")
        forge.expect('rendered')
        forge.expect('service/forgetest-[0-9-]+')
        forge.expect('deployment/forgetest-[0-9-]+')
        forge.expect('deployed')
        forge.expect('forgetest-[0-9-]+')
        forge.expect(pexpect.EOF)
        assert forge.wait() == 0
test_forge.py 文件源码 项目:forge 作者: datawire 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def do_test_rebuilder(tree, path):
    directory = mktree(FORGE_YAML + tree, MANGLE=MANGLE)
    forge = launch(directory, "forge build containers")
    forge.expect(pexpect.EOF)
    assert forge.wait() == 0
    assert run_image(directory).strip() == "hello"

    with open(os.path.join(directory, path), "write") as f:
        f.write('print("goodbye")\n')

    forge = launch(directory, "forge build containers")
    forge.expect(pexpect.EOF)
    assert forge.wait() == 0
    assert run_image(directory).strip() == "goodbye"

    forge = launch(directory, "forge clean")
    forge.expect("docker kill ")
    forge.expect(pexpect.EOF)
    assert forge.wait() == 0
anita.py 文件源码 项目:anita 作者: gson1703 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def halt(self):
        self.login()
        self.child.send("halt\n")
        try:
            # Wait for text confirming the halt, or EOF
            self.child.expect("(The operating system has halted)|(entering state S5)", timeout = 60)
        except pexpect.EOF:
            # Didn't see the text but got an EOF; that's OK.
            print "EOF"
        except pexpect.TIMEOUT, e:
            # This is unexpected but mostly harmless
            print "timeout waiting for halt confirmation:", e

# Calling this directly is deprecated, use Anita.login()
test_commands.py 文件源码 项目:ravel 作者: ravel-net 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def testCommands(self):
        p = pexpect.spawn(self.ravelCmd)
        p.expect("ravel>")
        p.sendline("exit")
        p.expect(pexpect.EOF)
tunnel.py 文件源码 项目:zanph 作者: zanph 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def _try_passwordless_openssh(server, keyfile):
    """Try passwordless login with shell ssh command."""
    if pexpect is None:
        raise ImportError("pexpect unavailable, use paramiko")
    cmd = 'ssh -f '+ server
    if keyfile:
        cmd += ' -i ' + keyfile
    cmd += ' exit'

    # pop SSH_ASKPASS from env
    env = os.environ.copy()
    env.pop('SSH_ASKPASS', None)

    ssh_newkey = 'Are you sure you want to continue connecting'
    p = pexpect.spawn(cmd, env=env)
    while True:
        try:
            i = p.expect([ssh_newkey, _password_pat], timeout=.1)
            if i==0:
                raise SSHException('The authenticity of the host can\'t be established.')
        except pexpect.TIMEOUT:
            continue
        except pexpect.EOF:
            return True
        else:
            return False
test_cli.py 文件源码 项目:scm 作者: rookiebulls 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
def exit(self):
        """Send Ctrl + D to exit.
        """
        self.cli.sendcontrol('d')
        self.cli.expect(pexpect.EOF)
ssh.py 文件源码 项目:watchmen 作者: lycclsltt 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def run(self, cmd):
        ssh_newkey = 'Are you sure you want to continue connecting'
        child = pexpect.spawn('ssh -l %s -p %s %s %s'%(self._user, self._port, self._host, cmd))
        i = child.expect([pexpect.TIMEOUT, ssh_newkey, 'password: '])
        if i == 0: #timeout
            raise Exception('ssh timeout')
        if i == 1: #SSH does not have the public key. Just accept it.
            child.sendline('yes')
            child.expect('password: ')
            i = child.expect([pexpect.TIMEOUT, 'password: '])
            if i == 0: #timeout
                raise Exception('ssh timeout')
        child.sendline(self._pwd)
        child.expect(pexpect.EOF)
        return child.before
pxssh.py 文件源码 项目:watchmen 作者: lycclsltt 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def logout (self):
        '''Sends exit to the remote shell.

        If there are stopped jobs then this automatically sends exit twice.
        '''
        self.sendline("exit")
        index = self.expect([EOF, "(?i)there are stopped jobs"])
        if index==1:
            self.sendline("exit")
            self.expect(EOF)
        self.close()


问题


面经


文章

微信
公众号

扫码关注公众号