Python3 + Jenkins API实践

背景

在使用接口自动化实践过程中，需要通过 Python3 + Jenkins 自动触发接口自动化脚本的JOB，过程中还涉及到通过 Parameter 参数去动态构建 JOB，在此总结下使用过程中遇到的问题以及方法。

使用方法（官网）

一、初始化 Jenkins 实例

1 2	# 此处的 password 可以通过使用 API Token 来替代, 避免密码泄漏。 server = jenkins.Jenkins(url, username=username, password=password)

在Jenkins上生成Token具体的操作步骤，按照下图中的数字标记进行相应的操作。

二、获取最后一次JOB的构建号&结果&状态

@classmethod
def get_job_status(cls, job_name, server):
  # 最近一次JOB的 构建号
  build_num = server.get_job_info(job_name)['lastBuild']['number']
  print("build num %s" % build_num)
  
  # 最近一次JOB的 结果 SUCCESS/FAILURE等 （JOB执行成功还是失败）
  special_build_id_info = server.get_build_info(job_name, build_num)
  print("指定Job result: %s" % special_build_id_info['result'])
  
  # 最后一次JOB的 状态 True/False （是否执行完成）
  status = server.get_build_info(job_name, build_num)['building']
  print("指定Job status: %s" % status)
  return status, build_num

三、构建执行JOB

def start_job(cls, job_name, server=None, params=None):
  server = server or cls.server
  params = params or dict()
  status, build_num = cls.get_job_status(job_name, server)
  if not status:
    server.build_job(job_name, parameters=params)
    return "开始执行"
  else:
    raise ValueError("已存在正在执行中的任务, 请勿重复执行")

遇到的问题

问题一：一秒内，同一个人快速调用多次 `start_job()` , 预期应该是仅仅第一次调用时返回 “开始执行”, 实际结果是：多次调用仍然会返回 “开始执行”。

原因：在调用 cls.get_job_status(job_name, server) 的时候，从第一次调用成功后（假设获取到的最后一次构建号为4），Jenkins平台上就开始准备构建该JOB，此时任务会放到 Build Queue 中进行一系列的准备，如果此时再去调用 cls.get_job_status() 时候，JOB还在 Build Queue 中等待的话，则获取到的构建号仍然为4, 并且 status 为 False（如下结果的输出），所以还是返回了 “开始执行”。

# 仅当 build num 变成5后，才触发了异常。
==========================
build num 4
指定Job result: SUCCESS
指定Job status: False
==========================
build num 4
指定Job result: SUCCESS
指定Job status: False
==========================
build num 5
指定Job result: None
指定Job status: True

解决方案一：通过将执行记录存储到数据库中，每次执行之前，先去查询下是否存在下个构建号的记录。（如果有需要还可以通过行锁来控制此次的插入操作，这样保证高并发下也能符合预期。）

解决方案二（推荐）：通过 Jenkins 提供的接口 get_queue_info() 方法获取当前队列中的JOB列表，如果JOB已经在队列中，则直接返回异常。

# 具体实现代码如下：job_name 是需要判断是否处于构建中的JOB名称， 再配合JOB的 status 来实现判断JOB任务是否已经开始构建。
queue_info = server.get_queue_info()
for single_queue_info in queue_info:
  if single_queue_info.get("task", dict()).get("name") == job_name:
    return False

return True

问题二： Jenkins 构建JOB时候，需要动态传入参数到 JOB中。

解决方案：先通过Jenkins安装对应的插件：

Build With Parameters 输入框式的参数
Persistent Parameter 下拉框式的参数

Jenkins中的使用方法如下截图，通过点击 Add Parameter 新增变量来构建JOB。

通过Python代码触发构建代码如下：

# 以字典形式传递参数进去即可，如上图的配置，代码中的参数格式：
#    params: {
#        "operator": "auto_job"
#    }

server.build_job(job_name, parameters=params)

总结

很多时候，在解决问题的时候（特别是对于问题的相关知识不是太了解的情况下），如果只是紧急修复问题的话，思考出来的方案可能都不是一个比较合适的方案。所以在空闲时候可以再回过来看看问题，说不定能够想到更多的解决方案。

参考链接

JENKINS-API 详解

背景

使用方法 （官网）