Python expect.Expect类代码示例

    def list(connection):
        '''
        list repositories
        '''
        RHUIManager.screen(connection, "repo")
        Expect.enter(connection, "l")
        # eating prompt!!
        pattern = re.compile('l\r\n(.*)\r\n-+\r\nrhui\s* \(repo\)\s* =>',
                re.DOTALL)
        ret = Expect.match(connection, pattern, grouplist=[1])[0]
        print ret
        reslist = map(lambda x: x.strip(), ret.split("\r\n"))
        print reslist
        repolist = []
        for line in reslist:
            # Readling lines and searching for repos
            if line == '':
                continue
            if "Custom Repositories" in line:
                continue
            if "Red Hat Repositories" in line:
                continue
            if "No repositories are currently managed by the RHUI" in line:
                continue
            repolist.append(line)

        Expect.enter(connection, 'q')
        return repolist

 def delete_cdses(connection, *cdses):
     '''
     unregister (delete) CDS instance from the RHUI
     '''
     RHUIManager.screen(connection, "cds")
     Expect.enter(connection, "d")
     RHUIManager.select_items(connection, *cdses)
     RHUIManager.quit(connection, timeout=30)

    def logout(connection, prefix=""):
        '''
        Logout from rhui-manager

        Use @param prefix to specify something to expect before exiting
        '''
        Expect.expect(connection, prefix + ".*rhui \(.*\) =>")
        Expect.enter(connection, "logout")

 def list_lines(connection, prompt='', enter_l=True):
     '''
     list items on screen returning a list of lines seen
     eats prompt!!!
     '''
     if enter_l:
         Expect.enter(connection, "l")
     match = Expect.match(connection, re.compile("(.*)" + prompt, re.DOTALL))
     return match[0].split('\r\n')

 def delete_repo(connection, repolist):
     '''
     delete a repository from the RHUI
     '''
     RHUIManager.screen(connection, "repo")
     Expect.enter(connection, "d")
     RHUIManager.select(connection, repolist)
     RHUIManager.proceed_with_check(connection, "The following repositories will be deleted:", repolist, ["Red Hat Repositories", "Custom Repositories"])
     RHUIManager.quit(connection)

    def quit(connection, prefix="", timeout=10):
        '''
        Quit from rhui-manager

        Use @param prefix to specify something to expect before exiting
        Use @param timeout to specify the timeout
        '''
        Expect.expect(connection, prefix + ".*rhui \(.*\) =>", timeout)
        Expect.enter(connection, "q")

 def sync_cds(connection, cdslist):
     '''
     sync an individual CDS immediately
     '''
     RHUIManager.screen(connection, "sync")
     Expect.enter(connection, "sc")
     RHUIManager.select(connection, cdslist)
     RHUIManager.proceed_with_check(connection, "The following CDS instances will be scheduled for synchronization:", cdslist)
     RHUIManager.quit(connection)

 def sync_cluster(connection, clusterlist):
     '''
     sync a CDS cluster immediately
     '''
     RHUIManager.screen(connection, "sync")
     Expect.enter(connection, "sl")
     RHUIManager.select(connection, clusterlist)
     RHUIManager.proceed_with_check(connection, "The following CDS clusters will be scheduled for synchronization:", clusterlist)
     RHUIManager.quit(connection)

 def delete_repo(connection, repolist):
     '''
     delete a repository from the RHUI
     '''
     RHUIManager.screen(connection, "repo")
     Expect.enter(connection, "d")
     RHUIManager.select(connection, repolist)
     RHUIManager.proceed_without_check(connection)
     Expect.expect(connection, ".*rhui \(" + "repo" + "\) =>")

    def check_for_package(connection, reponame, package):
        '''
        list packages in a repository
        '''
        RHUIManager.screen(connection, "repo")
        Expect.enter(connection, "p")
        RHUIManager.select_one(connection, reponame)
        Expect.expect(connection, "\(blank line for no filter\):")
        Expect.enter(connection, package)

        pattern = re.compile('.*only\.\r\n(.*)\r\n-+\r\nrhui\s* \(repo\)\s* =>',
                             re.DOTALL)
        ret = Expect.match(connection, pattern, grouplist=[1])[0]
        reslist = map(lambda x: x.strip(), ret.split("\r\n"))
        print reslist
        packagelist = []
        for line in reslist:
            if line == '':
                continue
            if line == 'Packages:':
                continue
            if line == 'No packages found that match the given filter.':
                continue
            if line == 'No packages in the repository.':
                continue
            packagelist.append(line)

        Expect.enter(connection, 'q')
        return packagelist

 def list(connection):
     '''
     return the list of currently managed CDSes
     '''
     RHUIManager.screen(connection, "cds")
     # eating prompt!!
     lines = RHUIManager.list_lines(connection, prompt=RHUIManagerCds.prompt)
     ret = Cds.parse(lines)
     # custom quitting; have eaten the prompt
     Expect.enter(connection, 'q')
     return [cds for _, cds in ret]

 def install_rpm_from_rhua(rhua_connection, connection, rpmpath):
     '''
     Transfer RPM package from RHUA host to the instance and install it
     @param rpmpath: path to RPM package on RHUA node
     '''
     tfile = tempfile.NamedTemporaryFile(delete=False)
     tfile.close()
     rhua_connection.sftp.get(rpmpath, tfile.name)
     connection.sftp.put(tfile.name, tfile.name + ".rpm")
     os.unlink(tfile.name)
     Expect.ping_pong(connection, "rpm -i " + tfile.name + ".rpm" + " && echo SUCCESS", "[^ ]SUCCESS", 60)

 def _sync_cds(self, cdslist):
     """ Sync cds """
     if (not "RHUA" in self.rs.Instances.keys()) or len(self.rs.Instances["RHUA"]) < 1:
         raise nose.exc.SkipTest("can't test without RHUA!")
     try:
         RHUIManagerSync.sync_cds(self.rs.Instances["RHUA"][0], cdslist)
     except ExpectFailed:
         # The CDS is not available for syncing so most probably it's syncing right now
         # Trying to check the status
         Expect.enter(self.rs.Instances["RHUA"][0], "b")
         RHUIManager.quit(self.rs.Instances["RHUA"][0])
     self._sync_wait_cds(cdslist)

 def command_output(connection, command, pattern_tuple, username="admin",
         password="admin"):
     """return output of a command based on pattern_tuple provided. Output
     is split to lines"""
     Expect.enter(connection, "pulp-admin -u %s -p %s %s" % \
             (username, password, command))
     # eats prompt!
     pattern, group_index = pattern_tuple
     ret = Expect.match(connection, pattern, grouplist=[group_index])[0]
     # reset prompt
     Expect.enter(connection, "")
     return ret.split('\r\n')

 def screen(connection, screen_name):
     '''
     Open specified rhui-manager screen
     '''
     if screen_name in ["repo", "cds", "loadbalancers", "sync", "identity", "users"]:
         key = screen_name[:1]
     elif screen_name == "client":
         key = "e"
     elif screen_name == "entitlements":
         key = "n"
     Expect.enter(connection, key)
     Expect.expect(connection, "rhui \(" + screen_name + "\) =>")