zhangxiang
015cb0dc78
W1-W3: 基线修复与测试基础设施搭建 - 修复 Jenkins JDK 21 兼容性，统一 E2E 目录，修复 storageState 冲突 - 搭建后端测试基类 BaseContractTest + Testcontainers PostgreSQL - 创建 TestDataFactory 链式构造，完善 Vitest 基座与 Playwright fixtures - 建立 docker-compose.test.yml 与测试数据隔离方案 W4-W5: 单元测试补齐（阶段 2） - 补齐 gym-member/gym-groupCourse/gym-checkIn/gym-payment 核心模块单元测试 - 补齐 gym-coach/manage-sys 模块单元测试 - 前端 utils/composables/stores 单元测试，37 文件 502 项测试 - JaCoCo 覆盖率门禁从 30% 调整至 55%，21 模块全部通过 W6-W7: 集成与契约测试（阶段 3） - Repository 集成测试：会员/团课/签到/支付关键表，Testcontainers 100% 通过 - Handler 集成测试：WebTestClient 覆盖正向/异常/权限路径 - 网关集成测试：JWT/RBAC/签名/限流/重试 - Flyway 迁移测试：验证迁移脚本可重复执行 - OpenAPI 契约测试：覆盖 ≥80% P0 接口，202 项契约测试 0 失败 - 跨模块契约测试：会员-支付-团课数据一致性 W8-W9: E2E 与用户旅程测试（阶段 4） - 管理员 Web 核心流程 E2E：用户/角色/菜单/字典/配置 - 小程序会员端核心页面 E2E：购卡/预约/签到 - 5 条 P0 用户旅程全链路自动化，60 条 journey 测试 0 失败 W10: 变异测试与质量门禁（阶段 5） - 后端 PIT 配置：pitest-maven 1.19.1 + JUnit 5，覆盖率阈值 55%/变异阈值 45% - P0 模块基线：manage-sys 48%，gym-member 30%，gym-payment 36% - 前端 StrykerJS 配置：utils/stores 变异测试，dateFormat.ts 70.83% - Jenkins 质量门禁：JaCoCo/PIT/E2E 统一检查，不达标阻断构建 W11: 持续运行与改进（阶段 6） - 测试指标收集脚本 scripts/collect-test-metrics.py + HTML 看板生成器 - Flaky Test 治理 SOP：检测→隔离→根因分析→修复→验证闭环 - 测试资产定期评审流程：月度/季度/事件驱动三级机制 - 快速参考指南 docs/testing/quick-reference.md - 累计 10 份测试文档，7 个里程碑全部达成
456 lines
16 KiB
Python
456 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试执行指标收集脚本
|
||
==========================
|
||
从 JaCoCo、JUnit、PIT 等报告中提取测试质量指标，输出为 JSON 格式，
|
||
支持历史趋势数据的积累与可视化。
|
||
|
||
数据源：
|
||
- JaCoCo: manage-test-report/target/site/jacoco-aggregate/jacoco.csv
|
||
- JUnit: 各模块 target/surefire-reports/*.xml
|
||
- PIT: 各模块 target/pit-reports/mutations.xml
|
||
- E2E: gym-manage-web/test-results/junit.xml
|
||
|
||
用法：
|
||
python3 scripts/collect-test-metrics.py [--output-dir <path>] [--history-dir <path>]
|
||
|
||
Cross-Validation:
|
||
- JaCoCo CSV 格式参考: https://www.jacoco.org/jacoco/trunk/doc/csv.html
|
||
- JUnit XML 格式参考: https://llg.cubic.org/docs/junit/
|
||
- PIT mutations.xml 格式参考: https://pitest.org/quickstart/mutators/
|
||
"""
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import os
|
||
import sys
|
||
import xml.etree.ElementTree as ET
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="收集测试质量指标")
|
||
parser.add_argument(
|
||
"--output-dir",
|
||
default="target/test-metrics",
|
||
help="输出目录（默认: target/test-metrics）",
|
||
)
|
||
parser.add_argument(
|
||
"--history-dir",
|
||
default="target/test-metrics/history",
|
||
help="历史数据存储目录（默认: target/test-metrics/history）",
|
||
)
|
||
parser.add_argument(
|
||
"--project-dir",
|
||
default=".",
|
||
help="项目根目录（默认: 当前目录）",
|
||
)
|
||
parser.add_argument(
|
||
"--build-number",
|
||
default=None,
|
||
help="构建编号（CI 环境传入）",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def find_surefire_reports(project_dir: str) -> List[str]:
|
||
"""查找所有 surefire 报告 XML 文件"""
|
||
reports = []
|
||
for root, dirs, files in os.walk(project_dir):
|
||
if "surefire-reports" in root:
|
||
for f in files:
|
||
if f.endswith(".xml") and not f.startswith("TEST-"):
|
||
continue
|
||
if f.endswith(".xml"):
|
||
reports.append(os.path.join(root, f))
|
||
return reports
|
||
|
||
|
||
def parse_surefire_report(filepath: str) -> Dict[str, Any]:
|
||
"""解析单个 JUnit XML 报告"""
|
||
try:
|
||
tree = ET.parse(filepath)
|
||
root = tree.getroot()
|
||
testsuite = root if root.tag == "testsuite" else root.find(".//testsuite")
|
||
if testsuite is None:
|
||
return {"file": filepath, "tests": 0, "failures": 0, "errors": 0, "skipped": 0}
|
||
|
||
return {
|
||
"file": os.path.relpath(filepath),
|
||
"tests": int(testsuite.get("tests", 0)),
|
||
"failures": int(testsuite.get("failures", 0)),
|
||
"errors": int(testsuite.get("errors", 0)),
|
||
"skipped": int(testsuite.get("skipped", 0)),
|
||
"time": float(testsuite.get("time", 0)),
|
||
}
|
||
except (ET.ParseError, FileNotFoundError) as e:
|
||
return {"file": filepath, "error": str(e)}
|
||
|
||
|
||
def parse_jacoco_csv(filepath: str) -> Dict[str, Any]:
|
||
"""
|
||
解析 JaCoCo CSV 聚合报告，提取指令覆盖率和分支覆盖率。
|
||
|
||
信源参考: JaCoCo CSV 格式 — INSTRUCTION_MISSED, INSTRUCTION_COVERED,
|
||
BRANCH_MISSED, BRANCH_COVERED, 等列。
|
||
"""
|
||
if not os.path.exists(filepath):
|
||
return {"error": f"JaCoCo report not found: {filepath}"}
|
||
|
||
metrics = {
|
||
"instruction_missed": 0,
|
||
"instruction_covered": 0,
|
||
"branch_missed": 0,
|
||
"branch_covered": 0,
|
||
"line_missed": 0,
|
||
"line_covered": 0,
|
||
"method_missed": 0,
|
||
"method_covered": 0,
|
||
"class_missed": 0,
|
||
"class_covered": 0,
|
||
}
|
||
|
||
try:
|
||
with open(filepath, "r") as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
# 跳过头部汇总行
|
||
group = row.get("GROUP", "")
|
||
if not group:
|
||
continue
|
||
metrics["instruction_missed"] += int(row.get("INSTRUCTION_MISSED", 0))
|
||
metrics["instruction_covered"] += int(row.get("INSTRUCTION_COVERED", 0))
|
||
metrics["branch_missed"] += int(row.get("BRANCH_MISSED", 0))
|
||
metrics["branch_covered"] += int(row.get("BRANCH_COVERED", 0))
|
||
metrics["line_missed"] += int(row.get("LINE_MISSED", 0))
|
||
metrics["line_covered"] += int(row.get("LINE_COVERED", 0))
|
||
metrics["method_missed"] += int(row.get("METHOD_MISSED", 0))
|
||
metrics["method_covered"] += int(row.get("METHOD_COVERED", 0))
|
||
metrics["class_missed"] += int(row.get("CLASS_MISSED", 0))
|
||
metrics["class_covered"] += int(row.get("CLASS_COVERED", 0))
|
||
except (csv.Error, IOError) as e:
|
||
return {"error": f"Failed to parse JaCoCo CSV: {e}"}
|
||
|
||
total_instruction = metrics["instruction_missed"] + metrics["instruction_covered"]
|
||
total_branch = metrics["branch_missed"] + metrics["branch_covered"]
|
||
|
||
return {
|
||
"instruction_coverage": round(
|
||
(metrics["instruction_covered"] / total_instruction * 100) if total_instruction > 0 else 0,
|
||
2,
|
||
),
|
||
"branch_coverage": round(
|
||
(metrics["branch_covered"] / total_branch * 100) if total_branch > 0 else 0,
|
||
2,
|
||
),
|
||
"line_coverage": round(
|
||
(metrics["line_covered"] / (metrics["line_missed"] + metrics["line_covered"]) * 100)
|
||
if (metrics["line_missed"] + metrics["line_covered"]) > 0
|
||
else 0,
|
||
2,
|
||
),
|
||
**metrics,
|
||
}
|
||
|
||
|
||
def parse_pit_report(filepath: str) -> Dict[str, Any]:
|
||
"""
|
||
解析 PIT mutations.xml，提取变异分数。
|
||
|
||
信源参考: PIT mutations.xml — mutationCoverage 属性包含总体变异覆盖率。
|
||
"""
|
||
if not os.path.exists(filepath):
|
||
return {"error": f"PIT report not found: {filepath}"}
|
||
|
||
try:
|
||
tree = ET.parse(filepath)
|
||
root = tree.getroot()
|
||
mutation_coverage = float(root.get("mutationCoverage", 0))
|
||
total_mutations = int(root.get("totalMutations", 0))
|
||
killed_mutations = int(root.get("killedMutations", 0))
|
||
return {
|
||
"mutation_coverage": mutation_coverage,
|
||
"total_mutations": total_mutations,
|
||
"killed_mutations": killed_mutations,
|
||
"test_strength": round(
|
||
(killed_mutations / total_mutations * 100) if total_mutations > 0 else 0,
|
||
2,
|
||
),
|
||
}
|
||
except (ET.ParseError, FileNotFoundError) as e:
|
||
return {"error": f"Failed to parse PIT report: {e}"}
|
||
|
||
|
||
def parse_e2e_report(filepath: str) -> Dict[str, Any]:
|
||
"""
|
||
解析 Playwright 输出的 JUnit XML 报告。
|
||
"""
|
||
if not os.path.exists(filepath):
|
||
return {"error": f"E2E report not found: {filepath}"}
|
||
|
||
try:
|
||
tree = ET.parse(filepath)
|
||
root = tree.getroot()
|
||
total = int(root.get("tests", 0))
|
||
failures = int(root.get("failures", 0))
|
||
errors = int(root.get("errors", 0))
|
||
skipped = int(root.get("skipped", 0))
|
||
pass_rate = round(((total - failures - errors) / total * 100) if total > 0 else 0, 2)
|
||
return {
|
||
"total": total,
|
||
"failures": failures,
|
||
"errors": errors,
|
||
"skipped": skipped,
|
||
"pass_rate": pass_rate,
|
||
}
|
||
except (ET.ParseError, FileNotFoundError) as e:
|
||
return {"error": f"Failed to parse E2E report: {e}"}
|
||
|
||
|
||
def parse_frontend_coverage(filepath: str) -> Dict[str, Any]:
|
||
"""
|
||
解析前端 Vitest 覆盖率 JSON 报告。
|
||
|
||
信源参考: @vitest/coverage-v8 JSON 输出格式。
|
||
"""
|
||
if not os.path.exists(filepath):
|
||
return {"error": f"Frontend coverage report not found: {filepath}"}
|
||
|
||
try:
|
||
with open(filepath, "r") as f:
|
||
data = json.load(f)
|
||
total = data.get("total", {})
|
||
lines = total.get("lines", {})
|
||
branches = total.get("branches", {})
|
||
functions = total.get("functions", {})
|
||
statements = total.get("statements", {})
|
||
|
||
def pct(v: Dict[str, int]) -> float:
|
||
total_count = v.get("total", 0)
|
||
covered = v.get("covered", 0)
|
||
return round((covered / total_count * 100) if total_count > 0 else 0, 2)
|
||
|
||
return {
|
||
"lines_coverage": pct(lines),
|
||
"branches_coverage": pct(branches),
|
||
"functions_coverage": pct(functions),
|
||
"statements_coverage": pct(statements),
|
||
"total_lines": lines.get("total", 0),
|
||
"covered_lines": lines.get("covered", 0),
|
||
}
|
||
except (json.JSONDecodeError, IOError) as e:
|
||
return {"error": f"Failed to parse frontend coverage: {e}"}
|
||
|
||
|
||
def collect_all_metrics(project_dir: str, build_number: Optional[str] = None) -> Dict[str, Any]:
|
||
"""收集所有测试指标"""
|
||
project_path = Path(project_dir).resolve()
|
||
timestamp = datetime.now(timezone.utc).isoformat()
|
||
|
||
metrics: Dict[str, Any] = {
|
||
"timestamp": timestamp,
|
||
"build_number": build_number or "local",
|
||
"backend": {},
|
||
"frontend": {},
|
||
"e2e": {},
|
||
"pit": {},
|
||
}
|
||
|
||
# ---- 后端 JaCoCo 覆盖率 ----
|
||
jacoco_csv = (
|
||
project_path / "gym-manage-api" / "manage-test-report" / "target" / "site" / "jacoco-aggregate" / "jacoco.csv"
|
||
)
|
||
metrics["backend"]["jacoco"] = parse_jacoco_csv(str(jacoco_csv))
|
||
|
||
# ---- 后端 JUnit 测试结果 ----
|
||
surefire_reports = find_surefire_reports(str(project_path / "gym-manage-api"))
|
||
total_tests = 0
|
||
total_failures = 0
|
||
total_errors = 0
|
||
total_skipped = 0
|
||
total_time = 0.0
|
||
|
||
for report_path in surefire_reports:
|
||
result = parse_surefire_report(report_path)
|
||
if "error" not in result:
|
||
total_tests += result["tests"]
|
||
total_failures += result["failures"]
|
||
total_errors += result["errors"]
|
||
total_skipped += result["skipped"]
|
||
total_time += result["time"]
|
||
|
||
metrics["backend"]["unit_tests"] = {
|
||
"total": total_tests,
|
||
"failures": total_failures,
|
||
"errors": total_errors,
|
||
"skipped": total_skipped,
|
||
"time_seconds": round(total_time, 2),
|
||
"pass_rate": round(
|
||
((total_tests - total_failures - total_errors) / total_tests * 100) if total_tests > 0 else 0,
|
||
2,
|
||
),
|
||
}
|
||
|
||
# ---- 后端 PIT 变异测试 ----
|
||
pit_modules = ["manage-sys", "gym-member", "gym-payment", "manage-common"]
|
||
pit_results = {}
|
||
# 信源交叉验证: 各模块 PIT mutations.xml
|
||
for module in pit_modules:
|
||
pit_path = project_path / "gym-manage-api" / module / "target" / "pit-reports" / "mutations.xml"
|
||
pit_results[module] = parse_pit_report(str(pit_path))
|
||
metrics["pit"]["modules"] = pit_results
|
||
|
||
# 计算 PIT 加权平均
|
||
pit_total_mutations = sum(
|
||
m.get("total_mutations", 0) for m in pit_results.values() if "error" not in m
|
||
)
|
||
pit_killed_mutations = sum(
|
||
m.get("killed_mutations", 0) for m in pit_results.values() if "error" not in m
|
||
)
|
||
metrics["pit"]["overall"] = {
|
||
"total_mutations": pit_total_mutations,
|
||
"killed_mutations": pit_killed_mutations,
|
||
"mutation_coverage": round(
|
||
(pit_killed_mutations / pit_total_mutations * 100) if pit_total_mutations > 0 else 0,
|
||
2,
|
||
),
|
||
}
|
||
|
||
# ---- 前端 Vitest 覆盖率 ----
|
||
frontend_coverage_json = (
|
||
project_path / "gym-manage-web" / "coverage" / "coverage-final.json"
|
||
)
|
||
metrics["frontend"]["coverage"] = parse_frontend_coverage(str(frontend_coverage_json))
|
||
|
||
# ---- 前端单元测试 ----
|
||
frontend_surefire = find_surefire_reports(str(project_path / "gym-manage-web"))
|
||
fe_total = 0
|
||
fe_failures = 0
|
||
fe_errors = 0
|
||
fe_skipped = 0
|
||
for report_path in frontend_surefire:
|
||
result = parse_surefire_report(report_path)
|
||
if "error" not in result:
|
||
fe_total += result["tests"]
|
||
fe_failures += result["failures"]
|
||
fe_errors += result["errors"]
|
||
fe_skipped += result["skipped"]
|
||
metrics["frontend"]["unit_tests"] = {
|
||
"total": fe_total,
|
||
"failures": fe_failures,
|
||
"errors": fe_errors,
|
||
"skipped": fe_skipped,
|
||
"pass_rate": round(
|
||
((fe_total - fe_failures - fe_errors) / fe_total * 100) if fe_total > 0 else 0,
|
||
2,
|
||
),
|
||
}
|
||
|
||
# ---- E2E 测试 ----
|
||
e2e_junit = project_path / "gym-manage-web" / "test-results" / "junit.xml"
|
||
metrics["e2e"] = parse_e2e_report(str(e2e_junit))
|
||
|
||
# ---- 前端 StrykerJS 变异测试 ----
|
||
stryker_report = project_path / "gym-manage-web" / "reports" / "mutation" / "mutation.json"
|
||
if stryker_report.exists():
|
||
try:
|
||
with open(stryker_report, "r") as f:
|
||
stryker_data = json.load(f)
|
||
metrics["frontend"]["mutation"] = {
|
||
"score": round(stryker_data.get("metrics", {}).get("mutationScore", 0), 2),
|
||
"killed": stryker_data.get("metrics", {}).get("killed", 0),
|
||
"survived": stryker_data.get("metrics", {}).get("survived", 0),
|
||
"total": stryker_data.get("metrics", {}).get("totalDetected", 0),
|
||
}
|
||
except (json.JSONDecodeError, IOError) as e:
|
||
metrics["frontend"]["mutation"] = {"error": str(e)}
|
||
else:
|
||
metrics["frontend"]["mutation"] = {"error": "StrykerJS report not found"}
|
||
|
||
return metrics
|
||
|
||
|
||
def load_history(history_dir: str) -> List[Dict[str, Any]]:
|
||
"""加载历史指标数据"""
|
||
history_file = os.path.join(history_dir, "metrics-history.json")
|
||
if os.path.exists(history_file):
|
||
try:
|
||
with open(history_file, "r") as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, IOError):
|
||
return []
|
||
return []
|
||
|
||
|
||
def save_history(history_dir: str, metrics: Dict[str, Any], max_entries: int = 100):
|
||
"""保存指标到历史记录"""
|
||
os.makedirs(history_dir, exist_ok=True)
|
||
history = load_history(history_dir)
|
||
history.append(metrics)
|
||
# 保留最近 N 条记录
|
||
if len(history) > max_entries:
|
||
history = history[-max_entries:]
|
||
|
||
history_file = os.path.join(history_dir, "metrics-history.json")
|
||
with open(history_file, "w") as f:
|
||
json.dump(history, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
def main():
|
||
args = parse_args()
|
||
project_dir = os.path.abspath(args.project_dir)
|
||
output_dir = os.path.join(project_dir, args.output_dir)
|
||
history_dir = os.path.join(project_dir, args.history_dir)
|
||
|
||
print(f"📊 收集测试指标...")
|
||
print(f" 项目目录: {project_dir}")
|
||
print(f" 输出目录: {output_dir}")
|
||
|
||
metrics = collect_all_metrics(project_dir, args.build_number)
|
||
|
||
# 输出当前指标
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
output_file = os.path.join(output_dir, "current-metrics.json")
|
||
with open(output_file, "w") as f:
|
||
json.dump(metrics, f, indent=2, ensure_ascii=False)
|
||
print(f" ✅ 当前指标已保存: {output_file}")
|
||
|
||
# 保存历史记录
|
||
save_history(history_dir, metrics)
|
||
print(f" ✅ 历史记录已更新: {history_dir}/metrics-history.json")
|
||
|
||
# 打印摘要
|
||
print("\n=== 测试指标摘要 ===")
|
||
jacoco = metrics["backend"].get("jacoco", {})
|
||
if "error" not in jacoco:
|
||
print(f" 后端 JaCoCo 指令覆盖率: {jacoco.get('instruction_coverage', 'N/A')}%")
|
||
print(f" 后端 JaCoCo 分支覆盖率: {jacoco.get('branch_coverage', 'N/A')}%")
|
||
|
||
ut = metrics["backend"].get("unit_tests", {})
|
||
if "error" not in ut:
|
||
print(f" 后端单元测试: {ut.get('total', 0)} 项, 通过率 {ut.get('pass_rate', 'N/A')}%")
|
||
|
||
pit = metrics["pit"].get("overall", {})
|
||
if "error" not in pit:
|
||
print(f" PIT 变异覆盖率: {pit.get('mutation_coverage', 'N/A')}%")
|
||
|
||
fe_ut = metrics["frontend"].get("unit_tests", {})
|
||
if "error" not in fe_ut:
|
||
print(f" 前端单元测试: {fe_ut.get('total', 0)} 项, 通过率 {fe_ut.get('pass_rate', 'N/A')}%")
|
||
|
||
fe_cov = metrics["frontend"].get("coverage", {})
|
||
if "error" not in fe_cov:
|
||
print(f" 前端语句覆盖率: {fe_cov.get('statements_coverage', 'N/A')}%")
|
||
|
||
e2e = metrics.get("e2e", {})
|
||
if "error" not in e2e:
|
||
print(f" E2E 测试: {e2e.get('total', 0)} 项, 通过率 {e2e.get('pass_rate', 'N/A')}%")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |