437 lines
22 KiB
Python
437 lines
22 KiB
Python
#!/usr/bin/python3
|
||
# -*- coding: UTF-8 -*-
|
||
# 设置utf-8 显示中文
|
||
"""
|
||
@Author: guozg
|
||
@File：handle_dependence_data.py
|
||
"""
|
||
|
||
import sys
|
||
from string import Template
|
||
import jsonpath
|
||
import pytest
|
||
import requests
|
||
|
||
sys.path.append("../")
|
||
from common.yml.get_yml_keys import GetYmlKeys
|
||
from config.get_conf_data import GetConfData
|
||
from common.util.handle_jsonfile import HandleJsonFile
|
||
from login.login import Login
|
||
from common.yml.handle_dependence.get_all_dependence_case import GetAllDepCase
|
||
from common.yml.handle_dependence.get_yml_func_data import GetFuncData
|
||
from common.yml.handle_checkdata import HandleCheck
|
||
from common.yml.handle_response import HandleResponse
|
||
|
||
|
||
class HandleDependenceData:
|
||
"""
|
||
处理case里的所有依赖数据，包括多层依赖以及嵌套依赖的数据.
|
||
其中多层依赖,仅限制于request与response.当为json为sql时,不支持.
|
||
"""
|
||
|
||
def __init__(self, yml_data: dict):
|
||
"""
|
||
处理数据依赖,将根据文件的类型,而进行不同的操作. \n
|
||
yml文件:对应的是依赖类型为 request 或 response
|
||
json文件: 对应的是依赖类型为: json
|
||
sql:
|
||
:param yml_data: 处理后的yml数据.(page层与case层已合并,且默认值已处理完成)
|
||
"""
|
||
|
||
self.__conf = GetConfData()
|
||
self.__ymlkeys = GetYmlKeys()
|
||
self.__login = Login()
|
||
self.__yml_data = yml_data
|
||
self.__depcase = GetAllDepCase()
|
||
self.__funcdata = GetFuncData()
|
||
self.__hjson = HandleJsonFile()
|
||
self.__hresponse = HandleResponse()
|
||
self.__hcheck = HandleCheck()
|
||
|
||
def run_all_dependence_case(self)->dict:
|
||
"""
|
||
运行所有的依赖接口(注意在运行之前,会先进行反转,然后再按顺序去执行) \n
|
||
当运行到最后一个接口后,会将当前的case的 依赖数据 等,进行相关的替换操作 \n
|
||
即:最后会将当前的case对应的ymldata 依赖的值替换后,再回传回去. \n
|
||
:return: dict
|
||
"""
|
||
# 获取所有的依赖数据,并进行反转.方便进行
|
||
rely_list = self.__depcase.get_all_case(self.__yml_data)[::-1]
|
||
print(f"依赖的接口数据，进行反转后如下：\n{rely_list}")
|
||
count = len(rely_list)
|
||
# 存储响应结果
|
||
res = None
|
||
# 存储上一次的requestdata(依赖数据里有request依赖)
|
||
last_request = None
|
||
# 保存token,无token 无法执行
|
||
token = None
|
||
# 登录类型 app 还是web
|
||
login_type = None
|
||
|
||
# **************** 需要用到的一些key ********************
|
||
api_data_key = self.__ymlkeys.get_api_data_key()
|
||
data_type_key = self.__ymlkeys.get_data_type_key()
|
||
req_data_key = self.__ymlkeys.get_request_data_key()
|
||
|
||
dep_login_key = self.__ymlkeys.get_dependence_login_key()
|
||
login_data_key = self.__ymlkeys.get_login_data_key()
|
||
login_type_key = self.__ymlkeys.get_login_type_key()
|
||
token_key = self.__ymlkeys.get_token_filepath_key()
|
||
|
||
app_str = self.__conf.get_app_str()
|
||
web_str = self.__conf.get_web_str()
|
||
|
||
request_str = self.__conf.get_request_str()
|
||
response_str = self.__conf.get_response_str()
|
||
|
||
# dependence_case key
|
||
dep_case_key = self.__ymlkeys.get_dependence_case_key()
|
||
# dependence_case_data key
|
||
dep_casedata_key = self.__ymlkeys.get_dependence_case_data_key()
|
||
# dependence_type key
|
||
dep_type_key = self.__ymlkeys.get_dependence_type_key()
|
||
# dependence_key key
|
||
dep_key = self.__ymlkeys.get_dependence_key()
|
||
# replace_key key
|
||
rep_key = self.__ymlkeys.get_replace_key()
|
||
# jsonfilepath key
|
||
jsonfilepath_key = self.__ymlkeys.get_jsonfilepath_key()
|
||
|
||
|
||
case_join_page_flag = self.__conf.get_case_join_page()
|
||
|
||
# 返回的结果中的key
|
||
caseid_key = self.__conf.get_caseid()
|
||
filetype_key = self.__conf.get_filetype()
|
||
filedata_key = self.__conf.get_filedata()
|
||
testinit_key = self.__conf.get_testinit()
|
||
json_str = self.__conf.get_json_str()
|
||
sql_str = self.__conf.get_sql_str()
|
||
|
||
if count > 0:
|
||
for item in rely_list:
|
||
# 要先获取caseid
|
||
caseid = item[caseid_key]
|
||
'''
|
||
此时要判断 item的 caseid 和类型的.当caseid = None时,此时要判断filetype的类型是json 还是 sql.
|
||
目前当为sql时,未做处理,后续的逻辑再补充.
|
||
当caseid 不为None时,表示依赖的类型为request 或 response
|
||
'''
|
||
if caseid == None:
|
||
# 获取filetype的值
|
||
filetype = item[filetype_key]
|
||
if filetype == json_str:
|
||
filedata = item[filedata_key]
|
||
# 对res 直接进行赋值
|
||
res = filedata
|
||
|
||
# 表示为sql,暂时不做处理.
|
||
else :
|
||
pytest.fail(f"当依赖类型为sql时,这块的逻辑还没有完善,暂时不支持.")
|
||
|
||
"""
|
||
1:由于依赖的数据里,可能存在着登录依赖,但是登录的yml数据与其他的数据有一定的差异,所以要优先判断是否有登录
|
||
2:要检测是否有登录,目前只需要检测caseid里是否有None字符串即可. demo/test_data.yml::test_init:::None
|
||
"""
|
||
|
||
#demo/test_data.yml::test_init:::None
|
||
elif "None" in caseid:
|
||
# 处理登录
|
||
casestr = caseid.split(case_join_page_flag)[0]
|
||
logindata = self.__funcdata.get_yml_func_data(casestr)
|
||
login_type = logindata[login_type_key]
|
||
# 获取登录请求参数
|
||
login_data = logindata[login_data_key]
|
||
if login_type == app_str:
|
||
res = self.__login.get_app_login_response(login_data)
|
||
# 表示是web api登录
|
||
else:
|
||
res = self.__login.get_web_login_response(login_data)
|
||
# 存储登录的请求参数
|
||
last_request = login_data
|
||
|
||
# 表示为正常的 request 或response 依赖.
|
||
else:
|
||
# 获取 依赖的接口的 ymldata数据
|
||
ymldata = item[filedata_key]
|
||
api_data = ymldata[api_data_key]
|
||
# 获取data 的类型,是data,还是json,或者params--暂时用不着
|
||
data_type = api_data[data_type_key]
|
||
# 获取请求参数 值
|
||
request_data = api_data[req_data_key]
|
||
# 是否要登录
|
||
whether_login = api_data[dep_login_key]
|
||
# 是否有数据依赖
|
||
whether_dependence = api_data[dep_case_key]
|
||
|
||
# 表示需要登录获取token
|
||
if whether_login == True:
|
||
logindata = api_data[login_data_key]
|
||
# 登录类型,app还是web
|
||
login_type = logindata[login_type_key]
|
||
# 登录的参数值
|
||
login_data_second = logindata[login_data_key]
|
||
# 获取token_path的值
|
||
token_path = logindata[token_key]
|
||
|
||
"""
|
||
不在判断是全局登录依赖,还是local登录依赖.直接判断 token_path的值,当有值时,直接取,否则就使用login_data
|
||
里的值进行登录
|
||
"""
|
||
from common.util.get_file_path import GetFilePath
|
||
fpath = GetFilePath()
|
||
flag = fpath.check_tokenfile_available(login_type,token_path)
|
||
# 表示json文件存在,且有效.
|
||
if flag == True:
|
||
token = self.__hjson.read_json_file_to_dict(token_path)
|
||
|
||
# 表示json文件无效.
|
||
else:
|
||
# 判断是哪种类型的登录
|
||
if login_type == app_str:
|
||
token = self.__login.get_app_token(login_data_second)
|
||
|
||
#表示为web登录
|
||
else:
|
||
token = self.__login.get_web_token(login_data_second)
|
||
|
||
if login_type == app_str:
|
||
token = {"headers": token}
|
||
|
||
# 表示为web登录
|
||
else:
|
||
token = {"cookies": token}
|
||
|
||
# # 表示不需要登录
|
||
# else:
|
||
"""
|
||
注意:当将依赖的case 反转后,只有执行第一个依赖的case时,不存在数据依赖,其余的case都存在数据依赖.
|
||
1:判断是否有数据依赖,如果有,则还要进行参数值的替换
|
||
2:当有数据依赖时,表示res(用于接收上个依赖接口请求响应的结果)有值
|
||
3:当有接口依赖时,还要区分是request请求参数的值的依赖 还是 response 值的依赖
|
||
4:为了以防万一,当为response依赖时,可以对res做判断,当为None时,直接将该case置为fail.
|
||
"""
|
||
if whether_dependence == True:
|
||
# 获取依赖数据
|
||
dep_data = api_data[dep_casedata_key]
|
||
# 获取依赖类型
|
||
dep_type = dep_data[dep_type_key]
|
||
|
||
# 要判断依赖类型是request还是response
|
||
if dep_type == request_str:
|
||
# 此时进行last_request是否为None的判断,当为None时,要将case置为fail
|
||
if last_request == None:
|
||
pytest.fail(f"在执行被依赖的case:{caseid}时,依赖的上一次的接口请求参数:None,故无法执行此被依赖的case"
|
||
f"所以要将当前的case(非被依赖的case)置为fail")
|
||
|
||
# 表示此时要从依赖的接口的请求参数里取值
|
||
else:
|
||
# 获取被依赖的key 和 要替换的key
|
||
dep_key_list = dep_data[dep_key]
|
||
replace_key_list = dep_data[rep_key]
|
||
# 进行到这一步,说明dependence_key 与 replace_key的个数是一致的.所以在此不需要再校验两者的个数是否相同
|
||
key_list_count = len(dep_key_list)
|
||
for i in range(key_list_count):
|
||
# 获取依赖的key
|
||
dependence_key = dep_key_list[i]
|
||
# 获取要替换的key
|
||
replace_key = replace_key_list[i]
|
||
# 从request请求参数里获取被依赖的结果
|
||
dependence_value = jsonpath.jsonpath(last_request, dependence_key)[0]
|
||
# # 使用模板
|
||
# request_data_tmp = Template(json.dumps(request_data))
|
||
# # 重新拼接request_data的值
|
||
# request_data = json.loads(
|
||
# request_data_tmp.substitute(**{replace_key: dependence_value}))
|
||
|
||
"""
|
||
不在使用模板替换技术,因为字段里存在.的情况,如 actionType.id 当出现这种情况时,使用模板替换,会报错
|
||
所以要注意上面几行的代码
|
||
"""
|
||
# 替换请求数据.
|
||
request_data[data_type][replace_key] = dependence_value
|
||
|
||
# 执行被依赖的接口.当需要登录时,表示要传入token
|
||
if whether_login == True:
|
||
res = requests.request(**request_data, **token)
|
||
# 表示不需要传入token
|
||
else:
|
||
res = requests.request(**request_data)
|
||
last_request = request_data
|
||
|
||
|
||
# 表示是 response 数据依赖
|
||
elif dep_type == response_str:
|
||
# 此时要进行res是否为None的判断,当为None时,要将case置为fail
|
||
if res == None:
|
||
pytest.fail(f"在执行被依赖的case:{caseid}时,上一次的接口响应结果为None,故无法执行此被依赖的case"
|
||
f"所以要将当前的case(非被依赖的case)置为fail")
|
||
# 此时要从依赖的结果里进行获取相应的字段值,来替换request请求的字段值
|
||
else:
|
||
# 获取被依赖的key 和 要替换的key
|
||
dep_key_list = dep_data[dep_key]
|
||
replace_key_list = dep_data[rep_key]
|
||
# 进行到这一步,说明dependence_key 与 replace_key的个数是一致的.所以在此不需要再校验两者的个数是否相同
|
||
key_list_count = len(dep_key_list)
|
||
|
||
for i in range(key_list_count):
|
||
# 获取依赖的key
|
||
dependence_key = dep_key_list[i]
|
||
# 获取要替换的key
|
||
replace_key = replace_key_list[i]
|
||
|
||
# 获取被依赖的结果
|
||
# 加一层判断,当直接从json文件里获取数据时,是没有json()方法的.
|
||
if isinstance(res,dict):
|
||
dependence_value = jsonpath.jsonpath(res, dependence_key)[0]
|
||
else:
|
||
dependence_value = jsonpath.jsonpath(res.json(), dependence_key)[0]
|
||
# # 使用模板
|
||
# request_data_tmp = Template(json.dumps(request_data))
|
||
# # 重新拼接request_data的值
|
||
# request_data = json.loads(
|
||
# request_data_tmp.substitute(**{replace_key: dependence_value}))
|
||
|
||
"""
|
||
不在使用模板替换技术,因为字段里存在.的情况,如 actionType.id 当出现这种情况时,使用模板替换,会报错
|
||
所以要注意上面几行的代码
|
||
"""
|
||
request_data[data_type][replace_key] = dependence_value
|
||
|
||
# 执行被依赖的接口.当需要登录时,表示要传入token
|
||
if whether_login == True:
|
||
res = requests.request(**request_data, **token)
|
||
# 表示不需要登录,即不需要传入token
|
||
else:
|
||
res = requests.request(**request_data)
|
||
last_request = request_data
|
||
|
||
|
||
# 表示为 sql 数据依赖,该逻辑暂时未实现
|
||
elif dep_type == sql_str:
|
||
pass
|
||
# 表示为json 数据依赖
|
||
else:
|
||
# 获取被依赖的key 和 要替换的key
|
||
dep_key_list = dep_data[dep_key]
|
||
replace_key_list = dep_data[rep_key]
|
||
# 进行到这一步,说明dependence_key 与 replace_key的个数是一致的.所以在此不需要再校验两者的个数是否相同
|
||
key_list_count = len(dep_key_list)
|
||
# 获取jsonfilepath的值
|
||
jsonfilepath = dep_data[jsonfilepath_key]
|
||
# 获取依赖的json数据
|
||
res = self.__hjson.read_json_file_to_dict(jsonfilepath)
|
||
|
||
for i in range(key_list_count):
|
||
# 获取依赖的key
|
||
dependence_key = dep_key_list[i]
|
||
# 获取要替换的key
|
||
replace_key = replace_key_list[i]
|
||
|
||
# 获取被依赖的结果
|
||
# 加一层判断,当直接从json文件里获取数据时,是没有json()方法的.
|
||
if isinstance(res, dict):
|
||
dependence_value = jsonpath.jsonpath(res, dependence_key)[0]
|
||
else:
|
||
dependence_value = jsonpath.jsonpath(res.json(), dependence_key)[0]
|
||
# # 使用模板
|
||
# request_data_tmp = Template(json.dumps(request_data))
|
||
# # 重新拼接request_data的值
|
||
# request_data = json.loads(
|
||
# request_data_tmp.substitute(**{replace_key: dependence_value}))
|
||
|
||
"""
|
||
不在使用模板替换技术,因为字段里存在.的情况,如 actionType.id 当出现这种情况时,使用模板替换,会报错
|
||
所以要注意上面几行的代码
|
||
"""
|
||
request_data[data_type][replace_key] = dependence_value
|
||
|
||
# 执行被依赖的接口.当需要登录时,表示要传入token
|
||
if whether_login == True:
|
||
res = requests.request(**request_data, **token)
|
||
# 表示不需要登录,即不需要传入token
|
||
else:
|
||
res = requests.request(**request_data)
|
||
last_request = request_data
|
||
|
||
|
||
|
||
# 表示没有数据依赖
|
||
else:
|
||
# 当需要登录时, 表示要传入token
|
||
if whether_login == True:
|
||
res = requests.request(**request_data, **token)
|
||
|
||
else:
|
||
# 存储响应结果
|
||
res = requests.request(**request_data)
|
||
# 在实际的执行接口的过程中，有时会出现无响应或反应的结果为None的情况
|
||
text = res.text
|
||
if text == None or text == "":
|
||
res = None
|
||
|
||
# 存储请求参数
|
||
last_request = request_data
|
||
|
||
"""检测是否保存响应结果"""
|
||
self.__hresponse.handle_response(res, ymldata)
|
||
|
||
"""数据检查,同时按配置保存校验结果"""
|
||
self.__hcheck.handle_checkdata(res, ymldata)
|
||
|
||
|
||
# # 在返回结果时,要先判断所依赖的数据类型是request,还是response
|
||
# if self.__yml_data[self.api_data_key][self.dependence_type_key] == self.request_str:
|
||
# return last_request
|
||
# else:
|
||
# return res.json()
|
||
|
||
# 在返回结果时,要先判断所依赖的数据类型是request,还是response,然后将传入的ymldata数据直接进行替换
|
||
dependence_data = self.__yml_data[api_data_key][dep_casedata_key]
|
||
dependence_type = dependence_data[dep_type_key]
|
||
|
||
dependence_key_list = dependence_data[dep_key]
|
||
replace_key_list = dependence_data[rep_key]
|
||
key_list_count = len(dependence_key_list)
|
||
# 获取传入的ymldata里的requestdata
|
||
requestdata = self.__yml_data[api_data_key][req_data_key]
|
||
data_type = self.__yml_data[api_data_key][data_type_key]
|
||
for i in range(key_list_count):
|
||
# 获取依赖的key
|
||
dependence_key = dependence_key_list[i]
|
||
# 获取要替换的key
|
||
replace_key = replace_key_list[i]
|
||
if dependence_type == request_str:
|
||
# 从request请求参数里获取被依赖的结果
|
||
dependence_value = jsonpath.jsonpath(last_request, dependence_key)[0]
|
||
else:
|
||
|
||
# 先判断res 是否为None
|
||
if res == None:
|
||
pytest.fail(f"在执行被依赖的接口或获取被依赖的数据时，响应结果为None，故将case置为fail")
|
||
else:
|
||
# 表示是直接从json文件里获取的响应结果
|
||
if isinstance(res,dict):
|
||
dependence_value = jsonpath.jsonpath(res, dependence_key)[0]
|
||
else:
|
||
# 获取被依赖的结果
|
||
dependence_value = jsonpath.jsonpath(res.json(), dependence_key)[0]
|
||
"""
|
||
由于依赖的字段里，也存在带.的情况，如 actionType.id。当出现这种情况时，就不能使用模板替换技术，
|
||
此时进行进行赋值即可。
|
||
下面的两行代码，需要注释掉，直接进行赋值
|
||
"""
|
||
# # 使用模板
|
||
# request_data_tmp = Template(json.dumps(requestdata))
|
||
# # 重新拼接request_data的值
|
||
# requestdata = json.loads(request_data_tmp.substitute(**{replace_key: dependence_value}))
|
||
|
||
requestdata[data_type][replace_key] = dependence_value
|
||
|
||
# 再重对传入的ymldata里的request_data重新赋值
|
||
self.__yml_data[api_data_key][self.__ymlkeys.get_request_data_key()] = requestdata
|
||
# 返回拼接请求参数值后的ymldata
|
||
return self.__yml_data
|
||
|
||
|