226 lines
11 KiB
Python
226 lines
11 KiB
Python
#!/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()
|