python 中的 HP ALM 12 rest api 帮助

Question

对于 HP ALM 12.20.3264。

使用 python 2.7.9 - 希望完成自动化任务。其中 - 需要完成以下任务：

1. 连接 HP ALM  成功。请在下面找到成功消息。

   Log :(u'Open ALM session success', u'AUTH URL:', u'https://abcdefgh.com/qcbin/authentication-point/authenticate', u'HEADERS:', {'Cookie': None, 'Content-Type': 'application/xml', 'Accept': 'application/xml', 'KeepAlive': 'true'})
   

2. 从测试中获取测试用例名称信息 -> TestLab  失败并出现以下错误：

   (u'[ALMSession] Get ALM function with errors', 401, "Authentication failed. Browser based integrations - to login append '?login-form-required=y' to the url you tried to access.", u'PATH:', u'https://abcdefgh.com/qcbin/rest/domains/CORE_PRODUCTS/projects/NEO/Testing/TestLab', u'HEADERS:', {'Cookie': 'LWSSO_COOKIE_KEY=kFvs5DG2lK918ErK8Kf11u1bua_1bjLYpuPxw-1QCLBd3Pu4DoXZzCoVjuzMckASy-_87uA-5hGBnLd-atrhiMaRxkD2Ed79frDzx-qzWCCw-V0lSeWOXTWt57L-HdA9ZzWb3biMqaEnEdQvokPZteJKSgsXyMVqqRQgUrj3bB-ybLNuWngycagsTkLGnshoaNdqGaW6H_UVu7tOsNQxK2on3rMrbnqe2UrP6gPzyViBMPKFPRvuwhb_bsgPF8L3GdfWTbKg7u5Fz6cxq_eerwe2G8PrwFe2PzRC5D2VCHyxxAvk4trI4eUx4U5cVMPZ;Path=/;HTTPOnly', 'Content-Type': 'application/xml', 'Accept': 'application/xml', 'KeepAlive': 'true'})
   

3. 更新测试用例状态 - 通过/失败 ==> 尚未实施

能否请您帮助理解 HP ALM 12 公开的 RESTful api。

以下是我的示例 Python 脚本：

import requests
import xml.etree.ElementTree as ET

class ALMUrl:
    def __init__(self, qcurl, domain, project):
        self.__base = u'https://' + qcurl + u'/qcbin'
        self.__auth = self.__base + u'/authentication-point/authenticate'
        self.__logout = self.__base + u'/authentication-point/logout'
        self.__work = self.__base + u'/rest/domains/' + domain + u'/projects/' + project

    def get_auth(self):
        return self.__auth

    def get_logout(self):
        return self.__logout

    def __getattr__(self, *args):
        result = self.__work
        for arg in args:
            result += '/' + arg
        return result

class ALMSession:
    def __init__(self, login, password):
        try:

            self.__headers = {"Accept":"application/xml",
                              "Content-Type":"application/xml",
                              "KeepAlive":"true",
                              "Cookie": None}#"Authorization":"Basic " + base64.b64encode(login + ':' + password)}
            self.__user_pass = (login, password)
        except:
            print(u"Exception while creating ALMSession", self.__headers, self.__h)

  def Open(self, ALMUrl):
        #head, context = self.__h.request(ALMUrl.get_auth(), "GET", headers=self.__headers)
        r = requests.get(ALMUrl.get_auth(), auth=self.__user_pass)
        #if head.status is 200:
        if r.status_code is 200:
            print(u"Open ALM session success", u'AUTH URL:', ALMUrl.get_auth(), u'HEADERS:', self.__headers)
            self.__headers["Cookie"] = r.headers['set-cookie']                
            return 0
        else:
            print(u"Open ALM session", r.status_code, r.reason, u'AUTH URL:', ALMUrl.get_auth(), u'HEADERS:', self.__headers)
            return int(r.status_code)

    def Close(self, ALMUrl):
        if self.__headers["Cookie"] is not None:
            r = requests.get(ALMUrl.get_logout(), headers=self.__headers, auth=self.__user_pass)
            if r.status_code is 200:
                print(u"Close ALM session success", u'LOGOUT URL:', ALMUrl.get_logout(), u'HEADERS:', self.__headers)
                return 0
            else:
                print(u"Close ALM session", r.status_code, r.reason, u'LOGOUT URL:', ALMUrl.get_logout(), u'HEADERS:', self.__headers)
                return int(r.status_code)
        else:
            print(u"Close ALM session", u"1", u"httplib2.Http was not initialized")
            return 1
   def Get(self, ALMUrl, *args):
        if self.__headers["Cookie"] is not None:
            r = requests.get(ALMUrl.__getattr__(*args), headers=self.__headers, auth=self.__user_pass)
            if r.status_code == 200:
                print(u"[ALMSession] Get success", u"URL:", ALMUrl.__getattr__(*args), u"HEADERS:", self.__headers)
                res = []
                self.parse_xml(r.content, res)
                return 0, res
            elif r.status_code == 500:
                try:
                    if isinstance(r.text, unicode):
                        exceptionxml = ET.fromstring(r.text.encode('utf8','ignore'))
                    else:
                        exceptionxml = ET.fromstring(r.text)
                    print(u"[ALMSession] Get ALM function with errors", exceptionxml[0].text, exceptionxml[1].text, u"PATH:", ALMUrl.__getattr__(*args), u"HEADERS:", self.__headers)
                except ET.ParseError:
                    print(u"[ALMSession] Get ALM function with errors, returned message is not XML", u"PATH:", ALMUrl.__getattr__(*args), u"HEADERS:", self.__headers, ET.ParseError.message)
                return int(r.status_code), None
            else:
                print(u"[ALMSession] Get ALM function with errors", r.status_code, r.reason, u"PATH:", ALMUrl.__getattr__(*args), u"HEADERS:", self.__headers)

                return int(r.status_code), None
        else:
            print(u"[ALMSession] Get ALM function with errors", u"1", u"httplib2.Http not initialized")
            return 1, None

    def Update(self, ALMUrl, data, *args):
        if self.__headers["Cookie"] is not None:
            r = requests.put(ALMUrl.__getattr__(*args),
                              headers=self.__headers,
                              data=data,
                              auth=self.__user_pass)
            if r.status_code == 200:
                print(u"[ALMSession] Update success", u"URL:", ALMUrl.__getattr__(*args))
                return 0
            elif r.status_code == 500:
                if isinstance(r.text, unicode):
                    exceptionxml = ET.fromstring(r.text.encode('utf8','ignore'))
                else:
                    exceptionxml = ET.fromstring(r.text)
                print(u"[ALMSession] Update ALM function with errors", exceptionxml[0].text, exceptionxml[1].text, u"PATH:", ALMUrl.__getattr__(*args), u"DATA:", data, u"HEADERS:", self.__headers)
                return int(r.status_code)
            else:
                print(u"[ALMSession] Update ALM function with errors", r.status_code, r.reason, u"PATH:", ALMUrl.__getattr__(*args), u"DATA:", data, u"HEADERS:", self.__headers)
                return int(r.status_code)
        else:
            print(u"[ALMSession] Update ALM function with errors", u"1", u"httplib2.Http not initialized")
            return 1

if __name__ == '__main__':   

    qcurl = "almint.eu.abc.com"

    qcuname = "abc"
    qcpwd = "acb"
    qcdomain = "CORE_PRODUCTS"
    qcproject = "NEO"

    objALMUrl = ALMUrl(qcurl,qcdomain,qcproject)
    objALMSession = ALMSession(qcuname,qcpwd)
    objALMSession.Open(objALMUrl)
    objALMSession.Get(objALMUrl,"Testing/TestLab")  

    objALMSession.Close(objALMUrl)

Answer 1

以下代码涵盖了您的大部分要求。简而言之，此代码从 Jenkins 获取量角器测试的输出，并在 HP ALM 中创建测试集（如果不存在）并更新测试状态并附加报告。

要了解端点列表，请在您喜欢的浏览器中输入以下内容 <<ALM_SERVER>>/qcbin/rest/resouce-list

要了解限制和架构详细信息，请参阅 HP ALM 中的 GoTo 帮助。

import re
import json
import requests
import datetime
import time
import sys
from requests.auth import HTTPBasicAuth

protractor_result_file = './combined_result.json'

almUserName = ""
almPassword = ""
almDomain = ""
almProject = ""

almURL = "https://---/qcbin/"
authEndPoint = almURL + "authentication-point/authenticate"
qcSessionEndPoint = almURL + "rest/site-session"
qcLogoutEndPoint = almURL + "authentication-point/logout"
midPoint = "rest/domains/" + almDomain + "/projects/"

mydate = datetime.datetime.now()
testSetName = ""
assignmentGroup = ""
parser_temp_dic = {}

cookies = dict()

headers = {
    'cache-control': "no-cache"
}

'''
Function    :   alm_login
Description :   Authenticate user
Parameters  :   global parameter
                alm_username     -   ALM User
                alm_password     -   ALM Password
'''


def alm_login():
    response = requests.post(authEndPoint, auth=HTTPBasicAuth(almUserName, almPassword), headers=headers)
    if response.status_code == 200:
        cookieName = response.headers.get('Set-Cookie')
        LWSSO_COOKIE_KEY = cookieName[cookieName.index("=") + 1: cookieName.index(";")]
        cookies['LWSSO_COOKIE_KEY'] = LWSSO_COOKIE_KEY
    response = requests.post(qcSessionEndPoint, headers=headers, cookies=cookies)
    if response.status_code == 200 | response.status_code == 201:
        cookieName = response.headers.get('Set-Cookie').split(",")[1]
        QCSession = cookieName[cookieName.index("=") + 1: cookieName.index(";")]
        cookies['QCSession'] = QCSession
    return


'''
Function    :   alm_logout
Description :   terminate user session
Parameters  :   No Parameters
'''


def alm_logout():
    response = requests.post(qcLogoutEndPoint, headers=headers, cookies=cookies)
    print(response.headers.get('Expires'))
    return


'''
Function    :   parse_result
Description :   Parse protractor result file
Parameters  :   No Parameters
'''


def parse_result():
    try:
        f = open(protractor_result_file, 'r')
    except (FileNotFoundError) as err:
        print("File Not found error: {0}".format(err))
        return

    obj = json.load(f)
    test_set_id = find_test_set(find_test_set_folder(testSetPath), "test-sets")
    test_instance_data = "<Entities>"
    test_instance_data_put = "<Entities>"
    test_step_data = "<Entities>"

    # Get all the test id's if test plan folder exists already
    test_plan_details = find_test_plan_folder(testPlanPath)
    payload = {"query": "{test-folder.hierarchical-path['" + test_plan_details["hierarchical-path"] + "*']}",
               "fields": "id,name,steps", "page-size": 5000}
    response = requests.get(almURL + midPoint + "/tests", params=payload, headers=headers, cookies=cookies)
    all_tests = json.loads(response.text)

    # Get all test instance if test set exists already
    str_api = "test-instances"
    payload = {"query": "{cycle-id['" + test_set_id + "']}", "fields": "test-id", "page-size": 5000}
    response = requests.get(almURL + midPoint + "/" + str_api, params=payload, headers=headers, cookies=cookies)
    all_test_instance = json.loads(response.text)

    test_order = 0
    for spec in obj:
        for testSuite in spec:
            if len(spec[testSuite]['specs']) > 0:
                for test in spec[testSuite]['specs']:
                    outputTestName = re.sub('[^A-Za-z0-9\s]+', '', test['fullName']).strip()
                    # Check if the test case already exits in test plan
                    test_details = test_exists(outputTestName, all_tests)
                    test_case_exists = True
                    if len(test_details) == 0:
                        test_case_exists = False

                    if test_case_exists is True:
                        parser_temp_dic[int(test_details['id'])] = {'status': []}

                        # Check if test instance exists in test set
                    test_instance_exists = True
                    if test_case_exists == True:
                        parser_temp_dic[int(test_details['id'])]['status'].append(test['status'].capitalize())
                        if len(test_exists(test_details['id'], all_test_instance)) == 0:
                            test_instance_exists = False

                    if test_instance_exists is False and test_case_exists is True:
                        test_order += 1
                        test_instance_data = test_instance_data + "<Entity Type=" + chr(34) + "test-instance" + chr(
                            34) + "><Fields><Field Name=" + chr(
                            34) + "owner" + chr(34) + "><Value>" + almUserName + "</Value></Field><Field Name=" + chr(
                            34) + "subtype-id" + chr(
                            34) + "><Value>hp.qc.test-instance.MANUAL</Value></Field><Field Name=" + chr(
                            34) + "test-order" + chr(34) + "><Value>" + str(
                            test_order) + "</Value></Field><Field Name=" + chr(
                            34) + "cycle-id" + chr(
                            34) + "><Value>" + test_set_id + "</Value></Field><Field Name=" + chr(
                            34) + "test-id" + chr(34) + "><Value>" + str(
                            test_details['id']) + "</Value></Field></Fields></Entity>"
                        template_in = "{\"Type\": \"test-instance\", \"Fields\": [{\"Name\": \"id\", \"values\": [{\"value\"" \
                                      ": \"675\"}]}, {\"Name\": \"test-id\", \"values\": [{\"value\": \"" + str(
                            test_details['id']) + "\"}]}]}"
                        all_test_instance['entities'].append(json.loads(template_in))

    bulk_operation_post("test-instances", test_instance_data + "</Entities>", True, "POST")

    strAPI = "test-instances"
    payload = {"query": "{cycle-id['" + test_set_id + "']}", "fields": "id,test-id,test-config-id,cycle-id",
               "page-size": 5000}
    response = requests.get(almURL + midPoint + "/" + strAPI, params=payload, headers=headers, cookies=cookies)
    obj = json.loads(response.text)
    run_instance_post = "<Entities>"

    for entity in obj["entities"]:
        run_name = re.sub('[-:]', '_',
                          'automation_' + datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S'))
        temp_map = create_key_value(entity["Fields"])
        parser_temp_dic[int(temp_map['test-id'])]['testcycl-id'] = temp_map['id']
        parser_temp_dic[int(temp_map['test-id'])]['test-config-id'] = temp_map['test-config-id']
        parser_temp_dic[int(temp_map['test-id'])]['test-id'] = temp_map['test-id']
        parser_temp_dic[int(temp_map['test-id'])]['cycle-id'] = temp_map['cycle-id']
        # parser_temp_dic[int(temp_map['test-id'])]['status'].sort()
        status = "Passed"
        if 'Failed' in parser_temp_dic[int(temp_map['test-id'])]['status']:
            status = 'Failed'
        parser_temp_dic[int(temp_map['test-id'])]['final-status'] = status
        run_instance_post = run_instance_post + "<Entity Type=" + chr(34) + "run" + chr(
            34) + "><Fields><Field Name=" + chr(
            34) + "name" + chr(34) + "><Value>" + run_name + "</Value></Field><Field Name=" + chr(34) + "owner" + chr(
            34) + "><Value>" + almUserName + "</Value></Field><Field Name=" + chr(34) + "test-instance" + chr(
            34) + "><Value>1</Value></Field><Field Name=" + chr(34) + "testcycl-id" + chr(34) + "><Value>" + str(
            temp_map['id']) + "</Value></Field><Field Name=" + chr(34) + "cycle-id" + chr(34) + "><Value>" + str(
            temp_map['cycle-id']) + "</Value></Field><Field Name=" + chr(34) + "status" + chr(
            34) + "><Value>" + "Not Completed" + "</Value></Field><Field Name=" + chr(34) + "test-id" + chr(
            34) + "><Value>" + temp_map['test-id'] + "</Value></Field><Field Name=" + chr(34) + "subtype-id" + chr(
            34) + "><Value>hp.qc.run.MANUAL</Value></Field></Fields></Entity>"
    bulk_operation_post("runs", run_instance_post + "</Entities>", True, "POST")

    # ("*************\tRUNS\t*********************")

    payload = {"query": "{cycle-id['" + test_set_id + "']}", "fields": "id,test-id", "page-size": 5000}
    response = requests.get(almURL + midPoint + "/runs", params=payload, headers=headers, cookies=cookies)
    obj = json.loads(response.text)

    run_ids = []
    run_instance_put = "<Entities>"
    for entity in obj["entities"]:
        if len(entity["Fields"]) != 1:
            temp_map = create_key_value(entity["Fields"])
            parser_temp_dic[int(temp_map['test-id'])]['run-id'] = temp_map['id']
            run_ids.append(temp_map['id'])
            status = parser_temp_dic[int(temp_map['test-id'])]['final-status']
            run_instance_put = run_instance_put + "<Entity Type=" + chr(34) + "run" + chr(
                34) + "><Fields><Field Name=" + chr(
                34) + "id" + chr(34) + "><Value>" + str(temp_map['id']) + "</Value></Field><Field Name=" + chr(
                34) + "testcycl-id" + chr(34) + "><Value>" + str(
                parser_temp_dic[int(temp_map['test-id'])]['testcycl-id']) + "</Value></Field><Field Name=" + chr(
                34) + "status" + chr(
                34) + "><Value>" + status + "</Value></Field></Fields></Entity>"

    bulk_operation_post("runs", run_instance_put + "</Entities>", True, "PUT")

    # Upload result file
    payload = open("./screenshots/combined_result.html", 'rb')
    headers['Content-Type'] = "application/octet-stream"
    headers['slug'] = "protractor-test-results.html"
    response = requests.post(almURL + midPoint + "/" + "test-sets/" + str(test_set_id) + "/attachments/",
                             cookies=cookies, headers=headers,
                             data=payload)
    return


'''
Function    :   find_test_set_folder
Description :   This sends a couple of http request and authenticate the user
Parameters  :   1 Parameter
                test_set_path    -   ALM test set path
'''


def find_test_set_folder(test_set_path):
    json_str = json.loads(find_folder_id(test_set_path.split("\\"), "test-set-folders", 0, "id"))

    if 'entities' in json_str:
        return create_key_value(json_str['entities'][0]['Fields'])['id']
    else:
        return create_key_value(json_str['Fields'])['id']


'''
Function    :   find_test_set
Description :   This sends a couple of http request and authenticate the user
Parameters  :   1 Parameter
                test_set_path    -   ALM test set path
'''


def find_test_set(test_set_folder_id, strAPI):
    payload = {"query": "{name['" + testSetName + "'];parent-id[" + str(test_set_folder_id) + "]}", "fields": "id"}
    response = requests.get(almURL + midPoint + "/" + strAPI, params=payload, headers=headers, cookies=cookies)
    obj = json.loads(response.text)
    parentID = ""
    if obj["TotalResults"] >= 1:
        parentID = get_field_value(obj['entities'][0]['Fields'], "id")
        # print("test set id of " + testSetName + " is " + str(parentID))
    else:
        # print("Folder " + testSetName + " does not exists")
        data = "<Entity Type=" + chr(34) + strAPI[0:len(strAPI) - 1] + chr(34) + "><Fields><Field Name=" + chr(
            34) + "name" + chr(
            34) + "><Value>" + testSetName + "</Value></Field><Field Name=" + chr(34) + "parent-id" + chr(
            34) + "><Value>" + str(test_set_folder_id) + "</Value></Field><Field Name=" + chr(34) + "subtype-id" + chr(
            34) + "><Value>hp.qc.test-set.default</Value></Field> </Fields> </Entity>"
        response = requests.post(almURL + midPoint + "/" + strAPI, data=data, headers=headers, cookies=cookies)
        obj = json.loads(response.text)
        if response.status_code == 200 | response.status_code == 201:
            parentID = get_field_value(obj['Fields'], "id")
            # print("test set id of " + testSetName + " is " + str(parentID))

    return parentID


'''
Function    :   find_test_plan_folder
Description :   This sends a couple of http request and authenticate the user
Parameters  :   1 Parameter
                test_set_path    -   ALM test set path
'''


def find_test_plan_folder(test_plan_path):
    json_str = json.loads(find_folder_id(test_plan_path.split("\\"), "test-folders", 2, "id,hierarchical-path"))
    if 'entities' in json_str:
        return create_key_value(json_str['entities'][0]['Fields'])
    else:
        return create_key_value(json_str['Fields'])


'''
Function    :   find_folder_id
Description :   This sends a couple of http request and authenticate the user
Parameters  :   1 Parameter
                test_set_path    -   ALM test set path
'''


def find_folder_id(arrFolder, strAPI, parentID, fields):
    response = ""
    for folderName in arrFolder:
        payload = {"query": "{name['" + folderName + "'];parent-id[" + str(parentID) + "]}", "fields": fields}
        response = requests.get(almURL + midPoint + "/" + strAPI, params=payload, headers=headers, cookies=cookies)
        obj = json.loads(response.text)
        if obj["TotalResults"] >= 1:
            parentID = get_field_value(obj['entities'][0]['Fields'], "id")
            # print("folder id of " + folderName + " is " + str(parentID))
        else:
            # print("Folder " + folderName + " does not exists")
            data = "<Entity Type=" + chr(34) + strAPI[0:len(strAPI) - 1] + chr(34) + "><Fields><Field Name=" + chr(
                34) + "name" + chr(
                34) + "><Value>" + folderName + "</Value></Field><Field Name=" + chr(34) + "parent-id" + chr(
                34) + "><Value>" + str(parentID) + "</Value></Field></Fields> </Entity>"
            response = requests.post(almURL + midPoint + "/" + strAPI, data=data, headers=headers, cookies=cookies)
            obj = json.loads(response.text)
            if response.status_code == 200 | response.status_code == 201:
                parentID = get_field_value(obj['Fields'], "id")
                # print("folder id of " + folderName + " is " + str(parentID))
    return response.text


'''
Function    :   get_field_value
Description :   Find the value of matching json key
Parameters  :   2 Parameters
                obj         -   JSON object
                field_name  -   JSON KEY
'''


def get_field_value(obj, field_name):
    for field in obj:
        if field['Name'] == field_name:
            return field['values'][0]['value']


'''
Function    :   findTestCase
Description :   Check if given test case exists, if not create one
Parameters  :   3 parameters
                str_api          -   End point name
                str_test_name     -   Name of the test case
                parent_id        -   Test Plan folder id
'''


def test_exists(str_test_name, obj_json):
    str_exists = ''
    for test in obj_json['entities']:
        almtestname = re.sub('[^A-Za-z0-9\s_]+', '', test['Fields'][1]['values'][0]['value'].replace("_", " ")).strip()
        if almtestname == str_test_name:
            return create_key_value(test['Fields'])
    return str_exists


'''
Function    :   Post Test Case / Test Instance
Description :   Generic function to post multiple entities. Make sure to build the data in correct format
Parameters  :   3 parameters
                str_api          -   End point name
                data            -   Actual data to post
                bulk_operation   -   True or False
'''


def bulk_operation_post(str_api, data, bulk_operation, request_type):
    response = ""
    try:
        if bulk_operation:
            headers['Content-Type'] = "application/xml;type = collection"
        if request_type == 'POST':
            response = requests.post(almURL + midPoint + "/" + str_api, data=data, headers=headers, cookies=cookies)
        elif request_type == 'PUT':
            response = requests.put(almURL + midPoint + "/" + str_api, data=data, headers=headers, cookies=cookies)
    finally:
        headers['Content-Type'] = "application/xml"
        if response.status_code == 200 | response.status_code == 201:
            return response.text
    return response


'''
Function    :   remove_special_char
Description :   Function to remove non-acceptable characters
Parameters  :   1 parameter
                str_input   -   input string
'''


def remove_special_char(str_input):
    return re.sub('[^A-Za-z0-9\s_-]+', '', str_input).strip()


'''
Function    :   create_key_value
Description :   Function to generate key-value pair from json
Parameters  :   1 parameter
                obj_json   -   JSON Object
'''


def create_key_value(obj_json):
    final_dic = {}
    for elem in obj_json:
        if len(elem['values']) >= 1:
            if 'value' in elem['values'][0]:
                final_dic[elem["Name"]] = elem["values"][0]['value']
    return final_dic


'''

'''
'''
    CORE FUNCTION
'''


def update_results_alm():
    try:
        alm_login()
        headers['Accept'] = "application/json"
        headers['Content-Type'] = "application/xml"
        parse_result()
    finally:
        alm_logout()


if len(sys.argv) - 1 != 4:
    print('Build number is required.You have passed :', str(sys.argv), 'arguments.')
else:
    testSetName = sys.argv[1]
    testPlanPath = sys.argv[2]
    testSetPath = sys.argv[3]
    almProject = sys.argv[4]
    print(testSetName + "\n" + testPlanPath + "\n" + testSetPath + "\n" + almProject)
    midPoint += almProject
    update_results_alm()

Answer 2

您可以在此处找到代码 https://github.com/arunprabusamy/Python-Libraries/blob/main/alm_RestAPI/almLib.py

只需要三个参数 - 测试集 ID（周期 ID）、ALM 测试 ID、执行状态。使用 JSON 有效负载且简单。