GitLab CI 实现自动测试、合并分支并用钉钉发送消息

发表于 2023-03-05 阅读次数：本文字数： 801 阅读时长 ≈ 3 分钟

CI 任务触发需求

普通提交代码时不触发
当指定commit关键字触发
发起了 Merge Requests则自动触发
测试任务成功后自动合并分支

.gitlab-ci.yml

为了不在代码里体现秘钥信息，可在项目 CI/CD 设置里 variables 添加 token、secret，触发 CI 任务时可自动获取此变量

#每个 Job 任务结束后通过钉钉发送任务结果信息
after_script:
  - python dingtalk.py $TOKEN $SECRET $CI_JOB_STATUS $CI_JOB_URL $CI_JOB_NAME $CI_PROJECT_TITLE $GITLAB_USER_NAME
#step 1
build:
  tags:
    # 指定运行 Runner 标签
    - dev
  rules:
    # 当 commit 信息为 `run ci`开头或发起了 `Merge Requests`时触发
    - if: $CI_COMMIT_MESSAGE =~ /^run ci/ || $CI_PIPELINE_SOURCE == "merge_request_event"
  script:
    - do build job
#step 2
test:
  tags:
    # 指定运行 Runner 标签
    - dev
  rules:
    - if: $CI_COMMIT_MESSAGE =~ /^run ci/ || $CI_PIPELINE_SOURCE == "merge_request_event"
  #设置依赖关系
  needs:
    - Solve
  script:
    - do test job
#step 3: auto merge
Merge_Branch:
  tags:
    - dev
  rules:
    # 仅在有 MR 时运行
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
  #设置依赖关系
  needs:
    - Compare
  script:
    # 谁发起了MR就合并谁的MR请求
    - python /opt/gitlab-runner/gitlab-api.py $CI_PROJECT_ID $GITLAB_USER_LOGIN

dingtalk.py

通过dingtalkchatbot发送钉钉消息,先在钉钉群里添加一个自定义机器人，记录下URL token 和 secret 秘钥

#!/usr/bin/env python
# encoding=utf-8
__Author__ = "Actpi"

import sys
import argparse
from dingtalkchatbot.chatbot import DingtalkChatbot, ActionCard, CardItem


def send_notify(webhook,secret,message):
    # 初始化机器人小丁
    bot = DingtalkChatbot(webhook,secret,pc_slide=False)  # 方式一：通常初始化方式
    bot.send_link(title=message[0],text=message[1],message_url=message[2],pic_url=message[3])

def set_actioncard(args):
    # 失败时：任意任务都发通知
    if args.CI_JOB_STATUS == "failed":
        V_IMG = "https://s2.loli.net/2022/09/28/WpklwGHgMZuoCXU.png"
        V_STATUS = "执行失败 "
    # 成功时：仅 merge 任务发通知
    elif args.CI_JOB_STATUS == "success" and args.CI_JOB_NAME == "Merge_Branch":
        V_IMG = "https://s2.loli.net/2022/09/28/R8QKZqrXHUedmB5.png"
        V_STATUS = "执行成功 "
    else:
        sys.exit(10)
    V_TITLE = "任务 %s %s" %(args.CI_JOB_NAME,V_STATUS)
    V_TEXT = "项目:  %s  \n作者: %s" %(args.CI_PROJECT_TITLE,args.GITLAB_USER_NAME)
    message_url = args.CI_JOB_URL
    pic_url = V_IMG

    return [V_TITLE,V_TEXT,message_url,pic_url]

def get_args():
    parser = argparse.ArgumentParser(description='钉钉发送job结果消息、需要传入参数：')
    parser.add_argument('TOKEN', help='钉钉机器人 TOKEN')
    parser.add_argument('SECRET', help='钉钉机器人 SECRET')
    parser.add_argument('CI_JOB_STATUS', help='变量 CI_JOB_STATUS')
    parser.add_argument('CI_JOB_URL', help='变量 CI_JOB_URL')
    parser.add_argument('CI_JOB_NAME', help='变量 CI_JOB_NAME')
    parser.add_argument('CI_PROJECT_TITLE', help='变量 CI_PROJECT_TITLE')
    parser.add_argument('GITLAB_USER_NAME', help='变量 GITLAB_USER_NAME')
    return parser.parse_args()
    
def main():
    args = get_args()
    # 钉钉机器人 webhook 地址
    url = "https://oapi.dingtalk.com/robot/send?access_token="
    webhook = url + args.TOKEN
    #secret = 'SEC11b9...这里填写自己的加密设置密钥'
    secret = args.SECRET
    #设置发送的消息内容
    message = set_actioncard(args)
    send_notify(webhook,secret,message)
if __name__ == '__main__':
    main()

gitlab-api.py

用 python-gitlab api 封装完成分支合并的操作

#!/usr/bin/env python
# -*- coding: utf-8 -*-
__Author__ = "Actpi"


import sys
import gitlab
# doc: https://python-gitlab.readthedocs.io/en/stable/api-usage.html
url = "https://gitlab.com"
# personal access token
access_token = "xxxxxxxxxxxxxxxx"

def main():
    gl = gitlab.Gitlab(url,access_token)
    # get the project with id
    project = gl.projects.get(CI_PROJECT_ID)
    # json格式化
    # print(project.to_json(indent=4))
    # 获取当前打开的MR信息
    mrs = project.mergerequests.list(state='opened')
    if mrs:
        for mr in mrs:
            #谁触发了流水线就合并谁的MR请求
            if mr.author['username'] == COMMIT_USER:
                if mr.has_conflicts:
                    print("有代码冲突、请先解决冲突！")
                    sys.exit(1)
                else:
                    print("title: %s\ndescription: %s\ndetails: " %(mr.title,mr.description) )
                    if mr.merge_status == "can_be_merged":
                        print("merge "+mr.source_branch+" to "+mr.target_branch+" by "+mr.author['username']+" at " + mr.created_at)
                        mr.merge()
                    elif mr.merge_status == "unchecked":
                        print("当前 MR 状态未被检查确认，需要审核者在 MR 页面点击 Approved 按钮确认！")
                        sys.exit(1)
                    else :
                        print("当前MR：%s 状态为: %s, 不能合并，请检查MR详情！" %(mr.title,mr.merge_status))
                        sys.exit(1)
    else:
        print("No Opened Merge Request.")
        sys.exit(1)
if __name__ == '__main__':
    if len(sys.argv) > 2:
        CI_PROJECT_ID = sys.argv[1]
        CI_COMMIT_AUTHOR = sys.argv[2]
        print("CI_TRIGGER_AUTHOR: %s" % CI_COMMIT_AUTHOR)
        COMMIT_USER = CI_COMMIT_AUTHOR.split()[0]
        main()
    else:
        print("Error:Please Input parameter: PROJECT_ID and CI_COMMIT_AUTHOR")
        sys.exit(1)