import base64
import copy
import hashlib
import math
import random
import shutil
import zipfile
from os.path import exists
from typing import Optional, Tuple, List
import fire
from crypto.Cipher import AES
from faker import Faker
from requests import request, Response
import json
from urllib.parse import urljoin
import os
import time
fake = Faker('zh_CN')
version_url = "version"
interval = 10
# header
header = {
"Content-Type": "application/json",
}
# Environment Path
class EnvPath:
token = "TOKEN"
class Prefix:
UserServer = "api-user"
CaseServer = "api-case"
AssetServer = "api-asset"
TaskServer = "api-task"
# token
_token = os.environ.get(EnvPath.token)
""" 初始化配置相关接口 """
accept_url = Prefix.UserServer + "/users/agreement" # 接受协议
init_config_url = Prefix.TaskServer + "/system/initConfig" # 初始化配置
init_serve_url = Prefix.TaskServer + "/system/initServices" # 初始化服务
conn_master_url = Prefix.TaskServer + "/system/connectMaster" # 连接master
start_serve_url = Prefix.TaskServer + "/system/start" # 启动服务
""" 多车增加节点 """
add_worker_url = Prefix.TaskServer + "/worker" # 增加节点
""" 登录相关接口 """
user_check_url = Prefix.UserServer + "/users/check"
health = Prefix.UserServer + "/health" # 检查服务是否正常
cloud_login_url = Prefix.UserServer + "/users/login" # 云端登录
standalone_login_url = Prefix.UserServer + "/users/standalone" # 单机版登录 -- 已测试
login_check = Prefix.UserServer + "/users/check" # user check
""" 案例库目录相关接口 """
categories_url = Prefix.CaseServer + "/categories" # 案例库目录 --已测试
get_categories_url = Prefix.CaseServer + "/categories?withCases=false" # 获取所有案例库 --已测试
# get_cases_url = "api-case/cases?page=1&filter={category}&pageSize=100"
get_cases_url = Prefix.CaseServer + "/cases?page={page}&filter={category}&pageSize=100" # 获取当前案例库的所有案例 --已测试
delete_categories_url = Prefix.CaseServer + "/categories/{}" # 删除案例库目录
""" 上传文件相关接口 """
file_md5_url = Prefix.AssetServer + "/files/{}?raw={}" # 验证文件MD5 -- 已完成
upload_file_url = Prefix.AssetServer + "/files/{}/upload?raw={}" # 上传文件 -- 已完成
""" 测试案例相关接口 """
case_import_url = Prefix.AssetServer + "/cases/import" # 导入案例
case_def_url = Prefix.CaseServer + "/cases/{case_id}" # 案例概要 -- 已完成
case_details_url = Prefix.CaseServer + "/cases/{case_id}/data" # 案例详情 -- 已完成
delete_case_url = Prefix.CaseServer + "/trash" # 删除案例 -- 已完成
convert_replay_case_url = Prefix.TaskServer + "/tasks/{id}/convert2case" # 转换回放案例
update_case_url = "api-asset/cases/update" # 编辑案例
""" 测试任务相关接口 """
create_task_url = Prefix.TaskServer + "/tasks" # 创建task --已测试
stop_task_url = Prefix.TaskServer + "/tasks/stop" # 停止task --已完成
stop_sessions_url = Prefix.TaskServer + "/sessions/stop" # 会话停止 --已完成
delete_task_url = Prefix.TaskServer + "/tasks/delete" # 删除task --已完成
delete_sessions_url = Prefix.TaskServer + "/sessions/delete" # 删除停止 --已完成
result_url = Prefix.TaskServer + "/tasks/task?taskIds={task_id}" # 删除停止 案例运行结果详情 -- 已测试
pause_task_url = Prefix.TaskServer + "/sessions/{task_id}/pause" # 任务暂停
resume_task_url = Prefix.TaskServer + "/sessions/{task_id}/resume" # 任务暂停后开始
task_result_url = Prefix.TaskServer + "/tasks/task?taskIds={id}" # 删除停止 案例运行结果详情 -- 已测试
""" 主车相关接口 """
import_vehicle_url = Prefix.AssetServer + "/vehicles/import" # 导入主车 --已完成
delete_vehicle_url = Prefix.AssetServer + "/vehicles/{vehicles_id}" # 删除主车 --已完成
get_vehicle_url = Prefix.AssetServer + "/vehicles" # 获取主车信息 --已完成
""" 地图相关接口 """
import_map_url = Prefix.AssetServer + "/maps" # 导入地图 --已完成
delete_map_url = Prefix.CaseServer + "/maps/trash" # 删除地图 --已完成
get_map_url = Prefix.CaseServer + "/maps?pageSize=100" # 获取地图相关信息
road_service_url = Prefix.TaskServer + "/sce" # 启动路网服务
laneTypeIfInside_url = Prefix.TaskServer + "/sce/{map_id}/location/laneTypeIfInside" # 监测点是否在车道内
""" 数据驱动相关接口 """
import_data_driven_url = Prefix.AssetServer + "/dd" # 导入数据驱动源
delete_data_dirven_url = Prefix.AssetServer + "/dd/{id}" # 删除数据驱动源
""" 资源相关接口 """
import_asset_url = Prefix.AssetServer + "/assets" # 资源导入
asset_repeat_url = Prefix.AssetServer + "/assets/isrepeat" # 资源查重
get_asset_url = Prefix.AssetServer + "/assets?schema={}" # 获取资源信息
delete_asset_url = Prefix.AssetServer + "/assets/{}" # 删除对应资源
""" 案例导出相关 """
export_case_url = Prefix.AssetServer + "/cases/export" # 案例导出
download_case_url = Prefix.AssetServer + "/download/{pake_id}" # 案例下载
""" 测试案例集相关 """
get_suites_url = Prefix.AssetServer + "/suites?isProjectItem={}&opponent=openscenario" # 获取测试集
run_suite_url = Prefix.TaskServer + "/tasks" # 运行测试集
get_taskassemble_url = Prefix.TaskServer + "/tasks?finished=true&page=0&pageSize=12&own=true&expandedIds={task_set_id}" # 获取测试案例集合
queue_status_url = Prefix.TaskServer + "/tasks/queue?own=true" # 套件运行状态检查 -- 已测试
""" 案例判定相关 """
get_judgements_url = Prefix.AssetServer + "/cases/{caseId}/judgements"
# get_judgement_url = get_judgements_url + "/{judgementId}"
get_judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
""" 测试计划相关接口 """
testPlan_list_url = Prefix.TaskServer + "/testPlan?page=1&pageSize=20&options={options}" # 获取测试计划列表
testPlan_id_url = Prefix.TaskServer + "/testPlan/{id}" # 获取测试计划 / 更新测试计划（包含重命名）
create_testPlan_url = Prefix.TaskServer + "/testPlan" # 创建测试计划
del_testPlan_url = Prefix.TaskServer + "/testPlan/delete" # 删除多个测试计划
del_testPlan_single_url = Prefix.TaskServer + "/testPlan/{id}" # 删除单个测试计划
execute_testPlan_url = Prefix.TaskServer + "/testPlan/execute" # 执行测试计划
clone_testPlan_url = Prefix.TaskServer +"/testPlan/{id}/clone" # 复制测试计划
isrepeat_testPlan_url = Prefix.TaskServer +"/testPlan/isrepeat?name={name}&type={type}" # 测试计划名称是否已存在
# f"api-asset/cases/{caseId}/judgements/{judgementId}"
# url = f"api-asset/cases/{caseId}/judgements"
class UrlCodeType:
unquote = "unquote"
quote = "quote"
class File:
"""
文件父类
"""
def __init__(self, file_path: str):
if not exists(file_path):
raise FileNotFoundError
self._file_path = file_path
self._data = None
class FileReader(File):
"""
读取文件
"""
def __init__(self, file_path: str):
super(FileReader, self).__init__(file_path)
def read_byte(self) -> bytes:
with open(self._file_path, 'rb') as f:
return f.read()
def read_str(self) -> str:
with open(self._file_path, 'r', encoding='utf-8') as f:
return f.read()
def read_json(self) -> json:
with open(self._file_path, 'r', encoding='utf-8') as f:
return json.load(f)
class RquestApi():
def __init__(self):
super(RquestApi, self).__init__()
self._headers = {"Content-Type": "application/json", "ProjectId": 'default'}
self._token = os.environ.get(EnvPath.token)
self._host_ip = SimOneUrl
self.get = "GET"
self.post = "POST"
self.delete = "DELETE"
self.put = "PUT"
def send(self, method: str, url: str, **kwargs) -> Response:
"""
发送请求
method: 请求方法
url: 请求地址
kwargs: 请求参数
"""
self.headers.update({"Authorization": self.token})
complete_url = urljoin(self._host_ip, url)
data = dict()
data['method'] = method
data['url'] = complete_url
data['headers'] = self.headers
data.update(kwargs)
print(data)
response = request(verify=False, **data)
try:
print("response:%s" % str(response.content.decode("utf-8")))
self._status_code = response.status_code
self._response_code = response.json().get("code")
except:
pass
return response
@property
def token(self):
return self._token
@property
def headers(self):
return self._headers
@headers.setter
def headers(self, value):
self._headers = value
class LoginBusiness(RquestApi):
def cloud_login_api(self, userinfo: dict):
"""
云端登录API
@param userinfo:
@return:
"""
return self.send(self.post, cloud_login_url, json=userinfo).json()
def check_username_api(self, username: str):
"""
校验用户名称是否合格
@param username:
@return:
"""
payload = {'username': username}
return self.send(self.post, login_check, json=payload).json()
def getCheckToken(self) -> str:
"""
校验token是否合格
@return:
"""
getCheckUrl = os.path.join(SimOneUrl, user_check_url)
payload = {'username': loginData['username']}
response = self.send(self.post, getCheckUrl, json=payload)
checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
return checkToken
def pad(self, text):
"""
填充函数，使被加密数据的字节码长度是block_size的整数倍
@param text:
@return:
"""
length = AES.block_size
count = len(text.encode('utf-8'))
add = length - (count % length)
entext = text + (chr(add) * add)
return entext
def str_aes(self, string: str, key: str):
"""
aes 加密
@param string:
@param key:
@return:
"""
aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
res = aes.encrypt(self.pad(string).encode("utf8"))
msg = str(base64.b64encode(res), encoding="utf8")
return msg
def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
'''
@param string:
@param username:
@param key: AES加密的key
@return:
'''
check_token = self.check_username_api(username)
target_string = check_token["data"]["checkToken"] + string
msg = self.str_aes(target_string, key)
return msg
@property
def _cloud_std_login(self):
"""
云端版本登陆方式
@return:
"""
try:
new_login_data = copy.deepcopy(loginData)
new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
username=new_login_data['username'])
print('cloud login data:' + str(new_login_data))
response = self.cloud_login_api(userinfo=new_login_data)
print("-------", response)
print('cloud login request result:' + str(response))
token = response['data']['token']
os.environ[EnvPath.token] = token
print(token)
return token
except Exception as e:
print(str(e))
raise print('get token error ')
@property
def _enterprise_std_login(self):
"""
单机版本登录方式
@return:
"""
try:
getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
response = self.send(self.get, url=getLoginUrl)
token = json.loads(response.content.decode("utf-8"))['data']['token']
os.environ[EnvPath.token] = token
except Exception as e:
print(str(e))
token = ""
# raise AssertionError('get token error ')
return token
# 登录入口
class LoginPolicy():
standalone = "standalone"
cloud = "cloud"
def __new__(cls, env_name):
if env_name == cls.standalone:
return LoginBusiness()._enterprise_std_login
elif env_name == cls.cloud:
return LoginBusiness()._cloud_std_login
else:
return print("----------input error---------------")
class Suite(RquestApi):
def __init__(self):
super(Suite, self).__init__()
self._cyclical = 0
self.category_name = "Automation_" + str(random.randint(0, 9999))
def create_category_api(self, cate_name: str):
'''
创建案例库
@param cate_name: 案例库名
@return:
'''
payload = {"parentId": "", "name": cate_name}
return self.send(self.post, categories_url, json=payload).json()
def create_category(self, cate_name: str = None) -> str:
'''
创建案例库
@param cate_name:
@return:
'''
if not cate_name:
cate_name = self.category_name
print(f"案例库目录名称：{cate_name}")
response = self.create_category_api(cate_name)
cate_id = response["data"]["categoryId"]
print(f"案例库目录 id ：{cate_id}")
return cate_id
def get_category_api(self):
"""
获取案例库列表
@return: response
"""
return self.send(self.get, categories_url).json()
def get_category(self, userid: str = None) -> dict:
"""
获取案例库
@param userid: 用户id desc：admin->builtIn
@return:
"""
# 初始化一个包含data和parentId的字典，用于存储案例库信息
cate_dict = {"data": {}, "parentId": {}}
# 调用API获取案例库数据
response = self.get_category_api()
# 获取案例库总数
cate_dict["total"] = response["data"]["total"]
# 如果没有提供userid，获取所有案例库目录信息
if not userid:
print("获取当前所有的案例库目录信息")
# 遍历所有案例库，将名称和ID、名称和父ID存入字典
for one in response["data"]["categories"]:
cate_dict["data"].setdefault(one["name"], one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
else:
# 如果提供了userid，只获取该用户的案例库目录信息
print("获取用户{}的案例库目录信息".format(userid))
# 遍历所有案例库，筛选出属于指定用户的案例库
for one in response["data"]["categories"]:
if one["userId"] == userid:
# 将用户案例库的名称和ID、名称和父ID存入字典
cate_dict["data"].setdefault(one["name"], one["id"])
cate_dict["parentId"].setdefault(one["name"], one["parentId"])
# 打印获取到的案例库信息
print("获取到的案例库->%s" % str(cate_dict))
# 返回包含案例库信息的字典
return cate_dict
def get_category_case_api(self, cate_id: str):
"""
获取当前案例库所有案例
@param cate_name: 案例库名
@return:
"""
return self.send(self.get, get_cases_url.format(category=cate_id)).json()
def get_cases_category(self, cate_id: list):
"""
获取当前案例库的所有案例
@param cate_id: 案例库目录id
@return:
"""
def url_code(url: str, way: str):
from urllib import parse
if way == "unquote":
return parse.unquote(url)
elif way == "quote":
return parse.quote(url)
page_size = 100
# print("get all case id from api by category_id")
_filter = {"keyword": "", "categories": cate_id}
string = json.dumps(_filter).replace(" ", "")
filter = url_code(string, UrlCodeType.quote).replace("%", "%25")
response = self.send(self.get, get_cases_url.format(page="1", category=filter)).json()
# get case id
# print("get_cases_category 入参cate_id:%s" % cate_id)
cases_num = response['data']['total']
if cases_num >= page_size:
if cases_num % page_size == 0:
page = cases_num // page_size
else:
page = cases_num // page_size + 1
else:
page = 1
def update_case_id(response, page_size):
case_name = response['data']['caseDefs'][page_size]['name']
case_id = response['data']['caseDefs'][page_size]['id']
case_id_list.append(case_id)
cases_dict.setdefault(case_name, [case_id])
cases_dict = {}
case_id_list = []
# 如果案例数量是100以内
if cases_num <= page_size:
for i in range(cases_num):
update_case_id(response=response, page_size=i)
else:
# 如果案例数量超过100了，固定获取案例数量为100个就翻页
for i in range(page):
response = self.send(self.get, self.get_cases_url.format(page=str(i + 1), category=filter)).json()
# cases_num % page_size 是取整表示没有到达最后一页
# if i == 0 or i > 0 and cases_num % page_size == 0:
if i + 1 != page:
# print("xxx page_size:",page_size)
for j in range(page_size):
# print("xxx j:",j)
update_case_id(response=response, page_size=j)
# 到最后一页了，因为案例数量不固定所以需要单独计算获取的个数
else:
for j in range(cases_num % page_size):
update_case_id(response=response, page_size=j)
return cases_dict, case_id_list
def get_case_detail_api(self, case_id: str):
"""
获取案例信息
@param : /{case_id}/data
@return: response
"""
url = case_details_url.format(case_id=case_id)
return self.send(self.get, url).json()
def get_case_detail(self, case_id: str):
"""
@param case_id: 测试案例id
@return:
"""
response = self.get_case_detail_api(case_id)
print(f"调用 get_case_detail : {response}")
return response
def get_version(self):
res = self.send(self.get, url= version_url).json()
version = res["data"]["profile"]["version"]
return version
# def run_task(self, caseid: list, vehicle_id: str = None):
# """
# :param caseid:
# :return:
# """
#
# def creat_task_name():
# return "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
#
# taskname = creat_task_name()
# payload = {"caseIds": caseid, "taskName": taskname, "speed": 1}
# if vehicle_id:
# payload.update({"overrideVehicleId": vehicle_id})
# response = self.send(self.post, url=create_task_url, json=payload).json()
# print(f"run_task response：{response}") # taskIds为空时代表运行案例异常
# data = response['data']
# return data['sessionId'], data['taskIds']
def run_task(self, case_ids: List[str], vehicle_id: Optional[str] = None, withEvaluation: bool = False, version: str = "3.5.0", type: str = "worldsim")-> Tuple[str, List[str]]:
"""
运行测试任务集合
Run task set
@param caseid: caseid
@param vehicle_id: vehicle_id
@param withEvaluation: withEvaluation
@param version: version
@return:
"""
task_name = "AutoTest_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
payload = {}
if version >= "3.6.3":
payload = {"caseIds": case_ids, "taskName": task_name,
"type": type, "withEvaluation": False, "vehicleConfig": {
"Ego": {"instanceId": "Ego", "id": "default", "name": "主车", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "默认控制器", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器", "algorithmId": "SimOneDriver"}]}},
"notification": {"type": "none", "condition": "any"}, "enableStateMachine": False, "priority": 0}
elif version >= "3.5.0":
payload = {"caseIds": case_ids,
"taskName": task_name,
"withEvaluation": withEvaluation,
"controllers": [{"id": "Default", "name": "默认控制器", "algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器", "algorithmId": "SimOneDriver"}]}
elif version < "3.5.0":
payload = {"caseIds": case_ids, "taskName": task_name, "speed": 1}
# elif version > "3.1.1":
# payload = {"caseIds": case_ids, "taskName": task_name, "speed": 1}
# else:
# payload = {"caseIds":case_ids,"taskName":task_name,"withEvaluation":True,"type":"worldsim"} # 现代云端 内置案例有坑
if vehicle_id:
payload.update({"overrideVehicleId": vehicle_id})
response = self.send(self.post, url=create_task_url, json=payload).json()
data = response['data']
return data['sessionId'], data['taskIds']
def stop_task_api(self, payload: dict):
"""
task停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_task_url, json=payload).json()
def stop_sessions_api(self, payload: dict):
"""
sessions停止API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, stop_sessions_url, json=payload).json()
def stop_task(self, task_id: list, sessions_id: list):
"""
停止任务和会话
@param task_id:
@param sessions_id:
@return:
"""
task_payload = {"ids": task_id}
self.stop_task_api(task_payload)
sessions_id = {"ids": sessions_id}
self.stop_sessions_api(sessions_id)
def pause_task_api(self, task_id: str):
"""
暂停任务
@param task_id:
@return:
"""
return self.send(self.post, pause_task_url.format(task_id=task_id), json={}).json()
def resume_task_api(self, task_id: str):
"""
暂停任务后开始
@param task_id:
@return:
"""
return self.send(self.post, resume_task_url.format(task_id=task_id), json={}).json()
def delete_sessions_api(self, payload: dict):
"""
sessions删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_sessions_url, json=payload).json()
def delete_task_api(self, payload: dict):
"""
task删除API
@param payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
@return:
"""
return self.send(self.post, delete_task_url, json=payload).json()
def delete_task(self, sessions_id: list, task_id: list):
"""
删除任务和会话
@param sessions_id:
@param task_id:
@return:
"""
sessions_id = {"ids": sessions_id}
self.delete_sessions_api(sessions_id)
task_payload = {"ids": task_id}
self.delete_task_api(task_payload)
def import_vehicle_api(self, payload: dict):
"""
导入主车
@param payload: eg {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
@return:
"""
return self.send(self.post, import_vehicle_url, json=payload).json()
def get_vehicle_data(self, file_path: str):
vehicle_info = FileReader(file_path).read_json()
vehicle_id = vehicle_info.get("id")
return vehicle_info, vehicle_id
def import_vehicle(self, file_path: str) -> str:
"""
导入主车
@param file_path: 主车文件
@return: 主车id
"""
vehicle_data, vehicle_id = self.get_vehicle_data(file_path)
payload = {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
result = self.import_vehicle_api(payload)
return result["data"]["id"]
def get_vehicle_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get, get_vehicle_url).json()
def get_vehicle(self, userid: str = None):
"""
根据用户信息获取主车信息
@param userid:
@return:
"""
global vehicle_data
vehicle_data = {}
res = self.get_vehicle_api()
# print(f"get_vehicle_api response ------------------------->: {res}")
data = res["data"]["list"]["byId"]
id_list = res["data"]["list"]["allIds"]
if userid:
for id in id_list:
if data[id]["userId"] == userid:
vehicle_data.setdefault(data[id]["name"], data[id]["id"])
else:
for id in id_list:
vehicle_data.setdefault(data[id]["userId"], {}).update({data[id]["name"]: data[id]["id"]})
print("获取到的主车信息->：{}".format(str(vehicle_data)))
return vehicle_data
def get_vehicle_id(self, vehicle_name: list) -> list and dict:
"""
根据主车名称获取主车ID
@param vehicle_name:
@return:
"""
vehicle_id_dict = {}
vehicle_id_list = []
vehicle_list = []
for i in vehicle_name:
for k, v in self.get_vehicle().items():
vehicle_list.append(v)
for vehicle_dict in vehicle_list:
# print("vehicle_dict", vehicle_dict)
vehicle_dict_key = dict(vehicle_dict).items()
for vehicle_dict_k, vehicle_dict_v in vehicle_dict_key:
if i in vehicle_dict_k:
vehicle_name_id = vehicle_dict[i]
vehicle_id_list.append(vehicle_name_id)
vehicle_id_dict.setdefault(i, vehicle_name_id)
print("vehicle_id_list->:", vehicle_id_list, "\nvehicle_id_dict->:", vehicle_id_dict)
return vehicle_id_list, vehicle_id_dict
def delete_vehicles_api(self, vehicles_id: str):
"""
删除主车
@param vehicles_id: 主车id
@return:
"""
url = delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
def delete_vehicle(self, id: str):
"""
删除主车的方法
@param id: 主车id
@return:
"""
response = self.delete_vehicles_api(id)
def get_suite_api(self, isProjectItem : bool):
"""
@param isProjectItem : True(团队) or False(False)
@return:
"""
url = get_suites_url.format(isProjectItem)
return self.send(self.get, url=url).json()
def get_suite(self, suite_name: str = None , isProjectItem : bool = True):
"""
@param isProjectItem : true(团队) or false(个人)
:return:
"""
if isProjectItem: isProjectItem = "true"
else: isProjectItem = "false"
response = self.get_suite_api(isProjectItem)
data = response['data']
suite_dict = {}
for id in data['allIds']:
suite_dict.setdefault(data['byId'][id]['name'], {}).update({"caseIds": data['byId'][id]['caseIds']})
print("suite_dict", suite_dict)
try :
if not suite_name:
return suite_dict
else:
# print("suite_dict[suite_name]:",suite_dict[suite_name])
return suite_dict[suite_name]
except Exception as e:
print("suite_name不存在")
def run_suite_api(self, payload: dict):
"""
run_suite
@param payload: payload
@return:
"""
return self.send(self.post, url=run_suite_url, json=payload).json()
def run_suite(self, suite_name: str, taskName: str = None, vehicle_id: str = None):
if taskName is None:
def creat_task_name():
return "taskName_" + time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
taskName = creat_task_name()
caseid_list = self.get_suite(suite_name)['caseIds']
payload = {
'caseIds': caseid_list,
'taskName': taskName,
"type" : "worldsim",
"enableStateMachine" : False,
"vehicleConfig" : {
"Ego" : {
"instanceId" : "Ego" ,
"id" : "default",
"name" : "主车",
"classId" :"MKZ",
"algorithms" : [
{
"id" : "Default",
"name" : "默认控制器",
"algorithmId": "SimOneDriver",
}
]
}
},
"withEvaluation" : False
}
if vehicle_id:
# vehicle_module = VehicleBusiness()
# vehicle_module.get_vehicle()
payload.update({"overrideVehicleId": vehicle_id})
response = self.run_suite_api(payload)
data = response['data']
return data['sessionId'], data['taskIds']
def suite_queue_api(self):
"""
suite_queue
@return:
"""
return self.send(self.get, url=queue_status_url).json()
def suite_queue_check(self, n=10):
"""
@param n:cycle index
@return:
"""
if n == 1:
print("-----------------------The test case run fail--------------------------------------")
return False
while (1):
try:
assert self.suite_queue_api()
response = self.suite_queue_api()
assert response["code"] == 0
except Exception as e:
return -1
queue_info = response["data"]
running, pending, waiting = queue_info["running"], queue_info["pending"], queue_info["waiting"]
if running == 0 and pending == 0 and waiting == 0:
print("-----------------------The test case run finish--------------------------------------")
return True
else:
time.sleep(interval)
def get_result(self, task_id: list, index: int = None):
"""
:param task_id:
:return:
"""
task_id_str = ",".join(task_id)
# print(task_id_str)
response = self.send(self.get, url=result_url.format(task_id=task_id_str)).json()
data = response["data"]["tasks"]
result = []
for one in data:
result.append({"pass": one["pass"], "is_ended": True if one["key"] == "ended" else False})
if not index:
return result
else:
return result[index]
def is_ended(self, task_id: list, case_name=None):
"""
判断案例是否运行结束
@param task_id: 任务id
@param case_name: 案例名称
@return:
"""
interval = 10
while (1):
result = self.get_result(task_id, -1)
if result["is_ended"]:
print('Case status：end of run')
if case_name:
print(f'Name of the current ending case->{case_name}')
return False
else:
print('Case status：in progress')
if case_name:
print(f'Name of the current running case->{case_name}')
time.sleep(interval)
def get_taskassemble_api(self, task_set_id):
'''
获取测试案例集
@param
@return:
'''
return self.send(self.get, get_taskassemble_url.format(task_set_id=task_set_id)).json()
def get_task_id_of_task_set(self, task_set_id: str) -> list:
"""
获取任务集合中所有测试案例的task_id
@param task_set_id: 任务集合id
@return: list
"""
response = self.get_taskassemble_api(task_set_id)
_task_id_list = response["data"]["list"]
# print("---------------------------------task_id_list-----------------------")
# print(_task_id_list)
index = 0
task_id_list = []
for i in _task_id_list:
index += 1
# print(f"------------------------第{index}个元素----------------")
# print(i)
if i['parentId'] == task_set_id:
# print(i)
task_id_list.append(i["id"])
# print("task_id_list",task_id_list)
return task_id_list
def get_task_result_api(self, task_id: str):
"""
获取指定task的结果
@param task_id:
@return:
"""
return self.send(self.get, task_result_url.format(id=task_id)).json()
def get_task_result(self, task_id_list: list = None) -> list:
"""
获取任务集中的 案例名称，caseId，task_id,运行结果
@param task_id_list: 任务id列表
@param task_set_id: 任务集合id
@return:
"""
task_result_list = []
print(f"task_id_list ===== >>> {task_id_list} ")
for task_id in task_id_list:
result = self.get_task_result_api(task_id)
case_name = result["data"]["tasks"][0]["case"]["name"]
case_id = result["data"]["tasks"][0]["caseId"]
task_id = result["data"]["tasks"][0]["id"]
task_result = result["data"]["tasks"][0]["pass"]
if task_result != True: task_result = False
task_result_list.append({"case_name": case_name,
"case_id": case_id,
"task_id": task_id,
"task_result": task_result})
print("task_result_list->", task_result_list)
return task_result_list
def import_map_api(self, payload: dict, files: dict):
"""
导入地图API
@param payload:传参字典
@param files: 传参文件
@return:
"""
self.headers = {"projectId": "default"}
return self.send(self.post, import_map_url, data=payload, files=files).json()
def import_map(self, xodr_path: str, thumbnail_path: str):
"""
导入地图方法
@param xodr_path: 地图文件
@param thumbnail_path: 地图对应的图像
@return:
"""
map_name = "map_" + str(random.randint(0, 9999))
payload = {"params": json.dumps(
{"category": "customized", "id": "", "name": os.path.splitext(os.path.basename(xodr_path))[0], "size": 512,
"ppm": 10, "bgColor": "#dddddd", "reproject": True, "reprojectOrigin": False, "reprojectOriginLat": 0,
"reprojectOriginLng": 0, "tags": [], "notes": "",
"header": {"minX": -210.20535534122396, "minY": -149.68815701999185, "minZ": -1.862645149230957e-9,
"maxX": 237.95535534122394, "maxY": 135.43815701999196, "maxZ": 2.7940070024635385e-9,
"centerX": 97.12480158531203, "centerY": 24.463606820251727, "centerZ": 100,
"localEnuExt": "6378137,0,0;0,1,0;0,0,1;1,0,0"}})}
files = {'xodr': open(xodr_path, 'rb'), "thumbnail": open(thumbnail_path, 'rb')}
result = self.import_map_api(payload, files)
if result["code"] == 0:
print(f"import map {xodr_path}success")
else:
raise f"import map {xodr_path} fail"
def get_map_api(self):
"""
获取地图
@return:
"""
return self.send(self.get, get_map_url).json()
def get_map_id(self, userid: str = None, map_name: str = None):
"""
按条件获取地图信息，默认获取所有地图
@param userid: 用户id
@param map_name: 地图名字
@return:
"""
global map_data
map_data = {}
try:
result = self.get_map_api()
# print(len(result['data']['maps']))
map_info = result['data']['maps']
if userid:
for map in map_info:
if map["userId"] == userid:
map_data.setdefault(map["name"], map["id"])
elif map_name:
for map in map_info:
if map["name"] == map_name:
map_data = {map["name"]: map["id"]}
else:
for map in map_info:
map_data.setdefault(map["userId"], {}).update({map["name"]: map["id"]})
return map_data
except Exception as e:
print(e)
def delete_map_api(self, payload: dict):
"""
删除地图
@param payload: eg：{ids: ["27a0bbc4-5d4d-48c1-9187-62df794dadae"]}
@return:
"""
return self.send(self.post, delete_map_url, json=payload).json()
def delete_map(self, id: list):
"""
批量删除地图
@param id: 地图id列表
@return:
"""
print("删除的地图列表-》{}".format(str(id)))
payload = {"ids": id}
self.delete_map_api(payload)
def get_judgements_api(self, caseId, judgementId=None):
"""
获取判定信息API
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
if judgementId:
url = get_judgement_url.format(caseId=caseId, judgementId=judgementId)
else:
url = get_judgements_url.format(caseId=caseId)
response = self.send(self.get, url).json()
return response
def get_judgements_method(self, caseId: str, judgementId: str = None):
"""
获取判定信息
@param caseId: 案例ID
@param judgementId: 判定ID
@return:
"""
response = self.get_judgements_api(caseId, judgementId)
print("得到的扩展判定信息：", response["data"][1])
return response
def update_judgement_api(self, caseId, judgementId="collision", payload: dict = None):
"""
@param caseId: caseId
@param judgementId: judgementId
@return:
"""
if not payload:
payload = {"schema": "judgement", "settings": {"logLevel": "error", "action": "failure", "logInfo": ""},
"scope": {"size": {"x": 10, "y": 10, "z": 0}, "heading": {"w": 1, "x": 0, "y": 0, "z": 0},
"position": {"x": 0, "y": 0, "z": 0},
"type": "global"}, "builtIn": True,
"name": "碰撞", "lock": True, "id": "collision",
"type": "collision", "category": "general",
"conditions": [], "userId": "admin",
"enabled": True}
response = self.send(self.put, get_judgement_url.format(caseId=caseId, judgementId=judgementId),
data=json.dumps(payload))
return response
def update_judgement_method(self, caseId, judgementId="collision"):
__collision = {"schema": "judgement", "settings": {"logLevel": "error", "action": "failure", "logInfo": ""},
"scope": {"size": {"x": 10, "y": 10, "z": 0}, "heading": {"w": 1, "x": 0, "y": 0, "z": 0},
"position": {"x": 0, "y": 0, "z": 0},
"type": "global"}, "builtIn": True,
"name": "碰撞", "lock": True, "id": "collision",
"type": "collision", "category": "general",
"conditions": [], "userId": "admin",
"enabled": True}
payload = json.dumps(__collision)
judgement_url = "api-asset/cases/{caseId}/judgements/{judgementId}"
response = self.send(self.put, judgement_url.format(caseId=caseId, judgementId=judgementId), data=payload)
print("更新扩展判定信息")
print("status_code:", response.status_code)
""" ——————————————————————————【0721新增编辑案例及依赖的接口】———————————————————————— """
def update_case(self, case_id: str, **kwargs):
"""
编辑案例
@param case_id: 案例ID
@param map_name: 地图名称
@param map_id: 地图ID
@param vehicle_id: 主车ID
@param AbsoluteTargetSpeed: 主车初始状态绝对目标速度
@param WorldPosition_x: 主车初始状态世界位置x
@param WorldPosition_y: 主车初始状态世界位置y
@param WorldPosition_z: 主车初始状态世界位置z
@return: 更新主车或地图等信息后的案例
"""
# 提取参数（设置默认值）
map_name = kwargs.get("map_name")
map_id = kwargs.get("map_id")
vehicle_id = kwargs.get("vehicle_id")
absolute_target_speed = kwargs.get("absolute_target_speed", 0.0)
worldposition_x = kwargs.get("worldposition_x", 0.0)
worldposition_y = kwargs.get("worldposition_y", 0.0)
worldposition_z = kwargs.get("worldposition_z", 0.0)
case_def = self.case_def_api(case_id=case_id)
case_data = self.case_data_api(case_id=case_id)
# print("临时测试 原case_def：", case_def)
# print("临时测试 原case_data：", case_data)
# print("update_case case_id:", case_id)
# print("update_case map_name:", map_name)
# print("update_case map_id:", map_id)
# print("update_case vehicle_id:", vehicle_id)
# 修改主车
if vehicle_id:
vehicle_name = self.get_vehicle_name(vehicle_id)
case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"][
"name"] = vehicle_name
case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"]["Properties"][
"Property"][0]["value"] = vehicle_id
for key in case_data["data"]["openSCENARIO"]["Entities"]["ScenarioObject"][0]["Vehicle"]["Properties"][
"Property"]:
if key["name"] == 'model':
key["value"] = vehicle_id
print(key)
if key["name"] == 'name':
key["value"] = vehicle_name
print(key)
# 修改地图
if map_id:
case_def["data"]["data"]["mapId"] = map_id
case_def["data"]["data"]["mapName"] = map_name.split(".")[0]
case_data["data"]["openSCENARIO"]["RoadNetwork"]["LogicFile"]["filepath"] = map_id + "." + \
map_name.split(".")[1]
case_payload = {"casedef": case_def["data"]["data"],
"casedata": case_data["data"],
"needRefresh": True}
# 修改主车初始状态世界位置
if worldposition_x:
privates = case_data["data"]["openSCENARIO"]['Storyboard']['Init']['Actions']['Private']
for private in privates:
if private["entityRef"] == "Ego":
for PrivateAction in private["PrivateAction"]:
if PrivateAction.get('TeleportAction'):
PrivateAction['TeleportAction']['Position']['WorldPosition']['x'] = worldposition_x
PrivateAction['TeleportAction']['Position']['WorldPosition']['y'] = worldposition_y
PrivateAction['TeleportAction']['Position']['WorldPosition']['z'] = worldposition_z
# 修改主车初始状态绝对目速度
if absolute_target_speed:
privates = case_data["data"]["openSCENARIO"]['Storyboard']['Init']['Actions']['Private']
for private in privates:
if private["entityRef"] == "Ego":
for PrivateAction in private["PrivateAction"]:
if PrivateAction.get('LongitudinalAction'):
PrivateAction['LongitudinalAction']['SpeedAction']['SpeedActionTarget']['AbsoluteTargetSpeed']['value'] = absolute_target_speed
response = self.update_case_api(payload=case_payload)
# print("临时测试 update_case response",response)
if response['code'] != 0:
raise Exception("编辑案例异常，接口返回：" + response)
return response
def case_def_api(self, case_id: str):
"""
案例概要
@param case_id:
@return:
"""
return self.send(self.get, case_def_url.format(case_id=case_id)).json()
def case_data_api(self, case_id: str):
"""
案例详情
@param case_id:
@return:
"""
return self.send(self.get, case_details_url.format(case_id=case_id)).json()
def get_vehicle_name(self, vehicle_id: str):
res = self.get_vehicle_api()
print(f"get_vehicle_api response: {res}")
data = res["data"]["list"]["byId"]
for key in data:
if key == vehicle_id:
print(f"找到主车ID为{vehicle_id}的主车名称：{data[key]['name']}")
return data[key]["name"]
print(f"没有找到主车ID为{vehicle_id}的主车")
return None
def update_case_api(self, payload: dict):
"""
案例编辑
@param payload:
@return:
"""
# print("********* update_case_api payload********",payload)
self.headers.update({"Content-Type": "application/json"})
return self.send(self.post, update_case_url, data=json.dumps(payload)).json()
def testPlan_list_api(self):
"""
获取测试计划列表
"""
options= {"search":"","types":[],"vehicleIds":[],"algorithms":[],"triggerTypes":[],"notifications":[],"category":"worldsim"}
json_str = json.dumps(options, indent=4, ensure_ascii=False) # 缩进美化，保留非ASCII字符
# encoded_url = quote(json_str, safe="/:?=") # 保留路径分隔符、查询符号和等号
return self.send(self.get, url=testPlan_list_url.format(options=json_str)).json()
def test_testPlan_list(self):
"""
获取测试计划列表
"""
response = self.testPlan_list_api()
return response
def get_testPlan_api(self, PlanId):
"""
获取测试计划
"""
return self.send(self.get, url= testPlan_id_url.format(id=PlanId)).json()
def test_get_testPlan(self, PlanId):
"""
获取测试计划
@param PlanId: 测试计划ID
@return: response
"""
response = self.get_testPlan_api(PlanId)
return response
def create_testPlan_api(self, categoryIds, name, withEvaluation, evaluationPresetId, Default):
"""
创建测试计划
"""
payload = {
"data": {"caseSuiteIds": [], "categoryIds": categoryIds, "name": name,
"type": "1", "vehicleConfig": {"Ego": {"instanceId": "Ego", "id": "default", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "默认控制器",
"algorithmId": Default},
{"id": "AutoDrive", "name": "自驾控制器",
"algorithmId": "SimOneDriver"}],
"name": "主车"}}, "triggerType": "manual",
"withEvaluation": False, "overrideJudgementId": "-1", "evaluationPresetId": evaluationPresetId,
"evaluationOperatorId": "-1", "notification": {"type": "none", "condition": "any"},
"enableStateMachine": False}}
return self.send(self.post, create_testPlan_url, json=payload).json()
def test_create_testPlan(self, caseId : list ,test_case_title : str ,withEvaluation , evaluationPresetId , Default):
"""
@param caseId: 案例ID
@param test_case_title: 测试计划名称
@param withEvaluation: 是否进行评估
@param evaluationPresetId: 评估预设ID
@param Default: 主车算法
@return: response
"""
response = self.create_testPlan_api(caseId, test_case_title,
withEvaluation = withEvaluation,
evaluationPresetId= 1,
Default="SimOneDriver")
return response
def update_testPlan_api(self, PlanId, categoryIds, name):
"""
更新测试计划
"""
payload = {
"data": {"caseSuiteIds": [], "categoryIds": categoryIds, "name": name,
"type": "1", "vehicleConfig": {"Ego": {"instanceId": "Ego", "id": "default", "classId": "MKZ",
"algorithms": [{"id": "Default", "name": "默认控制器",
"algorithmId": "SimOneDriver"},
{"id": "AutoDrive", "name": "自驾控制器",
"algorithmId": "Manual"}],
"name": "主车"}}, "triggerType": "manual",
"withEvaluation": False, "overrideJudgementId": "-1", "evaluationPresetId": 1,
"evaluationOperatorId": "-1", "notification": {"type": "none", "condition": "any"},
"enableStateMachine": False}}
return self.send(self.put, url=testPlan_id_url.format(id=PlanId), json=payload).json()
def test_update_testPlan(self, PlanId : str ,caseId : list, NewCaseTitle):
"""
@param PlanId: 测试计划ID
@param caseId: 案例ID
@param NewCaseTitle: 新的测试计划名称
@return: update_response
"""
update_response = self.update_testPlan_api(PlanId, caseId, NewCaseTitle)
return update_response
def delfel_testPlan_single_api(self, PlanID:str):
"""
删除单个测试计划
"""
response = self.send(self.delete, url=del_testPlan_single_url.format(id=PlanID)).json()
return response
def delfel_testPlan_single(self,PlanId : str ):
"""
删除单个测试计划
@param PlanId: 测试计划ID
@return: response
"""
response = self.delfel_testPlan_single_api(PlanId)
return response
def del_testPlan_api(self, ids:list):
"""
删除多个测试计划
"""
payload = {"ids": ids}
return self.send(self.post, url=del_testPlan_url, json=payload).json()
def test_del_testPlan(self, planID):
"""
删除多个测试计划
@param planID: 测试计划ID
@return: del_response
"""
del_response = self.del_testPlan_api([planID])
return del_response
def execute_testPlan_api(self, ids:list):
"""
执行测试计划
"""
payload = {"ids":ids}
return self.send(self.post, url=execute_testPlan_url, json=payload).json()
def test_execute_testPlan(self, PlanId : list):
"""
执行测试计划
@param planID: 测试计划ID
@return: response
"""
response = self.execute_testPlan_api(PlanId)
return response
def clone_testPlan_api(self, id):
"""
克隆测试计划
"""
payload = {}
return self.send(self.post, url=clone_testPlan_url.format(id=id), json=payload).json()
def test_clone_testPlan(self, PlanId : str):
"""
克隆测试计划
@param planID: 测试计划ID
@return: response
"""
response = self.clone_testPlan_api(PlanId)
return response
def isrepeat_testPlan_api(self, name, type):
"""
判断测试计划名称是否重复
"""
return self.send(self.get, url=isrepeat_testPlan_url.format(name=name, type=type)).json()
def test_isrepeat_testPlan(self, PlanName):
"""
判断测试计划名称是否重复
@param PlanName: 测试计划名称
@return: response
"""
response = self.isrepeat_testPlan_api(PlanName + time.strftime("_%H:%M:%S", time.localtime()), "worldsim")
return response
def main(category_name: str, case_name: str = None, vehicle_name=None,version: str = None):
"""
运行任务的入口主函数
@param category_name: category_name
@param case_name:
@param vehicle_name:vehicle_name
@param version: version
@return:
"""
suite = Suite()
# suite.create_category() # 创建案例库测试通过
if not version:
version = suite.get_version()
print("请求simone version接口提取版本号为：" + version)
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
# # 编辑案例，参数：案例ID、初始化状态主车绝对目标速度、初始状态主车世界位置
# suite.update_case(case_id = "f1c2b1f0-9cce-11f0-b520-f5c71358353c", absolute_target_speed = 111.123, worldposition_x = -334.9999998954549, worldposition_y = -3.000050967810888, worldposition_z = 1.1)
# 主车控制相关逻辑
if vehicle_name:
vehicle_dict = suite.get_vehicle_id([vehicle_name])[1]
if vehicle_name not in vehicle_dict.keys():
print(f"vehicle_name：{vehicle_name} inexistence")
return
vehicle_id = vehicle_dict[vehicle_name]
else:
vehicle_id = None
print(f"vehicle_name:{vehicle_name},vehicle_id:{vehicle_id}")
# 案例名称逻辑
if case_name:
case_id_list = cases_dict[case_name]
print(f" case_name:{case_name}")
# 启动案例
session_id, task_ids = suite.run_task(case_id_list, vehicle_id=vehicle_id, version=version)
time.sleep(10)
# suite.stop_task(task_id=task_ids, sessions_id=[session_id])
# 等待案例运行完毕
falg = suite.suite_queue_check()
if falg is True:
# 获loginData取案例运行结果
suite.get_task_result(task_ids)
SimOneUrl = "http://127.0.0.1:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
# LoginPolicy("cloud")
LoginPolicy("standalone")
if __name__ == '__main__':
# main("转向冲突", "转向冲突8",None, "3.7.0")
fire.Fire(main)
# eg:
# 1. Run one case :python runSimOneCase.py --category_name='入门案例' --case_name='构建标准案例2.0'
# 2. Run one case and switch the vehicle :python runSimOneCase.py --category_name='入门案例' --case_name='构建标准案例2.0' --vehicle_name='手动控制-默认'
# 3. Run all cases under the case category :python runSimOneCase.py --category_name='入门案例'
# 4. Run all cases under the case category and switch the vehicle :python runSimOneCase.py --category_name='入门案例’ --vehicle_name='手动控制-默认'