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

(2023-05-12) 支持测试完成后，通过企业微信发送pytest-html测试结果

Browse Source
This commit is contained in:
floraachy
2023-05-18 08:45:02 +08:00
parent 34a6762438
commit 950675b751
3 changed files with 229 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
case_utils/send_result_handle.py
View File

@@ -7,9 +7,10 @@
from loguru import logger
from common_utils.yagmail_handle import YagEmailServe
from config.global_vars import NotificationType
from config.settings import SEND_RESULT_TYPE, email, ding_talk
from config.settings import SEND_RESULT_TYPE, email, ding_talk, wechat
from common_utils.bs4_handle import SoupAPI
from common_utils.dingding_handle import DingTalkBot
from common_utils.webchat_handle import WechatBot
def get_test_info_from_html_report(html_report_path):
@@ -111,6 +112,26 @@ def send_dingding(webhook_url, secret, title, text):
logger.error(f"发送钉钉通知异常， 错误信息：{e}")
def send_wechat(webhook_url, content, attachment=None):
"""
发送企业微信消息
"""
try:
wechat = WechatBot(webhook_url=webhook_url)
msg = wechat.send_markdown(content=content)
if msg:
if attachment:
file = wechat.send_file(wechat.upload_file(attachment))
if file:
logger.info(f"发送企业微信通知(包括文本以及附件)成功~")
else:
logger.error(f"发送企业微信通知(附件)失败~")
else:
logger.error(f"发送企业微信（文本）失败~")
except Exception as e:
logger.error(f"发送企业微信通知异常， 错误信息：{e}")
def send_result(results, attachment_path=None):
"""
根据用户配置，采取指定方式，发送测试结果
@@ -120,22 +141,22 @@ def send_result(results, attachment_path=None):
content = f"""
各位同事, 大家好:
  自动化用例于{results.get('start_time', None)}开始运行，运行时长：{results.get('runs_time', None)}， 目前已执行完成。
### 自动化用例于{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('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} %
#### 执行结果如下:
- 用例运行总数: {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} %
**********************************
附件为具体的测试报告，详细情况可下载附件进程查看， 非相关负责人员可忽略此消息。谢谢。
@@ -153,7 +174,15 @@ def send_result(results, attachment_path=None):
send_dingding(webhook_url=ding_talk["webhook_url"], secret=ding_talk["secret"], title=subject, text=content)
# 发送企业微信通知
elif SEND_RESULT_TYPE == NotificationType.WECHAT.value:
pass
send_wechat(webhook_url=wechat["webhook_url"], content=content, attachment=attachment_path)
# 全部渠道都发送通知
else:
pass
# 发送邮件
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)
# 发送钉钉通知
send_dingding(webhook_url=ding_talk["webhook_url"], secret=ding_talk["secret"], title=subject, text=content)
# 发送企业微信
send_wechat(webhook_url=wechat["webhook_url"], content=content, attachment=attachment_path)

178
common_utils/webchat_handle.py Normal file
View File

@@ -0,0 +1,178 @@
# -*- coding: utf-8 -*-
# @Time : 2023/5/11 15:01
# @Author : chenyinhua
# @File : webchat_handle.py
# @Software: PyCharm
# @Desc: 企业微信机器人
import os
from requests import request
from loguru import logger
import base64
import hashlib
import re
class WechatBot:
"""
企业微信机器人
当前自定义机器人支持文本（text）、markdown（markdown）、图片（image）、图文（news）, 文件（file）五种消息类型。
机器人的text/markdown类型消息支持在content中使用<@userid>扩展语法来@群成员
"""
def __init__(self, webhook_url):
"""
:param webhook_url: 机器人的WebHook_url
"""
self.webhook_url = webhook_url
self.headers = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
def send_text(self, content, mentioned_list=[], mentioned_mobile_list=[]):
"""
发送文本消息
:param content: 文本内容，最长不超过2048个字节，必须是utf8编码
:param mentioned_list: userid的列表，提醒群中的指定成员(@某个成员)，@all表示提醒所有人，如果开发者获取不到userid，可以使用mentioned_mobile_list
:param mentioned_mobile_list: 手机号列表，提醒手机号对应的群成员(@某个成员)，@all表示提醒所有人
"""
payload = {
"msgtype": "text",
"text": {
"content": content,
"mentioned_list": mentioned_list,
"mentioned_mobile_list": mentioned_mobile_list
}
}
response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
if response.json().get("errcode") == 0:
logger.debug(f"通过企业微信发送文本消息成功：{response.json()}")
return True
else:
logger.error(f"通过企业微信发送文本消息失败：{response.text}")
return False
def send_markdown(self, content):
"""
发送markdown消息
目前支持的markdown语法是如下的子集：
1. 标题 （支持1至6级标题，注意#与文字中间要有空格）
2. 加粗
3. 链接
4. 行内代码段（暂不支持跨行）
5. 引用
6. 字体颜色(只支持3种内置颜色), 绿色（color="info"），灰色（color="comment"），橙红色（color="warning"）
:param content: markdown内容，最长不超过4096个字节，必须是utf8编码
"""
payload = {
"msgtype": "markdown",
"markdown": {
"content": content
}
}
response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
if response.json().get("errcode") == 0:
logger.debug(f"通过企业微信发送md消息成功：{response.json()}")
return True
else:
logger.error(f"通过企业微信发送md消息失败：{response.text}")
return False
def send_picture(self, image_path):
"""
发送图片消息
:param image_path: 图片的绝对路径
"""
with open(image_path, "rb") as f:
image_data = f.read()
payload = {
"msgtype": "image",
"image": {
"base64": base64.b64encode(image_data).decode("utf-8"), # # 将图片数据转换成Base64编码格式
"md5": hashlib.md5(image_data).hexdigest() # # 计算图片的MD5值
}
}
response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
if response.json().get("errcode") == 0:
logger.debug(f"通过企业微信发送图片消息成功：{response.json()}")
return True
else:
logger.error(f"通过企业微信发送图片失败：{response.text}")
return False
def send_text_picture(self, articles: list):
"""
发送图文消息
:param articles: 图文消息，一个图文消息支持1到8条图文, 包括如下字段
1. title: 标题，不超过128个字节，超过会自动截断
2. description: 非必填，描述，不超过512个字节，超过会自动截断
3. url: 点击后跳转的链接。
4. picurl: 非必填，图文消息的图片链接，支持JPG、PNG格式，较好的效果为大图 1068*455，小图150*150。
"""
payload = {
"msgtype": "news",
"news": {
"articles": [
]
}
}
for article in articles:
payload["news"]["articles"].append(
{
"title": article.get("title"),
"description": article.get("description", ""),
"url": article.get("url"),
"picurl": article.get("picurl", "")
}
)
response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
if response.json().get("errcode") == 0:
logger.debug(f"通过企业微信发送图文消息成功：{response.json()}")
return True
else:
logger.error(f"通过企业微信发送图文失败：{response.text}")
return False
def upload_file(self, file_path):
"""
上传文件到企业微信服务器(要求文件大小在5B~20M之间)
注意：素材上传得到media_id，该media_id仅三天内有效；media_id只能是对应上传文件的机器人可以使用
:param file_path: 文件绝对路径
"""
token_regex = r"key=([\w-]+)"
match = re.search(token_regex, self.webhook_url)
token = match.group(1)
url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key={token}&type=file"
headers = {
"Content-Type": "multipart/form-data;"
}
with open(file_path, "rb") as f:
files = {"media": (os.path.basename(file_path), f.read())}
response = request(url=url, method="POST", files=files, headers=headers)
if response.json().get("errcode") == 0:
media_id = response.json().get("media_id")
logger.debug(f"上传文件成功，media_id= {media_id}")
return media_id
else:
logger.error(f"上传文件失败：{response.text}")
return False
def send_file(self, media_id):
"""
发送文件
:param media_id: 文件id，通过下文的文件上传接口获取
"""
payload = {
"msgtype": "file",
"file": {
"media_id": media_id,
}
}
response = request(url=self.webhook_url, method="POST", json=payload, headers=self.headers)
if response.json().get("errcode") == 0:
logger.debug(f"通过企业微信发送文件消息成功：{response.json()}")
return True
else:
logger.error(f"通过企业微信发送文件消息失败：{response.text}")
return False

5
config/settings.py
View File

@@ -64,3 +64,8 @@ ding_talk = {
"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=***********",
"secret": "***********"
}
# ------------------------------------ 企业微信相关配置 ----------------------------------------------------#
wechat = {
"webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=********",
}