Jenkins執行python腳本

構建選擇Excute Windows batch command

 下面是python腳本，注意字符集 GBK

 

runtest.py

#-*-coding:GBK -*- 
import sys
import time
import pymysql
import requests


# print(sys.argv[1])

ids_out = []
# 用","分割，得到列表id_list
id_list = sys.argv[1].split(",")
for id_str in id_list:    
    # 用~分割，1001這種會變成['1001']，其他會變成['1003', '1005']，因為range不包含后面的值，所以要加1，取[1003,10006)
    for i in range(int(id_str.split("~")[0]), int(id_str.split("~")[-1]) + 1):
        ids_out.append(i)
        
print("需要執行的用例集：", ids_out)

json = {"ids": ids_out}

r = requests.post('http://10.138.60.185:9999/step_allexcute_data1/', json=json)

print(r.status_code, r.text)


# 遍歷列表，查詢出用例名和執行時間在10分鍾內的結果
print("\n")
print("序號   用例編號    用例名稱            執行結果")
suc_num = 0
sum_num = 1
for id in ids_out:
    try:
        sql = "select step_name,api_result,actual_params,expect_params from vcredit_resultdetail where step_id = %s and create_time > DATE_SUB(NOW(), INTERVAL 10 MINUTE)" % id
        # 執行第一次查詢
        conn = pymysql.connect(host="10.138.60.185", user="root", password="fyw554193", database="automate")
        cursor = conn.cursor()
        cursor.execute(sql)
        data = cursor.fetchone()
        cursor.close()
        conn.close()
        # 如果查詢不到結果，則10秒后再查，執行100次，打印"id,用例名：用例正在執行，10秒后繼續查詢"
        i = 0
        while (not data) and (i < 100):
            # print("%s: 未查到執行結果，10秒后繼續查詢。。。" % id)
            time.sleep(10)
            conn = pymysql.connect(host="10.138.60.185", user="root", password="fyw554193", database="automate")
            cursor = conn.cursor()
            cursor.execute(sql)
            data = cursor.fetchone()
            i += 1
        # 查詢到結果，則正常打印
        if data[1] == 1:
            suc_num += 1
            print("%s      %s        %s            用例執行通過" % (sum_num,id, data[0]))
        else:
            print("%s      %s        %s            用例執行未通過,實際結果為%s,預期結果為%s" % (sum_num,id, data[0], data[1], data[2]))
    # 失敗則打印異常
    except Exception:
        print("%s      %s                      接口發生異常" % (sum_num,id))
    sum_num += 1
print("\n")
print("總計運行%s個測試用例，成功%s個，失敗%s個" %(len(ids_out), suc_num, len(ids_out)-suc_num))

 

record.py

#-*-coding:GBK -*- 
import sys
import pymysql
import requests
import xml.dom.minidom
from requests.auth import HTTPBasicAuth

'''通過Jenkins API獲取項目構建結果'''
auth = HTTPBasicAuth('admin', 'admin')
url = "http://10.138.60.82:8080/api/python?pretty=true"
r = requests.get(url=url, auth=auth)
print(r.text)

dic = eval(r.text)

for i in dic["jobs"]:
    if i["name"] == 'build':
        bu_color = i["color"]
    elif i["name"] == 'scanning':
        sc_color = i["color"]

print(bu_color, sc_color)

'''通過pom.xml獲取版本信息'''
pom_path = sys.argv[1]
print(pom_path)
dom = xml.dom.minidom.parse(pom_path)
rootdata = dom.documentElement
itemlist = rootdata.getElementsByTagName('version')
version = itemlist[0].firstChild.data
print(version)

'''把結果轉化並插入數據庫'''
bu_status = 1 if bu_color == "blue" else 0
sc_status = 1 if sc_color == "blue" else 0

sql = "insert into vcredit_buildrecord(requireName,ProjectName,build,isCis,create_time) VALUE ('%s','%s',%s,%s,NOW())" \
      % (version, "ProjectName", bu_status, sc_status)

conn = pymysql.connect(host='10.138.30.104', user='root', password='080305', database='testquality', charset='UTF8')
cursor = conn.cursor()
cursor.execute(sql)
cursor.close()
conn.commit()
conn.close()

 

codeLine.py

#-*-coding:GBK -*- 
import re
import sys
import xml.dom.minidom
import pymysql

with open("D:\\jenkins\\jobs\\git\\gitlog.txt") as f:
    txt = f.read()

print(txt)

p1 = r"(?<= \| )\d+?(?= \+\+)"
pattern1 = re.compile(p1)
num = int(pattern1.findall(txt)[0])

print(num)

'''通過pom.xml獲取版本信息'''
pom_path = sys.argv[1]
print(pom_path)
dom = xml.dom.minidom.parse(pom_path)
rootdata = dom.documentElement
itemlist = rootdata.getElementsByTagName('version')
version = itemlist[0].firstChild.data
print(version)

'''更新數據庫'''
sql = "UPDATE `vcredit_requirequality` SET `codeLine`=%s WHERE `requireName`='%s'" % (num, version)

print(sql)

conn = pymysql.connect(host='10.138.30.104', user='root', password='080305', database='testquality', charset='UTF8')
cursor = conn.cursor()
cursor.execute(sql)
cursor.close()
conn.commit()
conn.close()