PICT 组合 +PageObject+Python asyncio 接口测试

小爸爸 · 发表于 2017-6-28 11:27:28

说明
全对偶接口生成参数
功能
python3.4 asyncio aiohttp
unittest参数化
pageobject
数据维护用的YMAL
基于PICT全对偶生成接口参
代码分析
aiohttp asyncio 封装

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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()

复制代码

请求参数的处理
<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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

<pre style='border-width: 1px 0px; margin: 0px -15px 20px; padding: 5px 15px; border-radius: 0px; color: rgb(68, 68, 68); text-transform: none; line-height: 1.4285; text-indent: 0px; letter-spacing: 0.45px; overflow: auto; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; border-top-color: rgb(240, 240, 240); border-bottom-color: rgb(240, 240, 240); border-top-style: solid; border-bottom-style: solid; display: block; word-break: break-all; word-wrap: break-word; box-sizing: border-box; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px; border-image: initial;' class="highlight plaintext"><code style='margin: 2px; padding: 0px !important; border-radius: 3px; border: currentColor !important; color: rgb(68, 68, 68) !important; line-height: 18px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px !important; display: block; white-space: pre-wrap; word-break: break-all; box-sizing: border-box; background-color: rgb(249, 249, 249) !important;'>title: XXXX接口测试

host: baidu.com

port: 8443

protocol: https://

header: {account": "XX", "Content-Type": "application/json; charset=UTF-8","secrectKey": "XXX=","appID": "XX"}

</code></pre><p style='margin: 0px 0px 20px; color: rgb(34, 37, 39); text-transform: none; line-height: 26px; text-indent: 0px; letter-spacing: 0.45px; font-family: Helvetica, Arial, "PingFang SC", "Noto Sans", Roboto, "Microsoft Yahei", sans-serif; font-size: 15px; font-style: normal; font-weight: normal; word-spacing: 0px; white-space: normal; box-sizing: border-box; orphans: 2; widows: 2; background-color: rgb(255, 255, 255); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'> 
复制代码
配置用例yaml

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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



复制代码

小爸爸 · 发表于 2017-6-28 11:32:13

本帖最后由小爸爸于 2017-6-28 11:35 编辑

PageObject

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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()
)

复制代码

代码入口实例

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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()

复制代码

结果执行过程

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>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'}]]
......

复制代码

草帽路飞UU · 发表于 2017-6-28 17:32:27

楼主你这是针对参数少的，像我们公司接口都是20多个参数，难道也这样去排列组合？那接口多的话是不是太耗时间

乐哈哈yoyo · 发表于 2017-6-28 17:32:57

有个疑问, 一组要求顺序POST的请求, 可以用 asyncio 吗

		自动登录	找回密码
密码			(注-册)加入51Testing

[转贴] PICT 组合 +PageObject+Python asyncio 接口测试

站长推荐 /1