HP ALM 12 rest api help in python
For HP ALM 12.20.3264.
Using python 2.7.9 - wish to complete an automation task. For which - need to achieve below tasks:
Connect HP ALM Success . Please find below success message.
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'})Get test case name information from Testing -> TestLab Failed with below error:
(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'})Update test case state- pass/Fail ==> Yet to Implement
Could you please help understanding the restful api’s exposed by HP ALM 12.
Below is my sample Python script:
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
Below code covers most of your requirement. In short, this code takes the output of protractor test from Jenkins and create test set (If not exist) in HP ALM and update the test status and attaches the report.
To know the list of end points, enter the following in your favourite browser
<<ALM_SERVER>>/qcbin/rest/resouce-list
To understand the limitations and the schema details, GoTo Help in HP ALM.
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()