# Copyright 2022 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Provide helpers for running Fuchsia's `ffx`.""" import logging import os import json import subprocess import sys import tempfile from contextlib import AbstractContextManager from typing import IO, Iterable, List, Optional from common import run_continuous_ffx_command, run_ffx_command, SDK_ROOT RUN_SUMMARY_SCHEMA = \ 'https://fuchsia.dev/schema/ffx_test/run_summary-8d1dd964.json' def get_config(name: str) -> Optional[str]: """Run a ffx config get command to retrieve the config value.""" try: return run_ffx_command(cmd=['config', 'get', name], capture_output=True).stdout.strip() except subprocess.CalledProcessError as cpe: # A return code of 2 indicates no previous value set. if cpe.returncode == 2: return None raise class ScopedFfxConfig(AbstractContextManager): """Temporarily overrides `ffx` configuration. Restores the previous value upon exit.""" def __init__(self, name: str, value: str) -> None: """ Args: name: The name of the property to set. value: The value to associate with `name`. """ self._old_value = None self._new_value = value self._name = name def __enter__(self): """Override the configuration.""" # Cache the old value. self._old_value = get_config(self._name) if self._new_value != self._old_value: run_ffx_command(cmd=['config', 'set', self._name, self._new_value]) return self def __exit__(self, exc_type, exc_val, exc_tb) -> bool: if self._new_value == self._old_value: return False # Allow removal of config to fail. remove_cmd = run_ffx_command(cmd=['config', 'remove', self._name], check=False) if remove_cmd.returncode != 0: logging.warning('Error when removing ffx config %s', self._name) # Explicitly set the value back only if removing the new value doesn't # already restore the old value. if self._old_value is not None and \ self._old_value != get_config(self._name): run_ffx_command(cmd=['config', 'set', self._name, self._old_value]) # Do not suppress exceptions. return False class FfxTestRunner(AbstractContextManager): """A context manager that manages a session for running a test via `ffx`. Upon entry, an instance of this class configures `ffx` to retrieve files generated by a test and prepares a directory to hold these files either in a specified directory or in tmp. On exit, any previous configuration of `ffx` is restored and the temporary directory, if used, is deleted. The prepared directory is used when invoking `ffx test run`. """ def __init__(self, results_dir: Optional[str] = None) -> None: """ Args: results_dir: Directory on the host where results should be stored. """ self._results_dir = results_dir self._custom_artifact_directory = None self._temp_results_dir = None self._debug_data_directory = None def __enter__(self): if self._results_dir: os.makedirs(self._results_dir, exist_ok=True) else: self._temp_results_dir = tempfile.TemporaryDirectory() self._results_dir = self._temp_results_dir.__enter__() return self def __exit__(self, exc_type, exc_val, exc_tb) -> bool: if self._temp_results_dir: self._temp_results_dir.__exit__(exc_type, exc_val, exc_tb) self._temp_results_dir = None # Do not suppress exceptions. return False def run_test(self, component_uri: str, test_args: Optional[Iterable[str]] = None, node_name: Optional[str] = None, test_realm: Optional[str] = None) -> subprocess.Popen: """Starts a subprocess to run a test on a target. Args: component_uri: The test component URI. test_args: Arguments to the test package, if any. node_name: The target on which to run the test. Returns: A subprocess.Popen object. """ command = [ 'test', 'run', '--output-directory', self._results_dir, ] if test_realm: command.append("--realm") command.append(test_realm) command.append(component_uri) if test_args: command.append('--') command.extend(test_args) return run_continuous_ffx_command(command, node_name, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) def _parse_test_outputs(self): """Parses the output files generated by the test runner. The instance's `_custom_artifact_directory` member is set to the directory holding output files emitted by the test. This function is idempotent, and performs no work if it has already been called. """ if self._custom_artifact_directory: return run_summary_path = os.path.join(self._results_dir, 'run_summary.json') try: with open(run_summary_path) as run_summary_file: run_summary = json.load(run_summary_file) except IOError: logging.exception('Error reading run summary file.') return except ValueError: logging.exception('Error parsing run summary file %s', run_summary_path) return assert run_summary['schema_id'] == RUN_SUMMARY_SCHEMA, \ 'Unsupported version found in %s' % run_summary_path run_artifact_dir = run_summary.get('data', {})['artifact_dir'] for artifact_path, artifact in run_summary.get( 'data', {})['artifacts'].items(): if artifact['artifact_type'] == 'DEBUG': self._debug_data_directory = os.path.join( self._results_dir, run_artifact_dir, artifact_path) break if run_summary['data']['outcome'] == "NOT_STARTED": logging.critical('Test execution was interrupted. Either the ' 'emulator crashed while the tests were still ' 'running or connection to the device was lost.') sys.exit(1) # There should be precisely one suite for the test that ran. suites_list = run_summary.get('data', {}).get('suites') if not suites_list: logging.error('Missing or empty list of suites in %s', run_summary_path) return suite_summary = suites_list[0] # Get the top-level directory holding all artifacts for this suite. artifact_dir = suite_summary.get('artifact_dir') if not artifact_dir: logging.error('Failed to find suite\'s artifact_dir in %s', run_summary_path) return # Get the path corresponding to artifacts for artifact_path, artifact in suite_summary['artifacts'].items(): if artifact['artifact_type'] == 'CUSTOM': self._custom_artifact_directory = os.path.join( self._results_dir, artifact_dir, artifact_path) break def get_custom_artifact_directory(self) -> str: """Returns the full path to the directory holding custom artifacts emitted by the test or None if the directory could not be discovered. """ self._parse_test_outputs() return self._custom_artifact_directory def get_debug_data_directory(self): """Returns the full path to the directory holding debug data emitted by the test, or None if the path cannot be determined. """ self._parse_test_outputs() return self._debug_data_directory def run_symbolizer(symbol_paths: List[str], input_fd: IO, output_fd: IO, raw_bytes: bool = False) -> subprocess.Popen: """Runs symbolizer that symbolizes |input| and outputs to |output|.""" symbolize_cmd = ([ 'debug', 'symbolize', '--', '--omit-module-lines', '--build-id-dir', os.path.join(SDK_ROOT, '.build-id') ]) for path in symbol_paths: symbolize_cmd.extend(['--ids-txt', path]) if raw_bytes: encoding = None else: encoding = 'utf-8' return run_continuous_ffx_command(symbolize_cmd, stdin=input_fd, stdout=output_fd, stderr=subprocess.STDOUT, encoding=encoding, close_fds=True)