1# Copyright 2022 The Chromium Authors 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4"""Functions used in both v1 and v2 scripts.""" 5 6import json 7import os 8import platform 9import stat 10 11from typing import Iterable, List, Tuple 12 13 14_FILTER_DIR = 'testing/buildbot/filters' 15_SSH_KEYS = os.path.expanduser('~/.ssh/fuchsia_authorized_keys') 16_CHROME_HEADLESS = 'CHROME_HEADLESS' 17_SWARMING_SERVER = 'SWARMING_SERVER' 18 19class VersionNotFoundError(Exception): 20 """Thrown when version info cannot be retrieved from device.""" 21 22 23def get_ssh_keys() -> str: 24 """Returns path of Fuchsia ssh keys.""" 25 26 return _SSH_KEYS 27 28 29def running_unattended() -> bool: 30 """Returns true if running non-interactively. 31 32 When running unattended, confirmation prompts and the like are suppressed. 33 """ 34 35 # TODO(crbug.com/40884247): Change to mixin based approach. 36 # And remove SWARMING_SERVER check when it's no longer needed by dart, 37 # eureka and flutter to partially revert https://crrev.com/c/4112522. 38 return _CHROME_HEADLESS in os.environ or _SWARMING_SERVER in os.environ 39 40 41def force_running_unattended() -> None: 42 """Treats everything as running non-interactively.""" 43 if not running_unattended(): 44 os.environ[_CHROME_HEADLESS] = '1' 45 assert running_unattended() 46 47 48def force_running_attended() -> None: 49 """Treats everything as running interactively.""" 50 if running_unattended(): 51 os.environ.pop(_CHROME_HEADLESS, None) 52 os.environ.pop(_SWARMING_SERVER, None) 53 assert not running_unattended() 54 55 56def get_host_arch() -> str: 57 """Retrieve CPU architecture of the host machine. """ 58 host_arch = platform.machine() 59 # platform.machine() returns AMD64 on 64-bit Windows. 60 if host_arch in ['x86_64', 'AMD64']: 61 return 'x64' 62 if host_arch in ['aarch64', 'arm64']: 63 return 'arm64' 64 raise NotImplementedError('Unsupported host architecture: %s' % host_arch) 65 66 67def add_exec_to_file(file: str) -> None: 68 """Add execution bits to a file. 69 70 Args: 71 file: path to the file. 72 """ 73 file_stat = os.stat(file) 74 os.chmod(file, file_stat.st_mode | stat.S_IXUSR) 75 76 77def get_ssh_prefix(host_port_pair: Tuple[str, int]) -> List[str]: 78 """Get the prefix of a barebone ssh command.""" 79 return [ 80 'ssh', '-F', 81 os.path.join(os.path.dirname(__file__), 'sshconfig'), 82 host_port_pair[0], '-p', 83 str(host_port_pair[1]) 84 ] 85 86 87def install_symbols(package_paths: Iterable[str], 88 fuchsia_out_dir: str) -> None: 89 """Installs debug symbols for a package into the GDB-standard symbol 90 directory located in fuchsia_out_dir.""" 91 92 symbol_root = os.path.join(fuchsia_out_dir, '.build-id') 93 for path in package_paths: 94 package_dir = os.path.dirname(path) 95 ids_txt_path = os.path.join(package_dir, 'ids.txt') 96 with open(ids_txt_path, 'r') as f: 97 for entry in f: 98 build_id, binary_relpath = entry.strip().split(' ') 99 binary_abspath = os.path.abspath( 100 os.path.join(package_dir, binary_relpath)) 101 symbol_dir = os.path.join(symbol_root, build_id[:2]) 102 symbol_file = os.path.join(symbol_dir, build_id[2:] + '.debug') 103 if not os.path.exists(symbol_dir): 104 os.makedirs(symbol_dir) 105 106 if os.path.islink(symbol_file) or os.path.exists(symbol_file): 107 # Clobber the existing entry to ensure that the symlink's 108 # target is up to date. 109 os.unlink(symbol_file) 110 os.symlink(os.path.relpath(binary_abspath, symbol_dir), 111 symbol_file) 112 113 114# TODO(crbug.com/42050403): Until one can send files to the device when running 115# a test, filter files must be read from the test package. 116def map_filter_file_to_package_file(filter_file: str) -> str: 117 """Returns the path to |filter_file| within the test component's package.""" 118 119 if not _FILTER_DIR in filter_file: 120 raise ValueError('CFv2 tests only support registered filter files ' 121 'present in the test package') 122 return '/pkg/' + filter_file[filter_file.index(_FILTER_DIR):] 123 124 125# TODO(crbug.com/40938340): Rename to get_product_version. 126def get_sdk_hash(system_image_dir: str) -> Tuple[str, str]: 127 """Read version of hash in pre-installed package directory. 128 Returns: 129 Tuple of (product, version) of image to be installed. 130 """ 131 132 with open(os.path.join(system_image_dir, 133 'product_bundle.json')) as product: 134 # The product_name in the json file does not match the name of the image 135 # flashed to the device. 136 return (os.path.basename(os.path.normpath(system_image_dir)), 137 json.load(product)['product_version']) 138