160 lines
8.1 KiB
Python
160 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise a private installed .app through LaunchServices, not its executable."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import platform
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from typing import BinaryIO
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1]
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
from conformance.harness import ConformanceContext, ResolvedApp, RuntimeClient, require
|
|
|
|
|
|
def verify_rendered_text(ctx: ConformanceContext, probe: Path, log: BinaryIO) -> None:
|
|
marker = "PEBREL TEXT READY"
|
|
# Clear the echoed command before printing the marker; only rendered output
|
|
# can satisfy the screenshot observation.
|
|
ctx.prompt("printf '\\033[2J\\033[H%s%s\\n' 'PEBREL TEXT ' 'READY'")
|
|
ctx.wait_for_line(re.compile(r"^" + re.escape(marker) + r"$"))
|
|
screenshot = ctx.artifact_dir / "installed-app.png"
|
|
deadline = time.monotonic() + 15
|
|
while time.monotonic() < deadline:
|
|
require(capture_screenshot(screenshot, log) == "captured", "installed app screenshot is unavailable")
|
|
result = subprocess.run([str(probe), str(screenshot), marker],
|
|
stdout=log, stderr=subprocess.STDOUT, timeout=15, check=False)
|
|
require(result.returncode in (0, 1), "screenshot text recognition failed")
|
|
if result.returncode == 0:
|
|
return
|
|
time.sleep(0.1)
|
|
raise RuntimeError("installed app did not render the terminal text marker")
|
|
|
|
|
|
def capture_screenshot(destination: Path, log: BinaryIO) -> str:
|
|
try:
|
|
capture = subprocess.run(["/usr/sbin/screencapture", "-x", str(destination)],
|
|
stdout=log, stderr=subprocess.STDOUT, timeout=15, check=False)
|
|
if capture.returncode == 0 and destination.is_file() and destination.stat().st_size > 0:
|
|
return "captured"
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
return "unavailable"
|
|
|
|
|
|
def owned_processes(executable: Path) -> dict[int, str]:
|
|
listing = subprocess.check_output(["/bin/ps", "-axo", "pid=,comm="], text=True)
|
|
result = {}
|
|
for line in listing.splitlines():
|
|
fields = line.strip().split(None, 1)
|
|
if len(fields) == 2 and fields[1] == str(executable):
|
|
process_id = int(fields[0])
|
|
birth = subprocess.run(["/bin/ps", "-p", str(process_id), "-o", "lstart="],
|
|
capture_output=True, text=True, check=False).stdout.strip()
|
|
if birth:
|
|
result[process_id] = birth
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--app", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
report = {"status": "failed", "launch_method": "launchservices",
|
|
"commit": os.environ.get("GITHUB_SHA", "local"),
|
|
"architecture": "aarch64" if platform.machine() == "arm64" else platform.machine(),
|
|
"os_version": platform.mac_ver()[0], "quarantine_assessment": "not_tested"}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
require(sys.platform == "darwin", "LaunchServices requires native macOS")
|
|
with tempfile.TemporaryDirectory(prefix="pebrel-installed-smoke-") as temporary:
|
|
root = Path(temporary).resolve()
|
|
installed = root / "Applications" / "Pebrel Preview.app"
|
|
subprocess.run(["/usr/bin/ditto", str(args.app.resolve()), str(installed)], check=True)
|
|
subprocess.run(["/usr/bin/codesign", "--verify", "--deep", "--strict", str(installed)], check=True)
|
|
app = ResolvedApp(installed)
|
|
logs = args.output.parent / f"{args.output.stem}-artifacts"
|
|
ctx = ConformanceContext(app, "macos", root / "config", root / "work", logs)
|
|
ctx.prepare()
|
|
launcher = None
|
|
with (logs / "launchservices.log").open("wb") as log:
|
|
try:
|
|
text_probe = root / "macos-text"
|
|
subprocess.run([
|
|
"/usr/bin/xcrun", "swiftc", str(SCRIPTS / "conformance/macos_text.swift"),
|
|
"-module-cache-path", str(root / "swift-cache"), "-o", str(text_probe),
|
|
], stdout=log, stderr=subprocess.STDOUT, timeout=90, check=True)
|
|
launcher = subprocess.Popen(
|
|
["/usr/bin/open", "-n", "-W", "-a", str(installed),
|
|
"--env", f"PEBREL_CONFIG_DIR={ctx.config_dir}",
|
|
"--env", f"NEBULA_CONFIG_DIR={ctx.config_dir}",
|
|
"--env", "PATH=/usr/bin:/bin:/usr/sbin:/sbin",
|
|
"--env", "LANG=", "--env", "LC_ALL=", "--env", "LC_CTYPE="],
|
|
cwd="/", stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT,
|
|
)
|
|
deadline = time.monotonic() + 45
|
|
while time.monotonic() < deadline:
|
|
if ctx.port_file.is_file():
|
|
try:
|
|
client = RuntimeClient.from_port_file(ctx.port_file, timeout=2)
|
|
snapshot = client.request("runtime.snapshot")["result"]
|
|
require(snapshot["process_id"] in owned_processes(app.executable),
|
|
"runtime endpoint is not owned by this installed copy")
|
|
ctx.client = client
|
|
ctx.description = client.request("runtime.describe")["result"]
|
|
ctx.refresh_targets(snapshot)
|
|
break
|
|
except (OSError, ValueError, KeyError, RuntimeError):
|
|
pass
|
|
require(launcher.poll() is None, "LaunchServices exited before the app was ready")
|
|
time.sleep(0.1)
|
|
else:
|
|
raise RuntimeError("installed application did not become ready in 45 seconds")
|
|
ctx.detect_shell()
|
|
ctx.prompt('printf "NEBULA_GUI_LOCALE=%s_END\\n" "${LC_ALL:-${LC_CTYPE:-${LANG:-}}}"')
|
|
ctx.wait_for_line(re.compile(r"NEBULA_GUI_LOCALE=[^\r\n]*UTF-?8_END", re.IGNORECASE))
|
|
ctx.prompt('printf "NEBULA_GUI_CWD=%s_END\\n" "$PWD"')
|
|
ctx.wait_for_line(re.compile(r"NEBULA_GUI_CWD=" + re.escape(str(Path.home())) + r"_END"))
|
|
verify_rendered_text(ctx, text_probe, log)
|
|
report.update(screenshot="captured", rendered_text=True)
|
|
for window in ctx.snapshot().get("windows", []):
|
|
ctx.api("window.close", {"window_id": window["id"]})
|
|
deadline = time.monotonic() + 5
|
|
while time.monotonic() < deadline and owned_processes(app.executable):
|
|
time.sleep(0.1)
|
|
require(not owned_processes(app.executable), "installed application did not quit after closing its windows")
|
|
launcher.wait(timeout=5)
|
|
report.update(status="passed", utf8_locale=True, home_cwd=True)
|
|
finally:
|
|
for process_id, birth in owned_processes(app.executable).items():
|
|
if owned_processes(app.executable).get(process_id) == birth:
|
|
os.kill(process_id, signal.SIGKILL)
|
|
if launcher is not None:
|
|
try:
|
|
launcher.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
launcher.kill()
|
|
launcher.wait(timeout=5)
|
|
app.close()
|
|
except Exception as error:
|
|
report["error"] = f"{type(error).__name__}: {error}"
|
|
print(report["error"], file=sys.stderr)
|
|
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return 0 if report["status"] == "passed" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|