提交 bbfffa46 authored 作者: 陈泽健's avatar 陈泽健

feat(dingtalk): 钉钉报告通知版面改版为分区卡片式

- 三形态统一标题「UI自动化定时报告」(原「UI自动化告警」),仅以 emoji 区分: 全部通过 / ️ 存在失败 / 🚨 执行异常
- 不打印失败用例明细:版面为分区卡片式(标题/任务/被测系统/汇总统计/失败提示/耗时/链接),块间 \n\n 分隔适配钉钉 markdown 折叠行为
- 保留一句失败统计提示( 存在 N 项失败用例,请及时查看报告),@手机号仅失败/异常形态携带
- _shrink_message 简化为整体截断兜底;测试更新为 29 项覆盖新版面
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 4b8d773b
...@@ -6,13 +6,16 @@ ...@@ -6,13 +6,16 @@
作者:czj 作者:czj
创建日期:2026-09-07 创建日期:2026-09-07
最后修改:2026-09-07 最后修改:2026-09-08
实现说明: 实现说明:
- 接入模式为钉钉群机器人 Webhook + 加签(可选),与系统设置中已有的 - 接入模式为钉钉群机器人 Webhook + 加签(可选),与系统设置中已有的
「钉钉对接配置(企业内部应用)」相互独立、互不影响; 「钉钉对接配置(企业内部应用)」相互独立、互不影响;
- 消息为 markdown msgtype,按执行结果分三形态: - 消息为 markdown msgtype,三形态统一标题「UI自动化定时报告」,仅以 emoji 区分状态:
✅ 全部通过 / ⚠️ 存在失败用例 / 🚨 执行异常(无用例结果,如看门狗中断); ✅ 全部通过 / ⚠️ 存在失败用例 / 🚨 执行异常(无用例结果,如看门狗中断);
- 版面为分区卡片式:标题 → 任务名 → 被测系统 → 汇总统计 → 失败提示 → 耗时 → 报告链接;
不列举失败用例明细(明细看报告);
- ⚠️ 钉钉 markdown 会把单个 \n 折叠为空格,段落间必须用 \n\n 分行;
- 网络请求使用 requests + 线程池(run_in_executor),避免阻塞事件循环; - 网络请求使用 requests + 线程池(run_in_executor),避免阻塞事件循环;
- 发送失败仅记录日志,不影响定时任务调度推进。 - 发送失败仅记录日志,不影响定时任务调度推进。
...@@ -49,12 +52,6 @@ REQUEST_TIMEOUT = 10 ...@@ -49,12 +52,6 @@ REQUEST_TIMEOUT = 10
# 消息长度保护阈值(钉钉 markdown 正文上限 2048 字节,预留余量) # 消息长度保护阈值(钉钉 markdown 正文上限 2048 字节,预留余量)
MAX_MESSAGE_BYTES = 2000 MAX_MESSAGE_BYTES = 2000
# 失败用例明细最大条数(超出折叠)
MAX_FAILED_ITEMS = 5
# 单条错误摘要最大长度
MAX_ERROR_LENGTH = 80
# secret 脱敏占位符(接口返回与「保持原值」判定) # secret 脱敏占位符(接口返回与「保持原值」判定)
SECRET_MASK = "******" SECRET_MASK = "******"
...@@ -181,7 +178,7 @@ async def send_markdown( ...@@ -181,7 +178,7 @@ async def send_markdown(
# ==================== 消息构建 ==================== # ==================== 消息构建 ====================
def _flatten_error(text: str, max_length: int = MAX_ERROR_LENGTH) -> str: def _flatten_error(text: str, max_length: int = 80) -> str:
"""压平换行并截断错误摘要""" """压平换行并截断错误摘要"""
flat = " ".join(str(text or "").split()) flat = " ".join(str(text or "").split())
if len(flat) > max_length: if len(flat) > max_length:
...@@ -189,18 +186,6 @@ def _flatten_error(text: str, max_length: int = MAX_ERROR_LENGTH) -> str: ...@@ -189,18 +186,6 @@ def _flatten_error(text: str, max_length: int = MAX_ERROR_LENGTH) -> str:
return flat return flat
def _case_error_summary(case: CaseResult) -> str:
"""
用例错误摘要:用例级 error_message 优先,为空取首个失败步骤的 error
"""
if case.error_message:
return _flatten_error(case.error_message)
for step in case.steps_result or []:
if step.get("status") == "failed" and step.get("error"):
return _flatten_error(step["error"])
return "未知错误"
def _report_url(execution_id: str) -> str: def _report_url(execution_id: str) -> str:
"""拼报告链接;PLATFORM_BASE_URL 未配置时返回空串""" """拼报告链接;PLATFORM_BASE_URL 未配置时返回空串"""
base = (settings.PLATFORM_BASE_URL or "").rstrip("/") base = (settings.PLATFORM_BASE_URL or "").rstrip("/")
...@@ -225,11 +210,11 @@ def build_execution_message( ...@@ -225,11 +210,11 @@ def build_execution_message(
notify_mobiles: Optional[List[str]] = None, notify_mobiles: Optional[List[str]] = None,
) -> Tuple[str, str, List[str]]: ) -> Tuple[str, str, List[str]]:
""" """
按执行结果构建三形态消息 按执行结果构建三形态消息(分区卡片式版面,不列举失败明细)
Args: Args:
execution: Execution 记录(含 passed/failed/skipped/pass_rate/duration/status 等) execution: Execution 记录(含 passed/failed/skipped/pass_rate/duration/status 等)
failed_results (List[CaseResult]): 失败用例结果列表 failed_results (List[CaseResult]): 失败用例结果列表(仅用于统计,不展示明细)
task_name (str): 定时任务名称 task_name (str): 定时任务名称
notify_mobiles (Optional[List[str]]): 配置的被@手机号(仅失败/异常告警形态使用) notify_mobiles (Optional[List[str]]): 配置的被@手机号(仅失败/异常告警形态使用)
...@@ -242,14 +227,24 @@ def build_execution_message( ...@@ -242,14 +227,24 @@ def build_execution_message(
skipped = execution.skipped or 0 skipped = execution.skipped or 0
pass_rate = execution.pass_rate if execution.pass_rate is not None else 0.0 pass_rate = execution.pass_rate if execution.pass_rate is not None else 0.0
# 报告链接行(未配置 PLATFORM_BASE_URL 时省略) # 报告链接块(未配置 PLATFORM_BASE_URL 时省略)
link_line = ""
url = _report_url(execution.id) url = _report_url(execution.id)
if url: link_block = f"📎 [点击查看完整报告]({url})" if url else ""
link_line = f"\n\n📎 [查看完整报告]({url})"
# 被测系统行(未配置 TARGET_URL 时省略) # 被测系统块(未配置 TARGET_URL 时省略)
target_line = _target_system_line() target_block = _target_system_line()
# 耗时/完成时间块(形态 1/2 共用)
duration_line = f"⏱ 耗时 {execution.duration:.0f}s" if execution.duration else ""
finish_line = (
f"⏰ 完成 {execution.end_time.strftime('%Y-%m-%d %H:%M:%S')}"
if execution.end_time else ""
)
time_block = " | ".join(x for x in (duration_line, finish_line) if x)
def _join_blocks(blocks: List[str]) -> str:
"""块间用空行连接:钉钉 markdown 会把单个 \n 折叠为空格,必须 \n\n 才能换行"""
return "\n\n".join(b for b in blocks if b)
# ---- 形态 3:执行异常(无用例结果,如看门狗中断) ---- # ---- 形态 3:执行异常(无用例结果,如看门狗中断) ----
if total == 0 and not failed_results: if total == 0 and not failed_results:
...@@ -258,111 +253,72 @@ def build_execution_message( ...@@ -258,111 +253,72 @@ def build_execution_message(
execution.end_time.strftime("%Y-%m-%d %H:%M:%S") execution.end_time.strftime("%Y-%m-%d %H:%M:%S")
if execution.end_time else time.strftime("%Y-%m-%d %H:%M:%S") if execution.end_time else time.strftime("%Y-%m-%d %H:%M:%S")
) )
title = f"UI自动化异常 - {task_name}" blocks = [
lines = [ "🚨 **【UI自动化定时报告】**",
f"🚨 **【UI自动化异常】{task_name}**", task_name,
] target_block,
if target_line:
lines.append(target_line)
lines += [
"",
"执行未产出任何用例结果,可能被看门狗中断或执行引擎异常。", "执行未产出任何用例结果,可能被看门狗中断或执行引擎异常。",
"",
f"原因:{reason}", f"原因:{reason}",
"",
f"⏰ 时间 {finish_at}", f"⏰ 时间 {finish_at}",
link_block,
] ]
text = "\n".join(lines) + link_line return (
return title, text, list(notify_mobiles or []) f"UI自动化定时报告 - {task_name}",
_join_blocks(blocks),
# 汇总行(形态 1/2 共用) list(notify_mobiles or []),
summary_line = (
f"📊 汇总:总数 {total} / ✅ 通过 {passed} / ❌ 失败 {failed}"
f" / ⏭ 跳过 {skipped} / 通过率 {pass_rate}%"
)
duration_line = f"⏱ 耗时 {execution.duration:.0f}s" if execution.duration else ""
finish_line = (
f"⏰ 完成 {execution.end_time.strftime('%Y-%m-%d %H:%M:%S')}"
if execution.end_time else ""
) )
time_line = " | ".join(x for x in (duration_line, finish_line) if x)
# ---- 形态 2:存在失败用例(⚠️ 告警形态,@指定人) ---- # 汇总区(形态 1/2 共用):跳过为 0 时不展示,减少噪音
if failed > 0 or failed_results: pass_fail_items = [f"✅ 通过 {passed}", f"❌ 失败 {failed}"]
title = f"UI自动化告警 - {task_name}" if skipped:
lines = [ pass_fail_items.append(f"⏭ 跳过 {skipped}")
f"⚠️ **【UI自动化告警】{task_name}**", summary_blocks = [
] f"📊 汇总:总数 {total}",
if target_line: " | ".join(pass_fail_items),
lines.append(target_line) f"📈 通过率:{pass_rate}%",
lines += [
"",
summary_line,
"",
f"🔴 **失败用例({len(failed_results)}项):**",
] ]
for case in failed_results[:MAX_FAILED_ITEMS]:
lines.append(f"• {case.case_name or case.case_id or '未知用例'}:{_case_error_summary(case)}")
if len(failed_results) > MAX_FAILED_ITEMS:
lines.append(f"• ...(还有 {len(failed_results) - MAX_FAILED_ITEMS} 项,详见报告)")
if time_line:
lines.append("")
lines.append(time_line)
text = "\n".join(lines) + link_line
return title, text, list(notify_mobiles or [])
# ---- 形态 1:全部通过(✅ 报告形态,不@任何人) ---- # ---- 形态 1:全部通过(✅ 报告形态,不@任何人) ----
title = f"UI自动化报告 - {task_name}" if failed <= 0 and not failed_results:
lines = [ blocks = [
f"✅ **【UI自动化报告】{task_name}**", "✅ **【UI自动化定时报告】**",
task_name,
target_block,
*summary_blocks,
time_block,
link_block,
] ]
if target_line: return f"UI自动化定时报告 - {task_name}", _join_blocks(blocks), []
lines.append(target_line)
lines += [ # ---- 形态 2:存在失败用例(⚠️ 告警形态,@指定人,仅提示不列明细) ----
"", failed_count = failed or len(failed_results)
summary_line, blocks = [
"⚠️ **【UI自动化定时报告】**",
task_name,
target_block,
*summary_blocks,
f"❌ 存在 {failed_count} 项失败用例,请及时查看报告",
time_block,
link_block,
] ]
if time_line: return (
lines.append("") f"UI自动化定时报告 - {task_name}",
lines.append(time_line) _join_blocks(blocks),
text = "\n".join(lines) + link_line list(notify_mobiles or []),
return title, text, [] )
def _shrink_message(title: str, text: str, at_mobiles: List[str]) -> Tuple[str, List[str]]: def _shrink_message(title: str, text: str, at_mobiles: List[str]) -> Tuple[str, List[str]]:
""" """
超长保护:正文超 MAX_MESSAGE_BYTES 时逐条丢弃失败明细行, 超长保护:正文超 MAX_MESSAGE_BYTES 时按安全字符数截断并追加提示。
全部丢完仍超长则截断正文,追加折叠提示。 (失败明细已不再列举,正常不会触发;兜底防异常长的 reason 等场景)
""" """
if len(text.encode("utf-8")) <= MAX_MESSAGE_BYTES: if len(text.encode("utf-8")) <= MAX_MESSAGE_BYTES:
return text, at_mobiles return text, at_mobiles
lines = text.split("\n")
# 从后往前找失败明细行(• 开头且非折叠提示),逐条移除
for i in range(len(lines) - 1, -1, -1):
if len(text.encode("utf-8")) <= MAX_MESSAGE_BYTES:
break
if lines[i].startswith("•") and "详见报告" not in lines[i]:
lines.pop(i)
text = "\n".join(lines)
# 明细行有删减时补一行折叠提示
hint = "• (明细过长已折叠,详见报告)"
if hint not in text:
# 插到 🔴 标题行之后
for j, ln in enumerate(lines):
if ln.startswith("🔴"):
lines.insert(j + 1, hint)
break
else:
lines.append(hint)
text = "\n".join(lines)
# 仍超长:硬截断(保汇总与链接所在尾部不保)
if len(text.encode("utf-8")) > MAX_MESSAGE_BYTES:
# 按字符截断到安全长度(每字符最多 3 字节 + 余量) # 按字符截断到安全长度(每字符最多 3 字节 + 余量)
safe_chars = MAX_MESSAGE_BYTES // 3 - 20 safe_chars = MAX_MESSAGE_BYTES // 3 - 20
text = text[:safe_chars] + "\n\n(消息过长已截断,详见报告)" return text[:safe_chars] + "\n\n(消息过长已截断,详见报告)", at_mobiles
return text, at_mobiles
# ==================== 业务入口 ==================== # ==================== 业务入口 ====================
......
...@@ -6,11 +6,12 @@ ...@@ -6,11 +6,12 @@
覆盖: 覆盖:
- 加签算法 build_sign(HMAC-SHA256 + base64 + urlencode,与钉钉官方算法一致) - 加签算法 build_sign(HMAC-SHA256 + base64 + urlencode,与钉钉官方算法一致)
- 三形态消息构建(✅ 报告 / ⚠️ 告警 / 🚨 异常) - 三形态消息构建(✅ 定时报告 / ⚠️ 存在失败 / 🚨 执行异常),统一标题「UI自动化定时报告」
- 失败明细 > 5 条折叠、错误摘要截断、报告链接按 PLATFORM_BASE_URL 取舍 - 分区卡片式版面:块间 \n\n 分隔(钉钉 markdown 单 \n 折叠为空格)、不列失败明细、失败仅一句提示
- 超长消息 _shrink_message 保护(正文 < 2048 字节上限) - 错误摘要截断、报告链接按 PLATFORM_BASE_URL 取舍
- 超长消息 _shrink_message 整体截断保护(正文 < 2048 字节上限)
- secret 脱敏与「****** 保持原值」语义(get_config_masked / 服务层) - secret 脱敏与「****** 保持原值」语义(get_config_masked / 服务层)
- send_markdown 的 payload 组装(at.atMobiles 仅告警/异常形态携带) - send_markdown 的 payload 组装(at.atMobiles 仅失败/异常形态携带)
- notify_execution_result 手动发送路径(报告中心「发通知」按钮,FakeDB 模拟查询) - notify_execution_result 手动发送路径(报告中心「发通知」按钮,FakeDB 模拟查询)
作者:czj 作者:czj
...@@ -27,10 +28,8 @@ import pytest ...@@ -27,10 +28,8 @@ import pytest
from app.config import settings from app.config import settings
from app.services.dingtalk_notify_service import ( from app.services.dingtalk_notify_service import (
MAX_FAILED_ITEMS,
MAX_MESSAGE_BYTES, MAX_MESSAGE_BYTES,
SECRET_MASK, SECRET_MASK,
_case_error_summary,
_flatten_error, _flatten_error,
_report_url, _report_url,
_shrink_message, _shrink_message,
...@@ -113,18 +112,19 @@ class TestMessageForms: ...@@ -113,18 +112,19 @@ class TestMessageForms:
TASK = "每日核心回归" TASK = "每日核心回归"
def test_all_pass_form(self): def test_all_pass_form(self):
"""✅ 全部通过:无失败块、不@任何人""" """✅ 全部通过:定时报告标题、无失败提示、不@任何人"""
title, text, at_mobiles = build_execution_message( title, text, at_mobiles = build_execution_message(
make_execution(), [], self.TASK, notify_mobiles=["13800000001"] make_execution(), [], self.TASK, notify_mobiles=["13800000001"]
) )
assert title == f"UI自动化报告 - {self.TASK}" assert title == f"UI自动化定时报告 - {self.TASK}"
assert "✅" in text assert "✅" in text and "UI自动化定时报告" in text
assert "总数 26" in text and "通过率 100.0%" in text assert "总数 26" in text and "通过率:100.0%" in text
assert "失败用例" not in text
assert "🔴" not in text assert "🔴" not in text
assert at_mobiles == [] assert at_mobiles == []
def test_failure_form_with_at_mobiles(self): def test_failure_form_with_at_mobiles(self):
"""⚠️ 存在失败:告警标题 + 失败明细 + @配置手机号""" """⚠️ 存在失败:告警标题 + 失败提示 + @配置手机号"""
execution = make_execution( execution = make_execution(
total_cases=26, passed=20, failed=6, skipped=0, pass_rate=76.9 total_cases=26, passed=20, failed=6, skipped=0, pass_rate=76.9
) )
...@@ -135,22 +135,35 @@ class TestMessageForms: ...@@ -135,22 +135,35 @@ class TestMessageForms:
title, text, at_mobiles = build_execution_message( title, text, at_mobiles = build_execution_message(
execution, failed, self.TASK, notify_mobiles=["13800000001", "13900000002"] execution, failed, self.TASK, notify_mobiles=["13800000001", "13900000002"]
) )
assert title == f"UI自动化告警 - {self.TASK}" assert title == f"UI自动化定时报告 - {self.TASK}"
assert "⚠️" in text and "🔴" in text assert "⚠️" in text and "UI自动化定时报告" in text
assert "失败用例(2项)" in text assert "❌ 存在 6 项失败用例,请及时查看报告" in text
assert "用例一:Timeout 30000ms exceeded" in text
# 用例级 error_message 为空时回退首个失败步骤 error
assert "用例二:元素未找到" in text
assert at_mobiles == ["13800000001", "13900000002"] assert at_mobiles == ["13800000001", "13900000002"]
def test_failure_form_overflow_five(self): def test_failure_form_never_lists_details(self):
"""失败明细 > 5 条:仅展示 5 条 + 折叠行""" """失败用例再多也不列明细:无用例名/错误摘要/• 行,仅一句统计提示"""
execution = make_execution(total_cases=10, passed=3, failed=7, pass_rate=30.0) execution = make_execution(total_cases=10, passed=3, failed=7, pass_rate=30.0)
failed = [make_failed_case(f"失败用例{i}", error=f"错误{i}") for i in range(7)] failed = [make_failed_case(f"失败用例{i}", error=f"错误{i}") for i in range(7)]
_, text, _ = build_execution_message(execution, failed, self.TASK) _, text, _ = build_execution_message(execution, failed, self.TASK)
detail_lines = [ln for ln in text.split("\n") if ln.startswith("• 失败用例")] assert "失败用例0" not in text and "错误1" not in text
assert len(detail_lines) == MAX_FAILED_ITEMS assert "•" not in text and "🔴" not in text
assert "还有 2 项,详见报告" in text assert "❌ 存在 7 项失败用例,请及时查看报告" in text
def test_failure_hint_uses_execution_failed_first(self):
"""失败提示数:execution.failed 优先,为 0 时回退 failed_results 数"""
execution = make_execution(total_cases=10, passed=9, failed=1, pass_rate=90.0)
failed = [make_failed_case("用例A"), make_failed_case("用例B")]
_, text, _ = build_execution_message(execution, failed, self.TASK)
assert "❌ 存在 1 项失败用例" in text
def test_lines_separated_by_blank_line(self):
"""钉钉 markdown 单个 \n 会被折叠为空格:所有行必须以 \\n\\n 分隔"""
execution = make_execution()
_, text, _ = build_execution_message(execution, [], self.TASK)
body_lines = text.split("\n\n")
assert len(body_lines) >= 5 # 标题/任务名/汇总×3/时间/链接
for block in body_lines:
assert "\n" not in block # 块内不允许再出现裸 \n
def test_exception_form_zero_results(self): def test_exception_form_zero_results(self):
"""🚨 0 条用例结果(看门狗中断):异常形态 + @手机号""" """🚨 0 条用例结果(看门狗中断):异常形态 + @手机号"""
...@@ -161,8 +174,9 @@ class TestMessageForms: ...@@ -161,8 +174,9 @@ class TestMessageForms:
title, text, at_mobiles = build_execution_message( title, text, at_mobiles = build_execution_message(
execution, [], self.TASK, notify_mobiles=["13800000001"] execution, [], self.TASK, notify_mobiles=["13800000001"]
) )
assert title == f"UI自动化异常 - {self.TASK}" assert title == f"UI自动化定时报告 - {self.TASK}"
assert "🚨" in text and "看门狗" in text assert "🚨" in text and "UI自动化定时报告" in text
assert "看门狗" in text
assert at_mobiles == ["13800000001"] assert at_mobiles == ["13800000001"]
def test_link_omitted_without_base_url(self, monkeypatch): def test_link_omitted_without_base_url(self, monkeypatch):
...@@ -174,7 +188,7 @@ class TestMessageForms: ...@@ -174,7 +188,7 @@ class TestMessageForms:
def test_link_appended_with_base_url(self, monkeypatch): def test_link_appended_with_base_url(self, monkeypatch):
monkeypatch.setattr(settings, "PLATFORM_BASE_URL", "http://192.168.5.44/") monkeypatch.setattr(settings, "PLATFORM_BASE_URL", "http://192.168.5.44/")
_, text, _ = build_execution_message(make_execution(), [], self.TASK) _, text, _ = build_execution_message(make_execution(), [], self.TASK)
assert "📎 [查看完整报告](http://192.168.5.44/api/reports/generate/exec_test_001)" in text assert "📎 [点击查看完整报告](http://192.168.5.44/api/reports/generate/exec_test_001)" in text
def test_target_system_line_in_all_forms(self, monkeypatch): def test_target_system_line_in_all_forms(self, monkeypatch):
"""🌐 被测系统行:三形态均展示(取 TARGET_URL 去协议前缀)""" """🌐 被测系统行:三形态均展示(取 TARGET_URL 去协议前缀)"""
...@@ -193,7 +207,7 @@ class TestMessageForms: ...@@ -193,7 +207,7 @@ class TestMessageForms:
class TestErrorSummary: class TestErrorSummary:
"""错误摘要:压平 + 截断""" """错误摘要:压平 + 截断(仅异常形态 reason 使用)"""
def test_flatten_and_truncate(self): def test_flatten_and_truncate(self):
long_error = "line1\nline2\n" + "x" * 200 long_error = "line1\nline2\n" + "x" * 200
...@@ -202,44 +216,32 @@ class TestErrorSummary: ...@@ -202,44 +216,32 @@ class TestErrorSummary:
assert len(flat) == 80 + 1 # 截断 + 省略号 assert len(flat) == 80 + 1 # 截断 + 省略号
assert flat.endswith("…") assert flat.endswith("…")
def test_case_summary_prefers_error_message(self): def test_flatten_custom_length(self):
case = make_failed_case(error="用例级错误", steps=[{"status": "failed", "error": "步骤级错误"}]) flat = _flatten_error("错" * 100, max_length=150)
assert _case_error_summary(case) == "用例级错误" assert len(flat) == 100 # 未超 150,不截断
def test_case_summary_falls_back_to_step(self):
case = make_failed_case(error=None, steps=[{"status": "passed"}, {"status": "failed", "error": "步骤级错误"}])
assert _case_error_summary(case) == "步骤级错误"
def test_case_summary_unknown(self):
assert _case_error_summary(make_failed_case()) == "未知错误"
class TestShrinkMessage: class TestShrinkMessage:
"""超长保护""" """超长保护:整体截断兜底"""
def test_short_message_untouched(self): def test_short_message_untouched(self):
text = "短消息" text = "短消息"
shrunk, at = _shrink_message("t", text, ["138"]) shrunk, at = _shrink_message("t", text, ["138"])
assert shrunk == text and at == ["138"] assert shrunk == text and at == ["138"]
def test_overlong_message_shrinks_below_limit(self): def test_overlong_message_hard_truncated(self):
# 构造 8 条超长失败明细 → 必然超 2000 字节 # 构造超 2000 字节正文(正常版面不会触发;兜底异常长 reason 等场景)
lines = ["⚠️ **【UI自动化告警】长消息任务**", "", "📊 汇总", "", "🔴 **失败用例(8项):**"] text = "⚠️ **【UI自动化定时报告】**\n\n" + "长" * 1500
for i in range(8):
lines.append(f"• 失败用例{i}:{'错' * 120}")
lines.append("\n\n📎 [查看完整报告](http://x/api/reports/generate/e1)")
text = "\n".join(lines)
assert len(text.encode("utf-8")) > MAX_MESSAGE_BYTES assert len(text.encode("utf-8")) > MAX_MESSAGE_BYTES
shrunk, _ = _shrink_message("t", text, []) shrunk, _ = _shrink_message("t", text, [])
assert len(shrunk.encode("utf-8")) <= MAX_MESSAGE_BYTES assert len(shrunk.encode("utf-8")) <= MAX_MESSAGE_BYTES
assert "明细过长已折叠" in shrunk assert "消息过长已截断,详见报告" in shrunk
def test_extreme_message_hard_truncated(self): def test_truncation_preserves_head(self):
# 单条超巨明细(无 • 之外的可行折叠空间也须兜底截断) text = "头部标记\n\n" + "长" * 1500
text = "⚠️ 标题\n" + "• " + "错" * 3000
shrunk, _ = _shrink_message("t", text, []) shrunk, _ = _shrink_message("t", text, [])
assert len(shrunk.encode("utf-8")) <= MAX_MESSAGE_BYTES assert shrunk.startswith("头部标记")
class TestConfigMasking: class TestConfigMasking:
...@@ -381,7 +383,7 @@ class TestNotifyExecutionResult: ...@@ -381,7 +383,7 @@ class TestNotifyExecutionResult:
assert message == "Webhook 未配置" assert message == "Webhook 未配置"
async def test_manual_send_alert_form(self, monkeypatch): async def test_manual_send_alert_form(self, monkeypatch):
"""有失败用例 → ⚠️ 告警形态,标题含执行名,携带 @手机号""" """有失败用例 → ⚠️ 告警形态,统一标题,仅失败提示不列明细,携带 @手机号"""
captured = {} captured = {}
async def fake_send(webhook_url, secret, title, text, at_mobiles=None): async def fake_send(webhook_url, secret, title, text, at_mobiles=None):
...@@ -400,8 +402,10 @@ class TestNotifyExecutionResult: ...@@ -400,8 +402,10 @@ class TestNotifyExecutionResult:
ok, message = await svc.notify_execution_result("exec_test_001", "全量回归-0908", db=db) ok, message = await svc.notify_execution_result("exec_test_001", "全量回归-0908", db=db)
assert ok is True and message == "发送成功" assert ok is True and message == "发送成功"
assert captured["title"] == "UI自动化告警 - 全量回归-0908" assert captured["title"] == "UI自动化定时报告 - 全量回归-0908"
assert "⚠️" in captured["text"] and "用例A:超时" in captured["text"] assert "⚠️" in captured["text"]
assert "❌ 存在 7 项失败用例,请及时查看报告" in captured["text"]
assert "用例A" not in captured["text"] and "超时" not in captured["text"]
assert captured["at"] == ["13800000001"] assert captured["at"] == ["13800000001"]
assert captured["url"].startswith(self.WEBHOOK) assert captured["url"].startswith(self.WEBHOOK)
assert captured["secret"] == "SECabc" assert captured["secret"] == "SECabc"
...@@ -420,7 +424,7 @@ class TestNotifyExecutionResult: ...@@ -420,7 +424,7 @@ class TestNotifyExecutionResult:
ok, _ = await svc.notify_execution_result("exec_test_001", "全量回归-0908", db=db) ok, _ = await svc.notify_execution_result("exec_test_001", "全量回归-0908", db=db)
assert ok is True assert ok is True
assert captured["title"] == "UI自动化报告 - 全量回归-0908" assert captured["title"] == "UI自动化定时报告 - 全量回归-0908"
assert captured["at"] == [] assert captured["at"] == []
async def test_manual_send_dingtalk_error_propagates(self, monkeypatch): async def test_manual_send_dingtalk_error_propagates(self, monkeypatch):
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论