139 lines
5.1 KiB
Python
139 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate and inspect a ship test manifest."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from ship_lib import load_yaml
|
|
|
|
POLICIES = {"required", "not_applicable"}
|
|
CHANGES = {"added", "changed"}
|
|
KINDS = {"unit", "integration", "e2e", "static"}
|
|
MODES = {"schema", "verify", "regress"}
|
|
|
|
|
|
def _issue(issues: list[str], message: str) -> None:
|
|
issues.append(message)
|
|
|
|
|
|
def _mapping(value: Any, path: str, issues: list[str]) -> dict[str, Any] | None:
|
|
if not isinstance(value, dict):
|
|
_issue(issues, f"{path} 必须是 mapping")
|
|
return None
|
|
return value
|
|
|
|
|
|
def validate_manifest(
|
|
manifest_path: Path,
|
|
*,
|
|
expected_ship_id: str | None = None,
|
|
policy: str = "required",
|
|
mode: str = "schema",
|
|
) -> tuple[list[str], dict[str, Any] | None]:
|
|
issues: list[str] = []
|
|
if mode not in MODES:
|
|
return [f"mode 无效: {mode!r}"], None
|
|
try:
|
|
data = load_yaml(manifest_path)
|
|
except Exception as exc: # noqa: BLE001
|
|
return [f"无法解析 test manifest: {exc}"], None
|
|
root = _mapping(data, "test-manifest", issues)
|
|
if root is None:
|
|
return issues, None
|
|
|
|
ship_id = root.get("shipId")
|
|
if not isinstance(ship_id, str) or not ship_id.strip():
|
|
_issue(issues, "test-manifest.shipId 不能为空")
|
|
elif expected_ship_id and ship_id != expected_ship_id:
|
|
_issue(issues, f"test-manifest.shipId 不一致: {ship_id!r} != {expected_ship_id!r}")
|
|
|
|
tests = root.get("tests")
|
|
if not isinstance(tests, list):
|
|
_issue(issues, "test-manifest.tests 必须是列表")
|
|
tests = []
|
|
ids: set[str] = set()
|
|
for index, item in enumerate(tests):
|
|
prefix = f"tests[{index}]"
|
|
row = _mapping(item, prefix, issues)
|
|
if row is None:
|
|
continue
|
|
test_id = row.get("id")
|
|
if not isinstance(test_id, str) or not test_id.strip():
|
|
_issue(issues, f"{prefix}.id 不能为空")
|
|
elif test_id in ids:
|
|
_issue(issues, f"重复测试 id: {test_id}")
|
|
else:
|
|
ids.add(test_id)
|
|
for field in ("runner", "target", "selector"):
|
|
if not isinstance(row.get(field), str) or not row[field].strip():
|
|
_issue(issues, f"{prefix}.{field} 不能为空")
|
|
if row.get("kind") not in KINDS:
|
|
_issue(issues, f"{prefix}.kind 无效: {row.get('kind')!r}")
|
|
if row.get("change") not in CHANGES:
|
|
_issue(issues, f"{prefix}.change 无效: {row.get('change')!r}")
|
|
covers = row.get("covers")
|
|
if not isinstance(covers, list) or not covers or not all(
|
|
isinstance(value, str) and value.strip() for value in covers
|
|
):
|
|
_issue(issues, f"{prefix}.covers 必须是非空字符串列表")
|
|
|
|
if policy not in POLICIES:
|
|
_issue(issues, f"test policy 无效: {policy!r}")
|
|
gap = root.get("testGap")
|
|
if gap is not None:
|
|
gap_row = _mapping(gap, "testGap", issues)
|
|
if gap_row is not None:
|
|
if not isinstance(gap_row.get("reason"), str) or not gap_row["reason"].strip():
|
|
_issue(issues, "testGap.reason 不能为空")
|
|
if mode in {"verify", "regress"} and not (
|
|
isinstance(gap_row.get("acceptedBy"), str)
|
|
and gap_row["acceptedBy"].strip()
|
|
):
|
|
_issue(issues, "testGap 在 verify/regress 前必须有 acceptedBy")
|
|
|
|
if mode in {"verify", "regress"}:
|
|
if policy == "required" and not tests and gap is None:
|
|
_issue(issues, "policy=required 时必须登记至少一个测试，或填写已批准的 testGap")
|
|
if policy == "not_applicable" and tests:
|
|
_issue(issues, "policy=not_applicable 时 tests 必须为空")
|
|
if policy == "not_applicable" and gap is None:
|
|
_issue(issues, "policy=not_applicable 时必须说明 testGap.reason")
|
|
|
|
return issues, root
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="校验并列出 ship 测试 manifest")
|
|
parser.add_argument("manifest", help="test-manifest.yaml 路径")
|
|
parser.add_argument("--ship-id")
|
|
parser.add_argument("--policy", default="required", choices=sorted(POLICIES))
|
|
parser.add_argument("--mode", default="schema", choices=sorted(MODES))
|
|
parser.add_argument("--json", action="store_true", dest="as_json")
|
|
args = parser.parse_args()
|
|
|
|
issues, data = validate_manifest(
|
|
Path(args.manifest),
|
|
expected_ship_id=args.ship_id,
|
|
policy=args.policy,
|
|
mode=args.mode,
|
|
)
|
|
if args.as_json:
|
|
print(json.dumps({"ok": not issues, "issues": issues, "tests": (data or {}).get("tests", [])}, ensure_ascii=False))
|
|
elif issues:
|
|
for issue in issues:
|
|
print(f"✗ {issue}", file=sys.stderr)
|
|
else:
|
|
print(f"✓ test manifest 通过: {args.manifest}")
|
|
for row in (data or {}).get("tests", []):
|
|
print(f"- {row['id']}: {row['runner']} {row['target']} {row['selector']}")
|
|
return 1 if issues else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|