使用参考手册¶

1. 步骤描述¶

创建一个python文件依次将接口路径、依赖、请求方法粘贴进去
顺序粘贴对应的 功能实现 模块代码
登录功能必须要有并且在首位调用所有的操作都需要依赖登录功能携带的token
粘贴调用示例并稍微根据自己的环境做一些参数修改比如请求路径修改，使用案例库案例名称修改等，即可成功调用API
修改以下几处位置：

# 路径地址与云端账户需要修改为自己对应的账户
SimOneUrl = "http://127.0.0.1:30083"
loginData = {
    "username": 'admin',
    "password": 'admin',
}


if __name__ == '__main__':
    # 云端登录
    LoginPolicy("cloud")
    # 单机版本登录
    LoginPolicy("standalone")

备注：附录参考代码是根据步骤描述中的顺序进行粘贴并可以成功运行的代码示例，如有疑惑请查看附录参考代码！！

2. 接口路径与依赖库¶

import base64
import copy
import inspect
import random
from os.path import exists

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

# 路径地址与云端账户需要修改为自己对应的账户
SimOneUrl = "http://127.0.0.1:30083/"
loginData = {
    "username": 'admin',
    "password": 'admin',
}

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)


"""  登录相关接口  """
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_details_url = Prefix.CaseServer + "/cases/{case_id}/data"  # 案例详情
delete_case_url = Prefix.CaseServer + "/trash"  # 删除案例
convert_replay_case_url = Prefix.TaskServer + "/tasks/{id}/convert2case"  # 转换回放案例
case_def_url = Prefix.CaseServer + "/cases/{case_id}"  # 案例概要
update_case_url = Prefix.AssetServer + "/cases/update"  # 编辑案例 # doing

"""  测试任务相关接口  """
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}"  # 删除停止  案例运行结果详情
task_info = Prefix.TaskServer + "/tasks/task"  # 删除停止  案例运行结果详情
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"  # get获取主车信息 post创建主车
get_vehicles_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 + "/{pake_id}"  # 案例下载

"""  测试案例集相关  """
get_suites_url = Prefix.AssetServer + "/suites"  # 获取测试集
run_suite_url = Prefix.TaskServer + "/tasks"  # 运行测试集
get_task_list_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 = "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}"  # 测试计划名称是否已存在


"""测试报告导出相关"""
export_report_url = Prefix.TaskServer + "/sessions/{sessions_id}/report" # 导出测试报告
download_report_url = "/report/{sessions_id}_zh-Hans.pdf" # 下载测试报告

3. 接口请求方法¶

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)
        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