feat(tester): require caller artifact locators instead of --release
Mobile runs now take --artifact android|ios=<path-or-url> from the caller and create a fresh isolated worktree each time. Drop CI release/manifest lookup.
This commit is contained in:
@@ -1,9 +1,10 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Acquire and provision a CI mobile test bundle.
|
||||
"""Provision caller-provided mobile artifacts for tester.
|
||||
|
||||
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.
|
||||
This is shared tester infrastructure. It never replaces real Android/iOS
|
||||
execution with HTTP calls: it only obtains, verifies and installs the
|
||||
artifact the caller supplied, then the catalog runner drives the mobile
|
||||
session.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -30,7 +31,6 @@ 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"}
|
||||
|
||||
@@ -39,8 +39,8 @@ class EnvironmentFailure(RuntimeError):
|
||||
"""An external environment or recoverable provisioning failure."""
|
||||
|
||||
|
||||
class ManifestFailure(RuntimeError):
|
||||
"""A malformed or mismatched CI bundle contract."""
|
||||
class ArtifactFailure(RuntimeError):
|
||||
"""A missing or unusable caller-provided artifact locator."""
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -102,78 +102,6 @@ def fetch_bytes(url: str, *, attempts: int = 3, token: str = "") -> bytes:
|
||||
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:
|
||||
@@ -182,21 +110,89 @@ def sha256_file(path: Path) -> str:
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def download_artifact(platform: str, item: dict[str, Any], destination: Path, context: RuntimeContext) -> dict[str, Any]:
|
||||
def parse_artifacts(values: list[str]) -> dict[str, str]:
|
||||
pairs: dict[str, str] = {}
|
||||
for raw in values:
|
||||
platform, sep, value = raw.partition("=")
|
||||
platform = platform.strip().lower()
|
||||
value = value.strip()
|
||||
if not sep or platform not in PLATFORMS or not value:
|
||||
raise ArtifactFailure(f"--artifact 格式必须是 android=<路径或 URL> 或 ios=<路径或 URL>: {raw}")
|
||||
if platform in pairs:
|
||||
raise ArtifactFailure(f"--artifact 重复平台: {platform}")
|
||||
pairs[platform] = value
|
||||
return pairs
|
||||
|
||||
|
||||
def local_path(locator: str) -> Path | None:
|
||||
parsed = urllib.parse.urlsplit(locator)
|
||||
if parsed.scheme in ("http", "https"):
|
||||
return None
|
||||
if parsed.scheme == "file":
|
||||
return Path(parsed.path)
|
||||
if parsed.scheme:
|
||||
raise ArtifactFailure(f"不支持的产物地址 scheme: {parsed.scheme}")
|
||||
return Path(locator).expanduser()
|
||||
|
||||
|
||||
def obtain_artifact(platform: str, locator: str, destination: Path, context: RuntimeContext) -> dict[str, Any]:
|
||||
path = local_path(locator)
|
||||
if path is not None:
|
||||
if not path.is_file():
|
||||
raise ArtifactFailure(f"{platform} 产物不是现有文件: {path}")
|
||||
resolved = path.resolve()
|
||||
return {
|
||||
"path": str(resolved),
|
||||
"locator": str(resolved),
|
||||
"sha256": sha256_file(resolved),
|
||||
"attempts": 1,
|
||||
"source": "file",
|
||||
"verified": True,
|
||||
}
|
||||
|
||||
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}
|
||||
destination.write_bytes(fetch_bytes(locator, attempts=1, token=os.environ.get("GITEA_TOKEN", "")))
|
||||
return {
|
||||
"path": str(destination),
|
||||
"locator": redact_url(locator),
|
||||
"sha256": sha256_file(destination),
|
||||
"attempts": attempt,
|
||||
"source": "url",
|
||||
"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}")
|
||||
raise EnvironmentFailure(f"{platform} 下载失败（{context.retries['download']} 次）: {last}")
|
||||
|
||||
|
||||
def catalog_identities(root: Path) -> dict[str, str]:
|
||||
path = root / ".pouch" / "tester" / "catalog.yaml"
|
||||
if not path.is_file():
|
||||
return {}
|
||||
try:
|
||||
import yaml
|
||||
except ImportError:
|
||||
return {}
|
||||
try:
|
||||
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||||
except (OSError, yaml.YAMLError):
|
||||
return {}
|
||||
artifacts = data.get("artifacts")
|
||||
if not isinstance(artifacts, dict):
|
||||
return {}
|
||||
identities: dict[str, str] = {}
|
||||
android = artifacts.get("android")
|
||||
ios = artifacts.get("ios")
|
||||
if isinstance(android, dict) and isinstance(android.get("package"), str) and android["package"]:
|
||||
identities["android"] = android["package"]
|
||||
if isinstance(ios, dict) and isinstance(ios.get("bundle_id"), str) and ios["bundle_id"]:
|
||||
identities["ios"] = ios["bundle_id"]
|
||||
return identities
|
||||
|
||||
|
||||
def android_identity(path: Path, context: RuntimeContext) -> str:
|
||||
@@ -336,6 +332,18 @@ def sanitize(value: str) -> str:
|
||||
return re.sub(r"[^A-Za-z0-9._-]+", "-", value)
|
||||
|
||||
|
||||
def resolve_commit(explicit: str, context: RuntimeContext) -> str:
|
||||
if explicit:
|
||||
if not COMMIT.fullmatch(explicit):
|
||||
raise ArtifactFailure("--commit 必须是 40 位 commit SHA")
|
||||
return explicit
|
||||
git = shutil.which("git") or "git"
|
||||
result = run_command([git, "rev-parse", "HEAD"], context=context, cwd=context.root)
|
||||
if result.code != 0 or not COMMIT.fullmatch(result.output):
|
||||
raise EnvironmentFailure(f"无法读取当前 commit: {context.root}")
|
||||
return result.output
|
||||
|
||||
|
||||
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)
|
||||
@@ -349,22 +357,13 @@ def ensure_worktree(commit: str, context: RuntimeContext, parent: Path | None =
|
||||
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:]}")
|
||||
token = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + os.urandom(3).hex()
|
||||
path = (base / f"{sanitize(commit[:12])}-{token}").resolve()
|
||||
if path.exists():
|
||||
raise EnvironmentFailure(f"隔离 worktree 路径已存在，拒绝覆盖: {path}")
|
||||
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}")
|
||||
@@ -376,7 +375,7 @@ def platforms(raw: str) -> list[str]:
|
||||
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 且不能重复")
|
||||
raise ArtifactFailure("--platform 只能是 android,ios 且不能重复")
|
||||
return values
|
||||
|
||||
|
||||
@@ -395,7 +394,6 @@ def execute(args: argparse.Namespace) -> int:
|
||||
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,
|
||||
@@ -405,31 +403,37 @@ def execute(args: argparse.Namespace) -> int:
|
||||
"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']}-"))
|
||||
locators = parse_artifacts(args.artifacts)
|
||||
missing = [platform for platform in requested if platform not in locators]
|
||||
extra = [platform for platform in locators if platform not in requested]
|
||||
if missing:
|
||||
raise ArtifactFailure("缺少调用方产物位置: " + ", ".join(f"--artifact {item}=<路径或 URL>" for item in missing))
|
||||
if extra:
|
||||
raise ArtifactFailure("产物平台超出 --platform: " + ", ".join(extra))
|
||||
record["artifacts_requested"] = {platform: redact_url(locator) if local_path(locator) is None else locator for platform, locator in locators.items()}
|
||||
commit = resolve_commit(args.commit, context)
|
||||
worktree, actual_commit = ensure_worktree(commit, context, Path(args.worktree_root).resolve() if args.worktree_root else None)
|
||||
record["worktree"] = {"path": str(worktree), "expected_commit": commit, "actual_commit": actual_commit, "isolated": True}
|
||||
artifact_dir = Path(tempfile.mkdtemp(prefix="pouch-tester-artifacts-"))
|
||||
record["artifact_dir"] = str(artifact_dir)
|
||||
prepared: dict[str, tuple[dict[str, Any], dict[str, str]]] = {}
|
||||
expected = catalog_identities(context.root)
|
||||
prepared: dict[str, tuple[dict[str, Any], dict[str, 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}")
|
||||
result["artifact"] = obtain_artifact(platform, locators[platform], path, context)
|
||||
identity = android_identity(Path(result["artifact"]["path"]), context) if platform == "android" else ios_identity(Path(result["artifact"]["path"]))
|
||||
wanted = expected.get(platform)
|
||||
if wanted and identity != wanted:
|
||||
raise EnvironmentFailure(f"{platform} 身份不匹配: expected={wanted} 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:
|
||||
prepared[platform] = (result, node, identity)
|
||||
except (EnvironmentFailure, ArtifactFailure) 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:
|
||||
@@ -438,10 +442,9 @@ def execute(args: argparse.Namespace) -> int:
|
||||
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():
|
||||
for platform, (result, node, identity) 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)
|
||||
installed = install_android(node, Path(result["artifact"]["path"]), identity, context) if platform == "android" else install_ios(node, Path(result["artifact"]["path"]), identity, context)
|
||||
result.update({"status": "ready", "install": installed, "failure_class": None})
|
||||
except EnvironmentFailure as exc:
|
||||
failures[platform] = str(exc)
|
||||
@@ -450,7 +453,7 @@ def execute(args: argparse.Namespace) -> int:
|
||||
if not failures:
|
||||
record["product_test"].update({"status": "pending", "failure_class": None})
|
||||
record["evidence"] = context.evidence
|
||||
except (EnvironmentFailure, ManifestFailure) as exc:
|
||||
except (EnvironmentFailure, ArtifactFailure) 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")}
|
||||
@@ -462,34 +465,18 @@ def execute(args: argparse.Namespace) -> int:
|
||||
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")
|
||||
root = argparse.ArgumentParser(prog="tester", description="Obtain and provision caller-provided mobile artifacts")
|
||||
sub = root.add_subparsers(dest="command", required=True)
|
||||
run = sub.add_parser("run")
|
||||
run.add_argument("--release", required=True)
|
||||
run.add_argument("--artifact", action="append", dest="artifacts", default=[], metavar="PLATFORM=PATH_OR_URL", help="调用方产物，可重复，例如 android=dist/app.apk 或 ios=https://...")
|
||||
run.add_argument("--platform", default="android,ios")
|
||||
run.add_argument("--commit", default="", help="本轮隔离 worktree 的 40 位 commit；缺省使用当前 HEAD")
|
||||
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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user