From a47e89970212160e3e5eade4da7fdb89a1f3d830 Mon Sep 17 00:00:00 2001 From: laily Date: Thu, 10 Sep 2026 16:34:38 +0800 Subject: [PATCH] feat(tester): add catalog-driven black-box test skill Maintain and run project test catalogs under .pouch/tester, filtered by category, id, and risk (safe/act/harm). Scripts generate the human CSV from cases YAML. --- skills/tester/README.md | 50 ++ skills/tester/SKILL.md | 90 +++ skills/tester/references/levels.md | 40 ++ skills/tester/references/maintain.md | 38 ++ skills/tester/references/run.md | 75 +++ skills/tester/references/schema.md | 68 +++ skills/tester/scripts/catalog.py | 634 ++++++++++++++++++++ skills/tester/scripts/generate_cases_csv.py | 194 ++++++ skills/tester/templates/catalog.yaml | 26 + skills/tester/templates/local.yaml.example | 4 + 10 files changed, 1219 insertions(+) create mode 100644 skills/tester/README.md create mode 100644 skills/tester/SKILL.md create mode 100644 skills/tester/references/levels.md create mode 100644 skills/tester/references/maintain.md create mode 100644 skills/tester/references/run.md create mode 100644 skills/tester/references/schema.md create mode 100755 skills/tester/scripts/catalog.py create mode 100755 skills/tester/scripts/generate_cases_csv.py create mode 100644 skills/tester/templates/catalog.yaml create mode 100644 skills/tester/templates/local.yaml.example diff --git a/skills/tester/README.md b/skills/tester/README.md new file mode 100644 index 0000000..5c3bb48 --- /dev/null +++ b/skills/tester/README.md @@ -0,0 +1,50 @@ +# tester + +维护并执行项目里的黑盒测试目录。用例按分类放在 `.pouch/tester/`，可以列出、增改，也可以按分类或编号去跑。 + +## 什么时候使用 + +- 「列出测试」「一共有哪些用例」 +- 「加一条相机开盖的测试」「完善 TC-CAM-003 的验证」 +- 「跑 smoke」「测 CAM」「跑 TC-INF-001 和 TC-CAM-001」 +- 「现场复测」 + +不管 ACK。这是项目自己的测试目录。 + +## 使用前准备 + +- 目录在项目根的 `.pouch/tester/`。没有就说「初始化 tester」。 +- 跑真实环境前，复制 `.pouch/tester/local.yaml.example` 为 `local.yaml`，填 `HOST`。也可以用环境变量 `HX_TEST_HOST`。 +- 会动硬件、会改数据的用例带风险级别。默认只跑最低档（只读）。要跑更高风险，需要你明确说，例如「风险到 hardware」。 + +## 使用示例 + +```text +列出测试用例 +``` + +```text +加一条测试：分类 CAM，风险 hardware，开盖再关盖，用脚本打 /api/camera_3d +``` + +```text +跑 INF 分类的 smoke +``` + +```text +跑 TC-CAM-001,TC-CAM-005，风险保持默认 +``` + +## Agent 会做什么 + +1. 用脚本读 `.pouch/tester/`，按分类统计或筛出你点名的用例，不会把整库塞进对话。 +2. 改用例前先给你看将要写入的内容，你确认再保存。 +3. 执行时：脚本类直接跑命令；浏览器和移动模拟器先初始化一次，同一轮里复用。 +4. 超出你允许的风险级别会跳过，不会偷偷执行。 +5. 跑完给出按分类计数的结果，以及 `pass / fail / environment-fail / skipped`。 + +## 如何判断完成 + +- **列出**：能看到分类表和总条数；`.pouch/tester/cases.csv` 已由 skill 里的 `generate_cases_csv.py` 从 `cases/` 重新生成。 +- **维护**：目标 id 能 `查看` 到，且检查通过。 +- **运行**：有本次 `runs/` 快照，每条都有判定和证据；没填 HOST 时会停，而不是编造失败。 diff --git a/skills/tester/SKILL.md b/skills/tester/SKILL.md new file mode 100644 index 0000000..707ea65 --- /dev/null +++ b/skills/tester/SKILL.md @@ -0,0 +1,90 @@ +--- +name: tester +description: >- + 维护并执行项目黑盒测试目录（.pouch/tester）：列出/增改用例，按分类或 + id 跑脚本、浏览器或移动模拟器。触发词：/tester、列出测试、完善测试用例、 + 按分类测、跑 smoke、现场复测。默认风险最低档，不猜 HOST。 +--- + +# Tester：项目测试目录 + +开始时解析本 `SKILL.md` 所在目录为 ``。项目根用 +`git rev-parse --show-toplevel`。目录固定为 `/.pouch/tester/`。 + +用脚本读写目录，不要把 `cases/` 整库灌进上下文： + +```bash +python3 /scripts/catalog.py \ + --root [参数] +``` + +## 选择模式 + +- 还没有 `.pouch/tester/catalog.yaml`，或用户说初始化 tester：执行「初始化」。 +- 列出、有哪些测试、一共多少条：执行「列出」。 +- 看某一条怎么测：执行「查看」。 +- 加一条 / 完善 / 改验证 / 退役：执行「维护」。 +- 检查目录是否完整：执行「检查」。 +- 跑测试、按分类测、按 id 测、跑 smoke：执行「运行」。 + +## 边界（所有模式） + +- 不猜 HOST / Base URL / 是否已授权。缺 HOST 则停，记 environment-fail，不用 localhost。 +- 默认风险上限是 catalog 里的 `default_max_risk`。超出上限的条目标 `skipped`，禁止执行。 +- `risk` 只能是 `safe` / `act` / `harm`。`suite=smoke` 只能配 `risk=safe`。规则见 [levels.md](references/levels.md)。 +- `surface=browser` 必须开真实页面；`surface=mobile` 必须挂模拟器 session。二者都不得改成只打 API。 +- 浏览器、移动模拟器按 catalog `sessions` 初始化一次，同一次运行内复用。 +- 环境失败（连不上、缺浏览器、下游超时、session 挂了）不是产品失败。 +- 不改产品代码，不提交，不把成绩单当操作手册。 +- 禁止把完整 `cases/` 贴进 prompt。筛选用 `select`，单条用 `show`。 + +## 初始化 + +用户点名才跑。已有 `catalog.yaml` 则不覆盖，转入「检查」。 + +1. 建 `.pouch/tester/{cases,runs}/`。 +2. 若不存在，从 `/templates/` 拷 `catalog.yaml` 和 `local.yaml.example`。 +3. 不写 `local.yaml` 的 HOST。不猜地址。 +4. 跑 `catalog.py validate --root `。无用例是「部分完成」，不是失败。 + +```text +## tester 初始化：完成 | 部分完成 | 阻塞 + +已具备: … +待配置: 路径 + 字段 + 缺了会挡住哪步 +下一步: 一句话 +``` + +## 列出 + +```bash +python3 /scripts/catalog.py list --root +``` + +把脚本打印的分类表原样给用户。不要手算条数。清单 CSV 由独立脚本从 `cases/*.yaml` 生成： + +```bash +python3 /scripts/generate_cases_csv.py --root +``` + +`list` 结束时也会跑它。告诉用户路径 `.pouch/tester/cases.csv`。不要手改 CSV，改用例走「维护」后再生成。 + +## 查看 + +```bash +python3 /scripts/catalog.py show --root --id +``` + +只展示这一条的 how / expected / risk / surface。字段对不上时再读 [schema.md](references/schema.md)。 + +## 检查 + +只读。跑 `validate`。HOST 是否已配置只报「已设 / 未设」，不要把 `local.yaml` 全文贴出来。用初始化同一套报告，标题改为 `## tester 检查：…`。 + +## 维护 + +读 [maintain.md](references/maintain.md)。写 `suite`/`risk` 时读 [levels.md](references/levels.md)。先 `show` 或 `next-id`，拟好条目给用户确认再写文件，然后 `validate`。 + +## 运行 + +读 [run.md](references/run.md)。先 `select` 得到命中列表，再按 surface 分流。未指定范围时等同「suite=smoke 且风险上限=默认」。 diff --git a/skills/tester/references/levels.md b/skills/tester/references/levels.md new file mode 100644 index 0000000..b2521b8 --- /dev/null +++ b/skills/tester/references/levels.md @@ -0,0 +1,40 @@ +# suite 与 risk + +维护或用户问套件/安全级别时读本文件。两套词对所有项目相同，不要自造名字。 + +## suite：跑哪一批 + +只表示耗时和依赖，不表示危不危险。危不危险看 `risk`。 + +业界常见还有 sanity（窄回归，和 smoke 重叠）、regression（≈本 skill 的 full）、acceptance、nightly。本 skill **不采用** sanity/acceptance：和 smoke/full 分不清，且已有 category、id、risk 三层筛选。 + +固定三个词： + +| 值 | 何时用 | +|----|--------| +| `smoke` | 最短路径证明系统还活着。必须快、少依赖、**`risk` 只能是 `safe`**。 | +| `full` | 日常回归里其余自动化用例。说「跑 full」仍受风险上限约束。 | +| `slow` | 太慢或环境太重，不进日常 full：真实焊接、付费 API、要工件/账号、跑很久。 | + +口诀： + +1. 只回答「活着吗、主入口能不能用」→ `smoke`。 +2. 要特殊现场/很久才能跑完 → `slow`。 +3. 其余自动化 → `full`。 +4. 拿不准 → `full`。不要把会动真机或会删数据的放进 `smoke`。 + +未指定套件时只跑 `smoke`。`跑 full` 不含 `slow`。 + +## risk：副作用有多大 + +三个短词，网站 / App / 机器人共用。`catalog.yaml` 的 `risks` 必须按此顺序，且以 `safe` 开头： + +| 值 | 含义 | 网站 | App | 机器人 | +|----|------|------|-----|--------| +| `safe` | 只看，不留下变化 | 打开页、只读 API | 打开界面 | 读位姿、查状态 | +| `act` | 会动手，一般可补救 | 提交表单、改配置 | 发消息、保存 | 点动、使能、开盖扫描 | +| `harm` | 难恢复或伤人/伤钱/伤环境 | 删库、真实扣费 | 清数据、内购 | 急停、断电、真焊 | + +默认只跑 `safe`。项目细分闸（点动 vs 焊接）用可选 `gate`，不要再发明 risk 名。 + +`suite=smoke` 且 `risk!=safe` 为非法。 diff --git a/skills/tester/references/maintain.md b/skills/tester/references/maintain.md new file mode 100644 index 0000000..5c37f4b --- /dev/null +++ b/skills/tester/references/maintain.md @@ -0,0 +1,38 @@ +# tester 维护用例 + +只在用户要加、改、退役时读本文件。写之前必须把完整条目亮给用户，确认后再落盘。 + +## 加一条 + +1. 分类必须已在 `catalog.yaml` 的 `categories` 里。没有就先改 catalog（仍要确认）。 +2. 取编号： + +```bash +python3 /scripts/catalog.py next-id --root --category +``` + +3. 拟一条符合 [schema.md](schema.md) 的用例。`suite`/`risk` 按 [levels.md](levels.md)：`smoke` 只能 `safe`；会动手用 `act`，难恢复用 `harm`。细分授权用 `gate`，不要发明新的 risk 名。 +4. `surface=browser|mobile` 必须填 `session`，且 catalog 里已声明。 +5. `what` 和 `expected` 写成可观察信号，禁止「应该正常」。 +6. 给用户看拟写内容。确认后写入 `cases/.yaml` 的 `cases` 列表末尾。 +7. `validate`。失败则修好再结束。通过后跑 `generate_cases_csv.py --root ` 刷新 `.pouch/tester/cases.csv`。不要手改这份 CSV。 + +## 完善 / 更新 + +1. `show --id `。 +2. 只改用户点名的字段（常见：`how`、`expected`、`risk`、`notes`）。 +3. 展示 diff 或更新后的整条。确认后写回。 +4. `validate`。通过后跑 `generate_cases_csv.py` 刷新 `cases.csv`。 + +## 退役 + +把 `status` 改为 `retired`，不要删文件、不要复用 id。`notes` 写原因。确认后写回，再跑 `generate_cases_csv.py`。 + +## 禁止 + +- 未确认就写 +- 把 browser/mobile 改成 script 来「省事」 +- 为了能在默认风险下跑而把 `risk` 改低 +- 一次塞进无关的一批用例 +- 手改 `runs/` 冒充执行记录 +- 手改 `cases.csv`（必须从 `cases/*.yaml` 生成） diff --git a/skills/tester/references/run.md b/skills/tester/references/run.md new file mode 100644 index 0000000..a36d40a --- /dev/null +++ b/skills/tester/references/run.md @@ -0,0 +1,75 @@ +# tester 运行 + +只在用户要跑测试时读本文件。 + +## 1. 选范围 + +未指定范围 → `--suite smoke`，风险上限 = `default_max_risk`。 + +用户说分类、id、suite、风险上限时，原样传给 select： + +```bash +python3 /scripts/catalog.py select --root \ + [--category CAT] [--id ID]... [--suite smoke|full|slow] \ + [--max-risk ] [--surface script|browser|mobile] +``` + +`select` 已排除 `retired` 和风险高于上限的条目（后者若用户点了具体 id，会在 JSON 的 `skipped` 里说明，不执行）。 + +无命中：停下说明筛选条件，不要扩大范围。 + +## 2. 运行前闸 + +- `HOST` 未设（环境变量 / `HX_TEST_HOST` / `local.yaml` 都没有）：全部停，environment-fail。 +- 命中条的 `risk` 在 catalog `auth_env` 里有对应变量：变量不是 `true` 则该条 `skipped`。有 `gate` 时还要 `catalog.gates` 对应变量为 `true`。即使用户点了 id 也不执行。 +- 若 catalog 声明了 `status_probe`：先请求它。命中 `stop_if` → 整轮停，environment-fail。 + +## 3. Session + +看命中条用到的 `session` 键： + +- 尚未初始化：按 catalog `sessions..init` 做。`reuse: persist` 的先 `probe`，活着就附着。 +- 本轮已初始化：直接用，不要重开浏览器或模拟器。 +- `reuse: run`：整轮结束再关。`persist`：不关。 +- 中途 session 挂了：后续同 surface 标 environment-fail，不要标产品失败。 + +## 4. 执行 + +按 select 返回的顺序，一条一条来。默认失败即停（产品 `fail` 或探测到的不安全状态）。用户说继续 / keep-going 才往后跑。 + +**script** + +有 `how.run`： + +```bash +python3 /scripts/catalog.py run-script --root --id +``` + +`run-script` 只跑命令，返回 `executed` + stdout/exit（或 skipped / environment-fail）。**pass/fail 仍由 Agent 对照 `expected` 判定**，不要把 curl 退出码 0 当成产品通过。 + +没有 `run`：按 `how.steps` 做，证据对照 `expected`。不要发明 steps 里没有的调用。 + +**browser / mobile** + +挂到已有 session，执行 `how.steps`，对照 `expected`。必须留下可见文本、URL 或交互证据。 + +## 5. 记录与报告 + +每条：`id, result, evidence, notes`。`result` 只能是 `pass` / `fail` / `environment-fail` / `skipped`。 + +写入 `.pouch/tester/runs/-<范围>.yaml`，包含筛选条件、HOST 是否已设（可以写 HOST 值）、session 是否复用、逐条结果。 + +对话报告： + +```text +## tester 运行：完成 | 部分完成 | 阻塞 + +范围: … +风险上限: … +session: browser=复用|新开|未用；mobile=… +计数: pass=N fail=N environment-fail=N skipped=N +失败: id + 一句证据 +跳过: id + 原因 +快照: .pouch/tester/runs/ +下一步: 一句话 +``` diff --git a/skills/tester/references/schema.md b/skills/tester/references/schema.md new file mode 100644 index 0000000..a326360 --- /dev/null +++ b/skills/tester/references/schema.md @@ -0,0 +1,68 @@ +# tester 目录字段 + +项目根 `.pouch/tester/`： + +``` +catalog.yaml 分类、风险阶梯、session、变量 +local.yaml.example HOST 与授权开关模板 +local.yaml 本机覆盖，不提交 +cases/.yaml 一个分类一个文件（执行源） +cases.csv 给人看的扁平清单；skill 的 generate_cases_csv.py 读 cases/*.yaml 生成 +runs/ 每次运行快照 +``` + +## catalog.yaml + +| 字段 | 含义 | +|------|------| +| `categories` | `{id, title}` 列表，id 与 `cases/.yaml` 对应 | +| `risks` | 从低到高，只能是 `safe` → `act` → `harm` 的一段，且以 `safe` 开头。含义见 [levels.md](levels.md) | +| `default_max_risk` | 未指定时的运行上限，必须是 `risks` 里的值 | +| `vars` | 可替换变量，如 `API_BASE: http://${HOST}:8095` | +| `auth_env` | 风险名 → 环境变量名。该档执行前变量必须为 `true` | +| `gates` | 可选。项目细分闸名 → 环境变量。用在用例的 `gate` 上，不要拿它代替 `risk` | +| `sessions` | 可复用会话。`kind`: `browser` / `mobile`；`reuse`: `run`（本轮结束关掉）或 `persist`（probe 活着就附着） | +| `id_pattern` | 用例 id 正则 | + +`local.yaml` 与环境变量覆盖 `HOST` 和授权开关。查找顺序：环境变量 > `local.yaml` > `catalog.yaml` 的 vars。也认 `HX_TEST_HOST` 作为 `HOST`。 + +## 一条用例 + +写在对应分类文件的 `cases:` 列表里。 + +| 字段 | 必填 | 含义 | +|------|------|------| +| `id` | 是 | 匹配 `id_pattern` | +| `category` | 是 | 必须等于文件声明的 category | +| `title` | 是 | 短名 | +| `what` | 是 | 一句可观察的「测什么」 | +| `status` | 是 | `active` / `retired` | +| `suite` | 是 | `smoke` / `full` / `slow`，判定见 [levels.md](levels.md) | +| `risk` | 是 | `safe` / `act` / `harm` | +| `gate` | 否 | 项目细分闸，必须在 catalog `gates` 里 | +| `surface` | 是 | `script` / `browser` / `mobile` | +| `session` | browser/mobile 必填 | 指向 catalog `sessions` 的键 | +| `how` | 是 | 怎么测，见下 | +| `expected` | 是 | 怎么过，至少一条 `{kind, value}` | +| `maps_to` | 否 | 对人文档的功能项 id | +| `requires` | 否 | 前置标签 | +| `skip_unless` | 否 | 额外跳过条件（环境变量表达式） | +| `setup` | 否 | 人话前置 | +| `cleanup` | 否 | 跑完回退 | +| `notes` | 否 | 备注 | + +## how + +- `script`：`how.run` 是一条可在项目根执行的命令（可多行脚本）。没有 `run` 时 Agent 按 `how.steps` 做，仍不得发明未写明的请求。 +- `browser`：`how.url` + `how.steps`。`expected.kind` 用 `visible-text` / `url` / `interaction`。 +- `mobile`：`how.steps`，挂到 `session` 指定的模拟器。`expected` 同样是可观察信号。 + +可选 `how.timeout`，秒，默认 60。 + +## expected.kind + +`api-status` / `api-field` / `script-exit` / `visible-text` / `url` / `interaction` + +## 判定 + +`pass` / `fail` / `environment-fail` / `skipped` diff --git a/skills/tester/scripts/catalog.py b/skills/tester/scripts/catalog.py new file mode 100755 index 0000000..45586d1 --- /dev/null +++ b/skills/tester/scripts/catalog.py @@ -0,0 +1,634 @@ +#!/usr/bin/env python3 +"""tester catalog CLI: list / select / show / validate / next-id / run-script.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +from datetime import datetime, timezone +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) + +SURFACES = ("script", "browser", "mobile") +STATUSES = ("active", "retired") +SUITES = ("smoke", "full", "slow") +CANONICAL_RISKS = ("safe", "act", "harm") +RESULTS = ("pass", "fail", "environment-fail", "skipped") +EXPECTED_KINDS = ( + "api-status", + "api-field", + "script-exit", + "visible-text", + "url", + "interaction", +) + + +def catalog_dir(root: Path) -> Path: + return root / ".pouch" / "tester" + + +def load_yaml(path: Path) -> Any: + text = path.read_text(encoding="utf-8") + data = yaml.safe_load(text) + return data + + +def dump_yaml(data: Any) -> str: + return yaml.safe_dump( + data, + allow_unicode=True, + sort_keys=False, + default_flow_style=False, + ) + + +def truthy(value: Any) -> bool: + if isinstance(value, bool): + return value + return str(value).strip().lower() in ("1", "true", "yes", "on") + + +def load_catalog_meta(root: Path) -> dict[str, Any]: + path = catalog_dir(root) / "catalog.yaml" + if not path.is_file(): + raise SystemExit(f"缺少 {path}") + data = load_yaml(path) + if not isinstance(data, dict): + raise SystemExit(f"{path} 必须是 mapping") + return data + + +def load_local(root: Path) -> dict[str, Any]: + path = catalog_dir(root) / "local.yaml" + if not path.is_file(): + return {} + data = load_yaml(path) or {} + if not isinstance(data, dict): + raise SystemExit(f"{path} 必须是 mapping") + return data + + +def case_files(root: Path) -> list[Path]: + d = catalog_dir(root) / "cases" + if not d.is_dir(): + return [] + return sorted(p for p in d.glob("*.yaml") if p.is_file()) + + +def load_all_cases(root: Path) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for path in case_files(root): + data = load_yaml(path) or {} + if not isinstance(data, dict): + raise SystemExit(f"{path} 必须是 mapping") + cat = data.get("category") + title = data.get("title", "") + cases = data.get("cases") or [] + if not isinstance(cases, list): + raise SystemExit(f"{path}: cases 必须是列表") + for i, case in enumerate(cases): + if not isinstance(case, dict): + raise SystemExit(f"{path}: cases[{i}] 必须是 mapping") + item = dict(case) + item["_file"] = str(path.relative_to(root)) + if "category" not in item and cat: + item["category"] = cat + if "_file_category" not in item: + item["_file_category"] = cat + item["_file_title"] = title + out.append(item) + return out + + +def resolve_vars(root: Path, meta: dict[str, Any]) -> dict[str, str]: + merged: dict[str, str] = {} + for key, value in (meta.get("vars") or {}).items(): + merged[str(key)] = "" if value is None else str(value) + local = load_local(root) + for key, value in local.items(): + if value is None: + continue + merged[str(key)] = str(value).lower() if isinstance(value, bool) else str(value) + auth_keys = set() + for mapping in (meta.get("auth_env") or {}, meta.get("gates") or {}): + if isinstance(mapping, dict): + auth_keys.update(str(value) for value in mapping.values() if value) + for key, value in os.environ.items(): + if key in merged or key in auth_keys or key.startswith("AUTHORIZE_") or key in { + "HOST", + "HX_TEST_HOST", + "WEB_BASE", + "API_BASE", + "WS_URL", + "STATUS_BASE", + }: + merged[key] = value + if not merged.get("HOST") and merged.get("HX_TEST_HOST"): + merged["HOST"] = merged["HX_TEST_HOST"] + if os.environ.get("HX_TEST_HOST") and not os.environ.get("HOST"): + merged["HOST"] = os.environ["HX_TEST_HOST"] + for _ in range(10): + changed = False + for key, value in list(merged.items()): + new = substitute(value, merged) + if new != value: + merged[key] = new + changed = True + if not changed: + break + return merged + + +def substitute(text: str, vars_map: dict[str, str]) -> str: + def repl(match: re.Match[str]) -> str: + name = match.group(1) + return vars_map.get(name, match.group(0)) + + return re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", repl, text) + + +def risk_index(meta: dict[str, Any], name: str) -> int: + risks = list(meta.get("risks") or []) + if name not in risks: + raise SystemExit(f"未知风险: {name}（catalog risks={risks}）") + return risks.index(name) + + +def host_set(vars_map: dict[str, str]) -> bool: + host = (vars_map.get("HOST") or "").strip() + return bool(host) and "${HOST}" not in host + + +def export_csv_script() -> Path: + return Path(__file__).resolve().parent / "generate_cases_csv.py" + + +def run_export_csv(root: Path) -> int: + proc = subprocess.run( + [sys.executable, str(export_csv_script()), "--root", str(root)], + check=False, + ) + return int(proc.returncode) + + +def cmd_list(root: Path) -> int: + meta = load_catalog_meta(root) + cases = load_all_cases(root) + risks = list(meta.get("risks") or []) + 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)}") + print() + headers = ["分类", "标题", "条数", "smoke"] + risks + list(SURFACES) + print("| " + " | ".join(headers) + " |") + print("| " + " | ".join("---" for _ in headers) + " |") + + cat_ids: list[str] = [] + for c in meta.get("categories") or []: + if isinstance(c, dict) and c.get("id"): + cat_ids.append(str(c["id"])) + for c in cases: + cid = str(c.get("category") or "") + if cid and cid not in cat_ids: + cat_ids.append(cid) + + for cat in cat_ids: + group = [c for c in active if c.get("category") == cat] + title = (cats_meta.get(cat) or {}).get("title") or ( + group[0].get("_file_title") if group else "" + ) + row = [cat, str(title), str(len(group)), str(sum(1 for c in group if c.get("suite") == "smoke"))] + for risk in risks: + row.append(str(sum(1 for c in group if c.get("risk") == risk))) + for surface in SURFACES: + row.append(str(sum(1 for c in group if c.get("surface") == surface))) + print("| " + " | ".join(row) + " |") + print() + print("跑法: 跑 <分类> | 跑 | 跑 smoke | 跑 <分类> 风险到 ") + vars_map = resolve_vars(root, meta) + print(f"HOST: {'已设' if host_set(vars_map) else '未设'}") + code = run_export_csv(root) + if code == 0: + print(f"清单: {catalog_dir(root).relative_to(root) / 'cases.csv'}") + return code + + +def parse_ids(values: list[str]) -> list[str]: + out: list[str] = [] + for raw in values: + for part in str(raw).split(","): + item = part.strip() + if item: + out.append(item) + return out + + +def cmd_select(args: argparse.Namespace, root: Path) -> int: + meta = load_catalog_meta(root) + cases = load_all_cases(root) + max_risk = args.max_risk or meta.get("default_max_risk") or "safe" + max_i = risk_index(meta, max_risk) + wanted = set(parse_ids(args.ids or [])) + selected: list[dict[str, Any]] = [] + skipped: list[dict[str, str]] = [] + missing = set(wanted) + + for case in cases: + cid = str(case.get("id") or "") + if case.get("status", "active") != "active": + if cid in wanted: + skipped.append({"id": cid, "reason": "retired"}) + missing.discard(cid) + continue + if wanted: + if cid not in wanted: + continue + missing.discard(cid) + else: + if args.category and case.get("category") != args.category: + continue + if args.suite and case.get("suite") != args.suite: + continue + if args.surface and case.get("surface") != args.surface: + continue + risk = str(case.get("risk") or "safe") + try: + r_i = risk_index(meta, risk) + except SystemExit: + skipped.append({"id": cid, "reason": f"未知风险 {risk}"}) + continue + if r_i > max_i: + skipped.append({"id": cid, "reason": f"risk={risk} 超过上限 {max_risk}"}) + continue + public = {k: v for k, v in case.items() if not str(k).startswith("_")} + selected.append(public) + + if wanted and missing: + print( + json.dumps({"error": "找不到 active 用例: " + ", ".join(sorted(missing))}, ensure_ascii=False), + file=sys.stderr, + ) + return 2 + + print( + json.dumps( + { + "max_risk": max_risk, + "count": len(selected), + "skipped": skipped, + "cases": selected, + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + +def cmd_show(root: Path, case_id: str) -> int: + for case in load_all_cases(root): + if case.get("id") == case_id: + public = {k: v for k, v in case.items() if not str(k).startswith("_")} + print(dump_yaml(public), end="") + return 0 + print(f"找不到用例: {case_id}", file=sys.stderr) + return 2 + + +def validate(root: Path) -> list[str]: + errors: list[str] = [] + meta_path = catalog_dir(root) / "catalog.yaml" + if not meta_path.is_file(): + return [f"缺少 {meta_path}"] + try: + meta = load_catalog_meta(root) + except SystemExit as exc: + return [str(exc)] + + cats = meta.get("categories") + if not isinstance(cats, list): + errors.append("catalog.categories 必须是列表") + cats = [] + cat_ids = [] + for i, item in enumerate(cats): + if not isinstance(item, dict) or not item.get("id"): + errors.append(f"catalog.categories[{i}] 需要 id") + else: + cat_ids.append(str(item["id"])) + risks = meta.get("risks") + if not isinstance(risks, list) or not risks: + errors.append("catalog.risks 必须是非空列表") + risks = [] + if risks and risks[0] != "safe": + errors.append("catalog.risks 必须以 safe 开头") + last_pos = -1 + for name in risks: + if name not in CANONICAL_RISKS: + errors.append( + f"非法 risk 名 {name!r}，只能是 safe/act/harm" + ) + continue + pos = CANONICAL_RISKS.index(name) + if pos <= last_pos: + errors.append("catalog.risks 必须按 safe < act < harm 升序且不重复") + last_pos = pos + default_max = meta.get("default_max_risk") + if default_max not in risks: + errors.append(f"default_max_risk={default_max!r} 不在 risks 中") + gates = meta.get("gates") or {} + if gates and not isinstance(gates, dict): + errors.append("catalog.gates 必须是 mapping") + gates = {} + auth_env = meta.get("auth_env") or {} + if auth_env and not isinstance(auth_env, dict): + errors.append("catalog.auth_env 必须是 mapping") + auth_env = {} + sessions = meta.get("sessions") or {} + if not isinstance(sessions, dict): + errors.append("catalog.sessions 必须是 mapping") + sessions = {} + pattern = meta.get("id_pattern") or r"^.+$" + try: + id_re = re.compile(pattern) + except re.error as exc: + errors.append(f"id_pattern 无效: {exc}") + id_re = re.compile(r"^.+$") + + seen: dict[str, str] = {} + try: + cases = load_all_cases(root) + except SystemExit as exc: + return errors + [str(exc)] + + for case in cases: + cid = str(case.get("id") or "") + where = f"{case.get('_file')}:{cid or '?'}" + if not cid: + errors.append(f"{where}: 缺少 id") + continue + if not id_re.match(cid): + errors.append(f"{where}: id 不匹配 {pattern}") + if cid in seen: + errors.append(f"{where}: id 重复（也出现在 {seen[cid]}）") + seen[cid] = str(case.get("_file")) + cat = str(case.get("category") or "") + file_cat = str(case.get("_file_category") or "") + if file_cat and cat and cat != file_cat: + errors.append(f"{where}: category={cat} 与文件 category={file_cat} 不一致") + if cat and cat_ids and cat not in cat_ids: + errors.append(f"{where}: category={cat} 不在 catalog.categories") + if case.get("status", "active") not in STATUSES: + errors.append(f"{where}: 非法 status") + if case.get("suite") not in SUITES: + errors.append(f"{where}: 非法 suite") + if case.get("risk") not in risks: + errors.append(f"{where}: 非法 risk={case.get('risk')}") + if case.get("suite") == "smoke" and case.get("risk") != "safe": + errors.append(f"{where}: smoke 只能 risk=safe") + gate = case.get("gate") + if gate: + if gate not in gates: + errors.append(f"{where}: gate={gate} 未在 catalog.gates 声明") + if case.get("risk") in (None, "safe"): + errors.append(f"{where}: risk=safe 不要填 gate") + risk = case.get("risk") + if risk and risk != "safe" and risk not in auth_env and not gate: + errors.append(f"{where}: {risk} 需要 auth_env.{risk} 或 gate") + surface = case.get("surface") + if surface not in SURFACES: + errors.append(f"{where}: 非法 surface={surface}") + if not str(case.get("title") or "").strip(): + errors.append(f"{where}: 缺少 title") + if not str(case.get("what") or "").strip(): + errors.append(f"{where}: 缺少 what") + how = case.get("how") + if not isinstance(how, dict): + errors.append(f"{where}: how 必须是 mapping") + how = {} + if surface == "script" and not how.get("run") and not how.get("steps"): + errors.append(f"{where}: script 需要 how.run 或 how.steps") + if surface in ("browser", "mobile"): + session = case.get("session") + if not session: + errors.append(f"{where}: {surface} 需要 session") + elif session not in sessions: + errors.append(f"{where}: session={session} 未在 catalog.sessions 声明") + if surface == "browser" and not how.get("url") and not how.get("steps"): + errors.append(f"{where}: browser 需要 how.url 或 how.steps") + if surface == "mobile" and not how.get("steps"): + errors.append(f"{where}: mobile 需要 how.steps") + expected = case.get("expected") + if not isinstance(expected, list) or not expected: + errors.append(f"{where}: expected 必须是非空列表") + else: + for j, sig in enumerate(expected): + if not isinstance(sig, dict) or not sig.get("kind") or not sig.get("value"): + errors.append(f"{where}: expected[{j}] 需要 kind 与 value") + elif sig.get("kind") not in EXPECTED_KINDS: + errors.append(f"{where}: expected[{j}].kind 非法") + return errors + + +def cmd_validate(root: Path) -> int: + errors = validate(root) + if errors: + for err in errors: + print(f"✗ {err}", file=sys.stderr) + print(f"validate: {len(errors)} 个问题", file=sys.stderr) + 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)}") + return 0 + + +def cmd_export_csv(root: Path) -> int: + return run_export_csv(root) + + +def cmd_next_id(root: Path, category: str) -> int: + meta = load_catalog_meta(root) + cat_ids = [c.get("id") for c in meta.get("categories") or [] if isinstance(c, dict)] + if category not in cat_ids: + print(f"未知分类: {category} 可选: {cat_ids}", file=sys.stderr) + return 2 + prefix = f"TC-{category}-" + max_n = 0 + for case in load_all_cases(root): + cid = str(case.get("id") or "") + if case.get("category") != category: + continue + match = re.match(rf"TC-{re.escape(category)}-(\d+)", cid) + if match: + max_n = max(max_n, int(match.group(1))) + print(f"{prefix}{max_n + 1:03d}") + return 0 + + +def authorized(vars_map: dict[str, str], meta: dict[str, Any], case: dict[str, Any]) -> bool: + risk = str(case.get("risk") or "safe") + risks = list(meta.get("risks") or ["safe"]) + lowest = risks[0] if risks else "safe" + if risk != lowest: + env_name = (meta.get("auth_env") or {}).get(risk) + if env_name and not truthy(vars_map.get(str(env_name), "")): + return False + if not env_name and not case.get("gate"): + return False + gate = case.get("gate") + if gate: + gate_env = (meta.get("gates") or {}).get(gate) + if not gate_env or not truthy(vars_map.get(str(gate_env), "")): + return False + return True + + +def cmd_run_script(root: Path, case_id: str) -> int: + meta = load_catalog_meta(root) + vars_map = resolve_vars(root, meta) + case = next((c for c in load_all_cases(root) if c.get("id") == case_id), None) + if not case: + print(json.dumps({"id": case_id, "result": "environment-fail", "evidence": "找不到用例"}, ensure_ascii=False)) + 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)) + return 0 + if case.get("surface") != "script": + print( + json.dumps( + {"id": public_id, "result": "environment-fail", "evidence": "run-script 只能跑 surface=script"}, + ensure_ascii=False, + ) + ) + return 2 + if not host_set(vars_map): + print( + json.dumps( + {"id": public_id, "result": "environment-fail", "evidence": "HOST 未设"}, + ensure_ascii=False, + ) + ) + return 2 + risk = str(case.get("risk") or "safe") + if not authorized(vars_map, meta, case): + print( + json.dumps( + {"id": public_id, "result": "skipped", "evidence": f"风险 {risk} 未授权"}, + ensure_ascii=False, + ) + ) + return 0 + how = case.get("how") or {} + run = how.get("run") + if not run: + print( + json.dumps( + {"id": public_id, "result": "environment-fail", "evidence": "没有 how.run，改由 Agent 按 steps 执行"}, + ensure_ascii=False, + ) + ) + return 2 + script = substitute(str(run), vars_map) + timeout = float(how.get("timeout") or 60) + env = os.environ.copy() + env.update({k: v for k, v in vars_map.items() if v is not None}) + try: + proc = subprocess.run( + script, + shell=True, + cwd=str(root), + env=env, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired: + print( + json.dumps( + { + "id": public_id, + "result": "environment-fail", + "evidence": f"timeout {timeout}s", + }, + ensure_ascii=False, + ) + ) + return 1 + output = ((proc.stdout or "") + ("\n" + proc.stderr if proc.stderr else "")).strip() + evidence = output[-1200:] + print( + json.dumps( + { + "id": public_id, + "result": "executed", + "exit": proc.returncode, + "evidence": evidence, + }, + ensure_ascii=False, + ) + ) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parent = argparse.ArgumentParser(add_help=False) + parent.add_argument("--root", default=".", help="项目根") + parser = argparse.ArgumentParser(description="tester catalog") + sub = parser.add_subparsers(dest="cmd", required=True) + + sub.add_parser("list", help="按分类统计", parents=[parent]) + p_sel = sub.add_parser("select", help="筛选用例 JSON", parents=[parent]) + p_sel.add_argument("--category") + p_sel.add_argument("--id", action="append", dest="ids", default=[]) + p_sel.add_argument("--suite", choices=SUITES) + p_sel.add_argument("--max-risk") + p_sel.add_argument("--surface", choices=SURFACES) + p_show = sub.add_parser("show", help="打印一条用例", parents=[parent]) + p_show.add_argument("--id", required=True, dest="case_id") + sub.add_parser("validate", help="校验目录", parents=[parent]) + sub.add_parser("export-csv", help="写出给人看的 cases.csv", parents=[parent]) + p_next = sub.add_parser("next-id", help="分配下一个 TC--NNN", parents=[parent]) + p_next.add_argument("--category", required=True) + p_run = sub.add_parser("run-script", help="执行一条 how.run", parents=[parent]) + p_run.add_argument("--id", required=True, dest="case_id") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + root = Path(args.root).resolve() + if args.cmd == "list": + return cmd_list(root) + if args.cmd == "select": + return cmd_select(args, root) + if args.cmd == "show": + return cmd_show(root, args.case_id) + if args.cmd == "validate": + return cmd_validate(root) + if args.cmd == "export-csv": + return cmd_export_csv(root) + if args.cmd == "next-id": + return cmd_next_id(root, args.category) + if args.cmd == "run-script": + return cmd_run_script(root, args.case_id) + parser.error(f"未知命令 {args.cmd}") + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/tester/scripts/generate_cases_csv.py b/skills/tester/scripts/generate_cases_csv.py new file mode 100755 index 0000000..c4b87e2 --- /dev/null +++ b/skills/tester/scripts/generate_cases_csv.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""从 .pouch/tester/cases/*.yaml 生成给人看的 cases.csv。不要手改 CSV。""" + +from __future__ import annotations + +import argparse +import csv +import sys +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) + +FIELDS = ( + "category", + "id", + "maps_to", + "title", + "what", + "suite", + "risk", + "gate", + "surface", + "session", + "status", + "requires", + "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), + str(case.get("id") or ""), + ), + ) + + +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: + writer.writerow( + { + "category": case.get("category") or "", + "id": case.get("id") or "", + "maps_to": case.get("maps_to") or "", + "title": case.get("title") or "", + "what": one_line(case.get("what")), + "suite": case.get("suite") or "", + "risk": case.get("risk") or "", + "gate": case.get("gate") or "", + "surface": case.get("surface") or "", + "session": case.get("session") or "", + "status": case.get("status") or "active", + "requires": one_line(case.get("requires")), + "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="输出路径，默认 /.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()) diff --git a/skills/tester/templates/catalog.yaml b/skills/tester/templates/catalog.yaml new file mode 100644 index 0000000..633c6df --- /dev/null +++ b/skills/tester/templates/catalog.yaml @@ -0,0 +1,26 @@ +version: 1 +updatedAt: "1970-01-01T00:00:00+00:00" +project: + name: "" +id_pattern: "^[A-Z]+-[A-Za-z0-9-]+$" +categories: [] +risks: + - safe + - act + - harm +default_max_risk: safe +vars: {} +auth_env: + act: AUTHORIZE_ACT + harm: AUTHORIZE_HARM +gates: {} +sessions: + browser: + kind: browser + reuse: run + init: "打开目标页面" + mobile: + kind: mobile + reuse: persist + probe: "" + init: "" diff --git a/skills/tester/templates/local.yaml.example b/skills/tester/templates/local.yaml.example new file mode 100644 index 0000000..440f6ff --- /dev/null +++ b/skills/tester/templates/local.yaml.example @@ -0,0 +1,4 @@ +# 复制为 .pouch/tester/local.yaml 后填写。不要提交 local.yaml。 +HOST: "" +AUTHORIZE_ACT: false +AUTHORIZE_HARM: false