#!/usr/bin/env python3 # Copyright 2025 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Shared utility functions for iOS test runners.""" import uuid import dataclasses import glob import json import os import re import subprocess import threading from typing import Any, Dict, List, Optional, Tuple import shlex def _load_env() -> None: """Loads environment variables from a .env file if it exists.""" tools_dir = os.path.dirname(os.path.abspath(__file__)) repo_root = os.path.abspath(os.path.join(tools_dir, '..', '..')) paths_to_try = [ os.path.join(os.getcwd(), '.env'), os.path.join(repo_root, '.env'), ] loaded = set() for path in paths_to_try: path = os.path.abspath(path) if path in loaded or not os.path.exists(path): continue loaded.add(path) try: with open(path, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue if '=' not in line: continue key, val = line.split('=', 1) key = key.strip() val = val.strip() if ((val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'"))): val = val[1:-1] if key not in os.environ: os.environ[key] = val except Exception as e: print(f"Warning: Failed to read .env file at {path}: {e}") # Automatically load environment variables from .env files on import. _load_env() @dataclasses.dataclass class Simulator: """Dataclass for simulator information.""" name: str udid: str os_version: str state: str is_available: bool @property def booted(self) -> bool: """Returns True if the simulator is in the 'Booted' state.""" return self.state == 'Booted' @property def parsed_os_version(self) -> Tuple[int, ...]: """Returns the OS version string as a tuple of integers.""" return tuple(map(int, self.os_version.split('.'))) @property def display_string(self) -> str: """Returns a formatted string for displaying simulator details.""" booted_status = (f"({Colors.GREEN}Booted{Colors.RESET})" if self.booted else "") return (f"{self.name} (OS: {self.os_version}, UDID: {self.udid}) " f"{booted_status}") class Colors: """ANSI color codes for terminal output.""" HEADER = '\033[35m' # Magenta BLUE = '\033[34m' CYAN = '\033[36m' GREEN = '\033[32m' WARNING = '\033[33m' # Yellow FAIL = '\033[31m' # Red BOLD = '\033[1m' RESET = '\033[0m' def print_command(command_list: List[str]) -> None: """Prints a shell command, quoting arguments for safe copy-pasting.""" quoted_command = ' '.join(shlex.quote(arg) for arg in command_list) print(f'{quoted_command}\n') def print_header(header_text: str) -> None: """Prints a formatted header string with special colors.""" print(f'\n{Colors.HEADER}{Colors.BOLD}{header_text}{Colors.RESET}') class SimulatorManager: """Manages and queries the list of available simulators.""" def __init__(self): """Initializes the manager, fetching and parsing the simulator list.""" self.simulators: List[Simulator] = [] try: output = subprocess.check_output( ['xcrun', 'simctl', 'list', 'devices', '--json'], encoding='utf-8') all_devices = json.loads(output).get('devices', {}) for runtime, devices in all_devices.items(): # Only process iOS runtimes. if 'iOS' not in runtime: continue os_version_full = runtime.split('iOS-')[-1].replace('-', '.') # Use only major.minor version for xcodebuild compatibility. os_version = '.'.join(os_version_full.split('.')[:2]) for device in devices: self.simulators.append( Simulator(name=device.get('name', ''), udid=device.get('udid', ''), os_version=os_version, state=device.get('state', ''), is_available=device.get( 'isAvailable', False))) except (subprocess.CalledProcessError, json.JSONDecodeError) as e: print(f"Error fetching simulator list: {e}") def list_available_simulators(self): """Prints a formatted list of available simulators.""" print("\nAvailable Simulators:") for sim in self.simulators: if sim.is_available: print(f" - {sim.display_string}") def find_specific_device(self, identifier: str) -> Optional[Simulator]: """Finds a specific device by name or UDID.""" for sim in self.simulators: if sim.name.lower() == identifier.lower( ) or sim.udid == identifier: return sim return None def find_best_available_device(self) -> Optional[Simulator]: """Finds the best available iPhone simulator to use as a default.""" best_candidate = None best_sdk_version = (0, ) best_iphone_version = 0 for sim in self.simulators: if not sim.is_available: continue current_sdk_version = sim.parsed_os_version match = re.match(r'^iPhone (\d+)', sim.name) if match: iphone_version = int(match.group(1)) if current_sdk_version > best_sdk_version: best_sdk_version = current_sdk_version best_iphone_version = iphone_version best_candidate = sim elif current_sdk_version == best_sdk_version: if iphone_version > best_iphone_version: best_iphone_version = iphone_version best_candidate = sim return best_candidate def find_device_by_type_and_version( self, device_type: str, os_version: Optional[str] = None) -> Optional[Simulator]: """Finds a device that matches a specific type and OS version.""" best_candidate = None best_sdk_version = (0, ) for sim in self.simulators: if not sim.is_available or sim.name.lower() != device_type.lower(): continue if os_version: # Truncate input os_version to major.minor for comparison. os_version_truncated = '.'.join(os_version.split('.')[:2]) if sim.os_version == os_version_truncated: return sim else: current_sdk_version = sim.parsed_os_version if current_sdk_version > best_sdk_version: best_sdk_version = current_sdk_version best_candidate = sim return best_candidate def find_and_boot_simulator(device_type: Optional[str], os_version: Optional[str]) -> Optional[Simulator]: """Finds the requested simulator, booting it if necessary. Args: device_type: The device type to look for (e.g., 'iPhone 15 Pro'). os_version: The OS version to look for (e.g., '17.5'). Returns: A Simulator object for the selected device, or None on failure. """ print_header("--- Selecting Simulator ---") env_device = os.environ.get('IOS_SIMULATOR_DEFAULT_DEVICE') env_os = os.environ.get('IOS_SIMULATOR_DEFAULT_OS') if not device_type and env_device: device_type = env_device print(f"Using default simulator device from environment: {device_type}") if not os_version and env_os: os_version = env_os print(f"Using default simulator OS from environment: {os_version}") sim_manager = SimulatorManager() if not sim_manager.simulators: return None simulator_to_use = None # 1. If a specific device is requested, try to find it. if device_type: is_udid = False try: uuid.UUID(device_type) is_udid = True except ValueError: pass if is_udid: for sim in sim_manager.simulators: if sim.udid.lower() == device_type.lower(): simulator_to_use = sim break if not simulator_to_use: print(f"Could not find a simulator with UDID '{device_type}'.") return None else: simulator_to_use = sim_manager.find_device_by_type_and_version( device_type, os_version) if not simulator_to_use: print(f"Could not find a simulator for device '{device_type}' " f"and OS '{os_version}'.") sim_manager.list_available_simulators() return None # 2. If no device was specified, look for an already booted one. if not simulator_to_use: for sim in sim_manager.simulators: if sim.booted: simulator_to_use = sim break # 3. If still no device, find the best available default. if not simulator_to_use: print(f"{Colors.BLUE}No simulator booted. " f"Finding the newest available iPhone...{Colors.RESET}") simulator_to_use = sim_manager.find_best_available_device() if not simulator_to_use: print("Could not find a suitable default iPhone simulator.") sim_manager.list_available_simulators() return None print( f"{Colors.BLUE}Device: {simulator_to_use.display_string}{Colors.RESET}" ) # Boot the selected device if it's not already running. if not simulator_to_use.booted: print( f"\n{Colors.CYAN}Simulator '{simulator_to_use.display_string}' is " f"not booted. Booting...{Colors.RESET}") try: subprocess.check_call( ['xcrun', 'simctl', 'boot', simulator_to_use.udid]) except subprocess.CalledProcessError as e: print(f"Error booting simulator: {e}") return None return simulator_to_use class SimulatorLogStreamer: """Streams logs from the simulator in the background.""" def __init__(self, simulator_udid: str, out_dir: str, scheme: str, predicate: str, show_in_stdout: bool = False): self.simulator_udid = simulator_udid self.out_dir = out_dir self.scheme = scheme self.predicate = predicate self.show_in_stdout = show_in_stdout self.proc = None self.safe_scheme = re.sub(r'[^a-zA-Z0-9_]', '_', scheme) self.log_file_path = os.path.join( out_dir, f'app_logs_{self.safe_scheme}_{os.getpid()}.log') self.thread = None self.log_file = None def start(self): os.makedirs(self.out_dir, exist_ok=True) # Keep only the 5 most recent log files for this scheme to limit # disk usage. try: pattern = os.path.join(self.out_dir, f'app_logs_{self.safe_scheme}_*.log') existing_logs = glob.glob(pattern) if len(existing_logs) >= 5: existing_logs.sort(key=os.path.getmtime) for old_log in existing_logs[:-4]: os.remove(old_log) except Exception: pass cmd = [ 'xcrun', 'simctl', 'spawn', self.simulator_udid, 'log', 'stream', '--level', 'debug', '--style', 'syslog', '--predicate', self.predicate ] self.log_file = open(self.log_file_path, 'w', encoding='utf-8', errors='replace') self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding='utf-8', errors='replace') def read_loop(): for line in self.proc.stdout: self.log_file.write(line) self.log_file.flush() if self.show_in_stdout: print(f"{Colors.CYAN}[AppLog]{Colors.RESET} {line.strip()}") self.thread = threading.Thread(target=read_loop, daemon=True) self.thread.start() print( f"{Colors.BLUE}Streaming app logs to: " f"{self.log_file_path}{Colors.RESET}" ) def stop(self): if self.proc: self.proc.terminate() try: self.proc.wait(timeout=2) except subprocess.TimeoutExpired: self.proc.kill() self.proc.wait() if self.thread: self.thread.join() if self.log_file: self.log_file.close()