laily
12e00bd594
Keep pouch naming and .pouch/ack project state, and bring in ACK regression mode, deployer test-environment binding, and manage-release updates from main.
379 lines
13 KiB
Python
379 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""校验 ACK 项目回归目录及其任务引用。
|
||
|
||
权威结构位于 templates/regression.schema.json。jsonschema 是可选依赖；内置规则
|
||
始终检查用例 ID、验收信号和跨文件引用。本脚本只解析数据，不执行 steps。
|
||
|
||
用法:
|
||
python3 validate_regression.py .pouch/ack/regression.yaml
|
||
python3 validate_regression.py .pouch/ack/regression.yaml \
|
||
--tasks .pouch/ack/tasks.yaml
|
||
|
||
退出码: 0 通过 / 1 校验失败 / 2 环境或用法错误。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from yaml_subset import (
|
||
DuplicateKeyError,
|
||
YamlSubsetError,
|
||
load_json_unique,
|
||
load_yaml_subset,
|
||
make_unique_pyyaml_loader,
|
||
)
|
||
|
||
|
||
ID_RE = re.compile(r"^REG-[A-Za-z0-9][A-Za-z0-9-]*$")
|
||
TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
|
||
RELATIVE_PATH_RE = re.compile(r"^[A-Za-z0-9._/-]+$")
|
||
STATUSES = {"active", "retired"}
|
||
SUITES = {"smoke", "full"}
|
||
SURFACES = {"browser", "api"}
|
||
SOURCE_KINDS = {"feature", "bug"}
|
||
EXPECTED_KINDS = {"visible-text", "api-status", "api-field", "url", "interaction"}
|
||
CASE_FIELDS = {
|
||
"id",
|
||
"title",
|
||
"status",
|
||
"source",
|
||
"suite",
|
||
"surface",
|
||
"setup",
|
||
"steps",
|
||
"expected",
|
||
"automationRef",
|
||
}
|
||
SOURCE_FIELDS = {"taskId", "kind"}
|
||
EXPECTED_FIELDS = {"kind", "value"}
|
||
TOP_LEVEL_FIELDS = {"version", "updatedAt", "project", "cases"}
|
||
|
||
|
||
def _nonempty(value: Any) -> bool:
|
||
return isinstance(value, str) and bool(value.strip())
|
||
|
||
|
||
def _mapping(value: Any) -> bool:
|
||
return isinstance(value, dict)
|
||
|
||
|
||
def _reject_unknown(
|
||
value: dict[str, Any],
|
||
allowed: set[str],
|
||
where: str,
|
||
errors: list[str],
|
||
) -> None:
|
||
for field in sorted(set(value) - allowed):
|
||
errors.append(f"{where}: 未知字段 {field!r}")
|
||
|
||
|
||
def load_yaml(path: Path, label: str) -> dict[str, Any]:
|
||
try:
|
||
content = path.read_text(encoding="utf-8")
|
||
except OSError as exc:
|
||
sys.stderr.write(f"{label}读取失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
|
||
if path.suffix.lower() == ".json":
|
||
try:
|
||
data = load_json_unique(content)
|
||
except (json.JSONDecodeError, DuplicateKeyError) as exc:
|
||
sys.stderr.write(f"{label} JSON 解析失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
else:
|
||
try:
|
||
import yaml # type: ignore
|
||
except ImportError:
|
||
try:
|
||
data = load_yaml_subset(content)
|
||
except YamlSubsetError as exc:
|
||
sys.stderr.write(f"{label} YAML 子集解析失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
else:
|
||
try:
|
||
data = yaml.load(content, Loader=make_unique_pyyaml_loader(yaml))
|
||
except yaml.YAMLError as exc: # type: ignore
|
||
sys.stderr.write(f"{label} YAML 解析失败: {exc}\n")
|
||
raise SystemExit(1)
|
||
|
||
if not isinstance(data, dict):
|
||
sys.stderr.write(f"{label}顶层必须是对象（mapping）\n")
|
||
raise SystemExit(1)
|
||
return data
|
||
|
||
|
||
def infer_project_root(document_path: Path) -> Path | None:
|
||
lexical = document_path.expanduser().absolute()
|
||
parent = lexical.parent
|
||
if parent.name == "ack" and parent.parent.name == "docs":
|
||
return parent.parent.parent.resolve()
|
||
for candidate in (parent, *parent.parents):
|
||
if (candidate / ".git").exists():
|
||
return candidate.resolve()
|
||
return None
|
||
|
||
|
||
def _validate_expected(value: Any, where: str, errors: list[str]) -> None:
|
||
if not isinstance(value, list) or not value:
|
||
errors.append(f"{where}: 必须是非空列表")
|
||
return
|
||
for index, item in enumerate(value):
|
||
item_where = f"{where}[{index}]"
|
||
if not _mapping(item):
|
||
errors.append(f"{item_where}: 必须是对象")
|
||
continue
|
||
_reject_unknown(item, EXPECTED_FIELDS, item_where, errors)
|
||
if item.get("kind") not in EXPECTED_KINDS:
|
||
errors.append(f"{item_where}.kind: 必须是 {sorted(EXPECTED_KINDS)}")
|
||
if not _nonempty(item.get("value")):
|
||
errors.append(f"{item_where}.value: 必须是非空字符串")
|
||
|
||
|
||
def validate_builtin(data: dict[str, Any]) -> list[str]:
|
||
errors: list[str] = []
|
||
_reject_unknown(data, TOP_LEVEL_FIELDS, "<root>", errors)
|
||
if data.get("version") != 1 or isinstance(data.get("version"), bool):
|
||
errors.append("version 必须是整数 1")
|
||
if not _nonempty(data.get("updatedAt")):
|
||
errors.append("updatedAt 必须是非空字符串")
|
||
|
||
project = data.get("project")
|
||
if not _mapping(project):
|
||
errors.append("project 必须是对象")
|
||
else:
|
||
_reject_unknown(project, {"name"}, "project", errors)
|
||
if not _nonempty(project.get("name")):
|
||
errors.append("project.name 必须是非空字符串")
|
||
|
||
cases = data.get("cases")
|
||
if not isinstance(cases, list):
|
||
errors.append("cases 必须是列表")
|
||
return errors
|
||
|
||
seen: set[str] = set()
|
||
for index, case in enumerate(cases):
|
||
where = f"cases[{index}]"
|
||
if not _mapping(case):
|
||
errors.append(f"{where}: 必须是对象")
|
||
continue
|
||
_reject_unknown(case, CASE_FIELDS, where, errors)
|
||
missing = sorted(CASE_FIELDS - {"automationRef"} - set(case))
|
||
for field in missing:
|
||
errors.append(f"{where}.{field}: 必填")
|
||
|
||
case_id = case.get("id")
|
||
if not isinstance(case_id, str) or ID_RE.fullmatch(case_id) is None:
|
||
errors.append(f"{where}.id: 必须使用 REG-<id> 格式")
|
||
elif case_id in seen:
|
||
errors.append(f"{where}.id: 不能重复 {case_id!r}")
|
||
else:
|
||
seen.add(case_id)
|
||
|
||
if not _nonempty(case.get("title")):
|
||
errors.append(f"{where}.title: 必须是非空字符串")
|
||
if case.get("status") not in STATUSES:
|
||
errors.append(f"{where}.status: 必须是 {sorted(STATUSES)}")
|
||
if case.get("suite") not in SUITES:
|
||
errors.append(f"{where}.suite: 必须是 {sorted(SUITES)}")
|
||
if case.get("surface") not in SURFACES:
|
||
errors.append(f"{where}.surface: 必须是 {sorted(SURFACES)}")
|
||
if not _nonempty(case.get("setup")):
|
||
errors.append(f"{where}.setup: 必须是非空字符串")
|
||
|
||
source = case.get("source")
|
||
if not _mapping(source):
|
||
errors.append(f"{where}.source: 必须是对象")
|
||
else:
|
||
_reject_unknown(source, SOURCE_FIELDS, f"{where}.source", errors)
|
||
task_id = source.get("taskId")
|
||
if not isinstance(task_id, str) or TASK_ID_RE.fullmatch(task_id) is None:
|
||
errors.append(f"{where}.source.taskId: 必须是任务 ID")
|
||
if source.get("kind") not in SOURCE_KINDS:
|
||
errors.append(f"{where}.source.kind: 必须是 {sorted(SOURCE_KINDS)}")
|
||
|
||
steps = case.get("steps")
|
||
if (
|
||
not isinstance(steps, list)
|
||
or not steps
|
||
or any(not _nonempty(step) for step in steps)
|
||
):
|
||
errors.append(f"{where}.steps: 必须是非空字符串列表")
|
||
|
||
_validate_expected(case.get("expected"), f"{where}.expected", errors)
|
||
|
||
automation_ref = case.get("automationRef")
|
||
if automation_ref is not None:
|
||
if (
|
||
not isinstance(automation_ref, str)
|
||
or automation_ref.startswith("/")
|
||
or "\\" in automation_ref
|
||
or ".." in automation_ref.split("/")
|
||
or RELATIVE_PATH_RE.fullmatch(automation_ref) is None
|
||
):
|
||
errors.append(f"{where}.automationRef: 必须是项目内相对路径")
|
||
return errors
|
||
|
||
|
||
def validate_with_schema(data: dict[str, Any], schema_path: Path) -> list[str]:
|
||
import jsonschema # type: ignore
|
||
|
||
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
||
validator = jsonschema.Draft7Validator(schema)
|
||
errors = []
|
||
for error in sorted(validator.iter_errors(data), key=lambda item: list(item.path)):
|
||
location = "/".join(str(part) for part in error.path) or "<root>"
|
||
errors.append(f"[schema] {location}: {error.message}")
|
||
return errors
|
||
|
||
|
||
def validate_tasks_link(
|
||
regression: dict[str, Any],
|
||
tasks: dict[str, Any],
|
||
) -> list[str]:
|
||
errors: list[str] = []
|
||
project = tasks.get("project")
|
||
if not isinstance(project, dict):
|
||
return ["[tasks] project 必须是对象"]
|
||
regression_file = project.get("regressionFile")
|
||
if regression_file != ".pouch/ack/regression.yaml":
|
||
errors.append(
|
||
"[tasks] project.regressionFile 必须固定为 .pouch/ack/regression.yaml"
|
||
)
|
||
if not isinstance(tasks.get("regressionRuns"), list):
|
||
errors.append("引用 regressionFile 的任务板必须包含 regressionRuns 列表")
|
||
|
||
regression_project = regression.get("project")
|
||
if (
|
||
isinstance(regression_project, dict)
|
||
and _nonempty(regression_project.get("name"))
|
||
and _nonempty(project.get("name"))
|
||
and regression_project["name"] != project["name"]
|
||
):
|
||
errors.append("regression.project.name 必须与 tasks.project.name 一致")
|
||
|
||
case_ids = {
|
||
case.get("id")
|
||
for case in regression.get("cases") or []
|
||
if isinstance(case, dict) and isinstance(case.get("id"), str)
|
||
}
|
||
task_ids = {
|
||
task.get("id")
|
||
for task in tasks.get("tasks") or []
|
||
if isinstance(task, dict) and isinstance(task.get("id"), str)
|
||
}
|
||
for index, task in enumerate(tasks.get("tasks") or []):
|
||
if not isinstance(task, dict):
|
||
continue
|
||
where = f"[tasks] tasks[{index}]"
|
||
refs = task.get("regressionRefs")
|
||
if refs is None:
|
||
continue
|
||
if not isinstance(refs, list):
|
||
errors.append(f"{where}.regressionRefs: 必须是列表")
|
||
continue
|
||
seen: set[str] = set()
|
||
for ref_index, ref in enumerate(refs):
|
||
ref_where = f"{where}.regressionRefs[{ref_index}]"
|
||
if not isinstance(ref, str) or ID_RE.fullmatch(ref) is None:
|
||
errors.append(f"{ref_where}: 必须使用 REG-<id> 格式")
|
||
continue
|
||
if ref in seen:
|
||
errors.append(f"{ref_where}: 不能重复 {ref!r}")
|
||
seen.add(ref)
|
||
if ref not in case_ids:
|
||
errors.append(f"{ref_where}: 未知用例 {ref!r}")
|
||
|
||
for index, run in enumerate(tasks.get("regressionRuns") or []):
|
||
if not isinstance(run, dict):
|
||
continue
|
||
where = f"[tasks] regressionRuns[{index}]"
|
||
for case_id in run.get("caseIds") or []:
|
||
if isinstance(case_id, str) and case_id not in case_ids:
|
||
errors.append(f"{where}.caseIds: 未知用例 {case_id!r}")
|
||
source_task_ids = run.get("taskIds") or []
|
||
if isinstance(source_task_ids, list):
|
||
for task_id in source_task_ids:
|
||
if isinstance(task_id, str) and task_id not in task_ids:
|
||
errors.append(f"{where}.taskIds: 未知任务 {task_id!r}")
|
||
return errors
|
||
|
||
|
||
def validate_all(
|
||
data: dict[str, Any],
|
||
schema_path: Path,
|
||
*,
|
||
use_schema: bool | None = None,
|
||
) -> tuple[list[str], str]:
|
||
if use_schema is None:
|
||
try:
|
||
import jsonschema # type: ignore # noqa: F401
|
||
except ImportError:
|
||
use_schema = False
|
||
else:
|
||
use_schema = schema_path.is_file()
|
||
|
||
errors = validate_builtin(data)
|
||
if use_schema and schema_path.is_file():
|
||
errors = validate_with_schema(data, schema_path) + errors
|
||
mode = f"schema ({schema_path.name}) + 内置结构"
|
||
else:
|
||
mode = "内置结构"
|
||
return list(dict.fromkeys(errors)), mode
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description="校验 ACK 项目回归目录")
|
||
parser.add_argument(
|
||
"regression", nargs="?", default=".pouch/ack/regression.yaml"
|
||
)
|
||
parser.add_argument("--tasks", help="关联的 .pouch/ack/tasks.yaml")
|
||
parser.add_argument("--schema", help="regression.schema.json 路径（默认自动探测）")
|
||
args = parser.parse_args(argv)
|
||
|
||
regression_path = Path(args.regression)
|
||
if not regression_path.is_file():
|
||
sys.stderr.write(f"找不到回归目录: {regression_path}\n")
|
||
return 2
|
||
|
||
schema_path = (
|
||
Path(args.schema)
|
||
if args.schema
|
||
else Path(__file__).resolve().parent.parent
|
||
/ "templates"
|
||
/ "regression.schema.json"
|
||
)
|
||
if args.schema and not schema_path.is_file():
|
||
sys.stderr.write(f"找不到 schema: {schema_path}\n")
|
||
return 2
|
||
|
||
data = load_yaml(regression_path, "回归目录")
|
||
errors, mode = validate_all(data, schema_path)
|
||
|
||
if args.tasks:
|
||
tasks_path = Path(args.tasks)
|
||
if not tasks_path.is_file():
|
||
sys.stderr.write(f"找不到任务板: {tasks_path}\n")
|
||
return 2
|
||
tasks = load_yaml(tasks_path, "任务板")
|
||
errors.extend(validate_tasks_link(data, tasks))
|
||
mode += " + tasks 引用"
|
||
|
||
if errors:
|
||
sys.stderr.write(f"回归目录校验失败（{mode}），共 {len(errors)} 项：\n")
|
||
for error in errors:
|
||
sys.stderr.write(f"- {error}\n")
|
||
return 1
|
||
|
||
sys.stdout.write(f"回归目录校验通过（{mode}）\n")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|