前言

去年，我们进行了项目的拆分，拆分后的各个子系统也都逐步的改成了通过接口进行数据的交换，接口测试也被提上日程。经过一段时间的探索，接口自动化测试方案越来越完善，今天给大家做个详细的讲解。

方案

目前我们的接口都是使用的http协议，其测试的基本原理是模拟前端（客户端）向服务器发送数据，得到相应的响应数据，从而判断接口是否可以正常的进行数据交换。在测试的过程中尝试过两种方式，一种是利用性能测试工具Jmeter模拟客户端发起http请求，另外一种是使用python脚本直接编写脚本模拟客户端发起http请求。

利用Jmeter工具配置，需要对如何利用Jmeter进行性能测试熟悉，通过相应的配置可完成，但不够灵活，比如某些字段需要经过特定的加密处理，不能通过Jmeter直接完成。

所以选择直接用python脚本进行，模拟http请求也就几行代码就可完成。但只是模拟请求不是最终的目标，也需要易用，不会编码的人也会维护我们的测试用例，所以形成了现在的形态，遵循了测试框架的一些基本原则，业务逻辑与测试脚本分离，测试脚本与测试数据分离。大致框架如下图所示：

目录结构如下：

所有的测试用例使用Excel统一管理，测试数据根据需要可以选择配置在Excel中或者保存在测试数据文件中。测试用例格式如下：

日志格式如下：

测试完成后可将异常的接口通过邮件发送给相关人。以上是接口测试方案的大致介绍，下面给大家说说具体怎么配置用例。

如何进行测试

测试的核心脚本已经搭建好，后续不会有太大的改动，维护测试用例的Excel表格即可完成后续接口的测试，不管是新接口的测试还是老接口的回归，那如何编写一个接口的测试用例呢？

1、      打开测试用例的Excel表格，填写用例编号、接口描述信息，被测接口的域名和请求地址。

2、      选择接口请求的方式，目前有两种，一种是POST，一种是GET，根据实际情况选择。

3、      选择接口接收数据的方式，目前有三种，Form类型，请求的数据会进行urlencode编码，一般都是这种类型，官网的接口主要是这种；Data类型，以文本的形式直接请求接口，不经过urlencode编码，引擎的接口大部分是这种，选择Data类型时，请求的数据有两种，一种是直接在Excel中配置json字符串，一种是填写文本文件路径，文件中也是json字符串，主要在于post的数据很大时，比如保存案例，在Excel中不好管理。File类型表示上传文件，在测试上传时选择File类型。

4、      配置需要向接口发送的数据，如下图所示，需要根据上一步中选择的类型配置正确的测试数据，除了填写的是文件路径外，数据必须是标准的json格式字符串。

测试数据中，可以带参数，格式为${parameter}，此处的参数必须在后面的关联（Correlation）字段中有赋值，在后面的关联字段配置给大家详细介绍。其中内置了四个参数，分别是：${randomEmail}（随机邮箱地址）、${randomTel}（随机手机号码）、${timestamp}（当前时间戳）、${session}（session id，默认为None）以及${hashPassword}（hash加密密码，明文123456）。

5、      配置数据是否需要编码加密，目前有三种，不加密，MD5加密和DES加密。这是根据我们自身项目的特点加的选项，引擎有几个接口需要进行MD5加密，场景秀的接口都经过了DES加密。

6、      配置检查点，检查点的目的是校验接口返回的数据是否是我们期望的。

7、      配置关联，在接口的测试过程中，两个接口常常会有相关性，比如引擎新建案例需要先登录官网，那么，就需要做前后接口数据的关联。前面步骤已经提到过，在配置测试数据的时候可以配置参数，那么，关联的配置就是为了给这些参数赋值的，格式如下：${parameter}=[level1][level2][level3]，多个参数中间用半角的分号（;）隔开，如下图所示。关联参数有两部分组成，等号前面是参数名称，需要跟测试数据中配置的参数名称保持一致，等号后面的部分是获取当前接口返回值的，因为接口返回值都是json格式的字符串，所以[level1]表示第一层级的指定key的值，[level1][level2]表示获取第一层级指定key的值中的指定key的值，有点绕，我们举例说明，大家就明白了。

登录接口的返回值是：{"data":"http:\/\/my.test.liveapp.com.cn\/admin\/myapp\/applist","success":true,"message":"6tdehonrs6mg9metjqprfind16"}

后续的操作都需要是登录状态，所以需要得到session id，那么参数就可以这么写：${session}=[message]，得到的值就是6tdehonrs6mg9metjqprfind16。

保存案例接口的返回值是：{"ecode":0,"msg":"SUCCESS","data":[{"$id":"55d43d077f8b9ad56b8b4576","page_id":115323,"page_order":0},……

后续的操作需要mongo id和page id，那么参数可以这样写：${mongo_id}=[data][0][$id];${page_id}=[data][0][page_id]，就可以分别得到55d43d077f8b9ad56b8b4576和115323。这里大家发现会出现数字，是因为”data”的值是一个列表，而不是字典，没有相应的key，所以可以用数字代替，从0开始计算。

8、      最后一步，配置用例是否执行，只有Yes和No两种选项，这个很好理解，就不多解释了。

以上就是配置一条用例的过程，配置完成后，保存Excel文件，提交到SVN即可，Jenkins接口测试的项目已经配置好，在每次引擎项目构建之后都会自动构建接口测试项目。

如果大家还有什么疑问，可以找我一起探讨。

附代码如下（Github：https://github.com/TronGeek/InterfaceTest）：


#!/usr/bin/env python
#coding=utf8

# Todo：接口自动化测试
# Author：归根落叶
# Blog：http://this.ispenn.com

import json
import http.client,mimetypes
from urllib.parse import urlencode
import random
import time
import re
import logging
import os,sys
try:
import xlrd
except:
os.system(\'pip install -U xlrd\')
import xlrd
try:
from pyDes import *
except ImportError as e:
os.system(\'pip install -U pyDes --allow-external pyDes --allow-unverified pyDes\')
from pyDes import *
import hashlib
import base64
import smtplib
from email.mime.text import MIMEText

log_file = os.path.join(os.getcwd(),\'log/liveappapi.log\')
log_format = \'[%(asctime)s] [%(levelname)s] %(message)s\'
logging.basicConfig(format=log_format,filename=log_file,filemode=\'w\',level=logging.DEBUG)
console = logging.StreamHandler()
console.setLevel(logging.DEBUG)
formatter = logging.Formatter(log_format)
console.setFormatter(formatter)
logging.getLogger(\'\').addHandler(console)

#获取并执行测试用例
def runTest(testCaseFile):
testCaseFile = os.path.join(os.getcwd(),testCaseFile)
if not os.path.exists(testCaseFile):
logging.error(\'测试用例文件不存在！！！\')
sys.exit()
testCase = xlrd.open_workbook(testCaseFile)
table = testCase.sheet_by_index(0)
errorCase = []
correlationDict = {}
correlationDict[\'${hashPassword}\'] = hash1Encode(\'123456\')
correlationDict[\'${session}\'] = None
for i in range(1,table.nrows):
correlationDict[\'${randomEmail}\'] = \'\'.join(random.sample(\'abcdefghijklmnopqrstuvwxyz\',6)) + \'@automation.test\'
correlationDict[\'${randomTel}\'] = \'186\' + str(random.randint(10000000,99999999))
correlationDict[\'${timestamp}\'] = int(time.time())
if table.cell(i,10).value.replace(\'\n\',\'\').replace(\'\r\',\'\') != \'Yes\':
continue
num = str(int(table.cell(i,0).value)).replace(\'\n\',\'\').replace(\'\r\',\'\')
api_purpose = table.cell(i,1).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
api_host = table.cell(i,2).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
request_url = table.cell(i,3).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
request_method = table.cell(i,4).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
request_data_type = table.cell(i,5).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
request_data = table.cell(i,6).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
encryption = table.cell(i,7).value.replace(\'\n\',\'\').replace(\'\r\',\'\')
check_point = table.cell(i,8).value
correlation = table.cell(i,9).value.replace(\'\n\',\'\').replace(\'\r\',\'\').split(\';\')
for key in correlationDict:
if request_url.find(key) > 0:
request_url = request_url.replace(key,str(correlationDict[key]))
if request_data_type == \'Form\':
dataFile = request_data
if os.path.exists(dataFile):
fopen = open(dataFile,encoding=\'utf-8\')
request_data = fopen.readline()
fopen.close()
for keyword in correlationDict:
if request_data.find(keyword) > 0:
request_data = request_data.replace(keyword,str(correlationDict[keyword]))
try:
if encryption == \'MD5\':
request_data = json.loads(request_data)
status,md5 = getMD5(api_host,urlencode(request_data).replace("%27","%22"))
if status != 200:
logging.error(num + \' \' + api_purpose + "[ " + str(status) + " ], 获取md5验证码失败！！！")
continue
request_data = dict(request_data,**{"sign":md5.decode("utf-8")})
request_data = urlencode(request_data).replace("%27","%22")
elif encryption == \'DES\':
request_data = json.loads(request_data)
request_data = urlencode({\'param\':encodePostStr(request_data)})
else:
request_data = urlencode(json.loads(request_data))
except Exception as e:
logging.error(num + \' \' + api_purpose + \' 请求的数据有误，请检查[Request Data]字段是否是标准的json格式字符串！\')
continue
elif request_data_type == \'Data\':
dataFile = request_data
if os.path.exists(dataFile):
fopen = open(dataFile,encoding=\'utf-8\')
request_data = fopen.readline()
fopen.close()
for keyword in correlationDict:
if request_data.find(keyword) > 0:
request_data = request_data.replace(keyword,str(correlationDict[keyword]))
request_data = request_data.encode(\'utf-8\')
elif request_data_type == \'File\':
dataFile = request_data
if not os.path.exists(dataFile):
logging.error(num + \' \' + api_purpose + \' 文件路径配置无效，请检查[Request Data]字段配置的文件路径是否存在！！！\')
continue
fopen = open(dataFile,\'rb\')
data = fopen.read()
fopen.close()
request_data = \'\'\'
------WebKitFormBoundaryDf9uRfwb8uzv1eNe
Content-Disposition:form-data;name="file";filename="%s"
Content-Type:
Content-Transfer-Encoding:binary

%s
------WebKitFormBoundaryDf9uRfwb8uzv1eNe--
\'\'\' % (os.path.basename(dataFile),data)
status,resp = interfaceTest(num,api_purpose,api_host,request_url,request_data,check_point,request_method,request_data_type,correlationDict[\'${session}\'])
if status != 200:
errorCase.append((num + \' \' + api_purpose,str(status),\'http://\'+api_host+request_url,resp))
continue
for j in range(len(correlation)):
param = correlation[j].split(\'=\')
if len(param) == 2:
if param[1] == \'\' or not re.search(r\'^\[\',param[1]) or not re.search(r\'\]$\',param[1]):
logging.error(num + \' \' + api_purpose + \' 关联参数设置有误，请检查[Correlation]字段参数格式是否正确！！！\')
continue
value = resp
for key in param[1][1:-1].split(\'][\'):
try:
temp = value[int(key)]
except:
try:
temp = value[key]
except:
break
value = temp
correlationDict[param[0]] = value
return errorCase

# 接口测试
def interfaceTest(num,api_purpose,api_host,request_url,request_data,check_point,request_method,request_data_type,session):
headers = {\'Content-Type\':\'application/x-www-form-urlencoded; charset=UTF-8\',
\'X-Requested-With\':\'XMLHttpRequest\',
\'Connection\':\'keep-alive\',
\'Referer\':\'http://\' + api_host,
\'User-Agent\':\'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/43.0.2357.134 Safari/537.36\'}
if session is not None:
headers[\'Cookie\'] = \'session=\' + session
if request_data_type == \'File\':
headers[\'Content-Type\'] = \'multipart/form-data;boundary=----WebKitFormBoundaryDf9uRfwb8uzv1eNe;charset=UTF-8\'
elif request_data_type == \'Data\':
headers[\'Content-Type\'] = \'text/plain; charset=UTF-8\'

conn = http.client.HTTPConnection(api_host)
if request_method == \'POST\':
conn.request(\'POST\',request_url,request_data,headers=headers)
elif request_method == \'GET\':
conn.request(\'GET\',request_url+\'?\'+request_data,headers=headers)
else:
logging.error(num + \' \' + api_purpose + \' HTTP请求方法错误，请确认[Request Method]字段是否正确！！！\')
return 400,request_method
response = conn.getresponse()
status = response.status
resp = response.read()
if status == 200:
resp = resp.decode(\'utf-8\')
if re.search(check_point,str(resp)):
logging.info(num + \' \' + api_purpose + \' 成功, \' + str(status) + \', \' + str(resp))
return status,json.loads(resp)
else:
logging.error(num + \' \' + api_purpose + \' 失败！！！, [ \' + str(status) + \' ], \' + str(resp))
return 2001,resp
else:
logging.error(num + \' \' + api_purpose + \' 失败！！！, [ \' + str(status) + \' ], \' + str(resp))
return status,resp.decode(\'utf-8\')

#获取md5验证码
def getMD5(url,postData):
headers = {\'Content-Type\':\'application/x-www-form-urlencoded; charset=UTF-8\',
\'X-Requested-With\':\'XMLHttpRequest\'}
conn = http.client.HTTPConnection(\'this.ismyhost.com\')
conn.request(\'POST\',\'/get_isignature\',postData,headers=headers)
response = conn.getresponse()
return response.status,response.read()

# hash1加密
def hash1Encode(codeStr):
hashobj = hashlib.sha1()
hashobj.update(codeStr.encode(\'utf-8\'))
return hashobj.hexdigest()

# DES加密
def desEncode(desStr):
k = des(\'secretKEY\', padmode=PAD_PKCS5)
encodeStr = base64.b64encode(k.encrypt(json.dumps(desStr)))
return encodeStr

# 字典排序
def encodePostStr(postData):
keyDict = {\'key\':\'secretKEY\'}
mergeDict = dict(postData, **keyDict)
mergeDict = sorted(mergeDict.items())
postStr = \'\'
for i in mergeDict:
postStr = postStr + i[0] + \'=\' + i[1] + \'&\'
postStr = postStr[:-1]
hashobj = hashlib.sha1()
hashobj.update(postStr.encode(\'utf-8\'))
token = hashobj.hexdigest()
postData[\'token\'] = token
return desEncode(postData)

#发送通知邮件
def sendMail(text):
sender = \'no-reply@myhost.cn\'
receiver = [\'penn@myhost.cn\']
mailToCc = [\'penn@myhost.cn\']
subject = \'[AutomantionTest]接口自动化测试报告通知\'
smtpserver = \'smtp.exmail.qq.com\'
username = \'no-reply@myhost.cn\'
password = \'password\'

msg = MIMEText(text,\'html\',\'utf-8\')
msg[\'Subject\'] = subject
msg[\'From\'] = sender
msg[\'To\'] = \';\'.join(receiver)
msg[\'Cc\'] = \';\'.join(mailToCc)
smtp = smtplib.SMTP()
smtp.connect(smtpserver)
smtp.login(username, password)
smtp.sendmail(sender, receiver + mailToCc, msg.as_string())
smtp.quit()

def main():
errorTest = runTest(\'TestCase/TestCasePre.xlsx\')
if len(errorTest) > 0:
html = \'接口自动化定期扫描，共有 \' + str(len(errorTest)) + \' 个异常接口，列表如下：\' + \'

\'
for test in errorTest:
html = html + \'

\'
html = html + \'

\'
#sendMail(html)

if __name__ == \'__main__\':
main()

作者：归根落叶， 转自：https://www.ispenn.com/2015/08/interface-test-automation-scheme-details