laily/.pouch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update tester

Browse Source
This commit is contained in:
laily
2026-09-16 14:40:33 +08:00
parent 2783946cec
commit bdfd2ff342
8 changed files with 259 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/tester/scripts/catalog.py
+42 -6
View File
@@ -20,7 +20,11 @@ except ImportError: # pragma: no cover
raise SystemExit(2)
SURFACES = ("script", "browser", "mobile")
STATUSES = ("active", "retired")
STATUSES = ("active", "retired", "pending")
STATUS_SKIP_REASON = {
"retired": "retired",
"pending": "pending（暂缓，见 notes）",
}
SUITES = ("smoke", "full", "slow")
CANONICAL_RISKS = ("safe", "act", "harm")
RESULTS = ("pass", "fail", "environment-fail", "skipped")
@@ -86,6 +90,27 @@ def case_files(root: Path) -> list[Path]:
return sorted(p for p in d.glob("*.yaml") if p.is_file())
def category_rank(root: Path) -> dict[str, int]:
path = catalog_dir(root) / "catalog.yaml"
order: list[str] = []
if path.is_file():
data = load_yaml(path) or {}
if isinstance(data, dict):
for item in data.get("categories") or []:
if isinstance(item, dict) and item.get("id"):
order.append(str(item["id"]))
return {name: index for index, name in enumerate(order)}
def sort_by_catalog(root: Path, cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
rank = category_rank(root)
for case in cases:
cat = str(case.get("category") or "")
if cat and cat not in rank:
rank[cat] = len(rank)
return sorted(cases, key=lambda case: rank.get(str(case.get("category") or ""), 999))
def load_all_cases(root: Path) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for path in case_files(root):
@@ -108,7 +133,7 @@ def load_all_cases(root: Path) -> list[dict[str, Any]]:
item["_file_category"] = cat
item["_file_title"] = title
out.append(item)
return out
return sort_by_catalog(root, out)
def resolve_vars(root: Path, meta: dict[str, Any]) -> dict[str, str]:
@@ -189,7 +214,8 @@ def cmd_list(root: Path) -> int:
cats_meta = {c["id"]: c for c in meta.get("categories") or [] if isinstance(c, dict) and "id" in c}
active = [c for c in cases if c.get("status", "active") == "active"]
retired = [c for c in cases if c.get("status") == "retired"]
print(f"## 测试目录 active {len(active)} / retired {len(retired)}")
pending = [c for c in cases if c.get("status") == "pending"]
print(f"## 测试目录 active {len(active)} / pending {len(pending)} / retired {len(retired)}")
print()
headers = ["分类", "标题", "条数", "smoke"] + risks + list(SURFACES)
print("| " + " | ".join(headers) + " |")
@@ -222,6 +248,9 @@ def cmd_list(root: Path) -> int:
code = run_export_csv(root)
if code == 0:
print(f"清单: {catalog_dir(root).relative_to(root) / 'cases.csv'}")
run_csvs = sorted((catalog_dir(root) / "runs").glob("*.csv"))
if run_csvs:
print(f"成绩单: {run_csvs[-1].relative_to(root)}")
return code
@@ -249,7 +278,11 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
cid = str(case.get("id") or "")
if case.get("status", "active") != "active":
if cid in wanted:
skipped.append({"id": cid, "reason": "retired"})
status = str(case.get("status"))
skipped.append({
"id": cid,
"reason": STATUS_SKIP_REASON.get(status, status),
})
missing.discard(cid)
continue
if wanted:
@@ -450,7 +483,8 @@ def cmd_validate(root: Path) -> int:
return 1
cases = load_all_cases(root)
active = sum(1 for c in cases if c.get("status", "active") == "active")
print(f"✓ validate 通过 active={active} total={len(cases)}")
pending = sum(1 for c in cases if c.get("status") == "pending")
print(f"✓ validate 通过 active={active} pending={pending} total={len(cases)}")
return 0
@@ -504,7 +538,9 @@ def cmd_run_script(root: Path, case_id: str) -> int:
return 2
public_id = case.get("id")
if case.get("status") != "active":
print(json.dumps({"id": public_id, "result": "skipped", "evidence": "retired"}, ensure_ascii=False))
status = str(case.get("status"))
reason = STATUS_SKIP_REASON.get(status, status)
print(json.dumps({"id": public_id, "result": "skipped", "evidence": reason}, ensure_ascii=False))
return 0
if case.get("surface") != "script":
print(
skills/tester/scripts/generate_cases_csv.py
+35 -5
View File
@@ -32,6 +32,11 @@ FIELDS = (
"how",
"expected",
"notes",
"last_run",
"last_result",
"last_executed",
"last_evidence",
"last_notes",
)
@@ -139,25 +144,45 @@ def sort_cases(root: Path, cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
rank = {name: index for index, name in enumerate(order)}
return sorted(
cases,
key=lambda case: (
rank.get(str(case.get("category") or ""), 999),
str(case.get("id") or ""),
),
key=lambda case: rank.get(str(case.get("category") or ""), 999),
)
def last_run_lookup(root: Path) -> tuple[str, dict[str, dict[str, Any]]]:
from generate_run_csv import latest_run_yaml, load_run, results_by_id
run_path = latest_run_yaml(root)
if run_path is None:
return "", {}
try:
run = load_run(run_path)
except SystemExit:
return "", {}
try:
rel = str(run_path.relative_to(root))
except ValueError:
rel = str(run_path)
return rel, results_by_id(run)
def export_cases_csv(root: Path, out: Path | None = None) -> Path:
from generate_run_csv import executed_cell
cases = sort_cases(root, load_cases(root))
last_run, last_by_id = last_run_lookup(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 "")
last = last_by_id.get(cid) or {}
last_result = str(last.get("result") or "") if last else ""
writer.writerow(
{
"category": case.get("category") or "",
"id": case.get("id") or "",
"id": cid,
"maps_to": case.get("maps_to") or "",
"title": case.get("title") or "",
"what": one_line(case.get("what")),
@@ -172,6 +197,11 @@ def export_cases_csv(root: Path, out: Path | None = None) -> Path:
"how": how_brief(case.get("how")),
"expected": one_line(case.get("expected")),
"notes": one_line(case.get("notes")),
"last_run": last_run if last else "",
"last_result": last_result,
"last_executed": executed_cell(last_result) if last else "",
"last_evidence": one_line(last.get("evidence")) if last else "",
"last_notes": one_line(last.get("notes")) if last else "",
}
)
return path
skills/tester/scripts/generate_run_csv.py
+162
View File
@@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""从 runs/*.yaml 生成给人看的成绩单 CSV。不要手改 CSV。"""
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)
from generate_cases_csv import load_cases, one_line, resolve_root, tester_dir
RESULTS = ("pass", "fail", "environment-fail", "skipped")
EXECUTED_YES = frozenset({"pass", "fail", "environment-fail"})
FIELDS = (
"id",
"result",
"executed",
"evidence",
"notes",
"category",
"title",
"suite",
"risk",
"surface",
"maps_to",
)
def executed_cell(result: str) -> str:
if result in EXECUTED_YES:
return "yes"
if result == "skipped":
return "no"
return ""
def run_yaml_files(root: Path) -> list[Path]:
runs = tester_dir(root) / "runs"
if not runs.is_dir():
return []
return sorted(path for path in runs.glob("*.yaml") if path.is_file())
def latest_run_yaml(root: Path) -> Path | None:
files = run_yaml_files(root)
return files[-1] if files else None
def load_run(path: Path) -> dict[str, Any]:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
raise SystemExit(f"{path} 必须是 mapping")
results = data.get("results")
if not isinstance(results, list):
raise SystemExit(f"{path} 缺少 results 列表")
return data
def results_by_id(run: dict[str, Any]) -> dict[str, dict[str, Any]]:
out: dict[str, dict[str, Any]] = {}
for item in run.get("results") or []:
if not isinstance(item, dict) or not item.get("id"):
continue
out[str(item["id"])] = item
return out
def case_index(root: Path) -> dict[str, dict[str, Any]]:
out: dict[str, dict[str, Any]] = {}
for case in load_cases(root):
cid = str(case.get("id") or "")
if cid:
out[cid] = case
return out
def export_run_csv(
root: Path,
run_path: Path | None = None,
out: Path | None = None,
refresh_cases: bool = True,
) -> Path:
path = run_path or latest_run_yaml(root)
if path is None:
raise SystemExit(f"找不到运行快照: {tester_dir(root) / 'runs'}/*.yaml")
path = path.resolve()
run = load_run(path)
cases = case_index(root)
dest = out or path.with_suffix(".csv")
dest.parent.mkdir(parents=True, exist_ok=True)
with dest.open("w", encoding="utf-8-sig", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=list(FIELDS), extrasaction="ignore")
writer.writeheader()
for item in run.get("results") or []:
if not isinstance(item, dict) or not item.get("id"):
continue
cid = str(item["id"])
result = str(item.get("result") or "")
case = cases.get(cid) or {}
writer.writerow(
{
"id": cid,
"result": result,
"executed": executed_cell(result),
"evidence": one_line(item.get("evidence")),
"notes": one_line(item.get("notes")),
"category": case.get("category") or "",
"title": case.get("title") or "",
"suite": case.get("suite") or "",
"risk": case.get("risk") or "",
"surface": case.get("surface") or "",
"maps_to": case.get("maps_to") or "",
}
)
if refresh_cases:
from generate_cases_csv import export_cases_csv
export_cases_csv(root)
return dest
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="从 runs/*.yaml 生成成绩单 CSV")
parser.add_argument("--root", default=".", help="项目根")
parser.add_argument("--run", help="运行快照 yaml，默认 runs/ 里文件名最新的一份")
parser.add_argument("--out", help="输出路径，默认与 yaml 同名 .csv")
parser.add_argument(
"--no-refresh-cases",
action="store_true",
help="不要顺便刷新 cases.csv 的 last_* 列",
)
args = parser.parse_args(argv)
root = resolve_root(args.root)
run_path = Path(args.run).resolve() if args.run else None
out = Path(args.out).resolve() if args.out else None
path = export_run_csv(
root,
run_path=run_path,
out=out,
refresh_cases=not args.no_refresh_cases,
)
counts: Counter[str] = Counter()
with path.open(encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))
for row in rows:
counts[row.get("result") or ""] += 1
bits = " ".join(f"{name}={counts.get(name, 0)}" for name in RESULTS)
print(f"wrote {path} rows={len(rows)} {bits}")
return 0
if __name__ == "__main__":
raise SystemExit(main())