floraachy/apiautotest
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(2023-05-11) 支持测试完成后，通过钉钉发送pytest-html测试结果

Browse Source
This commit is contained in:
floraachy
2023-05-17 17:33:01 +08:00
parent af7ffc1de8
commit 34a6762438
3 changed files with 322 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
case_utils/send_result_handle.py
+64 -37
View File
@@ -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
common_utils/dingding_handle.py
+252
View File
@@ -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'&timestamp={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
config/settings.py
+6
View File
@@ -58,3 +58,9 @@ email = {
"host": "smtp.qq.com",
"to": ["******", "******"] # 收件人邮箱
}
# ------------------------------------ 钉钉相关配置 ----------------------------------------------------#
ding_talk = {
"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=***********",
"secret": "***********"
}