搭建支持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，你就可以通过断言完成预期结果的自动验证了。
