🧾 项目背景        

        在自动化测试中，及时获取测试失败的结果是非常关键的一环。为了实现这一目标，我们可以通过企业微信机器人将测试失败信息实时推送到指定群聊中，以便团队成员快速响应问题。
        本文将分享一个使用 Python 封装企业微信 Webhook 接口，实现自动化测试失败结果推送功能的实践案例，并附上完整的可运行代码。

🧰 技术栈

• Python 3.x
• Requests 库（用于发送 HTTP 请求）
• 正则表达式（提取关键信息）
• 企业微信机器人 Webhook

📦 脚本功能概述

该脚本主要完成以下几个任务：

1. 过滤出失败的测试用例；
2. 构建 Markdown 格式的失败报告；
3. 发送消息到企业微信机器人；
4. 支持长消息截断并提示查看完整报告；
5. 提取关键错误信息并格式化显示。

💡 完整代码与详解

以下是一个完整的、可用于生产环境的企业微信推送脚本，包含异常处理、日志记录、Markdown 消息构建等核心逻辑。

📄 文件结构

project_root/
├── config/
│   └── setting.py        # 配置 Webhook URL
└── utils/
    └── wechat_bot.py     # 推送脚本主文件

📝 setting.py 示例配置

# config/setting.py
WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxxxxxxxx"

🧠 wechat_bot.py 主要逻辑

✅ 导入依赖库

import requests
import json
from typing import List, Dict
from datetime import datetime
import re
import logging
from config.setting import WEBHOOK_URL

🛠 初始化日志

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

📮 推送函数：send_to_wechat_robot

def send_to_wechat_robot(WEBHOOK_URL: str, test_results: List[Dict]) -> None:
    """
    将失败的测试结果推送到企业微信机器人。

    :param WEBHOOK_URL: 企业微信机器人的 Webhook URL
    :param test_results: 测试结果列表，每个结果为字典格式
    """
    # 过滤出失败的测试结果
    failed_results = [result for result in test_results if result.get("status") == "failed"]

    if not failed_results:
        print("✅ 没有失败的测试用例，无需推送！")
        return

    # 构造消息内容
    markdown_content = generate_markdown_content(failed_results)

    # 添加消息长度限制逻辑
    MAX_MESSAGE_LENGTH = 1900
    if len(markdown_content) > MAX_MESSAGE_LENGTH:
        # 截断内容，并添加提示
        markdown_content = markdown_content[:MAX_MESSAGE_LENGTH]
        markdown_content += "---\n"
        markdown_content += "\n⚠️⚠️⚠️ **提示**: 消息已截断，原始内容过长，请查看完整报告！ ⚠️⚠️⚠️\n"
        markdown_content += "🤖 *本消息由数据监测自动发送*\n"

    message = {
        "msgtype": "markdown",
        "markdown": {
            "content": markdown_content
        }
    }

    try:
        # 发送 POST 请求
        response = requests.post(WEBHOOK_URL, data=json.dumps(message), headers={"Content-Type": "application/json"})
        if response.status_code == 200:
            print("🟥 失败测试结果已成功推送至企业微信机器人！")
        else:
            print(f"❌ 消息推送失败，状态码: {response.status_code}, 响应内容: {response.text}")
    except Exception as e:
        print(f"⚠️ 消息推送发生异常: {str(e)}")

📝 Markdown 内容生成函数：generate_markdown_content

def generate_markdown_content(test_results: List[Dict]) -> str:
    content = "# ❌ 自动化测试 - 失败结果报告\n"
    content += "> 📅 **推送时间**: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + "\n\n"
    content += "---\n"

    for result in test_results:
        name = result.get("name", "未知用例")
        short_name = extract_short_name(name)

        raw_message = str(result.get("excinfo_value", "无详细信息")).strip()

        interface_value_match = re.search(
            r"实际值\s*[\[\('\"\{]?([^)'\"\]\}\s]+)[\]\)'\"\}]?\s*(?:不等于预期值|与预期值不一致)",
            raw_message
        )

        indicator_value_match = re.search(
            r"实际值\s*[\[\('\"\{]?([^)'\"\]\}\s]+)[\]\)'\"\}]?\s*(?:等于预期值|与预期值一致)",
            raw_message
        )

        if interface_value_match:
            actual_value = interface_value_match.group(1)
            value_type = "接口状态返回值"
        elif indicator_value_match:
            actual_value = indicator_value_match.group(1)
            value_type = "指标返回值"
        else:
            actual_value = "未提取到具体值"
            value_type = "未知类型"

        def escape_markdown(text: str):
            return re.sub(r"([_*\[\]()~`>#+\-=|{}.!\\])", r"\\\1", text) if text else "-"

        actual_value_escaped = escape_markdown(actual_value)

        content += f"📌 **指标名称**: `{short_name}`\n"

        if value_type == "接口状态返回值":
            content += f"🔴 **错误详情**: '{value_type}为：{actual_value_escaped}'\n"
        elif value_type == "指标返回值":
            content += f"🔵 **错误详情**: '{value_type}为：{actual_value_escaped}'\n"
        else:
            content += f"⚠️ **错误详情**: '{actual_value}'\n"

        content += "---\n"

    content += "🤖 *本消息由数据监测自动发送*\n"
    return content

🔍 提取短名函数：extract_short_name

def extract_short_name(full_name: str) -> str:
    """
    提取简短的用例名称。

    :param full_name: 完整的测试用例名称（如 tests/test_销售沙盘.yaml::盘数-应收及回款-【全年回款】-M线-老张）
    :return: 简短的用例名称（如 盘数-应收及回款-【全年回款】-M线-老张）
    """
    parts = full_name.split("::")
    if len(parts) > 1:
        return parts[-1].strip()
    return full_name.strip()

🧪 示例调用入口

if __name__ == '__main__':
    # 示例：测试结果数据（模拟从 pytest 收集到的失败测试用例）
    test_results = [
        {
            "name": "盘数-年上刊目标-【全年回款】-M线-老张",
            "status": "failed",
            "excinfo_value": "处理验证规则时发生错误: 验证失败（eq比较实际值和预期值是否相等）: 实际值 500 不等于预期值 200"
        },
        {
            "name": "盘客户-171动作-【集团TOP-打样转化率-实际】-M线-老张",
            "status": "failed",
            "excinfo_value": "处理验证规则时发生错误: 验证失败（not_eq比较实际值和预期值是否不相等）: 实际值 - 等于预期值 -"
        },
        {
            "name": "盘客户-171动作-【集团TOP-一层KP见面率】-M线-老张",
            "status": "failed",
            "excinfo_value": "验证失败（not_eq比较实际值和预期值是否不相等）: 实际值 None 等于预期值 None"
        }
    ]

    # 调用推送方法
    send_to_wechat_robot(WEBHOOK_URL, test_results)

🎨 消息格式示例

# ❌ 自动化测试 - 失败结果报告
> 📅 **推送时间**: 2023-10-10 14:30:00

---
📌 **指标名称**: `盘数-年上刊目标-【全年回款】-M线-老张`
🔴 **错误详情**: '接口状态返回值为：500'
---
🤖 *本消息由数据监测自动发送*

⚙️ 配置与部署建议

• 在 config/setting.py 中配置你的企业微信 Webhook 地址；
• 可结合 CI/CD 工具（如 Jenkins、GitLab CI）触发推送逻辑；
• 若需更复杂的通知逻辑，可扩展支持图文、卡片等多种消息类型；
• 支持集成到 Pytest、Allure、Jenkins Pipeline 等自动化测试平台中。

✅ 总结

        通过封装企业微信 Webhook 接口，我们可以非常方便地实现自动化测试失败通知机制。该脚本具有良好的扩展性，可以轻松集成到各类自动化测试框架中，提升团队协作效率和问题响应速度。

        如果你也在做自动化测试告警相关的工作，不妨试试这个方法！