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 6import logging 7import os 8import json 9import subprocess 10import sys 11import tempfile 12 13from contextlib import AbstractContextManager 14from typing import IO, Iterable, List, Optional 15 16from common import run_continuous_ffx_command, run_ffx_command, SDK_ROOT 17 18RUN_SUMMARY_SCHEMA = \ 19 'https://fuchsia.dev/schema/ffx_test/run_summary-8d1dd964.json' 20 21 22def 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 35class 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 78def test_connection(target_id: Optional[str]) -> None: 79 """Run an echo test to verify that the device can be connected to.""" 80 81 run_ffx_command(cmd=('target', 'echo'), target_id=target_id) 82 83 84class FfxTestRunner(AbstractContextManager): 85 """A context manager that manages a session for running a test via `ffx`. 86 87 Upon entry, an instance of this class configures `ffx` to retrieve files 88 generated by a test and prepares a directory to hold these files either in a 89 specified directory or in tmp. On exit, any previous configuration of 90 `ffx` is restored and the temporary directory, if used, is deleted. 91 92 The prepared directory is used when invoking `ffx test run`. 93 """ 94 95 def __init__(self, results_dir: Optional[str] = None) -> None: 96 """ 97 Args: 98 results_dir: Directory on the host where results should be stored. 99 """ 100 self._results_dir = results_dir 101 self._custom_artifact_directory = None 102 self._temp_results_dir = None 103 self._debug_data_directory = None 104 105 def __enter__(self): 106 if self._results_dir: 107 os.makedirs(self._results_dir, exist_ok=True) 108 else: 109 self._temp_results_dir = tempfile.TemporaryDirectory() 110 self._results_dir = self._temp_results_dir.__enter__() 111 return self 112 113 def __exit__(self, exc_type, exc_val, exc_tb) -> bool: 114 if self._temp_results_dir: 115 self._temp_results_dir.__exit__(exc_type, exc_val, exc_tb) 116 self._temp_results_dir = None 117 118 # Do not suppress exceptions. 119 return False 120 121 def run_test(self, 122 component_uri: str, 123 test_args: Optional[Iterable[str]] = None, 124 node_name: Optional[str] = None) -> subprocess.Popen: 125 """Starts a subprocess to run a test on a target. 126 Args: 127 component_uri: The test component URI. 128 test_args: Arguments to the test package, if any. 129 node_name: The target on which to run the test. 130 Returns: 131 A subprocess.Popen object. 132 """ 133 command = [ 134 'test', 'run', '--output-directory', self._results_dir, 135 component_uri 136 ] 137 if test_args: 138 command.append('--') 139 command.extend(test_args) 140 return run_continuous_ffx_command(command, 141 node_name, 142 stdout=subprocess.PIPE, 143 stderr=subprocess.STDOUT) 144 145 def _parse_test_outputs(self): 146 """Parses the output files generated by the test runner. 147 148 The instance's `_custom_artifact_directory` member is set to the 149 directory holding output files emitted by the test. 150 151 This function is idempotent, and performs no work if it has already been 152 called. 153 """ 154 if self._custom_artifact_directory: 155 return 156 157 run_summary_path = os.path.join(self._results_dir, 'run_summary.json') 158 try: 159 with open(run_summary_path) as run_summary_file: 160 run_summary = json.load(run_summary_file) 161 except IOError: 162 logging.exception('Error reading run summary file.') 163 return 164 except ValueError: 165 logging.exception('Error parsing run summary file %s', 166 run_summary_path) 167 return 168 169 assert run_summary['schema_id'] == RUN_SUMMARY_SCHEMA, \ 170 'Unsupported version found in %s' % run_summary_path 171 172 run_artifact_dir = run_summary.get('data', {})['artifact_dir'] 173 for artifact_path, artifact in run_summary.get( 174 'data', {})['artifacts'].items(): 175 if artifact['artifact_type'] == 'DEBUG': 176 self._debug_data_directory = os.path.join( 177 self._results_dir, run_artifact_dir, artifact_path) 178 break 179 180 if run_summary['data']['outcome'] == "NOT_STARTED": 181 logging.critical('Test execution was interrupted. Either the ' 182 'emulator crashed while the tests were still ' 183 'running or connection to the device was lost.') 184 sys.exit(1) 185 186 # There should be precisely one suite for the test that ran. 187 suites_list = run_summary.get('data', {}).get('suites') 188 if not suites_list: 189 logging.error('Missing or empty list of suites in %s', 190 run_summary_path) 191 return 192 suite_summary = suites_list[0] 193 194 # Get the top-level directory holding all artifacts for this suite. 195 artifact_dir = suite_summary.get('artifact_dir') 196 if not artifact_dir: 197 logging.error('Failed to find suite\'s artifact_dir in %s', 198 run_summary_path) 199 return 200 201 # Get the path corresponding to artifacts 202 for artifact_path, artifact in suite_summary['artifacts'].items(): 203 if artifact['artifact_type'] == 'CUSTOM': 204 self._custom_artifact_directory = os.path.join( 205 self._results_dir, artifact_dir, artifact_path) 206 break 207 208 def get_custom_artifact_directory(self) -> str: 209 """Returns the full path to the directory holding custom artifacts 210 emitted by the test or None if the directory could not be discovered. 211 """ 212 self._parse_test_outputs() 213 return self._custom_artifact_directory 214 215 def get_debug_data_directory(self): 216 """Returns the full path to the directory holding debug data 217 emitted by the test, or None if the path cannot be determined. 218 """ 219 self._parse_test_outputs() 220 return self._debug_data_directory 221 222 223def run_symbolizer(symbol_paths: List[str], input_fd: IO, 224 output_fd: IO) -> 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 return run_continuous_ffx_command(symbolize_cmd, 234 stdin=input_fd, 235 stdout=output_fd, 236 stderr=subprocess.STDOUT) 237