[雪峰磁针石博客]python3快速入门教程11命令行自动化工具

[雪峰磁针石博客]python3快速入门教程

命令行自动化与pexpect

简介

Pexpect 是一个用来启动交互式命令行子程序并对其进行自动控制的 Python 工具模块。 Pexpect 可以用来和像 ssh、ftp、passwd、telnet 等命令行程序进行自动交互。可广泛用于自动化运维和测试，实现同时控制多台设备和自动化。Linux中的知名装包软件就使用了Pexpect。 Pexpect在IBM，alibaba，google等公司有广泛使用，在https://pypi.python.org/pypi/pexpect 的日下载量一万左右。

纯python实现，依赖pty模块(不支持Windows)。

最新英文版文档参见：http://pexpect.readthedocs.org/en/latest/

安装：

版本要求：Python 2.6、3.2 或以上

快速入门

ssh登录是常用的操作，过程如下：

# ssh test@172.17.100.18
test@172.17.100.18's password: 
Last login: Tue Mar 15 17:53:01 2016 from 172.17.100.19

下面我们用pexpect来自动实现这个过程：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pexpect
child = pexpect.spawn('ssh test@172.17.100.18')
child.expect('password: ') # 等待password:字符出现
print(child.before + child.after) # 输出password:前后的字符
child.sendline('123456')    # 发送密码
child.expect(']\$')  # 等待]$字符出现
print(child.before + child.after)
child.interact() # 把ssh的连接交给用户控制。

上面最后一句在非交互式的情况下要关闭连接，用child.close()替换即可。

上 述操作除了interact外，都可以用python的ssh模块：paramiko代替。不过对于一些同时支持telenet、ftp、ssh等协议命 令行的通信设备，可以用pexpect通杀。telenet、ftp、ssh等从协议的层次联系，pexpect会更接近用户使用，更加适合自动化测试。

两个重要方法： expect()和send() (以及sendline() )。expect可以接受正则表达式作为参数。

before包含预期字符串之前的信息， after包含匹配模式及其以后的内容。

批量操作多台服务器

• 功能：实现同时对多台linux服务器通过ssh执行同一命令。

• 技术基础: python pexpect，不支持windows。

• 参数：

  • 固定参数pwd：远程服务器密码，用户名目前写死是root，可自行修改。

  • 可选参数-c CMDS：要执行的命令，比如："ls -l","cd /home/test && test.py&如果不选择，会从当前目前的cmd.txt读取。

  • 可选参数-s SERVERS：目标服务器，比如192.168.0.1,最后一位数可以用-表示一个区间，分号用于分割不同的ip。如果不选择，会从当前目前的ip.txt读取。

库文件：common.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# Author:         Rongzhong Xu
# CreateDate: 2014-05-06

import os

import pexpect


class Ssh(object):

    client = None

    @classmethod
    def connect(cls, ip, username="root", password="123456", prompt=']#',
                silent=False):

        # Ssh to remote server
        ssh_newkey = 'Are you sure you want to continue connecting'
        child = pexpect.spawn('ssh ' + username + '@' + ip, maxread=5000)

        i = 1
        # Enter password
        while i != 0:
            i = child.expect([prompt, 'assword:*', ssh_newkey, pexpect.TIMEOUT,
                              'key.*? failed'])
            if not silent:
                print(child.before, child.after)
            if i == 0:  # find prompt
                pass
            elif i == 1:  # Enter password
                child.send(password + "\r")
            if i == 2:  # SSH does not have the public key. Just accept it.
                child.sendline('yes\r')
            if i == 3:  # Timeout
                raise Exception('ERROR TIMEOUT! SSH could not login. ')
            if i == 4:  # new key
                print(child.before, child.after)
                os.remove(os.path.expanduser('~') + '/.ssh/known_hosts')

        Ssh.client = child

    @classmethod
    def command(cls, cmd, prompt=']#', silent=False):
        Ssh.client.buffer = ''
        Ssh.client.send(cmd + "\r")
        # Ssh.client.setwinsize(400,400)
        Ssh.client.expect(prompt)
        if not silent:
            print(Ssh.client.before, Ssh.client.after)
        return Ssh.client.before, Ssh.client.after

    def close(cls,):
        Ssh.client.close()

主脚本batch.py ：

#!/usr/bin/python
# -*- coding: utf-8 -*-
# Author:         Rongzhong Xu
# CreateDate: 2014-05-06
import argparse
import common
import argparse

parser = argparse.ArgumentParser()
parser.add_argument('pwd', action="store",help=u'password')
parser.add_argument('-c', dest='cmds', action="store", help=u'command')
parser.add_argument('-s', dest='servers', action="store", help=u'hosts')
parser.add_argument('--version', action='version',
    version='%(prog)s 1.1 Rongzhong xu 2014 05 08')

options = parser.parse_args()
servers = []
if options.servers:
    raw_server = options.servers.split(';') # ips' decollator is  semicolon
    for server in raw_server:
        if '-' in server: # - means continuous ips
            server_list = server.split('.')
            base = '.'.join(server_list[:3])
            indices = server_list[-1].split('-')
            start, end = indices

            for item in range(int(start),int(end)+1):
                servers.append('{0}.{1}'.format(base,item))
        else:
            servers.append(server)
else:
    for item in open('ip.txt'):
        servers.append(item)

cmds = []
if options.cmds:
    cmds.append(options.cmds)
else:
    for item in open('cmd.txt'):
        servers.append(item)

for host in servers:
    print()
    print("*"*80)
    print("\nConnect to host: {0}".format(host))
    c = common.Ssh()
    c.connect(host,'root',options.pwd)
    for item in cmds:
        c.command(item)
    c.close()

执行演示：

# ./batch.py -husage: batch.py [-h] [-c CMDS] [-s SERVERS] [--version] pwdpositional arguments:  pwd         password
 
optional arguments:
  -h, --help  show this help message and exit
  -c CMDS     command
  -s SERVERS  hosts
  --version   show program's version number and exit
 
# ./batch.py password -s "192.168.0.71-76;123.1.149.26" -c "cat /etc/redhat-release"
 
********************************************************************************
 
Connect to host: 192.168.0.71
Last login: Thu May  8 17:04:02 2014 from 183.14.8.49
[root@localhost ~ ]# cat /etc/redhat-release
CentOS release 5.8 (Final)
[root@localhost ~ ]#
********************************************************************************
 
Connect to host: 192.168.0.72
Last login: Thu May  8 17:03:05 2014 from 192.168.0.232
[root@localhost ~ ]# cat /etc/redhat-release
CentOS release 5.8 (Final)
[root@localhost ~ ]#
********************************************************************************
 
Connect to host: 192.168.0.73
Last login: Thu May  8 17:02:29 2014 from 192.168.0.232
[root@localhost ~ ]# cat /etc/redhat-release
CentOS release 5.8 (Final)
[root@localhost ~ ]#
********************************************************************************
 
Connect to host: 192.168.0.74
Last login: Thu May  8 17:02:32 2014 from 192.168.0.232
[root@localhost ~ ]# cat /etc/redhat-release
CentOS release 5.8 (Final)
[root@localhost ~ ]#
********************************************************************************
 
Connect to host: 192.168.0.75
root@192.168.0.75's p assword:  
Last login: Thu May  8 17:02:56 2014 from 192.168.0.232[root@localhost ~ ]# cat /etc/redhat-releaseCentOS release 6.4 (Final)[root@localhost ~ ]#********************************************************************************
 
Connect to host: 192.168.0.76
Last login: Thu May  8 17:03:00 2014 from 192.168.0.232[root@localhost ~ ]# cat /etc/redhat-releaseCentOS release 5.8 (Final)[root@localhost ~ ]#********************************************************************************
 
Connect to host: 123.1.149.26
Last login: Thu May  8 16:46:56 2014 from 183.56.157.199[root@LINUX ~ ]# cat /etc/redhat-releaseRed Hat Enterprise Linux Server release 6.5 (Santiago)[root@LINUX ~ ]#[root@AutoTest batch]#

其他命令自动化工具

https://pypi.python.org/pypi/Fabric/ python远程执行与部署库，运维的最爱。

https://fedorahosted.org/func/

https://pypi.python.org/pypi/pyserial/

https://pypi.python.org/pypi/paramiko/1.16.0

https://docs.python.org/2/library/subprocess.html

https://pypi.python.org/pypi/sh

API概览

EOF与TIMEOUT

EOF(End Of File)与TIMEOUT可以在expect方法中使用，它们不是正则表达式，而是常量。

源于异常，而不是BaseException例外。从BaseException直接继承的例外情况赶上他们几乎总是错误的做法保留。

如果子进程已经退出，读取子进程的输出会引发EOF异常。此时子进程的输出全部在before中。

expect()接受的参数是正则表达式或正则表达式列表，可匹配多个可选的响应。比如ssh登录的各种情况处理：

class Ssh(object):
    client = None
    @classmethod
    def connect(cls, ip, username="root", password="123456", prompt=']#',
                silent=False):
        # Ssh to remote server
        ssh_newkey = 'Are you sure you want to continue connecting'
        child = pexpect.spawn('ssh ' + username + '@' + ip, maxread=5000)
        i = 1
        # Enter password
        while i != 0:
            i = child.expect([prompt, 'assword:*', ssh_newkey, pexpect.TIMEOUT,
                              'key.*? failed'])
            if not silent:
                print(child.before + child.after)
            if i == 0:  # find prompt
                pass
            elif i == 1:  # Enter password
                child.send(password + "\r")
            if i == 2:  # SSH does not have the public key. Just accept it.
                child.sendline('yes\r')
            if i == 3:  # Timeout
                raise Exception('ERROR TIMEOUT! SSH could not login. ')
            if i == 4:  # new key
                print(child.before, child.after)
                os.remove(os.path.expanduser('~') + '/.ssh/known_hosts')
        Ssh.client = child
         
    @classmethod
    def command(cls, cmd, prompt=']#', silent=False):
        Ssh.client.buffer = ''
        Ssh.client.send(cmd + "\r")
        # Ssh.client.setwinsize(400,400)
        Ssh.client.expect(prompt)
        if not silent:
            print(Ssh.client.before + Ssh.client.after)
        return Ssh.client.before, Ssh.client.after
         
    @classmethod
    def close(cls,):
        Ssh.client.close()

expect()的超时默认为30秒，超时时生成TIMEOUT异常。可以修改：

# Wait no more than 2 minutes (120 seconds) for password prompt.
child.expect('password:', timeout=120)

行尾处理

Pexpect匹配的正则表达式与标准的有些差异，默认是非贪婪匹配。Pexpect一次读取一个字符读，这样行尾标识$失去了意义。行尾用"\r\n"(CR/LF)表示。Pexpect中的"\n"实际对应"\r\n"。匹配行尾的操作如下：

child.expect('\r\n')

尽量用：

child.expect ('.+')

而不是：

child.expect ('.*')

API文档

pexpect.screen和pexpect.ANSI在版本4已经不推荐使用，建议用pyte替代。

class spawn

def __init__(self, command, args=[], timeout=30, maxread=2000,
        searchwindowsize=None, logfile=None, cwd=None, env=None,
        ignore_sighup=True):

调用示例：

child = pexpect.spawn('/usr/bin/ftp')
child = pexpect.spawn('/usr/bin/ssh user@example.com')
child = pexpect.spawn('ls -latr /tmp')
child = pexpect.spawn('/usr/bin/ftp', [])
child = pexpect.spawn('/usr/bin/ssh', ['user@example.com'])
child = pexpect.spawn('ls', ['-latr', '/tmp'])

pexpect不能解释shell元字符，比如 (>, |, or *)，需要启动shell来解决该问题：

child = pexpect.spawn('/bin/bash -c "ls -l | grep LOG > logs.txt"')
child.expect(pexpect.EOF)

日志输出：

child = pexpect.spawn('some_command')
fout = open('mylog.txt','wb')
child.logfile = fout

输出到stdout

# In Python 2:
child = pexpect.spawn('some_command')
child.logfile = sys.stdout# 
 
In Python 3, spawnu should be used to give str to stdout:
child = pexpect.spawnu('some_command')
child.logfile = sys.stdout

可爱的python测试开发库 请在github上点赞，谢谢！
python中文库文档汇总
[雪峰磁针石博客]python3标准库-中文版
[雪峰磁针石博客]python3快速入门教程
接口自动化性能测试线上培训大纲
python测试开发自动化测试数据分析人工智能自学每周一练
更多内容请关注 雪峰磁针石：简书

• 技术支持qq群： 144081101(后期会录制视频存在该群群文件) 591302926 567351477 钉钉免费群：21745728

• 道家技术-手相手诊看相中医等钉钉群21734177 qq群：391441566 184175668 338228106 看手相、面相、舌相、抽签、体质识别。服务费50元每人次起。请联系钉钉或者微信pythontesting