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