From 34a6762438c84132572babcd20f2d58579c0a92c Mon Sep 17 00:00:00 2001 From: floraachy <1622042529@qq.com> Date: Wed, 17 May 2023 17:33:01 +0800 Subject: [PATCH] =?UTF-8?q?(2023-05-11)=20=E6=94=AF=E6=8C=81=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E5=AE=8C=E6=88=90=E5=90=8E=EF=BC=8C=E9=80=9A=E8=BF=87?= =?UTF-8?q?=E9=92=89=E9=92=89=E5=8F=91=E9=80=81pytest-html=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E7=BB=93=E6=9E=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- case_utils/send_result_handle.py | 101 ++++++++----- common_utils/dingding_handle.py | 252 +++++++++++++++++++++++++++++++ config/settings.py | 6 + 3 files changed, 322 insertions(+), 37 deletions(-) create mode 100644 common_utils/dingding_handle.py diff --git a/case_utils/send_result_handle.py b/case_utils/send_result_handle.py index 534a846..58cb0e1 100644 --- a/case_utils/send_result_handle.py +++ b/case_utils/send_result_handle.py @@ -4,11 +4,12 @@ # @File : send_result_handle.py # @Software: PyCharm # @Desc: +from loguru import logger from common_utils.yagmail_handle import YagEmailServe -from config.settings import email from config.global_vars import NotificationType -from config.settings import SEND_RESULT_TYPE +from config.settings import SEND_RESULT_TYPE, email, ding_talk from common_utils.bs4_handle import SoupAPI +from common_utils.dingding_handle import DingTalkBot def get_test_info_from_html_report(html_report_path): @@ -77,53 +78,79 @@ def get_test_info_from_html_report(html_report_path): return results +def send_email(user, pwd, host, subject, contents, to, attachments): + """ + 发送邮件 + """ + try: + yag = YagEmailServe(user=user, password=pwd, host=host) + info = { + "subject": subject, + "contents": contents, + "to": to, + "attachments": attachments + + } + yag.send_email(info) + except Exception as e: + logger.error(f"发送邮件通知异常， 错误信息：{e}") + + +def send_dingding(webhook_url, secret, title, text): + """ + 发送钉钉消息 + """ + try: + dingding = DingTalkBot(webhook_url=webhook_url, secret=secret) + res = dingding.send_markdown(title=title, text=text, is_at_all=True) + if res: + logger.info(f"发送钉钉通知成功~") + else: + logger.error(f"发送钉钉通知失败~") + except Exception as e: + logger.error(f"发送钉钉通知异常， 错误信息：{e}") + + def send_result(results, attachment_path=None): """ 根据用户配置，采取指定方式，发送测试结果 """ + # 通知内容 + subject = f"{results.get('project_name', None)} 接口自动化报告_{results.get('start_time', None)}" + content = f""" + 各位同事, 大家好: + +   自动化用例于{results.get('start_time', None)}开始运行，运行时长：{results.get('runs_time', None)}， 目前已执行完成。 + ----------------------------------------------------------------------------------------------------------- +   测试人：{results.get('tester', None)} / {results.get('dept', None)} +   测试平台：{results.get('platform', None)} / {results.get('python_version', None)} +   测试环境：{results.get('project_env', None)} + --------------------------------------------------------------------------------------------------------------- +     执行结果如下: +     用例运行总数: {results.get("total_cases", None)} 个 +     通过用例个数（passed）: {results.get("passed", None)} 个 +     失败用例个数（failed）: {results.get("failed", None)} 个 +     异常用例个数（error）: {results.get("error", None)} 个 +     跳过用例个数（skipped）: {results.get("skipped", None)} 个 +     预期失败用例个数（xfailed）: {results.get("xfailed", None)} 个 +     意外通过用例个数（xpassed）: {results.get("xpassed", None)} 个 +     失败重试用例个数 * 次数之和（rerun）: {results.get("rerun", None)} 个 +     成 功 率: {(results.get("passed") / results.get("total_cases")) * 100} % + + ********************************** + 附件为具体的测试报告，详细情况可下载附件进程查看， 非相关负责人员可忽略此消息。谢谢。 + """ # 默认不发送任何通知 if SEND_RESULT_TYPE == NotificationType.DEFAULT.value: pass # 发送邮件通知 elif SEND_RESULT_TYPE == NotificationType.EMAIL.value: - email_settings = email - yag = YagEmailServe(user=email_settings.get("user"), password=email_settings.get("password"), - host=email_settings.get("host")) - info = { - "subject": f"{results.get('project_name', None)} 接口自动化报告_{results.get('start_time', None)}", - "contents": f""" - 各位同事, 大家好: - -     自动化用例于{results.get('start_time', None)}开始运行，运行时长：{results.get('runs_time', None)}， 目前已执行完成。 - ----------------------------------------------------------------------------------------------------------- -     测试人：{results.get('tester', None)} / {results.get('dept', None)} -     测试平台：{results.get('platform', None)} / {results.get('python_version', None)} -     测试环境：{results.get('project_env', None)} - --------------------------------------------------------------------------------------------------------------- -     执行结果如下: -         用例运行总数: {results.get("total_cases", None)} 个 -         通过用例个数（passed）: {results.get("passed", None)} 个 -         失败用例个数（failed）: {results.get("failed", None)} 个 -         异常用例个数（error）: {results.get("error", None)} 个 -         跳过用例个数（skipped）: {results.get("skipped", None)} 个 -         预期失败用例个数（xfailed）: {results.get("xfailed", None)} 个 -         意外通过用例个数（xpassed）: {results.get("xpassed", None)} 个 -         失败重试用例个数 * 次数之和（rerun）: {results.get("rerun", None)} 个 -         成 功 率: {(results.get("passed") / results.get("total_cases")) * 100} % - - ********************************** - jenkins地址：https://xxxxxxxxx - 详细情况可登录jenkins平台查看，非相关负责人员可忽略此消息。谢谢。 - """, - "to": email.get("to"), - "attachments": attachment_path - - } - yag.send_email(info) + send_email(user=email.get("user"), pwd=email.get("password"), host=email.get("host"), subject=subject, + contents=content, to=email.get("to"), attachments=attachment_path) # 发送钉钉通知 elif SEND_RESULT_TYPE == NotificationType.DING_TALK.value: - pass + send_dingding(webhook_url=ding_talk["webhook_url"], secret=ding_talk["secret"], title=subject, text=content) # 发送企业微信通知 elif SEND_RESULT_TYPE == NotificationType.WECHAT.value: pass diff --git a/common_utils/dingding_handle.py b/common_utils/dingding_handle.py new file mode 100644 index 0000000..3ea6c87 --- /dev/null +++ b/common_utils/dingding_handle.py @@ -0,0 +1,252 @@ +# -*- coding: utf-8 -*- +# @Time : 2023/5/11 9:47 +# @Author : chenyinhua +# @File : dingding_handle.py +# @Software: PyCharm +# @Desc: 钉钉通知封装 + +import hmac +import hashlib +import base64 +import urllib.parse +import time +import urllib.request +from loguru import logger +from requests import request + + +class DingTalkBot: + """ + 钉钉机器人 + """ + + def __init__(self, webhook_url, secret=None): + """ + :param secret: 安全设置的加签秘钥 + :param webhook_url: 机器人没有加签的WebHook_url + """ + # 适配钉钉机器人的加签模式和关键字模式/白名单IP模式 + if secret: + timestamp = str(round(time.time() * 1000)) + sign = self.get_sign(secret, timestamp) + self.webhook_url = webhook_url + f'×tamp={timestamp}&sign={sign}' # 最终url，url+时间戳+签名 + else: + self.webhook_url = webhook_url + self.headers = { + "Content-Type": "application/json", + "Charset": "UTF-8" + } + + def get_sign(self, secret, timestamp): + """ + 根据时间戳 + "sign" 生成密钥 + 把timestamp+"\n"+密钥当做签名字符串，使用HmacSHA256算法计算签名，然后进行Base64 encode，最后再把签名参数再进行urlEncode，得到最终的签名（需要使用UTF-8字符集）。 + :return: + """ + string_to_sign = f'{timestamp}\n{secret}'.encode('utf-8') + hmac_code = hmac.new( + secret.encode('utf-8'), + string_to_sign, + digestmod=hashlib.sha256).digest() + + sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) + return sign + + def send_text(self, content, mobiles=None, is_at_all=False): + """ + 发送文本消息 + :param content: 发送的内容 + :param mobiles: 被艾特的用户的手机号码，格式是列表，注意需要在content里面添加@人的手机号码 + :param is_at_all: 是否艾特所有人，布尔类型，true为艾特所有人，false为不艾特 + """ + if mobiles: + if isinstance(mobiles, list): + payload = { + "msgtype": "text", + "text": { + "content": content + }, + "at": { + "atMobiles": mobiles, + "isAtAll": False + } + } + for mobile in mobiles: + payload["text"]["content"] += f"@{mobile}" + else: + raise TypeError("mobiles类型错误 不是list类型.") + else: + payload = { + "msgtype": "text", + "text": { + "content": content + }, + "at": { + "atMobiles": "", + "isAtAll": is_at_all + } + } + response = request(url=self.webhook_url, json=payload, headers=self.headers, method="POST") + if response.json().get("errcode") == 0: + logger.debug(f"send_text发送钉钉消息成功：{response.json()}") + return True + else: + logger.error(f"send_text发送钉钉消息失败：{response.text}") + return False + + def send_link(self, title, text, message_url, pic_url=None): + """ + 发送链接消息 + :param title: 消息标题 + :param text: 消息内容，如果太长只会部分展示 + :param message_url: 点击消息跳转的url地址 + :param pic_url: 图片url + """ + payload = { + "msgtype": "link", + "link": { + "title": title, + "text": text, + "picUrl": pic_url, + "messageUrl": message_url + } + } + response = request(url=self.webhook_url, json=payload, headers=self.headers, method="POST") + if response.json().get("errcode") == 0: + logger.debug(f"send_link发送钉钉消息成功：{response.json()}") + return True + else: + logger.error(f"send_link发送钉钉消息失败：{response.text}") + return False + + def send_markdown(self, title, text, mobiles=None, is_at_all=False): + """ + 发送markdown消息 + 目前仅支持md语法的子集，如标题，引用，文字加粗，文字斜体，链接，图片，无序列表，有序列表 + :param title: 消息标题，首屏回话透出的展示内容 + :param text: 消息内容，markdown格式 + :param mobiles: 被艾特的用户的手机号码，格式是列表，注意需要在text里面添加@人的手机号码 + :param is_at_all: 是否艾特所有人，布尔类型，true为艾特所有人，false为不艾特 + """ + if mobiles: + if isinstance(mobiles, list): + payload = { + "msgtype": "markdown", + "markdown": { + "title": title, + "text": text + }, + "at": { + "atMobiles": mobiles, + "isAtAll": False + } + } + for mobile in mobiles: + payload["markdown"]["text"] += f" @{mobile}" + else: + raise TypeError("mobiles类型错误 不是list类型.") + else: + payload = { + "msgtype": "markdown", + "markdown": { + "title": title, + "text": text + }, + "at": { + "atMobiles": "", + "isAtAll": is_at_all + } + } + response = request(url=self.webhook_url, json=payload, headers=self.headers, method="POST") + if response.json().get("errcode") == 0: + logger.debug(f"send_markdown发送钉钉消息成功：{response.json()}") + return True + else: + logger.error(f"send_markdown发送钉钉消息失败：{response.text}") + return False + + def send_action_card_single(self, title, text, single_title, single_url, btn_orientation=0): + """ + 发送消息卡片(整体跳转ActionCard类型) + :param title: 消息标题 + :param text: 消息内容，md格式消息 + :param single_title: 单个按钮的标题 + :param single_url: 点击singleTitle按钮后触发的URL + :param btn_orientation: 0-按钮竖直排列，1-按钮横向排列 + """ + payload = { + "msgtype": "actionCard", + "actionCard": { + "title": title, + "text": text, + "singleTitle": single_title, + "singleURL": single_url, + "btnOrientation": btn_orientation, + } + + } + response = request(url=self.webhook_url, json=payload, headers=self.headers, method="POST") + if response.json().get("errcode") == 0: + logger.debug(f"send_action_card_single发送钉钉消息成功：{response.json()}") + return True + else: + logger.error(f"send_action_card_single发送钉钉消息失败：{response.text}") + return False + + def send_action_card_split(self, title, text, btns, btn_orientation=0): + """ + 发送消息卡片(独立跳转ActionCard类型) + :param title: 消息标题 + :param text: 消息内容，md格式消息 + :param btns: 列表嵌套字典类型，"btns": [{"title": "内容不错", "actionURL": "https://www.dingtalk.com/"}, ......] + :param btn_orientation: 0-按钮竖直排列，1-按钮横向排列 + """ + payload = { + "msgtype": "actionCard", + "actionCard": { + "title": title, + "text": text, + "btns": [], + "btnOrientation": btn_orientation, + } + + } + for btn in btns: + payload["actionCard"]["btns"].append({ + "title": btn.get("title"), + "actionURL": btn.get("action_url") + }) + response = request(url=self.webhook_url, json=payload, headers=self.headers, method="POST") + if response.json().get("errcode") == 0: + logger.debug(f"send_action_card_split发送钉钉消息成功：{response.json()}") + return True + else: + logger.error(f"send_action_card_split发送钉钉消息失败：{response.text}") + return False + + def send_feed_card(self, links_msg): + """ + 发送多组消息卡片(FeedCard类型) + :param links_msg: 列表嵌套字典类型，每一个字段包括如下参数：title(单条信息文本), messageURL(点击单条信息后的跳转链接), picURL(单条信息后面图片的url) + """ + payload = { + "msgtype": "feedCard", + "feedCard": { + "links": [] + } + } + for link in links_msg: + payload["feedCard"]["links"].append( + { + "title": link.get("title"), + "messageURL": link.get("messageURL"), + "picURL": link.get("picURL") + } + ) + response = request(url=self.webhook_url, json=payload, headers=self.headers, method="POST") + if response.json().get("errcode") == 0: + logger.debug(f"send_feed_card发送钉钉消息成功：{response.json()}") + return True + else: + logger.error(f"send_feed_card发送钉钉消息失败：{response.text}") + return False diff --git a/config/settings.py b/config/settings.py index d6c8a5a..2d5720a 100644 --- a/config/settings.py +++ b/config/settings.py @@ -58,3 +58,9 @@ email = { "host": "smtp.qq.com", "to": ["******", "******"] # 收件人邮箱 } + +# ------------------------------------ 钉钉相关配置 ----------------------------------------------------# +ding_talk = { + "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=***********", + "secret": "***********" +}