88 lines
3.7 KiB
Python
88 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
# -------------------------------
|
||
# @文件：dingding_request.py
|
||
# @时间：2024/4/10 上午9:36
|
||
# @作者：caiweichao
|
||
# @功能描述：发送钉钉第三方机器人工具类
|
||
# -------------------------------
|
||
from util.basic.analysis_yaml import AnalysisYaml
|
||
from util.basic.http_request import HttpRequest
|
||
from util.auto_test.get_allure_data import CaseCount
|
||
|
||
import time
|
||
import hmac
|
||
import hashlib
|
||
import base64
|
||
import urllib.parse
|
||
|
||
from util.basic.log import Log
|
||
|
||
|
||
class DingdingRequest:
|
||
def __init__(self, robot_name):
|
||
# 获取对应的钉钉机器人配置
|
||
robot_info: dict = AnalysisYaml().get_dingding_robot_config(robot_name=robot_name)
|
||
self.test_team = robot_info.get('test_team')
|
||
try:
|
||
if robot_info.get("secret"):
|
||
timestamp = str(round(time.time() * 1000))
|
||
secret = str(robot_info.get("secret"))
|
||
secret_enc = secret.encode('utf-8')
|
||
string_to_sign = '{}\n{}'.format(timestamp, secret)
|
||
string_to_sign_enc = string_to_sign.encode('utf-8')
|
||
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
|
||
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
|
||
self.res_url = robot_info.get('url') + f"×tamp={timestamp}&sign={sign}"
|
||
else:
|
||
self.res_url = robot_info.get("url")
|
||
self.header = {"Content-Type": "application/json", "Charset": "UTF-8"}
|
||
except KeyError:
|
||
raise KeyError(f'机器人:{robot_name}的配置文件录入错误，请检查!')
|
||
except Exception:
|
||
raise Exception('未知异常请检查')
|
||
|
||
def send_msg(self, title: str, msg: str, is_at=None):
|
||
res_msg = {
|
||
"msgtype": "markdown",
|
||
"markdown": {
|
||
"title": title,
|
||
"text": msg
|
||
},
|
||
"at": {
|
||
"atMobiles": f"[{is_at}]"
|
||
}
|
||
}
|
||
res = HttpRequest(method="post", url=self.res_url, data=res_msg, headers=self.header)
|
||
Log.debug(f"钉钉机器人请求结果:{res}")
|
||
|
||
def send_test_report(self, report_url):
|
||
# 构建测试报告数据
|
||
detail = CaseCount()
|
||
total_case = detail.total_cases
|
||
pass_case = f'<font color=\"#009900\">{detail.pass_case}</font>'
|
||
fail_case = detail.fail_case if detail.fail_case == 0 else f"<font color=\"#FF0000\">{detail.fail_case}</font>"
|
||
broken_case = 0 if detail.broken_case == 0 else f"<font color=\"#8B4513\">{detail.broken_case}</font>"
|
||
skip_case = 0 if detail.skipped_case == 0 else f"<font color=\"#808080\">{detail.skipped_case}</font>"
|
||
# 处理测试报告详情
|
||
report_info = (f"#### {self.test_team}:用例执行明细\n"
|
||
f"| 描述 | 执行明细 \n| --- | --- \n"
|
||
f"| 执行用例总数 | {total_case} \n"
|
||
f"| 通过用例数 | {pass_case} \n"
|
||
f"| 失败用例数 | {fail_case} \n"
|
||
f"| 异常用例数 | {broken_case} \n"
|
||
f"| 跳过用例数 | {skip_case} \n"
|
||
f"| 测试报告链接 | [点击查看]({report_url})"
|
||
)
|
||
test_detail = {
|
||
"msgtype": "markdown",
|
||
"markdown": {
|
||
"title": "测试结果",
|
||
"text": report_info
|
||
}
|
||
}
|
||
# 执行用例为空时不会进行钉钉推送
|
||
if total_case != 0:
|
||
HttpRequest(method='post', url=self.res_url, data=test_detail, headers=self.header)
|
||
|
||
if __name__ == '__main__':
|
||
DingdingRequest(robot_name='DEMO_ROBOT_NAME').send_test_report(report_url="https://www.baidu.com") |