2023年7月10日发(作者：)

python接⼝⾃动化测试框架pdf_Python接⼝⾃动化测试框架Python接⼝⾃动化测试框架在⾃动化的测试体系中，包含了UI⾃动化测试和接⼝⾃动化测试，UI⾃动化实现的前提是软件版本进⼊稳定期，UI界⾯稳定、变动少，相⽐较之下接⼝⾃动化，接⼝受外界因素的影响较少，维护成本低，能够在最短时间发现问题。⼀、浅谈接⼝测试1、什么是接⼝测试：API测试⼜称为接⼝测试，主要⽤于检测外部系统与系统之间以及内部各个⼦系统之间的交互点，是对系统接⼝功能进⾏测试的⼀种⼿段，也是集成测试的⼀部分，通过直接控制被测应⽤的接⼝(API)来确定是否在功能、可靠性、性能和安全⽅⾯达到预期的软件测试活动。2、如何测试接⼝：检查接⼝返回的数据是否与预期结果⼀致。检查接⼝的容错性，假如传递数据的类型错误时是否可以处理。接⼝参数的边界值。例如，传递的参数⾜够⼤或为负数时，接⼝是否可以正常处理。接⼝的性能，http请求接⼝⼤多与后端代码逻辑、执⾏的SQL语句性能、算法等相关。接⼝的安全性，外部调⽤的接⼝尤为重要。3、接⼝测试的意义、⽬的：接⼝测试的核⼼意义、⽬的在于：以保证系统的正确和稳定为核⼼，以持续集成为⼿段，提⾼测试效率，提升⽤户体验，降低产品研发成本。⼆、定义框架⽬录分层接⼝⾃动化框架的没有统⼀标准，可以根据实际需要⾃定义，以满⾜功能测试⽬标要求为⽬的。⼀个基本的接⼝⾃动化测试框架需要满⾜的功能：测试⽤例管理、各种配置信息管理、数据库操作、⽇志打印、报告输出、邮件发送"""API_Autotest/|-- API_Case/ #测试⽤例| | |-- PreviewRelease_01_ #预发布环境注册登录接⼝case| | |-- PreviewRelease_02_ #预发布环境主流程接⼝case| | |-- PreviewRelease_03_ #预发布环境.......接⼝case| | |-- Test_01_ #测试环境注册登录接⼝case| | |-- Test_02_ #测试环境主流程接⼝case| | |-- Test_03_ #测试环境......接⼝case| | ||-- data/ ###配置信息| |-- conf_ #账号密码等配置信息| |-- custom_ #向接⼝请求的请求的参数变量、keys、请求头| |-- export_ #接⼝url| |-- request_ #向接⼝请求的请求参数||-- logic/ ###主要逻辑| |-- export_ #接⼝实现主逻辑| |-- #数据库操作类| |-- log_ #⽇志打印类| |-- public_ #公⽤函数类| |-- send_ #发送邮件类||-- log/ #⽇志| |-- Case--2_log #根据运⾏⽇期保存操作⽇志| |-- Case--2_log||-- report/ #测试报告| |-- #根据运⾏⽇期保存测试报告⽇志| |-- ||-- #readme||-- #接⼝case运⾏管理类"""三、知识技能储备萝⼘青菜各有所爱，每个⼈⼼中的接⼝⾃动化测试框架也各不相同，想实现⼀个基础功能完备的接⼝测试框架需要的Python知识如下：1、Python基础知识列举了⼀些需要掌握的基础知识2、主要依赖的库下⾯介绍的库都是数据库操作、⽇志打印、报告输出、邮件发送功能实现所依赖的库：a、数据库操作数据库操作主要pymysql库，下⾯为代码⽰例：importpymysqlimportdatetime,timeimportosfrom _dict importConfDatafrom _print _class importPublicclassConnectDatabase():"""连接数据库类"""def __init__(self):=LOG()_data=ConfData()#连接数据库tion_110 = t(host=__conf_data("database_110", "host"),port=__conf_data("database_110", "port"),user=__conf_data("database_110", "user"),password=__conf_data("database_110", "password"),db=__conf_data("database_110", "db"),charset='utf8',#以字典形式展⽰所查询数据cursorclass=rsor)#保存错误⽇志的⽂件名称_name = _new_file(path=(e(e(__file__)), "log"))def select_sql(self,**kwargs):"""根据传⼊参数执⾏数据库查询操作:param args::param kwargs: database:(database_110)选择数据库、table：表名、condition：where条件:return:"""database_name= ("database")field_name= ("field", "*")table_name= ("table")where_condition= ("condition")if database_name == "database_110":try:with tion_() as cursor:sql= "SELECT %s FROM %s WHERE %s;"data=(field_name, table_name, where_condition)e(sql%data)tion_()result=ne()returnresultexceptException as e:_warning("database", _name, "select_error：%s" %e)def select_sql_all(self, *args, **kwargs):"""根据传⼊参数执⾏数据库查询操作:param args::param kwargs: database:(database_110)选择数据库、table：表名、condition：where条件:return:"""database_name= ("database")field_name= ("field", "*")table_name= ("table")where_condition= ("condition")if database_name == "database_110":try:with tion_() as cursor:sql= "SELECT %s FROM %s WHERE %s;"data=(field_name, table_name, where_condition)print((sql %data))e(sql%data)tion_()result=ll()returnresultexceptException as e:_warning("database", _name, "select_error：%s" %e)def update_sql(self, *args, **kwargs):"""根据传⼊参数执⾏数据库更新操作:param args::param kwargs: database:(database_110)选择数据库、table：表名、set：更新的字段和值、condition：where条件:return:"""database_name= ("database")table_name= ("table")set_value= ("set")where_condition= ("condition")if database_name == "database_110":try:with tion_() as cursor:sql= "UPDATE %s SET %s WHERE %s;"data=(table_name, set_value, where_condition)#print(sql%data)e(sql %data)tion_()()exceptException as e:_warning("database", _name, "update_error：%s" %e)tion_ck()def delete_sql(self, *args, **kwargs):"""根据传⼊参数执⾏数据库删除操作:param args::param kwargs: database:(database_110)选择数据库、table：表名、condition：where条件:return:"""database_name= ("database")table_name= ("table")where_condition= ("condition")if database_name == "database_110":try:with tion_() as cursor:sql= "DELETE from %s where %s;"data=(table_name, where_condition)e(sql%data)tion_()ntexceptException as e:_warning("database", _name, "delete_error：%s" %e)tion_ck()def insert_sql(self, *args, **kwargs):"""根据传⼊参数执⾏数据库插⼊操作:param args::param kwargs: database:(database_110)选择数据库、sql:需要插⼊的sql语句:return:"""database_name= ("database")insert_sql= ("sql")if database_name == "database_110":try:with tion_() as cursor:e(insert_sql)tion_()ntexceptException as e:_warning("database", _name, "insert_error：%s" %e)tion_ck()def mysql_function(self, *args, **kwargs):"""根据传⼊参数执⾏数据库函数操作:param args::param kwargs: database:(database_110)选择数据库、function_name:函数名称，data_id：数据ID，phone:电话:return:"""database_name= ("database")function_name= ("function_name")data_id= ("data_id")phone= ("phone")product_id= ("product_id")if database_name == "database_110":try:with tion_() as cursor:oc(function_name,args=(data_id,phone,product_id,))tion_()exceptException as e:_warning("database", _name, "mysql_function：%s" %e)tion_ck()if __name__ == "__main__":b=ConnectDatabase()data_id= _short_id()⽰例代码--删减版PS：pymsql库操作mysql数据库增、删、查、改、调⽤函数b、⽇志打印⽇志打印依赖logging库，下⾯为代码⽰例：importloggingimportosimportsysimportdatetimelog_path= (e(e(__file__)), "log")(log_path)classLOG(object):"""⽇志打印类"""def __init__(self):_time= ().strftime("%Y%m%d%H%M")def log_info(self, *args):"""根据传⼊参数打印普通⽇志:param arg[0]log的功能模块名，arg[1] 保存log的⽂件名，arg[2]要打印的⽇志内容:return 返回logger 对象"""#创建⼀个logger对象logger =ger(args[0])el()#创建⼀个向屏幕输⼊的handler对象ch =Handler()el()#创建⼀个像⽂件输⼊的handler对象log_file = (log_path, "%s--%s_log" % (args[1], _time))fh= ndler(log_file, mode="a+", encoding="utf-8")el()#设置log输⼊格式formatter = ter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')matter(formatter)matter(formatter)#logger，添加handler对象dler(ch)dler(fh)(args[2])#在记录⽇志之后移除句柄, 不然会重复打印⽇志Handler(fh)Handler(ch)returnlogger@staticmethoddef log_warning(*args):"""根据传⼊参数错误⽇志:param arg[0]log的功能模块名，arg[1]⽂件名 arg[2] 需要打印的内容:return 返回logger 对象"""#创建⼀个logger对象logger =ger(args[0])el()#创建⼀个向屏幕输⼊的handler对象ch =Handler()el()#创建⼀个像⽂件输⼊的handler对象log_file = (log_path, args[1])fh= ndler(log_file, mode="a+", encoding="utf-8")el()#设置log输⼊格式formatter = ter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')matter(formatter)matter(formatter)#logger，添加handler对象dler(ch)dler(fh)g(args[2])#在记录⽇志之后移除句柄, 不然会重复打印⽇志Handler(fh)Handler(ch)print("aaa")returnlogger#@staticmethoddeflog_debug(self,message):"""根据传⼊参数错误⽇志:param arg[0]log的功能模块名，arg[1]⽂件名 arg[2] 需要打印的内容:return 返回logger 对象"""#创建Loggerlogger =ger()el()#创建Handler#终端HandlerconsoleHandler =Handler()el()#⽂件HandlerfileHandler = ndler('', mode='w', encoding='UTF-8')el()#Formatterformatter = ter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')matter(formatter)matter(formatter)#添加到Logger中dler(consoleHandler)dler(fileHandler)(message)#if __name__ == "__main__":#A=LOG()#_debug('123')⽰例代码--删减版c、报告输⼊报告输出依赖库主要为requests、unittest、下⾯为代码⽰例：importrequestsfrom _url importExportERLfrom t_dict importRequestDatafrom _printimportLOGfrom se importConnectDatabasefrom _variableimportGlobalVariableclassExportLogic(object):"""ExportLogic 接⼝主逻辑类"""def __init__(self):t=RequestData()=ExportERL()tDatabase=ConnectDatabase()=LOG()_variable=GlobalVariable()@staticmethoddef request_post(*args, **kwargs):"""传⼊请求参数,返回json请求结果:param args::param kwargs: json:请求数据, url：请求路径， header:请求头:return:"""request_data= ("json")request_url= ("url")request_header= ("header")response_json= (url=request_url, data=request_data, headers=request_header).json()returnresponse_json@staticmethoddef request_post_json(*args, **kwargs):"""传⼊请求参数,返回json请求结果:param args::param kwargs: json:请求数据, url：请求路径， header:请求头:return:"""request_data= ("json")request_url= ("url")request_header= ("header")response_json= (url=request_url, json=request_data, headers=request_header)returnresponse_json@staticmethoddef request_post_file(*args, **kwargs):"""上传接⼝独有:param args::param kwargs: json:请求数据, url：请求路径， header:请求头, file:⽂件路径:return:"""request_data= ("json")request_url= ("url")request_header= ("header")request_file= ("files")response_json= (url=request_url, data=request_data, files=request_file,headers=request_header).json()returnresponse_jsonif __name__ == "__main__":b= ExportLogic()requests库演⽰代码--删减版importunittestimportosimportHTMLTestReportCNimportdatetimefrom _logic _variable importGlobalVariablefrom _url importExportERLfrom _class t_dict importRequestDatafrom _print importLOGfrom se _dict importConfDatafrom _emailimportSendEmailimportrandom,requestsclassLZExportCase(se):"""新浪有借有还接⼝测试⽤例"""#全距PHONE =Public().create_phone()#提交五项资料的HeaderHEADER = GlobalVariable().get_header('lz_Header')#修改⼿机号的HeaderHEADER1 = GlobalVariable().get_header('lz_Header')BASEID= ''PERIODNUM= ''IMGID=''Repayid= ''Periodnum_= ''defsetUp(self):_logic=ExportLogic()_url=ExportERL()=Public()t_data=RequestData()_variable=GlobalVariable()=LOG()tDatabase=ConnectDatabase()_data=ConfData()rd= GlobalVariable().get_variable("password")=Public().create_phone()ne= GlobalVariable().get_variable("YJ_newphone")#_path =(e(e(__file__)), "report")_path = e(e(__file__))= ((e(e(e(__file__))),"static"),"")deftearDownC(self):pass#((_path, ""))#(((, "log"), "Export_log"))deftest_085_FileLoad(self):"""Case--上传⽂件(saveType正常)"""url= __export_url("LZ_test_url", "File")request_json= t__request_data('File')file_= open(, 'rb')file={"file": ('', file_, 'image/jpeg')}request_json["saveType"] = '1'request_json= _md5(keys="lz_test_keys", json=request_json)_info(self._(), __conf_data("log_name", "LZ"),"%s request data：%s" %(self._testMethodName, request_json))response_json= _t_post_file(url=url, json=request_json, files=file, header=)= response_json['data']['id']_info(self._(), __conf_data("log_name", "LZ"),"%s response data：%s" %(self._testMethodName, response_json))Equal(__conf_PromptMsg("prompt_msg", "LZ",'data_correct')['msg'], response_json['msg']) Equal(__conf_PromptMsg("prompt_msg", "LZ",'data_correct')['code'], response_json['code'])file_.close()deftest_086_HeadImg(self):"""Case--修改头像(headimgId正常)"""url= __export_url("LZ_test_url","HeadImg")request_json= t__request_data('HeadImg')request_json["headimgId"] =equest_json= _md5(keys="lz_test_keys", json=request_json)_info(self._(), __conf_data("log_name", "LZ"),"%s request data：%s" %(self._testMethodName, request_json))response_json= _t_post(url=url, json=request_json, header=)_info(self._(), __conf_data("log_name", "LZ"),"%s response data：%s" %(self._testMethodName, response_json))Equal(__conf_PromptMsg("prompt_msg", "LZ",'data_correct')['msg'], response_json['msg']) Equal(__conf_PromptMsg("prompt_msg", "LZ",'data_correct')['code'],response_json['code'])deftest_087_HeadImg(self):"""Case--修改头像(headimgId错误)"""url=__export_url("LZ_test_url", "HeadImg")request_json= t__request_data('HeadImg')request_json["headimgId"] = '3213adcx213vcv'request_json= _md5(keys="lz_test_keys", json=request_json)_info(self._(), __conf_data("log_name", "LZ"),"%s request data：%s" %(self._testMethodName, request_json))response_json= _t_post(url=url, json=request_json, header=)_info(self._(), __conf_data("log_name", "LZ"),"%s response data：%s" %(self._testMethodName, response_json))Equal(__conf_PromptMsg("prompt_msg", "LZ",'headimgId_ERROR')['msg'], response_json['msg'])and Equal(__conf_PromptMsg("prompt_msg", "LZ",'headimgId_ERROR')['code'],response_json['code'])deftest_088_HeadImg(self):"""Case--修改头像(headimgId为空)"""url=__export_url("LZ_test_url", "HeadImg")request_json= t__request_data('HeadImg')request_json["headimgId"] = ''request_json= _md5(keys="lz_test_keys", json=request_json)_info(self._(), __conf_data("log_name", "LZ"),"%s request data：%s" %(self._testMethodName, request_json))response_json= _t_post(url=url, json=request_json, header=)_info(self._(), __conf_data("log_name", "LZ"),"%s response data：%s" %(self._testMethodName, response_json))Equal(__conf_PromptMsg("prompt_msg", "LZ",'headimgId_NULL')['msg'], response_json['msg'])and Equal(__conf_PromptMsg("prompt_msg", "LZ",'headimgId_NULL')['code'],response_json['code'])if __name__ == '__main__':Send_mail=SendEmail()now_time= ().strftime("%Y%m%d%H%M")module_name= me(__file__).split(".")[0]module= __import__(module_name)path= (e(e(e(__file__))), "report")logo_path= ((path, "static"),"")fp= open((path, "report--%" % now_time), "wb")runner= stRunner(stream=fp)all_suite=stsFromModule(module)(all_suite)()_email(Send_mail)unittest库演⽰代码--删减版d、邮件发送邮件发送依赖两个python内置库smtplib、email：om email _dict importConfDatafrom _class importPublicclassSendEmail(object):def __init__(self):_data=ConfData()= __conf_data("email", "sender")er= __conf_data("email", "receiver")_server= __conf_data("email", "SMTP_server")me= __conf_data("email", "username")rd= __conf_data("email", "password")t= __conf_data("email", "content")_time= ().strftime("%Y%m%d%H%M")_path= (e(e(__file__)), "report")_path= (e(e(__file__)), "log")defcreate_msg(self,receiver):"""此函数主要构建收发邮件联系⼈、邮件正⽂、邮件附件、邮件标题:return:"""#构建邮件正⽂msg =ltipart()msg["from"] =msg["to"] =receivermsg['subject'] = "API⾃动化测试报告"txt=xt(t)(txt)#构建邮件附件之⼀：⾃动化测试报告report_name = _new_file(path=_path)report_path=(_path, report_name)report= xt(open(report_path, "rb").read(),'html', 'utf-8')report["Content-Type"] = 'application/octet-stream'_header('Content-Disposition', 'attachment', filename=('gbk', '',report_name))_base64(report)(report)#构建邮件附件之⼀：测试⽇志log_name = _new_file(path=_path)log_path=(_path, log_name)info_log= xt(open(log_path, 'rb').read(), 'base64', 'utf-8')info_log["Content-Type"] = 'application/octet-stream'info__header('Content-Disposition', 'attachment', filename=('gbk','', log_name))_base64(info_log)(info_log)returnmsgdef send_email(self,*args):"""创建实例，发送邮件:return:"""receiver= __conf_data("email", "%s_receiver" % args[0][0]) if args ermsg=_msg(receiver)smtp= _SSL(_server, 465) #在Linux端使⽤ssL⽅式连接邮箱服务器#t(_server, 465) # 在windows端使⽤connect⽅式连接邮箱服务器(me, rd)il(, (","), _string())()if __name__ == "__main__":s=SendEmail()_email()⽰例代码