Simon
476 lines
18 KiB
Python
476 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
# =============================================================================
|
||
# IT 智能服务台 — 自动化测试套件全景生成器
|
||
# =============================================================================
|
||
# 用途:
|
||
# 1. 扫描 src/backend/tests/ 下所有 test_*.py 文件（含子目录）
|
||
# 2. 调用 pytest --collect-only 统计每个文件的用例数
|
||
# 3. 按主题分类（基于文件名关键字 + 目录前缀）
|
||
# 4. 输出 Markdown 表格到 docs/03-测试文档/00-测试规范/测试套件全景.md
|
||
#
|
||
# 使用方式:
|
||
# cd src/backend
|
||
# python ../../scripts/test_inventory.py
|
||
# python ../../scripts/test_inventory.py --dry-run # 仅打印不写文件
|
||
# python ../../scripts/test_inventory.py --baseline 1311/88/4 # 填入最近一次 pytest 跑分
|
||
#
|
||
# 关联文档:
|
||
# - 测试方法论指南: docs/03-测试文档/00-测试规范/测试方法论指南.md
|
||
# - 测试套件全景: docs/03-测试文档/00-测试规范/测试套件全景.md
|
||
# - 整改记录 #12: docs/04-运维文档/部署运维/00-文档规范化整改记录.md
|
||
# =============================================================================
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
from typing import Dict, List, Tuple
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# 主题分类器（基于文件名关键字）
|
||
# -----------------------------------------------------------------------------
|
||
TOPIC_RULES: List[Tuple[str, List[str]]] = [
|
||
("🔐 鉴权 / 认证 / RBAC", [
|
||
r"auth", r"otp", r"mfa", r"rbac", r"admin_user", r"admin_ip",
|
||
r"sensitive_words_auth", r"high_risk_guard", r"token_anomaly",
|
||
]),
|
||
("🤖 自动化 / 工作流", [
|
||
r"automation", r"global_intent", r"information_item",
|
||
r"session_manager", r"timeout_cleaner",
|
||
]),
|
||
("📋 审批工作流", [
|
||
r"approval", r"asset_approval",
|
||
]),
|
||
("💬 消息 / 路由 / 会话", [
|
||
r"message_", r"messages_uuid", r"nontext_message", r"voice_asr",
|
||
r"routing_service", r"ai_reply",
|
||
]),
|
||
("👥 会话协作 / 邀请", [
|
||
r"conversation", r"collaboration", r"employee_", r"invite_",
|
||
]),
|
||
("🧠 知识 / RAG", [
|
||
r"knowledge", r"bugfix_ki", r"confidence_gate", r"topic_detector",
|
||
]),
|
||
("🛡️ 内容审核", [
|
||
r"content_moderation",
|
||
]),
|
||
("🎯 诊断 / 推荐 / 评分 / Wingman", [
|
||
r"triage", r"recommend", r"scoring", r"wingman",
|
||
r"match_keywords", r"evaluation",
|
||
]),
|
||
("🏢 企微 / H5 集成", [
|
||
r"wecom_crypto", r"h5_",
|
||
]),
|
||
("🖥️ 资产 / 基础 / 工具", [
|
||
r"env_gating", r"exclusion", r"p2_p3", r"response_contract",
|
||
r"api_basic", r"backend_observer", r"avatar_service",
|
||
r"meetingroom", r"org_tree", r"neo4j_client", r"todo_integration",
|
||
r"byod",
|
||
]),
|
||
("🔌 WS / 实时推送", [
|
||
r"ws_",
|
||
]),
|
||
("🧪 Tier 1 API / 通用", [
|
||
r"tier1_api", r"agents",
|
||
]),
|
||
("⚠️ 预存在 collection 错误", [
|
||
r"approval_detect_intent", r"byod",
|
||
]),
|
||
]
|
||
|
||
TOPIC_ORDER = [t[0] for t in TOPIC_RULES]
|
||
|
||
|
||
def classify(filename: str) -> str:
|
||
"""根据文件名关键字匹配主题。"""
|
||
name = filename.lower()
|
||
if any(re.search(p, name) for p in TOPIC_RULES[-1][1]):
|
||
return TOPIC_RULES[-1][0]
|
||
for topic, patterns in TOPIC_RULES[:-1]:
|
||
if any(re.search(p, name) for p in patterns):
|
||
return topic
|
||
return "🗂️ 其他 / 未分类"
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# pytest --collect-only 解析
|
||
# -----------------------------------------------------------------------------
|
||
PYTEST_COLLECT_CMD = [
|
||
sys.executable, "-m", "pytest",
|
||
"tests/",
|
||
"--collect-only",
|
||
"-q",
|
||
"--no-header",
|
||
]
|
||
|
||
|
||
def collect_per_file(backend_dir: Path, ignore_files: List[str]) -> Dict[str, int]:
|
||
"""扫描 tests/ 下所有 test_*.py（含子目录），用 pytest --collect-only 统计用例数。
|
||
|
||
返回 key 格式：
|
||
- tests/test_foo.py → "test_foo"
|
||
- tests/automation/test_foo.py → "automation/test_foo"
|
||
"""
|
||
counts: Dict[str, int] = defaultdict(int)
|
||
|
||
# 1. 找到所有测试文件（含子目录）
|
||
test_files: List[Path] = []
|
||
tests_dir = backend_dir / "tests"
|
||
for tf in tests_dir.glob("test_*.py"):
|
||
test_files.append(tf)
|
||
for sub in tests_dir.iterdir():
|
||
if sub.is_dir() and (sub / "__init__.py").exists():
|
||
for tf in sub.glob("test_*.py"):
|
||
test_files.append(tf)
|
||
|
||
# 2. 构造精确 pytest 命令（每个文件单独 collect，避开 collection 错误传染）
|
||
for tf in sorted(test_files):
|
||
rel = tf.relative_to(tests_dir).with_suffix("") # e.g. "test_foo" 或 "automation/test_foo"
|
||
key = str(rel).replace("\\", "/")
|
||
stem = tf.stem
|
||
if stem in ignore_files and "/" not in key:
|
||
counts[key] = 0
|
||
continue
|
||
cmd = [
|
||
sys.executable, "-m", "pytest",
|
||
str(tf.relative_to(backend_dir)),
|
||
"--collect-only", "-q", "--no-header",
|
||
]
|
||
try:
|
||
r = subprocess.run(
|
||
cmd, cwd=backend_dir, capture_output=True, text=True,
|
||
encoding="utf-8", timeout=60,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
counts[key] = -1
|
||
continue
|
||
# 解析 "X tests collected" 或 "X test collected"
|
||
text = r.stdout + r.stderr
|
||
m = re.search(r"(\d+)\s+tests?\s+collected", text)
|
||
if m:
|
||
counts[key] = int(m.group(1))
|
||
else:
|
||
counts[key] = 0 # collection 错误 → 0
|
||
|
||
# 3. 标 ignore 文件为 0
|
||
for f in ignore_files:
|
||
if f not in counts:
|
||
counts[f] = 0
|
||
return dict(counts)
|
||
|
||
|
||
def collect_errors(backend_dir: Path) -> List[Dict[str, str]]:
|
||
"""收集 collection 错误（import 损坏等）。"""
|
||
result = subprocess.run(
|
||
PYTEST_COLLECT_CMD,
|
||
cwd=backend_dir,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
)
|
||
errors = []
|
||
for line in (result.stdout + result.stderr).splitlines():
|
||
if "ERROR collecting" in line or "ImportError" in line or "ModuleNotFoundError" in line:
|
||
errors.append({"line": line.strip()})
|
||
return errors
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# 文件级说明（人工维护）
|
||
# -----------------------------------------------------------------------------
|
||
NOTE_BY_FILE: Dict[str, str] = {
|
||
# 鉴权
|
||
"test_admin_ip_whitelist": "后台 IP 白名单",
|
||
"test_admin_user": "后台管理员账号",
|
||
"test_agents_auth": "坐席鉴权",
|
||
"test_rbac_verification": "角色权限矩阵",
|
||
"test_auth_qrcode": "扫码登录",
|
||
"test_auth_unified": "统一认证（AUTH-03）",
|
||
"test_mfa": "MFA 旧路径（已迁移到 auth-otp）",
|
||
"test_otp_bind_flow": "OTP 绑定流程",
|
||
"test_otp_unified": "OTP 统一认证",
|
||
"test_token_anomaly": "Token 异常检测",
|
||
"test_sensitive_words_auth": "**本次新增**（BUG-通用-004 鉴权补漏回归）",
|
||
"test_high_risk_guard": "高危操作 OTP 守卫",
|
||
# 自动化
|
||
"test_automation_executor": "自动化执行器",
|
||
"test_automation_intent_router": "意图路由器",
|
||
"test_automation_services": "自动化服务层",
|
||
"test_automation_session_manager": "会话管理器",
|
||
"test_automation_unit": "单元测试",
|
||
"test_automation_approval": "自动化审批",
|
||
# automation 子目录
|
||
"automation/test_global_intent_router": "全局意图路由（子目录）",
|
||
"automation/test_information_item_service": "信息项服务（子目录）",
|
||
"automation/test_session_manager_pause_resume": "会话暂停恢复（子目录）",
|
||
"automation/test_timeout_cleaner": "超时清理（子目录）",
|
||
# 审批
|
||
"test_approval_state_machine": "审批状态机",
|
||
"test_approval_webhook": "审批 Webhook",
|
||
"test_asset_approval_urge": "资产审批催办",
|
||
"test_approval_detect_intent": "⚠️ collection 错（import 损坏，预存在）",
|
||
# 消息
|
||
"test_message_router": "消息路由器",
|
||
"test_message_dedup": "消息去重",
|
||
"test_message_experience": "体验优化",
|
||
"test_message_id_type_bug": "UUID 类型 bug 回归",
|
||
"test_messages_uuid": "UUID 消息",
|
||
"test_nontext_message": "非文本消息（图片/语音）",
|
||
"test_voice_asr": "语音转文字（百度 ASR）",
|
||
"test_routing_service": "路由服务（含 broken link）",
|
||
"test_ai_reply_gate": "AI 回复闸门",
|
||
"test_ai_reply_mode_api": "AI 回复模式 API",
|
||
# 协作
|
||
"test_conversation_grab": "会话抢占（摇人）",
|
||
"test_conversations": "会话管理",
|
||
"test_collaboration": "协作功能",
|
||
"test_employee_history_messages": "员工历史消息",
|
||
"test_employee_profile_service": "员工档案",
|
||
"test_invite_link": "邀请链接",
|
||
"test_invite_participant": "邀请参与者",
|
||
"test_invite_status": "邀请状态",
|
||
# 知识
|
||
"test_knowledge_iteration": "知识迭代",
|
||
"test_bugfix_ki_suggestions": "KI 建议 bug 修复回归",
|
||
"test_confidence_gate": "置信度闸门",
|
||
"test_topic_detector": "主题检测",
|
||
# 审核
|
||
"test_content_moderation": "敏感词命中（中台核心）",
|
||
# 诊断
|
||
"test_triage": "分诊",
|
||
"test_recommend_progress_service": "推荐进度",
|
||
"test_scoring_service": "评分服务",
|
||
"test_wingman": "Wingman 模块",
|
||
"test_wingman_service": "Wingman 服务层",
|
||
"test_match_keywords_email": "关键词匹配",
|
||
"test_evaluation": "评估",
|
||
# 企微
|
||
"test_wecom_crypto": "企微加解密",
|
||
"test_h5_oauth": "H5 OAuth",
|
||
"test_h5_shake": "H5 摇一摇",
|
||
"test_h5_asset_pipeline": "H5 资产推送管道",
|
||
"test_h5_mask_option_select": "H5 蒙层/选项/选择",
|
||
# 基础
|
||
"test_env_gating": "环境隔离",
|
||
"test_exclusion": "排除规则",
|
||
"test_p2_p3": "P2/P3 优先级",
|
||
"test_response_contract": "响应契约",
|
||
"test_api_basic": "API 基础冒烟",
|
||
"test_backend_observer": "后端观察者",
|
||
"test_avatar_service": "头像服务",
|
||
"test_meetingroom": "会议室预定",
|
||
"test_org_tree": "组织架构树",
|
||
"test_neo4j_client": "Neo4j 客户端",
|
||
"test_todo_integration": "待办集成",
|
||
"test_byod": "⚠️ collection 错（预存在）",
|
||
# WS
|
||
"test_ws_endpoints": "WebSocket 端点",
|
||
"test_ws_push_to_employee": "WS 推送给员工",
|
||
# Tier 1
|
||
"test_tier1_api": "Tier 1 API 通用",
|
||
"test_agents": "坐席基础",
|
||
# 快速回复
|
||
"test_quick_rules": "⚠️ collection 错（预存在）",
|
||
}
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Markdown 渲染
|
||
# -----------------------------------------------------------------------------
|
||
def render_markdown(
|
||
file_case_counts: Dict[str, int],
|
||
topics: Dict[str, List[Tuple[str, int]]],
|
||
errors: List[Dict[str, str]],
|
||
total_pct: float,
|
||
passed: int,
|
||
failed: int,
|
||
xfailed: int,
|
||
) -> str:
|
||
"""生成测试套件全景.md 的 Markdown 内容。"""
|
||
total_files = len(file_case_counts)
|
||
total_tests = sum(c for c in file_case_counts.values() if c > 0)
|
||
error_files = sum(1 for c in file_case_counts.values() if c == 0)
|
||
|
||
lines = [
|
||
"# 自动化测试套件全景（src/backend/tests/）",
|
||
"",
|
||
"> **版本**: v1.0 | **生效日期**: 2026-08-05 | **维护人**: Duckula",
|
||
"> 自动生成工具：`scripts/test_inventory.py`",
|
||
"> （手动重跑：`cd src/backend && python ../../scripts/test_inventory.py --baseline 1311/88/4`）",
|
||
"",
|
||
"本索引按 pytest 测试文件列出后端测试套件全貌，**与 `docs/03-测试文档/04-版本测试报告/` 互补**：",
|
||
"- 文档报告（TC/TR/BUG）：按需求/版本/缺陷分类",
|
||
"- 本文档（测试套件全景）：按 pytest 文件 + 主题分类",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 📊 总览数据",
|
||
"",
|
||
f"- **测试文件总数**: {total_files}（含子目录）",
|
||
f"- **测试用例总数**: {total_tests}",
|
||
f"- **预存在 collection 错误**: {error_files} 个文件（已忽略，不计入用例）",
|
||
f"- **最近一次 baseline**: {passed} passed / {failed} failed / {xfailed} xfailed (≈ {total_pct:.1f}%)",
|
||
f"- **关联规范**: [`测试方法论指南.md`](./测试方法论指南.md)（Tier 0 / Tier 1 / Tier 2 / E2E 四层）",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 📁 按主题分类（pytest 文件 → 主题）",
|
||
"",
|
||
]
|
||
|
||
for topic in TOPIC_ORDER:
|
||
if topic not in topics or not topics[topic]:
|
||
continue
|
||
items = sorted(topics[topic], key=lambda x: -x[1])
|
||
valid_count = sum(c for _, c in items if c > 0)
|
||
err_count = sum(1 for _, c in items if c == 0)
|
||
lines.append(f"### {topic}（{len(items)} 文件 / {valid_count} 用例" + (f" / {err_count} collection 错" if err_count else "") + "）")
|
||
lines.append("")
|
||
lines.append("| 测试文件 | 用例数 | 说明 |")
|
||
lines.append("|----------|--------|------|")
|
||
for key, count in items:
|
||
note = NOTE_BY_FILE.get(key, "")
|
||
if count == 0:
|
||
lines.append(f"| `tests/{key}.py` | ⚠️ 0 | {note or 'collection 错误'} |")
|
||
else:
|
||
lines.append(f"| `tests/{key}.py` | {count} | {note} |")
|
||
lines.append("")
|
||
|
||
# 预存在错误
|
||
if errors:
|
||
lines.append("---")
|
||
lines.append("")
|
||
lines.append("## ⚠️ 预存在 collection 错误（已知，跳过）")
|
||
lines.append("")
|
||
lines.append("| 错误信息 |")
|
||
lines.append("|----------|")
|
||
for e in errors[:10]:
|
||
lines.append(f"| `{e['line'][:120]}` |")
|
||
lines.append("")
|
||
|
||
lines.extend([
|
||
"---",
|
||
"",
|
||
"## 🚀 常用命令",
|
||
"",
|
||
"```bash",
|
||
"cd src/backend",
|
||
"",
|
||
"# 跑全部有效套件（跳过 3 个预损坏）",
|
||
"python -m pytest tests/ -q \\",
|
||
" --ignore=tests/test_approval_detect_intent.py \\",
|
||
" --ignore=tests/test_byod.py \\",
|
||
" --ignore=tests/test_quick_rules.py",
|
||
"",
|
||
"# 跑指定主题（按文件名匹配）",
|
||
"python -m pytest tests/test_sensitive_words_auth.py tests/test_content_moderation.py -v",
|
||
"",
|
||
"# 跑本次新增的鉴权回归（最关键）",
|
||
"python -m pytest tests/test_sensitive_words_auth.py -v",
|
||
"",
|
||
"# 重新生成本文档",
|
||
"python ../../scripts/test_inventory.py --baseline 1311/88/4",
|
||
"```",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 🔗 关联文档",
|
||
"",
|
||
"- 测试方法论指南: `./测试方法论指南.md`",
|
||
"- 测试分类索引: `../README.md`",
|
||
"- TC（功能测试用例）: `../03-功能测试用例/`",
|
||
"- TR（版本测试报告）: `../04-版本测试报告/`",
|
||
"- BUG 单: `../05-缺陷单/`",
|
||
"- 整改记录 #12: `../../04-运维文档/部署运维/00-文档规范化整改记录.md`",
|
||
"",
|
||
"---",
|
||
"",
|
||
f"> 最后更新：2026-08-05 | 自动生成 baseline：{passed}/{total_tests} 通过",
|
||
])
|
||
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Main
|
||
# -----------------------------------------------------------------------------
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="生成测试套件全景 Markdown")
|
||
parser.add_argument(
|
||
"--backend-dir",
|
||
default="src/backend",
|
||
help="backend 目录相对路径（默认 src/backend）",
|
||
)
|
||
parser.add_argument(
|
||
"--output",
|
||
default="docs/03-测试文档/00-测试规范/测试套件全景.md",
|
||
help="输出 Markdown 文件路径",
|
||
)
|
||
parser.add_argument(
|
||
"--ignore",
|
||
nargs="*",
|
||
default=["test_approval_detect_intent", "test_byod", "test_quick_rules"],
|
||
help="预存在 collection 错误，跳过 collect",
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="仅打印不写文件",
|
||
)
|
||
parser.add_argument(
|
||
"--baseline",
|
||
type=str,
|
||
default=None,
|
||
help="最近一次 pytest 跑分（'passed/failed/xfailed'，如 '1311/88/4'）",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
repo_root = Path(__file__).resolve().parent.parent
|
||
backend_dir = repo_root / args.backend_dir
|
||
output_path = repo_root / args.output
|
||
|
||
if not backend_dir.exists():
|
||
print(f"❌ backend 目录不存在: {backend_dir}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(f"🔍 扫描 {backend_dir}/tests/ ...", file=sys.stderr)
|
||
counts = collect_per_file(backend_dir, args.ignore)
|
||
print(f" 找到 {len(counts)} 个测试文件（耗时取决于 collection 速度）", file=sys.stderr)
|
||
|
||
topics: Dict[str, List[Tuple[str, int]]] = defaultdict(list)
|
||
for key, count in sorted(counts.items()):
|
||
topic = classify(key)
|
||
topics[topic].append((key, count))
|
||
|
||
errors = collect_errors(backend_dir)
|
||
|
||
if args.baseline:
|
||
parts = args.baseline.split("/")
|
||
passed, failed, xfailed = int(parts[0]), int(parts[1]), int(parts[2])
|
||
else:
|
||
passed, failed, xfailed = 0, 0, 0
|
||
total = passed + failed + xfailed
|
||
total_pct = 100 * passed / total if total else 0
|
||
|
||
markdown = render_markdown(counts, topics, errors, total_pct, passed, failed, xfailed)
|
||
|
||
if args.dry_run:
|
||
print(markdown)
|
||
return 0
|
||
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
output_path.write_text(markdown, encoding="utf-8")
|
||
valid_total = sum(c for c in counts.values() if c > 0)
|
||
print(f"✅ 已写入 {output_path}", file=sys.stderr)
|
||
print(f" {len(counts)} 文件 / {valid_total} 用例 / {sum(1 for c in counts.values() if c == 0)} collection 错", file=sys.stderr)
|
||
print(f" baseline: {passed} passed / {failed} failed / {xfailed} xfailed", file=sys.stderr)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|