之前写过组合测试生成参数在接口测试中的探索篇,现在正式运用的项目中
全对偶接口生成参数
aiohttp asyncio 封装
class request():
def __init__(self, **kwargs):
'''
http请求的封装，传入dict
:param req:
'''
self.req = kwargs
def get(self, url, param):
data = {}
_url = self.req["protocol"] + self.req["host"] + ":" + str(self.req["port"]) + url
print(_url +" get请求参数为:"+str(param))
try:
response = yield from aiohttp.request("GET", _url, headers=self.req["header"], params=param)
string = (yield from response.read()).decode('utf-8')
if response.status == 200:
data = json.loads(string)
else:
print("data fetch failed for")
print(response.content, response.status)
data["status_code"] = response.status
print(data)
except asyncio.TimeoutError:
print("访问失败")
except UnicodeDecodeError:
print("接口崩溃了")
return data
def post(self,url, param):
data = {}
_url = self.req["protocol"] + self.req["host"] + ':' + str(self.req["port"]) + url
print(_url + " post接口参数为:" + str(param))
try:
response = yield from aiohttp.request('POST', _url, data=json.dumps(param), headers=self.req["header"])
string = (yield from response.read()).decode('utf-8')
if response.status == 200:
data = json.loads(string)
else:
print("data fetch failed for")
print(response.content, response.status)
data["status_code"] = response.status
print(data)
except asyncio.TimeoutError:
print("访问失败")
return data
def asyn(fun):
loop = asyncio.get_event_loop()
tasks = asyncio.ensure_future(fun)
loop.run_until_complete(tasks)
# loop.close()
print('Task ret: {}'.format(tasks.result()))
return tasks.result()
url = '/XXX/login'
protocol = "https://"
host = "XXX"
port= 8443
header = {"account": "XXX", "Content-Type": "application/json; charset=UTF-8","secrectKey": "XXX=","appID": "app_2017030812000000011"}
f = request(header=header, host=host, protocol=protocol, port=port)
data = {'XXX': 'XXX', 'name': 'XXX'}
BaseAsy.asyn(f.post(url, param=data))
请求参数的处理
def paramsFilter(params):
'''
请求参数处理
:param params:
:return:
'''
result = {}
for wap_key in params:
for son_key in params[wap_key]:
if params[wap_key]["error"] == "0" or params[wap_key]["error"] == "1" or params[wap_key]["error"] == "2": # 过滤和处理相关请求参数
result[wap_key] = changeFormat(params[wap_key]["type"], params[wap_key]["input"])
break
return result
def changeFormat(key, param):
'''
请求参数类型处理
:param key:
:param param:
:return:
'''
param_type = {
"str": lambda: str(param),
"int": lambda: int(param),
"float": lambda: float(param),
"bool": lambda: bool(param)
}
return param_type[key]()
def readReq(param):
'''
读取请求的url
:param param:
:return:
'''
return param.split("|") # 1 用例id,2 用例介绍,3 url
def readParam(param):
'''
读取准备的pict参数
param1:...
param2:..
:return: list
'''
result = []
_param2 = ""
for item in param:
for key in item:
tempParam = item[key].split("&")
_param = ""
for tItem in tempParam:
tiParam = tItem.split("|")
if len(tiParam) == 5:
_param = _param + "," + key + ":error:" + tiParam[0] + ":input:" + tiParam[1] + ":type:" + tiParam[
2] + ":" + tiParam[3] + ":" + tiParam[4]
else:
_param = _param + "," + key + ":error:" + tiParam[0] + ":" + tiParam[1] + ":" + tiParam[2]
_param2 = _param2 + "," + _param
result.append(key + ":"+_param[1:])
break
return result
def readPictParam(paramRequestPath):
'''
读取本地e生成好了的接口请求参数
:param paramRequestPath: 已经处理好的pict参数路径
:return: list
'''
result = read(paramRequestPath)
l_result = []
if result:
for info in range(len(result)):
for item in range(len(result[info])):
t_result = result[info][item].split(",")
d_t = {}
for i in t_result:
temp = i.split(":")
t = {}
t[temp[1]] = temp[2]
if len(temp) > 5: #如果大于5，说明全部参数为8：如 ：' {'rep': 'dict', 'type': 'str', 'input': '""', 'error': '2'}
t[temp[3]] = temp[4]
t[temp[5]] = temp[6]
t[temp[7]] = temp[8]
else:
t[temp[3]] = temp[4] # 参数至少
d_t[temp[0]] = t
l_result.append(d_t)
return l_result
def pairPatchParam(**kwargs):
'''
pict生成请求参数
:param kwargs:
params: 请求的参数列表，类型为list
paramPath: 用例目录
paramRequestPath： 已生成用例目录
:return:
'''
for item in kwargs["params"]:
write(kwargs["paramPath"], item)
os.popen("pict " + kwargs["paramPath"] + ">" + kwargs["paramRequestPath"])
time.sleep(1)
配置 init.yaml
title: XXXX接口测试
host: baidu.com
port: 8443
protocol: https://
header: {account": "XX", "Content-Type": "application/json; charset=UTF-8","secrectKey": "XXX=","appID": "XX"}
配置用例 yaml
req: 1001|登录|/XX/login|POST
param:
- name: 0|swx458348|str|rep|dict&1|swx4583481|str|rep|dic&3|rep|dict
- XX: 0|fnNoaWt1bjE5ODk|str|rep|dict&1|fnNoaWt1bjE5ODk1|str|rep|dic&3|rep|dict
#error: 0正常，1错误的值，2类型错误,3不传字段，4后面再扩展如最大，最小
#rep：后面是检查点，支持
#{} ,对应key为Dict
#{[]},对应key为DictList
#{[{},{}]} 对应key为DictListDict
PageObject
class Login:
'''
kwargs:
path: 用例文件目录
initPath： 请求头部目录
'''
def __init__(self, **kwargs):
self.path = kwargs["path"] # 用例yaml目录
self.param = getYam(self.path)["param"] # 请求参数
self.req = getYam(self.path)["req"] # 请求url
self.readParam = readParam(self.param) # 读取并处理请求参数
pairPatchParam(params=self.readParam, paramPath=PATH("../Log/param.log"),
paramRequestPath=PATH("../Log/paramRequest.log")) # pict生成参数
self.getParam = readPictParam(paramRequestPath=PATH("../Log/paramRequest.log")) # 得到pict生成的参数
self.readReq = readReq(self.req) # 0 用例id,1 用例介绍,2 url,3 mehtod
print(self.readReq)
self.head = requestHead(kwargs["initPath"]) # initPath 请求头准备
print(self.head)
# self.head = requestHead(PATH("../yaml/init.yaml")) # protocol ,header,port,host,title
'''
发请求
'''
def operate(self):
for item in self.getParam:
param = paramsFilter(item) # 过滤接口,如果有其他加密，可以自行扩展
f = request(header=self.head["header"], host=self.head["host"], protocol=self.head["protocol"], port=self.head["port"])
if self.readReq[3] == "POST":
BaseAsy.asyn(f.post(self.readReq[2], param=param))
else:
BaseAsy.asyn(f.get(self.readReq[2], param=param))
test
from PageObject.PageLogin import Login
PATH = lambda p: os.path.abspath(
os.path.join(os.path.dirname(__file__), p)
)
class LoginTest(unittest.TestCase):
def testLogin(self):
login = Login(path=PATH("../yaml/login.yaml"), initPath=PATH("../yaml/init.yaml"))
login.operate()
)
from Base.BaseRunner import ParametrizedTestCase
from test.TestLogin import LoginTest
def runnerCase():
starttime = datetime.now()
suite = unittest.TestSuite()
suite.addTest(ParametrizedTestCase.parametrize(LoginTest))
unittest.TextTestRunner(verbosity=2).run(suite)
endtime = datetime.now()
if __name__ == '__main__':
runnerCase()
https://XXX post接口参数为:...
Task ret: {'status_code': 409}
https://XXX post接口参数为:{}
{'status_code': 400}
Task ret: {'status_code': 400}
https://XXX/login post接口参数为:....
{'status_code': 200, 'resultCode': 0, 'info': 'Success', '...
https:/...r/login post接口参数为:{'name': 'XXX', 'pwd': 'XXX'}
{'status_code': 409}
....
[[{'param': {'XX': 'AAAAAAAAAAAA'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {}, 'result': {'status_code': 400}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa', 'XX': 'AAAAAAAAAAAA'}, 'result': {'data': {'resultList': {'token': '5A2EFCE9F0C12EF99D8D18D930D7B71E:8CD9AFC8F3FB5B6B072FAE403A522D9040769B62FB557F90786DF4FFC1AF6356BA269B6E8905197B5B55359D0066146A', 'result': 'success'}}, 'resultCode': 0, 'info': 'Success', 'status_code': 200}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa', 'XX': 'AAAAAAAAAAAA1'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa1', 'XX': 'AAAAAAAAAAAA'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'XX': 'AAAAAAAAAAAA1'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa'}, 'result': {'status_code': 400}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa1'}, 'result': {'status_code': 400}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa1', 'XX': 'AAAAAAAAAAAA1'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}]]
......