由于请求是http类接口的核心,因此封装前考虑问题比较多:
<强> 1。对多种接口类型的支持,
<强> 2。连接异常时能够重连;
<强> 3。并发处理的选择,
<强> 4。使用方便,容易维护。
当前并未全部实现,后期会不断完善。重点提一下并发处理的选择:python的并发处理机制由于存在吉尔的原因,实现起来并不是很理想,综合考虑多进程,多线程,协程,在不考虑大并发性能测试的前提下使用了多线程——线程池的形式实现。使用的是
并发。期货模块。当前仅方便支持webservice接口。
# - * -编码:utf - 8 - *
进口的要求
从并发。期货进口ThreadPoolExecutor
从工具。配置进口配置#配置文件读取
从工具。日志导入日志#日志管理
从工具。工具导入decoLOG #日志装饰
“‘
功能:类的请求
使用方法:
郭可昌
作成时间:20180224
更新内容:
更新时间:
“‘
类请求(对象):
def __init__(自我):
自我。会话=requests.session ()
自我。头={}
# URL默认来源于配置文件,方便不同测试环境的切换,也可以动态设定
自我。URL=配置().getURL ()
#默认60年代,可以动态设定
自我。timeout=60
# http连接异常的场合,重新连接的次数,默认为3,可以动态设定
自我。iRetryNum=3
自我。errorMsg=" "
#内容={用例编号:响应数据}
自我。反应={}
#内容={用例编号:异常信息}
self.resErr={}
#原始后使用保留
# bodyData:请求的数据
@decoLOG
def post(自我,bodyData):
响应=没有
自我。errorMsg=" "
试一试:
响应=self.session.post(自我。URL, data=https://www.yisu.com/zixun/bodyData.encode (utf - 8),标题=自我。头,超时=self.timeout)
response.raise_for_status ()
除了例外e:
自我。errorMsg=str (e)
日志(). Log。HTTP错误(“请求异常,异常信息:% s“% self.errorMsg)
返回响应
#复数请求并发处理,采用线程池的形式,用例数祝辞线程池的容量:线程池的容量为并发数,否则,用例数为并发数
# dicDatas:{用例编号:用例数据}
@decoLOG
def req_all(自我、dicDatas iThreadNum=5):
如果len (dict (dicDatas)) & lt;1：
日志().logger.error(“没有测试对象,请确认后再尝试…”)
返回self.responses.clear ()
#请求用例集合转换(用例编的号,用例数据)
种子=[我因为我在dicDatas.items ())
self.responses.clear ()
#线程池并发执行,iThreadNum为并发数
遗嘱执行人ThreadPoolExecutor (iThreadNum):
executor.map (self.req_single、种子)
#返回所有请求的响应信息({用例编号:响应数据}),http连接异常:对应没有
返回self.responses
#用于单用例提交、http连接失败可以重新连接,最大重新连接数可以动态设定
def req_single(自我、listData reqType=皃ost”, iLoop=1):
响应=没有
#如果达到最大重连次数,连接后提交结束
如果iLoop==self.iRetryNum:
如果reqType==皃ost”:
试一试:
响应=requests.post(自我。URL, data=https://www.yisu.com/zixun/listData [1] .encode (utf - 8),标题=self.header,
超时=self.timeout)
response.raise_for_status ()
除了例外e:
#异常信息保存只在最大连接次数时进行,未达到最大连接次数,异常信息为空
自我。resErr [listData [0]]=str (e)
日志(). Log。HTTP错误(“请求异常,异常信息:% s【% d】“% (str (e), iLoop))
自我。反应(listData[0]]=响应
其他:
#未来:其他请求方法扩大
通过
#未达到最大连接数,如果出现异常,则重新连接尝试
其他:
如果reqType==皃ost”:
试一试:
响应=requests.post(自我。URL, data=https://www.yisu.com/zixun/listData [1] .encode (utf - 8),标题=self.header,
超时=self.timeout)
response.raise_for_status ()
除了例外e:
日志(). Log。HTTP错误(“请求异常,异常信息:% s【% d】“% (str (e), iLoop))
#重连次数递增
iLoop +=1
#进行重新连接
自我。req_single (listData reqType, iLoop)
#当前连接终止
回来没有
自我。反应(listData[0]]=响应
其他:
#未来:其他请求方法扩大
通过
#设定SoapAction,快捷完成webservice接口头设定
def setSoapAction(自我,soapAction):
自我。标题(“SOAPAction”)=SOAPAction
自我。标题(“内容类型”)=" text/xml; charset=utf - 8”
自我。标题(“连接”)=拔?
自我。标题(“用户代理”)=" InterfaceAutoTest-run "
