176 lines
6.4 KiB
Python
176 lines
6.4 KiB
Python
#!/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())
|