Files

476 lines
18 KiB
Python
Raw Permalink Normal View History

# -*- 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())