From bdfd2ff342d98bbbb5809a7cdbdbb1eb2a7c5322 Mon Sep 17 00:00:00 2001 From: laily Date: Wed, 16 Sep 2026 14:40:33 +0800 Subject: [PATCH] feat: update tester --- skills/tester/README.md | 4 +- skills/tester/SKILL.md | 2 +- skills/tester/references/maintain.md | 3 +- skills/tester/references/run.md | 13 +- skills/tester/references/schema.md | 9 +- skills/tester/scripts/catalog.py | 48 +++++- skills/tester/scripts/generate_cases_csv.py | 40 ++++- skills/tester/scripts/generate_run_csv.py | 162 ++++++++++++++++++++ 8 files changed, 259 insertions(+), 22 deletions(-) create mode 100644 skills/tester/scripts/generate_run_csv.py diff --git a/skills/tester/README.md b/skills/tester/README.md index 5c3bb48..f284694 100644 --- a/skills/tester/README.md +++ b/skills/tester/README.md @@ -45,6 +45,6 @@ ## 如何判断完成 -- **列出**：能看到分类表和总条数；`.pouch/tester/cases.csv` 已由 skill 里的 `generate_cases_csv.py` 从 `cases/` 重新生成。 +- **列出**：能看到分类表和总条数；`.pouch/tester/cases.csv` 已由 skill 里的 `generate_cases_csv.py` 从 `cases/` 重新生成。最近一次运行的结果在同一张表的 `last_result` / `last_executed` 列。 - **维护**：目标 id 能 `查看` 到，且检查通过。 -- **运行**：有本次 `runs/` 快照，每条都有判定和证据；没填 HOST 时会停，而不是编造失败。 +- **运行**：有本次 `runs/*.yaml` 快照和同名 `.csv` 成绩单（`executed` 列区分已跑 / 未跑）；没填 HOST 时会停，而不是编造失败。 diff --git a/skills/tester/SKILL.md b/skills/tester/SKILL.md index 707ea65..96b8029 100644 --- a/skills/tester/SKILL.md +++ b/skills/tester/SKILL.md @@ -67,7 +67,7 @@ python3 /scripts/catalog.py list --root python3 /scripts/generate_cases_csv.py --root ``` -`list` 结束时也会跑它。告诉用户路径 `.pouch/tester/cases.csv`。不要手改 CSV，改用例走「维护」后再生成。 +`list` 结束时也会跑它。告诉用户路径 `.pouch/tester/cases.csv`。不要手改 CSV，改用例走「维护」后再生成。有运行快照时，这份 CSV 末尾有 `last_result` / `last_executed`（`yes` 已跑，`no` 跳过，空=未进入最近一次运行）。 ## 查看 diff --git a/skills/tester/references/maintain.md b/skills/tester/references/maintain.md index 5c37f4b..df21743 100644 --- a/skills/tester/references/maintain.md +++ b/skills/tester/references/maintain.md @@ -35,4 +35,5 @@ python3 /scripts/catalog.py next-id --root --category - 为了能在默认风险下跑而把 `risk` 改低 - 一次塞进无关的一批用例 - 手改 `runs/` 冒充执行记录 -- 手改 `cases.csv`（必须从 `cases/*.yaml` 生成） +- 手改 `cases.csv`（必须从 `cases/*.yaml` 生成；`last_*` 列来自最近一次 `runs/*.yaml`） +- 手改 `runs/*.csv`（必须从对应 yaml 用 `generate_run_csv.py` 生成） diff --git a/skills/tester/references/run.md b/skills/tester/references/run.md index a36d40a..71a02ce 100644 --- a/skills/tester/references/run.md +++ b/skills/tester/references/run.md @@ -35,7 +35,7 @@ python3 /scripts/catalog.py select --root \ ## 4. 执行 -按 select 返回的顺序，一条一条来。默认失败即停（产品 `fail` 或探测到的不安全状态）。用户说继续 / keep-going 才往后跑。 +按 select 返回的顺序，一条一条来。select 按 `catalog.yaml` 的 `categories` 顺序，同一分类内按该 yaml 列表顺序（不是文件名字母序）。默认失败即停（产品 `fail` 或探测到的不安全状态）。用户说继续 / keep-going 才往后跑。 **script** @@ -57,7 +57,13 @@ python3 /scripts/catalog.py run-script --root --id 每条：`id, result, evidence, notes`。`result` 只能是 `pass` / `fail` / `environment-fail` / `skipped`。 -写入 `.pouch/tester/runs/-<范围>.yaml`，包含筛选条件、HOST 是否已设（可以写 HOST 值）、session 是否复用、逐条结果。 +写入 `.pouch/tester/runs/-<范围>.yaml`，包含筛选条件、HOST 是否已设（可以写 HOST 值）、session 是否复用、逐条结果。立刻用同一快照生成 CSV，不要手改： + +```bash +python3 /scripts/generate_run_csv.py --root --run +``` + +这会写出 `.pouch/tester/runs/-<范围>.csv`（`executed=yes|no`），并刷新 `cases.csv` 的 `last_result` / `last_executed` 列。对话里两个路径都报。 对话报告： @@ -70,6 +76,7 @@ session: browser=复用|新开|未用；mobile=… 计数: pass=N fail=N environment-fail=N skipped=N 失败: id + 一句证据 跳过: id + 原因 -快照: .pouch/tester/runs/ +快照: .pouch/tester/runs/.yaml +成绩单: .pouch/tester/runs/.csv 下一步: 一句话 ``` diff --git a/skills/tester/references/schema.md b/skills/tester/references/schema.md index a326360..56942b6 100644 --- a/skills/tester/references/schema.md +++ b/skills/tester/references/schema.md @@ -7,15 +7,16 @@ catalog.yaml 分类、风险阶梯、session、变量 local.yaml.example HOST 与授权开关模板 local.yaml 本机覆盖，不提交 cases/.yaml 一个分类一个文件（执行源） -cases.csv 给人看的扁平清单；skill 的 generate_cases_csv.py 读 cases/*.yaml 生成 -runs/ 每次运行快照 +cases.csv 给人看的扁平清单；generate_cases_csv.py 读 cases/*.yaml，并拼最近一次运行的 last_* 列 +runs/*.yaml 每次运行快照（执行源） +runs/*.csv 同一次运行的成绩单；generate_run_csv.py 从对应 yaml 生成 ``` ## catalog.yaml | 字段 | 含义 | |------|------| -| `categories` | `{id, title}` 列表，id 与 `cases/.yaml` 对应 | +| `categories` | `{id, title}` 列表，id 与 `cases/.yaml` 对应；列表顺序即 `select` / 执行顺序 | | `risks` | 从低到高，只能是 `safe` → `act` → `harm` 的一段，且以 `safe` 开头。含义见 [levels.md](levels.md) | | `default_max_risk` | 未指定时的运行上限，必须是 `risks` 里的值 | | `vars` | 可替换变量，如 `API_BASE: http://${HOST}:8095` | @@ -36,7 +37,7 @@ runs/ 每次运行快照 | `category` | 是 | 必须等于文件声明的 category | | `title` | 是 | 短名 | | `what` | 是 | 一句可观察的「测什么」 | -| `status` | 是 | `active` / `retired` | +| `status` | 是 | `active` / `pending` / `retired`。`pending` = 暂缓，仍保留在目录里但不执行，原因写在 `notes` | | `suite` | 是 | `smoke` / `full` / `slow`，判定见 [levels.md](levels.md) | | `risk` | 是 | `safe` / `act` / `harm` | | `gate` | 否 | 项目细分闸，必须在 catalog `gates` 里 | diff --git a/skills/tester/scripts/catalog.py b/skills/tester/scripts/catalog.py index 45586d1..38547de 100755 --- a/skills/tester/scripts/catalog.py +++ b/skills/tester/scripts/catalog.py @@ -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( diff --git a/skills/tester/scripts/generate_cases_csv.py b/skills/tester/scripts/generate_cases_csv.py index c4b87e2..735e995 100755 --- a/skills/tester/scripts/generate_cases_csv.py +++ b/skills/tester/scripts/generate_cases_csv.py @@ -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 diff --git a/skills/tester/scripts/generate_run_csv.py b/skills/tester/scripts/generate_run_csv.py new file mode 100644 index 0000000..e494ac3 --- /dev/null +++ b/skills/tester/scripts/generate_run_csv.py @@ -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())