tester v2: 术语对齐 pytest/JUnit/Robot，补齐 automation 与 not-run

字段改名（每项取舍的理由见新增的 references/concepts.md）： - 用例 status -> lifecycle, suite -> tag, surface -> driver, cleanup -> teardown, skip_unless -> assume；删除 requires - risk 取值 safe/act/harm -> readonly/mutating/destructive（单调三级，与 gate 正交） - catalog auth_env -> authorize；授权变量统一为 AUTHORIZE_MUTATING / AUTHORIZE_DESTRUCTIVE 新增用例字段 automation（automated/assisted/manual），默认 automated，并在 validate 里机器校验：manual 禁用 driver 与 how.run 且必须有 how.steps；assisted 至少一步以 [人工] 开头。select 新增 --automation。结果词新增 not-run，区分「主动不跑」（skipped：retired/pending/超风险上限/未授权/assume 不成立）与「没轮到」（失败即停后、被分类前置挡住整组、需人工而无人值守）。 status_probe 定为分类级前置条件：catalog 顶层为默认，categories[] 同名键覆盖（缺省继承 / false 关闭 / mapping 整体替换）。select 按分类返回生效值，并把 probe.url 用 vars 展开（solve Agent 无法自行解析 ${VAR} 的问题），同批返回 vars 供 how 里的 ${...} 替换。新增 references/concepts.md：术语表（每词一句唯一定义 + 业界锚点 + 反例）、v1->v2 映射、以及 risk/automation/not-run/status_probe 的设计理由。其余文档（SKILL/README/schema/levels/run/maintain）与三个脚本（catalog、 generate_cases_csv、generate_run_csv）同步更新。
2026-09-16 23:07:08 +08:00
parent ca3156ed93
commit cc9af04a40
12 changed files with 443 additions and 143 deletions
@@ -19,15 +19,16 @@ 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", "pending")
-STATUS_SKIP_REASON = {
+DRIVERS = ("script", "browser", "mobile")
+LIFECYCLES = ("active", "retired", "pending")
+LIFECYCLE_SKIP_REASON = {
    "retired": "retired",
    "pending": "pending（暂缓，见 notes）",
 }
-SUITES = ("smoke", "full", "slow")
-CANONICAL_RISKS = ("safe", "act", "harm")
-RESULTS = ("pass", "fail", "environment-fail", "skipped")
+TAGS = ("smoke", "full", "slow")
+CANONICAL_RISKS = ("readonly", "mutating", "destructive")
+AUTOMATIONS = ("automated", "assisted", "manual")
+RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
 EXPECTED_KINDS = (
    "api-status",
    "api-field",
@@ -146,7 +147,7 @@ def resolve_vars(root: Path, meta: dict[str, Any]) -> dict[str, str]:
            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 {}):
+    for mapping in (meta.get("authorize") 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():
@@ -212,12 +213,12 @@ def cmd_list(root: Path) -> int:
    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"]
-    pending = [c for c in cases if c.get("status") == "pending"]
+    active = [c for c in cases if c.get("lifecycle", "active") == "active"]
+    retired = [c for c in cases if c.get("lifecycle") == "retired"]
+    pending = [c for c in cases if c.get("lifecycle") == "pending"]
    print(f"## 测试目录  active {len(active)} / pending {len(pending)} / retired {len(retired)}")
    print()
-    headers = ["分类", "标题", "条数", "smoke"] + risks + list(SURFACES)
+    headers = ["分类", "标题", "条数", "smoke"] + risks + list(DRIVERS)
    print("| " + " | ".join(headers) + " |")
    print("| " + " | ".join("---" for _ in headers) + " |")

@@ -235,11 +236,11 @@ def cmd_list(root: Path) -> int:
        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"))]
+        row = [cat, str(title), str(len(group)), str(sum(1 for c in group if c.get("tag") == "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)))
+        for driver in DRIVERS:
+            row.append(str(sum(1 for c in group if c.get("driver") == driver)))
        print("| " + " | ".join(row) + " |")
    print()
    print("跑法: 跑 <分类> | 跑 <id,id> | 跑 smoke | 跑 <分类> 风险到 <risk>")
@@ -268,6 +269,75 @@ def public_case(case: dict[str, Any]) -> dict[str, Any]:
    return {k: v for k, v in case.items() if not str(k).startswith("_")}


+PROBE_SOURCES = ("global", "category", "disabled", "none", "invalid")
+
+
+def effective_status_probe(meta: dict[str, Any], category: str) -> tuple[Any, str]:
+    """分类生效的 status_probe（分类级前置条件）。
+
+    返回 (probe, source)；probe 为 None 表示该分类不做探测。
+    source: global（继承顶层）/ category（分类覆盖）/ disabled（分类置 false）/
+    none（顶层和分类都没声明）/ invalid（分类值非法）。
+    """
+    cats = {
+        str(c.get("id")): c
+        for c in meta.get("categories") or []
+        if isinstance(c, dict) and c.get("id")
+    }
+    cat = cats.get(category) or {}
+    if "status_probe" in cat:
+        value = cat.get("status_probe")
+        if value is False:
+            return None, "disabled"
+        if isinstance(value, dict):
+            return value, "category"
+        return None, "invalid"
+    probe = meta.get("status_probe")
+    if isinstance(probe, dict):
+        return probe, "global"
+    return None, "none"
+
+
+def resolve_probe(probe: Any, vars_map: dict[str, str]) -> Any:
+    """把探测定义里的 ${VAR} 展开，Agent 拿到就能直接请求。"""
+    if not isinstance(probe, dict):
+        return probe
+    out = dict(probe)
+    url = out.get("url")
+    if isinstance(url, str):
+        out["url"] = substitute(url, vars_map)
+    stop_if = out.get("stop_if")
+    if isinstance(stop_if, dict):
+        out["stop_if"] = {
+            (substitute(str(key), vars_map) if "${" in str(key) else key): (
+                [
+                    substitute(str(item), vars_map)
+                    if isinstance(item, str) and "${" in item
+                    else item
+                    for item in value
+                ]
+                if isinstance(value, list)
+                else value
+            )
+            for key, value in stop_if.items()
+        }
+    return out
+
+
+def probe_errors(where: str, probe: Any) -> list[str]:
+    """校验一份 status_probe 定义。允许值为 mapping；false 由调用方处理。"""
+    if not isinstance(probe, dict):
+        return [f"{where}: status_probe 必须是 mapping 或 false"]
+    errors: list[str] = []
+    url = probe.get("url")
+    if not isinstance(url, str) or not url.strip():
+        errors.append(f"{where}.url 必填")
+    stop_if = probe.get("stop_if")
+    if stop_if is not None and not isinstance(stop_if, dict):
+        errors.append(f"{where}.stop_if 必须是 mapping（字段 -> 命中值列表）")
+    return errors
+
+
 def run_together_groups(meta: dict[str, Any]) -> list[dict[str, Any]]:
    raw = meta.get("run_together") or []
    if not isinstance(raw, list):
@@ -294,6 +364,7 @@ def expand_run_together(
    selected: list[dict[str, Any]],
    skipped: list[dict[str, str]],
    max_i: int,
+    automation_filter: str = "automated",
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    groups = run_together_groups(meta)
    if not groups:
@@ -318,10 +389,14 @@ def expand_run_together(
                members.append(selected_map[cid])
                continue
            case = by_id.get(cid)
-            if not case or case.get("status", "active") != "active":
+            if not case or case.get("lifecycle", "active") != "active":
                skipped.append({"id": cid, "reason": "run_together 搭档不是 active"})
                continue
-            risk = str(case.get("risk") or "safe")
+            automation = str(case.get("automation") or "automated")
+            if automation_filter != "any" and automation != automation_filter:
+                skipped.append({"id": cid, "reason": f"run_together 搭档 automation={automation}（需人工）"})
+                continue
+            risk = str(case.get("risk") or "readonly")
            try:
                r_i = risk_index(meta, risk)
            except SystemExit:
@@ -354,8 +429,9 @@ def expand_run_together(
 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_risk = args.max_risk or meta.get("default_max_risk") or "readonly"
    max_i = risk_index(meta, max_risk)
+    automation_filter = args.automation or "automated"
    wanted = set(parse_ids(args.ids or []))
    selected: list[dict[str, Any]] = []
    skipped: list[dict[str, str]] = []
@@ -363,12 +439,12 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:

    for case in cases:
        cid = str(case.get("id") or "")
-        if case.get("status", "active") != "active":
+        if case.get("lifecycle", "active") != "active":
            if cid in wanted:
-                status = str(case.get("status"))
+                lifecycle = str(case.get("lifecycle"))
                skipped.append({
                    "id": cid,
-                    "reason": STATUS_SKIP_REASON.get(status, status),
+                    "reason": LIFECYCLE_SKIP_REASON.get(lifecycle, lifecycle),
                })
                missing.discard(cid)
            continue
@@ -379,11 +455,18 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
        else:
            if args.category and case.get("category") != args.category:
                continue
-            if args.suite and case.get("suite") != args.suite:
+            if args.tag and case.get("tag") != args.tag:
                continue
-            if args.surface and case.get("surface") != args.surface:
+            if args.driver and case.get("driver") != args.driver:
                continue
-        risk = str(case.get("risk") or "safe")
+        automation = str(case.get("automation") or "automated")
+        if automation_filter != "any" and automation != automation_filter:
+            skipped.append({
+                "id": cid,
+                "reason": f"automation={automation}（需人工），本轮只跑 automation={automation_filter}",
+            })
+            continue
+        risk = str(case.get("risk") or "readonly")
        try:
            r_i = risk_index(meta, risk)
        except SystemExit:
@@ -401,15 +484,27 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
        )
        return 2

-    selected, together = expand_run_together(meta, cases, selected, skipped, max_i)
+    selected, together = expand_run_together(meta, cases, selected, skipped, max_i, automation_filter)
+
+    # 按分类解析生效的 status_probe，让 Agent 不再自己按 catalog 猜作用范围。
+    vars_map = resolve_vars(root, meta)
+    status_probes: dict[str, dict[str, Any]] = {}
+    for case in selected:
+        cat = str(case.get("category") or "")
+        if cat and cat not in status_probes:
+            probe, source = effective_status_probe(meta, cat)
+            status_probes[cat] = {"source": source, "probe": resolve_probe(probe, vars_map)}

    print(
        json.dumps(
            {
                "max_risk": max_risk,
+                "automation": automation_filter,
                "count": len(selected),
+                "vars": vars_map,
                "skipped": skipped,
                "run_together": together,
+                "status_probe": status_probes,
                "cases": selected,
            },
            ensure_ascii=False,
@@ -449,22 +544,28 @@ def validate(root: Path) -> list[str]:
            errors.append(f"catalog.categories[{i}] 需要 id")
        else:
            cat_ids.append(str(item["id"]))
+        if isinstance(item, dict) and "status_probe" in item:
+            value = item.get("status_probe")
+            if value is not False:
+                errors.extend(probe_errors(f"catalog.categories[{i}].status_probe", value))
+    if "status_probe" in meta:
+        errors.extend(probe_errors("catalog.status_probe", meta.get("status_probe")))
    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 开头")
+    if risks and risks[0] != "readonly":
+        errors.append("catalog.risks 必须以 readonly 开头")
    last_pos = -1
    for name in risks:
        if name not in CANONICAL_RISKS:
            errors.append(
-                f"非法 risk 名 {name!r}，只能是 safe/act/harm"
+                f"非法 risk 名 {name!r}，只能是 readonly/mutating/destructive"
            )
            continue
        pos = CANONICAL_RISKS.index(name)
        if pos <= last_pos:
-            errors.append("catalog.risks 必须按 safe < act < harm 升序且不重复")
+            errors.append("catalog.risks 必须按 readonly < mutating < destructive 升序且不重复")
        last_pos = pos
    default_max = meta.get("default_max_risk")
    if default_max not in risks:
@@ -473,10 +574,10 @@ def validate(root: Path) -> list[str]:
    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 = {}
+    authorize = meta.get("authorize") or {}
+    if authorize and not isinstance(authorize, dict):
+        errors.append("catalog.authorize 必须是 mapping")
+        authorize = {}
    sessions = meta.get("sessions") or {}
    if not isinstance(sessions, dict):
        errors.append("catalog.sessions 必须是 mapping")
@@ -535,26 +636,27 @@ def validate(root: Path) -> list[str]:
            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("lifecycle", "active") not in LIFECYCLES:
+            errors.append(f"{where}: 非法 lifecycle")
+        if case.get("tag") not in TAGS:
+            errors.append(f"{where}: 非法 tag")
        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")
+        if case.get("tag") == "smoke" and case.get("risk") != "readonly":
+            errors.append(f"{where}: tag=smoke 只能 risk=readonly")
        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")
+            if case.get("risk") in (None, "readonly"):
+                errors.append(f"{where}: risk=readonly 不要填 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 risk and risk != "readonly" and risk not in authorize and not gate:
+            errors.append(f"{where}: {risk} 需要 authorize.{risk} 或 gate")
+        automation = str(case.get("automation") or "automated")
+        if automation not in AUTOMATIONS:
+            errors.append(f"{where}: 非法 automation={case.get('automation')}")
+            automation = "automated"
        if not str(case.get("title") or "").strip():
            errors.append(f"{where}: 缺少 title")
        if not str(case.get("what") or "").strip():
@@ -563,18 +665,51 @@ def validate(root: Path) -> list[str]:
        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")
+        steps = how.get("steps") or []
+        if automation == "manual":
+            if case.get("driver"):
+                errors.append(f"{where}: automation=manual 不要填 driver")
+            if how.get("run"):
+                errors.append(f"{where}: automation=manual 不要填 how.run")
+            if not steps:
+                errors.append(f"{where}: automation=manual 需要 how.steps")
+        else:
+            driver = case.get("driver")
+            if driver not in DRIVERS:
+                errors.append(f"{where}: 非法 driver={driver}")
+            if driver == "script" and not how.get("run") and not steps:
+                errors.append(f"{where}: script 需要 how.run 或 how.steps")
+            if driver in ("browser", "mobile"):
+                session = case.get("session")
+                if not session:
+                    errors.append(f"{where}: {driver} 需要 session")
+                elif session not in sessions:
+                    errors.append(f"{where}: session={session} 未在 catalog.sessions 声明")
+                if driver == "browser" and not how.get("url") and not steps:
+                    errors.append(f"{where}: browser 需要 how.url 或 how.steps")
+                if driver == "mobile" and not steps:
+                    errors.append(f"{where}: mobile 需要 how.steps")
+            if automation == "assisted":
+                if not steps:
+                    errors.append(f"{where}: automation=assisted 需要 how.steps")
+                elif not any(str(step).startswith("[人工]") for step in steps):
+                    errors.append(f"{where}: automation=assisted 的 how.steps 至少一步以 [人工] 开头")
+        assume = case.get("assume")
+        if assume is not None:
+            if not isinstance(assume, str) or not assume.strip():
+                errors.append(f"{where}: assume 必须是非空字符串")
+            else:
+                forbidden: set[str] = set()
+                risk_name = str(case.get("risk") or "")
+                if risk_name and risk_name != "readonly":
+                    env = authorize.get(risk_name)
+                    if env:
+                        forbidden.add(str(env))
+                if gate and gate in gates:
+                    forbidden.add(str(gates[gate]))
+                for var in sorted(forbidden):
+                    if re.search(rf"\b{re.escape(var)}\b", assume):
+                        errors.append(f"{where}: assume 重复了 {var}（已由 risk/gate 授权）")
        expected = case.get("expected")
        if not isinstance(expected, list) or not expected:
            errors.append(f"{where}: expected 必须是非空列表")
@@ -597,8 +732,8 @@ def cmd_validate(root: Path) -> int:
        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")
-    pending = sum(1 for c in cases if c.get("status") == "pending")
+    active = sum(1 for c in cases if c.get("lifecycle", "active") == "active")
+    pending = sum(1 for c in cases if c.get("lifecycle") == "pending")
    print(f"✓ validate 通过  active={active} pending={pending} total={len(cases)}")
    return 0

@@ -627,11 +762,11 @@ def cmd_next_id(root: Path, category: str) -> int:


 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"
+    risk = str(case.get("risk") or "readonly")
+    risks = list(meta.get("risks") or ["readonly"])
+    lowest = risks[0] if risks else "readonly"
    if risk != lowest:
-        env_name = (meta.get("auth_env") or {}).get(risk)
+        env_name = (meta.get("authorize") 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"):
@@ -652,15 +787,24 @@ def cmd_run_script(root: Path, case_id: str) -> int:
        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":
-        status = str(case.get("status"))
-        reason = STATUS_SKIP_REASON.get(status, status)
+    if case.get("lifecycle", "active") != "active":
+        lifecycle = str(case.get("lifecycle"))
+        reason = LIFECYCLE_SKIP_REASON.get(lifecycle, lifecycle)
        print(json.dumps({"id": public_id, "result": "skipped", "evidence": reason}, ensure_ascii=False))
        return 0
-    if case.get("surface") != "script":
+    automation = str(case.get("automation") or "automated")
+    if automation != "automated":
        print(
            json.dumps(
-                {"id": public_id, "result": "environment-fail", "evidence": "run-script 只能跑 surface=script"},
+                {"id": public_id, "result": "not-run", "evidence": f"automation={automation}，需人工，run-script 不执行"},
+                ensure_ascii=False,
+            )
+        )
+        return 0
+    if case.get("driver") != "script":
+        print(
+            json.dumps(
+                {"id": public_id, "result": "environment-fail", "evidence": "run-script 只能跑 driver=script"},
                ensure_ascii=False,
            )
        )
@@ -673,7 +817,7 @@ def cmd_run_script(root: Path, case_id: str) -> int:
            )
        )
        return 2
-    risk = str(case.get("risk") or "safe")
+    risk = str(case.get("risk") or "readonly")
    if not authorized(vars_map, meta, case):
        print(
            json.dumps(
@@ -745,9 +889,15 @@ def build_parser() -> argparse.ArgumentParser:
    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("--tag", choices=TAGS)
    p_sel.add_argument("--max-risk")
-    p_sel.add_argument("--surface", choices=SURFACES)
+    p_sel.add_argument("--driver", choices=DRIVERS)
+    p_sel.add_argument(
+        "--automation",
+        choices=[*AUTOMATIONS, "any"],
+        default=None,
+        help="默认 automated；any 表示不按人工参与度筛选",
+    )
    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])