mirror of https://gitee.com/a529548204/apitest.git
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
# coding:utf-8
|
|
"""
|
|
@author: 井松
|
|
@contact: 529548204@qq.com
|
|
@file: relevance.py
|
|
@time: 2021/11/15 17:13
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
|
|
import jsonpath
|
|
from scripts.mkDir import mk_dir
|
|
from scripts.readYamlFile import ini_yaml
|
|
from config.confManage import dir_manage
|
|
|
|
"""
|
|
response: 执行接口,根据jsonpath 读取返回结果的指定值 存入缓存
|
|
body: 执行前 把body转化成json格式 根据jsonpath 读取指定参数值 存入缓存
|
|
|
|
取
|
|
body: get请求 根据name replace参数 post 根据path 更新json数据
|
|
"""
|
|
cachepath = dir_manage("${pro_dir}$") + dir_manage("${cache_dir}$")
|
|
|
|
|
|
def valueHandle(data: str):
|
|
param_dict = {}
|
|
param_list = data.split("&")
|
|
for param in param_list:
|
|
param_dict[param.split("=")[0]] = param.split("=")[1]
|
|
return param_dict
|
|
|
|
|
|
class Cache(object):
|
|
def __init__(self, path=cachepath):
|
|
self.path = path
|
|
mk_dir(path)
|
|
self.del_list = os.listdir(self.path)
|
|
|
|
def set(self, key, value):
|
|
with open(self.path + key + ".text", 'w', encoding="utf-8") as f:
|
|
f.write(str(value))
|
|
|
|
def get(self, key):
|
|
|
|
if key + ".text" in self.del_list:
|
|
with open(self.path + key + ".text", 'r', encoding="utf-8") as f:
|
|
value = f.read()
|
|
return value
|
|
else:
|
|
raise ValueError("{}不存在".format(key))
|
|
|
|
def set_many(self, data: dict):
|
|
for i in data:
|
|
with open(self.path + i + ".text", 'w', encoding="utf-8") as f:
|
|
f.write(str(data[i]))
|
|
|
|
def clear_all_cache(self):
|
|
|
|
del_list = os.listdir(self.path)
|
|
for f in del_list:
|
|
file_path = os.path.join(self.path, f)
|
|
if os.path.isfile(file_path):
|
|
os.remove(file_path)
|
|
elif os.path.isdir(file_path):
|
|
shutil.rmtree(file_path)
|
|
|
|
def clear_cache(self, key: str):
|
|
file_path = os.path.join(self.path, key + ".text")
|
|
if os.path.isfile(file_path):
|
|
os.remove(file_path)
|
|
elif os.path.isdir(file_path):
|
|
shutil.rmtree(file_path)
|
|
|
|
|
|
class Relevance(object):
|
|
|
|
def __init__(self):
|
|
self.caches = Cache()
|
|
|
|
def respons_cache(self, data, respons):
|
|
values = jsonpath.jsonpath(respons, data['path'])
|
|
if not values:
|
|
raise ValueError("path错误")
|
|
self.caches.set(key=data["name"], value=values[0])
|
|
|
|
def body_cache(self, data, param):
|
|
if isinstance(param, dict):
|
|
values = jsonpath.jsonpath(param, data['path'])
|
|
|
|
self.caches.set(key=data["name"], value=values[0])
|
|
else:
|
|
values = jsonpath.jsonpath(valueHandle(param), data['path'])
|
|
if not values:
|
|
raise ValueError("path错误")
|
|
self.caches.set(key=data["name"], value=values[0])
|
|
|
|
def relevance(self, data: dict, bodys=None, res=None):
|
|
if data is not None:
|
|
for i in data:
|
|
if i["cachefrom"] == 'body':
|
|
self.body_cache(i, bodys)
|
|
elif i["cachefrom"] == 'response':
|
|
self.respons_cache(i, res)
|
|
else:
|
|
raise TypeError("datasfrom错误")
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
s = {'regionList': {'name': '区域列表', 'host': 'host', 'address': '/v1/apps/region/', 'method': 'get', 'relevance': [
|
|
# {'relevancetype': 'set', 'datasfrom': 'reponse', 'path': '$.data', 'name': "ceshi"},
|
|
# {'relevancetype': 'get', 'datasfrom': 'reponse', 'path': '$.data', 'name': "ceshi"},
|
|
{'relevancetype': 'set', 'datasfrom': 'body', 'path': "$.code", 'name': "code"},
|
|
{'relevancetype': 'set', 'datasfrom': 'url', 'path': None, 'name': None}], 'relevanceCheck': None}}
|
|
ress = {
|
|
"data": {
|
|
"haha": 123
|
|
}
|
|
}
|
|
bo = "code=200&msg=success"
|
|
|
|
s1 = ini_yaml("urltest.yml", r"D:\apitest\test_suite\datas\saasApp")
|
|
rels = Relevance()
|
|
|
|
rels.relevance(s1, bodys=bo, res=ress)
|