Python 实现Metersphere平台API调用

实践环境

Python 3.9.13

安装依赖包

pip install pycryptodome
pip install requests

Metersphere v2.0.12

代码实现

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