添加企微测试报告推送

README.md

@@ -28,7 +28,7 @@
 - 兼容 yaml / json 两种文件格式编写测试数据
 - 测试用例自动生成，可以根据测试数据文件自动生成测试用例
 - 自动测试报告，html, allure
-- 自动测试结果通知，飞书，钉钉，~~企业微信~~，邮箱
+- 自动测试结果通知，飞书，钉钉，企业微信，邮箱
 - ......
 
 ## 流程图

httpfpt/core/conf.toml

@@ -48,6 +48,13 @@ proxies.http = ''
 proxies.https = ''
 send = false
 
+# 企业微信
+[wechat]
+webhook = ''
+proxies.http = ''
+proxies.https = ''
+send = false
+
 
 # 请求发送
 [request]

httpfpt/core/get_conf.py

@@ -67,6 +67,18 @@ def __init__(self) -> None:
         }
         self.FEISHU_SEND = glom(self.__config, 'feishu.send')
 
+        # 企业微信
+        self.WECHAT_WEBHOOK = glom(self.__config, 'wechat.webhook')
+        self.WECHAT_PROXY = {
+            'http': glom(self.__config, 'wechat.proxies.http')
+            if glom(self.__config, 'wechat.proxies.http') != ''
+            else None,
+            'https': glom(self.__config, 'wechat.proxies.https')
+            if glom(self.__config, 'wechat.proxies.https') != ''
+            else None,
+        }
+        self.WECHAT_SEND = glom(self.__config, 'wechat.send')
+
         # 请求发送
         self.REQUEST_TIMEOUT = glom(self.__config, 'request.timeout')
         self.REQUEST_VERIFY = glom(self.__config, 'request.verify')

httpfpt/run.py

@@ -29,6 +29,7 @@
 from httpfpt.utils.send_report.dingding import DingDing
 from httpfpt.utils.send_report.email import SendEmail
 from httpfpt.utils.send_report.feishu import FeiShu
+from httpfpt.utils.send_report.wechat import WeChat
 from httpfpt.utils.time_control import get_current_time
 
 
@@ -139,6 +140,9 @@ def startup(
     if config.FEISHU_SEND:
         FeiShu(test_result).send()
 
+    if config.WECHAT_SEND:
+        WeChat(test_result).send()
+
     if allure:
         if not os.path.exists(ALLURE_REPORT_ENV_FILE):
             shutil.copyfile(ALLURE_ENV_FILE, ALLURE_REPORT_ENV_FILE)

httpfpt/utils/send_report/wechat.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from httpfpt.common.log import log
+from httpfpt.core.get_conf import config
+
+
+class WeChat:
+    def __init__(self, content: dict):
+        self.content = content
+
+    def send(self) -> None:
+        # 发送企业微信消息
+        try:
+            import requests
+
+            headers = {'Content-Type': 'application/json; charset=utf-8', 'Connection': 'close'}
+            data = {
+                'msgtype': 'markdown',
+                'markdown': {
+                    'content': f"# {config.TEST_REPORT_TITLE}\n"
+                    f"> 👤 测试人员: **{config.TESTER_NAME}**\n"
+                    f"> 🤖 测试结果: **{self.content['result']}**\n"
+                    f"> ✅ 通过用例: <font color='info'>**{self.content['passed']}**</font>\n"
+                    f"> 🔧 失败用例: **{self.content['failed']}**\n"
+                    f"> ❌ 错误用例: **{self.content['error']}**\n"
+                    f"> ⚠️ 跳过用例: **{self.content['skipped']}**\n"
+                    f"> ⌛ 开始时间: **{self.content['started_time']}**\n"
+                    f"> ⏱️ 执行耗时: **{self.content['elapsed']}**\n"
+                    "> ➡️ 查看报告: [点击跳转](https://foryourself)"
+                },
+            }
+            response = requests.session().post(
+                url=config.WECHAT_WEBHOOK,
+                json=data,
+                headers=headers,
+                proxies=config.WECHAT_PROXY,  # type: ignore
+            )
+            response.raise_for_status()
+        except Exception as e:
+            log.error(f'企业微信消息发送异常: {e}')
+        else:
+            log.success('企业微信发送成功')