mirror of https://gitee.com/a529548204/apitest.git
116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
# coding:utf-8
|
|
import datetime
|
|
import json
|
|
import yaml
|
|
import threading
|
|
import os
|
|
import time
|
|
import logging
|
|
from config.confManage import dir_manage
|
|
from util.scripts import root_path
|
|
from util.tools.getFileNames import getfilepathlist
|
|
from util.tools.log import Log
|
|
from util.tools.redisData import RedisHandler
|
|
|
|
Log()
|
|
|
|
|
|
datapath = root_path + dir_manage("${test_suite}$") + dir_manage("${data_dir}$") + "/" + dir_manage("${test_name}$")
|
|
datalist = getfilepathlist(datapath, ".yml")
|
|
re = RedisHandler()
|
|
|
|
|
|
class ComplexEncoder(json.JSONEncoder):
|
|
# 重写json类 处理时间格式
|
|
def default(self, obj):
|
|
if isinstance(obj, datetime.datetime):
|
|
return obj.strftime('%Y-%m-%d %H:%M:%S')
|
|
elif isinstance(obj, datetime.date):
|
|
return obj.strftime('%Y-%m-%d')
|
|
else:
|
|
return json.JSONEncoder.default(self, obj)
|
|
|
|
|
|
def ini_allyaml(datalists):
|
|
"""
|
|
获取全部yaml数据
|
|
:return:
|
|
"""
|
|
|
|
for file in datalists:
|
|
alldata = {}
|
|
name = file.split("/")[-1].split(".")[0]
|
|
try:
|
|
with open(file, 'r', encoding="utf-8") as f:
|
|
file_data = yaml.safe_load(f.read())
|
|
for k, v in file_data.items():
|
|
file_data[k] = json.dumps(file_data[k], cls=ComplexEncoder)
|
|
re.set_many(file_data)
|
|
except UnicodeDecodeError:
|
|
with open(file, 'r') as f:
|
|
file_data = yaml.safe_load(f.read())
|
|
for k, v in file_data.items():
|
|
file_data[k] = json.dumps(file_data[k], cls=ComplexEncoder)
|
|
re.set_many(file_data)
|
|
|
|
|
|
def threading_read():
|
|
"""
|
|
双线程读取参数数据
|
|
:return:
|
|
"""
|
|
re.clear()
|
|
n1 = int(len(datalist) / 2)
|
|
# 获取文件列表数量
|
|
l1 = datalist[0:n1]
|
|
l2 = datalist[n1:n1 * 2]
|
|
l3 = datalist[n1 * 2:]
|
|
thread1 = threading.Thread(target=ini_allyaml, kwargs={"datalists": l1})
|
|
thread2 = threading.Thread(target=ini_allyaml, kwargs={"datalists": l2})
|
|
thread3 = threading.Thread(target=ini_allyaml, kwargs={"datalists": l3})
|
|
# 启动线程
|
|
thread1.start()
|
|
thread2.start()
|
|
thread3.start()
|
|
|
|
|
|
def ini_yaml(filename, path=datapath):
|
|
try:
|
|
with open(path + "/" + filename, 'r', encoding="utf-8") as f:
|
|
file_data = f.read()
|
|
data = yaml.safe_load(file_data)
|
|
|
|
return data
|
|
except UnicodeDecodeError:
|
|
with open(path + "/" + filename, 'r') as f:
|
|
file_data = f.read()
|
|
data = yaml.safe_load(file_data)
|
|
|
|
return data
|
|
|
|
|
|
def readredisdata(name):
|
|
rd = RedisHandler()
|
|
return json.loads(rd.get_key(name))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# print(datapath)
|
|
# get_yaml_data(r"F:\api2.0\config\runConfig.yml")
|
|
# t1 = time.time()
|
|
# ini_allyaml(datalist)
|
|
threading_read()
|
|
# s = ini_yaml("biaodan.yml")
|
|
# print(s)
|
|
# print(type(json.dumps(s,cls=ComplexEncoder)))
|
|
# t2 = time.time() - t1
|
|
# print(f"文件读取时间为{t2}")
|
|
|
|
# d = readRedisData("login")
|
|
# print(d)
|
|
# # case_level = runConfig_dict[0]["address"].format(**{"home_id": "123"})
|
|
# print(runConfig_dict)
|
|
|
|
# print(case_level)
|
|
# print(type(case_level))
|