为什么要做接口自动化框架
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 json class HttpClient(object): def __init__(self): pass def request(self, requestMethod, requestUrl, paramsType, requestData, headers =None, **kwargs): if requestMethod == "post": print("---", requestData, type(requestData)) if paramsType == "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 if paramsType == "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 response if __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 value if __name__ == "__main__": print(md5_encrypt("sfwe"))
3.4 封装Log
import logging import logging.config from 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 smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header from ProjVar.var import * import os import smtplib from email import encoders from email.mime.base import MIMEBase from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header from 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 = 2 API_requestUrl = 3 API_requestMothod = 4 API_paramsType = 5 API_apiTestCaseFileName = 6 API_active = 7 CASE_requestData = 1 CASE_relyData = 2 CASE_responseCode = 3 CASE_responseData = 4 CASE_dataStore = 5 CASE_checkPoint = 6 CASE_active = 7 CASE_status = 8 CASE_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_DATA from utils.md5_encrypt import md5_encrypt REQUEST_DATA = {"用户注册":{"1":{"username":"zhangsan", "password":"dfsdf23"}, "headers":{"cookie":"asdfwerw"}}} RESPONSE_DATA = {"用户注册":{"1":{"code":"00"}, "headers":{"age":2342}}} class GetRely(object): def __init__(self): pass @classmethod def get(self, dataSource, relyData, headSource = {}): print(type(dataSource)) print(dataSource) data = dataSource.copy() for key, value in relyData.items(): if key == "request": #说明应该去REQUEST_DATA中获取 for k, v in value.items(): interfaceName, case_idx = v.split("->") val = REQUEST_DATA[interfaceName][case_idx][k] if k == "password": data[k] = md5_encrypt(val) else: data[k] = val elif key == "response": # 应该去RESPONSE_DATA中获取 for k, v in value.items(): interfaceName, case_idx = v.split("->") data[k] = RESPONSE_DATA[interfaceName][case_idx][k] elif key == "headers": if headSource: for key, value in value.items(): if key == "request": for k, v in value.items(): for i in v: headSource[i] = REQUEST_DATA[k]["headers"][i] elif key == "response": for i, val in value.items(): for j in val: headSource[j] = RESPONSE_DATA[i]["headers"][j] return "%s" %data if __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_DATA class RelyDataStore(object): def __init__(self): pass @classmethod def do(cls, storePoint, apiName, caseId, request_source = {}, response_source = {}, req_headers={}, res_headers = {}): for key, value in storePoint.items(): if key == "request": # 说明需要存储的依赖数据来自请求参数,应该将数据存储到REQUEST_DATA for i in value: if i in request_source: val = request_source[i] if apiName not in REQUEST_DATA: # 说明存储数据的结构还未生成,需要指明数据存储结构 REQUEST_DATA[apiName]={str(caseId): {i: val}} else: #说明存储数据结构中最外层结构已存在 if str(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 for j in value: if j in response_source: val = response_source[j] if apiName not in RESPONSE_DATA: # 说明存储数据的结构还未生成,需要指明数据存储结构 RESPONSE_DATA[apiName]={str(caseId): {j: val}} else: #说明存储数据结构中最外层结构已存在 if str(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": for k, v in value.items(): if k == "request": # 说明需要往REQUEST_DATA变量中写入存储数据 for item in v: if item in 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变量中写入存储数据 for it in v: if it in 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 re class CheckResult(object): def __init__(self): pass @classmethod def check(self, responseObj, checkPoint): responseBody = responseObj.json() # responseBody = {"code": "", "userid": 12, "id": "12"} errorKey = {} for key, value in checkPoint.items(): if key in responseBody: if isinstance(value, (str, int)): # 等值校验 if responseBody[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"] if typeS == "N": # 说明是整形校验 if not isinstance(sourceData, int): errorKey[key] = sourceData else: errorKey[key] = "[%s] not exist" %key return errorKey if __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里面写结果
from config.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-8 import requests import json from action.get_rely import GetRely from config.public_data import * from utils.ParseExcel import ParseExcel from utils.HttpClient import HttpClient from action.data_store import RelyDataStore from action.check_result import CheckResult from action.write_result import write_result from utils.Log import * def main(): parseE = ParseExcel() parseE.loadWorkBook(file_path) sheetObj = parseE.getSheetByName("API") activeList = parseE.getColumn(sheetObj, API_active) for idx, cell in enumerate(activeList[1:], 2): if cell.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) for c_idx, col in enumerate(caseActiveObj[1:], 2): if col.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 ) # 获取到响应结果后,接下来进行数据依赖存储逻辑实现 if response.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()
框架待完善~~请各路神仙多多指教~~
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
稳了!魔兽国服回归的3条重磅消息!官宣时间再确认!
昨天有一位朋友在大神群里分享,自己亚服账号被封号之后居然弹出了国服的封号信息对话框。
这里面让他访问的是一个国服的战网网址,com.cn和后面的zh都非常明白地表明这就是国服战网。
而他在复制这个网址并且进行登录之后,确实是网易的网址,也就是我们熟悉的停服之后国服发布的暴雪游戏产品运营到期开放退款的说明。这是一件比较奇怪的事情,因为以前都没有出现这样的情况,现在突然提示跳转到国服战网的网址,是不是说明了简体中文客户端已经开始进行更新了呢?