接口自动化-yaml文件维护自动化用例

最近团长工作中收到了后台回归测试的任务，火急火燎的写了几个主要流程的接口自动化回归脚本。刚开始想着就几个接口那就直接调用吧，写出来是这样的：

class ops_ris():
def __init__(self,env,phase_no='IG_TELCHECK',risk_desc='AGENCY'):
self.env=env
self.phase_no=phase_no
self.risk_desc=risk_desc
def investigation(self):
"""调查处理无风险经办结案"""
#转派
self.transfercase()
#查询最近一笔xiejinggeng名下经办待处理案件号
# sql = "SELECT case_no FROM ig_case where handle_user='xiejinggeng'  and state in ('TCED','TCW')  ORDER BY id desc limit 1"
# self.case_no = DBmananger(self.env, 'ris').callMysql(sql)[0]["case_no"]
# 查询最近一笔case_no案件号IG_TELCHECK阶段对应的task_no
sql=f"select task_no from wf_task where biz_no='{self.case_no}' and phase_no='{self.phase_no}' order by id desc limit 1"
task_no=DBmananger(self.env, 'ris').callMysql(sql)[0]["task_no"]
#登录获取sid
self.sid=get_sid(self.env)
#调查经办结案
method,bizContent=invest_first(self.case_no,task_no)
self.header=headers(self.sid)
self.url=envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json', method='post',headers=self.header)
if res['flag'] =="S":
logger.debug(f"案件{self.case_no}已经办处理成功")
else:
logger.error(f"案件{self.case_no}经办处理失败，失败原因:{res['msg']}")
test_name = ["中介类型调查经办处理", "盗用类型调查经办处理", "伪冒类型调查经办处理", "其他类型调查经办处理"]
if self.risk_desc == "AGENCY":
test_Name = test_name[0]
elif self.risk_desc == "THEFT":
test_Name = test_name[1]
elif self.risk_desc == "FAKE":
test_Name = test_name[2]
else:
test_Name = test_name[3]
return {
"测试用例名称":test_Name,
"测试结果":"pass" if res['flag'] =="S" else "fail",
"原因":res["msg"],
"url":self.url,
"res": res,
"param":GetCommonContent(method, bizContent)
}
def transfercase(self):
"""调查案件案件转派"""
sql=f"select case_no from ig_case where risk_desc='{self.risk_desc}' and state in ('TCED','TCW')  ORDER BY id desc limit 1"
self.case_no = DBmananger(self.env, 'ris').callMysql(sql)[0]["case_no"]
sql=f"select task_no,biz_no from wf_task where phase_no='{self.phase_no}' and biz_no='{self.case_no}' ORDER BY id desc limit 1"
task_no = DBmananger(self.env, 'ris').callMysql(sql)[0]["task_no"]
self.case_no = DBmananger(self.env, 'ris').callMysql(sql)[0]["biz_no"]
method, bizContent = transfer(self.case_no, task_no)
# 登录获取sid
self.sid = get_sid(self.env)
self.header = headers(self.sid)
self.url = envhost[self.env]
self.res=apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',method='post', headers=self.header)
# 调查转派
if self.res['flag'] =="S":
logger.debug(f"案件{self.case_no}已经转派成功")
else:
logger.error(f"案件{self.case_no}经办转派失败，失败原因:{res['msg']}")
return {
"测试用例名称": "调查案子转派",
"测试结果": "pass" if self.res['flag'] == "S" else "fail",
"原因": self.res["msg"],
"url": self.url,
"res": self.res,
"param": GetCommonContent(method, bizContent)
}
def review(self):
"""调查案件复核"""
test_name = ["中介类型调查复核处理", "盗用类型调查复核处理", "伪冒类型调查复核处理", "其他类型调查复核处理"]
if self.risk_desc == "AGENCY":
test_Name = test_name[0]
elif self.risk_desc == "THEFT":
test_Name = test_name[1]
elif self.risk_desc == "FAKE":
test_Name = test_name[2]
else:
test_Name = test_name[3]
if self.res['flag'] == "F":
return {
"测试用例名称": test_Name,
"测试结果": "skip" ,
"原因": "经办失败跳过复核",
"url": self.url,
"res": '',
"param": ''
}
else:
#检测案件状态，结案直接pass  复核阶段调复核处理
sql=f"select state from ig_case where case_no='{self.case_no}' "
state = DBmananger(self.env, 'ris').callMysql(sql)[0]["state"]
if state=="PS":
logger.debug(f"案件{self.case_no}已经结案")
else:
sql=f"select task_no,state from wf_task where phase_no='IG_LASTINSTANCE' and biz_no='{self.case_no}' ORDER BY id desc limit 1"
task_no = DBmananger(self.env, 'ris').callMysql(sql)[0]["task_no"]
sql=f"select remark from m_opinion where case_no='{self.case_no}' and phase_no='IG_TELCHECK'"
remark=DBmananger(self.env, 'ris').callMysql(sql)[0]["remark"]
method, bizContent = transfer(self.case_no, task_no)
# 登录获取sid
self.sid = get_sid(self.env)
self.header = headers(self.sid)
self.url = envhost[self.env]
apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',method='post', headers=self.header)
method, bizContent = inreview(self.case_no, task_no,remark)
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json', method='post', headers=self.header)
# 调查复核结案
if res['flag'] == "S":
logger.debug(f"案件{self.case_no}结案成功")
else:
#再处理一次复核，有缓存
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json', method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.case_no}结案成功")
else:
logger.debug(f"案件{self.case_no}结案失败，原因：{res['msg']}")
# test_name = ["中介类型调查复核处理", "盗用类型调查复核处理", "伪冒类型调查复核处理", "其他类型调查复核处理"]
# if self.risk_desc == "AGENCY":
#     test_Name = test_name[0]
# elif self.risk_desc == "THEFT":
#     test_Name = test_name[1]
# elif self.risk_desc == "FAKE":
#     test_Name = test_name[2]
# else:
#     test_Name = test_name[3]
return {
"测试用例名称": test_Name,
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
class tel():
def __init__(self,env,investigate="Y",investigateType1="N",investigateType2="N",investigateType3="Y",investigateType4="N",biz_no=''):
self.env = env
#调查类型：investigateType1-中介，investigateType2-盗用，investigateType3-伪冒，investigateType4-其他
self.investigate=investigate
self.investigateType1=investigateType1
self.investigateType2 = investigateType2
self.investigateType3 = investigateType3
self.investigateType4 = investigateType4
self.biz_no=biz_no
def transfercase(self):
"""电核案件转派"""
#倒序查xiejinggeng名下电核最新一笔电核案子
if self.biz_no=='':
sql = "select user_no,task_no,biz_no from wf_task where  phase_no='AP_TELCHECK' and switch_flag='SW_AP_360_JIE' and state='DISPATHED'  ORDER BY id DESC  limit 1"
else:
sql=f"select user_no,task_no,biz_no from wf_task where biz_no ='{self.biz_no}' and phase_no='AP_TELCHECK' ORDER BY id DESC limit 1"
self.user_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["user_no"]
self.task_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["task_no"]
self.biz_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["biz_no"]
if self.user_no=="xiejinggeng":
# 登录获取sid
self.sid = get_sid(self.env)
method, bizContent = transfer_tel(self.biz_no,self.task_no,'genggeng')
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}转派成功")
else:
logger.debug(f"案件{self.biz_no}转派失败，原因：{res['msg']}")
elif self.user_no=="genggeng":
# 登录获取sid
self.sid = get_sid(self.env,login_name='genggeng')
method, bizContent = transfer_tel(self.biz_no, self.task_no, 'xiejinggeng')
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}转派成功")
else:
logger.debug(f"案件{self.biz_no}转派失败，原因：{res['msg']}")
else:
# 登录获取sid
self.sid = get_sid(self.env)
method, bizContent = transfer_tel(self.biz_no, self.task_no, 'genggeng')
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}转派成功")
else:
logger.debug(f"案件{self.biz_no}转派失败，原因：{res['msg']}")
return {
"测试用例名称": "电核案件转派",
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
def tel_check(self):
"""电话核查"""
#查下案子在谁名下
sql=f"select user_no,task_no,biz_no from wf_task where   phase_no='AP_TELCHECK' ORDER BY id DESC limit 1"
self.user_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["user_no"]
self.task_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["task_no"]
self.biz_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["biz_no"]
# 登录获取sid
self.sid = get_sid(self.env,login_name=self.user_no)
method, bizContent =  tel_check(self.biz_no,self.task_no,self.investigateType1,self.investigateType2,self.investigateType3,self.investigateType4,self.investigate)
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',
method='post', headers=self.header)
#investigateType1-中介，investigateType2-盗用，investigateType3-伪冒，investigateType4-其他
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}电核成功")
else:
logger.debug(f"案件{self.biz_no}电核失败，原因：{res['msg']}")
test_name=["电核转中介类型调查","电核转盗用类型调查","电核转伪冒类型调查","电核转其他类型调查"]
if self.investigateType1=="Y":
test_Name=test_name[0]
elif self.investigateType2=="Y":
test_Name=test_name[1]
elif self.investigateType3=="Y":
test_Name=test_name[2]
else:
test_Name = test_name[3]
return {
"测试用例名称": test_Name,
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
def final_jud(self):
"""终审"""
#先转派
sql = f"select user_no,task_no,biz_no from wf_task where biz_no ='{self.biz_no}' and phase_no='AP_LASTINSTANCE' ORDER BY id DESC limit 1"
self.user_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["user_no"]
self.task_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["task_no"]
self.biz_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["biz_no"]
if self.user_no=="xiejinggeng":
# 登录获取sid
self.sid = get_sid(self.env)
method, bizContent = transfer_tel(self.biz_no,self.task_no,'genggeng')
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}转派成功")
else:
logger.debug(f"案件{self.biz_no}转派失败，原因：{res['msg']}")
elif self.user_no=="genggeng":
# 登录获取sid
self.sid = get_sid(self.env,login_name='genggeng')
method, bizContent = transfer_tel(self.biz_no, self.task_no, 'xiejinggeng')
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}转派成功")
else:
logger.debug(f"案件{self.biz_no}转派失败，原因：{res['msg']}")
else:
# 登录获取sid
self.sid = get_sid(self.env)
method, bizContent = transfer_tel(self.biz_no, self.task_no, 'genggeng')
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent), parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}转派成功")
else:
logger.debug(f"案件{self.biz_no}转派失败，原因：{res['msg']}")
#终审
# 查下案子在谁名下
sql = f"select user_no,task_no,biz_no from wf_task where biz_no ='{self.biz_no}' and phase_no='AP_LASTINSTANCE' ORDER BY id DESC limit 1"
self.user_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["user_no"]
self.task_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["task_no"]
self.biz_no = DBmananger(self.env, 'apv_wf').callMysql(sql)[0]["biz_no"]
# 登录获取sid
self.sid = get_sid(self.env, login_name=self.user_no)
method, bizContent = fina_check(self.biz_no,self.task_no)
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent),
parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.biz_no}终审成功")
else:
logger.debug(f"案件{self.biz_no}终审失败，原因：{res['msg']}")
test_name = ["终审转中介类型调查", "终审转盗用类型调查", "终审转伪冒类型调查", "终审转其他类型调查"]
if self.investigateType1 == "Y":
test_Name = test_name[0]
elif self.investigateType2 == "Y":
test_Name = test_name[1]
elif self.investigateType3 == "Y":
test_Name = test_name[2]
else:
test_Name = test_name[3]
return {
"测试用例名称":  test_Name,
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
class Inquisition():
def __init__(self,env):
self.env=env
def Inquisitiontransfer(self):
"""征审质检转派"""
#查询最新待质检案子
sql="select appl_no from qc_case where source_channel='APV' and state='W_QC' and case_type='A24' order by id desc limit 1"
self.appl_no = DBmananger(self.env, 'ris').callMysql(sql)[0]["appl_no"]
#转派
# 登录获取sid
self.sid = get_sid(self.env)
method, bizContent = transfer_opinion(self.appl_no,"xiejinggeng")
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent),
parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.appl_no}征审质检转派给xiejinggeng成功")
else:
logger.debug(f"案件{self.appl_no}征审质检转派给xiejinggeng失败，原因：{res['msg']}")
return {
"测试用例名称": "征审质检案件转派",
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
def Inquisition_qc_case(self):
'''征审质检'''
# 登录获取sid
self.sid = get_sid(self.env)
method, bizContent = opinion(self.appl_no)
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent),
parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.appl_no}征审质检成功")
else:
logger.debug(f"案件{self.appl_no}征审质检失败，原因：{res['msg']}")
return {
"测试用例名称": "征审质检",
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
def Inquisition_feedback(self):
'''征审质检争议处理'''
# 登录获取sid
sql=f"select tel_check_user from qc_case where appl_no ={self.appl_no}"
self.tel_check_user = DBmananger(self.env, 'ris').callMysql(sql)[0]["tel_check_user"]
self.sid = get_sid(self.env,login_name=self.tel_check_user)
method, bizContent = feedback(self.appl_no)
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent),
parammode='json',
method='post', headers=self.header)
if res['flag'] == "S":
logger.debug(f"案件{self.appl_no}征审质检争议处理成功")
else:
logger.debug(f"案件{self.appl_no}征审质检争议处理失败，原因：{res['msg']}")
return {
"测试用例名称": "征审质检争议处理",
"测试结果": "pass" if res['flag'] == "S" else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
def Inquisition_opinionend(self):
'''征审质检定案'''
# 登录获取sid
self.sid = get_sid(self.env, login_name=self.tel_check_user)
method, bizContent = opinionend(self.appl_no)
self.header = headers(self.sid)
self.url = envhost[self.env]
res = apiRequests.callInterface(url=self.url, param=GetCommonContent(method, bizContent),
parammode='json',
method='post', headers=self.header)
sql=f"select state from qc_case where appl_no ={self.appl_no}"
self.state = DBmananger(self.env, 'ris').callMysql(sql)[0]["state"]
if res['flag'] == "S" and self.state=='VDT':
logger.debug(f"案件{self.appl_no}征审质检定案成功")
else:
logger.debug(f"案件{self.appl_no}征审质检定案失败，原因：{res['msg']},案件状态：{self.state}")
return {
"测试用例名称": "征审质检定案",
"测试结果": "pass" if res['flag'] == "S" and self.state=='VDT' else "fail",
"原因": res["msg"],
"url": self.url,
"res": res,
"param": GetCommonContent(method, bizContent)
}
def nowtime():
now_time = datetime.datetime.now()
now = datetime.datetime.strftime(now_time, '%Y-%m-%d %H:%M:%S')
# date_now=datetime.datetime.strftime(now_time, '%Y-%m-%d')
date = (int(time.mktime(time.strptime(now, "%Y-%m-%d %H:%M:%S"))))
return now, date

把接口数据直接写到程序里面了，运行起来很爽。不过写过的都知道维护会很麻烦，接口要是改动了要改代码，也不好扩展，还是数据分离吧，借鉴httprunner的方式，将用例放到yaml维护吧（放到excle也可以，之前有些过这种感兴趣的可以翻阅这个文档添加链接描述）
格式如下：

- CaseNo: 1
Extract:
biz_no:
sql-apv_wf: select ***
taskNo:
sql-apv_wf: select ***
user: 'genggeng'
headers:
fun: get_sid(login_name='xiejinggeng')
CaseNmae: 电核转派
Url: ''
Method: post
param:
method: qihoo.apv.ma.caseturn.update
bizContent:
taskNoList:
- key: $biz_no$
value: $taskNo$
productCode: 360借条
toUser: $user$
headers:
Cookie: $headers$

介绍：
Extract这个key里面的是变量相关的，支持sql操作、自定义函数，可通过正则区分sql还是自定义函数，如：sql-开头判断是数据库操作的，后面接操作的库名apv_wf；fun则是自定义函数，到置顶目录去查找并执行get_sid(login_name=‘xiejinggeng’)。变量用变量变量变量包括起来，然后替换。如下

from quality_management_logic.ops_new.PublicCenter.Case_Get_Data import *
from main import BASE_DIR
from quality_management_logic.commonCenter.logUtil import  getlog
from quality_management_logic.ops.publicCenter import apiRequests
from quality_management_logic.ops.dataCenter.env import envhost
logger = getlog(targetName='RunCase')
#yaml文件list
casepath=f"{BASE_DIR}/TestCaseCenter/"
casefilelist=File_Name(casepath)
def Run(env):
'''
执行用例的入口
:return:
'''
#读yaml用例文件转换成json格式数据
for yamlfile in casefilelist:
caselist=yaml_r(yamlfile)
logger.debug(f"获取用例list:{caselist}")
#用例中用到的sql、函数预执行替换成新用例
for case in caselist:
#存在变量就替换，反之直接执行用例
if 'Extract' in case.keys():
logger.debug(f"用例:{case.get('CaseNo')}存在变量:{case.get('Extract').keys()}")
for key in case.get('Extract').keys():
extract=to_extract(key)
#是sql
if GetSql(case.get('Extract')[key],env=env):
extract_value= GetSql(case.get('Extract')[key],env=env)
#是自定义函数
elif GetFun(case.get('Extract')[key],pattern='fun'):
extract_value = GetFun(case.get('Extract')[key],pattern='fun')
#是常量
else:
extract_value=case.get('Extract')[key]
#替换
# update_allvalues(caselist, key, extract_value)
update_allvalue(caselist, extract, extract_value)
logger.debug(f"获取用例list:{caselist}")
#开始执行替换后的测试用例
for case in caselist:
url=envhost[env]
param=case["param"]
headers=case["headers"]
res = apiRequests.callInterface(url=url, param=param, parammode='json', method='post', headers=headers)
if __name__=='__main__':
print(Run(env='stg1'))

大功告成

本文地址：https://blog.csdn.net/kairui_guxiaobai/article/details/110472505

黄山市民网：https://www.huangshanshimin.com/

相关文章