210 lines
6.8 KiB
Python
Executable File
210 lines
6.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""从 .pouch/tester/cases/*.yaml 生成给人看的 cases.csv。不要手改 CSV。
|
|
|
|
运行结果属于 runs/ 下的追加式记录，不回写本清单，避免多个工作区共同
|
|
刷新运行状态造成合并冲突。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from journey import suite as case_suite, test_type
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError: # pragma: no cover
|
|
print("需要 PyYAML: python3 -m pip install pyyaml", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
|
|
FIELDS = (
|
|
"category",
|
|
"id",
|
|
"test_type",
|
|
"suite",
|
|
"client_suite",
|
|
"platform",
|
|
"maps_to",
|
|
"title",
|
|
"what",
|
|
"tag",
|
|
"risk",
|
|
"gate",
|
|
"driver",
|
|
"automation",
|
|
"session",
|
|
"lifecycle",
|
|
"setup",
|
|
"how",
|
|
"expected",
|
|
"notes",
|
|
)
|
|
|
|
|
|
def tester_dir(root: Path) -> Path:
|
|
return root / ".pouch" / "tester"
|
|
|
|
|
|
def git_root(start: Path) -> Path | None:
|
|
cur = start.resolve()
|
|
if cur.is_file():
|
|
cur = cur.parent
|
|
for path in [cur, *cur.parents]:
|
|
if (path / ".git").exists() and (path / ".pouch" / "tester" / "cases").is_dir():
|
|
return path
|
|
return None
|
|
|
|
|
|
def resolve_root(explicit: str) -> Path:
|
|
path = Path(explicit).resolve()
|
|
if (path / ".pouch" / "tester" / "cases").is_dir():
|
|
return path
|
|
found = git_root(Path.cwd())
|
|
if found is not None:
|
|
return found
|
|
return path
|
|
|
|
|
|
def load_yaml(path: Path) -> Any:
|
|
return yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def one_line(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, list):
|
|
return " | ".join(part for part in (one_line(item) for item in value) if part)
|
|
if isinstance(value, dict):
|
|
if "kind" in value and "value" in value:
|
|
return f"{value['kind']}: {value['value']}"
|
|
return " | ".join(f"{key}={one_line(item)}" for key, item in value.items())
|
|
text = str(value).replace("\r\n", "\n").replace("\r", "\n")
|
|
return " | ".join(line.strip() for line in text.split("\n") if line.strip())
|
|
|
|
|
|
def how_brief(how: Any) -> str:
|
|
if not isinstance(how, dict):
|
|
return one_line(how)
|
|
bits: list[str] = []
|
|
if how.get("url"):
|
|
bits.append(f"url={how['url']}")
|
|
if how.get("run"):
|
|
bits.append("run: " + one_line(how["run"]))
|
|
if how.get("steps"):
|
|
bits.append(one_line(how["steps"]))
|
|
return " | ".join(bits)
|
|
|
|
|
|
def category_order(root: Path) -> list[str]:
|
|
catalog = tester_dir(root) / "catalog.yaml"
|
|
if not catalog.is_file():
|
|
return []
|
|
data = load_yaml(catalog) or {}
|
|
if not isinstance(data, dict):
|
|
return []
|
|
order: list[str] = []
|
|
for item in data.get("categories") or []:
|
|
if isinstance(item, dict) and item.get("id"):
|
|
order.append(str(item["id"]))
|
|
return order
|
|
|
|
|
|
def load_cases(root: Path) -> list[dict[str, Any]]:
|
|
cases_dir = tester_dir(root) / "cases"
|
|
if not cases_dir.is_dir():
|
|
raise SystemExit(f"找不到用例目录: {cases_dir}")
|
|
files = sorted(path for path in cases_dir.glob("*.yaml") if path.is_file())
|
|
if not files:
|
|
raise SystemExit(f"{cases_dir} 里没有 yaml")
|
|
out: list[dict[str, Any]] = []
|
|
for path in files:
|
|
data = load_yaml(path)
|
|
if not isinstance(data, dict):
|
|
raise SystemExit(f"{path} 必须是 mapping")
|
|
file_cat = data.get("category")
|
|
rows = data.get("cases") or []
|
|
if not isinstance(rows, list):
|
|
raise SystemExit(f"{path}: cases 必须是列表")
|
|
for index, case in enumerate(rows):
|
|
if not isinstance(case, dict):
|
|
raise SystemExit(f"{path}: cases[{index}] 必须是 mapping")
|
|
item = dict(case)
|
|
if not item.get("category") and file_cat:
|
|
item["category"] = file_cat
|
|
item["_file"] = str(path)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
def sort_cases(root: Path, cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
order = category_order(root)
|
|
for case in cases:
|
|
cat = str(case.get("category") or "")
|
|
if cat and cat not in order:
|
|
order.append(cat)
|
|
rank = {name: index for index, name in enumerate(order)}
|
|
return sorted(
|
|
cases,
|
|
key=lambda case: rank.get(str(case.get("category") or ""), 999),
|
|
)
|
|
|
|
|
|
def export_cases_csv(root: Path, out: Path | None = None) -> Path:
|
|
cases = sort_cases(root, load_cases(root))
|
|
path = out or (tester_dir(root) / "cases.csv")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(FIELDS), extrasaction="ignore")
|
|
writer.writeheader()
|
|
for case in cases:
|
|
cid = str(case.get("id") or "")
|
|
client_suite_value = case.get("client_suite") or (
|
|
case.get("suite") if test_type(case) == "journey" else ""
|
|
)
|
|
writer.writerow(
|
|
{
|
|
"category": case.get("category") or "",
|
|
"id": cid,
|
|
"test_type": test_type(case),
|
|
"suite": case_suite(case),
|
|
"client_suite": client_suite_value or "",
|
|
"platform": one_line(case.get("platform")),
|
|
"maps_to": case.get("maps_to") or "",
|
|
"title": case.get("title") or "",
|
|
"what": one_line(case.get("what")),
|
|
"tag": case.get("tag") or "",
|
|
"risk": case.get("risk") or "",
|
|
"gate": case.get("gate") or "",
|
|
"driver": case.get("driver") or "",
|
|
"automation": case.get("automation") or "automated",
|
|
"session": case.get("session") or "",
|
|
"lifecycle": case.get("lifecycle") or "active",
|
|
"setup": one_line(case.get("setup")),
|
|
"how": how_brief(case.get("how")),
|
|
"expected": one_line(case.get("expected")),
|
|
"notes": one_line(case.get("notes")),
|
|
}
|
|
)
|
|
return path
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="从 cases/*.yaml 生成 cases.csv")
|
|
parser.add_argument("--root", default=".", help="项目根")
|
|
parser.add_argument("--out", help="输出路径，默认 <root>/.pouch/tester/cases.csv")
|
|
args = parser.parse_args(argv)
|
|
root = resolve_root(args.root)
|
|
out = Path(args.out).resolve() if args.out else None
|
|
path = export_cases_csv(root, out)
|
|
n = sum(1 for _ in path.open(encoding="utf-8-sig")) - 1
|
|
print(f"wrote {path} rows={n}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|