diff --git a/README.md b/README.md index 96c3d61..ee96caa 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ - 兼容 yaml / json 两种文件格式编写测试数据 - 测试用例自动生成,可以根据测试数据文件自动生成测试用例 - 自动测试报告,html, allure -- 自动测试结果通知,飞书,钉钉,~~企业微信~~,邮箱 +- 自动测试结果通知,飞书,钉钉,企业微信,邮箱 - ...... ## 流程图 diff --git a/httpfpt/core/conf.toml b/httpfpt/core/conf.toml index 40c5dc7..24977df 100644 --- a/httpfpt/core/conf.toml +++ b/httpfpt/core/conf.toml @@ -48,6 +48,13 @@ proxies.http = '' proxies.https = '' send = false +# 企业微信 +[wechat] +webhook = '' +proxies.http = '' +proxies.https = '' +send = false + # 请求发送 [request] diff --git a/httpfpt/core/get_conf.py b/httpfpt/core/get_conf.py index b69783d..4bbf5e0 100644 --- a/httpfpt/core/get_conf.py +++ b/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') diff --git a/httpfpt/run.py b/httpfpt/run.py index 26f5650..9a48744 100644 --- a/httpfpt/run.py +++ b/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) diff --git a/httpfpt/utils/send_report/wechat.py b/httpfpt/utils/send_report/wechat.py new file mode 100644 index 0000000..50af58a --- /dev/null +++ b/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"> ✅ 通过用例: **{self.content['passed']}**\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('企业微信发送成功')