zoukankan      html  css  js  c++  java
  • 【转】运用钉钉机器人功能发送自动化结果报告

    背景:

    原有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 # 运行自动化用例
    复制代码

    (Tips:最后一行执行UI自动化用例的命令参数:retry 是用于用例失败后自动重试的,由于网络等原因导致UI自动化有几率加载不出元素(尽管有等待10s),导致用例失败。retry参数非robotframework自带,源码修改见另一篇博客:(【转载】扩展Robot Framework,实现失败用例自动再执行(失败重跑)

  • 相关阅读:
    docker log
    byte转String防止乱码
    SQL索引
    Redis 总结精讲
    如何保证消息队列是高可用的
    消息中间件(一)MQ详解及四大MQ比较
    @Bean和@Componet区别
    理解Spring的AOP和Ioc/DI就这么简单
    SpringBoot 基础
    《Linux 鸟哥私房菜》 第6章 Linux的文件权限与目录配置
  • 原文地址:https://www.cnblogs.com/Byronlyu/p/7216061.html
Copyright © 2011-2022 走看看