admin管理员组文章数量:1437305
Python 实现Metersphere平台API调用
实践环境
Python 3.9.13
安装依赖包
代码语言：javascript代码运行次数：0运行复制pip install pycryptodome
pip install requestsMetersphere v2.0.12
代码实现
代码语言：javascript代码运行次数：0运行复制# -*- coding:utf-8 -*-
import base64
import os
import time
import uuid
import json
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
# Metersphere平台-个人信息-API Keys-Access Key
METER_SPHERE_ACCESS_KEY = os.environ.get('METER_SPHERE_ACCESS_KEY', 'vTiFyYFTfVAZfbRc')
# Metersphere平台-个人信息-API Keys-Secret Key
METER_SPHERE_SECRET_KEY = os.environ.get('METER_SPHERE_SECRET_KEY', 'N4iKP6cqT8zJLfcx')
METER_SPHERE_HOST = os.environ.get('METER_SPHERE_HOST', '192.168.88.135')
METER_SPHERE_PORT = os.environ.get('METER_SPHERE_PORT', '8081')
METER_SPHERE_PROTOCOL = os.environ.get('METER_SPHERE_PROTOCOL', 'http')
class MeterSphereCli(object):
def __init__(self):
self.session = requests.session()
self.set_headers(METER_SPHERE_ACCESS_KEY, METER_SPHERE_SECRET_KEY)
self.host = METER_SPHERE_HOST
self.port = METER_SPHERE_PORT
self.protocol = METER_SPHERE_PROTOCOL
def aes_encrypt(self, text, secret_key, iv: bytes = None):
cipher = AES.new(secret_key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))
encrypted = cipher.encrypt(pad(text.encode('utf-8'), AES.block_size))
# 通过aes加密后，再base64加密
b_encrypted = base64.b64encode(encrypted)
return b_encrypted
def set_headers(self, access_key, secret_key):
time_stamp = int(round(time.time() * 1000))
combox_key = access_key + '|' + str(uuid.uuid4()) + '|' + str(time_stamp)
signature = self.aes_encrypt(combox_key, secret_key, access_key)
header = {'Content-Type': 'application/json', 'ACCEPT': 'application/json', 'accessKey': access_key,
'signature': signature.decode('UTF-8')}
self.session.headers.update(header)
def get_workspaces(self, search_keyword=None, return_when_keyword_matched=True):
'''获取用户工作空间列表
@:param search_keyword 搜索关键词，仅支持 名称搜索
'''
result = []
url = f'{self.protocol}://{self.host}:{self.port}/track/workspace/list/userworkspace'
res = self.session.get(url)
if res.status_code != 200:
print'获取用户工作空间列表出错, 服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'获取用户工作空间列表出错, 服务器返回：%s' % res.text)
return result
if search_keyword:
for item in res.get('data', []):
workspace_id = item.get('id')
name = item.get('name')
if search_keyword in name:
result.append({'id': workspace_id, 'name': name})
if return_when_keyword_matched:
return result
else:
for item in res.get('data', []):
workspace_id = item.get('id')
name = item.get('name')
result.append({'id': workspace_id, 'name': name})
return result
def get_workspace_projects(self, workspace_id, search_keyword=None, return_when_keyword_matched=True):
'''获取工作空间关联的项目
@:param search_keyword 搜索关键词，仅支持 名称搜索
'''
result = []
url = f'{self.protocol}://{self.host}:{self.port}/track/project/list/related'
req_body = {'workspaceId': workspace_id}
res = self.session.post(url, json=req_body)
if res.status_code != 200:
print'获取工作空间关联的项目列表出错, 服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'获取工作空间关联的项目列表出错, 服务器返回：%s' % res.text)
return result
if search_keyword:
for item in res.get('data', []):
project_id = item.get('id')
name = item.get('name')
if search_keyword in name:
result.append({'id': project_id, 'name': name})
if return_when_keyword_matched:
return result
else:
for item in res.get('data', []):
project_id = item.get('id')
name = item.get('name')
result.append({'id': project_id, 'name': name})
return result
def get_project_envs(self, project_id, search_keyword=None, return_when_keyword_matched=True):
'''获取项目环境列表
@:param search_keyword 支持环境名称、主机+端口 模糊搜索
'''
result = []
url = f'{self.protocol}://{self.host}:{self.port}/track/environment/list/{project_id}'
res = self.session.get(url)
if res.status_code != 200:
print'获取项目环境列表出错, 服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'获取项目环境列表出错, 服务器返回：%s' % res.text)
return result
if search_keyword:
for item in res.get('data', []):
env_id = item.get('id')
name = item.get('name')
env_config = item.get('config').strip()
if env_config:
env_config = json.loads(env_config)
http_config = env_config.get('httpConfig', {})
condition = http_config.get('conditions', [{}])[0]
protocol = condition.get('protocol')
socket = condition.get('socket')
if search_keyword in name or search_keyword in socket:
result.append({'id': env_id, 'name': name, 'protocol': protocol, 'host_port': socket})
if return_when_keyword_matched:
return result
else:
for item in res.get('data', []):
env_id = item.get('id')
name = item.get('name')
env_config = item.get('config').strip()
if env_config:
env_config = json.loads(env_config)
http_config = env_config.get('httpConfig', {})
condition = http_config.get('conditions', [{}])[0]
protocol = condition.get('protocol')
socket = condition.get('socket')
result.append({'id': env_id, 'name': name, 'protocol': protocol, 'host_port': socket})
return result
def get_project_testplans(self, project_id, page_no=1, page_size=10, search_conditions={}):
'''获取项目计划列表
@param page: search_conditions {'name': {'value': 'test_plan_name' , 'operator': 'like'}}
'''
result = []
url = f'{self.protocol}://{self.host}:{self.port}/track/test/plan/list/{page_no}/{page_size}'
combine = {}
if 'name' in search_conditions:
search_condition = search_conditions.get('name')
combine['name'] = {
"operator": search_condition.get('operator'),
"value": search_condition.get('value')
}
req_body = {
"components": [{
"key": "name",
"name": "MsTableSearchInput",
"label": "commons.name",
"operator": {
"value": "like",
"options": [{
"label": "commons.adv_search.operators.like",
"value": "like"
}, {
"label": "commons.adv_search.operators.not_like",
"value": "not like"
}]
}
}, {
"key": "updateTime",
"name": "MsTableSearchDateTimePicker",
"label": "commons.update_time",
"operator": {
"options": [{
"label": "commons.adv_search.operators.between",
"value": "between"
}, {
"label": "commons.adv_search.operators.gt",
"value": "gt"
}, {
"label": "commons.adv_search.operators.lt",
"value": "lt"
}]
}
}, {
"key": "createTime",
"name": "MsTableSearchDateTimePicker",
"label": "commons.create_time",
"operator": {
"options": [{
"label": "commons.adv_search.operators.between",
"value": "between"
}, {
"label": "commons.adv_search.operators.gt",
"value": "gt"
}, {
"label": "commons.adv_search.operators.lt",
"value": "lt"
}]
}
}, {
"key": "moduleIds",
"name": "MsTableSearchNodeTree",
"label": "test_track.case.module",
"operator": {
"value": "in",
"options": [{
"label": "commons.adv_search.operators.in",
"value": "in"
}, {
"label": "commons.adv_search.operators.not_in",
"value": "not in"
}]
},
"options": {
"url": "/plan/node/list",
"type": "POST",
"params": {}
}
}, {
"key": "principal",
"name": "MsTableSearchSelect",
"label": "test_track.plan.plan_principal",
"operator": {
"options": [{
"label": "commons.adv_search.operators.in",
"value": "in"
}, {
"label": "commons.adv_search.operators.not_in",
"value": "not in"
}, {
"label": "commons.adv_search.operators.current_user",
"value": "current user"
}]
},
"options": {
"url": "/user/project/member/list",
"labelKey": "name",
"valueKey": "id"
},
"props": {
"multiple": True
}
}, {
"key": "status",
"name": "MsTableSearchSelect",
"label": "test_track.plan.plan_status",
"operator": {
"options": [{
"label": "commons.adv_search.operators.in",
"value": "in"
}, {
"label": "commons.adv_search.operators.not_in",
"value": "not in"
}]
},
"options": [{
"label": "test_track.plan.plan_status_prepare",
"value": "Prepare"
}, {
"label": "test_track.plan.plan_status_running",
"value": "Underway"
}, {
"label": "test_track.plan.plan_status_completed",
"value": "Completed"
}, {
"label": "test_track.plan.plan_status_finished",
"value": "Finished"
}, {
"label": "test_track.plan.plan_status_archived",
"value": "Archived"
}],
"props": {
"multiple": True
}
}, {
"key": "stage",
"name": "MsTableSearchSelect",
"label": "test_track.plan.plan_stage",
"operator": {
"options": [{
"label": "commons.adv_search.operators.in",
"value": "in"
}, {
"label": "commons.adv_search.operators.not_in",
"value": "not in"
}]
},
"options": [{
"text": "冒烟测试",
"value": "smoke",
"system": True
}, {
"text": "系统测试",
"value": "system",
"system": True
}, {
"text": "回归测试",
"value": "regression",
"system": True
}],
"props": {
"multiple": True
}
}, {
"key": "tags",
"name": "MsTableSearchInput",
"label": "commons.tag",
"operator": {
"value": "like",
"options": [{
"label": "commons.adv_search.operators.like",
"value": "like"
}, {
"label": "commons.adv_search.operators.not_like",
"value": "not like"
}]
}
}, {
"key": "followPeople",
"name": "MsTableSearchSelect",
"label": "commons.follow_people",
"operator": {
"options": [{
"label": "commons.adv_search.operators.in",
"value": "in"
}, {
"label": "commons.adv_search.operators.current_user",
"value": "current user"
}]
},
"options": {
"url": "/user/ws/current/member/list",
"labelKey": "name",
"valueKey": "id"
},
"props": {
"multiple": True
}
}, {
"key": "actualStartTime",
"name": "MsTableSearchDateTimePicker",
"label": "test_track.plan.actual_start_time",
"operator": {
"options": [{
"label": "commons.adv_search.operators.between",
"value": "between"
}, {
"label": "commons.adv_search.operators.gt",
"value": "gt"
}, {
"label": "commons.adv_search.operators.ge",
"value": "ge"
}, {
"label": "commons.adv_search.operators.lt",
"value": "lt"
}, {
"label": "commons.adv_search.operators.le",
"value": "le"
}]
}
}, {
"key": "actualEndTime",
"name": "MsTableSearchDateTimePicker",
"label": "test_track.plan.actual_end_time",
"operator": {
"options": [{
"label": "commons.adv_search.operators.between",
"value": "between"
}, {
"label": "commons.adv_search.operators.gt",
"value": "gt"
}, {
"label": "commons.adv_search.operators.ge",
"value": "ge"
}, {
"label": "commons.adv_search.operators.lt",
"value": "lt"
}, {
"label": "commons.adv_search.operators.le",
"value": "le"
}]
}
}, {
"key": "planStartTime",
"name": "MsTableSearchDateTimePicker",
"label": "test_track.plan.planned_start_time",
"operator": {
"options": [{
"label": "commons.adv_search.operators.between",
"value": "between"
}, {
"label": "commons.adv_search.operators.gt",
"value": "gt"
}, {
"label": "commons.adv_search.operators.ge",
"value": "ge"
}, {
"label": "commons.adv_search.operators.lt",
"value": "lt"
}, {
"label": "commons.adv_search.operators.le",
"value": "le"
}]
}
}, {
"key": "planEndTime",
"name": "MsTableSearchDateTimePicker",
"label": "test_track.plan.planned_end_time",
"operator": {
"options": [{
"label": "commons.adv_search.operators.between",
"value": "between"
}, {
"label": "commons.adv_search.operators.gt",
"value": "gt"
}, {
"label": "commons.adv_search.operators.ge",
"value": "ge"
}, {
"label": "commons.adv_search.operators.lt",
"value": "lt"
}, {
"label": "commons.adv_search.operators.le",
"value": "le"
}]
}
}],
"orders": [],
"nodeIds": [],
"projectId": project_id,
"selectAll": False,
"unSelectIds": [],
"combine": combine
}
res = self.session.post(url, json=req_body)
if res.status_code != 200:
print'获取项目计划列表出错, 服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'获取项目计划列表出错, 服务器返回：%s' % res.text)
return result
data = res.get('data', {})
item_count = data.get('itemCount', 0)
while item_count > 0:
for test_plan in data.get('listObject', []):
id = test_plan.get('id')
name = test_plan.get('name')
project_id = test_plan.get('projectId')
result.append({'id':id, 'name': name, 'project_id': project_id})
item_count -= page_size
if item_count > 0:
page_no += 1
url = f'{self.protocol}://{self.host}:{self.port}/track/test/plan/list/{page_no}/{page_size}'
res = self.session.post(url, json=req_body)
if res.status_code != 200:
print'获取项目计划列表出错, 服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'获取项目计划列表出错, 服务器返回：%s' % res.text)
return result
data = res.get('data', {})
return result
def get_resource_pool(self, search_keyword=None, return_when_keyword_matched=True):
''' 获取运行资源池
@:param search_keyword 搜索关键词,支持名称模糊搜索
'''
result = []
url = f'{self.protocol}://{self.host}:{self.port}/track/testresourcepool/list/quota/valid'
res = self.session.get(url)
if res.status_code != 200:
print'获取资源池失败，服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'获取资源池失败，服务器返回：%s' % res.text)
return result
if search_keyword:
for item in res.get('data'):
pool_id = item.get('id')
name = item.get('name')
if search_keyword in name :
result.append({'id': pool_id, 'name': name})
if return_when_keyword_matched:
return result
else:
for item in res.get('data'):
pool_id = item.get('id')
name = item.get('name')
result.append({'id':pool_id, 'name':name})
return result
def call_testplan(self, resource_pool_id, poject_id, test_plan_id, env_id):
''' 运行测试计划 '''
result = {}
url = f'{self.protocol}://{self.host}:{self.port}/track/test/plan/run'
req_body = {
"mode": "serial",
"reportType": "iddReport",
"onSampleError": False,
"runWithinResourcePool": False,
"resourcePoolId": resource_pool_id,
"envMap": {
poject_id: env_id
},
"testPlanId": test_plan_id,
"projectId": poject_id,
"userId": "admin",
"triggerMode": "MANUAL",
"environmentType": "JSON",
"environmentGroupId": "",
"testPlanDefaultEnvMap": {
},
"requestOriginator": "TEST_PLAN",
"retryEnable": False,
"retryNum": 1,
"browser": "CHROME",
"headlessEnabled": False,
"executionWay": "RUN"
}
res = self.session.post(url, json=req_body)
if res.status_code != 200:
print'执行测试计划失败，服务器返回：%s' % res.text)
return result
res = res.json()
if res.get('success') != True:
print'执行测试计划失败，服务器返回：%s' % res.text)
return result
result = {'report_id': res.get('data')}
return result
ms_cli = MeterSphereCli()
if __name__ == '__main__':
testplan_id = None
project_id = None
project_env_id = None
res_pool_id = None
#
res = ms_cli.get_workspaces('默认工作空间')
if res:
workspace_id = res[0].get('id')
projects = ms_cli.get_workspace_projects(workspace_id, search_keyword='默认项目')
if projects:
project = projects[0]
project_id = project.get('id')
testplans = ms_cli.get_project_testplans(project_id, 1, page_size=10, search_conditions={'name': {'value': '全局前置准备', 'operator':'like'}})
if testplans:
testplan = testplans[0]
testplan_id = testplan.get('id')
#
if project_id:
project_envs = ms_cli.get_project_envs(project_id, '测试环境')
if project_envs:
project_env = project_envs[0]
project_env_id = project_env.get('id')
pools = ms_cli.get_resource_pool('LOCAL')
if pools:
pool = pools[0]
res_pool_id = pool.get('id')
if res_pool_id and project_id and testplan_id and project_env_id:
res = ms_cli.call_testplan(res_pool_id, project_id, testplan_id, project_env_id)本文标签： Python 实现Metersphere平台API调用
版权声明：本文标题：Python 实现Metersphere平台API调用 内容由网友自发贡献，该文观点仅代表作者本人， 转载请联系作者并注明出处：http://www.betaflare.com/biancheng/1747486446a2699605.html， 本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容，一经查实，本站将立刻删除。
发表评论