RESTful API 参考手册
接口路径
__version__= "3.1"
class Prefix:
UserServer = "api-user"
CaseServer = "api-case"
AssetServer = "api-asset"
TaskServer = "api-task"
接口请求方法
class RquestApi():
def __init__(self):
super(RquestApi, self).__init__()
self._headers = {"Content-Type": "application/json"}
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)
print("response:%s"%str(response.content.decode("utf-8")))
self._status_code = response.status_code
self._response_code = response.json().get("code")
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
登录相关接口
单机版登录
接口
def standalone_login_api(self):
"""
接口名称：单机版登录
请求方式：get
请求路径：standalone_login_url = Prefix.UserServer + "/users/standalone"
"""
return self.send(self.get, self.standalone_login_url).json()
登录实现
class Enterprise(RquestApi):
def __init__(self):
super(Enterprise, self).__init__()
self.label = "Enterprise"
@property
def _get_token(self):
try:
response = standalone_login_api()
print('standalone login request result:' + str(response))
token = response['data']['token']
print("token:%s" % token)
return token
except Exception as e:
print(str(e))
raise print(e)
云端版登录
接口
def cloud_login_api(self, userinfo: dict):
"""
接口名称：云端版登录
请求参数：userinfo #userinfo只能以字典形式传参
请求方式：post
请求路径：cloud_login_url = Prefix.UserServer + "/users/login"
参数示例：userinfo: body {"username":xxx, "password": xxx}
"""
return self.send(self.post, self.cloud_login_url, json=userinfo).json()
def check_username_api(self, username:str):
"""
接口名称：用户校验
请求参数：username #字符串形式传参
请求方式：post
请求路径：cloud_login_url = Prefix.UserServer + "/users/check"
"""
payload = {'username': username}
return self.send(self.post, self.login_check, json=payload).json()
加密方式
import base64
from Crypto.Cipher import AES
class Encryption:
@staticmethod
def pad(text):
"""
填充函数，使被加密数据的字节码长度是block_size的整数倍
"""
length = AES.block_size
count = len(text.encode('utf-8'))
add = length - (count % length)
entext = text + (chr(add) * add)
return entext
@staticmethod
def str_aes(string:str,key:str):
"""
aes 加密
@param string:
@param key:
@return:
"""
aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
res = aes.encrypt(Encryption.pad(string).encode("utf8"))
msg = str(base64.b64encode(res), encoding="utf8")
return msg
@staticmethod
def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
'''
@param string:
@param username:
@param key: AES加密的key
@return:
'''
check_token = check_username_api(username)
target_string = check_token["data"]["checkToken"] + string
msg = Encryption.str_aes(target_string, key)
return msg
登录实现
class Cloud(RquestApi):
def __init__(self, login_data):
super(Cloud, self).__init__()
self.label = "Cloud"
self.login_data = login_data
@property
def _get_token(self):
# print(self.title)
try:
new_login_data = copy.deepcopy(self.login_data)
new_login_data['password'] = Encryption.get_str_aes(string=new_login_data['password'],
username=new_login_data['username'])
print('cloud login data:' + str(new_login_data))
response = cloud_login_api(userinfo=new_login_data)
print('cloud login request result:' + str(response))
token = response['data']['token']
print(token)
return token
except Exception as e:
print(str(e))
raise print('get token error ')
案例库目录相关接口
"""============================案例库目录接口================================="""
def get_category_api(self):
"""
接口名称：获取案例库列表
请求方式：get
请求路径：categories_url = Prefix.CaseServer + "/categories"
"""
return self.send(self.get, self.categories_url).json()
def create_category_api(self, cate_name:str):
"""
接口名称：创建案例库
请求参数：cate_name #字符串形式传参
请求方式：post
请求路径：categories_url = Prefix.CaseServer + "/categories"
"""
payload = {"parentId": "", "name": cate_name}
return self.send(self.post,self.categories_url,json=payload).json()
def get_category_case_api(self,cate_id:str):
"""
接口名称：获取当前案例库所有案例
请求参数：cate_id #字符串形式传参
请求方式：get
请求路径：get_cases_url = Prefix.CaseServer + "/cases?page=1&filter={category}&pageSize=100"
"""
return self.send(self.get,self.get_cases_url.format(category=cate_id)).json()
def delete_category_api(self, cate_id:str):
'''
接口名称：删除指定案例库
请求参数：cate_id #字符串形式传参
请求方式：delet
请求路径：delete_categories_url = Prefix.CaseServer + "/categories/{}"
'''
return self.send(self.delete,self.delete_categories_url.format(cate_id)).json()
上传文件相关接口
"""============================上传文件相关接口================================="""
class UploadFile(InitRquest):
def file_md5_check(self,file_path:str,raw:bool=False):
"""
接口名称：文件md5检查
请求参数：file_path: 文件路径 #file_path只能以字符串形式输入
row #raw:只能以bool类型输入
请求方式：get
请求路径：file_md5_url = Prefix.AssetServer + "/files/{}?raw={}"
"""
global url
file_md5 = HashTools.file_md5(file_path)
if raw:
url = self.file_md5_url.format(file_md5,"true")
else:
url = self.file_md5_url.format(file_md5, "false")
print(file_md5)
return self.send(self.get,url).json()
def upload_file(self, file_path:str,index:int,chunks:int,category:str,file_md5:str,raw:bool=False):
"""
接口名称：上传文件
请求参数：file_path: 文件路径 #字符串形式传参
index: 上传序列 #整数形式传参
chunks: chunks总数 #整数形式传参
category: 类别 #字符串形式传参
file_md5: 文件MD5 #字符串形式传参
请求方式：post
请求路径：upload_file_url = Prefix.AssetServer + "/files/{}/upload?raw={}"
"""
global url
self.headers={}
file_size = os.path.getsize(file_path)
file_name = file_path.split(os.sep)[-1]
files = {'file': open(os.path.join(file_path), 'rb')}
if not file_md5:
file_md5 = HashTools.file_md5(file_path)
if raw:
url = self.upload_file_url.format(file_md5,"true")
else:
url = self.upload_file_url.format(file_md5, "false")
upload_payload = {"params": json.dumps({"filename": file_name,
"index": index,
"chunks": chunks,
"filesize": file_size,
"uploadId": Generator.random_uuid(),
"category": category})}
return self.send(self.post, url, data=upload_payload, files=files).json()
测试案例相关接口
"""============================案例接口================================="""
def import_case_api(self,payload:dict):
"""
接口名称：案例导入
请求参数：payload #字典形式传递
请求方式：post
请求路径：case_import_url = Prefix.AssetServer+"/cases/import"
"""
#需要以json形式传递参数
return self.send(self.post, self.case_import_url,data=json.dumps(payload)).json()
def get_case_detail_api(self, case_id:str):
"""
接口名称：获取案例信息
请求参数：case_id #字符串形式传递
请求方式：get
请求路径：case_details_url = Prefix.CaseServer + "/cases/{case_id}/data"
"""
return self.send(self.get, url).json()
def delete_case_api(self, case_id:list):
'''
接口名称：删除案例
请求参数：case_id：案例id #列表形式传递
请求方式：post
请求路径：delete_case_url = Prefix.CaseServer + "/trash"
'''
payload = {"id": case_id}
return self.send(self.post,self.delete_case_url,json = payload).json()
测试任务相关接口
"""============================测试任务相关接口================================="""
def create_task_api(self,payload:dict):
"""
接口名称：创建task
请求参数：payload #字典形式传参
请求方式：post
请求路径：create_task_url = Prefix.TaskServer + "/tasks"
参数示例：payload: eg {"caseIds": ["3dd03c09-3ba9-4f7a-b988-fde88296095a"],
"taskName": "task_name"}
"""
return self.send(self.post,self.create_task_url,json=payload).json()
def stop_task_api(self,payload:dict):
"""
接口名称：task停止
请求参数：payload #字典形式传参
请求方式：post
请求路径：stop_task_url = Prefix.TaskServer + "/tasks/stop"
参数示例：payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
"""
return self.send(self.post,self.stop_task_url,json=payload).json()
def stop_sessions_api(self,payload:dict):
"""
接口名称：sessions停止
请求参数：payload #字典形式传参
请求方式：post
请求路径：stop_sessions_url = Prefix.TaskServer + "/sessions/stop"
参数示例：payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
"""
return self.send(self.post,self.stop_sessions_url,json=payload).json()
def delete_sessions_api(self,payload:dict):
"""
接口名称：sessions删除
请求参数：payload #以字典形式传参
请求方式：post
请求路径：delete_sessions_url = Prefix.TaskServer + "/sessions/delete"
参数示例：payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
"""
return self.send(self.post,self.delete_sessions_url,json=payload).json()
def delete_task_api(self,payload:dict):
"""
接口名称：task删除
请求参数：payload #字典形式传参
请求方式：post
请求路径：delete_task_url = Prefix.TaskServer + "/tasks/delete"
参数示例：payload: eg {"ids":["f9e096f6-78ff-431a-a5e2-24a2de4117df"]}
"""
return self.send(self.post,self.delete_task_url,json=payload).json()
def get_task_result_api(self,task_id:str):
"""
接口名称：获取指定task的结果
请求参数：task_id
请求方式：get
请求路径：task_result_url = Prefix.TaskServer + "/tasks/task?taskIds={id}"
参数类型：task_id只能以字符串形式传参
"""
return self.send(self.get,self.task_result_url.format(id=task_id)).json()
def pause_task_api(self,task_id:str):
"""
接口名称：暂停任务
请求参数：task_id #字符串形式传参
请求方式：post
请求路径：pause_task_url = Prefix.TaskServer + "/sessions/{task_id}/pause"
"""
return self.send(self.post, self.pause_task_url.format(task_id=task_id),json={}).json()
def resume_task_api(self,task_id:str):
"""
接口名称：暂停任务后开始
请求参数：task_id #字符串形式传参
请求方式：post
请求路径：resume_task_url = Prefix.TaskServer + "/sessions/{task_id}/resume"
"""
return self.send(self.post, self.resume_task_url.format(task_id=task_id),json={}).json()
主车相关接口
"""============================主车相关接口================================="""
def import_vehicle_api(self,payload:dict):
"""
接口名称：导入主车
请求参数：payload #字典形式传参
请求方式：post
请求路径：import_vehicle_url = Prefix.AssetServer + "/vehicles/import"
参数示例：payload: eg {"vehicleData": {"byId": {vehicle_id: vehicle_data},
"allIds": [vehicle_id]}}
"""
return self.send(self.post,self.import_vehicle_url,json=payload).json()
def get_vehicle_api(self):
"""
接口名称：获取主车信息
请求方式：get
请求路径：get_vehicle_url = Prefix.AssetServer + "/vehicles"
"""
return self.send(self.get,self.get_vehicle_url).json()
def delete_vehicles_api(self, vehicles_id: str):
"""
接口名称：删除主车
请求参数：vehicles_id：主车id #字符串形式传参
请求方式：delete
请求路径：delete_vehicle_url = Prefix.AssetServer + "/vehicles/{vehicles_id}"
参数类型：vehicles_id只能以字符串形式传参
"""
url = self.delete_vehicle_url.format(vehicles_id=vehicles_id)
return self.send(self.delete, url)
地图相关接口
"""============================地图相关接口================================="""
def import_map_api(self,payload:dict,files:dict):
"""
接口名称：导入地图
请求参数：payload #字典形式传参
files #字典形式传参
请求方式：post
请求路径：import_map_url = Prefix.AssetServer+"/maps"
参数示例：
data: eg: {params:
{"category":"test_field","id":"member21","name":"member21",
"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,
"localEnuExt":"6378137,0,0;0,1,0;0,0,1;1,0,0"
}
}
}
files: {"xodr": binary, "thumbnail": binary}
"""
self.headers = {}
return self.send(self.post,self.import_map_url,data=payload,files=files).json()
def get_map_api(self):
"""
接口名称：获取地图
请求方式：get
请求路径：get_map_url = Prefix.CaseServer+"/maps?pageSize=100"
"""
return self.send(self.get, self.get_map_url).json()
def start_sce_api(self,payload:dict):
"""
接口名称：启动路网服务
请求参数：payload #字典形式传参
请求方式：post
请求路径：road_service_url = Prefix.TaskServer+"/sce"
参数示例：payload: {"mapId": mapId,"md5":md5}
"""
return self.send(self.post,self.road_service_url,json=payload).json()
def delete_map_api(self,payload:dict):
"""
接口名称：删除地图
请求参数：payload #字典形式传参
请求方式：post
请求路径：delete_map_url = Prefix.CaseServer + "/maps/trash"
参数示例：payload: eg：{ids: ["27a0bbc4-5d4d-48c1-9187-62df794dadae"]}
"""
return self.send(self.post, self.delete_map_url, json=payload).json()
def laneTypeIfInside_api(self,map_id:str,payload:dict):
"""
接口名称：坐标是否在地图车道内
请求参数： map_id: 地图id #字符串形式传参
payload: 位置坐标 #字典形式传参
请求方式：post
请求路径：laneTypeIfInside_url = Prefix.TaskServer+"/sce/{map_id}/location/laneTypeIfInside"
参数示例：payload: eg：{ids: ["27a0bbc4-5d4d-48c1-9187-62df794dadae"]}
"""
return self.send(self.post,self.laneTypeIfInside_url.format(map_id=map_id),json=payload).json()
数据驱动相关接口
"""============================数据驱动相关接口================================="""
class DataDrivenAPI(InitRquest):
"""数据驱动接口"""
def import_data_driven_api(self,file_path:str,dd_serial:str):
"""
接口名称：导入数据驱动的全流程的接口
请求参数：file_path: 数据驱动文件 #字符串形式传参
dd_serial: 导入后序列号，防止重复 #字符串形式传参
请求方式：post
请求路径：import_data_driven_url = Prefix.AssetServer+"/dd"
"""
global id
file_size = os.path.getsize(file_path)
filename = os.path.basename(file_path)
disfilename = filename.split(".")[0] + "_" + dd_serial
filesize = os.path.getsize(file_path)
# api/dd
body = {"datatype": "excel",
"displayname": disfilename,
"filename": filename,
"filesize": filesize
}
log.info("***************" + str(body))
response = self.send(self.post,self.import_data_driven_url,json=body)
assert response.status_code == 200, "******* request error:%s ****************" % response.status_code
if response.status_code == 200:
if response.json()['code'] == 0:
id = response.json()['data']['id']
# api/dd/{id}
body1 = {"status": "uploading"}
self.send(self.put, self.import_data_driven_url+"/{}".format(id), json=body1)
# /api/files/{file_md5}?raw=true
md5_res = UploadFile().file_md5_check(file_path,raw=True)
file_md5 = md5_res['data']["id"]
if md5_res['data']['size'] == 0:
chunks = DataTools.round_up(file_size / gp.chunk_size)
for i in range(chunks):
UploadFile().upload_file(file_path=file_path,index=i,chunks=chunks,category="rawdata",file_md5=file_md5,raw=True)
# api/dd/move
move_param = {"source": file_md5, "target": filename}
log.info("move_param:%s" % move_param)
response_move = self.send(self.post,self.import_data_driven_url+"/move", json=move_param)
log.info("response_move status_code:%s" % response_move.status_code)
assert response_move.status_code == 204, "******* request error:%s ****************" % response_move.status_code
# api/dd/{id}
body2 = {"status": "upload-ok"}
response = self.send(self.put, self.import_data_driven_url + "/{}".format(id), json=body2)
assert response.status_code == 204, "******* request error:%s ****************" % response.status_code
# # api/dd/{id}/start
url_start = self.import_data_driven_url+"/{}/start".format(id)
response = self.send(self.post, url_start, json={})
assert response.status_code == 201, "******* request error:%s ****************" % response.status_code
return id
else:
log.info("资源已存在")
def delete_data_driven_api(self,id:str):
"""
接口名称：删除指定数据驱动
请求参数：id:数据驱动源id #字符串形式传参
请求方式：delete
请求路径：delete_data_dirven_url = Prefix.AssetServer+"/dd/{id}"
"""
return self.send(self.delete,self.delete_data_dirven_url.format(id)).json()
资源相关接口
"""============================资源相关接口================================="""
def import_assets_api(self,payload:dict):
"""
接口名称：导入资源API
请求参数：payload #字典形式传参
请求方式：post
请求路径：import_asset_url = Prefix.AssetServer+"/assets"
"""
return self.send(self.post,self.import_asset_url,data=json.dumps(payload)).json()
def assets_isrepeat_api(self,payload:dict):
"""
接口名称：资源查重API
请求参数：payload #字典形式传参
请求方式：post
请求路径：import_asset_url = Prefix.AssetServer+"/assets"
请求参数示例：payload: eg：{'name': 'API控制', 'schema': 'controller'}
"""
return self.send(self.post,self.asset_repeat_url,json=payload).json()
def get_assets_api(self,schema:str):
"""
接口名称：获取指定资源信息
请求参数：schema #字符串形式传参
请求方式：get
请求路径：get_asset_url = Prefix.AssetServer+"/assets?schema={}"
"""
return self.send(self.get,self.get_asset_url.format(schema)).json()
def delete_assets_api(self,id:str):
"""
接口名称：删除指定的资源
请求参数：id: 资源id #字符串形式传参
请求方式：delete
请求路径：delete_asset_url = Prefix.AssetServer+"/assets/{}"
"""
return self.send(self.delete,self.delete_asset_url).json()