ffx_integration.py (9157B)
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 """Provide helpers for running Fuchsia's `ffx`.""" 5 6 import logging 7 import os 8 import json 9 import subprocess 10 import sys 11 import tempfile 12 13 from contextlib import AbstractContextManager 14 from typing import IO, Iterable, List, Optional 15 16 from common import run_continuous_ffx_command, run_ffx_command, SDK_ROOT 17 18 RUN_SUMMARY_SCHEMA = \ 19 'https://fuchsia.dev/schema/ffx_test/run_summary-8d1dd964.json' 20 21 22 def get_config(name: str) -> Optional[str]: 23 """Run a ffx config get command to retrieve the config value.""" 24 25 try: 26 return run_ffx_command(cmd=['config', 'get', name], 27 capture_output=True).stdout.strip() 28 except subprocess.CalledProcessError as cpe: 29 # A return code of 2 indicates no previous value set. 30 if cpe.returncode == 2: 31 return None 32 raise 33 34 35 class ScopedFfxConfig(AbstractContextManager): 36 """Temporarily overrides `ffx` configuration. Restores the previous value 37 upon exit.""" 38 39 def __init__(self, name: str, value: str) -> None: 40 """ 41 Args: 42 name: The name of the property to set. 43 value: The value to associate with `name`. 44 """ 45 self._old_value = None 46 self._new_value = value 47 self._name = name 48 49 def __enter__(self): 50 """Override the configuration.""" 51 52 # Cache the old value. 53 self._old_value = get_config(self._name) 54 if self._new_value != self._old_value: 55 run_ffx_command(cmd=['config', 'set', self._name, self._new_value]) 56 return self 57 58 def __exit__(self, exc_type, exc_val, exc_tb) -> bool: 59 if self._new_value == self._old_value: 60 return False 61 62 # Allow removal of config to fail. 63 remove_cmd = run_ffx_command(cmd=['config', 'remove', self._name], 64 check=False) 65 if remove_cmd.returncode != 0: 66 logging.warning('Error when removing ffx config %s', self._name) 67 68 # Explicitly set the value back only if removing the new value doesn't 69 # already restore the old value. 70 if self._old_value is not None and \ 71 self._old_value != get_config(self._name): 72 run_ffx_command(cmd=['config', 'set', self._name, self._old_value]) 73 74 # Do not suppress exceptions. 75 return False 76 77 78 class FfxTestRunner(AbstractContextManager): 79 """A context manager that manages a session for running a test via `ffx`. 80 81 Upon entry, an instance of this class configures `ffx` to retrieve files 82 generated by a test and prepares a directory to hold these files either in a 83 specified directory or in tmp. On exit, any previous configuration of 84 `ffx` is restored and the temporary directory, if used, is deleted. 85 86 The prepared directory is used when invoking `ffx test run`. 87 """ 88 89 def __init__(self, results_dir: Optional[str] = None) -> None: 90 """ 91 Args: 92 results_dir: Directory on the host where results should be stored. 93 """ 94 self._results_dir = results_dir 95 self._custom_artifact_directory = None 96 self._temp_results_dir = None 97 self._debug_data_directory = None 98 99 def __enter__(self): 100 if self._results_dir: 101 os.makedirs(self._results_dir, exist_ok=True) 102 else: 103 self._temp_results_dir = tempfile.TemporaryDirectory() 104 self._results_dir = self._temp_results_dir.__enter__() 105 return self 106 107 def __exit__(self, exc_type, exc_val, exc_tb) -> bool: 108 if self._temp_results_dir: 109 self._temp_results_dir.__exit__(exc_type, exc_val, exc_tb) 110 self._temp_results_dir = None 111 112 # Do not suppress exceptions. 113 return False 114 115 def run_test(self, 116 component_uri: str, 117 test_args: Optional[Iterable[str]] = None, 118 node_name: Optional[str] = None, 119 test_realm: Optional[str] = None) -> subprocess.Popen: 120 """Starts a subprocess to run a test on a target. 121 Args: 122 component_uri: The test component URI. 123 test_args: Arguments to the test package, if any. 124 node_name: The target on which to run the test. 125 Returns: 126 A subprocess.Popen object. 127 """ 128 command = [ 129 'test', 'run', '--output-directory', self._results_dir, 130 ] 131 if test_realm: 132 command.append("--realm") 133 command.append(test_realm) 134 command.append(component_uri) 135 if test_args: 136 command.append('--') 137 command.extend(test_args) 138 return run_continuous_ffx_command(command, 139 node_name, 140 stdout=subprocess.PIPE, 141 stderr=subprocess.STDOUT) 142 143 def _parse_test_outputs(self): 144 """Parses the output files generated by the test runner. 145 146 The instance's `_custom_artifact_directory` member is set to the 147 directory holding output files emitted by the test. 148 149 This function is idempotent, and performs no work if it has already been 150 called. 151 """ 152 if self._custom_artifact_directory: 153 return 154 155 run_summary_path = os.path.join(self._results_dir, 'run_summary.json') 156 try: 157 with open(run_summary_path) as run_summary_file: 158 run_summary = json.load(run_summary_file) 159 except IOError: 160 logging.exception('Error reading run summary file.') 161 return 162 except ValueError: 163 logging.exception('Error parsing run summary file %s', 164 run_summary_path) 165 return 166 167 assert run_summary['schema_id'] == RUN_SUMMARY_SCHEMA, \ 168 'Unsupported version found in %s' % run_summary_path 169 170 run_artifact_dir = run_summary.get('data', {})['artifact_dir'] 171 for artifact_path, artifact in run_summary.get( 172 'data', {})['artifacts'].items(): 173 if artifact['artifact_type'] == 'DEBUG': 174 self._debug_data_directory = os.path.join( 175 self._results_dir, run_artifact_dir, artifact_path) 176 break 177 178 if run_summary['data']['outcome'] == "NOT_STARTED": 179 logging.critical('Test execution was interrupted. Either the ' 180 'emulator crashed while the tests were still ' 181 'running or connection to the device was lost.') 182 sys.exit(1) 183 184 # There should be precisely one suite for the test that ran. 185 suites_list = run_summary.get('data', {}).get('suites') 186 if not suites_list: 187 logging.error('Missing or empty list of suites in %s', 188 run_summary_path) 189 return 190 suite_summary = suites_list[0] 191 192 # Get the top-level directory holding all artifacts for this suite. 193 artifact_dir = suite_summary.get('artifact_dir') 194 if not artifact_dir: 195 logging.error('Failed to find suite\'s artifact_dir in %s', 196 run_summary_path) 197 return 198 199 # Get the path corresponding to artifacts 200 for artifact_path, artifact in suite_summary['artifacts'].items(): 201 if artifact['artifact_type'] == 'CUSTOM': 202 self._custom_artifact_directory = os.path.join( 203 self._results_dir, artifact_dir, artifact_path) 204 break 205 206 def get_custom_artifact_directory(self) -> str: 207 """Returns the full path to the directory holding custom artifacts 208 emitted by the test or None if the directory could not be discovered. 209 """ 210 self._parse_test_outputs() 211 return self._custom_artifact_directory 212 213 def get_debug_data_directory(self): 214 """Returns the full path to the directory holding debug data 215 emitted by the test, or None if the path cannot be determined. 216 """ 217 self._parse_test_outputs() 218 return self._debug_data_directory 219 220 221 def run_symbolizer(symbol_paths: List[str], 222 input_fd: IO, 223 output_fd: IO, 224 raw_bytes: bool = False) -> subprocess.Popen: 225 """Runs symbolizer that symbolizes |input| and outputs to |output|.""" 226 227 symbolize_cmd = ([ 228 'debug', 'symbolize', '--', '--omit-module-lines', '--build-id-dir', 229 os.path.join(SDK_ROOT, '.build-id') 230 ]) 231 for path in symbol_paths: 232 symbolize_cmd.extend(['--ids-txt', path]) 233 if raw_bytes: 234 encoding = None 235 else: 236 encoding = 'utf-8' 237 return run_continuous_ffx_command(symbolize_cmd, 238 stdin=input_fd, 239 stdout=output_fd, 240 stderr=subprocess.STDOUT, 241 encoding=encoding, 242 close_fds=True)