cfc/docs/tdd/tools/test_attr.py

# -*- coding: utf-8 -*-
"""CFC 测试失败逐例归因（纯只读）。

用途：把 surefire XML 的红例展开为逐例明细，并按「异常类型 + 是否含生产栈帧」预分桶，
避免跨轮归因边界漂移（B-26/B-28/B-32 教训）。

用法：
  python test_attr.py --reports <cfc-backend>/target/surefire-reports --out-json attr.json

输出：
  - cases[]：class / case / kind(F|E) / type / msg / prod_frames / env / bucket
  - buckets：{bucket: 例数}
  - classes：{class: {F, E, env, buckets}}

预分桶口径（写死，跨轮一致）：
  env          环境/上下文加载失败（数据源连接类关键字）
  mockito      Mockito 误用（InvalidUseOfMatchers / UnnecessaryStubbing / WantedButNotInvoked 等）
  npe_test     NullPointer 且栈内无生产帧 → 测试装置（桩返回 null / 手工注入缺依赖）
  npe_prod     NullPointer 且栈内含生产帧 → 需研发判定
  assert       断言失败（AssertionFailedError / ComparisonFailure）→ 需人工判「文案漂移 or 契约不一致」
  other        其余（异常类型非上述）
"""
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|"
    r"Failed to instantiate.*ApplicationContext",
    re.S,
)
MOCK_TYPES = {
    "MockitoException", "InvalidUseOfMatchersException", "WantedButNotInvoked",
    "ArgumentsAreDifferent", "UnnecessaryStubbingException", "MissingMethodInvocationException",
    "PotentialStubbingProblem", "MockitoAssertionError",
}
ASSERT_TYPES = {"AssertionFailedError", "ComparisonFailure", "AssertionError"}
PROD_FRAME = re.compile(r"at (com\.etotem\.cfc\.[\w.]+)\.(\w+)\(")


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


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


def classify(typ, body, msg, env):
    if env:
        return "env"
    if typ in MOCK_TYPES:
        return "mockito"
    if typ in ("NullPointerException",):
        prod = 0
        for line in body.split("\n"):
            line = line.strip()
            if not line.startswith("at "):
                continue
            m = PROD_FRAME.search(line)
            if m and not m.group(1).split(".")[-1].endswith(("Test", "Tests", "IT")):
                prod += 1
        return "npe_prod" if prod else "npe_test"
    if typ in ASSERT_TYPES:
        return "assert"
    return "other"


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--reports", required=True)
    ap.add_argument("--out-json", required=True)
    a = ap.parse_args()

    cases, classes = [], {}
    for p in sorted(glob.glob(os.path.join(a.reports, "TEST-*.xml"))):
        txt = read(p)
        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 ""
            typ = (n.get("type") or "").split(".")[-1]
            msg = (n.get("message") or "").strip() or (body.split("\n")[0] if body else "")
            frames = [l.strip() for l in body.split("\n") if l.strip().startswith("at ")]
            prod = 0
            for f in frames:
                m = PROD_FRAME.search(f)
                if m and not m.group(1).split(".")[-1].endswith(("Test", "Tests", "IT")):
                    prod += 1
            first_frame = frames[0] if frames else ""
            # 环境红判定：类级 XML 命中环境关键字（整类上下文加载失败）
            env = bool(ENV_PAT.search(body))
            rec = {"class": cls, "case": tc.get("name"),
                   "kind": "F" if n.tag == "failure" else "E", "type": typ,
                   "msg": msg[:220], "prod_frames": prod, "first_frame": first_frame[:140],
                   "env": env, "bucket": classify(typ, body, msg, env)}
            cases.append(rec)
            c = classes.setdefault(cls, {"F": 0, "E": 0, "env": False, "cases": 0})
            c["F" if rec["kind"] == "F" else "E"] += 1
            c["env"] = c["env"] or env
            c["cases"] += 1

    buckets = {}
    for c in cases:
        buckets[c["bucket"]] = buckets.get(c["bucket"], 0) + 1
    out = {"total_red": len(cases), "buckets": buckets, "classes": classes, "cases": cases}
    with io.open(a.out_json, "w", encoding="utf-8") as f:
        f.write(json.dumps(out, ensure_ascii=False, indent=1))
    print("red_cases=%d buckets=%s classes=%d" % (len(cases), buckets, len(classes)))


if __name__ == "__main__":
    main()