laily
a47e899702
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.
635 lines
22 KiB
Python
Executable File
635 lines
22 KiB
Python
Executable File
#!/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("跑法: 跑 <分类> | 跑 <id,id> | 跑 smoke | 跑 <分类> 风险到 <risk>")
|
||
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-<CAT>-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())
|