feat: update
This commit is contained in:
@@ -95,6 +95,13 @@ def create_ship(
|
||||
(spec_dir / "risks.yaml").write_text("[]\n", encoding="utf-8")
|
||||
|
||||
ensure_stage_layout(dest)
|
||||
manifest = _replace(
|
||||
(templates / "test-manifest.template.yaml").read_text(encoding="utf-8"),
|
||||
mapping,
|
||||
)
|
||||
(dest / "2-develop" / "test-manifest.yaml").write_text(
|
||||
manifest, encoding="utf-8"
|
||||
)
|
||||
|
||||
issues = validate_ship_dir(dest)
|
||||
if issues:
|
||||
|
||||
@@ -223,6 +223,19 @@ def _validate_tasks(path: Path, ship_id: Any, kind: Any) -> list[str]:
|
||||
verification = row.get("verification")
|
||||
if not isinstance(verification, list) or not verification:
|
||||
_issue(issues, f"{prefix}.verification 必须是非空列表")
|
||||
test_plan = row.get("testPlan")
|
||||
if test_plan is not None:
|
||||
plan = _mapping(test_plan, f"{prefix}.testPlan", issues)
|
||||
if plan is not None:
|
||||
policy = plan.get("policy")
|
||||
if policy not in {"required", "not_applicable"}:
|
||||
_issue(
|
||||
issues,
|
||||
f"{prefix}.testPlan.policy 无效: {policy!r}(required|not_applicable)",
|
||||
)
|
||||
manifest = plan.get("manifest")
|
||||
if not isinstance(manifest, str) or not manifest.strip():
|
||||
_issue(issues, f"{prefix}.testPlan.manifest 不能为空")
|
||||
for i, item in enumerate(tasks):
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user