主车 / 测试集 / 地图 / 判定 / 结果示例
9. 主车相关
9.1 导入主车
API
"""
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()
"""
功能实现
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)
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(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 main(suite_name:str=None):
suite = Suite()
# suite.get_suite(suite_name)
suite.import_vehicle('c:/test.json')
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/sessions/delete', 'headers': {'Content-Type': 'application/json', 'Authorization': '5a3662a3-2fbb-4790-b40d-42f600e63801'}, 'json': {'ids': ['bc344bf1-4bd3-493e-a4d3-95f98dcc3e48']}}
response:{"code":0,"data":{}}
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/tasks/delete', 'headers': {'Content-Type': 'application/json', 'Authorization': '5a3662a3-2fbb-4790-b40d-42f600e63801'}, 'json': {'ids': ['18a0fafe-d1e2-404b-b5e8-b4f3ea4c8e86']}}
response:{"code":0,"msg":"success"}
9.2 删除主车
API
"""
def delete_vehicles_api(self, vehicles_id: str):
"""
删除主车
@param vehicles_id: 主车id
@return:
"""
url = self.delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
"""
功能实现
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:
"""
self.delete_vehicles_api(id)
def main(category_name:str,case_name:str=None,vehicle_name:str=None):
suite = Suite()
suite.get_category()
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
print(case_name)
# 主车控制相关逻辑
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}")
suite.delete_vehicle(vehicle_id)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("转向冲突", "转向冲突8", "test")
返回值
9.3 获取主车信息
API
"""
def get_vehicle_api(self):
"""
获取主车信息
@return:
"""
return self.send(self.get,self.get_vehicle_url).json()
"""
功能实现
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 main(category_name:str,case_name:str=None,vehicle_name:str=None):
suite = Suite()
suite = Suite()
suite.get_category()
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
# 主车控制相关逻辑
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)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("转向冲突", "转向冲突8", "手动控制-默认")
返回值
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-asset/vehicles', 'headers': {'Content-Type': 'application/json', 'Authorization': 'e5033d43-2368-4fd0-9c4b-299c4d3ba8d9'}}
10. 测试集相关
10.1 获取测试集
API
"""
def get_suite_api(self, isProjectItem : str):
"""
@param isProjectItem : True(团队) or False(False)
@return:
"""
url = get_suites_url.format(isProjectItem)
return self.send(self.get, url=url).json()
"""
运行测试集
def get_suite_api(self, isProjectItem : bool):
"""
@param isProjectItem : True(团队) or 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 main(suite_name:str=None):
suite = Suite()
suite.get_suite(suite_name)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("suite_name" , True)
返回值
response:{"code":0,"data":{"byId":{"ea6d0670-0db5-11ef-98b6-f5137907bac2":{"schema":"suite","id":"ea6d0670-0db5-11ef-98b6-f5137907bac2","userId":"standalone","parentId":"","name":"test","created":1715226044506,"lastModified":1715226044506,"createUserRole":1,"caseIds":["f5bbfb2c-619a-4812-90c2-75a84247a4db"]}},"allIds":["ea6d0670-0db5-11ef-98b6-f5137907bac2"]}}
suite_dict {'test': {'caseIds': ['f5bbfb2c-619a-4812-90c2-75a84247a4db']}}
10.2 运行测试集
API
"""
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
def run_suite_api(self, payload:dict):
"""
run_suite
@param payload: payload
@return:
"""
return self.send(self.post, url=self.run_suite_url, json=payload).json()
"""
功能实现
def get_suite_api(self):
"""
get_suite
@return:
"""
return self.send(self.get, url=get_suites_url).json()
def get_suite(self,suite_name:str=None):
"""
:return:
"""
response = self.get_suite_api()
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)
if not suite_name:
return suite_dict
else:
#print("suite_dict[suite_name]:",suite_dict[suite_name])
return suite_dict[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 main(suite_name:str=None):
suite = Suite()
# suite.get_suite(suite_name)
suite.run_suite(suite_name)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-task/tasks', 'headers': {'Content-Type': 'application/json', 'Authorization': '7d4c315b-dbd1-4f4b-8552-ff66be9346ec'}, 'json': {'caseIds': ['f5bbfb2c-619a-4812-90c2-75a84247a4db'], 'taskName': 'taskName_2024-05-09_11:49:56'}}
response:{"code":0,"data":{"sessionId":"5b4fc9fb-8da4-42f4-8eef-7e4522b85dd4","taskIds":["0b933b59-b4d6-49e8-b656-de52a8561fb6"]}}
10.3 获取运行案例集信息
API
"""
def suite_queue_api(self):
"""
suite_queue
@return:
"""
return self.send(self.get, url=self.queue_status_url).json()
"""
功能实现
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 main(suite_name:str=None):
suite = Suite()
# suite.get_suite(suite_name)
suite.run_suite(suite_name)
suite.suite_queue_check()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-task/tasks/queue?own=true', 'headers': {'Content-Type': 'application/json', 'Authorization': 'f083c671-f6fe-4d64-b775-3f0539eca7c9'}}
response:{"code":0,"data":{"total":2,"running":1,"pending":1,"waiting":1,"finished":0,"estimatedTaskDuration":87.22495652173913}}
10.4 获取任务集中的task_id
API
"""
def get_taskassemble_api(self):
'''
获取测试案例集
@param
@return:
'''
return self.send(self.get, get_taskassemble_url).json()
"""
功能实现
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 main(suite_name:str=None):
suite = Suite()
res = suite.run_suite(suite_name)
suite.get_task_id_of_task_set(res[1])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("test")
返回值
task_set_id_list [id1,id2,...]
11. 地图相关
11.1 导入地图
API
"""
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_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:
"""
payload = {"params": json.dumps(
{"category": "customized", "id": "new_" + Faker('zh_CN').uuid4(),
"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"导入地图{xodr_path}成功，id:{result['data']['mapId']}")
else:
print(f"导入地图{xodr_path}失败，返回结果：{result}")
raise f"导入地图{xodr_path}失败，返回结果：{result}"
res_dict = {"code": result["code"], "mapId": result["data"]["mapId"],
"name": os.path.basename(os.path.basename(xodr_path))}
return res_dict
def main():
suite = Suite()
suite.import_map(r"D:\Project\SimOneAutomation\simone_automation\Sysdata\Public\import\Map\xodr\标志牌测试.xodr",
r"D:\Project\SimOneAutomation\simone_automation\Sysdata\Public\import\Map\img\import_map.png"
)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
{'method': 'POST', 'url': 'http://172.31.9.85:30083/api-asset/maps', 'headers': {'Authorization': '92400a5e-0e20-4d34-bfb3-05f9344f4889'}, 'data': {'params': '{"category": "customized", "id": "", "name": "\\u6807\\u5fd7\\u724c\\u6d4b\\u8bd5", "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-09, "maxX": 237.95535534122394, "maxY": 135.43815701999196, "maxZ": 2.7940070024635385e-09, "centerX": 97.12480158531203, "centerY": 24.463606820251727, "centerZ": 100, "localEnuExt": "6378137,0,0;0,1,0;0,0,1;1,0,0"}}'}, 'files': {'xodr': <_io.BufferedReader name='D:\\Project\\SimOneAutomation\\simone_automation\\Sysdata\\Public\\import\\Map\\xodr\\标志牌测试.xodr'>, 'thumbnail': <_io.BufferedReader name='D:\\Project\\SimOneAutomation\\simone_automation\\Sysdata\\Public\\import\\Map\\img\\import_map.png'>}}
response:{"code":0,"data":{"mapId":"01ca5bba-d804-4d32-900f-742fa958d47f"}}
import map D:\Project\SimOneAutomation\simone_automation\Sysdata\Public\import\Map\xodr\标志牌测试.xodrsuccess
11.2 获取地图信息
API
"""
def get_map_api(self):
"""
获取地图
@return:
"""
return self.send(self.get, self.get_map_url).json()
"""
功能实现
def get_map_api(self):
"""
获取地图
@return:
"""
return self.send(self.get, self.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 main():
suite = Suite()
suite.get_map_id()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"data"...}
11.3 删除地图
API
"""
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_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 main():
suite = Suite()
suite.delete(["'92400a5e-0e20-4d34-bfb3-05f9344f4889'"])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"data"...}
12. 判定相关
12.1 获取案例判定信息
API
"""
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_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 main():
suite = Suite()
suite.get_judgements_api("ebd4cf06-b66f-4e03-9294-5f60f55d453a")
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
response:{"code":0,"msg":"success","data":[{"id":"timeout","name":"超时","type":"timeout","category":"general","enabled":true,"scope":{"type":"global","position":{"x":0,"y":0,"z":0},"heading":{"x":0,"y":0,"z":0,"w":1},"size":{"x":10,"y":10,"z":0}},"conditions":[{"variable":"timeout","value":600}],"settings":{"action":"failure","logLevel":"error","logInfo":""},"builtIn":true,"lock":true,"userId":"admin","schema":"judgement"},{"id":"collision","name":"碰撞","type":"collision","category":"general","enabled":true,"scope":{"type":"global","position":{"x":0,"y":0,"z":0},"heading":{"x":0,"y":0,"z":0,"w":1},"size":{"x":10,"y":10,"z":0}},"conditions":[],"settings":{"action":"failure","logLevel":"error","logInfo":""},"builtIn":true,"lock":true,"userId":"admin","schema":"judgement"}]}
12.2 更新判定信息
API
"""
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)
"""
功能实现
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)
def main(category_name: str, case_name: str):
"""
category_name:案例库名称
case_name：案例名称
reset:是否重置判定
"""
LoginPolicy("standalone")
suite = Suite()
suite.get_category()
# 获取caseid逻辑
cate_id = suite.get_category()
cate_name = [cate_id["data"][category_name]]
cases_dict, case_id_list = suite.get_cases_category(cate_name)
case_id = cases_dict[case_name][0]
suite.get_judgements_method(case_id)
suite.update_judgement_method(case_id)
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main("new", "test_4")
返回值
更新扩展判定信息
status_code: 202
13. 任务运行结果相关
13.1 获取运行案例集信息
API
"""
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_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 = []
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 main():
suite = Suite()
# 入参id taskid 可以用 run_task方法中的返回值获取
suite.get_task_result(["51ec8581-4563-49c4-8ce1-ba96962b278b"])
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
main()
返回值
task_result_list-> [{'case_name': 'test_4-copy', 'case_id': 'f5bbfb2c-619a-4812-90c2-75a84247a4db', 'task_id': '51ec8581-4563-49c4-8ce1-ba96962b278b', 'task_result': False}]
13.2 获取任务集列表信息
API
"""
def get_task_list_api(self):
"""
获取任务列表集合
"""
response = self.send(self.get, url=get_task_list_url).json()
"""
功能实现
def get_task_list_api(self):
"""
获取任务列表集合
"""
response = self.send(self.get, url=get_task_list_url).json()
调用示例
SimOneUrl = "http://172.31.9.85:30083/"
loginData = {
"username": 'admin',
"password": 'admin',
}
LoginPolicy("standalone")
if __name__ == '__main__':
test = Suite()
test.get_task_list_api()
返回值
response:{"code":0,"data":{"list":[{"parentId":null,"deleted":null,"deleteTime":null,"createAt":"2024-05-16T02:34:13.000Z","updateAt":"2024-05-16T02:35:08.000Z",....}
13.3 下载测试报告
API
"""
def export_report_api(self, sessions_id: str):
"""
导出测试报告API
@param sessions_id: sessions_id
@return: Response
"""
self.headers.update({"Language": "zh-Hans"})
return self.send(self.get, export_report_url.format(sessions_id=sessions_id)).json()
def download_report_api(self, sessions_id: str):
"""
下载测试报告API
@param sessions_id: sessions_id
@return: Response
"""
return self.send(self.get, download_report_url.format(sessions_id=sessions_id)).content
"""
功能实现
def export_report_api(self, sessions_id: str):
"""
导出测试报告API
@param sessions_id: sessions_id
@return: Response
"""
self.headers.update({"Language": "zh-Hans"})
return self.send(self.get, export_report_url.format(sessions_id=sessions_id)).json()
def download_report_api(self, sessions_id: str):
"""
下载测试报告API
@param sessions_id: sessions_id
@return: Response
"""
return self.send(self.get, download_report_url.format(sessions_id=sessions_id)).content
def download_report(self, sessions_id: str, download_path: str, download_name: str) -> None:
"""
下载测试报告
@param sessions_id: sessions_id
@return: Response
"""
self.export_report_api(sessions_id)
pdf_data = self.download_report_api(sessions_id)
download_pathname = os.path.join(download_path, download_name)
with open(download_pathname, mode="wb") as f:
f.write(pdf_data)
调用示例
LoginPolicy("standalone")
if __name__ == '__main__':
test = Suite()
# 测试任务集ID
test.download_report("54c0806b-ba80-4821-ae25-49f7c284f486", r"C:\Users\Administrator\Desktop\test", "test.pdf")
返回值
response:{"code":0,"data":{"success":true}}
测试报告二进制数据