为什么要做接口自动化框架

1、业务与配置的分离

2、数据与程序的分离；数据的变更不影响程序

3、有日志功能，实现无人值守

4、自动发送测试报告

5、不懂编程的测试人员也可以进行测试

正常接口测试的流程是什么？

确定接口测试使用的工具----->配置需要的接口参数----->进行测试----->检查测试结果----->生成测试报告

测试的工具：python+requests

接口测试用例：excel

一、接口框架如下：

1、action包：用来存放关键字函数

2、config包：用来存放配置文件

3、TestData：用来存放测试数据，excel表

4、Log包：用来存放日志文件

5、utils包：用来存放公共的类

6、运行主程序interface_auto_test.py

7、Readme.txt：告诉团队组员使用改框架需要注意的地方

二、接口的数据规范设计---Case设计

一个sheet对应数据库里面一张表

APIsheet存放

编号；从1开始

接口的名称(APIName)；

请求的url(RequestUrl)；

请求的方法(RequestMethod)；

传参的方式(paramsType)：post/get请求方法不一样

用例说明(APITestCase)

是否执行(Active)部分接口已测通，下次不用测试，直接把这里设置成N，跳过此接口

post与get的区别

查看post详情

post请求参数一般是json串，参数放在from表单里面；参数一般不可见，相对来说安全性高些

查看get详情

get请求参数一般直接放在url里面

2.1注册接口用例

RequestData：请求的数据

(开发制定的传参方式)

RelyData：数据依赖

ResponseCode：响应code

ResponseData：响应数据

DataStore：存储的依赖数据；如果存在数据库里面，在表里增加一个字段用来存依赖的数据

(存储的方式是编写接口自动化的人员来设定的存储方式)

CheckPoint：检查点

Active：是否执行

Status：执行用例的状态，方便查看用例是否执行成功

ErrorInfo：case运行失败，失败的错误信息；eg：是也本身的原因还是case设置失败，还是其他原因

2.2登录接口用例

RequestData：请求的数据

(开发制定的传参方式)

RelyData：数据依赖

(存储的方式是编写接口自动化的人员来设定的存储方式)

ResponseCode：响应code

ResponseData：响应数据

DataStore：存储的依赖数据；如果存在数据库里面，在表里增加一个字段用来存依赖的数据

(存储的方式是编写接口自动化的人员来设定的存储方式)

CheckPoint：检查点

Active：是否执行

Status：执行用例的状态，方便查看用例是否执行成功

ErrorInfo：case运行失败，失败的错误信息；eg：是也本身的原因还是case设置失败，还是其他原因

重点说明下RelyData：数据依赖

采取的是字典：key:value来存储数据格式；

{"request":{"username":"register->1","password":"register->1"},"response":{"code":"register->1"}}

格式化之后：

{

"request":{

"username":"register->1",

"password":"register->1"

},

"response":{

"code":"register->1"

}

}

三、创建utils包：用来存放公共的类

3.1 ParseExcel.py 操作封装excel的类(ParseExcel.py)

#encoding=utf-8

import openpyxl

from openpyxl.styles import Border, Side, Font

import time

class ParseExcel(object):

def __init__(self):

self.workbook = None

self.excelFile = None

self.font = Font(color = None) # 设置字体的颜色

# 颜色对应的RGB值

self.RGBDict = {'red': 'FFFF3030', 'green': 'FF008B00'}

def loadWorkBook(self, excelPathAndName):

# 将excel文件加载到内存，并获取其workbook对象

try:

self.workbook = openpyxl.load_workbook(excelPathAndName)

except Exception as err:

raise err

self.excelFile = excelPathAndName

return self.workbook

def getSheetByName(self, sheetName):

# 根据sheet名获取该sheet对象

try:

# sheet = self.workbook.get_sheet_by_name(sheetName)

sheet = self.workbook[sheetName]

return sheet

except Exception as err:

raise err

def getSheetByIndex(self, sheetIndex):

# 根据sheet的索引号获取该sheet对象

try:

# sheetname = self.workbook.get_sheet_names()[sheetIndex]

sheetname = self.workbook.sheetnames[sheetIndex]

except Exception as err:

raise err

# sheet = self.workbook.get_sheet_by_name(sheetname)

sheet = self.workbook[sheetname]

return sheet

def getRowsNumber(self, sheet):

# 获取sheet中有数据区域的结束行号

return sheet.max_row

def getColsNumber(self, sheet):

# 获取sheet中有数据区域的结束列号

return sheet.max_column

def getStartRowNumber(self, sheet):

# 获取sheet中有数据区域的开始的行号

return sheet.min_row

def getStartColNumber(self, sheet):

# 获取sheet中有数据区域的开始的列号

return sheet.min_column

def getRow(self, sheet, rowNo):

# 获取sheet中某一行,返回的是这一行所有的数据内容组成的tuple，

# 下标从1开始，sheet.rows[1]表示第一行

try:

rows = []

for row in sheet.iter_rows():

rows.append(row)

return rows[rowNo - 1]

except Exception as err:

raise err

def getColumn(self, sheet, colNo):

# 获取sheet中某一列,返回的是这一列所有的数据内容组成tuple，

# 下标从1开始，sheet.columns[1]表示第一列

try:

cols = []

for col in sheet.iter_cols():

cols.append(col)

return cols[colNo - 1]

except Exception as err:

raise err

def getCellOfValue(self, sheet, coordinate = None,

rowNo = None, colsNo = None):

# 根据单元格所在的位置索引获取该单元格中的值,下标从1开始,

# sheet.cell(row = 1, column = 1).value，

# 表示excel中第一行第一列的值

if coordinate != None:

try:

return sheet[coordinate]

except Exception as err:

raise err

elif coordinate is None and rowNo is not None and \

colsNo is not None:

try:

return sheet.cell(row = rowNo, column = colsNo).value

except Exception as err:

raise err

else:

raise Exception("Insufficient Coordinates of cell !")

def getCellOfObject(self, sheet, coordinate = None,

rowNo = None, colsNo = None):

# 获取某个单元格的对象，可以根据单元格所在位置的数字索引，

# 也可以直接根据excel中单元格的编码及坐标

# 如getCellObject(sheet, coordinate = 'A1') or

# getCellObject(sheet, rowNo = 1, colsNo = 2)

if coordinate != None:

try:

# return sheet.cell(coordinate = coordinate)

return sheet[coordinate]

except Exception as err:

raise err

elif coordinate == None and rowNo is not None and \

colsNo is not None:

try:

return sheet.cell(row = rowNo,column = colsNo)

except Exception as err:

raise err

else:

raise Exception("Insufficient Coordinates of cell !")

def writeCell(self, sheet, content, coordinate = None,

rowNo = None, colsNo = None, style = None):

#根据单元格在excel中的编码坐标或者数字索引坐标向单元格中写入数据，

# 下标从1开始，参style表示字体的颜色的名字,比如red，green

if coordinate is not None:

try:

# sheet.cell(coordinate = coordinate).value = content

sheet[coordinate] = content

if style is not None:

sheet[coordinate].\

font = Font(color = self.RGBDict[style])

self.workbook.save(self.excelFile)

except Exception as e:

raise e

elif coordinate == None and rowNo is not None and \

colsNo is not None:

try:

sheet.cell(row = rowNo,column = colsNo).value = content

if style:

sheet.cell(row = rowNo,column = colsNo).\

font = Font(color = self.RGBDict[style])

self.workbook.save(self.excelFile)

except Exception as e:

raise e

else:

raise Exception("Insufficient Coordinates of cell !")

def writeCellCurrentTime(self, sheet, coordinate = None,

rowNo = None, colsNo = None):

# 写入当前的时间，下标从1开始

now = int(time.time())  #显示为时间戳

timeArray = time.localtime(now)

currentTime = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)

if coordinate is not None:

try:

sheet.cell(coordinate = coordinate).value = currentTime

self.workbook.save(self.excelFile)

except Exception as e:

raise e

elif coordinate == None and rowNo is not None \

and colsNo is not None:

try:

sheet.cell(row = rowNo, column = colsNo

).value = currentTime

self.workbook.save(self.excelFile)

except Exception as e:

raise e

else:

raise Exception("Insufficient Coordinates of cell !")

if __name__ == '__main__':

# 测试代码

pe = ParseExcel()

pe.loadWorkBook(r'D:\ProgramSourceCode\Python Source Code\WorkSpace\InterfaceFrame2018\inter_test_data.xlsx')

sheetObj = pe.getSheetByName(u"API")

print("通过名称获取sheet对象的名字：", sheetObj.title)

# print help(sheetObj.rows)

print("通过index序号获取sheet对象的名字：", pe.getSheetByIndex(0).title)

sheet = pe.getSheetByIndex(0)

print(type(sheet))

print(pe.getRowsNumber(sheet))  #获取最大行号

print(pe.getColsNumber(sheet))  #获取最大列号

rows = pe.getRow(sheet, 1)  #获取第一行

for i in rows:

print(i.value)

# # 获取第一行第一列单元格内容

# print pe.getCellOfValue(sheet, rowNo = 1, colsNo = 1)

# pe.writeCell(sheet, u'我爱祖国', rowNo = 10, colsNo = 10)

# pe.writeCellCurrentTime(sheet, rowNo = 10, colsNo = 11)

3.2 封装get/post请求(HttpClient.py)

import requests

import jsonclassHttpClient(object):

def __init__(self):

pass

def request(self, requestMethod, requestUrl, paramsType,

requestData, headers =None, **kwargs):

ifrequestMethod =="post":

print("---", requestData, type(requestData))

ifparamsType =="form":

response = self.__post(url = requestUrl, data = json.dumps(eval(requestData)),

headers = headers, **kwargs)

return response

elif paramsType =="json":

response = self.__post(url = requestUrl, json = json.dumps(eval(requestData)),

headers = headers, **kwargs)

return response

elif requestMethod =="get":

request_url = requestUrl

ifparamsType =="url":

request_url ="%s%s"%(requestUrl, requestData)

response = self.__get(url = request_url,params= requestData, **kwargs)

return response

def __post(self, url, data = None, json = None, headers=None,**kwargs):

print("----")

response = requests.post(url=url, data = data, json=json, headers=headers)

return response

def __get(self, url, params= None, **kwargs):

response = requests.get(url,params=params, **kwargs)

return responseif__name__ =="__main__":

hc = HttpClient()

res = hc.request("get","http://39.106.41.11:8080/getBlogContent/","url",'2')

print(res.json())

3.3 封装MD5(md5_encrypt)

import hashlib

def md5_encrypt(text):

m5 = hashlib.md5()

m5.update(text.encode("utf-8"))

value = m5.hexdigest()

return valueif__name__ =="__main__":

print(md5_encrypt("sfwe"))

3.4 封装Log

import logging

import logging.configfrom config.public_data import baseDir

# 读取日志配置文件

logging.config.fileConfig(baseDir +"\config\Logger.conf")

# 选择一个日志格式

logger = logging.getLogger("example02")#或者example01

def debug(message):

# 定义dubug级别日志打印方法

logger.debug(message)

def info(message):

# 定义info级别日志打印方法

logger.info(message)

def warning(message):

# 定义warning级别日志打印方法

logger.warning(message)

3.5 封装发送Email类

import smtplibfrom email.mime.text import MIMETextfrom email.mime.multipart import MIMEMultipartfrom email.header import HeaderfromProjVar.varimport *import os

import smtplibfrom email import encodersfromemail.mime.base import MIMEBasefrom email.mime.text import MIMETextfrom email.mime.multipart import MIMEMultipartfrom email.header import Headerfrom email.utils import formataddr

def send_mail():

mail_host="smtp.126.com"  #设置服务器

mail_user="testman1980"    #用户名

mail_pass="wulaoshi1980"  #口令

sender ='testman1980@126.com'    receivers = ['2055739@qq.com',"testman1980@126.com"] # 接收邮件，可设置为你的QQ邮箱或者其他邮箱

# 创建一个带附件的实例

message = MIMEMultipart()

message['From'] = formataddr(["光荣之路吴老师","testman1980@126.com"])

message['To'] =','.join(receivers)

subject ='自动化测试执行报告'    message['Subject'] = Header(subject,'utf-8')

message["Accept-Language"]="zh-CN"    message["Accept-Charset"]="ISO-8859-1,utf-8,gbk"    # 邮件正文内容

message.attach(MIMEText('最新执行的自动化测试报告，请参阅附件内容！','plain','utf-8'))

# 构造附件1，传送测试结果的excel文件

att = MIMEBase('application','octet-stream')

att.set_payload(open(ProjDirPath+"\\testdata\\testdata.xlsx",'rb').read())

att.add_header('Content-Disposition','attachment', filename=('gbk','',"自动化测试报告.xlsx"))

encoders.encode_base64(att)

message.attach(att)

"""    # 构造附件2，传送当前目录下的 runoob.txt 文件

att2 = MIMEText(open('e:\\a.py','rb').read(),'base64','utf-8')

att2["Content-Type"] ='application/octet-stream'    att2["Content-Disposition"] ='attachment; filename="a.py"'    message.attach(att2)

"""try:

smtpObj = smtplib.SMTP(mail_host)

smtpObj.login(mail_user, mail_pass)

smtpObj.sendmail(sender, receivers, message.as_string())

print("邮件发送成功")

except smtplib.SMTPException as e:

print("Error: 无法发送邮件", e)if__name__ =="__main__":

send_mail()

四、 创建config包 用来存放公共的参数、配置文件、长时间不变的变量值

创建public_data.py

import os

# 整个项目的根目录绝对路劲

baseDir = os.path.dirname(os.path.dirname(__file__))

# 获取测试数据文件的绝对路径

file_path = baseDir +"/TestData/inter_test_data.xlsx"API_apiName =2API_requestUrl =3API_requestMothod =4API_paramsType =5API_apiTestCaseFileName =6API_active =7CASE_requestData =1CASE_relyData =2CASE_responseCode =3CASE_responseData =4CASE_dataStore =5CASE_checkPoint =6CASE_active =7CASE_status =8CASE_errorInfo =9# 存储请求参数里面依赖的数据

REQUEST_DATA = {}

# 存储响应对象中的依赖数据

RESPONSE_DATA = {}if__name__=="__main__":

print(file_path)

print(baseDir)

五、创建TestData目录，用来存放测试文件

inter_test_data.xlsx

六、创建action包，用来存放关键字函数

6.1 解决数据依赖 (GetRely.py)

from config.public_data import REQUEST_DATA, RESPONSE_DATAfrom utils.md5_encrypt import md5_encrypt

REQUEST_DATA = {"用户注册":{"1":{"username":"zhangsan","password":"dfsdf23"},

"headers":{"cookie":"asdfwerw"}}}

RESPONSE_DATA = {"用户注册":{"1":{"code":"00"},"headers":{"age":2342}}}classGetRely(object):

def __init__(self):

pass

@classmethod

def get(self, dataSource, relyData, headSource = {}):

print(type(dataSource))

print(dataSource)

data = dataSource.copy()

forkey, valuein relyData.items():

ifkey =="request":

#说明应该去REQUEST_DATA中获取

fork, vin value.items():

interfaceName, case_idx = v.split("->")

val = REQUEST_DATA[interfaceName][case_idx][k]

ifk =="password":

data[k] = md5_encrypt(val)

else:

data[k] = val

elif key =="response":

# 应该去RESPONSE_DATA中获取

fork, vin value.items():

interfaceName, case_idx = v.split("->")

data[k] = RESPONSE_DATA[interfaceName][case_idx][k]

elif key =="headers":

if headSource:

forkey, valuein value.items():

ifkey =="request":

fork, vin value.items():

foriin v:

headSource[i] = REQUEST_DATA[k]["headers"][i]

elif key =="response":

fori, valin value.items():

forjin val:

headSource[j] = RESPONSE_DATA[i]["headers"][j]

return"%s"%dataif__name__ =="__main__":

s = {"username":"","password":"","code":""}

h = {"cookie":"123","age":332}

rely = {"request": {"username":"用户注册->1","password":"用户注册->1"},

"response":{"code":"用户注册->1"},

"headers":{"request":{"用户注册":["cookie"]},"response":{"用户注册":["age"]}}

}

print(GetRely.get(s, rely, h))

6.2 解决数据存储(RelyDataStore.py)

from config.public_data import RESPONSE_DATA, REQUEST_DATAclassRelyDataStore(object):

def __init__(self):

pass

@classmethod

def do(cls, storePoint, apiName, caseId, request_source = {}, response_source = {}, req_headers={}, res_headers = {}):

forkey, valuein storePoint.items():

ifkey =="request":

# 说明需要存储的依赖数据来自请求参数,应该将数据存储到REQUEST_DATA

foriin value:

ifiin request_source:

val = request_source[i]

ifapiName notin REQUEST_DATA:

# 说明存储数据的结构还未生成，需要指明数据存储结构

REQUEST_DATA[apiName]={str(caseId): {i: val}}

else:

#说明存储数据结构中最外层结构已存在

ifstr(caseId)in REQUEST_DATA[apiName]:

REQUEST_DATA[apiName][str(caseId)][i] = val

else:

# 说明内层结构不完整，需要指明完整的结构

REQUEST_DATA[apiName][str(caseId)] = {i: val}

else:

print("请求参数中不存在字段"+ i)

elif key =="response":

#说明需要存储的依赖数据来自接口的响应body,应该将数据存储到RESPONSE_DATA

forjin value:

ifjin response_source:

val = response_source[j]

ifapiName notin RESPONSE_DATA:

# 说明存储数据的结构还未生成，需要指明数据存储结构

RESPONSE_DATA[apiName]={str(caseId): {j: val}}

else:

#说明存储数据结构中最外层结构已存在

ifstr(caseId)in RESPONSE_DATA[apiName]:

RESPONSE_DATA[apiName][str(caseId)][j] = val

else:

# 说明内层结构不完整，需要指明完整的结构

RESPONSE_DATA[apiName][str(caseId)] = {j: val}

else:

print("接口的响应body中不存在字段"+ j)

elif key =="headers":

fork, vin value.items():

ifk =="request":

# 说明需要往REQUEST_DATA变量中写入存储数据

foritemin v:

ifitemin req_headers:

header = req_headers[item]

if"headers"in REQUEST_DATA[apiName]:

REQUEST_DATA[apiName]["headers"][item] = header

else:

REQUEST_DATA[apiName]["headers"] = {item: header}

elif k =="response":

# 说明需要往RESPONSE_DATA变量中写入存储数据

foritin v:

ifitin res_headers:

header = res_headers[it]

if"headers"in RESPONSE_DATA[apiName]:

RESPONSE_DATA[apiName]["headers"][it] = header

else:

RESPONSE_DATA[apiName]["headers"] = {item: header}

print(REQUEST_DATA)

print(RESPONSE_DATA)if__name__ =="__main__":

r = {"username":"srwcx01","password":"wcx123wac1","email":"wcx@qq.com"}

req_h = {"cookie":"csdfw23"}

res_h = {"age":597232}

s = {"request": ["username","password"],"response": ["userid"],"headers":{"request":["cookie"],

"response":["age"]}}

res = {"userid":12,"code":"00"}

RelyDataStore.do(s,"register",1, r, res, req_headers=req_h, res_headers=res_h)

print(REQUEST_DATA)

print(RESPONSE_DATA)

6.3 校验数据结果(CheckResult.py)

import reclassCheckResult(object):

def __init__(self):

pass

@classmethod

def check(self, responseObj, checkPoint):

responseBody = responseObj.json()

# responseBody = {"code":"","userid":12,"id":"12"}

errorKey = {}

forkey, valuein checkPoint.items():

ifkeyin responseBody:

ifisinstance(value, (str,int)):

# 等值校验

ifresponseBody[key] != value:

errorKey[key] = responseBody[key]

elif isinstance(value, dict):

sourceData = responseBody[key]

if"value"in value:

# 模糊匹配校验

regStr = value["value"]

rg = re.match(regStr,"%s"%sourceData)

if not rg:

errorKey[key] = sourceData

elif "type"in value:

# 数据类型校验

typeS = value["type"]

iftypeS =="N":

# 说明是整形校验

ifnot isinstance(sourceData,int):

errorKey[key] = sourceData

else:

errorKey[key] ="[%s] not exist"%key

return errorKeyif__name__ =="__main__":

r = {"code":"00","userid":12,"id":12}

c = {"code":"00","userid": {"type":"N"},"id": {"value":"\d+"}}

print(CheckResult.check(r, c))

6.4 往excel里面写结果

fromconfig.public_data import *def write_result(wbObj, sheetObj, responseData, errorKey, rowNum):

try:

# 写响应body

wbObj.writeCell(sheetObj, content="%s"%responseData,

rowNo = rowNum, colsNo=CASE_responseData)

# 写校验结果状态及错误信息

if errorKey:

wbObj.writeCell(sheetObj, content="%s"%errorKey,

rowNo=rowNum, colsNo=CASE_errorInfo)

wbObj.writeCell(sheetObj, content="faild",

rowNo=rowNum, colsNo=CASE_status, style="red")

else:

wbObj.writeCell(sheetObj, content="pass",

rowNo=rowNum, colsNo=CASE_status, style="green")

except Exception as err:

raise err

七、创建Log目录用来存放日志

八、主函数

#encoding=utf-8import requests

import jsonfrom action.get_rely import GetRelyfromconfig.public_data import *from utils.ParseExcel import ParseExcelfrom utils.HttpClient import HttpClientfrom action.data_store import RelyDataStorefrom action.check_result import CheckResultfrom action.write_result import write_resultfromutils.Log import *def main():

parseE = ParseExcel()

parseE.loadWorkBook(file_path)

sheetObj = parseE.getSheetByName("API")

activeList = parseE.getColumn(sheetObj, API_active)

foridx, cellinenumerate(activeList[1:],2):

ifcell.value =="y":

#需要被执行

RowObj = parseE.getRow(sheetObj, idx)

apiName = RowObj[API_apiName -1].value

requestUrl = RowObj[API_requestUrl -1].value

requestMethod = RowObj[API_requestMothod -1].value

paramsType = RowObj[API_paramsType -1].value

apiTestCaseFileName = RowObj[API_apiTestCaseFileName -1].value

# 下一步读取用例sheet表，准备执行测试用例

caseSheetObj = parseE.getSheetByName(apiTestCaseFileName)

caseActiveObj = parseE.getColumn(caseSheetObj, CASE_active)

forc_idx, colinenumerate(caseActiveObj[1:],2):

ifcol.value =="y":

#需要执行的用例

caseRowObj = parseE.getRow(caseSheetObj, c_idx)

requestData = caseRowObj[CASE_requestData -1].value

relyData = caseRowObj[CASE_relyData -1].value

responseCode = caseRowObj[CASE_responseCode -1].value

responseData = caseRowObj[CASE_responseData -1].value

dataStore = caseRowObj[CASE_dataStore -1].value

checkPoint = caseRowObj[CASE_checkPoint -1].value

#发送接口请求之前需要做一下数据依赖的处理

if relyData:

logging.info("处理第%s个接口的第%s条用例的数据依赖！")

requestData = GetRely.get(eval(requestData), eval(relyData))

httpC = HttpClient()

response = httpC.request(requestMethod=requestMethod,

requestData=requestData,

requestUrl=requestUrl,

paramsType=paramsType

)

# 获取到响应结果后，接下来进行数据依赖存储逻辑实现

ifresponse.status_code ==200:

responseData = response.json()

# 进行依赖数据存储

if dataStore:

RelyDataStore.do(eval(dataStore), apiName, c_idx -1, eval(requestData), responseData)

# 接下来就是校验结果

else:

logging.info("接口【%s】的第【%s】条用例，不需要进行依赖数据存储！"%(apiName, c_idx))

if checkPoint:

errorKey = CheckResult.check(response, eval(checkPoint))

write_result(parseE, caseSheetObj, responseData, errorKey, c_idx)

else:

logging.info("接口【%s】的第【%s】条用例，执行失败，接口协议code非200！"%(apiName, c_idx))

else:

logging.info("第%s个接口的第%s条用例，被忽略执行！"%(idx -1, c_idx-1))

else:

logging.info("第%s行的接口被忽略执行！"%(idx -1))if__name__=="__main__":

main()

框架待完善~~请多多指教~~

笔者这边创建了一个软件测试技术交流群1125760266，欢迎大家加入，共同探讨学习！