搭建支持RESTful风格和WebSocker协议的接口自动化测

第一步，你要建立一个叫做 common.py 的公共的方法类。下面我给出的这段注释详细的代码，就是类似我们使用 Postman 的公共方法的封装，它可以完成 HTTP 协议的 GET 请求或 POST 请求的验证，并且和你的业务无关。

# 定义一个common的类，它的父类是object

class Common(object):

  # common的构造函数

  def __init__(self):

    # 被测系统的根路由

    self.url_root = 'http://127.0.0.1:12356'

  # 封装你自己的get请求，uri是访问路由，params是get请求的参数，如果没有默认为空

  def get(self, uri, params=''):

    # 拼凑访问地址

    url = self.url_root + uri + params

    # 通过get请求访问对应地址

    res = requests.get(url)

    # 返回request的Response结果，类型为requests的Response类型

    return res

  # 封装你自己的post方法，uri是访问路由，params是post请求需要传递的参数，如果没有参数这里为空

  def post(self, uri, params=''):

    # 拼凑访问地址

    url = self.url_root + uri

    if len(params) > 0:

      # 如果有参数，那么通过post方式访问对应的url，并将参数赋值给requests.post默认参数data

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.post(url, data=params)

    else:

      # 如果无参数，访问方式如下

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.post(url)

    return res

接下来，用你自己的 Common 类，修改第一个接口的单接口测试脚本，就可以得到下面的代码了。

# Python代码中引入requests库，引入后才可以在你的代码中使用对应的类以及成员函数

from common import Common

# 首页的路由

uri = '/'

# 实例化自己的Common

comm = Common()

#调用你自己在Common封装的get方法 ，返回结果存到了response_index中

response_index = comm.get(uri)

# 存储返回的response_index对象的text属性存储了访问主页的response信息，通过下面打印出来

print('Response内容：' + response_index.text)

从这段代码中你可以看到，与前面对应的单接口测试脚本相比，代码的行数有明显的减少，这也能减少你很多的工作量，与此同时，如果你有任何关于 HTTP 协议的操作，都可以在 Common 类中进行修改和完善。

如果使用你自己刚刚建立的公共类（在我们内部有时候喜欢把它叫做轮子，这是源于一句俚语“不用重复造轮子”，因为 Common 类就是重复被各个检测代码使用的“轮子”）修改一下第二个接口的单接口测试脚本，代码就会变成下面这个样子：

#登录页路由

uri = '/login'

# username变量存储用户名参数

username = 'criss'

# password变量存储密码参数

password = 'criss'

# 拼凑body的参数

payload = 'username=' + username + '&password=' + password

comm = Common()

response_login = comm.post(uri,params=payload)

print('Response内容：' + response_login.text)

当你有一些更加复杂的脚本时，你会发现两次代码的变化会变得更明显，也更易读。

那么。使用我们一起封装的框架来完成上面的多接口测试后，就会得到下面的代码：

# Python代码中引入requests库，引入后才可以在你的代码中使用对应的类以及成员函数

from common import Common

# 建立uri_index的变量，存储战场的首页路由

uri_index = '/'

# 实例化自己的Common

comm = Common()

#调用你自己在Common封装的get方法 ，返回结果存到了response_index中

response_index = comm.get(uri_index)

# 存储返回的response_index对象的text属性存储了访问主页的response信息，通过下面打印出来

print('Response内容：' + response_index.text)

# uri_login存储战场的登录

uri_login = '/login'

# username变量存储用户名参数

username = 'criss'

# password变量存储密码参数

password = 'criss'

# 拼凑body的参数

payload = 'username=' + username + '&password=' + password

comm = Common()

response_login = comm.post(uri_login,params=payload)

print('Response内容：' + response_login.text)

# uri_selectEq存储战场的选择武器

uri_selectEq = '/selectEq'

# 武器编号变量存储用户名参数

equipmentid = '10003'

# 拼凑body的参数

payload = 'equipmentid=' + equipmentid

comm = Common()

response_selectEq = comm.post(uri_selectEq,params=payload)

print('Response内容：' + response_selectEq.text)

# uri_kill存储战场的选择武器

uri_kill = '/kill'

# 武器编号变量存储用户名参数

enemyid = '20001'

# 拼凑body的参数

payload = 'enemyid=' + enemyid+"&equipmentid="+equipmentid

comm = Common()

response_kill = comm.post(uri_kill,params=payload)

print('Response内容：' + response_kill.text)

你可以看到，上面的代码大量重复了你自己写的通用类的调用，这个其实是可以合成一个的；同时，你再观察一下我们一起写的 Common 类，你会发现有一个 self.url_root = ‘http://127.0.0.1:12356’，如果这里这样写，你的 Common 就只能用来测试我们这个小系统了，除非你每次都去修改框架。

但是，任何一个框架的维护者，都不希望框架和具体逻辑强相关，因此这也是一个优化点，那么将上面的内容都修改后，代码就会变成下面这个样子：

# Python代码中引入requests库，引入后才可以在你的代码中使用对应的类以及成员函数

from common import Common

# 建立uri_index的变量，存储战场的首页路由

uri_index = '/'

# 实例化自己的Common

comm = Common('http://127.0.0.1:12356')

#调用你自己在Common封装的get方法 ，返回结果存到了response_index中

response_index = comm.get(uri_index)

# 存储返回的response_index对象的text属性存储了访问主页的response信息，通过下面打印出来

print('Response内容：' + response_index.text)

# uri_login存储战场的登录

uri_login = '/login'

# username变量存储用户名参数

username = 'criss'

# password变量存储密码参数

password = 'criss'

# 拼凑body的参数

payload = 'username=' + username + '&password=' + password

response_login = comm.post(uri_login,params=payload)

print('Response内容：' + response_login.text)

# uri_selectEq存储战场的选择武器

uri_selectEq = '/selectEq'

# 武器编号变量存储用户名参数

equipmentid = '10003'

# 拼凑body的参数

payload = 'equipmentid=' + equipmentid

response_selectEq = comm.post(uri_selectEq,params=payload)

print('Response内容：' + response_selectEq.text)

# uri_kill存储战场的选择武器

uri_kill = '/kill'

# 武器编号变量存储用户名参数

enemyid = '20001'

# 拼凑body的参数

payload = 'enemyid=' + enemyid+"&equipmentid="+equipmentid

response_kill = comm.post(uri_kill,params=payload)

print('Response内容：' + response_kill.text)

是不是比上一个节省了很多代码，同时也看的更加的易读了，那么我们封住好的Common就变成了如下的样子：

# 定义一个common的类，它的父类是object

class Common(object):

# common的构造函数

def __init__(self,url_root):

# 被测系统的跟路由

self.url_root = url_root

# 封装你自己的get请求，uri是访问路由，params是get请求的参数，如果没有默认为空

def get(self, uri, params=''):

# 拼凑访问地址

url = self.url_root + uri + params

# 通过get请求访问对应地址

res = requests.get(url)

# 返回request的Response结果，类型为requests的Response类型

return res

# 封装你自己的post方法，uri是访问路由，params是post请求需要传递的参数，如果没有参数这里为空

def post(self, uri, params=''):

# 拼凑访问地址

url = self.url_root + uri

if len(params) > 0:

# 如果有参数，那么通过post方式访问对应的url，并将参数赋值给requests.post默认参数data

# 返回request的Response结果，类型为requests的Response类型

res = requests.post(url, data=params)

else:

# 如果无参数，访问方式如下

# 返回request的Response结果，类型为requests的Response类型

res = req

你可以看到，在上面这段代码中，我主要是让我们 Common 类的构造函数接受了一个变量，这个变量就是被测系统的根路由。这样是不是就比上一个代码段节省了很多代码，同时也更加易读了？那么我们封装好的 Common 就变成了下面这个样子：

# 定义一个common的类，它的父类是object

class Common(object):

  # common的构造函数

  def __init__(self,url_root):

    # 被测系统的跟路由

    self.url_root = url_root

  # 封装你自己的get请求，uri是访问路由，params是get请求的参数，如果没有默认为空

  def get(self, uri, params=''):

    # 拼凑访问地址

    url = self.url_root + uri + params

    # 通过get请求访问对应地址

    res = requests.get(url)

    # 返回request的Response结果，类型为requests的Response类型

    return res

  # 封装你自己的post方法，uri是访问路由，params是post请求需要传递的参数，如果没有参数这里为空

  def post(self, uri, params=''): 

    # 拼凑访问地址 

    url = self.url_root + uri 

    if len(params) > 0: 

      # 如果有参数，那么通过post方式访问对应的url，并将参数赋值给requests.post默认参数data

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.post(url, data=params) 

    else:

      # 如果无参数，访问方式如下

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.post(url)

      return res

通过改造 Common 类的构造函数，这个类已经变成一个通用类了，无论是哪一个项目的接口测试，都可以使用它来完成 HTTP 协议的接口验证了。其实到这里，我们上面说的只能算是一个调试代码，还不能算是一个测试框架。上面这些代码所有的返回值都打印到控制台后，为了完成接口测试，你需要时时刻刻看着控制台，这还不能算是自动化，只能说是一个辅助小工具。

在这里，你应该让全部测试结果都存储到测试报告里面，同时通过一个测试驱动框架来完成各个模块的驱动，比如Python 的 Unittest 。因此，上面的 Common 类还需要和 Python 的 unittest 一起使用，才算是一个完美的测试框架。

让你的框架支持RESTful风格的接口

RESTful 接口的测试和原始的 HTTP 协议接口的测试，又有什么区别呢？这里面有两部分需要你特别关注：数据交换的承载方式和操作方式。

我先说说数据交换的承载方式，RESTful 风格的接口主要是以 JSON 格式来进行数据交换。

另外一个部分是操作方式，上面用了HTTP 协议的 Get 和 Post，其实 HTTP 协议有很多方法，但是我们仅仅用了这两种，而 RESTful 的规定，使 HTTP 的很多方法都被利用到了，比如说，Get 方法用来获取资源，Post 方法用来新建资源（或者更新资源）；再比如说，Put 方法用来更新资源、Delete 方法用来删除资源等等。

现在，我们已经可以借助开源库，解决数据交换的事情了，但是，RESTful 风格接口和普通 HTTP 接口相比，还有一个明显的区别，那就是 RESTful 规定了 HTTP 的每一个方法都做固定的事情，可我们原来框架中的 Common 类却只支持 Get 和 Post 方法，因此，你需要在 Common 类中加入 Delete 和 Put 方法的支持。具体的操作你可以依据下面这个代码段来完成：

  def put(self,uri,params=None):

    '''

    封装你自己的put方法，uri是访问路由，params是put请求需要传递的参数，如果没有参数这里为空

    :param uri: 访问路由 

    :param params: 传递参数，string类型，默认为None 

    :return: 此次访问的response

    ''' 

    url = self.url_root+uri

    if params is not None:

      # 如果有参数，那么通过put方式访问对应的url，并将参数赋值给requests.put默认参数data

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.put(url, data=params)

  else:

      # 如果无参数，访问方式如下

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.put(url)

  return res

def delete(self,uri,params=None):

  '''

  封装你自己的delete方法，uri是访问路由，params是delete请求需要传递的参数，如果没有参数这里为空

  :param uri: 访问路由

  :param params: 传递参数，string类型，默认为None

  :return: 此次访问的response

  '''

  url = self.url_root + uri

  if params is not None:

    # 如果有参数，那么通过delete方式访问对应的url，并将参数赋值给requests.delete默认参数data

    # 返回request的Response结果，类型为requests的Response类型

    res = requests.delete(url, data=params)

  else:

    # 如果无参数，访问方式如下

    # 返回request的Response结果，类型为requests的Response类型

    res = requests.delete(url)

  return res

在上面的代码中，你可以看到，我们为了实现 HTTP 协议的 Put 和 Delete 方法，自己封装了 put() 函数和 delete() 函数。其实，要实现 RESTful 风格的接口测试，你只要封装 HTTP 协议对应的 Method 方法就可以了，这样，你的框架就能完美的支持 RESTful 风格的接口了。完成了这个操作后，我们的 Common 类就既可以完成 HTTP 协议接口的测试，也可以完成 RESTful 接口的测试了。

将 WebSocket 接口封装进你的框架

库，因此我只要用它完成客户端的撰写，就可以进行接口测试了。这里，我写下了第一个 WebSocket 的调用代码（这里我们以 http://www.websocket.org/demos/echo/ 为例)，如下面图中所示，我在代码里面写了详细的注释，你肯定能看懂每一句话的意思。

#引入websocket的create_connection类

from websocket import create_connection

# 建立和WebSocket接口的链接

ws = create_connection("ws://echo.websocket.org")

# 打印日子

print("发送 'Hello, World'...")

# 发送Hello，World

ws.send("Hello, World")

# 将WebSocket的返回值存储result变量

result = ws.recv()

# 打印返回的result

print("返回"+result)

# 关闭WebSocket链接

ws.close()

不知道你发现没有，上面的代码和 HTTP 协议的接口类似，都是先和一个请求建立连接，然后发送信息。它们的区别是，WebSocket 是一个长连接，因此需要人为的建立连接，然后再关闭链接，而 HTTP 却并不需要进行这一操作。

我们上面封装了 Common 类，你可以在它的构造函数中，添加一个 API 类型的参数，以便于知道自己要做的是什么协议的接口，其中 http 代表 HTTP 协议接口，ws 代表 WebSocket 协议接口。由于 WebSocket 是一个长连接，我们在 Common 类析构函数中添加了关闭 ws 链接的代码，以释放 WebSocket 长连接。依据前面的交互流程，实现代码如下所示：

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# python代码中引入requests库，引入后才可以在你的代码中使用对应的类以及成员函数

import requests

from websocket import create_connection

# 定义一个common的类，它的父类是object

class Common(object):

  # common的构造函数

  def __init__(self,url_root,api_type):

    '''

    :param api_type:接口类似当前支持http、ws，http就是HTTP协议，ws是WebSocket协议

    :param url_root: 被测系统的根路由

    ''' 

    if api_type=='ws':

      self.ws = create_connection(url_root)

    elif api_type=='http':

      self.ws='null'

      self.url_root = url_root

  # ws协议的消息发送

  def send(self,params):

    '''

    :param params: websocket接口的参数

    :return: 访问接口的返回值

    '''

    self.ws.send(params)

    res = self.ws.recv()

    return res

  # common类的析构函数，清理没有用的资源

  def __del__(self):

    '''

    :return:

    '''

    if self.ws!='null":

      self.ws.close()

  def get(self, uri, params=None):

    '''

    封装你自己的get请求，uri是访问路由，params是get请求的参数，如果没有默认为空

    :param uri: 访问路由

    :param params: 传递参数，string类型，默认为None

    :return: 此次访问的response

    '''

    # 拼凑访问地址

    if params is not None:

      url = self.url_root + uri + params

    else: 

      url = self.url_root + uri

    # 通过get请求访问对应地址

    res = requests.get(url)

    # 返回request的Response结果，类型为requests的Response类型

    return res

  def post(self, uri, params=None):

    '''

    封装你自己的post方法，uri是访问路由，params是post请求需要传递的参数，如果没有参数这里为空

    :param uri: 访问路由

    :param params: 传递参数，string类型，默认为None

    :return: 此次访问的response

    '''

    # 拼凑访问地址

    url = self.url_root + uri

    if params is not None:

      # 如果有参数，那么通过post方式访问对应的url，并将参数赋值给requests.post默认参数data

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.post(url, data=params)

    else:

      # 如果无参数，访问方式如下

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.post(url) 

    return res

  def put(self,uri,params=None):

    '''

    封装你自己的put方法，uri是访问路由，params是put请求需要传递的参数，如果没有参数这里为空

    :param uri: 访问路由

    :param params: 传递参数，string类型，默认为None

    :return: 此次访问的response

    '''

    url = self.url_root+uri

    if params is not None:

      # 如果有参数，那么通过put方式访问对应的url，并将参数赋值给requests.put默认参数data

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.put(url, data=params)

    else:

      # 如果无参数，访问方式如下

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.put(url)

    return res

  def delete(self,uri,params=None):

    '''

    封装你自己的delete方法，uri是访问路由，params是delete请求需要传递的参数，如果没有参数这里为空

    :param uri: 访问路由

    :param params: 传递参数，string类型，默认为None

    :return: 此次访问的response

    '''

    url = self.url_root + uri

    if params is not None:

      # 如果有参数，那么通过put方式访问对应的url，并将参数赋值给requests.put默认参数data

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.delete(url, data=params)

    else:

      # 如果无参数，访问方式如下

      # 返回request的Response结果，类型为requests的Response类型

      res = requests.put(url)

    return res

那么，使用上述的 Common 类将上面那个流水账一样的脚本进行改造后，就得出了下面这段代码：

from common import Common

# 建立和WebSocket接口的链接

con = Common('ws://echo.websocket.org','ws')

# 获取返回结果

result = con.send('Hello, World...')

#打印日志

print(result)

#释放WebSocket的长连接

del con

现在，从改造后的代码中，你是不是更能体会到框架的魅力了？它能让代码变得更加简洁和易读，将 WebSocket 的协议封装到你的框架后，你就拥有了一个既包含 HTTP 协议又包含 WebSocket 协议的接口测试框架了，随着你不断地积累新协议，你的框架会越来越强大，你自己的秘密武器库也会不断扩充，随着你对它的不断完善，它会让你的接口测试工作越来越简单，越来越快速。

最后，我们需要将数据封装，例如把测试数据放在excel里，Excel 是在设计测试用例方面使用最多的一个工具，那么我们也就可以用 Excel 作为自己的参数存储文件。那么如何选取和调用参数呢？你可以看看我设计的参数类：

import json

import xlrd

class Param(object):

  def __init__(self,paramConf='{}'):

    self.paramConf = json.loads(paramConf)

  def paramRowsCount(self):

    pass

  def paramColsCount(self):

    pass

  def paramHeader(self):

    pass

  def paramAllline(self):

    pass

  def paramAlllineDict(self):

    pass

class XLS(Param):

  '''

  xls基本格式(如果要把xls中存储的数字按照文本读出来的话,纯数字前要加上英文单引号:

  第一行是参数的注释,就是每一行参数是什么

  第二行是参数名,参数名和对应模块的po页面的变量名一致

  第3~N行是参数

  最后一列是预期默认头Exp

  '''

  def __init__(self, paramConf):

    '''

    :param paramConf: xls 文件位置(绝对路径)

    '''

    self.paramConf = paramConf

    self.paramfile = self.paramConf['file']

    self.data = xlrd.open_workbook(self.paramfile)

    self.getParamSheet(self.paramConf['sheet'])

  def getParamSheet(self,nsheets):

    '''

    设定参数所处的sheet

    :param nsheets: 参数在第几个sheet中

    :return:

    '''

    self.paramsheet = self.data.sheets()[nsheets]

  def getOneline(self,nRow):

    '''

    返回一行数据

    :param nRow: 行数

    :return: 一行数据 []

    '''

    return self.paramsheet.row_values(nRow)

  def getOneCol(self,nCol):

    '''

    返回一列

    :param nCol: 列数

    :return: 一列数据 []

    '''

    return self.paramsheet.col_values(nCol)

  def paramRowsCount(self):

    '''

    获取参数文件行数

    :return: 参数行数 int

    '''

    return self.paramsheet.nrows

  def paramColsCount(self):

    '''

    获取参数文件列数(参数个数)

    :return: 参数文件列数(参数个数) int

    '''

    return self.paramsheet.ncols

  def paramHeader(self):

    '''

    获取参数名称

    :return: 参数名称[]

    '''

    return self.getOneline(1)

  def paramAlllineDict(self):

    '''

    获取全部参数

    :return: {{}},其中dict的key值是header的值

    '''

    nCountRows = self.paramRowsCount()

    nCountCols = self.paramColsCount()

    ParamAllListDict = {}

    iRowStep = 2

    iColStep = 0

    ParamHeader= self.paramHeader()

    while iRowStep < nCountRows:

    ParamOneLinelist=self.getOneline(iRowStep)

    ParamOnelineDict = {}

    while iColStep<nCountCols:

    ParamOnelineDict[ParamHeader[iColStep]]=ParamOneLinelist[iColStep]

    iColStep=iColStep+1

    iColStep=0

    ParamAllListDict[iRowStep-2]=ParamOnelineDict

    iRowStep=iRowStep+1

    return ParamAllListDict

  def paramAllline(self):

    '''    获取全部参数

    :return: 全部参数[[]]  '''

    nCountRows= self.paramRowsCount()

    paramall = []

    iRowStep =2

    while iRowStep<nCountRows:

    paramall.append(self.getOneline(iRowStep))

    iRowStep=iRowStep+1

    return paramall

  def __getParamCell(self,numberRow,numberCol):

    return self.paramsheet.cell_value(numberRow,numberCol)

class ParamFactory(object):

  def chooseParam(self,type,paramConf):

    map_ = {

    'xls': XLS(paramConf)

    }

    return map_[type

上面这个代码看着很多，但你不需要完全看得懂，你只需要知道它解决问题的思路和方法就可以了，思路就是通过统一抽象，建立一个公共处理数据的方式。你可以设计和使用简单工厂类的设计模式，这样如果多一种参数存储类型，再添加一个对应的处理类就可以了，这很便于你做快速扩展，也可以一劳永逸地提供统一数据的处理模式。

接下来，你就可以把这次测试的全部参数都存到 Excel 里面了，具体内容如下图所示：

通过上面的参数类你可以看出，在这个 Excel 文件中，第一行是给人读取的每一列参数的注释，而所有的 Excel 都是从第二行开始读取的，第二行是参数名和固定的表示预期结果的 exp。现在，我们使用 ParamFactory 类，再配合上面的这个 Excel，就可以完成”战场“系统“选择武器”接口的改造了，如下面这段代码所示：

#引入Common、ParamFactory类

from common import Common

from param import ParamFactory

import os

# uri_login存储战场的选择武器

uri_selectEq = '/selectEq'

comm = Common('http://127.0.0.1:12356',api_type='http')

# 武器编号变量存储武器编号，并且验证返回时是否有参数设计预期结果

# 获取当前路径绝对值

curPath = os.path.abspath('.')

# 定义存储参数的excel文件路径

searchparamfile = curPath+'/equipmentid_param.xls'

# 调用参数类完成参数读取，返回是一个字典，包含全部的excel数据除去excel的第一行表头说明

searchparam_dict = ParamFactory().chooseParam('xls',{'file':searchparamfile,'sheet':0}).paramAlllineDict()

i=0

while i<len(searchparam_dict):

  # 读取通过参数类获取的第i行的参数

  payload = 'equipmentid=' + searchparam_dict[i]['equipmentid']

  # 读取通过参数类获取的第i行的预期

  exp=searchparam_dict[i]['exp']

  # 进行接口测试

  response_selectEq = comm.post(uri_selectEq,params=payload)

  # 打印返回结果

  print('Response内容：' + response_selectEq.text)

  # 读取下一行excel中的数据

  i=i+1

这样再执行你的测试脚本，你就可以看到数据文件中的三条数据，已经都会顺序的自动执行了。那么后续如果将它付诸于你自己的技术栈，以及自己的测试驱动框架比如 Python 的unittest，你就可以通过断言完成预期结果的自动验证了。