# -*- coding: utf-8 -*- """ @File : message_template.py @Time : 2023/11/21 17:29 @Author : geekbing @LastEditTime : - @LastEditors : - @Description : 消息模板 """ from typing import Dict from django.conf import settings import time from datetime import datetime def parse_message(summary: Dict, **kwargs): """ 解析消息模板 :param summary: 测试报告摘要 :param kwargs: 其他参数 :return: """ task_name = summary["task_name"] rows_count = summary["stat"]["testsRun"] pass_count = summary["stat"]["successes"] fail_count = summary["stat"]["failures"] error_count = summary["stat"]["errors"] skipped_count = summary["stat"].get("skipped", 0) expected_failures_count = summary["stat"].get("expectedFailures", 0) unexpected_successes_count = summary["stat"].get("unexpectedSuccesses", 0) duration = "%.2fs" % summary["time"]["duration"] report_id = summary["report_id"] base_url = settings.IM_REPORT_SETTING.get("base_url") port = settings.IM_REPORT_SETTING.get("port") report_url = f"{base_url}:{port}/api/lunarlink/reports/{report_id}" executed = rows_count fail_rate = "{:.2%}".format(fail_count / executed) if executed > 0 else "0.00%" success_rate = "{:.2%}".format(pass_count / executed) if executed > 0 else "0.00%" case_count = kwargs.get("case_count") creator = kwargs.get("creator", "") updater = kwargs.get("updater", "") start_at = summary.get("time", {}).get("start_at", 0) if start_at: start_time = datetime.fromtimestamp(start_at).strftime("%Y-%m-%d %H:%M:%S") end_time = datetime.fromtimestamp(start_at + summary["time"]["duration"]).strftime("%Y-%m-%d %H:%M:%S") else: start_time = "" end_time = "" setup_hooks_duration = summary.get("time", {}).get("setup_hooks_duration", 0) teardown_hooks_duration = summary.get("time", {}).get("teardown_hooks_duration", 0) platform_info = summary.get("platform", {}) python_version = platform_info.get("python_version", "") httprunner_version = platform_info.get("httprunner_version", "") avg_duration = 0 if executed > 0: avg_duration = summary["time"]["duration"] / executed failed_cases = [] error_cases = [] details = summary.get("details", []) for detail in details: records = detail.get("records", []) for record in records: if record.get("status") in ["failure", "error"]: case_name = record.get("name", "未知用例") case_status = record.get("status", "") meta_data = record.get("meta_data", {}) case_url = meta_data.get("request", {}).get("url", "") case_method = meta_data.get("request", {}).get("method", "") case_error = record.get("attachment", "") if case_error: case_error_lines = case_error.split('\n') if len(case_error_lines) > 1: case_error = case_error_lines[0] if len(case_error_lines) > 2: case_error += f"\n... (共{len(case_error_lines)}行错误信息)" case_error = case_error.strip() else: case_error = "无错误信息" case_info = { "name": case_name, "url": case_url, "method": case_method, "error": case_error[:150] if case_error else "无错误信息" } if case_status == "failure": failed_cases.append(case_info) elif case_status == "error": error_cases.append(case_info) failed_cases = failed_cases[:10] error_cases = error_cases[:10] return { "task_name": task_name, "duration": duration, "case_count": case_count, "pass_count": pass_count, "error_count": error_count, "fail_count": fail_count, "fail_rate": fail_rate, "success_rate": success_rate, "report_url": report_url, "creator": creator, "updater": updater, "start_time": start_time, "end_time": end_time, "skipped_count": skipped_count, "expected_failures_count": expected_failures_count, "unexpected_successes_count": unexpected_successes_count, "setup_hooks_duration": setup_hooks_duration, "teardown_hooks_duration": teardown_hooks_duration, "python_version": python_version, "httprunner_version": httprunner_version, "avg_duration": avg_duration, "failed_cases": failed_cases, "error_cases": error_cases, } def email_msg_template( task_name, duration, case_count, pass_count, error_count, fail_count, fail_rate, report_url, success_rate, creator, updater, start_time, failed_cases, error_cases, end_time, skipped_count, expected_failures_count, unexpected_successes_count, setup_hooks_duration, teardown_hooks_duration, python_version, httprunner_version, avg_duration, ): """ 定制邮件报告消息模板（高级优化版） :param task_name: 任务名称 :param duration: 总耗时 :param case_count: 用例个数 :param pass_count: 成功接口个数 :param error_count: 异常接口个数 :param fail_count: 失败接口个数 :param fail_rate: 失败比例 :param report_url: 报告链接 :param success_rate: 成功率 :param creator: 创建人 :param updater: 更新人 :param start_time: 开始时间 :param failed_cases: 失败用例列表 :param error_cases: 异常用例列表 :param end_time: 结束时间 :param skipped_count: 跳过用例数 :param expected_failures_count: 预期失败数 :param unexpected_successes_count: 意外成功数 :param setup_hooks_duration: 前置钩子耗时 :param teardown_hooks_duration: 后置钩子耗时 :param python_version: Python版本 :param httprunner_version: HttpRunner版本 :param avg_duration: 平均用例耗时 :return: 邮件主题及 HTML 内容 """ email_subject = f"【自动化测试报告】{task_name}" status_icon = "✅" if fail_count == 0 and error_count == 0 else ("⚠️" if fail_count == 0 else "❌") if error_count == 0 and fail_count == 0: status_text = "测试通过" status_color = "#2ecc71" status_bg = "#e8f8f5" elif fail_count == 0: status_text = "部分异常" status_color = "#f39c12" status_bg = "#fef5e7" else: status_text = "测试失败" status_color = "#e74c3c" status_bg = "#fdedec" failed_cases_html = "" if failed_cases: failed_cases_html = "
" for idx, case in enumerate(failed_cases, 1): failed_cases_html += f"""
{idx} {case['name']}
{case['method']} {case['url'][:60]}...
❌ {case['error']}
""" failed_cases_html += "
" error_cases_html = "" if error_cases: error_cases_html = "
" for idx, case in enumerate(error_cases, 1): error_cases_html += f"""
{idx} {case['name']}
{case['method']} {case['url'][:60]}...
⚠️ {case['error']}
""" error_cases_html += "
" email_content = f""" 自动化测试报告

🚀 自动化测试报告

{task_name}

{status_icon} {status_text}
📊 测试概览
总耗时 {duration}
用例总数 {case_count}
成功率 {success_rate}
失败率 {fail_rate}
✅ 成功 {pass_count}
❌ 失败 {fail_count}
⚠️ 异常 {error_count}
📈 通过率 {success_rate}
⏭️ 跳过 {skipped_count}
🎯 预期失败 {expected_failures_count}
🎉 意外成功 {unexpected_successes_count}
⚡ 平均耗时 {avg_duration:.3f}s
🔧 前置钩子 {setup_hooks_duration:.3f}s
🔧 后置钩子 {teardown_hooks_duration:.3f}s
📋 详细信息
📝任务名称 {task_name}
⏱️执行耗时 {duration}
⚡平均耗时 {avg_duration:.3f}s
📅开始时间 {start_time if start_time else '未记录'}
🏁结束时间 {end_time if end_time else '未记录'}
👤创建人 {creator if creator else '未记录'}
✏️更新人 {updater if updater else '未记录'}
🐍Python版本 {python_version if python_version else '未记录'}
🏃HttpRunner版本 {httprunner_version if httprunner_version else '未记录'}
📈 测试结果分析
测试通过率 {success_rate}
成功: {pass_count} 失败: {fail_count} 异常: {error_count}
{f'
❌ 失败用例列表
{failed_cases_html}' if failed_cases else ''} {f'
⚠️ 异常用例列表
{error_cases_html}' if error_cases else ''} 👉 点击查看详细报告

本邮件由系统自动发出，请勿回复

如有疑问，请联系相关负责人

""" return { "subject": email_subject, "html_message": email_content, } def qy_msg_template( task_name, duration, case_count, pass_count, error_count, fail_count, fail_rate, report_url, msg_type: str = "markdown", ): """ 定制企业微信消息模板 :param task_name: :param duration: :param case_count: :param pass_count: :param error_count: :param fail_count: :param fail_rate: :param report_url: :param msg_type: :return: """ if msg_type == "markdown": msg_template = {"msgtype": "markdown", "markdown": {"content": ""}} content = f"""**接口自动化测试报告** \n >任务名称: {task_name} >总共耗时: {duration} >用例个数: {case_count} >成功接口: **{pass_count}** >异常接口: **{error_count}** >失败接口: **{fail_count}** >失败比例: **{fail_rate}** >测试报告: [点击查看]({report_url})""" msg_template["markdown"]["content"] = content return msg_template text = f" 任务名称: {task_name}\n 总共耗时: {duration}\n 成功接口: {pass_count}个\n 异常接口: {error_count}个\n 失败接口: {fail_count}个\n 失败比例: {fail_rate}\n 查看详情: {report_url}" return text def dd_msg_template( task_name, duration, case_count, pass_count, error_count, fail_count, fail_rate, report_url, msg_type: str = "markdown", creator=None, updater=None, ): """ 定制钉钉消息模板：包含自动评价、图标提示、markdown美观布局 :param task_name: 任务名称 :param duration: 总耗时 :param case_count: 用例个数 :param pass_count: 成功接口个数 :param error_count: 异常接口个数 :param fail_count: 失败接口个数 :param fail_rate: 失败比例（字符串或百分比） :param report_url: 报告链接 :param msg_type: 消息类型，默认 markdown :return: 钉钉消息内容（字典格式） """ # 安全转换失败比例 try: fr = float(str(fail_rate).strip().replace("%", "").replace("％", "")) except Exception: fr = 0.0 # 状态图标 status_icon = "✅" if fail_count == 0 and error_count == 0 else ("⚠️" if fail_count == 0 else "❌") # 自动评价等级 if error_count == 0 and fail_count == 0: evaluation = "测试结果优秀" evaluation_icon = "🟢" elif fr <= 5: evaluation = "测试结果良好" evaluation_icon = "🟡" elif fr <= 10: evaluation = "测试结果一般" evaluation_icon = "🟠" else: evaluation = "测试结果严重异常" evaluation_icon = "🔴" if msg_type == "markdown": content = ( f"## {status_icon} **接口自动化测试完成通知**\n\n" f"### 📌 任务名称：**{task_name}**\n" f"### 📈 自动评价：{evaluation_icon} `{evaluation}`\n\n" f"> ⏱️ **总耗时：** {duration} \n" f"> 📊 **执行用例数：** {case_count} \n" f"> ✅ **成功接口数：** {pass_count} \n" f"> ⚠️ **异常接口数：** {error_count} \n" f"> ❌ **失败接口数：** {fail_count} \n" f"> 📉 **失败比例：** {fail_rate} \n\n" f"> 👤 **创建人：** {creator if creator else '未记录'} \n" # 新增创建人显示 f"> ✏️ **更新人：** {updater if updater else '未记录'} \n\n" # 新增更新人显示 f"🔗 [👉 点击查看详细报告]({report_url})\n\n" f"📬 详细内容也可通过邮件查看。\n\n" "---\n" "如有异常，请及时处理！🎉" ) return {"msgtype": "markdown", "markdown": {"content": content}} else: # 普通 text 消息 fallback content = ( f"{status_icon} 接口自动化测试完成通知\n" f"任务名称：{task_name}\n" f"自动评价：{evaluation_icon} {evaluation}\n" f"总耗时：{duration}\n" f"执行用例数：{case_count}\n" f"成功接口数：{pass_count}\n" f"异常接口数：{error_count}\n" f"失败接口数：{fail_count}\n" f"失败比例：{fail_rate}\n" f"创建人：{creator}\n" f"更新人：{updater}\n" f"报告地址：{report_url}\n" f"详细内容请查看邮件，如有异常，请及时处理！。" ) return {"msgtype": "text", "text": {"content": content}}