# -*- coding: utf-8 -*-
|
"""
|
@File : parser.py
|
@Time : 2023/1/14 15:10
|
@Author : geekbing
|
@LastEditTime : -
|
@LastEditors : -
|
@Description : API用例解析
|
"""
|
import datetime
|
import json
|
import logging
|
|
from ast import literal_eval
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
from typing import Any, Dict, Generator, List, Tuple, Union
|
|
import json5
|
import requests
|
|
from lunarlink import models
|
from lunaruser import models as user_models
|
from lunarlink.utils.tree import (
|
get_all_ycatid,
|
get_tree_max_id,
|
get_tree_ycatid_mapping,
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
class Format:
|
"""
|
解析标准HttpRunner脚本 前端->后端
|
"""
|
|
def __init__(self, body: Dict, level: str = "test"):
|
"""初始化要解析的参数
|
|
body => {
|
header: header -> [{key:'', value:'', desc:''},],
|
request: request -> {
|
form: formData - > [{key: '', value: '', type: 1, desc: ''},],
|
json: jsonData -> {},-
|
params: paramsData -> [{key: '', value: '', type: 1, desc: ''},]
|
files: files -> {"fields","binary"}
|
},
|
extract: extract -> [{key:'', value:'', desc:''}],
|
validate: validate -> [{expect: '', actual: '', comparator: 'equals', type: 1},],
|
variables: variables -> [{key: '', value: '', type: 1, desc: ''},],
|
hooks: hooks -> [{setup: '', teardown: ''},],
|
url: url -> string
|
method: method -> string
|
name: name -> string
|
}
|
"""
|
try:
|
self.name = body.pop("name", None)
|
self.__headers = body.get("header", {}).get("header", {})
|
self.__variables = body.get("variables", {}).get("variables", {})
|
self.__setup_hooks = body.get("hooks", {}).get("setup_hooks", {})
|
self.__teardown_hooks = body.get("hooks", {}).get("teardown_hooks", {})
|
|
if level == "test":
|
# 配置移除request参数
|
self.__params = (
|
body.get("request", {}).get("params", {}).get("params", {})
|
)
|
self.__data = body.get("request", {}).get("form", {}).get("data", {})
|
self.__json = body.get("request", {}).get("json", {})
|
self.__files = body.get("request", {}).get("files", {}).get("files", {})
|
|
self.__desc = {
|
"header": body.get("header", {}).get("desc", {}),
|
"data": body.get("request", {}).get("form", {}).get("desc", {}),
|
"files": body.get("request", {}).get("files", {}).get("desc", {}),
|
"params": body.get("request", {}).get("params", {}).get("desc", {}),
|
"variables": body.get("variables", {}).get("desc", {}),
|
}
|
self.url = body.pop("url", None)
|
self.method = body.pop("method", None)
|
|
self.__times = body.pop("times", None)
|
self.__extract = body.get("extract", {}).get("extract", {})
|
self.__validate = body.pop("validate", {}).get("validate", {})
|
self.__desc["extract"] = body.get("extract", {}).get("desc", {})
|
else:
|
self.__params = {}
|
self.__data = {}
|
self.__json = {}
|
self.__files = {}
|
|
if level == "config":
|
self.__desc = {
|
"header": body.get("header", {}).get("desc", {}),
|
"variables": body.get("variables", {}).get("desc", {}),
|
}
|
|
self.base_url = body.pop("base_url")
|
self.is_default = body.pop("is_default")
|
self.__parameters = body["parameters"].pop("parameters")
|
self.__desc["parameters"] = body["parameters"].pop("desc")
|
|
self.__level = level
|
self.testcase = None
|
self.project = body.pop("project", None)
|
self.relation = body.pop("nodeId", None)
|
# lunarlink的API没有rig_id字段，需要兼容
|
self.rig_id = body.get("rig_id", 0)
|
self.rig_env = body.get("rig_env", 0)
|
except KeyError:
|
pass
|
|
def parse(self):
|
"""
|
返回标准化HttpRunner "desc" 字段, 执行时需去除
|
:return:
|
"""
|
if not hasattr(self, "rig_id"):
|
self.rig_id = None
|
|
if not hasattr(self, "rig_env"):
|
self.rig_env = 0
|
|
test = {}
|
if self.__level == "test":
|
test = {
|
"name": self.name,
|
"rig_id": self.rig_id,
|
"times": self.__times,
|
"request": {"url": self.url, "method": self.method, "verify": False},
|
"desc": self.__desc,
|
}
|
|
if self.__extract:
|
test["extract"] = self.__extract
|
if self.__validate:
|
test["validate"] = self.__validate
|
|
elif self.__level == "config":
|
test = {
|
"name": self.name,
|
"request": {
|
"base_url": self.base_url,
|
},
|
"desc": self.__desc,
|
}
|
|
if self.__parameters:
|
test["parameters"] = self.__parameters
|
|
if self.__headers:
|
test["request"]["headers"] = self.__headers
|
if self.__params:
|
test["request"]["params"] = self.__params
|
if self.__data:
|
test["request"]["data"] = self.__data
|
if self.__json:
|
test["request"]["json"] = self.__json
|
# 兼容一些接口需要传空json
|
if self.__json == {}:
|
test["request"]["json"] = {}
|
if self.__files:
|
test["request"]["files"] = self.__files
|
if self.__variables:
|
test["variables"] = self.__variables
|
if self.__setup_hooks:
|
test["setup_hooks"] = self.__setup_hooks
|
if self.__teardown_hooks:
|
test["teardown_hooks"] = self.__teardown_hooks
|
|
self.testcase = test
|
|
|
class Parse:
|
"""
|
标准HttpRunner脚本解析至前端 后端->前端
|
"""
|
|
def __init__(self, body: Dict, level: str = "test"):
|
"""
|
body: => {
|
"name": "get token with $user_agent, $os_platform, $app_version",
|
"request": {
|
"url": "/api/get-token",
|
"method": "POST",
|
"headers": {
|
"app_version": "$app_version",
|
"os_platform": "$os_platform",
|
"user_agent": "$user_agent"
|
},
|
"json": {
|
"sign": "${get_sign($user_agent, $device_sn, $os_platform, $app_version)}"
|
},
|
"extract": [
|
{"token": "content.token"}
|
],
|
"validate": [
|
{"eq": ["status_code", 200]},
|
{"eq": ["headers.Content-Type", "application/json"]},
|
{"eq": ["content.success", true]}
|
],
|
"setup_hooks": [],
|
"teardown_hooks": []
|
}
|
"""
|
self.name = body.get("name")
|
self.__request = body.get("request") #
|
self.__variables = body.get("variables")
|
self.__setup_hooks = body.get("setup_hooks", [])
|
self.__teardown_hooks = body.get("teardown_hooks", [])
|
self.__desc = body.get("desc")
|
|
if level == "test":
|
self.__times = body.get("times", 1) # 如果导入没有times 默认为1
|
self.__extract = body.get("extract")
|
self.__validate = body.get("validate")
|
|
if level == "config":
|
self.__parameters = body.get("parameters")
|
|
self.__level = level
|
self.testcase = None
|
|
@staticmethod
|
def __get_type(content: Any) -> Tuple:
|
"""返回data_type 默认string"""
|
var_type = {
|
"str": 1,
|
"int": 2,
|
"float": 3,
|
"bool": 4,
|
"list": 5,
|
"dict": 6,
|
"NoneType": 7,
|
}
|
|
key = str(type(content).__name__)
|
|
# 黑魔法，为了兼容值是int，但又是$引用变量的情况
|
if key == "str" and "$int" in content:
|
return var_type["int"], content
|
|
if key == "NoneType":
|
return var_type["NoneType"], content
|
|
if key in ["list", "dict"]:
|
content = json.dumps(content, ensure_ascii=False)
|
else:
|
content = str(content)
|
|
return var_type[key], content
|
|
def parse_http(self):
|
"""解析成标准前端脚本格式"""
|
|
init = [
|
{
|
"key": "",
|
"value": "",
|
"desc": "",
|
}
|
]
|
|
init_p = [
|
{
|
"key": "",
|
"value": "",
|
"desc": "",
|
"type": 1,
|
}
|
]
|
|
# 初始化test结构
|
test = {
|
"name": self.name,
|
"header": init,
|
"request": {
|
"data": init_p,
|
"params": init_p,
|
"json_data": "",
|
},
|
"variables": init_p,
|
"hooks": [
|
{
|
"setup": "",
|
"teardown": "",
|
}
|
],
|
}
|
|
if self.__level == "test":
|
test["times"] = self.__times
|
test["method"] = self.__request["method"]
|
test["url"] = self.__request["url"]
|
test["validate"] = [
|
{
|
"expect": "",
|
"actual": "",
|
"comparator": "equals",
|
"type": 1,
|
}
|
]
|
test["extract"] = init
|
|
if self.__extract:
|
test["extract"] = []
|
for content in self.__extract:
|
for key, value in content.items():
|
test["extract"].append(
|
{
|
"key": key,
|
"value": value,
|
"desc": self.__desc["extract"][key],
|
}
|
)
|
|
if self.__validate:
|
test["validate"] = []
|
for content in self.__validate:
|
for key, value in content.items():
|
obj = Parse.__get_type(value[1])
|
# 兼容旧的断言
|
desc = ""
|
if len(value) >= 3:
|
# value[2]为None时，设置为''
|
desc = value[2] or ""
|
|
test["validate"].append(
|
{
|
"expect": obj[1],
|
"actual": value[0],
|
"comparator": key,
|
"type": obj[0],
|
"desc": desc,
|
}
|
)
|
elif self.__level == "config":
|
test["base_url"] = self.__request["base_url"]
|
test["parameters"] = init
|
|
if self.__parameters:
|
test["parameters"] = []
|
for content in self.__parameters:
|
for key, value in content.items():
|
test["parameters"].append(
|
{
|
"key": key,
|
"value": Parse.__get_type(value)[1],
|
"desc": self.__desc["parameters"][key],
|
}
|
)
|
|
if self.__request.get("headers"):
|
test["header"] = []
|
for key, value in self.__request.pop("headers").items():
|
test["header"].append(
|
{"key": key, "value": value, "desc": self.__desc["header"][key]}
|
)
|
|
if self.__request.get("data"):
|
test["request"]["data"] = []
|
for key, value in self.__request.pop("data").items():
|
obj = Parse.__get_type(value)
|
test["request"]["data"].append(
|
{
|
"key": key,
|
"value": obj[1],
|
"type": obj[0],
|
"desc": self.__desc["data"][key],
|
}
|
)
|
|
if self.__request.get("params"):
|
test["request"]["params"] = []
|
for key, value in self.__request.pop("params").items():
|
test["request"]["params"].append(
|
{
|
"key": key,
|
"value": value,
|
"type": 1,
|
"desc": self.__desc["params"][key],
|
}
|
)
|
|
if self.__request.get("json"):
|
test["request"]["json_data"] = json.dumps(
|
self.__request.pop("json"),
|
indent=4,
|
separators=(",", ": "),
|
ensure_ascii=False,
|
)
|
|
if self.__variables:
|
test["variables"] = []
|
for content in self.__variables:
|
for key, value in content.items():
|
obj = Parse.__get_type(value)
|
test["variables"].append(
|
{
|
"key": key,
|
"value": obj[1],
|
"desc": self.__desc["variables"][key],
|
"type": obj[0],
|
}
|
)
|
|
if self.__setup_hooks or self.__teardown_hooks:
|
test["hooks"] = []
|
if len(self.__setup_hooks) > len(self.__teardown_hooks):
|
for index in range(0, len(self.__setup_hooks)):
|
teardown = ""
|
if index < len(self.__teardown_hooks):
|
teardown = self.__teardown_hooks[index]
|
test["hooks"].append(
|
{
|
"setup": self.__setup_hooks[index],
|
"teardown": teardown,
|
}
|
)
|
else:
|
for index in range(0, len(self.__teardown_hooks)):
|
setup = ""
|
if index < len(self.__setup_hooks):
|
setup = self.__setup_hooks[index]
|
test["hooks"].append(
|
{
|
"setup": setup,
|
"teardown": self.__teardown_hooks[index],
|
}
|
)
|
|
self.testcase = test
|
|
|
class Yapi:
|
def __init__(
|
self,
|
yapi_base_url: str,
|
token: str,
|
faster_project_id: int,
|
):
|
self.__yapi_base_url = yapi_base_url
|
self.__token = token
|
self.faster_project_id = faster_project_id
|
self.api_info: List = []
|
self.api_ids: List = []
|
# self.category_info: List = []
|
# api基础信息，不包含请求报文
|
self.api_list_url = self.__yapi_base_url + "/api/interface/list"
|
# api详情，包含详细的请求报文
|
self.api_details_url = self.__yapi_base_url + "/api/interface/get"
|
# api所有分组目录，也包含了api的基础信息
|
self.category_info_url = self.__yapi_base_url + "/api/interface/list_menu"
|
|
def get_category_info(self) -> Dict:
|
"""获取接口菜单列表
|
:return:
|
"""
|
try:
|
res = requests.get(
|
self.category_info_url, params={"token": self.__token}
|
).json()
|
except Exception as e:
|
logger.error(f"获取yapi的目录失败：{e}")
|
else:
|
if res["errcode"] == 0:
|
return res
|
else:
|
return {"errcode": 1, "errmsg": "获取yapi的目录失败！", "data": []}
|
|
def get_api_uptime_mapping(self):
|
"""yapi所有api的更新时间映射关系，{api_id: api_up_time}
|
:return:
|
"""
|
""
|
category_info_list = self.get_category_info()
|
mapping = {}
|
for category_info in category_info_list["data"]:
|
category_detail = category_info.get("list", [])
|
for category in category_detail:
|
api_id = category["_id"]
|
up_time = category["up_time"]
|
mapping[api_id] = up_time
|
return mapping
|
|
def get_category_id_name_mapping(self):
|
"""获取yapi的分组信息映射关系，{category_id: category_name}
|
:return:
|
"""
|
|
try:
|
res = self.get_category_info()
|
except Exception as e:
|
logger.error(f"获取yapi的目录失败：{e}")
|
else:
|
if res["errcode"] == 0:
|
# {'category_id': 'category_name'}
|
category_id_name_mapping = {}
|
for category_info in res["data"]:
|
# 排除为空的分组
|
if category_info.get("list"):
|
category_name = category_info.get("name")
|
category_id = category_info.get("_id")
|
category_id_name_mapping[category_id] = category_name
|
return category_id_name_mapping
|
|
def get_api_info_list(self):
|
"""获取接口列表数据
|
:return:
|
"""
|
try:
|
res = requests.get(
|
self.api_list_url,
|
params={
|
"token": self.__token,
|
"page": 1,
|
"limit": 100000,
|
},
|
).json()
|
except Exception as e:
|
logger.error(f"获取api list失败: {e}")
|
else:
|
if res["errcode"] == 0:
|
return res
|
|
def get_api_ids(self) -> List:
|
"""
|
获取yapi的api_ids
|
:return:
|
"""
|
api_list = self.get_api_info_list()
|
return [api["id"] for api in api_list["data"]["list"]]
|
|
def get_batch_api_detail(self, api_ids: List[int]) -> Generator[dict, None, None]:
|
"""
|
获取yapi的所有api的详细信息
|
:param api_ids:
|
:return:
|
"""
|
token = self.__token
|
session = requests.Session() # 创建一个 Session 对象
|
|
def fetch_api_detail(api_id):
|
try:
|
response = session.get(
|
f"{self.api_details_url}?token={token}&id={api_id}"
|
)
|
response.raise_for_status() # 如果状态码不是200，会引发HTTPError异常
|
res = response.json()
|
return res["data"]
|
except requests.HTTPError as http_err:
|
logger.error(f"HTTP error occurred: {http_err}")
|
except Exception as e:
|
logger.error(f"Error occurred: {e}")
|
|
with ThreadPoolExecutor(max_workers=10) as executor:
|
futures = {executor.submit(fetch_api_detail, api_id) for api_id in api_ids}
|
for future in as_completed(futures):
|
api_detail = future.result()
|
if api_detail is not None:
|
yield api_detail # 使用 yield 关键字，返回一个生成器
|
|
@staticmethod
|
def get_variable_default_value(
|
variable_type: str, variable_value: Union[Dict, Any]
|
):
|
"""
|
获取变量默认值
|
:param variable_type:
|
:param variable_value:
|
:return:
|
"""
|
|
if isinstance(variable_value, dict) is False:
|
return ""
|
variable_type = variable_type.lower()
|
if variable_type in ("integer", "number", "bigdecimal"):
|
return variable_value.get("default", 0)
|
elif variable_type == "date":
|
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
elif variable_type == "string":
|
return ""
|
|
return ""
|
|
def create_relation_id(self, project_id):
|
"""创建yapi所属目录
|
:param project_id:
|
:return:
|
"""
|
category_id_name_mapping: Dict = self.get_category_id_name_mapping()
|
obj = models.Relation.objects.get(project_id=project_id, type=1)
|
eval_tree: List = literal_eval(obj.tree)
|
yapi_catids: List = [yapi_catid for yapi_catid in get_all_ycatid(eval_tree, [])]
|
|
if category_id_name_mapping is None:
|
return
|
for cat_id, cat_name in category_id_name_mapping.items():
|
if cat_id not in yapi_catids:
|
tree_id = get_tree_max_id(tree=eval_tree)
|
base_tree_node = {
|
"id": tree_id + 1,
|
"yapi_catid": cat_id,
|
"label": cat_name,
|
"children": [],
|
}
|
eval_tree.append(base_tree_node)
|
|
obj.tree = json.dumps(eval_tree, ensure_ascii=False)
|
obj.save()
|
|
def yapi2faster(self, source_api_info):
|
"""yapi单个api转成faster格式
|
:param source_api_info:
|
:return:
|
"""
|
|
logger.info(f"正在处理yapi的接口id是{source_api_info.get('_id')}")
|
api_info_template = {
|
"header": {
|
"header": {},
|
"desc": {},
|
},
|
"request": {
|
"form": {
|
"data": {},
|
"desc": {},
|
},
|
"json": {},
|
"params": {
|
"params": {},
|
"desc": {},
|
},
|
"files": {
|
"files": {},
|
"desc": {},
|
},
|
},
|
"extract": {
|
"extract": [],
|
"desc": {},
|
},
|
"validate": {
|
"validate": [],
|
},
|
"variables": {
|
"variables": [],
|
"desc": {},
|
},
|
"hooks": {
|
"setup_hooks": [],
|
"teardown_hooks": [],
|
},
|
"url": "",
|
"method": "",
|
"name": "",
|
"times": 1,
|
"nodeId": 0,
|
"project": self.faster_project_id,
|
}
|
|
default_validator = {"equals": ["status_code", 200]}
|
api_info_template["validate"]["validate"].append(default_validator)
|
|
# 限制api的名称最大长度，避免溢出
|
api_info_template["name"] = source_api_info.get("title", "默认api名称")[:100]
|
|
# path中{var}替换成$var格式
|
api_info_template["url"] = (
|
source_api_info.get("path", "").replace("{", "$").replace("}", "")
|
)
|
api_info_template["method"] = source_api_info.get("method", "GET")
|
|
# yapi的分组id
|
api_info_template["yapi_catid"] = source_api_info["catid"]
|
api_info_template["yapi_id"] = source_api_info["_id"]
|
|
# 十位时间戳
|
api_info_template["yapi_add_time"] = source_api_info.get("add_time", "")
|
api_info_template["yapi_up_time"] = source_api_info.get("up_time", "")
|
|
# yapi原作者名
|
api_info_template["yapi_username"] = source_api_info.get("username", "")
|
|
req_body_type = source_api_info.get("req_body_type")
|
req_body_other = source_api_info.get("req_body_other", "")
|
if req_body_type == "json" and req_body_other != "":
|
try:
|
req_body = json.loads(req_body_other)
|
except json.decoder.JSONDecodeError:
|
# TODO: 解析带注释的json req_body没起作用
|
req_body = json5.loads(req_body_other, encoding="utf8")
|
except Exception as e:
|
logger.error(
|
f"yapi: {source_api_info['_id']}, req_body json loads failed: {source_api_info.get('req_body_other', e)}"
|
)
|
else:
|
# TODO: 递归遍历properties所有节点
|
if isinstance(req_body, dict):
|
req_body_properties = req_body.get("properties")
|
if isinstance(req_body_properties, dict):
|
for field_name, field_value in req_body_properties.items():
|
if isinstance(field_value, dict) is False:
|
continue
|
any_of = field_value.get("anyOf")
|
if isinstance(any_of, list):
|
if len(any_of) > 0:
|
field_value: dict = any_of[0]
|
|
field_type = field_value.get("type", "unKnow")
|
if field_type == "unKnow":
|
logger.error(
|
f'yapi: {source_api_info["_id"]}, req_body json type is unKnow'
|
)
|
|
if not (field_type == "array" or field_type == "object"):
|
self.set_ordinary_variable(
|
api_info_template=api_info_template,
|
field_name=field_name,
|
field_type=field_type,
|
field_value=field_value,
|
)
|
|
if field_type == "array":
|
items: dict = field_value["items"]
|
|
# 特殊字段处理，通用的查询条件
|
if field_name == "conditions":
|
set_customized_variable(api_info_template, items)
|
else:
|
items_type: str = items.get("type")
|
if items_type != "array" and items_type != "object":
|
self.set_ordinary_variable(
|
api_info_template=api_info_template,
|
field_name=field_name,
|
field_type=field_type,
|
field_value=field_value,
|
)
|
|
if field_type == "object":
|
properties: dict = field_value.get("properties")
|
if properties and isinstance(properties, dict):
|
for (
|
property_name,
|
property_value,
|
) in properties.items():
|
field_type = property_value["type"]
|
if not (
|
field_type == "array"
|
or field_type == "object"
|
):
|
self.set_ordinary_variable(
|
api_info_template=api_info_template,
|
field_name=property_name,
|
field_type=field_type,
|
field_value=property_value,
|
)
|
|
req_query: List = source_api_info.get("req_query", [])
|
if req_query:
|
for param in req_query:
|
param_name = param["name"]
|
param_desc = param.get("desc", "")
|
param_example = param.get("example", "")
|
api_info_template["request"]["params"]["params"][
|
param_name
|
] = f"${param_name}"
|
api_info_template["request"]["params"]["desc"][param_name] = param_desc
|
api_info_template["variables"]["variables"].append(
|
{param_name: param_example}
|
)
|
api_info_template["variables"]["desc"][param_name] = param_desc
|
|
req_body_form: List = source_api_info.get("req_body_form", [])
|
if req_body_form:
|
for data in req_body_form:
|
form_name = data.get("name")
|
form_desc = data.get("desc", "")
|
form_example = data.get("example", "")
|
api_info_template["request"]["form"]["data"][
|
form_name
|
] = f"${form_name}"
|
api_info_template["request"]["form"]["desc"][form_name] = form_desc
|
api_info_template["variables"]["variables"].append(
|
{form_name: form_example}
|
)
|
api_info_template["variables"]["desc"][form_name] = form_desc
|
|
req_params: List = source_api_info.get("req_params", [])
|
if req_params:
|
for param in req_params:
|
param_name = param.get("name")
|
param_desc = param.get("desc", "")
|
param_example = param.get("example", "")
|
api_info_template["variables"]["variables"].append(
|
{param_name: param_example}
|
)
|
api_info_template["variables"]["desc"][param_name] = param_desc
|
|
return api_info_template
|
|
def set_ordinary_variable(
|
self, api_info_template, field_name, field_type, field_value
|
):
|
api_info_template["request"]["json"][field_name] = f"${field_name}"
|
api_info_template["variables"]["variables"].append(
|
{field_name: self.get_variable_default_value(field_type, field_value)}
|
)
|
api_info_template["variables"]["desc"][field_name] = field_value.get(
|
"description", ""
|
)
|
|
def get_parsed_apis(self, api_info) -> List:
|
"""
|
批量创建fastapi格式的api
|
:param api_info:
|
:return: 返回多个 API 类的实例
|
"""
|
|
apis = [
|
self.yapi2faster(api) for api in api_info if isinstance(api, dict) is True
|
]
|
proj = models.Project.objects.get(id=self.faster_project_id)
|
obj = models.Relation.objects.get(project_id=self.faster_project_id, type=1)
|
yapi_user = user_models.MyUser.objects.filter(name="yapi").first()
|
yapi_user_id = yapi_user.id if yapi_user else None
|
eval_tree: List = literal_eval(obj.tree)
|
tree_ycatid_mapping = get_tree_ycatid_mapping(value=eval_tree)
|
api_instances = []
|
for api in apis:
|
format_api = Format(api)
|
format_api.parse()
|
yapi_catid: int = api["yapi_catid"]
|
api_body = {
|
"name": format_api.name,
|
"body": format_api.testcase,
|
"url": format_api.url,
|
"method": format_api.method,
|
"project": proj,
|
"relation": tree_ycatid_mapping.get(yapi_catid, 0),
|
# 直接从yapi原来的api中获取
|
"yapi_catid": yapi_catid,
|
"yapi_id": api["yapi_id"],
|
"yapi_add_time": api["yapi_add_time"],
|
"yapi_up_time": api["yapi_up_time"],
|
"yapi_username": api["yapi_username"],
|
# 默认为yapi用户
|
"creator_id": yapi_user_id,
|
}
|
api_instances.append(models.API(**api_body))
|
|
return api_instances
|
|
@staticmethod
|
def merge_api(
|
api_instances: List, apis_imported_from_yapi: List
|
) -> Tuple[List, List]:
|
"""
|
将 yapi 获取的 api 和已导入测试平台的 api 进行合并
|
两种情况：
|
1. parsed_api.yapi_id不存在测试平台
|
2. yapi的id已经存在测试平台，新获取的 parsed_api.yapi_up_time > imported_api.yapi_up_time
|
:param api_instances: 解析后的 API 实例
|
:param apis_imported_from_yapi: 原 api 信息
|
:return: 返回要更新的 API 实例和要新增的 API 实例
|
"""
|
imported_apis_mapping = {
|
api.yapi_id: api.yapi_up_time for api in apis_imported_from_yapi
|
}
|
imported_apis_index = {
|
api.yapi_id: index for index, api in enumerate(apis_imported_from_yapi)
|
}
|
|
new_api_instances = []
|
update_api_instances = []
|
imported_apis_ids = set(imported_apis_mapping.keys())
|
for api in api_instances:
|
yapi_id = api.yapi_id
|
# parsed_api.yapi_id不存在测试平台
|
if yapi_id not in imported_apis_ids:
|
new_api_instances.append(api)
|
else:
|
# yapi的id已经存在测试平台
|
imported_yapi_up_time = imported_apis_mapping[yapi_id]
|
if api.yapi_up_time > int(imported_yapi_up_time):
|
index = imported_apis_index[yapi_id]
|
imported_api = apis_imported_from_yapi[index]
|
imported_api.method = api.method
|
imported_api.name = api.name
|
imported_api.url = api.url
|
imported_api.body = api.body
|
imported_api.yapi_up_time = api.yapi_up_time
|
|
update_api_instances.append(imported_api)
|
|
return update_api_instances, new_api_instances
|
|
def get_create_or_update_apis(self, imported_apis_mapping):
|
"""
|
返回需要新增和更新的api_id
|
imported_apis_mapping: {yapi_id: yapi_up_time}
|
新增：
|
yapi_id不存在测试平台imported_apis_mapping中
|
更新：
|
yapi_id存在测试平台imported_apis_mapping, 且up_time大于测试平台的
|
:param imported_apis_mapping:
|
:return:
|
"""
|
api_uptime_mapping: Dict = self.get_api_uptime_mapping()
|
|
create_ids = []
|
update_ids = []
|
for yapi_id, yapi_up_time in api_uptime_mapping.items():
|
imported_yapi_up_time = imported_apis_mapping.get(yapi_id)
|
if not imported_yapi_up_time:
|
# 新增
|
create_ids.append(yapi_id)
|
elif yapi_up_time > int(imported_yapi_up_time):
|
# 更新
|
update_ids.append(yapi_id)
|
|
return create_ids, update_ids
|
|
|
# 特殊字段conditions
|
def set_customized_variable(api_info_template, items):
|
if items["type"] == "object":
|
properties: dict = items["properties"]
|
attr_name: dict = properties.get("attributeName", {})
|
attribute_name_enum: list = attr_name.get("enum", [""])
|
if len(attribute_name_enum) == 0:
|
attribute_name_enum = [""]
|
target_value: list = [f"${value}" for value in attribute_name_enum]
|
# 查询条件字段默认模板
|
api_info_template["request"]["json"]["conditions"] = {
|
"attributeName": f"${attribute_name_enum[0]}",
|
"rangeType": "$rangeType",
|
"targetValue": target_value,
|
}
|
for attr in attribute_name_enum:
|
api_info_template["variables"]["variables"].append({attr: ""})
|
api_info_template["variables"]["desc"][attr] = attr_name.get(
|
"description", ""
|
)
|
|
# 查询条件比较类型
|
range_type: dict = properties.get("rangeType", {})
|
range_type_enum: list = range_type.get("enum", [""])
|
api_info_template["variables"]["variables"].append(
|
{"rangeType": range_type_enum[0]}
|
)
|
api_info_template["variables"]["desc"][
|
"rangeType"
|
] = f"条件匹配方式：{','.join(range_type_enum)}"
|
|
# 默认排序
|
api_info_template["request"]["json"]["orderBy"] = [
|
{
|
"attributeName": f"${attribute_name_enum[0]}",
|
"rankType": "DESC",
|
}
|
]
|
|
|
def format_json(value):
|
"""
|
将一个JSON对象格式化为易读的字符串。
|
|
如果输入值因为任何原因无法被序列化为JSON，就返回原始输入值。
|
|
:param value: 需要格式化的JSON对象
|
:return: 如果成功，返回格式化后的字符串；否则，返回原始输入值。
|
"""
|
try:
|
return json.dumps(value, indent=4, separators=(",", ": "), ensure_ascii=False)
|
except (TypeError, OverflowError):
|
return value
|
