安装DingtalkChatbot:
pip install DingtalkChatbot
钉钉配置机器人:
智能群助手
给机器人取一个名字~
webhook需要保存好,需要传入钉钉DingtalkChatbot()方法内
自定义关键词,这里需要注意一下,自定义关键词内的文案在msg内一定要有,否则无法触发机器人报警
钉钉模块内有很多种类型的消息场景,我的目的是接口自动化巡检,所以只需要消息场景就可以,其他的场景感兴趣的可以查一下场景字段参数~
把自己想要的日志内容写在msg内,并且单独封装一个方法获取需要的消息参数内容
# WebHook地址from dingtalkchatbot.chatbot import DingtalkChatbotclass DingDing: def __init__(self): self.yhn = "9z7_66n6cmbqy" self.webhook = "https://oapi.dingtalk.com/robot/send?access_token=6e3cc2451e89506e630b5be62de31ef28501a32bd2ef1181aaaa968527624b9b" def run_error(self, *args): xiaoding = DingtalkChatbot(self.webhook) xiaoding.send_text( msg=f'监测预警,测试用例数量:{args[0]},用例异常数量: {args[1]}' f'n测试用例名称:{args[2]}n日志内容:{args[3]}' f'n链接地址:{args[4]} 请注意及时处理!', at_dingtalk_ids=[self.yhn], is_at_all=False ) def run_normal(self, *args): xiaoding = DingtalkChatbot(self.webhook) xiaoding.send_text( msg=f'监测预警,接口自动化case执行完毕n测试用例数量:{args[0]},用例异常数量: {args[1]}' f'n测试报告地址:{args[2]} 请关注~', at_dingtalk_ids=[self.yhn], is_at_all=False ) def run_trigger(self, list_dict, if_print=True): """ 触发钉钉机器人 :param list_dict: case_list列表 :param if_print: 是否需要过滤测试通过的case True为是不会触发机器人 False不会过滤 :return: """ for dict_list in list_dict: kwargs = dict_list errorCount = kwargs["error_count"] testRun = kwargs["testRun"] case_name = kwargs["case_name"] error = kwargs["error"] fail_count = kwargs["failure"] count = errorCount + fail_count # todo 缺少报告地址获取 report_url = "完善中,后期替换~" if count != 0: if error != "": self.run_error(testRun, count, case_name, error, report_url) elif not if_print: error = "当前case执行通过" self.run_error(testRun, errorCount, case_name, error, report_url) else: return self.run_normal(testRun, errorCount, report_url)
书接上回,上一章写的是html报告与unittest执行方法的封装,我的目的是获取到一些报告内的用例名称/错误参数/执行用例数量/与用例异常数量,通过机器人发送到群内,所以需要取出数据,然后传给钉钉机器人方法
下面是接着上一章的两个方法获取下来result参数内容,然后通过字典的形式储存,因为是多条case,所以每执行一条需要在list内储存一份,这样每条case的数据就都有了,然后再通过封装好的
DingDing().run_trigger(result_list_dict, True)
钉钉的方法把参数带入,这里需要注意的一点就是,error异常与fail失败是分开的数量,所以要分别储存,然后相加数量来判断是否有异常case~
def run_dingDing(way, run_list, test_py=None): result_list_dict = [] global result if way == 1: result = run_py(run_list) elif way == 2: result = run_case(test_py, run_list) error_count = result.error_count testRun = result.testsRun failure = result.failure_count result_list = list(result.result) for rel in result_list: dict_result = { "error_count": error_count, "testRun": testRun, "failure": failure, "case_doc": rel[1]._testMethodDoc, "case_name": rel[1]._testMethodName, "error": rel[3] } result_list_dict.append(dict_result) # print(f"dict_result{dict_result}") # print(f"result_list_dict{result_list_dict}") DingDing().run_trigger(result_list_dict, True) return result_list_dictif __name__ == '__main__': file_list = ["test_shiyan.py", "MeetSpringFestival.py"] # run_py(file_list) case_list = ["test_case_01", "test_case_02"] # run_case(test_sample, case_list) dingDing_list = [2, case_list, test_sample] run_dingDing(*dingDing_list)
下期会加入定时任务,定时执行接口自动化巡检触发钉钉~