python利用jenkins模块操作jenkins
安装python-jenkins
可以使用pip命令来安装python-jenkins模块：
pip install python-jenkins
操作jenkins
接下来就是连接和操作jenkins，写了个class，直接上代码
class Jenkins():
def __init__(self, url, username, password):
# jenkins服务连接URL
self.url = url
# jenkins登陆的用户名密码
self.username = username
self.password = password
# 连接jenkins
self.server = jenkins.Jenkins(self.url, self.username, self.password)
def create_job(self, job_name, credentials_id, config_xml, command, node_name):
"""
创建job任务
:param job_name: 任务名称
:param config_xml: job任务参数设置
:param command: job任务中shell脚本
:param node_name: 执行任务的几点名称
:return:
"""
# 这步就是将job任务中的shell脚本那部分添加到config_xml中，在config_xml中需要提前设置好{command}才可以这样做哈
config_xml = config_xml.replace("{command}", command).rstrip()
config_xml = config_xml.decode('utf-8')
try:
self.server.create_job(job_name, config_xml)
except Exception as e:
err_one = "job[{}] already exists".format(job_name)
if str(e) != err_one:
return False, e
return True, ""
def get_job_xml(self, job_name):
"""
获取job xml配置信息
:param job_name:
:return:
"""
return self.server.get_job_config(job_name)
def get_node_config(self, node_name):
"""
获取节点配置信息
:param node_name:
:return:
"""
return self.server.get_node_config(node_name)
def create_node(self, node_name, credentials_id, sshhost, remotefs, labels=None, numExecutors=1, nodeDescription=None):
"""
创建节点
:param node_name: 节点名称
:param credentials_id: 在Jenkins中配置的凭据ID
:param sshhost: 主机
:param remoteFS: 远程工作目录
:param labels: 标签
:param numExecutors: Number of executors(Jenkins 可以在此节点上执行并发构建的最大数目)
:param nodeDescription: 描述
:return:
"""
try:
self.server.create_node(
node_name,
numExecutors=numExecutors,
nodeDescription=nodeDescription,
remoteFS=remotefs,
labels=labels,
launcher= "hudson.plugins.sshslaves.SSHLauncher",
launcher_params={
'stapler-class': 'hudson.slaves.SSHLauncher',
'host': sshhost,
'port': 22,
'credentialsId': credentials_id,
'jnlp': {
'workDirSettings': {
'innerDir': 'remoting',
'failIfWorkDirIsCaseInsensitive': False,
'failIfWorkDirExists': False,
'createWorkDir': True,
'disabled': False,
'deleteWorkDir': False,
'stickyReferenceFiles': ''
}
}
}
)
except Exception as e:
err_one = "node[{}] already exists".format(node_name)
if str(e) != err_one:
return False, e
return True, ""
def get_build_info(self, job_name, last_build_number=0):
"""
获取job 信息
:param job_name: jenkins任务名称
:param last_build_number: jenkins build number
:return: dict
"""
return self.server.get_job_info(job_name, last_build_number)
def get_job_build_number(self, job_name):
"""
获取job number
:param job_name: jenkins任务名称
:return: int
"""
try:
build_number = self.get_build_info(job_name)["lastBuild"]["number"]
except:
# 取不出来值是因为还没有执行过任务，所以返回默认值0
build_number = 0
return build_number
def build_job(self, job_name, params):
"""
执行job
:param params:
:return:
"""
self.server.build_job(job_name, params)
hope_build_number = -1
# 获取创建的任务number
while True:
last_build_number = self.get_job_build_number(job_name)
if last_build_number != hope_build_number:
hope_build_number = last_build_number + 1
else:
break
time.sleep(1)
# 获取自动化执行结果
while True:
result = self.get_build_info_status(job_name, hope_build_number)
if result is not None:
break
time.sleep(1)
return hope_build_number, result
def get_build_console_ouput(self, job_name, last_build_number):
"""
获取项目控制台日志
:param job_name: jenkins任务名称
:param last_build_number: jenkins build number
:return: str
"""
return self.server.get_build_console_output(name=job_name, number=last_build_number)
def get_build_info_status(self, job_name, last_build_number):
"""
获取任务状态
:param job_name: jenkins任务名称
:param last_build_number: jenkins build number
:return:str 状态有4种：SUCCESS| FAILURE| ABORTED| pending
"""
builds = self.get_build_info(job_name, last_build_number)["builds"]
for b in builds:
if b["number"] == last_build_number:
result = b["result"]
return result
def get_build_ouput_url(self, job_name, job_number):
"""
获取日志输出链接地址
:param job_name: jenkins任务名称
:param job_number: jenkins build number
:return: str
"""
return "{jenkins_url}/job/{job_name}/{job_number}/console".format(jenkins_url=self.url, job_name=job_name, job_number=job_number)
def get_build_report_url(self, job_name, job_number, report_name):
"""
获取自动化报告结果链接地址
:param job_name: jenkins任务名称
:param job_number: jenkins build number
:param report_name: jenkins测试报告html文件名称
:return: str
"""
return "{jenkins_url}/job/{job_name}/{job_number}/artifact/{report_name}".format(jenkins_url=self.url, job_name=job_name, job_number=job_number, report_name=report_name)
暂时写了些常用的。其实还有删除、修改等操作，可以看下底层代码中有哪些方法可以用。可玩性还是很高的。就不详细写了。
就这样，下课！！！