feat(tester): provision CI mobile bundles in shared skill

2026-10-14 17:29:12 +08:00
parent db92baf3a7
commit 706c553778
8 changed files with 693 additions and 9 deletions
@@ -0,0 +1,494 @@
+#!/usr/bin/env python3
+"""Acquire and provision a CI mobile test bundle.
+
+This is shared tester infrastructure.  It never replaces real Android/iOS
+execution with HTTP calls: it only downloads, verifies and installs the
+release artifact before the catalog runner drives the mobile session.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import plistlib
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+import zipfile
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from platform import system as platform_system
+from typing import Any
+
+
+SHA256 = re.compile(r"^[0-9a-fA-F]{64}$")
+COMMIT = re.compile(r"^[0-9a-fA-F]{40}$")
+PLATFORMS = {"android", "ios"}
+
+
+class EnvironmentFailure(RuntimeError):
+    """An external environment or recoverable provisioning failure."""
+
+
+class ManifestFailure(RuntimeError):
+    """A malformed or mismatched CI bundle contract."""
+
+
+@dataclass
+class CommandResult:
+    code: int
+    output: str
+
+
+@dataclass
+class RuntimeContext:
+    root: Path
+    retries: dict[str, int] = field(default_factory=lambda: {"download": 3, "start": 3, "install": 3})
+    evidence: list[dict[str, Any]] = field(default_factory=list)
+
+    def record(self, command: str, result: CommandResult) -> None:
+        self.evidence.append({
+            "command": command,
+            "status": "ok" if result.code == 0 else "error",
+            "output": result.output[-800:],
+        })
+
+
+def redact_url(value: str) -> str:
+    parsed = urllib.parse.urlsplit(value)
+    if not parsed.query:
+        return value
+    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "<redacted>", ""))
+
+
+def run_command(args: list[str], *, context: RuntimeContext | None = None, cwd: Path | None = None) -> CommandResult:
+    try:
+        proc = subprocess.run(args, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False)
+        result = CommandResult(proc.returncode, proc.stdout.strip())
+    except OSError as exc:
+        result = CommandResult(127, str(exc))
+    if context:
+        context.record(" ".join(args), result)
+    return result
+
+
+def fetch_bytes(url: str, *, attempts: int = 3, token: str = "") -> bytes:
+    if url.startswith("file://"):
+        return Path(urllib.parse.urlsplit(url).path).read_bytes()
+    if not urllib.parse.urlsplit(url).scheme:
+        return Path(url).read_bytes()
+    last: Exception | None = None
+    for attempt in range(1, attempts + 1):
+        try:
+            headers = {"User-Agent": "pouch-tester/1"}
+            if token:
+                headers["Authorization"] = f"token {token}"
+            request = urllib.request.Request(url, headers=headers)
+            with urllib.request.urlopen(request, timeout=30) as response:
+                return response.read()
+        except (OSError, urllib.error.URLError) as exc:
+            last = exc
+            if attempt < attempts:
+                time.sleep(min(attempt, 2))
+    raise EnvironmentFailure(f"下载失败（{attempts} 次）: {redact_url(url)}: {last}")
+
+
+def fetch_json(url: str, *, attempts: int = 3, token: str = "") -> dict[str, Any]:
+    try:
+        value = json.loads(fetch_bytes(url, attempts=attempts, token=token).decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ManifestFailure(f"manifest 不是有效 JSON: {redact_url(url)}: {exc}") from exc
+    if not isinstance(value, dict):
+        raise ManifestFailure("manifest 顶层必须是 mapping")
+    return value
+
+
+def validate_manifest(data: dict[str, Any], requested_release: str | None = None) -> None:
+    if not isinstance(data, dict) or data.get("schema_version") != 1:
+        raise ManifestFailure("manifest.schema_version 必须为 1")
+    release = data.get("release")
+    commit = data.get("commit")
+    if not isinstance(release, str) or not release:
+        raise ManifestFailure("manifest.release 不能为空")
+    if requested_release and release != requested_release:
+        raise ManifestFailure(f"manifest.release={release} 与请求的 release={requested_release} 不匹配")
+    if not isinstance(commit, str) or not COMMIT.fullmatch(commit):
+        raise ManifestFailure("manifest.commit 必须是 40 位 commit SHA")
+    artifacts = data.get("artifacts")
+    if not isinstance(artifacts, dict):
+        raise ManifestFailure("manifest.artifacts 必须是 mapping")
+    identities = {"android": ("package", "page.writes"), "ios": ("bundle_id", "page.writes")}
+    for platform, (key, expected) in identities.items():
+        item = artifacts.get(platform)
+        if not isinstance(item, dict):
+            raise ManifestFailure(f"缺少 artifacts.{platform}")
+        url = item.get("url")
+        if not isinstance(url, str) or not url:
+            raise ManifestFailure(f"artifacts.{platform}.url 不能为空")
+        if not urllib.parse.urlsplit(url).scheme and not Path(url).is_file():
+            raise ManifestFailure(f"artifacts.{platform}.url 不是 URL 或现有文件: {url}")
+        digest = item.get("sha256")
+        if not isinstance(digest, str) or not SHA256.fullmatch(digest):
+            raise ManifestFailure(f"artifacts.{platform}.sha256 必须是 64 位十六进制")
+        if item.get(key) != expected:
+            raise ManifestFailure(f"artifacts.{platform}.{key} 必须为 {expected}")
+
+
+def resolve_manifest(release: str, *, attempts: int = 3) -> tuple[dict[str, Any], str]:
+    token = os.environ.get("GITEA_TOKEN", "")
+    source = os.environ.get("TEST_BUNDLE_MANIFEST_URL", "").strip()
+    if not source:
+        candidate = Path(release)
+        if candidate.is_file() or release.startswith(("http://", "https://", "file://")):
+            source = release
+        else:
+            api = os.environ.get("GITEA_API", "").rstrip("/")
+            if not api:
+                raise EnvironmentFailure("缺少 TEST_BUNDLE_MANIFEST_URL 或 GITEA_API；不要手工复制/安装产物")
+            release_data = fetch_json(f"{api}/releases/tags/{urllib.parse.quote(release, safe='')}", attempts=attempts, token=token)
+            assets = [item for item in release_data.get("assets", []) if item.get("name") == "test-bundle.manifest.json"]
+            if len(assets) != 1:
+                raise EnvironmentFailure(f"release {release} 没有唯一 test-bundle.manifest.json；请等待 CI 双平台发布完成")
+            asset = assets[0]
+            source = asset.get("browser_download_url") or asset.get("download_url") or ""
+            if not source:
+                source = f"{api}/releases/{release_data['id']}/assets/{asset['id']}"
+    if urllib.parse.urlsplit(source).scheme:
+        data = fetch_json(source, attempts=attempts, token=token)
+    else:
+        try:
+            data = json.loads(Path(source).read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ManifestFailure(f"读取 manifest 失败: {source}: {exc}") from exc
+    requested = None if Path(release).is_file() or urllib.parse.urlsplit(release).scheme else release
+    validate_manifest(data, requested_release=requested)
+    return data, redact_url(source)
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def download_artifact(platform: str, item: dict[str, Any], destination: Path, context: RuntimeContext) -> dict[str, Any]:
+    last: Exception | None = None
+    for attempt in range(1, context.retries["download"] + 1):
+        try:
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            destination.write_bytes(fetch_bytes(item["url"], attempts=1, token=os.environ.get("GITEA_TOKEN", "")))
+            actual = sha256_file(destination)
+            if actual.lower() != item["sha256"].lower():
+                raise EnvironmentFailure(f"{platform} sha256 不匹配: expected={item['sha256']} actual={actual}")
+            return {"path": str(destination), "url": redact_url(item["url"]), "sha256": actual, "attempts": attempt, "verified": True}
+        except (OSError, EnvironmentFailure) as exc:
+            last = exc
+            if attempt < context.retries["download"]:
+                time.sleep(min(attempt, 2))
+    raise EnvironmentFailure(f"{platform} 下载/校验失败（{context.retries['download']} 次）: {last}")
+
+
+def android_identity(path: Path, context: RuntimeContext) -> str:
+    commands: list[list[str]] = []
+    if shutil.which("apkanalyzer"):
+        commands.append(["apkanalyzer", "manifest", "application-id", str(path)])
+    if shutil.which("aapt2"):
+        commands.append(["aapt2", "dump", "badging", str(path)])
+    if shutil.which("aapt"):
+        commands.append(["aapt", "dump", "badging", str(path)])
+    if not commands:
+        raise EnvironmentFailure("缺少 apkanalyzer/aapt2/aapt；请在 Android tester 节点配置 Android SDK build-tools")
+    for command in commands:
+        result = run_command(command, context=context)
+        if result.code == 0:
+            match = re.search(r"(?:name|application-id)[=:]['\"]?([A-Za-z0-9_.]+)", result.output)
+            if match:
+                return match.group(1)
+    raise EnvironmentFailure("APK 身份读取失败；请检查 Android SDK build-tools")
+
+
+def ios_identity(path: Path) -> str:
+    try:
+        with zipfile.ZipFile(path) as archive:
+            candidates = [name for name in archive.namelist() if re.fullmatch(r"Payload/[^/]+\.app/Info\.plist", name)]
+            if len(candidates) != 1:
+                raise EnvironmentFailure("IPA 中没有唯一 Payload/*.app/Info.plist")
+            plist = plistlib.loads(archive.read(candidates[0]))
+    except (OSError, zipfile.BadZipFile, plistlib.InvalidFileException) as exc:
+        raise EnvironmentFailure(f"IPA 解包/读取 Info.plist 失败: {exc}") from exc
+    bundle_id = plist.get("CFBundleIdentifier")
+    if not isinstance(bundle_id, str) or not bundle_id:
+        raise EnvironmentFailure("IPA Info.plist 缺少 CFBundleIdentifier")
+    return bundle_id
+
+
+def retry_command(args: list[str], *, context: RuntimeContext, action: str) -> CommandResult:
+    last = CommandResult(1, "")
+    for attempt in range(1, context.retries[action] + 1):
+        last = run_command(args, context=context)
+        if last.code == 0:
+            return last
+        if attempt < context.retries[action]:
+            time.sleep(min(attempt, 2))
+    raise EnvironmentFailure(f"{action} 失败（{context.retries[action]} 次）: {' '.join(args)}\n{last.output[-500:]}")
+
+
+def android_preflight(context: RuntimeContext) -> dict[str, str]:
+    adb = os.environ.get("ADB_BIN") or shutil.which("adb")
+    if not adb:
+        raise EnvironmentFailure("Android 节点缺少 adb；请安装/配置 Android SDK platform-tools")
+    retry_command([adb, "start-server"], context=context, action="start")
+    devices = run_command([adb, "devices", "-l"], context=context)
+    available = [line.split()[0] for line in devices.output.splitlines()[1:] if re.search(r"\sdevice\s", line)]
+    serial = os.environ.get("ANDROID_SERIAL", "").strip()
+    if serial and serial not in available:
+        raise EnvironmentFailure(f"ANDROID_SERIAL={serial} 不在 device 状态；请解锁/授权设备后检查 adb devices -l")
+    if not serial and len(available) != 1:
+        if not available:
+            raise EnvironmentFailure("没有 Android device；请连接、解锁、开启 USB 调试并接受 RSA 授权")
+        raise EnvironmentFailure(f"检测到多个 Android device ({', '.join(available)})；请设置 ANDROID_SERIAL")
+    return {"adb": adb, "serial": serial or available[0], "session_lifecycle": "reused-or-started"}
+
+
+def ios_preflight(context: RuntimeContext) -> dict[str, str]:
+    if platform_system() != "Darwin":
+        raise EnvironmentFailure("iOS 必须由 tester.ios 的 macOS 节点执行；Linux 不能伪装 Simulator")
+    xcrun = shutil.which("xcrun")
+    if not xcrun:
+        raise EnvironmentFailure("iOS 节点缺少 xcrun；请安装 Xcode Command Line Tools")
+    devices = run_command([xcrun, "simctl", "list", "devices", "available"], context=context)
+    if devices.code != 0:
+        raise EnvironmentFailure("simctl 不可用；请安装 Xcode 并确认已接受许可")
+    wanted = os.environ.get("IOS_SIMULATOR_UDID", "").strip()
+    matches: list[tuple[str, str, str]] = []
+    for line in devices.output.splitlines():
+        match = re.search(r"^\s*(.+?)\s+\(([0-9A-Fa-f-]{36})\)\s+\((Booted|Shutdown)\)", line)
+        if match:
+            matches.append((match.group(2), match.group(1), match.group(3)))
+    selected = next((item for item in matches if item[0] == wanted), None) if wanted else None
+    selected = selected or next((item for item in matches if item[2] == "Booted"), None) or (matches[0] if matches else None)
+    if not selected:
+        raise EnvironmentFailure("没有可用 iOS Simulator；请在 Mac 上安装一个可用 runtime/device")
+    udid, name, state = selected
+    if state != "Booted":
+        retry_command([xcrun, "simctl", "boot", udid], context=context, action="start")
+    retry_command([xcrun, "simctl", "bootstatus", udid, "-b"], context=context, action="start")
+    return {"xcrun": xcrun, "udid": udid, "name": name, "session_lifecycle": "reused" if state == "Booted" else "started"}
+
+
+def check_network(context: RuntimeContext, node: dict[str, str] | None = None) -> None:
+    host = os.environ.get("HOST", "").strip()
+    api = os.environ.get("API_BASE", "").strip() or (f"http://{host}" if host else "")
+    if not api:
+        raise EnvironmentFailure("缺少 HOST/API_BASE；请在 tester.local.yaml 配置测试服务地址")
+    for attempt in range(1, context.retries["start"] + 1):
+        try:
+            with urllib.request.urlopen(urllib.request.Request(api.rstrip("/") + "/v1/me"), timeout=5) as response:
+                if response.status < 500:
+                    break
+        except urllib.error.HTTPError as exc:
+            if exc.code < 500:
+                break
+        except (OSError, urllib.error.URLError):
+            pass
+        if attempt == context.retries["start"]:
+            raise EnvironmentFailure(f"测试服务不可达: {api}；请检查 HOST、VPN、DNS 和局域网后重跑")
+        time.sleep(min(attempt, 2))
+    if node and "adb" in node:
+        if shutil.which("ping"):
+            result = run_command([node["adb"], "-s", node["serial"], "shell", "ping", "-c", "1", "-W", "3", host], context=context)
+            if result.code != 0:
+                raise EnvironmentFailure(f"Android device 无法访问 HOST={host}；请检查设备 Wi-Fi/DNS/VPN")
+    elif node:
+        result = run_command([node["xcrun"], "simctl", "spawn", node["udid"], "/usr/bin/curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", api.rstrip("/") + "/v1/me"], context=context)
+        if result.code != 0 or result.output.strip() in {"", "000"} or result.output.strip().startswith("5"):
+            raise EnvironmentFailure(f"iOS Simulator 无法访问 {api}；请检查 Mac 的 VPN/DNS/局域网权限")
+
+
+def install_android(node: dict[str, str], path: Path, package: str, context: RuntimeContext) -> dict[str, Any]:
+    retry_command([node["adb"], "-s", node["serial"], "install", "-r", str(path)], context=context, action="install")
+    result = run_command([node["adb"], "-s", node["serial"], "shell", "pm", "path", package], context=context)
+    if result.code != 0 or "package:" not in result.output:
+        raise EnvironmentFailure(f"安装后未找到 Android package {package}；请检查签名/设备空间")
+    return {"serial": node["serial"], "package": package, "installed": True, "verified": True}
+
+
+def install_ios(node: dict[str, str], path: Path, bundle_id: str, context: RuntimeContext) -> dict[str, Any]:
+    retry_command([node["xcrun"], "simctl", "install", node["udid"], str(path)], context=context, action="install")
+    result = run_command([node["xcrun"], "simctl", "get_app_container", node["udid"], bundle_id], context=context)
+    if result.code != 0 or not result.output:
+        raise EnvironmentFailure(f"安装后未找到 iOS bundle {bundle_id}；请检查签名/Simulator 架构")
+    return {"udid": node["udid"], "simulator": node["name"], "bundle_id": bundle_id, "installed": True, "verified": True}
+
+
+def sanitize(value: str) -> str:
+    return re.sub(r"[^A-Za-z0-9._-]+", "-", value)
+
+
+def ensure_worktree(commit: str, context: RuntimeContext, parent: Path | None = None) -> tuple[Path, str]:
+    git = shutil.which("git") or "git"
+    check = run_command([git, "rev-parse", "--show-toplevel"], context=context, cwd=context.root)
+    if check.code != 0:
+        raise EnvironmentFailure(f"目标目录不是 Git worktree: {context.root}")
+    repo = Path(check.output).resolve()
+    verify = run_command([git, "rev-parse", f"{commit}^{{commit}}"], context=context, cwd=repo)
+    if verify.code != 0:
+        fetched = run_command([git, "fetch", "--no-tags", "origin", commit], context=context, cwd=repo)
+        if fetched.code != 0:
+            raise EnvironmentFailure(f"目标 commit 不在本地且无法从 origin 获取: {commit}")
+    base = parent or Path(os.environ.get("TESTER_WORKTREE_ROOT", Path(tempfile.gettempdir()) / "pouch-tester" / "worktrees"))
+    base.mkdir(parents=True, exist_ok=True)
+    path = (base / sanitize(commit[:12])).resolve()
+    listing = run_command([git, "worktree", "list", "--porcelain"], context=context, cwd=repo)
+    managed = any(re.search(rf"^worktree {re.escape(str(path))}$", block, re.M) for block in listing.output.split("\n\n"))
+    if path.exists() and not managed:
+        raise EnvironmentFailure(f"隔离 worktree 路径已存在但不是 Git worktree，拒绝覆盖: {path}")
+    if managed:
+        current = run_command([git, "-C", str(path), "rev-parse", "HEAD"], context=context)
+        if current.code != 0 or current.output.lower() != commit.lower():
+            removed = run_command([git, "worktree", "remove", "--force", str(path)], context=context, cwd=repo)
+            if removed.code != 0:
+                raise EnvironmentFailure(f"worktree commit 不匹配且自动修复失败: {path}")
+            managed = False
+    if not managed:
+        added = run_command([git, "worktree", "add", "--detach", str(path), commit], context=context, cwd=repo)
+        if added.code != 0:
+            raise EnvironmentFailure(f"创建精确 commit worktree 失败: {path}\n{added.output[-500:]}")
+    actual = run_command([git, "-C", str(path), "rev-parse", "HEAD"], context=context)
+    if actual.code != 0 or actual.output.lower() != commit.lower():
+        raise EnvironmentFailure(f"worktree commit 不匹配: expected={commit} actual={actual.output}")
+    return path, actual.output
+
+
+def platforms(raw: str) -> list[str]:
+    values = [item.strip().lower() for item in raw.split(",") if item.strip()]
+    if not values:
+        return ["android", "ios"]
+    if any(item not in PLATFORMS for item in values) or len(values) != len(set(values)):
+        raise ManifestFailure("--platform 只能是 android,ios 且不能重复")
+    return values
+
+
+def write_run(root: Path, record: dict[str, Any]) -> Path:
+    runs = root / ".pouch" / "tester" / "runs"
+    runs.mkdir(parents=True, exist_ok=True)
+    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-bundle-" + os.urandom(3).hex()
+    path = runs / f"{run_id}.yaml"
+    path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return path
+
+
+def execute(args: argparse.Namespace) -> int:
+    requested = platforms(args.platform)
+    context = RuntimeContext(Path(args.root).resolve())
+    record: dict[str, Any] = {
+        "schema_version": 1,
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "release_request": args.release,
+        "platforms_requested": requested,
+        "partial": bool(args.partial),
+        "binding": args.binding,
+        "node": os.environ.get("TESTER_NODE", "local"),
+        "terminal": {"handle": os.environ.get("ORCA_TERMINAL_HANDLE"), "created": os.environ.get("TESTER_TERMINAL_CREATED", "false").lower() == "true", "closed": os.environ.get("TESTER_TERMINAL_CLOSED", "unknown")},
+        "product_test": {"status": "not-run", "failure_class": "automatic-repair-failed", "note": "准备完成后由 tester catalog 驱动真实移动 session；本工具不以 API 替代移动测试"},
+        "platforms": {},
+    }
+    try:
+        manifest, source = resolve_manifest(args.release, attempts=context.retries["download"])
+        record["artifact_manifest"] = manifest
+        record["manifest_source"] = source
+        worktree, actual_commit = ensure_worktree(manifest["commit"], context, Path(args.worktree_root).resolve() if args.worktree_root else None)
+        record["worktree"] = {"path": str(worktree), "expected_commit": manifest["commit"], "actual_commit": actual_commit, "isolated": True}
+        artifact_dir = Path(tempfile.mkdtemp(prefix=f"pouch-tester-{manifest['release']}-"))
+        record["artifact_dir"] = str(artifact_dir)
+        prepared: dict[str, tuple[dict[str, Any], dict[str, str]]] = {}
+        failures: dict[str, str] = {}
+        for platform in requested:
+            item = manifest["artifacts"][platform]
+            result: dict[str, Any] = {"node": "tester.android" if platform == "android" else "tester.ios", "status": "preparing"}
+            record["platforms"][platform] = result
+            try:
+                path = artifact_dir / ("app.apk" if platform == "android" else "app.ipa")
+                result["artifact"] = download_artifact(platform, item, path, context)
+                identity = android_identity(path, context) if platform == "android" else ios_identity(path)
+                expected = item["package"] if platform == "android" else item["bundle_id"]
+                if identity != expected:
+                    raise EnvironmentFailure(f"{platform} 身份不匹配: expected={expected} actual={identity}")
+                node = android_preflight(context) if platform == "android" else ios_preflight(context)
+                check_network(context, node)
+                result.update({"identity": identity, "node_preflight": node, "status": "preflight-passed"})
+                prepared[platform] = (result, node)
+            except EnvironmentFailure as exc:
+                failures[platform] = str(exc)
+                result.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc)})
+        if failures and not args.partial:
+            for platform in requested:
+                if platform not in failures:
+                    record["platforms"][platform].update({"status": "not-run", "reason": "全量 preflight 闸门被另一平台阻断"})
+            record.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "next_action": "修复列出的平台环境后重跑；默认不会静默降级"})
+        else:
+            for platform, (result, node) in prepared.items():
+                try:
+                    item = manifest["artifacts"][platform]
+                    installed = install_android(node, Path(result["artifact"]["path"]), item["package"], context) if platform == "android" else install_ios(node, Path(result["artifact"]["path"]), item["bundle_id"], context)
+                    result.update({"status": "ready", "install": installed, "failure_class": None})
+                except EnvironmentFailure as exc:
+                    failures[platform] = str(exc)
+                    result.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc)})
+            record.update({"status": "environment-fail" if failures else "ready", "failure_class": "automatic-repair-failed" if failures else None})
+            if not failures:
+                record["product_test"].update({"status": "pending", "failure_class": None})
+        record["evidence"] = context.evidence
+    except (EnvironmentFailure, ManifestFailure) as exc:
+        record.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc), "evidence": context.evidence})
+    path = write_run(context.root, record)
+    output = {"status": record["status"], "run": str(path), "platforms": record["platforms"], "worktree": record.get("worktree")}
+    print(json.dumps(output, ensure_ascii=False, indent=2), file=sys.stderr if record["status"] != "ready" else sys.stdout)
+    return 0 if record["status"] == "ready" else 1
+
+
+def manifest_validate(args: argparse.Namespace) -> int:
+    try:
+        data = json.loads(Path(args.path).read_text(encoding="utf-8"))
+        validate_manifest(data, requested_release=args.release)
+    except (OSError, json.JSONDecodeError, ManifestFailure) as exc:
+        print(f"manifest invalid: {exc}", file=sys.stderr)
+        return 1
+    print(json.dumps(data, ensure_ascii=False, indent=2))
+    return 0
+
+
+def parser() -> argparse.ArgumentParser:
+    root = argparse.ArgumentParser(prog="tester", description="Acquire and provision a CI mobile test bundle")
+    sub = root.add_subparsers(dest="command", required=True)
+    run = sub.add_parser("run")
+    run.add_argument("--release", required=True)
+    run.add_argument("--platform", default="android,ios")
+    run.add_argument("--partial", action="store_true")
+    run.add_argument("--root", default=".")
+    run.add_argument("--worktree-root", default="")
+    run.add_argument("--binding", default="")
+    run.set_defaults(func=execute)
+    manifest = sub.add_parser("manifest")
+    manifest_sub = manifest.add_subparsers(dest="manifest_command", required=True)
+    validate = manifest_sub.add_parser("validate")
+    validate.add_argument("path")
+    validate.add_argument("--release", default=None)
+    validate.set_defaults(func=manifest_validate)
+    return root
+
+
+if __name__ == "__main__":
+    parsed = parser().parse_args()
+    raise SystemExit(parsed.func(parsed))