運用釘釘機器人功能發送自動化結果報告


轉載自博友:濤哥愛吃面  http://www.cnblogs.com/hito/p/6561300.html

背景:

原有jenkins自動化結果通知通過郵件發送,容易埋沒在郵箱的大量報警郵件中,結合作為工作聊天工具釘釘機器人,可以實現自動化開始及結果的消息、報告通知

實現:

1.機器人的創建

(具體步驟及相關協議:https://open-doc.dingtalk.com/docs/doc.htm?spm=a219a.7386797.0.0.TKQBVI&treeId=257&articleId=105735&docType=1

拿到的webhook地址格式如下:

https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx

簡略說下機器人群通知調用:請求地址即上方的webhook地址,POST,請求體支持三種類型text,link,markdown,我用到的只有text(自動化執行開始通知)和link(結果報告輸出)

2.腳本編寫

需求:

自動化開始執行,釘釘群接收自動化開始執行的通知;自動化執行結束后,釘釘群接收到自動化的執行結果報告,能看到執行結果(成功或失敗),包版本號,點擊可打開RF的report.html。

設計:

1.腳本傳參類型,以判斷開始或結束;

2.自動化執行前,調用腳本發送開始執的方法(簡單發送text消息);

3.結果報告輸出,while循環判斷自動化的執行結果log文件中是否有結果,拿到結果后即發送結果報告文件

源碼:

復制代碼
# coding:utf-8
import urllib2
import json
import sys
import time
import os


class dingHook(object):

    def __init__(self, pro):
        self.user = "tech"
        self.pro = pro
        self.url = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx"

    def request(self, url, method, data=None, head={}):
        request = urllib2.Request(url=url, headers=head)
        request.get_method = lambda: method
        httpRes = urllib2.urlopen(request, data)
        content = httpRes.read()
        httpRes.close()
        return content

    def post_start(self):
        data = {}
        data["msgtype"] = "text"
        data["text"] = {}
        data["text"]["content"] = "FROM 67 jenkins:開始執行【%s】自動化......" % self.pro
        data = json.dumps(data)
        head = {"Content-Type": "application/json"}
        content = self.request(self.url, "POST", data, head)
        return content

    def post_link(self, text, title, picurl="", messageurl=""):
        data = {}
        data["msgtype"] = "link"
        data["link"] = {}
        data["link"]["text"] = text
        data["link"]["title"] = title
        data["link"]["picUrl"] = picurl
        data["link"]["messageUrl"] = messageurl
        data = json.dumps(data)
        head = {"Content-Type": "application/json"}
        content = self.request(self.url, "POST", data, head)
        print content
        return content

    def post_result(self):
        i = 0
        result = "FAILURE"
        # 獲取當前jenkins構建版本號
        get_buildnum = "cat /home/%s/.jenkins/jobs/%s/nextBuildNumber" % (
            self.user, self.pro)
        output = os.popen(get_buildnum)
        buildNumber = str(int(output.read().strip()) - 1)
        # 獲取jenkins自動化執行結果,判斷是否執行結束
        get_buildresult = "grep Finished /home/%s/.jenkins/jobs/%s/builds/%s/log" % (
            self.user, self.pro, buildNumber)
        cat_version = "cat /home/%s/local/testcase/%s_report/version" % (self.user,self.pro)
        # 獲取結果報告中的返回結果
        get_RF_result = "grep 'All Tests' /home/%s/.jenkins/jobs/%s/builds/%s/robot-plugin/output.xml" % (
            self.user, self.pro, buildNumber)
        # jenkins自動化結果報告地址
        messageurl = "http://10.16.29.167:18080/jenkins/job/%s/%s/robot/report/report.html" % (
            self.pro, buildNumber)
        # 釘釘結果報告消息圖片
        picurl = "xxxxx.jpg"
        # 釘釘結果報告標題
        title = "【%s】自動化執行結果" % self.pro
        while i <= 1000:
            i = i + 1
            try:
                output_result = os.popen(get_buildresult)
                build_result = output_result.read()
                print build_result
                if "No such file or directory" in build_result:
                    time.sleep(10)
                    continue
                if "Finished" in build_result:
                    output_rf_result = os.popen(get_RF_result)
                    rf_result = output_rf_result.read().strip()
                    case_count_list = rf_result.split('"')
                    fail_case = int(case_count_list[1])
                    suc_case = int(case_count_list[3])
                    if fail_case == 0:
                        result = "SUCCESS-成功:%d-失敗:%d" % (suc_case,fail_case)
                    else:
                        result = "FAILURE-成功:%d-失敗:%d" % (suc_case,fail_case)
                    output_version = os.popen(cat_version)
                    version = output_version.read()
                    if "zip" in version:
                        result = "【"+result+"】【"+version+"】"
                    content = self.post_link(
                        text=result, title=title, messageurl=messageurl,picurl=picurl)
                    return content
                time.sleep(10)
            except Exception, e:
                print e
                time.sleep(10)
                continue

    def ding_hook(self, itype="start"):
        print itype
        if itype == "start":
            self.post_start()
        elif itype == "end":
            self.post_result()

if __name__ == '__main__':
    dh = dingHook(sys.argv[1])
    print dh.ding_hook(sys.argv[2])
復制代碼

 

3.jenkins構建配置

配置jenkins項目Execute shell

復制代碼
source ~/.bash_profile # 加載當前用戶環境配置文件
cd /home/test/local/testcase # 進入自動化用例目錄
python dingHook.py UI start # 發送自動化開始執行消息
BUILD_ID=dontKillMe # 聲明不殺死啟動進程
python dingHook.py UI end & # 后台啟動監控執行結果進程,拿到自動化執行結果后發送
pybot --retry 3 -L debug -d /home/test/local/testcase/UI_report/ UI # 運行自動化用例
復制代碼


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM