Yb-QaAutomationHub/output/compute_coverage.py

# -*- coding: utf-8 -*-
"""确定性覆盖率追溯计算：需求 REQ -> 测试点 TP -> 测试用例。

用法: python output/compute_coverage.py
输出: 控制台统计 + output/logs/coverage_metrics.json
"""
from __future__ import annotations

import json
import re
import sys
from collections import defaultdict
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO / "scripts"))
sys.stdout.reconfigure(encoding="utf-8")

from export_excel import load_markdown_table  # noqa: E402

BASE = "物流公司管理需求"
CASES = REPO / "output" / "test_cases" / f"{BASE}_测试用例.md"
POINTS = REPO / "output" / "test_points" / f"{BASE}_测试点.md"
ANALYSIS = REPO / "output" / "analysis" / f"{BASE}_分析.md"

TP_RE = re.compile(r"TP-[A-Z]-\d{3}")
REQ_RE = re.compile(r"REQ-\d{3}")
RISK_RE = re.compile(r"RISK-[A-Z]+")

# ---------- 1. 用例 ----------
headers, rows = load_markdown_table(CASES)
idx = {name: headers.index(name) for name in headers}
cases = []
for row in rows:
    cases.append({
        "id": row[idx["用例编号"]].strip(),
        "module": row[idx["模块"]].strip(),
        "priority": row[idx["优先级"]].strip(),
        "type": row[idx["类型"]].strip(),
        "note": row[idx["备注"]],
    })

tp_to_cases: dict[str, list[str]] = defaultdict(list)
case_tps: dict[str, list[str]] = {}
no_tp_cases: list[str] = []
risks_in_cases: dict[str, list[str]] = defaultdict(list)

for case in cases:
    tps = sorted(set(TP_RE.findall(case["note"])))
    case_tps[case["id"]] = tps
    if not tps:
        no_tp_cases.append(case["id"])
    for tp in tps:
        tp_to_cases[tp].append(case["id"])
    for risk in sorted(set(RISK_RE.findall(case["note"]))):
        risks_in_cases[risk].append(case["id"])

# ---------- 2. 测试点 ----------
point_text = POINTS.read_text(encoding="utf-8")
blocks: dict[str, dict] = {}
current: str | None = None
for line in point_text.splitlines():
    m = re.match(r"^### (TP-[A-Z]-\d{3}):\s*(.*)$", line)
    if m:
        current = m.group(1)
        blocks[current] = {"title": m.group(2).strip(), "priority": "", "reqs": []}
        continue
    if current is None:
        continue
    pm = re.match(r"^- 优先级:\s*(P\d)", line)
    if pm:
        blocks[current]["priority"] = pm.group(1)
    blocks[current]["reqs"].extend(REQ_RE.findall(line))

tp_priority = {tp: blk["priority"] for tp, blk in blocks.items()}
tp_reqs = {tp: sorted(set(blk["reqs"])) for tp, blk in blocks.items()}

# ---------- 3. 需求条目 ----------
analysis_text = ANALYSIS.read_text(encoding="utf-8")
req_ids: list[str] = []
for line in analysis_text.splitlines():
    m = re.match(r"^\|\s*(REQ-\d{3})\s*\|", line)
    if m and m.group(1) not in req_ids:
        req_ids.append(m.group(1))

# ---------- 4. 统计 ----------
total_tp = len(blocks)
covered_tp = [tp for tp in blocks if tp_to_cases.get(tp)]
p0_tp = [tp for tp, p in tp_priority.items() if p == "P0"]
p1_tp = [tp for tp, p in tp_priority.items() if p == "P1"]
p0_covered = [tp for tp in p0_tp if tp_to_cases.get(tp)]
p1_covered = [tp for tp in p1_tp if tp_to_cases.get(tp)]

req_to_tp: dict[str, list[str]] = defaultdict(list)
for tp, reqs in tp_reqs.items():
    for req in reqs:
        req_to_tp[req].append(tp)
req_without_tp = [r for r in req_ids if not req_to_tp.get(r)]

# 反向：用例 -> 需求
req_to_cases: dict[str, list[str]] = defaultdict(list)
for tp, cids in tp_to_cases.items():
    for req in tp_reqs.get(tp, []):
        req_to_cases[req].extend(cids)
for req in req_to_cases:
    req_to_cases[req] = sorted(set(req_to_cases[req]))

metrics = {
    "case_total": len(cases),
    "tp_total": total_tp,
    "tp_covered": len(covered_tp),
    "tp_coverage": round(len(covered_tp) / total_tp * 100, 2) if total_tp else 0,
    "p0_tp_total": len(p0_tp),
    "p0_tp_covered": len(p0_covered),
    "p0_tp_coverage": round(len(p0_covered) / len(p0_tp) * 100, 2) if p0_tp else 100,
    "p1_tp_total": len(p1_tp),
    "p1_tp_covered": len(p1_covered),
    "p1_tp_coverage": round(len(p1_covered) / len(p1_tp) * 100, 2) if p1_tp else 100,
    "req_total": len(req_ids),
    "req_with_tp": len(req_ids) - len(req_without_tp),
    "req_coverage": round((len(req_ids) - len(req_without_tp)) / len(req_ids) * 100, 2) if req_ids else 0,
    "req_without_tp": req_without_tp,
    "uncovered_tps": [tp for tp in blocks if not tp_to_cases.get(tp)],
    "no_tp_cases": no_tp_cases,
    "risk_coverage": {k: sorted(set(v)) for k, v in risks_in_cases.items()},
    "tp_to_cases": {tp: sorted(set(v)) for tp, v in tp_to_cases.items()},
    "req_to_tp": {k: sorted(set(v)) for k, v in req_to_tp.items()},
    "req_to_cases": req_to_cases,
    "tp_priority": tp_priority,
    "module_case_count": {},
}
mod_count: dict[str, int] = defaultdict(int)
for case in cases:
    mod_count[case["module"]] += 1
metrics["module_case_count"] = dict(sorted(mod_count.items()))

out = REPO / "output" / "logs" / "coverage_metrics.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(metrics, ensure_ascii=False, indent=2), encoding="utf-8")

print(f"用例总数: {metrics['case_total']}")
print(f"测试点总数: {total_tp}（P0={len(p0_tp)} P1={len(p1_tp)}）")
print(f"测试点->用例覆盖率: {metrics['tp_coverage']}% ({len(covered_tp)}/{total_tp})")
print(f"P0 测试点->用例覆盖率: {metrics['p0_tp_coverage']}% ({len(p0_covered)}/{len(p0_tp)})")
print(f"P1 测试点->用例覆盖率: {metrics['p1_tp_coverage']}% ({len(p1_covered)}/{len(p1_tp)})")
print(f"需求->测试点覆盖率: {metrics['req_coverage']}% ({metrics['req_with_tp']}/{len(req_ids)})")
print(f"无测试点覆盖的需求条目: {req_without_tp or '无'}")
print(f"无用例覆盖的测试点: {metrics['uncovered_tps'] or '无'}")
print(f"无 TP 追溯号的用例: {no_tp_cases or '无'}")
print(f"风险覆盖: " + "; ".join(f"{k}={len(set(v))}条用例" for k, v in sorted(risks_in_cases.items())))
print(f"已写出: {out}")