test(tester): validate CSV regression summary totals

2026-10-15 11:08:33 +08:00
parent 9ba9b3da0f
commit baed23d02a
4 changed files with 337 additions and 1 deletions
@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Validate a tester CSV/YAML pair and summarize statuses from CSV only."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+try:
+    import yaml
+except ImportError:  # pragma: no cover
+    print("需要 PyYAML: python3 -m pip install pyyaml", file=sys.stderr)
+    raise SystemExit(2)
+
+
+RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
+RESULT_SET = frozenset(RESULTS)
+
+
+class SummaryError(ValueError):
+    """Raised when the CSV/YAML pair cannot be safely summarized."""
+
+
+def _format_ids(ids: set[str]) -> str:
+    return ", ".join(sorted(ids)) or "(none)"
+
+
+def _duplicate_ids(ids: list[str]) -> list[str]:
+    counts = Counter(ids)
+    return sorted(case_id for case_id, count in counts.items() if count > 1)
+
+
+def load_csv(path: Path) -> list[dict[str, str]]:
+    """Load CSV rows and reject malformed IDs/results with diagnostics."""
+    try:
+        handle = path.open(encoding="utf-8-sig", newline="")
+    except OSError as exc:
+        raise SummaryError(f"无法读取 CSV {path}: {exc}") from exc
+
+    with handle:
+        reader = csv.DictReader(handle)
+        if not reader.fieldnames:
+            raise SummaryError(f"CSV {path} 缺少表头")
+        missing_fields = [field for field in ("id", "result") if field not in reader.fieldnames]
+        if missing_fields:
+            raise SummaryError(f"CSV {path} 缺少必需列: {', '.join(missing_fields)}")
+        rows = list(reader)
+
+    ids: list[str] = []
+    errors: list[str] = []
+    for line_number, row in enumerate(rows, start=2):
+        case_id = (row.get("id") or "").strip()
+        result = (row.get("result") or "").strip()
+        ids.append(case_id)
+        if not case_id:
+            errors.append(f"CSV 第 {line_number} 行 testcase id 缺失")
+        if not result:
+            errors.append(f"CSV 第 {line_number} 行 result 缺失 (testcase id={case_id or '(missing)'})")
+        elif result not in RESULT_SET:
+            errors.append(
+                f"CSV 第 {line_number} 行 result 未知: {result!r} "
+                f"(testcase id={case_id or '(missing)'}; 允许值={', '.join(RESULTS)})"
+            )
+
+    duplicates = _duplicate_ids([case_id for case_id in ids if case_id])
+    if duplicates:
+        errors.append(f"CSV testcase id 重复: {_format_ids(set(duplicates))}")
+    if errors:
+        raise SummaryError("CSV 校验失败:\n- " + "\n- ".join(errors))
+    return rows
+
+
+def load_yaml_results(path: Path) -> list[dict[str, Any]]:
+    """Load only YAML results structure needed for count/ID validation."""
+    try:
+        data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise SummaryError(f"无法读取 YAML {path}: {exc}") from exc
+    except yaml.YAMLError as exc:
+        raise SummaryError(f"YAML 解析失败 {path}: {exc}") from exc
+
+    if not isinstance(data, dict) or not isinstance(data.get("results"), list):
+        raise SummaryError(f"YAML {path} 缺少 results 列表")
+
+    results = data["results"]
+    ids: list[str] = []
+    errors: list[str] = []
+    for index, item in enumerate(results, start=1):
+        if not isinstance(item, dict):
+            errors.append(f"YAML results[{index}] 不是 mapping")
+            continue
+        case_id = str(item.get("id") or "").strip()
+        ids.append(case_id)
+        if not case_id:
+            errors.append(f"YAML results[{index}] testcase id 缺失")
+
+    duplicates = _duplicate_ids([case_id for case_id in ids if case_id])
+    if duplicates:
+        errors.append(f"YAML testcase id 重复: {_format_ids(set(duplicates))}")
+    if errors:
+        raise SummaryError("YAML 校验失败:\n- " + "\n- ".join(errors))
+    return results
+
+
+def summarize(csv_path: Path, yaml_path: Path) -> tuple[Counter[str], dict[str, int]]:
+    """Validate the pair and return counts calculated from CSV result values."""
+    rows = load_csv(csv_path)
+    yaml_results = load_yaml_results(yaml_path)
+    csv_ids = [(row.get("id") or "").strip() for row in rows]
+    yaml_ids = [str(item.get("id") or "").strip() for item in yaml_results]
+    csv_id_set = set(csv_ids)
+    yaml_id_set = set(yaml_ids)
+    diagnostics: list[str] = []
+
+    if len(rows) != len(csv_id_set):
+        diagnostics.append(
+            f"CSV 数据行总数与唯一 testcase id 数不一致: rows={len(rows)} unique_ids={len(csv_id_set)}"
+        )
+    if len(rows) != len(yaml_results):
+        diagnostics.append(
+            f"CSV 数据行总数与 YAML results 数量不一致: csv_rows={len(rows)} yaml_results={len(yaml_results)}"
+        )
+    if csv_id_set != yaml_id_set:
+        diagnostics.append(
+            "CSV/YAML testcase id 集合不一致: "
+            f"仅 CSV 有 [{_format_ids(csv_id_set - yaml_id_set)}]; "
+            f"仅 YAML 有 [{_format_ids(yaml_id_set - csv_id_set)}]"
+        )
+    if diagnostics:
+        raise SummaryError("CSV/YAML 汇总校验失败:\n- " + "\n- ".join(diagnostics))
+
+    counts: Counter[str] = Counter((row.get("result") or "").strip() for row in rows)
+    total = sum(counts[name] for name in RESULTS)
+    if total != len(rows):
+        raise SummaryError(
+            "CSV 状态计数和不等于数据行总数: "
+            f"count_sum={total} csv_rows={len(rows)}"
+        )
+    metadata = {
+        "csv_rows": len(rows),
+        "csv_unique_ids": len(csv_id_set),
+        "yaml_results": len(yaml_results),
+        "count_sum": total,
+    }
+    return counts, metadata
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="校验 tester CSV/YAML 并从 CSV result 汇总状态")
+    parser.add_argument("--csv", required=True, type=Path, help="运行成绩单 CSV")
+    parser.add_argument("--yaml", required=True, type=Path, help="对应运行快照 YAML")
+    args = parser.parse_args(argv)
+    try:
+        counts, metadata = summarize(args.csv, args.yaml)
+    except SummaryError as exc:
+        print(f"汇总失败: {exc}", file=sys.stderr)
+        return 1
+
+    bits = " ".join(f"{name}={counts[name]}" for name in RESULTS)
+    print(f"summary source=CSV.result {bits} total={metadata['count_sum']}")
+    print(
+        "validated "
+        f"csv_rows={metadata['csv_rows']} "
+        f"csv_unique_ids={metadata['csv_unique_ids']} "
+        f"yaml_results={metadata['yaml_results']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())