#!/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_mobile_capabilities_are_assumed_and_not_a_preflight_gate(self): data = self.config() config_path = self.write_config(data) env = os.environ.copy() env["PREFLIGHT_RUNNER_SCENARIO"] = "capability-missing" output = self.root / "capabilities.yaml" process = subprocess.run([ sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "capabilities.csv"), ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) report = __import__("yaml").safe_load(output.read_text(encoding="utf-8")) self.assertEqual(process.returncode, 0) self.assertEqual(report["status"], "ready") self.assertFalse(any("capability-" in row["check_id"] for row in report["checks"])) 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) self.assertNotIn("capability_probe", calls) if __name__ == "__main__": unittest.main()