fix tester multi-root fan-out execution

skills/tester/scripts/catalog.py

@@ -4,11 +4,16 @@
 from __future__ import annotations
 
 import argparse
+import contextlib
+import io
 import json
 import os
 import re
 import subprocess
 import sys
+import urllib.parse
+import urllib.request
+import zipfile
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -59,6 +64,60 @@ def catalog_path(root: Path) -> Path:
     return catalog_dir(root) / "catalog.yaml"
 
 
+def discover_or_self(root: Path) -> list[Path]:
+    """Return all independent tester projects below a parent repository."""
+    root = root.resolve()
+    return discover_catalog_roots(root)
+
+
+def cmd_fanout(args: argparse.Namespace, roots: list[Path]) -> int:
+    """Run a catalog command independently for every discovered tester root."""
+    results: list[dict[str, Any]] = []
+    for child_root in roots:
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            if args.cmd == "list":
+                exit_code = cmd_list(child_root)
+            elif args.cmd == "select":
+                exit_code = cmd_select(args, child_root)
+            elif args.cmd == "show":
+                exit_code = cmd_show(child_root, args.case_id)
+            elif args.cmd == "validate":
+                exit_code = cmd_validate(child_root)
+            elif args.cmd == "export-csv":
+                exit_code = cmd_export_csv(child_root)
+            elif args.cmd == "run-script":
+                exit_code = cmd_run_script(child_root, args.case_id)
+            else:
+                exit_code = 2
+                print(f"不支持从父仓库 fan-out 命令: {args.cmd}")
+        raw = output.getvalue().strip()
+        try:
+            result: Any = json.loads(raw) if raw else None
+        except json.JSONDecodeError:
+            result = raw
+        results.append({
+            "root": str(child_root),
+            "exit": exit_code,
+            "result": result,
+        })
+
+    if args.cmd == "select":
+        print(json.dumps({"mode": "fan-out", "roots": results}, ensure_ascii=False, indent=2))
+    else:
+        for item in results:
+            print(f"\n## tester root: {item['root']} (exit={item['exit']})")
+            result = item["result"]
+            if isinstance(result, (dict, list)):
+                print(json.dumps(result, ensure_ascii=False, indent=2))
+            elif result:
+                print(result)
+    return 0 if all(item["exit"] == 0 for item in results) else 2
+
+
+FANOUT_COMMANDS = {"list", "select", "show", "validate", "export-csv", "run-script"}
+
+
 def discover_catalog_roots(root: Path) -> list[Path]:
     """Find tester projects owned by a repository and its Git submodules.
 
@@ -560,6 +619,142 @@ def expand_run_together(
     return selected, hit_groups
 
 
+def parse_platform_pairs(values: list[str], label: str) -> tuple[dict[str, str], list[str]]:
+    pairs: dict[str, str] = {}
+    errors: list[str] = []
+    for raw in values:
+        platform, sep, value = raw.partition("=")
+        platform = platform.strip().lower()
+        value = value.strip()
+        if not sep or platform not in ("android", "ios") or not value:
+            errors.append(f"{label} 格式必须是 android=<值> 或 ios=<值>: {raw}")
+            continue
+        if platform in pairs:
+            errors.append(f"{label} 重复平台: {platform}")
+            continue
+        pairs[platform] = value
+    return pairs, errors
+
+
+def safe_locator(locator: str) -> str:
+    parsed = urllib.parse.urlsplit(locator)
+    if parsed.scheme in ("http", "https"):
+        path = parsed.path or "/"
+        return f"{parsed.scheme}://{parsed.netloc}{path}"
+    return locator
+
+
+def artifact_ready(locator: str) -> tuple[bool, str]:
+    parsed = urllib.parse.urlsplit(locator)
+    if parsed.scheme in ("http", "https"):
+        try:
+            request = urllib.request.Request(locator, method="HEAD")
+            with urllib.request.urlopen(request, timeout=10) as response:
+                if 200 <= response.status < 400:
+                    return True, f"HTTP {response.status}"
+                return False, f"HTTP {response.status}"
+        except Exception as exc:
+            return False, f"下载地址不可达: {type(exc).__name__}: {exc}"
+
+    path = Path(locator).expanduser()
+    if not path.is_file():
+        return False, "文件不存在"
+    if path.stat().st_size <= 0:
+        return False, "文件为空"
+    if path.suffix.lower() in (".apk", ".ipa") and not zipfile.is_zipfile(path):
+        return False, "不是有效的 APK/IPA zip 产物"
+    return True, f"文件 {path.stat().st_size} bytes"
+
+
+def runtime_binding(root: Path, binding_id: str) -> tuple[bool, str]:
+    config_path = root / ".pouch" / "agent.yaml"
+    if not config_path.is_file():
+        return False, f"缺少 {config_path}"
+    try:
+        config = load_yaml(config_path) or {}
+    except Exception as exc:
+        return False, f"读取 agent.yaml 失败: {exc}"
+    binding = (config.get("bindings") or {}).get(binding_id)
+    if not isinstance(binding, dict):
+        return False, "binding 不存在"
+    if binding.get("adapter") != "tester":
+        return False, f"adapter={binding.get('adapter')!r}，不是 tester"
+    location = binding.get("location")
+    if not location:
+        return False, "未配置 location"
+    return True, f"location={location}"
+
+
+def mobile_preflight(
+    root: Path,
+    meta: dict[str, Any],
+    selected: list[dict[str, Any]],
+    platforms_raw: list[str],
+    artifacts_raw: list[str],
+    bindings_raw: list[str],
+) -> dict[str, Any]:
+    mobile_selected = [case for case in selected if case.get("driver") == "mobile"]
+    if not mobile_selected:
+        return {"required": False, "status": "not-needed", "checks": [], "errors": []}
+
+    platforms = [item.strip().lower() for item in platforms_raw if item.strip()]
+    errors: list[str] = []
+    if not platforms:
+        errors.append("选中了 mobile 用例，必须显式提供 --platform android/ios")
+    if len(platforms) != len(set(platforms)):
+        errors.append("--platform 不能重复")
+    invalid_platforms = [item for item in platforms if item not in ("android", "ios")]
+    if invalid_platforms:
+        errors.append("非法平台: " + ", ".join(invalid_platforms))
+
+    artifacts, pair_errors = parse_platform_pairs(artifacts_raw, "--artifact")
+    errors.extend(pair_errors)
+    bindings, pair_errors = parse_platform_pairs(bindings_raw, "--binding")
+    errors.extend(pair_errors)
+    contract = meta.get("artifacts")
+    if not isinstance(contract, dict) or not contract:
+        errors.append("catalog 缺少 artifacts 产物契约")
+
+    checks: list[dict[str, Any]] = []
+    for platform in platforms:
+        if platform not in ("android", "ios"):
+            continue
+        spec = (contract or {}).get(platform) if isinstance(contract, dict) else None
+        if not isinstance(spec, dict):
+            errors.append(f"{platform}: catalog.artifacts 未声明")
+        if platform not in artifacts:
+            errors.append(f"{platform}: 缺少 --artifact {platform}=<路径或下载地址>")
+        else:
+            ok, detail = artifact_ready(artifacts[platform])
+            checks.append({
+                "platform": platform,
+                "artifact": safe_locator(artifacts[platform]),
+                "artifact_ok": ok,
+                "artifact_detail": detail,
+            })
+            if not ok:
+                errors.append(f"{platform}: 产物不可用: {detail}")
+
+        binding_id = bindings.get(platform, f"tester.{platform}")
+        ok, detail = runtime_binding(root, binding_id)
+        checks.append({
+            "platform": platform,
+            "binding": binding_id,
+            "binding_ok": ok,
+            "binding_detail": detail,
+        })
+        if not ok:
+            errors.append(f"{platform}: binding {binding_id} 不可用: {detail}")
+
+    return {
+        "required": True,
+        "status": "blocked" if errors else "ready",
+        "platforms": platforms,
+        "checks": checks,
+        "errors": errors,
+    }
+
+
 def cmd_select(args: argparse.Namespace, root: Path) -> int:
     meta = load_catalog_meta(root)
     cases = load_all_cases(root)
@@ -620,6 +815,18 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
 
     selected, together = expand_run_together(meta, cases, selected, skipped, max_i, automation_filter)
 
+    preflight = mobile_preflight(
+        root,
+        meta,
+        selected,
+        args.platform or [],
+        args.artifacts or [],
+        args.bindings or [],
+    )
+    if preflight["status"] == "blocked":
+        print(json.dumps({"preflight": preflight}, ensure_ascii=False, indent=2))
+        return 2
+
     # 按分类解析生效的 status_probe，让 Agent 不再自己按 catalog 猜作用范围。
     vars_map = resolve_vars(root, meta)
     status_probes: dict[str, dict[str, Any]] = {}
@@ -636,6 +843,8 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
                 "automation": automation_filter,
                 "count": len(selected),
                 "vars": vars_map,
+                "artifacts": meta.get("artifacts") or {},
+                "preflight": preflight,
                 "skipped": skipped,
                 "run_together": together,
                 "status_probe": status_probes,
@@ -718,6 +927,28 @@ def validate(root: Path) -> list[str]:
     if not isinstance(sessions, dict):
         errors.append("catalog.sessions 必须是 mapping")
         sessions = {}
+    artifacts = meta.get("artifacts")
+    if artifacts is not None:
+        if not isinstance(artifacts, dict):
+            errors.append("catalog.artifacts 必须是 mapping")
+        else:
+            for platform, spec in artifacts.items():
+                where = f"catalog.artifacts.{platform}"
+                if platform not in ("android", "ios"):
+                    errors.append(f"{where}: 平台只能是 android/ios")
+                if not isinstance(spec, dict):
+                    errors.append(f"{where}: 必须是 mapping")
+                    continue
+                if "required" in spec and not isinstance(spec["required"], bool):
+                    errors.append(f"{where}.required 必须是 boolean")
+                if platform == "android" and not spec.get("package"):
+                    errors.append(f"{where}: Android 需要 package")
+                if platform == "ios" and not spec.get("bundle_id"):
+                    errors.append(f"{where}: iOS 需要 bundle_id")
+                if platform == "android" and spec.get("bundle_id"):
+                    errors.append(f"{where}: Android 不应填 bundle_id")
+                if platform == "ios" and spec.get("package"):
+                    errors.append(f"{where}: iOS 不应填 package")
     pattern = meta.get("id_pattern") or r"^.+$"
     try:
         id_re = re.compile(pattern)
@@ -1057,6 +1288,29 @@ def build_parser() -> argparse.ArgumentParser:
         default=None,
         help="默认 automated；any 表示不按人工参与度筛选",
     )
+    p_sel.add_argument(
+        "--platform",
+        action="append",
+        choices=["android", "ios"],
+        default=[],
+        help="mobile 运行平台，可重复；选中 mobile 用例时必填",
+    )
+    p_sel.add_argument(
+        "--artifact",
+        action="append",
+        dest="artifacts",
+        default=[],
+        metavar="PLATFORM=PATH_OR_URL",
+        help="本轮移动产物，可重复，例如 android=dist/app.apk",
+    )
+    p_sel.add_argument(
+        "--binding",
+        action="append",
+        dest="bindings",
+        default=[],
+        metavar="PLATFORM=TESTER_BINDING",
+        help="平台 tester binding，可重复；默认 tester.<platform>",
+    )
     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])
@@ -1073,6 +1327,19 @@ def main(argv: list[str] | None = None) -> int:
     parser = build_parser()
     args = parser.parse_args(argv)
     root = Path(args.root).resolve()
+    if args.cmd != "roots":
+        roots = discover_or_self(root)
+        if len(roots) > 1:
+            if args.cmd in FANOUT_COMMANDS:
+                return cmd_fanout(args, roots)
+            print(
+                f"发现多个 tester 根目录，命令 {args.cmd} 不是可 fan-out 的执行命令；"
+                "请对具体项目根目录执行该维护命令。",
+                file=sys.stderr,
+            )
+            return 2
+        if len(roots) == 1:
+            root = roots[0]
     if args.cmd == "list":
         return cmd_list(root)
     if args.cmd == "select":