import argparse
-import asyncio
-import fcntl
import json
-import os
-import plistlib
import re
import shutil
import subprocess
import sys
-import tempfile
-from contextlib import asynccontextmanager
-from datetime import datetime
from pathlib import Path
DECODE_ARGS = ("UTF-8", "backslashreplace")
# The system log prefixes each line:
-# 2025-01-17 16:14:29.090 Df iOSTestbed[23987:1fd393b4] (Python) ...
-# 2025-01-17 16:14:29.090 E iOSTestbed[23987:1fd393b4] (Python) ...
+# 2025-01-17 16:14:29.093742+0800 iOSTestbed[23987:1fd393b4] ...
+# 2025-01-17 16:14:29.093742+0800 iOSTestbed[23987:1fd393b4] ...
LOG_PREFIX_REGEX = re.compile(
r"^\d{4}-\d{2}-\d{2}" # YYYY-MM-DD
- r"\s+\d+:\d{2}:\d{2}\.\d+" # HH:MM:SS.sss
- r"\s+\w+" # Df/E
+ r"\s+\d+:\d{2}:\d{2}\.\d+\+\d{4}" # HH:MM:SS.ssssss+ZZZZ
r"\s+iOSTestbed\[\d+:\w+\]" # Process/thread ID
- r"\s+\(Python\)\s" # Logger name
)
-# Work around a bug involving sys.exit and TaskGroups
-# (https://github.com/python/cpython/issues/101515).
-def exit(*args):
- raise MySystemExit(*args)
-
-
-class MySystemExit(Exception):
- pass
-
-
-class SimulatorLock:
- # An fcntl-based filesystem lock that can be used to ensure that
- def __init__(self, timeout):
- self.filename = Path(tempfile.gettempdir()) / "python-ios-testbed"
- self.timeout = timeout
-
- self.fd = None
-
- async def acquire(self):
- # Ensure the lockfile exists
- self.filename.touch(exist_ok=True)
-
- # Try `timeout` times to acquire the lock file, with a 1 second pause
- # between each attempt. Report status every 10 seconds.
- for i in range(0, self.timeout):
- try:
- fd = os.open(self.filename, os.O_RDWR | os.O_TRUNC, 0o644)
- fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
- except OSError:
- os.close(fd)
- if i % 10 == 0:
- print("... waiting", flush=True)
- await asyncio.sleep(1)
- else:
- self.fd = fd
- return
-
- # If we reach the end of the loop, we've exceeded the allowed number of
- # attempts.
- raise ValueError("Unable to obtain lock on iOS simulator creation")
-
- def release(self):
- # If a lock is held, release it.
- if self.fd is not None:
- # Release the lock.
- fcntl.flock(self.fd, fcntl.LOCK_UN)
- os.close(self.fd)
- self.fd = None
-
-
-# All subprocesses are executed through this context manager so that no matter
-# what happens, they can always be cancelled from another task, and they will
-# always be cleaned up on exit.
-@asynccontextmanager
-async def async_process(*args, **kwargs):
- process = await asyncio.create_subprocess_exec(*args, **kwargs)
- try:
- yield process
- finally:
- if process.returncode is None:
- # Allow a reasonably long time for Xcode to clean itself up,
- # because we don't want stale emulators left behind.
- timeout = 10
- process.terminate()
- try:
- await asyncio.wait_for(process.wait(), timeout)
- except TimeoutError:
- print(
- f"Command {args} did not terminate after {timeout} seconds "
- f" - sending SIGKILL"
- )
- process.kill()
-
- # Even after killing the process we must still wait for it,
- # otherwise we'll get the warning "Exception ignored in __del__".
- await asyncio.wait_for(process.wait(), timeout=1)
-
-
-async def async_check_output(*args, **kwargs):
- async with async_process(
- *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kwargs
- ) as process:
- stdout, stderr = await process.communicate()
- if process.returncode == 0:
- return stdout.decode(*DECODE_ARGS)
- else:
- raise subprocess.CalledProcessError(
- process.returncode,
- args,
- stdout.decode(*DECODE_ARGS),
- stderr.decode(*DECODE_ARGS),
- )
-
-
# Select a simulator device to use.
-async def select_simulator_device():
+def select_simulator_device():
# List the testing simulators, in JSON format
- raw_json = await async_check_output(
- "xcrun", "simctl", "list", "-j"
- )
+ raw_json = subprocess.check_output(["xcrun", "simctl", "list", "-j"])
json_data = json.loads(raw_json)
# Any device will do; we'll look for "SE" devices - but the name isn't
for devicetype in json_data["devicetypes"]
if devicetype["productFamily"] == "iPhone"
and (
- ("iPhone " in devicetype["name"] and devicetype["name"].endswith("e"))
+ (
+ "iPhone " in devicetype["name"]
+ and devicetype["name"].endswith("e")
+ )
or "iPhone SE " in devicetype["name"]
)
)
return se_simulators[-1][1]
-# Return a list of UDIDs associated with booted simulators
-async def list_devices():
- try:
- # List the testing simulators, in JSON format
- raw_json = await async_check_output(
- "xcrun", "simctl", "--set", "testing", "list", "-j"
- )
- json_data = json.loads(raw_json)
-
- # Filter out the booted iOS simulators
- return [
- simulator["udid"]
- for runtime, simulators in json_data["devices"].items()
- for simulator in simulators
- if runtime.split(".")[-1].startswith("iOS") and simulator["state"] == "Booted"
- ]
- except subprocess.CalledProcessError as e:
- # If there's no ~/Library/Developer/XCTestDevices folder (which is the
- # case on fresh installs, and in some CI environments), `simctl list`
- # returns error code 1, rather than an empty list. Handle that case,
- # but raise all other errors.
- if e.returncode == 1:
- return []
- else:
- raise
-
-
-async def find_device(initial_devices, lock):
- while True:
- new_devices = set(await list_devices()).difference(initial_devices)
- if len(new_devices) == 0:
- await asyncio.sleep(1)
- elif len(new_devices) == 1:
- udid = new_devices.pop()
- print(f"{datetime.now():%Y-%m-%d %H:%M:%S}: New test simulator detected")
- print(f"UDID: {udid}", flush=True)
- lock.release()
- return udid
- else:
- exit(f"Found more than one new device: {new_devices}")
-
-
-async def log_stream_task(initial_devices, lock):
- # Wait up to 5 minutes for the build to complete and the simulator to boot.
- udid = await asyncio.wait_for(find_device(initial_devices, lock), 5 * 60)
-
- # Stream the iOS device's logs, filtering out messages that come from the
- # XCTest test suite (catching NSLog messages from the test method), or
- # Python itself (catching stdout/stderr content routed to the system log
- # with config->use_system_logger).
+def xcode_test(location, simulator, verbose):
+ # Build and run the test suite on the named simulator.
args = [
- "xcrun",
- "simctl",
- "--set",
- "testing",
- "spawn",
- udid,
- "log",
- "stream",
- "--style",
- "compact",
- "--predicate",
- (
- 'senderImagePath ENDSWITH "/iOSTestbedTests.xctest/iOSTestbedTests"'
- ' OR senderImagePath ENDSWITH "/Python.framework/Python"'
- ),
- ]
-
- async with async_process(
- *args,
- stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT,
- ) as process:
- suppress_dupes = False
- while line := (await process.stdout.readline()).decode(*DECODE_ARGS):
- # Strip the prefix from each log line
- line = LOG_PREFIX_REGEX.sub("", line)
- # The iOS log streamer can sometimes lag; when it does, it outputs
- # a warning about messages being dropped... often multiple times.
- # Only print the first of these duplicated warnings.
- if line.startswith("=== Messages dropped "):
- if not suppress_dupes:
- suppress_dupes = True
- sys.stdout.write(line)
- else:
- suppress_dupes = False
- sys.stdout.write(line)
- sys.stdout.flush()
-
-
-async def xcode_test(location, simulator, verbose):
- # Run the test suite on the named simulator
- print("Starting xcodebuild...", flush=True)
- args = [
- "xcodebuild",
- "test",
"-project",
str(location / "iOSTestbed.xcodeproj"),
"-scheme",
"iOSTestbed",
"-destination",
f"platform=iOS Simulator,name={simulator}",
- "-resultBundlePath",
- str(location / f"{datetime.now():%Y%m%d-%H%M%S}.xcresult"),
"-derivedDataPath",
str(location / "DerivedData"),
]
- if not verbose:
- args += ["-quiet"]
+ verbosity_args = [] if verbose else ["-quiet"]
+
+ print("Building test project...")
+ subprocess.run(
+ ["xcodebuild", "build-for-testing"] + args + verbosity_args,
+ check=True,
+ )
- async with async_process(
- *args,
+ print("Running test project...")
+ # Test execution *can't* be run -quiet; verbose mode
+ # is how we see the output of the test output.
+ process = subprocess.Popen(
+ ["xcodebuild", "test-without-building"] + args,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
- ) as process:
- while line := (await process.stdout.readline()).decode(*DECODE_ARGS):
- sys.stdout.write(line)
- sys.stdout.flush()
+ )
+ while line := (process.stdout.readline()).decode(*DECODE_ARGS):
+ # Strip the timestamp/process prefix from each log line
+ line = LOG_PREFIX_REGEX.sub("", line)
+ sys.stdout.write(line)
+ sys.stdout.flush()
- status = await asyncio.wait_for(process.wait(), timeout=1)
- exit(status)
+ status = process.wait(timeout=5)
+ exit(status)
def clone_testbed(
sys.exit(13)
print("Cloning testbed project:")
- print(f" Cloning {source}...", end="", flush=True)
+ print(f" Cloning {source}...", end="")
shutil.copytree(source, target, symlinks=True)
print(" done")
sim_framework_path = xc_framework_path / "ios-arm64_x86_64-simulator"
if framework is not None:
if framework.suffix == ".xcframework":
- print(" Installing XCFramework...", end="", flush=True)
+ print(" Installing XCFramework...", end="")
if xc_framework_path.is_dir():
shutil.rmtree(xc_framework_path)
else:
)
print(" done")
else:
- print(" Installing simulator framework...", end="", flush=True)
+ print(" Installing simulator framework...", end="")
if sim_framework_path.is_dir():
shutil.rmtree(sim_framework_path)
else:
):
# XCFramework is a relative symlink. Rewrite the symlink relative
# to the new location.
- print(" Rewriting symlink to XCframework...", end="", flush=True)
+ print(" Rewriting symlink to XCframework...", end="")
orig_xc_framework_path = (
- source
- / xc_framework_path.readlink()
+ source / xc_framework_path.readlink()
).resolve()
xc_framework_path.unlink()
xc_framework_path.symlink_to(
sim_framework_path.is_symlink()
and not sim_framework_path.readlink().is_absolute()
):
- print(" Rewriting symlink to simulator framework...", end="", flush=True)
+ print(" Rewriting symlink to simulator framework...", end="")
# Simulator framework is a relative symlink. Rewrite the symlink
# relative to the new location.
orig_sim_framework_path = (
- source
- / "Python.XCframework"
- / sim_framework_path.readlink()
+ source / "Python.XCframework" / sim_framework_path.readlink()
).resolve()
sim_framework_path.unlink()
sim_framework_path.symlink_to(
print(" Using pre-existing iOS framework.")
for app_src in apps:
- print(f" Installing app {app_src.name!r}...", end="", flush=True)
+ print(f" Installing app {app_src.name!r}...", end="")
app_target = target / f"iOSTestbed/app/{app_src.name}"
if app_target.is_dir():
shutil.rmtree(app_target)
print(f"Successfully cloned testbed: {target.resolve()}")
-def update_plist(testbed_path, args):
- # Add the test runner arguments to the testbed's Info.plist file.
- info_plist = testbed_path / "iOSTestbed" / "iOSTestbed-Info.plist"
- with info_plist.open("rb") as f:
- info = plistlib.load(f)
+def update_test_plan(testbed_path, args):
+ # Modify the test plan to use the requested test arguments.
+ test_plan_path = testbed_path / "iOSTestbed.xctestplan"
+ with test_plan_path.open("r", encoding="utf-8") as f:
+ test_plan = json.load(f)
- info["TestArgs"] = args
+ test_plan["defaultOptions"]["commandLineArgumentEntries"] = [
+ {"argument": arg} for arg in args
+ ]
- with info_plist.open("wb") as f:
- plistlib.dump(info, f)
+ with test_plan_path.open("w", encoding="utf-8") as f:
+ json.dump(test_plan, f, indent=2)
-async def run_testbed(simulator: str | None, args: list[str], verbose: bool=False):
+def run_testbed(simulator: str | None, args: list[str], verbose: bool = False):
location = Path(__file__).parent
- print("Updating plist...", end="", flush=True)
- update_plist(location, args)
- print(" done.", flush=True)
+ print("Updating test plan...", end="")
+ update_test_plan(location, args)
+ print(" done.")
if simulator is None:
- simulator = await select_simulator_device()
- print(f"Running test on {simulator}", flush=True)
-
- # We need to get an exclusive lock on simulator creation, to avoid issues
- # with multiple simulators starting and being unable to tell which
- # simulator is due to which testbed instance. See
- # https://github.com/python/cpython/issues/130294 for details. Wait up to
- # 10 minutes for a simulator to boot.
- print("Obtaining lock on simulator creation...", flush=True)
- simulator_lock = SimulatorLock(timeout=10*60)
- await simulator_lock.acquire()
- print("Simulator lock acquired.", flush=True)
-
- # Get the list of devices that are booted at the start of the test run.
- # The simulator started by the test suite will be detected as the new
- # entry that appears on the device list.
- initial_devices = await list_devices()
+ simulator = select_simulator_device()
+ print(f"Running test on {simulator}")
- try:
- async with asyncio.TaskGroup() as tg:
- tg.create_task(log_stream_task(initial_devices, simulator_lock))
- tg.create_task(xcode_test(location, simulator=simulator, verbose=verbose))
- except* MySystemExit as e:
- raise SystemExit(*e.exceptions[0].args) from None
- except* subprocess.CalledProcessError as e:
- # Extract it from the ExceptionGroup so it can be handled by `main`.
- raise e.exceptions[0]
- finally:
- simulator_lock.release()
+ xcode_test(location, simulator=simulator, verbose=verbose)
def main():
run.add_argument(
"--simulator",
help=(
- "The name of the simulator to use (eg: 'iPhone 16e'). Defaults to ",
- "the most recently released 'entry level' iPhone device."
- )
+ "The name of the simulator to use (eg: 'iPhone 16e'). Defaults to "
+ "the most recently released 'entry level' iPhone device. Device "
+ "architecture and OS version can also be specified; e.g., "
+ "`--simulator 'iPhone 16 Pro,arch=arm64,OS=26.0'` would run on "
+ "an ARM64 iPhone 16 Pro simulator running iOS 26.0."
+ ),
)
run.add_argument(
- "-v", "--verbose",
+ "-v",
+ "--verbose",
action="store_true",
help="Enable verbose output",
)
clone_testbed(
source=Path(__file__).parent.resolve(),
target=Path(context.location).resolve(),
- framework=Path(context.framework).resolve() if context.framework else None,
+ framework=Path(context.framework).resolve()
+ if context.framework
+ else None,
apps=[Path(app) for app in context.apps],
)
elif context.subcommand == "run":
if test_args:
if not (
- Path(__file__).parent / "Python.xcframework/ios-arm64_x86_64-simulator/bin"
+ Path(__file__).parent
+ / "Python.xcframework/ios-arm64_x86_64-simulator/bin"
).is_dir():
print(
f"Testbed does not contain a compiled iOS framework. Use "
)
sys.exit(20)
- asyncio.run(
- run_testbed(
- simulator=context.simulator,
- verbose=context.verbose,
- args=test_args,
- )
+ run_testbed(
+ simulator=context.simulator,
+ verbose=context.verbose,
+ args=test_args,
)
else:
- print(f"Must specify test arguments (e.g., {sys.argv[0]} run -- test)")
+ print(
+ f"Must specify test arguments (e.g., {sys.argv[0]} run -- test)"
+ )
print()
parser.print_help(sys.stderr)
sys.exit(21)
projects / thirdparty / Python / cpython.git / commitdiff
