cfc/docs/tdd/tools/test_report_metrics.py

# -*- coding: utf-8 -*-
"""CFC 测试度量工具（纯只读）—— 聚合 surefire + 逐例归因 + 场景/用例覆盖。

用途：让每轮测试报告的数字可复现（B-15 度量脚本丢失、B-32 口径漂移）。

用法：
  # 1) 聚合 + 归因（后端 surefire 报告目录）
  python test_report_metrics.py --reports <cfc-backend>/target/surefire-reports --out-json agg.json

  # 2) 场景/用例覆盖（仓库根）
  python test_report_metrics.py --repo-root <repo> --scene

口径（写死在代码里，跨轮必须一致）：
  - 场景 = user-scenarios/*.md 中 `##`/`### 场景 <ID>：` 标题
  - TC   = test-cases/*.md 中 `| TC-xxx |` 表格行
  - 异常流 TC = 编号含 `-N<数字>` 或 类型列含 异常/边界/负向/权限
  - 场景绑定 = 场景 ID 或其映射 TC 编号出现在测试资产中
  - 环境红 = 堆栈/消息命中 ApplicationContext/JDBC 连接类关键字
  - 恒等式 = @Test − 类级@Disabled 内用例 + surefire 占位 + Σ(参数化展开−1) = 执行数
"""
import argparse
import glob
import io
import json
import os
import re
import xml.etree.ElementTree as ET

ENV_PAT = re.compile(
    r"Failed to load ApplicationContext|Communications link failure|Connection refused|"
    r"Could not create connection|Access denied for user|Unknown database|"
    r"CannotGetJdbcConnection|Unable to acquire JDBC Connection|Failed to instantiate.*ApplicationContext",
    re.S,
)
MOCK_TYPES = {"MockitoException", "InvalidUseOfMatchersException", "WantedButNotInvoked",
              "ArgumentsAreDifferent", "UnnecessaryStubbingException"}


def read(p):
    with io.open(p, encoding="utf-8", errors="ignore") as f:
        return f.read()


def xml_flag(txt, key):
    m = re.search(key + r'="(\d+)"', txt)
    return int(m.group(1)) if m else 0


def simple(name):
    return (name or "").split(".")[-1]


def collect_surefire(reports_dir):
    """XML 为主、TXT 兜底；返回 {类简单名: stats}。"""
    stats = {}
    for p in glob.glob(os.path.join(reports_dir, "TEST-*.xml")):
        txt = read(p)
        try:
            root = ET.parse(p).getroot()
        except Exception:
            continue
        cls = simple(root.get("name") or os.path.basename(p)[5:-4])
        rec = {"tests": xml_flag(txt, "tests"), "failures": xml_flag(txt, "failures"),
               "errors": xml_flag(txt, "errors"), "skipped": xml_flag(txt, "skipped"),
               "env": bool(ENV_PAT.search(txt)), "time": 0.0}
        m = re.search(r'time="([0-9.]+)"', txt)
        if m:
            rec["time"] = float(m.group(1))
        prev = stats.get(cls)
        if prev is None or rec["tests"] > prev["tests"]:
            stats[cls] = rec
        else:
            prev["env"] = prev["env"] or rec["env"]
    for p in glob.glob(os.path.join(reports_dir, "*.txt")):
        cls = simple(os.path.basename(p)[:-4])
        if cls in stats:
            continue
        txt = read(p)
        m = re.search(r"Tests run:\s*(\d+),\s*Failures:\s*(\d+),\s*Errors:\s*(\d+),\s*Skipped:\s*(\d+)", txt)
        if not m:
            continue
        stats[cls] = {"tests": int(m.group(1)), "failures": int(m.group(2)),
                      "errors": int(m.group(3)), "skipped": int(m.group(4)),
                      "env": bool(ENV_PAT.search(txt)), "time": 0.0}
    return stats


def source_classes(backend_root):
    out = set()
    for p in glob.glob(os.path.join(backend_root, "src", "test", "java", "**", "*.java"), recursive=True):
        fn = os.path.basename(p)
        if fn.endswith(("Test.java", "Tests.java", "IT.java")):
            out.add(fn[:-5])
    return out


def identity(backend_root, executed):
    """恒等式校验：base = @Test − 类级@Disabled 内用例 + 占位；delta 应为参数化净增。

    注意：@Disabled 必须紧贴类声明（注解块内）才算「类级」——方法级 @Disabled
    （如 EnergySystemFlowTest:69）会自带 1 条 skipped，不参与 base 扣减。
    """
    test_cnt = 0
    param_cnt = 0
    disabled_inner = 0
    placeholders = 0
    disabled_classes = []
    for p in glob.glob(os.path.join(backend_root, "src", "test", "java", "**", "*.java"), recursive=True):
        txt = read(p)
        n_test = len(re.findall(r"^\s*@Test\b", txt, re.M))
        test_cnt += n_test
        param_cnt += len(re.findall(r"^\s*@ParameterizedTest\b", txt, re.M))
        m = re.search(r"^\s*(?:public\s+|final\s+|abstract\s+)*class\s+(\w+)", txt, re.M)
        if m:
            head = txt[:m.start()]
            # 类声明前最后一个非注解语句之后的注解块
            block = "\n".join(head.split("\n")[-40:])
            if re.search(r"^\s*@Disabled\b", block, re.M):
                disabled_inner += n_test
                placeholders += 1
                disabled_classes.append(m.group(1))
    base = test_cnt - disabled_inner + placeholders
    return {"at_test": test_cnt, "parameterized_methods": param_cnt,
            "disabled_classes": disabled_classes,
            "disabled_inner": disabled_inner, "placeholders": placeholders,
            "base_executed": base, "actual_executed": executed,
            "param_expansion_net": executed - base}


def attribute(reports_dir):
    rows = []
    T = re.compile(r"at (com\.etotem\.cfc\.[\w.]+)\.(\w+)\(")
    for p in glob.glob(os.path.join(reports_dir, "TEST-*.xml")):
        try:
            root = ET.parse(p).getroot()
        except Exception:
            continue
        cls = simple(root.get("name"))
        for tc in root.iter("testcase"):
            nodes = list(tc.findall("failure")) + list(tc.findall("error"))
            if not nodes:
                continue
            n = nodes[0]
            body = n.text or ""
            frames = [l.strip() for l in body.split("\n") if l.strip().startswith("at ")]
            prod = []
            for f in frames:
                m = T.search(f)
                if m and not m.group(1).split(".")[-1].endswith(("Test", "Tests", "IT")):
                    prod.append(f)
            typ = (n.get("type") or "").split(".")[-1]
            rows.append({"class": cls, "case": tc.get("name"),
                         "kind": "F" if n.tag == "failure" else "E", "type": typ,
                         "msg": ((n.get("message") or "").strip() or (body.split("\n")[0] if body else ""))[:200],
                         "prod_frames": len(prod)})
    return rows


def scene_metrics(repo_root):
    scenes = []
    scene_re = re.compile(r"^#{2,3}\s*场景\s*([A-Z]{1,3}-\d+)", re.M)
    per = {}
    for p in sorted(glob.glob(os.path.join(repo_root, "tests", "user-scenarios", "*.md"))):
        ids = scene_re.findall(read(p))
        per[os.path.basename(p)] = len(ids)
        scenes += ids
    tcs = []
    for p in sorted(glob.glob(os.path.join(repo_root, "tests", "test-cases", "*.md"))):
        for line in read(p).splitlines():
            m = re.match(r"^\|\s*(TC-[A-Z]{1,4}-[0-9A-Za-z\-]+?)\s*\|", line)
            if m:
                cols = [c.strip() for c in line.strip().strip("|").split("|")]
                tcs.append({"id": m.group(1), "scene": cols[1] if len(cols) > 1 else "",
                            "type": cols[5] if len(cols) > 5 else ""})
    def is_exc(t):
        return bool(re.search(r"-N\d+", t["id"])) or any(k in t["type"] for k in ("异常", "边界", "负向", "权限"))
    mapped = set()
    for t in tcs:
        for sid in re.findall(r"[A-Z]{1,3}-\d+", t["scene"]):
            mapped.add(sid)
    # 测试资产编号索引
    toks = set()
    globs = [repo_root + r"\cfc-backend\src\test\java\**\*.java",
             repo_root + r"\cfc-frontend\__tests__\**\*.js",
             repo_root + r"\cfc-web\src\**\*.spec.js", repo_root + r"\cfc-web\src\**\*.test.js",
             repo_root + r"\tests\**\*.spec.js", repo_root + r"\cfc-langgraph\tests\**\*.py"]
    assets = 0
    for g in globs:
        for p in glob.glob(g, recursive=True):
            if "node_modules" in p:
                continue
            assets += 1
            for m in re.finditer(r"(?<![A-Za-z0-9])(TC-[A-Z0-9]{1,6}(?:-[A-Za-z0-9]{1,8}){0,4})(?![0-9A-Za-z])", read(p)):
                toks.add(m.group(1))
    spec_txt = "\n".join(read(p) for p in glob.glob(repo_root + r"\tests\**\*.spec.js", recursive=True))
    lit = [s for s in scenes
           if re.search(r"(?<![A-Za-z0-9])" + re.escape(s) + r"(?![0-9A-Za-z])", spec_txt)]
    n = len(scenes)
    exc = sum(1 for t in tcs if is_exc(t))
    return {
        "scenes": n, "per_scenario_file": per,
        "scenes_mapped_to_tc": len([s for s in scenes if s in mapped]),
        "scene_mapped_ratio": round(len([s for s in scenes if s in mapped]) * 100.0 / n, 1) if n else 0,
        "scenes_literal_in_spec": len(lit),
        "scene_spec_ratio": round(len(lit) * 100.0 / n, 1) if n else 0,
        "tc_total": len(tcs), "tc_exception": exc,
        "tc_exception_ratio": round(exc * 100.0 / len(tcs), 1) if tcs else 0,
        "tc_with_asset_anchor": len([t for t in tcs if t["id"] in toks]),
        "assets_scanned": assets,
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--reports", help="surefire-reports 目录")
    ap.add_argument("--repo-root", help="仓库根（用于恒等式与场景度量）")
    ap.add_argument("--scene", action="store_true", help="输出场景/用例覆盖")
    ap.add_argument("--out-json")
    a = ap.parse_args()

    out = {}
    if a.reports:
        st = collect_surefire(a.reports)
        tot = {"classes": len(st), "tests": 0, "failures": 0, "errors": 0, "skipped": 0, "time": 0.0}
        reds = []
        for cls, s in st.items():
            for k in ("tests", "failures", "errors", "skipped", "time"):
                tot[k] += s[k]
            if s["failures"] + s["errors"]:
                reds.append({"class": cls, "F": s["failures"], "E": s["errors"], "env": s["env"], "tests": s["tests"]})
        env_red = sum(r["F"] + r["E"] for r in reds if r["env"])
        passed = tot["tests"] - tot["failures"] - tot["errors"] - tot["skipped"]
        out["backend"] = dict(tot, passed=passed,
                              pass_rate=round(passed * 100.0 / tot["tests"], 2) if tot["tests"] else 0,
                              pass_rate_ex_env=round((passed + env_red) * 100.0 / tot["tests"], 2) if tot["tests"] else 0,
                              red_classes=len(reds), red_cases=tot["failures"] + tot["errors"],
                              env_red_cases=env_red, nonenv_red_cases=tot["failures"] + tot["errors"] - env_red,
                              time_sum_s=round(tot["time"], 1),
                              reds=sorted(reds, key=lambda x: -(x["F"] + x["E"])))
        out["attr"] = attribute(a.reports)
        if a.repo_root:
            b = os.path.join(a.repo_root, "cfc-backend")
            out["identity"] = identity(b, tot["tests"])
            out["silent_missing"] = sorted(source_classes(b) - set(st))
    if a.scene and a.repo_root:
        out["scene"] = scene_metrics(a.repo_root)

    txt = json.dumps(out, ensure_ascii=False, indent=1)
    if a.out_json:
        with io.open(a.out_json, "w", encoding="utf-8") as f:
            f.write(txt)
    b = out.get("backend")
    if b:
        print("classes=%d tests=%d pass=%d F=%d E=%d S=%d rate=%.2f%% (ex-env %.2f%%) red_classes=%d red_cases=%d"
              % (b["classes"], b["tests"], b["passed"], b["failures"], b["errors"], b["skipped"],
                 b["pass_rate"], b["pass_rate_ex_env"], b["red_classes"], b["red_cases"]))
    if "identity" in out:
        print("identity:", out["identity"])
    if "silent_missing" in out:
        print("silent_missing=%d %s" % (len(out["silent_missing"]), out["silent_missing"][:5]))
    if "scene" in out:
        print("scene:", {k: v for k, v in out["scene"].items() if k != "per_scenario_file"})


if __name__ == "__main__":
    main()