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 17 18class VersionNotFoundError(Exception): 19 """Thrown when version info cannot be retrieved from device.""" 20 21 22def get_ssh_keys() -> str: 23 """Returns path of Fuchsia ssh keys.""" 24 25 return _SSH_KEYS 26 27 28def running_unattended() -> bool: 29 """Returns true if running non-interactively. 30 31 When running unattended, confirmation prompts and the like are suppressed. 32 """ 33 34 # TODO(crbug/1401387): Change to mixin based approach. 35 return 'SWARMING_SERVER' in os.environ 36 37 38def get_host_arch() -> str: 39 """Retrieve CPU architecture of the host machine. """ 40 host_arch = platform.machine() 41 # platform.machine() returns AMD64 on 64-bit Windows. 42 if host_arch in ['x86_64', 'AMD64']: 43 return 'x64' 44 if host_arch in ['aarch64', 'arm64']: 45 return 'arm64' 46 raise NotImplementedError('Unsupported host architecture: %s' % host_arch) 47 48 49def add_exec_to_file(file: str) -> None: 50 """Add execution bits to a file. 51 52 Args: 53 file: path to the file. 54 """ 55 file_stat = os.stat(file) 56 os.chmod(file, file_stat.st_mode | stat.S_IXUSR) 57 58 59def parse_host_port(host_port_pair: str) -> Tuple[str, int]: 60 """Parses a host name or IP address and a port number from a string of 61 any of the following forms: 62 - hostname:port 63 - IPv4addy:port 64 - [IPv6addy]:port 65 66 Returns: 67 A tuple of the string host name/address and integer port number. 68 69 Raises: 70 ValueError if `host_port_pair` does not contain a colon or if the 71 substring following the last colon cannot be converted to an int. 72 """ 73 74 host, port = host_port_pair.rsplit(':', 1) 75 76 # Strip the brackets if the host looks like an IPv6 address. 77 if len(host) >= 4 and host[0] == '[' and host[-1] == ']': 78 host = host[1:-1] 79 return (host, int(port)) 80 81 82def get_ssh_prefix(host_port_pair: str) -> List[str]: 83 """Get the prefix of a barebone ssh command.""" 84 85 ssh_addr, ssh_port = parse_host_port(host_port_pair) 86 sshconfig = os.path.join(os.path.dirname(__file__), 'sshconfig') 87 return ['ssh', '-F', sshconfig, ssh_addr, '-p', str(ssh_port)] 88 89 90def install_symbols(package_paths: Iterable[str], 91 fuchsia_out_dir: str) -> None: 92 """Installs debug symbols for a package into the GDB-standard symbol 93 directory located in fuchsia_out_dir.""" 94 95 symbol_root = os.path.join(fuchsia_out_dir, '.build-id') 96 for path in package_paths: 97 package_dir = os.path.dirname(path) 98 ids_txt_path = os.path.join(package_dir, 'ids.txt') 99 with open(ids_txt_path, 'r') as f: 100 for entry in f: 101 build_id, binary_relpath = entry.strip().split(' ') 102 binary_abspath = os.path.abspath( 103 os.path.join(package_dir, binary_relpath)) 104 symbol_dir = os.path.join(symbol_root, build_id[:2]) 105 symbol_file = os.path.join(symbol_dir, build_id[2:] + '.debug') 106 if not os.path.exists(symbol_dir): 107 os.makedirs(symbol_dir) 108 109 if os.path.islink(symbol_file) or os.path.exists(symbol_file): 110 # Clobber the existing entry to ensure that the symlink's 111 # target is up to date. 112 os.unlink(symbol_file) 113 os.symlink(os.path.relpath(binary_abspath, symbol_dir), 114 symbol_file) 115 116 117# TODO(crbug.com/1279803): Until one can send files to the device when running 118# a test, filter files must be read from the test package. 119def map_filter_file_to_package_file(filter_file: str) -> str: 120 """Returns the path to |filter_file| within the test component's package.""" 121 122 if not _FILTER_DIR in filter_file: 123 raise ValueError('CFv2 tests only support registered filter files ' 124 'present in the test package') 125 return '/pkg/' + filter_file[filter_file.index(_FILTER_DIR):] 126 127 128# TODO(crbug.com/1496426): Rename to get_product_version. 129def get_sdk_hash(system_image_dir: str) -> Tuple[str, str]: 130 """Read version of hash in pre-installed package directory. 131 Returns: 132 Tuple of (product, version) of image to be installed. 133 """ 134 135 with open(os.path.join(system_image_dir, 136 'product_bundle.json')) as product: 137 # The product_name in the json file does not match the name of the image 138 # flashed to the device. 139 return (os.path.basename(os.path.normpath(system_image_dir)), 140 json.load(product)['product_version']) 141