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("
")
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("
")
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()