From c2eb207bd2b3499c88d2beac9f606ba07843d5ed Mon Sep 17 00:00:00 2001 From: laily Date: Thu, 15 Oct 2026 18:04:59 +0800 Subject: [PATCH] feat(test-preflight): add verify and prepare skill --- skills/test-preflight/README.md | 13 + skills/test-preflight/SKILL.md | 51 ++ skills/test-preflight/agents/openai.yaml | 3 + .../references/api_reference.md | 4 + .../test-preflight/references/boundaries.md | 44 ++ skills/test-preflight/references/schema.md | 85 +++ skills/test-preflight/scripts/preflight.py | 549 ++++++++++++++++++ .../tests/fixtures/contract_runner.py | 56 ++ skills/test-preflight/tests/test_preflight.py | 209 +++++++ 9 files changed, 1014 insertions(+) create mode 100644 skills/test-preflight/README.md create mode 100644 skills/test-preflight/SKILL.md create mode 100644 skills/test-preflight/agents/openai.yaml create mode 100644 skills/test-preflight/references/api_reference.md create mode 100644 skills/test-preflight/references/boundaries.md create mode 100644 skills/test-preflight/references/schema.md create mode 100644 skills/test-preflight/scripts/preflight.py create mode 100644 skills/test-preflight/tests/fixtures/contract_runner.py create mode 100644 skills/test-preflight/tests/test_preflight.py diff --git a/skills/test-preflight/README.md b/skills/test-preflight/README.md new file mode 100644 index 0000000..e339cf7 --- /dev/null +++ b/skills/test-preflight/README.md @@ -0,0 +1,13 @@ +# test-preflight + +Reusable preflight for tester matrices. It supports read-only `verify` and +test-environment-only `prepare`, and writes per-check YAML/CSV evidence for +artifacts, tester nodes, devices, network, services, and fixture/cleanup +contracts. + +Run the self-tests with: + +```bash +python3 -m unittest discover -s skills/test-preflight/tests +python3 /home/ace/.codex/skills/.system/skill-creator/scripts/quick_validate.py skills/test-preflight +``` diff --git a/skills/test-preflight/SKILL.md b/skills/test-preflight/SKILL.md new file mode 100644 index 0000000..03fa5ef --- /dev/null +++ b/skills/test-preflight/SKILL.md @@ -0,0 +1,51 @@ +--- +name: test-preflight +description: >- + Verify or prepare a mobile/integration test environment from a declared matrix: + artifacts, tester bindings, HOST/PUBLIC_HOST, service probes, and fixture/cleanup + contracts. Use before tester runs; it is not a deployment, build, or business-data skill. +--- + +# test-preflight + +Use this skill before `integration` or `journey` tests. It answers whether the +declared environment can start a run and emits one YAML and one CSV evidence +file. It never logs in, creates business data, builds/downloads an artifact, or +deploys production. + +## Modes + +- `verify` is read-only: validate the matrix, artifact identity, binding, actual + tester-node platform/device state, DNS/TCP/HTTP probes, backend probes, and + fixture/cleanup declarations. It must not install, launch, or deploy. +- `prepare` is side-effecting only within the test environment: call the + caller-supplied deployer command contract, run the same checks, then use the + tester-node command contract to install and launch the already supplied APK or + IPA. It never calls a builder and refuses `environment: prod`. + +The command is: + +```bash +python3 /scripts/preflight.py verify \ + --config --yaml --csv +python3 /scripts/preflight.py prepare \ + --config --yaml --csv +``` + +Read [references/schema.md](references/schema.md) for the input contract and +[references/boundaries.md](references/boundaries.md) before wiring a real +agent-runtime/deployer adapter. The included command-contract fixture is the +safe way to test remote-node behavior without pretending that this machine is a +Mac or that a deployment happened. + +Overall status is `ready`, `blocked`, or `environment-fail`: + +- `blocked` means a required declaration, artifact, binding, runner, fixture, + cleanup contract, or authorization is absent/unsafe. +- `environment-fail` means a declared environment exists but DNS/TCP/HTTP, + device/toolchain, installation, launch, or deployment execution failed. +- `ready` means every recorded check passed. A ready preflight permits the + caller to continue with `integration`; this skill does not run test cases. + +Do not treat a preflight report as a product result. A preflight failure means +no client session may be created; the tester owns `not-run` propagation. diff --git a/skills/test-preflight/agents/openai.yaml b/skills/test-preflight/agents/openai.yaml new file mode 100644 index 0000000..5ed725e --- /dev/null +++ b/skills/test-preflight/agents/openai.yaml @@ -0,0 +1,3 @@ +interface: + display_name: "Test Preflight" + short_description: "Verify or prepare tester environments" diff --git a/skills/test-preflight/references/api_reference.md b/skills/test-preflight/references/api_reference.md new file mode 100644 index 0000000..8ea6b77 --- /dev/null +++ b/skills/test-preflight/references/api_reference.md @@ -0,0 +1,4 @@ +# Reference moved + +Read [schema.md](schema.md) for the matrix and evidence contract, and +[boundaries.md](boundaries.md) for agent-runtime/deployer/builder integration. diff --git a/skills/test-preflight/references/boundaries.md b/skills/test-preflight/references/boundaries.md new file mode 100644 index 0000000..202ee04 --- /dev/null +++ b/skills/test-preflight/references/boundaries.md @@ -0,0 +1,44 @@ +# Integration boundaries + +`test-preflight` is an orchestration and evidence layer. It consumes existing +skills instead of implementing their policy or storage formats. + +| Boundary | Caller-supplied contract | What preflight does | What it must not do | +| --- | --- | --- | --- | +| agent-runtime | resolved `tester.` binding: binding id, node, environment, project/path, and a node runner | validate that the binding is explicit and route node probes through the runner | parse Orca state, invent a host, or create/reuse a terminal | +| deployer | `prepare` only: an argv command returning a JSON deployment receipt for `environment: test` | call it once and record its receipt/status | copy Compose/Argo logic, SSH directly, or accept production | +| builder/obtain | caller-provided local artifact path (and optional remote path), hash, source commit, package/bundle identity | hash and inspect the exact file; pass it to install contract | build, guess an old artifact, or silently download/upload | + +## Deployer command contract + +`prepare` requires: + +```yaml +deployment: + environment: test + command: [python3, /path/to/existing-deployer-adapter.py] +``` + +The command receives JSON on stdin: + +```json +{"operation":"prepare","environment":"test","project_root":"/abs/project","host":"...","public_host":"..."} +``` + +It must return `{ "status": "ready", "observed": "..." }` (an optional +`checks` mapping is recorded). Non-zero/invalid output is an +`environment-fail`; absent command or a non-test environment is `blocked`. +The adapter owns deployer-specific configuration, receipts, cleanup, and +authorization. + +## Device contract + +The node runner owns remote execution. For a local node the implementation uses +read-only `adb`/`xcrun simctl` checks in `verify`; `prepare` alone invokes +`adb install`/launch or `simctl install`/launch. For a remote node the same +operations are JSON requests to the runner. Missing xcrun on Linux, Android +`unauthorized`/`offline`, a missing device, or a failed install/start is an +`environment-fail`, not a product failure. + +No client session is created by this skill. The runner's `install` and `start` +operations are the only device mutations, and are never called by `verify`. diff --git a/skills/test-preflight/references/schema.md b/skills/test-preflight/references/schema.md new file mode 100644 index 0000000..0af5aac --- /dev/null +++ b/skills/test-preflight/references/schema.md @@ -0,0 +1,85 @@ +# test-preflight matrix schema + +The input is YAML. `mode` is supplied by the CLI, so a file may be reused for +`verify` and `prepare`. + +```yaml +schema_version: 1 +project_root: /absolute/project/root +host: https://test.example.invalid +public_host: https://public-test.example.invalid # optional +service_probes: + - id: api-health + url: "{HOST}/health" + expected_status: [200] +fixtures: + required: true + references: [isolated-test-account] +cleanup: + required: true + steps: [delete-run-data] +matrix: + - platform: android + tester: + binding: tester.android + node: local + runner: {kind: local} + artifact: + path: /absolute/dist/writes.apk + sha256: 012345... + package: page.writes + source_commit: 40-character-commit-sha + identity_command: [apkanalyzer, manifest, application-id, "{artifact}"] +``` + +Required matrix fields are `platform` (`android` or `ios`), `tester.binding`, +`tester.node`, and `artifact.path`. Artifact identity is required as +`package` for Android or `bundle_id` for iOS. `sha256` and `source_commit` are +optional but recommended; when present they are checked exactly. + +`service_probes` must contain at least one URL. `{HOST}` and `{PUBLIC_HOST}` +are expanded from the top-level values. Each probe is checked from the actual +tester node as DNS, TCP, and HTTP. A status outside 2xx–4xx fails transport +readiness; `expected_status` can require an exact allow-list. HTTP 401/403 can +therefore be a valid reachability result when explicitly expected. + +`fixtures` and `cleanup` are declarations, not commands. A required fixture +needs non-empty `references`; a required cleanup needs non-empty `steps` or +`run`. For a read-only run, use `required: false` with a non-empty `reason`. +The skill never executes these values. + +## Tester-node contract + +`tester.runner.kind: local` executes platform and network probes on the current +node. It is only valid for `node: local`/`localhost`. A non-local node must use +`kind: contract` and a non-empty argv `command`. The command receives one JSON +request on stdin and returns one JSON object on stdout; no shell string is +accepted. + +Requests have `operation` (`platform_probe`, `network_probe`, `install`, or +`start`), `platform`, `node`, and operation-specific fields. The response is: + +```json +{ + "status": "ready", + "observed": "short evidence", + "checks": { + "macos": {"status": "ready", "observed": "Darwin 24", "expected": "Darwin"} + } +} +``` + +`checks` is optional; each key becomes a separate evidence row. A non-zero, +invalid, or unavailable runner is `environment-fail`. The adapter is expected +to execute on the resolved tester node, not on the coordinator. See +[boundaries.md](boundaries.md). + +## Evidence + +YAML contains `mode`, overall `status`, timestamps, a summary, and `checks`. +CSV uses these stable columns: + +`check_id,mode,platform,node,target,observed,expected,status,remediation,timestamp` + +Every check must use one of `ready`, `blocked`, or `environment-fail`; the +report does not invent `fail` or `not-run` testcase results. diff --git a/skills/test-preflight/scripts/preflight.py b/skills/test-preflight/scripts/preflight.py new file mode 100644 index 0000000..fde8bcb --- /dev/null +++ b/skills/test-preflight/scripts/preflight.py @@ -0,0 +1,549 @@ +#!/usr/bin/env python3 +"""Run the test-preflight verify/prepare contract.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import os +import platform as host_platform +import plistlib +import re +import shutil +import socket +import subprocess +import sys +import urllib.error +import urllib.parse +import urllib.request +import zipfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping + +try: + import yaml +except ImportError as exc: # pragma: no cover + raise SystemExit("test-preflight 需要 PyYAML；请安装运行依赖后重试") from exc + + +PLATFORMS = {"android", "ios"} +STATUSES = {"ready", "blocked", "environment-fail"} +EVIDENCE_FIELDS = [ + "check_id", "mode", "platform", "node", "target", "observed", "expected", + "status", "remediation", "timestamp", +] +SHA256 = re.compile(r"^[0-9a-fA-F]{64}$") +COMMIT = re.compile(r"^[0-9a-fA-F]{40}$") + + +def now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def text(value: Any) -> str: + if isinstance(value, (dict, list)): + return json.dumps(value, ensure_ascii=False, sort_keys=True) + return "" if value is None else str(value) + + +def status_from(value: Any, default: str = "environment-fail") -> str: + return value if value in STATUSES else default + + +def sha256(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 command_text(command: list[str], *, payload: Mapping[str, Any] | None = None) -> tuple[int, str]: + try: + process = subprocess.run( + command, + input=(json.dumps(payload, ensure_ascii=False) if payload is not None else None), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + check=False, + ) + except OSError as exc: + return 127, str(exc) + return process.returncode, process.stdout.strip() + + +class Evidence: + def __init__(self, mode: str) -> None: + self.mode = mode + self.started_at = now() + self.rows: list[dict[str, str]] = [] + + def add( + self, + check_id: str, + *, + platform: str = "all", + node: str = "coordinator", + target: str = "", + observed: Any = "", + expected: Any = "", + status: str, + remediation: str = "", + ) -> None: + if status not in STATUSES: + raise ValueError(f"invalid evidence status: {status}") + self.rows.append({ + "check_id": check_id, + "mode": self.mode, + "platform": platform, + "node": node, + "target": target, + "observed": text(observed), + "expected": text(expected), + "status": status, + "remediation": remediation, + "timestamp": now(), + }) + + def overall(self) -> str: + statuses = {row["status"] for row in self.rows} + if "environment-fail" in statuses: + return "environment-fail" + if "blocked" in statuses: + return "blocked" + return "ready" + + def payload(self) -> dict[str, Any]: + return { + "schema_version": 1, + "mode": self.mode, + "status": self.overall(), + "started_at": self.started_at, + "finished_at": now(), + "summary": {state: sum(row["status"] == state for row in self.rows) for state in sorted(STATUSES)}, + "checks": self.rows, + } + + +def load_config(path: Path) -> dict[str, Any]: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError("配置根必须是 mapping") + return data + + +def expand(value: str, config: Mapping[str, Any]) -> str: + return value.replace("{HOST}", text(config.get("host"))).replace("{PUBLIC_HOST}", text(config.get("public_host"))) + + +def normalize_url(value: str) -> str: + return value if urllib.parse.urlsplit(value).scheme else "http://" + value.lstrip("/") + + +def argv(value: Any) -> list[str] | None: + if not isinstance(value, list) or not value or not all(isinstance(item, str) and item for item in value): + return None + return value + + +def validate_shape(config: Mapping[str, Any], evidence: Evidence) -> bool: + ok = True + if config.get("schema_version") != 1: + evidence.add("config-schema", target="config", observed=config.get("schema_version"), expected="1", status="blocked", remediation="将 schema_version 设为 1") + ok = False + if not isinstance(config.get("project_root"), str) or not Path(text(config.get("project_root"))).is_absolute(): + evidence.add("config-project-root", target="project_root", observed=config.get("project_root"), expected="absolute project_root", status="blocked", remediation="提供项目根绝对路径") + ok = False + if not isinstance(config.get("host"), str) or not config["host"].strip(): + evidence.add("config-host", target="HOST", expected="non-empty host URL", status="blocked", remediation="提供 HOST") + ok = False + probes = config.get("service_probes") + if not isinstance(probes, list) or not probes: + evidence.add("config-service-probes", target="service_probes", expected="at least one probe", status="blocked", remediation="声明服务 URL 探针") + ok = False + else: + for index, probe in enumerate(probes): + if not isinstance(probe, dict) or not isinstance(probe.get("url"), str) or not probe["url"].strip(): + evidence.add(f"config-service-probe-{index}", target=f"service_probes[{index}]", expected="url", status="blocked", remediation="为每个服务探针提供 URL") + ok = False + matrix = config.get("matrix") + if not isinstance(matrix, list) or not matrix: + evidence.add("config-matrix", target="matrix", expected="non-empty list", status="blocked", remediation="声明至少一个平台矩阵行") + return False + seen: set[str] = set() + for index, row in enumerate(matrix): + prefix = f"matrix-{index}" + if not isinstance(row, dict): + evidence.add(prefix, target=f"matrix[{index}]", expected="mapping", status="blocked", remediation="修正矩阵行") + ok = False + continue + platform_name = text(row.get("platform")).lower() + if platform_name not in PLATFORMS or platform_name in seen: + evidence.add(f"{prefix}-platform", platform=platform_name or "unknown", target=f"matrix[{index}].platform", expected="unique android or ios", status="blocked", remediation="为每个平台保留唯一矩阵行") + ok = False + seen.add(platform_name) + tester = row.get("tester") + if not isinstance(tester, dict) or not isinstance(tester.get("binding"), str) or not tester["binding"].startswith("tester."): + evidence.add(f"{prefix}-binding", platform=platform_name, target=f"matrix[{index}].tester.binding", expected="tester.", status="blocked", remediation="由 agent-runtime 解析并传入 tester binding") + ok = False + if not isinstance(tester, dict) or not isinstance(tester.get("node"), str) or not tester["node"].strip(): + evidence.add(f"{prefix}-node", platform=platform_name, target=f"matrix[{index}].tester.node", expected="actual tester node", status="blocked", remediation="提供实际 tester 节点名") + ok = False + artifact = row.get("artifact") + identity_key = "package" if platform_name == "android" else "bundle_id" + if not isinstance(artifact, dict) or not isinstance(artifact.get("path"), str) or not artifact["path"].strip(): + evidence.add(f"{prefix}-artifact", platform=platform_name, target=f"matrix[{index}].artifact.path", expected="caller-provided artifact path", status="blocked", remediation="提供本轮 APK/IPA，不要让 preflight 猜产物") + ok = False + if not isinstance(artifact, dict) or not isinstance(artifact.get(identity_key), str) or not artifact[identity_key].strip(): + evidence.add(f"{prefix}-identity", platform=platform_name, target=f"matrix[{index}].artifact.{identity_key}", expected="declared artifact identity", status="blocked", remediation=f"声明 artifact.{identity_key}") + ok = False + if isinstance(artifact, dict) and artifact.get("sha256") is not None and not SHA256.fullmatch(text(artifact.get("sha256"))): + evidence.add(f"{prefix}-sha256", platform=platform_name, target=f"matrix[{index}].artifact.sha256", expected="64 hex characters", status="blocked", remediation="提供正确 SHA-256 或删除字段") + ok = False + if isinstance(artifact, dict) and artifact.get("source_commit") is not None and not COMMIT.fullmatch(text(artifact.get("source_commit"))): + evidence.add(f"{prefix}-commit", platform=platform_name, target=f"matrix[{index}].artifact.source_commit", expected="40-character commit SHA", status="blocked", remediation="提供产物来源 commit SHA") + ok = False + return ok + + +def artifact_identity(path: Path, platform_name: str, artifact: Mapping[str, Any]) -> tuple[str | None, str]: + if platform_name == "ios": + try: + with zipfile.ZipFile(path) as archive: + names = [name for name in archive.namelist() if re.fullmatch(r"Payload/[^/]+\.app/Info\.plist", name)] + if len(names) != 1: + return None, "IPA 中没有唯一 Payload/*.app/Info.plist" + plist = plistlib.loads(archive.read(names[0])) + value = plist.get("CFBundleIdentifier") + return (value, "Info.plist CFBundleIdentifier") if isinstance(value, str) else (None, "Info.plist 缺少 CFBundleIdentifier") + except (OSError, zipfile.BadZipFile, plistlib.InvalidFileException) as exc: + return None, f"IPA 读取失败: {exc}" + command = argv(artifact.get("identity_command")) + if command: + command = [item.replace("{artifact}", str(path)) for item in command] + else: + tool = shutil.which("apkanalyzer") or shutil.which("aapt2") or shutil.which("aapt") + if not tool: + return None, "缺少 apkanalyzer/aapt2/aapt；无法读取 APK package identity" + command = [tool, "manifest", "application-id", str(path)] if tool.endswith("apkanalyzer") else [tool, "dump", "badging", str(path)] + code, output = command_text(command) + if code != 0: + return None, output[-500:] or f"identity command exited {code}" + match = re.search(r"(?:application-id|name)\s*[:=]\s*['\"]?([A-Za-z0-9_.]+)", output) + if not match: + match = re.search(r"package:\s+name='([^']+)'", output) + if not match: + match = re.fullmatch(r"\s*([A-Za-z0-9_.]+)\s*", output) + return (match.group(1), output[-500:]) if match else (None, "identity command output did not contain package") + + +def record_artifact(evidence: Evidence, row: Mapping[str, Any], platform_name: str, node: str, index: int) -> bool: + artifact = row.get("artifact") or {} + path = Path(text(artifact.get("path"))).expanduser() + prefix = f"{platform_name}-{index}-artifact" + if not path.is_file(): + evidence.add(prefix, platform=platform_name, node=node, target=str(path), expected="existing caller-provided artifact", status="blocked", remediation="补齐 APK/IPA 路径；不要默认构建") + return False + actual_hash = sha256(path) + expected_hash = artifact.get("sha256") + if expected_hash and actual_hash.lower() != text(expected_hash).lower(): + evidence.add(prefix + "-hash", platform=platform_name, node=node, target=str(path), observed=actual_hash, expected=expected_hash, status="blocked", remediation="确认产物路径或 SHA-256 与本轮构建身份一致") + return False + evidence.add(prefix + "-hash", platform=platform_name, node=node, target=str(path), observed=actual_hash, expected=expected_hash or "file SHA-256 recorded", status="ready") + identity, observed = artifact_identity(path, platform_name, artifact) + expected_identity = text(artifact.get("package" if platform_name == "android" else "bundle_id")) + if identity != expected_identity: + evidence.add(prefix + "-identity", platform=platform_name, node=node, target=str(path), observed=identity or observed, expected=expected_identity, status="environment-fail", remediation="使用与 catalog 身份一致的 APK/IPA，并重新校验产物") + return False + evidence.add(prefix + "-identity", platform=platform_name, node=node, target=str(path), observed=identity, expected=expected_identity, status="ready") + if artifact.get("source_commit"): + evidence.add(prefix + "-commit", platform=platform_name, node=node, target=str(path), observed=artifact.get("source_commit"), expected=artifact.get("source_commit"), status="ready") + return True + + +def local_platform_probe(platform_name: str) -> tuple[str, dict[str, dict[str, Any]]]: + if platform_name == "android": + adb = shutil.which("adb") + if not adb: + return "environment-fail", {"adb": {"status": "environment-fail", "observed": "missing", "expected": "Android SDK platform-tools", "remediation": "在 tester.android 节点安装并配置 adb"}} + code, output = command_text([adb, "devices", "-l"]) + states = [line.split() for line in output.splitlines()[1:] if line.strip()] + device = next((parts[0] for parts in states if len(parts) > 1 and parts[1] == "device"), "") + bad = next((parts for parts in states if len(parts) > 1 and parts[1] in {"unauthorized", "offline"}), None) + if bad: + return "environment-fail", {"adb": {"status": "ready", "observed": output, "expected": "adb responds"}, "device": {"status": "environment-fail", "observed": " ".join(bad), "expected": "authorized device", "remediation": "解锁 Android 设备并接受 RSA 授权；offline 时重启 adb/设备"}} + if code != 0 or not device: + return "environment-fail", {"adb": {"status": "environment-fail", "observed": output, "expected": "one authorized device", "remediation": "连接并授权 Android device/emulator"}} + boot_code, boot = command_text([adb, "-s", device, "shell", "getprop", "sys.boot_completed"]) + booted = boot_code == 0 and boot.strip() == "1" + return ("ready" if booted else "environment-fail", { + "adb": {"status": "ready", "observed": adb, "expected": "adb available"}, + "device": {"status": "ready", "observed": device, "expected": "authorized device"}, + "boot": {"status": "ready" if booted else "environment-fail", "observed": boot, "expected": "sys.boot_completed=1", "remediation": "等待 emulator boot complete 后重跑"}, + }) + if host_platform.system() != "Darwin": + return "environment-fail", {"macos": {"status": "environment-fail", "observed": host_platform.system(), "expected": "Darwin/macOS tester.ios node", "remediation": "将 tester.ios 指向可用 macOS，不要在 Linux 伪装 iOS"}} + xcrun = shutil.which("xcrun") + if not xcrun: + return "environment-fail", {"xcrun": {"status": "environment-fail", "observed": "missing", "expected": "xcrun and simctl", "remediation": "在 Mac 安装 Xcode Command Line Tools/Xcode"}} + code, output = command_text([xcrun, "simctl", "list", "devices", "available"]) + booted = any("(Booted)" in line for line in output.splitlines()) + return ("ready" if code == 0 and booted else "environment-fail", { + "macos": {"status": "ready", "observed": "Darwin", "expected": "Darwin/macOS"}, + "xcrun": {"status": "ready" if code == 0 else "environment-fail", "observed": output[-500:], "expected": "simctl available", "remediation": "安装 Xcode 并接受许可"}, + "boot": {"status": "ready" if booted else "environment-fail", "observed": "Booted" if booted else output[-300:], "expected": "Booted simulator", "remediation": "启动目标 Simulator 后重跑"}, + }) + + +def runner_call(runner: Mapping[str, Any], request: Mapping[str, Any]) -> tuple[str, str, dict[str, Any]]: + command = argv(runner.get("command")) + if not command: + return "blocked", "runner command missing", {} + code, output = command_text(command, payload=request) + if code != 0: + return "environment-fail", output[-800:], {} + try: + result = json.loads(output) + except json.JSONDecodeError: + return "environment-fail", output[-800:], {} + if not isinstance(result, dict): + return "environment-fail", "runner returned non-object", {} + return status_from(result.get("status")), text(result.get("observed", "")), result + + +def add_runner_checks(evidence: Evidence, result: Mapping[str, Any], *, prefix: str, platform_name: str, node: str, default_expected: str) -> str: + checks = result.get("checks") + if isinstance(checks, dict) and checks: + statuses: list[str] = [] + for name, detail in checks.items(): + detail = detail if isinstance(detail, dict) else {"observed": detail} + status = status_from(detail.get("status"), status_from(result.get("status"))) + statuses.append(status) + evidence.add(f"{prefix}-{name}", platform=platform_name, node=node, target=text(detail.get("target", name)), observed=detail.get("observed", result.get("observed", "")), expected=detail.get("expected", default_expected), status=status, remediation=text(detail.get("remediation", ""))) + return "environment-fail" if "environment-fail" in statuses else ("blocked" if "blocked" in statuses else "ready") + status = status_from(result.get("status")) + evidence.add(prefix, platform=platform_name, node=node, target=prefix, observed=result.get("observed", ""), expected=default_expected, status=status, remediation=text(result.get("remediation", ""))) + return status + + +def require_runner_checks(evidence: Evidence, result: Mapping[str, Any], *, prefix: str, platform_name: str, node: str, required: set[str]) -> bool: + checks = result.get("checks") + missing = sorted(required - set(checks)) if isinstance(checks, dict) else sorted(required) + if not missing: + return True + evidence.add(prefix + "-contract", platform=platform_name, node=node, target="runner response", observed={"missing": missing}, expected=sorted(required), status="blocked", remediation="让 tester 节点 adapter 返回每个必需的 adb/boot、macOS/xcrun/simctl 或 DNS/TCP/HTTP 证据") + return False + + +def network_local(url: str) -> list[dict[str, Any]]: + parsed = urllib.parse.urlsplit(url) + host = parsed.hostname or "" + port = parsed.port or (443 if parsed.scheme == "https" else 80) + try: + addresses = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except OSError as exc: + return [{"name": "dns", "status": "environment-fail", "observed": str(exc), "expected": "resolvable hostname", "remediation": "检查 tester 节点 DNS/VPN/网络"}] + rows: list[dict[str, Any]] = [{"name": "dns", "status": "ready", "observed": sorted({item[4][0] for item in addresses}), "expected": "resolvable hostname"}] + try: + with socket.create_connection((host, port), timeout=5): + rows.append({"name": "tcp", "status": "ready", "observed": f"{host}:{port}", "expected": "TCP connection"}) + except OSError as exc: + rows.append({"name": "tcp", "status": "environment-fail", "observed": str(exc), "expected": f"TCP {host}:{port}", "remediation": "检查端口、防火墙和测试节点到 HOST 的路由"}) + return rows + try: + with urllib.request.urlopen(urllib.request.Request(url, method="GET"), timeout=5) as response: + code = response.status + rows.append({"name": "http", "status": "ready" if 200 <= code < 500 else "environment-fail", "observed": code, "expected": "HTTP 2xx-4xx", "remediation": "检查后端健康状态" if code >= 500 else ""}) + except urllib.error.HTTPError as exc: + rows.append({"name": "http", "status": "ready" if 200 <= exc.code < 500 else "environment-fail", "observed": exc.code, "expected": "HTTP 2xx-4xx", "remediation": "检查后端服务，不把 5xx 当成可用" if exc.code >= 500 else ""}) + except (OSError, urllib.error.URLError) as exc: + rows.append({"name": "http", "status": "environment-fail", "observed": str(exc), "expected": "HTTP response", "remediation": "检查服务监听、代理、TLS 和节点网络"}) + return rows + + +def check_network(evidence: Evidence, config: Mapping[str, Any], row: Mapping[str, Any], platform_name: str, node: str, index: int) -> bool: + tester = row.get("tester") or {} + runner = tester.get("runner") if isinstance(tester, dict) else None + is_local = node in {"local", "localhost", "127.0.0.1"} + statuses: list[str] = [] + for probe_index, probe in enumerate(config.get("service_probes") or []): + raw_url = text(probe.get("url")) + if "{PUBLIC_HOST}" in raw_url and not text(config.get("public_host")).strip(): + evidence.add(f"{platform_name}-{index}-network-{probe_index}-public-host", platform=platform_name, node=node, target=raw_url, expected="PUBLIC_HOST configured", status="blocked", remediation="提供 PUBLIC_HOST 或改用 HOST 探针") + statuses.append("blocked") + continue + url = normalize_url(expand(raw_url, config)) + if not urllib.parse.urlsplit(url).hostname: + evidence.add(f"{platform_name}-{index}-network-{probe_index}-url", platform=platform_name, node=node, target=url, expected="valid probe URL", status="blocked", remediation="修正 service_probes URL 或 HOST/PUBLIC_HOST") + statuses.append("blocked") + continue + if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"): + details = network_local(url) + expected_statuses = probe.get("expected_status") + for detail in details: + if detail["name"] == "http" and isinstance(expected_statuses, list) and detail["status"] == "ready" and detail["observed"] not in expected_statuses: + detail = {**detail, "status": "environment-fail", "expected": expected_statuses, "remediation": "检查服务探针的 expected_status 或后端响应"} + status = status_from(detail["status"]) + statuses.append(status) + evidence.add(f"{platform_name}-{index}-network-{probe_index}-{detail['name']}", platform=platform_name, node=node, target=url, observed=detail["observed"], expected=detail["expected"], status=status, remediation=detail.get("remediation", "")) + elif isinstance(runner, dict) and runner.get("kind") == "contract": + status, observed, result = runner_call(runner, {"operation": "network_probe", "platform": platform_name, "node": node, "url": url, "probe": probe}) + result = result or {"status": status, "observed": observed} + complete = require_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-network-{probe_index}", platform_name=platform_name, node=node, required={"dns", "tcp", "http"}) + network_status = add_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-network-{probe_index}", platform_name=platform_name, node=node, default_expected="DNS/TCP/HTTP reachable from actual tester node") + statuses.append(network_status if complete else "blocked") + else: + evidence.add(f"{platform_name}-{index}-network-{probe_index}", platform=platform_name, node=node, target=url, expected="node runner contract", status="blocked", remediation="远端 tester 必须提供 agent-runtime adapter runner") + statuses.append("blocked") + return "environment-fail" not in statuses and "blocked" not in statuses + + +def check_fixture_cleanup(evidence: Evidence, config: Mapping[str, Any], platform_name: str, node: str, index: int) -> bool: + ok = True + fixture = config.get("fixtures") + valid_fixture = isinstance(fixture, dict) and (not fixture.get("required", True) or bool(fixture.get("references"))) + if not valid_fixture: + evidence.add(f"{platform_name}-{index}-fixture", platform=platform_name, node=node, target="fixtures", expected="required fixture references", status="blocked", remediation="声明隔离测试账号/fixture；不要在运行中临时创建") + ok = False + else: + evidence.add(f"{platform_name}-{index}-fixture", platform=platform_name, node=node, target="fixtures", observed=fixture.get("references"), expected="declared fixture references", status="ready") + cleanup = config.get("cleanup") + valid_noop = isinstance(cleanup, dict) and cleanup.get("required") is False and bool(text(cleanup.get("reason")).strip()) + valid_steps = isinstance(cleanup, dict) and (bool(cleanup.get("steps")) or bool(text(cleanup.get("run")).strip())) + if not (valid_noop or valid_steps): + evidence.add(f"{platform_name}-{index}-cleanup", platform=platform_name, node=node, target="cleanup", expected="steps/run or explicit read-only reason", status="blocked", remediation="声明 teardown/rollback；清理缺失时禁止开始测试") + ok = False + else: + evidence.add(f"{platform_name}-{index}-cleanup", platform=platform_name, node=node, target="cleanup", observed=cleanup, expected="cleanup contract declared", status="ready") + return ok + + +def run(config: Mapping[str, Any], mode: str) -> Evidence: + evidence = Evidence(mode) + shape_ok = validate_shape(config, evidence) + deployment_ok = True + if mode == "prepare": + deployment = config.get("deployment") + environment = deployment.get("environment") if isinstance(deployment, dict) else None + command = argv(deployment.get("command")) if isinstance(deployment, dict) else None + if environment != "test": + evidence.add("deployment-environment", target="deployment.environment", observed=environment, expected="test", status="blocked", remediation="prepare 只允许 deployer 的 test 环境") + if not command: + evidence.add("deployment-command", target="deployment.command", expected="existing deployer adapter command", status="blocked", remediation="提供既有 deployer 的 JSON command contract；不要在 skill 内实现部署") + deployment_ok = False + elif not command: + evidence.add("deployment-command", target="deployment.command", expected="existing deployer adapter command", status="blocked", remediation="提供既有 deployer 的 JSON command contract；不要在 skill 内实现部署") + deployment_ok = False + else: + code, output = command_text(command, payload={"operation": "prepare", "environment": "test", "project_root": config.get("project_root"), "host": config.get("host"), "public_host": config.get("public_host")}) + if code != 0: + evidence.add("deployment", target="test", observed=output[-800:], expected="deployer receipt status=ready", status="environment-fail", remediation="按 deployer 返回证据修复测试后端，再重跑 prepare") + deployment_ok = False + else: + try: + receipt = json.loads(output) + except json.JSONDecodeError: + receipt = {} + status = status_from(receipt.get("status"), "environment-fail") if isinstance(receipt, dict) else "environment-fail" + evidence.add("deployment", target="test", observed=receipt.get("observed", output[-800:]) if isinstance(receipt, dict) else output[-800:], expected="deployer receipt status=ready", status=status, remediation="检查既有 deployer 的测试环境 receipt" if status != "ready" else "") + deployment_ok = status == "ready" + if not shape_ok: + return evidence + for index, row in enumerate(config["matrix"]): + platform_name = text(row.get("platform")).lower() + tester = row.get("tester") or {} + node = text(tester.get("node")) + binding = text(tester.get("binding")) + evidence.add(f"{platform_name}-{index}-binding", platform=platform_name, node=node, target=binding, observed=binding, expected="resolved tester. binding", status="ready") + artifact_ok = record_artifact(evidence, row, platform_name, node, index) + fixture_ok = check_fixture_cleanup(evidence, config, platform_name, node, index) + runner = tester.get("runner") if isinstance(tester, dict) else None + is_local = node in {"local", "localhost", "127.0.0.1"} + if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"): + platform_status, details = local_platform_probe(platform_name) + for name, detail in details.items(): + evidence.add(f"{platform_name}-{index}-platform-{name}", platform=platform_name, node=node, target=name, observed=detail.get("observed", ""), expected=detail.get("expected", "platform ready"), status=status_from(detail.get("status"), platform_status), remediation=detail.get("remediation", "")) + platform_ok = platform_status == "ready" + elif isinstance(runner, dict) and runner.get("kind") == "contract": + status, observed, result = runner_call(runner, {"operation": "platform_probe", "platform": platform_name, "node": node, "binding": binding}) + result = result or {"status": status, "observed": observed} + required = {"adb", "device", "boot"} if platform_name == "android" else {"macos", "xcrun", "simctl", "boot"} + complete = require_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-platform", platform_name=platform_name, node=node, required=required) + platform_status = add_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-platform", platform_name=platform_name, node=node, default_expected="platform/device ready") + platform_ok = complete and platform_status == "ready" + else: + evidence.add(f"{platform_name}-{index}-platform-runner", platform=platform_name, node=node, target=node, expected="node runner contract", status="blocked", remediation="远端 tester 必须由 agent-runtime 提供可执行 runner") + platform_ok = False + network_ok = check_network(evidence, config, row, platform_name, node, index) + row_ok = artifact_ok and fixture_ok and platform_ok and network_ok + if mode == "prepare": + artifact = row.get("artifact") or {} + identity = artifact.get("package" if platform_name == "android" else "bundle_id") + target_path = text(artifact.get("remote_path") or artifact.get("path")) + if not row_ok or not deployment_ok: + evidence.add(f"{platform_name}-{index}-install", platform=platform_name, node=node, target=target_path, observed="not-run", expected="verify and deployment ready", status="blocked", remediation="先修复前置检查；prepare 不伪造安装成功") + evidence.add(f"{platform_name}-{index}-start", platform=platform_name, node=node, target=text(identity), observed="not-run", expected="install ready", status="blocked", remediation="安装前置失败，未启动 app") + else: + install_request = {"operation": "install", "platform": platform_name, "node": node, "artifact": target_path, "identity": identity} + start_request = {"operation": "start", "platform": platform_name, "node": node, "identity": identity} + if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"): + if platform_name == "android": + adb = shutil.which("adb") or "adb" + serial = os.environ.get("ANDROID_SERIAL", "") + install_code, install_output = command_text([adb, "-s", serial, "install", "-r", target_path] if serial else [adb, "install", "-r", target_path]) + install_result = {"status": "ready" if install_code == 0 else "environment-fail", "observed": install_output} + launch = [adb, "-s", serial, "shell", "monkey", "-p", text(identity), "1"] if serial else [adb, "shell", "monkey", "-p", text(identity), "1"] + start_code, start_output = command_text(launch) if install_code == 0 else (1, "install failed") + start_result = {"status": "ready" if start_code == 0 else "environment-fail", "observed": start_output} + else: + install_result = {"status": "environment-fail", "observed": "local iOS install requires a macOS runner", "remediation": "use tester.ios contract runner on Mac"} + start_result = {"status": "blocked", "observed": "not-run", "remediation": "install failed"} + add_runner_checks(evidence, install_result, prefix=f"{platform_name}-{index}-install", platform_name=platform_name, node=node, default_expected="artifact installed") + add_runner_checks(evidence, start_result, prefix=f"{platform_name}-{index}-start", platform_name=platform_name, node=node, default_expected="app launched") + elif isinstance(runner, dict) and runner.get("kind") == "contract": + install_status, install_observed, install_result = runner_call(runner, install_request) + add_runner_checks(evidence, install_result or {"status": install_status, "observed": install_observed}, prefix=f"{platform_name}-{index}-install", platform_name=platform_name, node=node, default_expected="artifact installed") + if install_status == "ready": + start_status, start_observed, start_result = runner_call(runner, start_request) + add_runner_checks(evidence, start_result or {"status": start_status, "observed": start_observed}, prefix=f"{platform_name}-{index}-start", platform_name=platform_name, node=node, default_expected="app launched") + else: + evidence.add(f"{platform_name}-{index}-start", platform=platform_name, node=node, target=text(identity), observed="not-run", expected="install ready", status="blocked", remediation="安装失败，先修复 tester 节点") + return evidence + + +def write_evidence(evidence: Evidence, yaml_path: Path, csv_path: Path) -> dict[str, Any]: + payload = evidence.payload() + yaml_path.parent.mkdir(parents=True, exist_ok=True) + csv_path.parent.mkdir(parents=True, exist_ok=True) + yaml_path.write_text(yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), encoding="utf-8") + with csv_path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=EVIDENCE_FIELDS) + writer.writeheader() + writer.writerows(evidence.rows) + return {"status": payload["status"], "yaml": str(yaml_path), "csv": str(csv_path), "checks": len(evidence.rows), "summary": payload["summary"]} + + +def main() -> int: + parser = argparse.ArgumentParser(description="Verify or prepare a test environment and emit YAML/CSV evidence") + parser.add_argument("mode", choices=("verify", "prepare")) + parser.add_argument("--config", required=True, type=Path) + parser.add_argument("--yaml", dest="yaml_path", type=Path, default=Path("preflight.yaml")) + parser.add_argument("--csv", dest="csv_path", type=Path, default=Path("preflight.csv")) + args = parser.parse_args() + try: + config = load_config(args.config) + evidence = run(config, args.mode) + result = write_evidence(evidence, args.yaml_path, args.csv_path) + except (OSError, ValueError, yaml.YAMLError) as exc: + print(json.dumps({"status": "blocked", "error": str(exc)}, ensure_ascii=False), file=sys.stderr) + return 2 + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 if result["status"] == "ready" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/test-preflight/tests/fixtures/contract_runner.py b/skills/test-preflight/tests/fixtures/contract_runner.py new file mode 100644 index 0000000..f4d5e95 --- /dev/null +++ b/skills/test-preflight/tests/fixtures/contract_runner.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Deterministic runner fixture for test-preflight contract tests.""" + +import json +import os +import sys + +request = json.load(sys.stdin) +scenario = os.environ.get("PREFLIGHT_RUNNER_SCENARIO", "ready") +operation = request.get("operation") + +if operation in {"deployment", "prepare"}: + print(json.dumps({"status": "ready", "observed": "mock test deploy receipt"})) +elif operation == "platform_probe": + platform = request.get("platform") + if scenario == "android-unauthorized" and platform == "android": + print(json.dumps({"status": "environment-fail", "checks": { + "adb": {"status": "ready", "observed": "adb 35", "expected": "adb responds"}, + "device": {"status": "environment-fail", "observed": "emulator-5554 unauthorized", "expected": "authorized device", "remediation": "accept RSA authorization"}, + "boot": {"status": "environment-fail", "observed": "unknown", "expected": "boot complete"}, + }})) + elif scenario == "mac-unavailable" and platform == "ios": + print(json.dumps({"status": "environment-fail", "checks": { + "macos": {"status": "environment-fail", "observed": "Mac unavailable", "expected": "Darwin tester node", "remediation": "restore tester.ios Mac"}, + }})) + else: + print(json.dumps({"status": "ready", "checks": { + "adb": {"status": "ready", "observed": "adb 35", "expected": "adb responds"}, + "device": {"status": "ready", "observed": "authorized", "expected": "authorized device"}, + "boot": {"status": "ready", "observed": "boot complete", "expected": "boot complete"}, + "macos": {"status": "ready", "observed": "Darwin", "expected": "Darwin"}, + "xcrun": {"status": "ready", "observed": "xcrun", "expected": "xcrun"}, + "simctl": {"status": "ready", "observed": "available", "expected": "simctl"}, + }})) +elif operation == "network_probe": + if scenario == "host-unreachable": + print(json.dumps({"status": "environment-fail", "checks": { + "dns": {"status": "ready", "observed": "10.0.0.2", "expected": "DNS resolves"}, + "tcp": {"status": "environment-fail", "observed": "connection refused", "expected": "TCP reachable", "remediation": "check HOST route"}, + }})) + elif scenario == "http-failure": + print(json.dumps({"status": "environment-fail", "checks": { + "dns": {"status": "ready", "observed": "10.0.0.2", "expected": "DNS resolves"}, + "tcp": {"status": "ready", "observed": "443 open", "expected": "TCP reachable"}, + "http": {"status": "environment-fail", "observed": 503, "expected": "HTTP 2xx-4xx", "remediation": "check service health"}, + }})) + else: + print(json.dumps({"status": "ready", "checks": { + "dns": {"status": "ready", "observed": "10.0.0.2", "expected": "DNS resolves"}, + "tcp": {"status": "ready", "observed": "443 open", "expected": "TCP reachable"}, + "http": {"status": "ready", "observed": 200, "expected": "HTTP 2xx-4xx"}, + }})) +elif operation in {"install", "start"}: + print(json.dumps({"status": "ready", "observed": operation + "ed mock artifact"})) +else: + print(json.dumps({"status": "environment-fail", "observed": "unknown operation"})) diff --git a/skills/test-preflight/tests/test_preflight.py b/skills/test-preflight/tests/test_preflight.py new file mode 100644 index 0000000..03b7799 --- /dev/null +++ b/skills/test-preflight/tests/test_preflight.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import json +import os +import plistlib +import subprocess +import sys +import tempfile +import unittest +import zipfile +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +import preflight # noqa: E402 + + +FIXTURE_RUNNER = Path(__file__).resolve().parent / "fixtures" / "contract_runner.py" + + +class PreflightTests(unittest.TestCase): + def setUp(self) -> None: + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + self.apk = self.root / "writes.apk" + self.apk.write_bytes(b"mock apk") + self.ipa = self.root / "writes.ipa" + with zipfile.ZipFile(self.ipa, "w") as archive: + archive.writestr("Payload/Writes.app/Info.plist", plistlib.dumps({"CFBundleIdentifier": "page.writes"})) + self.identity = self.root / "identity.py" + self.identity.write_text("print('page.writes')\n", encoding="utf-8") + + def tearDown(self) -> None: + self.temp.cleanup() + + def runner(self, scenario: str = "ready") -> dict[str, object]: + return {"kind": "contract", "command": [sys.executable, str(FIXTURE_RUNNER)]} + + def config(self, *, platform: str = "android", artifact: Path | None = None, node: str = "tester.android", scenario: str = "ready") -> dict[str, object]: + artifact = artifact or self.apk + artifact_row: dict[str, object] = { + "path": str(artifact), + "package": "page.writes" if platform == "android" else None, + "bundle_id": "page.writes" if platform == "ios" else None, + } + if platform == "android": + artifact_row["identity_command"] = [sys.executable, str(self.identity)] + artifact_row.pop("bundle_id") + else: + artifact_row.pop("package") + return { + "schema_version": 1, + "project_root": str(self.root), + "host": "https://test.example.invalid", + "public_host": "https://public.example.invalid", + "service_probes": [{"id": "health", "url": "{HOST}/health", "expected_status": [200]}], + "fixtures": {"required": True, "references": ["isolated-account"]}, + "cleanup": {"required": True, "steps": ["cleanup-run-data"]}, + "matrix": [{ + "platform": platform, + "tester": {"binding": "tester.android" if platform == "android" else "tester.ios", "node": node, "runner": self.runner(scenario)}, + "artifact": artifact_row, + }], + } + + def write_config(self, data: dict[str, object]) -> Path: + path = self.root / "matrix.yaml" + import yaml + path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") + return path + + def invoke(self, mode: str, data: dict[str, object]) -> tuple[int, dict[str, object], dict[str, object]]: + config = self.write_config(data) + yaml_path, csv_path = self.root / f"{mode}.yaml", self.root / f"{mode}.csv" + env = os.environ.copy() + env["PREFLIGHT_RUNNER_SCENARIO"] = "ready" + process = subprocess.run([ + sys.executable, str(ROOT / "scripts" / "preflight.py"), mode, + "--config", str(config), "--yaml", str(yaml_path), "--csv", str(csv_path), + ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + import yaml + return process.returncode, yaml.safe_load(yaml_path.read_text(encoding="utf-8")), {"stdout": process.stdout, "stderr": process.stderr} + + def test_missing_artifact_is_blocked_and_report_has_contract_columns(self): + data = self.config(artifact=self.root / "missing.apk") + code, report, _ = self.invoke("verify", data) + self.assertEqual(code, 1) + self.assertEqual(report["status"], "blocked") + row = next(row for row in report["checks"] if row["check_id"] == "android-0-artifact") + self.assertEqual(row["status"], "blocked") + self.assertEqual(set(row), set(preflight.EVIDENCE_FIELDS)) + + def test_android_unauthorized_is_environment_fail(self): + data = self.config() + config = self.write_config(data) + out = self.root / "unauthorized.yaml" + env = os.environ.copy() + env["PREFLIGHT_RUNNER_SCENARIO"] = "android-unauthorized" + process = subprocess.run([ + sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config), "--yaml", str(out), "--csv", str(self.root / "u.csv") + ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + report = json.loads(json.dumps(__import__("yaml").safe_load(out.read_text(encoding="utf-8")))) + self.assertEqual(process.returncode, 1) + self.assertEqual(report["status"], "environment-fail") + self.assertEqual(next(row for row in report["checks"] if row["check_id"].endswith("-device"))["status"], "environment-fail") + + def test_linux_without_xcrun_is_environment_fail(self): + data = self.config(platform="ios", artifact=self.ipa, node="local") + data["matrix"][0]["tester"]["runner"] = {"kind": "local"} + with mock.patch.object(preflight.host_platform, "system", return_value="Linux"), mock.patch.object(preflight, "network_local", return_value=[{"name": "dns", "status": "ready", "observed": "ok", "expected": "DNS"}, {"name": "tcp", "status": "ready", "observed": "ok", "expected": "TCP"}, {"name": "http", "status": "ready", "observed": 200, "expected": "HTTP"}]): + result = preflight.run(data, "verify") + self.assertEqual(result.overall(), "environment-fail") + row = next(row for row in result.rows if row["check_id"].endswith("-platform-macos")) + self.assertEqual(row["status"], "environment-fail") + + def test_mac_unavailable_contract_is_environment_fail(self): + data = self.config(platform="ios", artifact=self.ipa) + config_path = self.write_config(data) + env = os.environ.copy() + env["PREFLIGHT_RUNNER_SCENARIO"] = "mac-unavailable" + output = self.root / "mac.yaml" + process = subprocess.run([ + sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "mac.csv") + ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + import yaml + report = yaml.safe_load(output.read_text(encoding="utf-8")) + self.assertEqual(process.returncode, 1) + self.assertEqual(report["status"], "environment-fail") + self.assertTrue(any(row["check_id"].endswith("-macos") and row["status"] == "environment-fail" for row in report["checks"])) + + def test_missing_fixture_or_cleanup_is_blocked(self): + data = self.config() + data["fixtures"] = {"required": True, "references": []} + data["cleanup"] = {"required": True, "steps": []} + result = preflight.run(data, "verify") + self.assertEqual(result.overall(), "blocked") + self.assertTrue(any(row["check_id"].endswith("-fixture") and row["status"] == "blocked" for row in result.rows)) + self.assertTrue(any(row["check_id"].endswith("-cleanup") and row["status"] == "blocked" for row in result.rows)) + + def test_host_unreachable_from_actual_runner_is_environment_fail(self): + data = self.config() + config_path = self.write_config(data) + env = os.environ.copy() + env["PREFLIGHT_RUNNER_SCENARIO"] = "host-unreachable" + output = self.root / "host.yaml" + process = subprocess.run([ + sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "host.csv") + ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + import yaml + report = yaml.safe_load(output.read_text(encoding="utf-8")) + self.assertEqual(process.returncode, 1) + self.assertEqual(report["status"], "environment-fail") + self.assertTrue(any("network" in row["check_id"] and row["status"] == "environment-fail" for row in report["checks"])) + + def test_http_500_is_environment_fail_not_product_fail(self): + data = self.config() + config_path = self.write_config(data) + env = os.environ.copy() + env["PREFLIGHT_RUNNER_SCENARIO"] = "http-failure" + output = self.root / "http.yaml" + process = subprocess.run([ + sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "http.csv") + ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + import yaml + report = yaml.safe_load(output.read_text(encoding="utf-8")) + self.assertEqual(process.returncode, 1) + http_row = next(row for row in report["checks"] if row["check_id"].endswith("-http")) + self.assertEqual(http_row["observed"], "503") + self.assertEqual(http_row["status"], "environment-fail") + + def test_prepare_without_deployer_is_blocked_and_never_installs(self): + data = self.config() + code, report, _ = self.invoke("prepare", data) + self.assertEqual(code, 1) + self.assertEqual(report["status"], "blocked") + self.assertFalse(any(row["check_id"].endswith("-install") and row["status"] == "ready" for row in report["checks"])) + self.assertTrue(any(row["check_id"] == "deployment-command" for row in report["checks"])) + + def test_prepare_mock_contract_calls_install_and_start(self): + data = self.config() + data["deployment"] = {"environment": "test", "command": [sys.executable, str(FIXTURE_RUNNER)]} + code, report, _ = self.invoke("prepare", data) + self.assertEqual(code, 0) + self.assertEqual(report["status"], "ready") + ids = {row["check_id"] for row in report["checks"]} + self.assertIn("android-0-install", ids) + self.assertIn("android-0-start", ids) + + def test_verify_never_calls_install_or_start(self): + data = self.config() + calls: list[str] = [] + original = preflight.runner_call + + def record(runner, request): + calls.append(str(request["operation"])) + return original(runner, request) + + with mock.patch.object(preflight, "runner_call", side_effect=record): + result = preflight.run(data, "verify") + self.assertEqual(result.overall(), "ready") + self.assertNotIn("install", calls) + self.assertNotIn("start", calls) + + +if __name__ == "__main__": + unittest.main()