# -*-coding:utf-8 -*- # # --------------------------------------------------------------------------- # ProjectName: ${PROJECT_NAME} # FileName: ${FILE_NAME} # Author: shen # Datetime: ${DATE} ${TIME} # Description: # # --------------------------------------------------------------------------- import requests.sessions from APIAutoTest_shen_2 import log_decorator from APIAutoTest_shen_2.common.db import DB from APIAutoTest_shen_2.common.read_basic_ini import ReadBasicIni from APIAutoTest_shen_2.common.read_excel import ReadExcel class RequestsMethod: @log_decorator def __init__(self): """管理被测系统的登录状态""" #创建db类对象 self.db = DB() #创建session对象 self.bpm_session =requests.sessions.Session() #配置登录数据 login_url = ReadBasicIni().get_host("bpm") + "/auth" login_data = {"username":"admin","password":"WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="} #获取token token = self.bpm_session.request(method="post",url=login_url,json=login_data).json().get("token") #将获取到的token更新到session对象的headers中 self.bpm_session.headers["Authorization"] = "Bearer "+token @log_decorator def close(self): """关闭数据的连接""" self.db.close() # @log_decorator def request_all(self,req_url,req_mime,req_method,case_data,sql_data,update_key,sql_type): #封装一个公共的请求方法,先执行sql语句再发送请求 #执行删除sql语句 if sql_type =="delete": self.db.delete(sql_data) #再执行查询的sql语句 elif sql_type == "select": select_result = self.db.select(sql_data) #将查询结果更新到用例数据中 case_data[update_key] = select_result #判断sql语句是否为select|delete,或者是否为delete|select elif sql_type == "select|delete" or sql_type == "delete|select": #执行删除语句 self.db.delete(sql_data["delete"]) #执行查询语句 select_result = self.db.select(sql_data["select"]) #将查询到的结果更新到用例中 case_data[update_key] = select_result #发送请求 # requests传参关心的是传参的方式,不关心请求的方法 #判断媒体类型是否为query|json,json|query表示请求体使用json传参,同时也在地址栏传参 if req_mime == "query|json" or req_mime == "json|query": #requests支持地址栏和请求体同时传参,请求体中每次只能使用一个关键字传参 return self.bpm_session.request(method=req_method,url=req_url,json=case_data["json"],params=case_data["query"]) #判断媒体类型是否为application/json或json,使用关键字json传参 elif req_mime == "application/json" or req_mime == "json": return self.bpm_session.request(method=req_method,url=req_url,json=case_data) #判断媒体类型是否为application/x-www-form-urlencoded或form,使用关键字data传参 elif req_mime == "application/x-www-form-urlencoded" or req_mime == "form": return self.bpm_session.request(method=req_method,url=req_url,data=case_data) #判断媒体类型是否为multipart/form-data或form-data,使用关键字files传参 elif req_mime == "multipart/form-data" or req_mime == "form-data": return self.bpm_session.request(method=req_method,url=req_url,files=case_data) elif req_mime == "query": return self.bpm_session.request(method=req_method, url=req_url,params=case_data) elif req_mime is None: return self.bpm_session.request(method=req_method, url=req_url) else: return ValueError("媒体类型的值错误") if __name__ == '__main__': req = RequestsMethod() print(req.bpm_session.headers) # data =['http://36.139.193.99:8088/api/demension/v1/dem/addDem', 'json', 'post', {'code': 'test_dem', 'description': '测试添加的维度', 'isDefault': 1, 'name': '测试维度'}, {'message': '添加维度成功!'}, 'delete from uc_demension where CODE_="test_dem";', 'delete', None] # res = req.request_all(req_url=data[0],req_method=data[2],req_mime=data[1],case_data=data[3],sql_type=data[-2],sql_data=data[-3],update_key=data[-1]) # print(res.text) # res = ReadExcel("zs").get_data() # for data in res: # res1 = req.request_all(req_url=data[0],req_method=data[2],req_mime=data[1],case_data=data[3],sql_type=data[-2],sql_data=data[-3],update_key=data[-1]) # print(res1.json())