#!/usr/bin/env vpython3 # Copyright 2022 The Chromium Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """A helper tool for running Fuchsia's `ffx`. """ # Enable use of the print() built-in function. from __future__ import print_function import argparse import contextlib import errno import json import logging import os import re import shutil import subprocess import sys import tempfile import common import log_manager sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), 'test'))) from compatible_utils import parse_host_port def get_ffx_path(): """Returns the full path to `ffx`.""" return os.path.join(common.SDK_ROOT, 'tools', common.GetHostArchFromPlatform(), 'ffx') def format_host_port(host, port): """Formats a host name or IP address and port number into a host:port string. """ # Wrap `host` in brackets if it looks like an IPv6 address return ('[%s]:%d' if ':' in host else '%s:%d') % (host, port) class FfxRunner(): """A helper to run `ffx` commands.""" def __init__(self, log_manager): self._ffx = get_ffx_path() self._log_manager = log_manager def _run_repair_command(self, output): """Scans `output` for a self-repair command to run and, if found, runs it. If logging is enabled, `ffx` is asked to emit its own logs to the log directory. Returns: True if a repair command was found and ran successfully. False otherwise. """ # Check for a string along the lines of: # "Run `ffx doctor --restart-daemon` for further diagnostics." match = re.search('`ffx ([^`]+)`', output) if not match or len(match.groups()) != 1: return False # No repair command found. args = match.groups()[0].split() # Tell ffx to include the configuration file without prompting in case # logging is enabled. with self.scoped_config('doctor.record_config', 'true'): # If the repair command is `ffx doctor` and logging is enabled, add the # options to emit ffx logs to the logging directory. if len(args) and args[0] == 'doctor' and \ self._log_manager.IsLoggingEnabled(): args.extend( ('--record', '--output-dir', self._log_manager.GetLogDirectory())) try: self.run_ffx(args, suppress_repair=True) except subprocess.CalledProcessError as cpe: return False # Repair failed. return True # Repair succeeded. def run_ffx(self, args, check=True, suppress_repair=False): """Runs `ffx` with the given arguments, waiting for it to exit. If `ffx` exits with a non-zero exit code, the output is scanned for a recommended repair command (e.g., "Run `ffx doctor --restart-daemon` for further diagnostics."). If such a command is found, it is run and then the original command is retried. This behavior can be suppressed via the `suppress_repair` argument. Args: args: A sequence of arguments to ffx. check: If True, CalledProcessError is raised if ffx returns a non-zero exit code. suppress_repair: If True, do not attempt to find and run a repair command. Returns: A string containing combined stdout and stderr. Raises: CalledProcessError if `check` is true. """ log_file = self._log_manager.Open('ffx_log') \ if self._log_manager.IsLoggingEnabled() else None command = [self._ffx] command.extend(args) logging.debug(command) if log_file: print(command, file=log_file) repair_succeeded = False try: # TODO(grt): Switch to subprocess.run() with encoding='utf-8' when p3 is # supported. process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout_data, stderr_data = process.communicate() stdout_data = stdout_data.decode('utf-8') stderr_data = stderr_data.decode('utf-8') if check and process.returncode != 0: # TODO(grt): Pass stdout and stderr as two args when p2 support is no # longer needed. raise subprocess.CalledProcessError( process.returncode, command, '\n'.join((stdout_data, stderr_data))) except subprocess.CalledProcessError as cpe: if log_file: log_file.write('Process exited with code %d. Output: %s\n' % (cpe.returncode, cpe.output.strip())) # Let the exception fly unless a repair command is found and succeeds. if suppress_repair or not self._run_repair_command(cpe.output): raise repair_succeeded = True # If the original command failed but a repair command was found and # succeeded, try one more time with the original command. if repair_succeeded: return self.run_ffx(args, check, suppress_repair=True) stripped_stdout = stdout_data.strip() stripped_stderr = stderr_data.strip() if log_file: if process.returncode != 0 or stripped_stderr: log_file.write('Process exited with code %d.' % process.returncode) if stripped_stderr: log_file.write(' Stderr:\n%s\n' % stripped_stderr) if stripped_stdout: log_file.write(' Stdout:\n%s\n' % stripped_stdout) if not stripped_stderr and not stripped_stdout: log_file.write('\n') elif stripped_stdout: log_file.write('%s\n' % stripped_stdout) logging.debug( 'ffx command returned %d with %s%s', process.returncode, ('output "%s"' % stripped_stdout if stripped_stdout else 'no output'), (' and error "%s".' % stripped_stderr if stripped_stderr else '.')) return stdout_data def open_ffx(self, args): """Runs `ffx` with some arguments. Args: args: A sequence of arguments to ffx. Returns: A subprocess.Popen object. """ log_file = self._log_manager.Open('ffx_log') \ if self._log_manager.IsLoggingEnabled() else None command = [self._ffx] command.extend(args) logging.debug(command) if log_file: print(command, file=log_file) try: # TODO(grt): Add encoding='utf-8' when p3 is supported. return subprocess.Popen(command, stdin=open(os.devnull, 'r'), stdout=subprocess.PIPE, stderr=subprocess.STDOUT) except: logging.exception('Failed to open ffx') if log_file: print('Exception caught while opening ffx: %s' % str(sys.exc_info[1])) raise @contextlib.contextmanager def scoped_config(self, name, value): """Temporarily overrides `ffx` configuration. Args: name: The name of the property to set. value: The value to associate with `name`. Returns: Yields nothing. Restores the previous value upon exit. """ assert value is not None # Cache the current value. old_value = None try: old_value = self.run_ffx(['config', 'get', name]).strip() except subprocess.CalledProcessError as cpe: if cpe.returncode != 2: raise # The failure was for something other than value not found. # Set the new value if it is different. if value != old_value: self.run_ffx(['config', 'set', name, value]) try: yield None finally: if value == old_value: return # There is no need to restore an old value. # Clear the new value. self.run_ffx(['config', 'remove', name]) if old_value is None: return # Did removing the new value restore the original value on account of it # either being the default or being set in a different scope? if (self.run_ffx(['config', 'get', name], check=False).strip() == old_value): return # If not, explicitly set the original value. self.run_ffx(['config', 'set', name, old_value]) def list_targets(self): """Returns the (possibly empty) list of targets known to ffx. Returns: The list of targets parsed from the JSON output of `ffx target list`. """ json_targets = self.run_ffx(['target', 'list', '-f', 'json']) if not json_targets: return [] try: return json.loads(json_targets) except ValueError: # TODO(grt): Change to json.JSONDecodeError once p3 is supported. return [] def list_active_targets(self): """Gets the list of targets and filters down to the targets that are active. Returns: An iterator over active FfxTargets. """ targets = [ FfxTarget.from_target_list_json(self, json_target) for json_target in self.list_targets() ] return filter(lambda target: target.get_ssh_address(), targets) def remove_stale_targets(self, address): """Removes any targets from ffx that are listening at a given address. Args: address: A string representation of the target's ip address. """ for target in self.list_targets(): if target['rcs_state'] == 'N' and address in target['addresses']: self.run_ffx(['target', 'remove', address]) @contextlib.contextmanager def scoped_target_context(self, address, port): """Temporarily adds a new target. Args: address: The IP address at which the target is listening. port: The port number on which the target is listening. Yields: An FfxTarget for interacting with the target. """ target_id = format_host_port(address, port) # -n allows `target add` to skip waiting for the device to come up, # as this can take longer than the default wait period. self.run_ffx(['target', 'add', '-n', target_id]) try: yield FfxTarget.from_address(self, address, port) finally: self.run_ffx(['target', 'remove', target_id], check=False) def get_node_name(self, address, port): """Returns the node name for a target given its SSH address. Args: address: The address at which the target's SSH daemon is listening. port: The port number on which the daemon is listening. Returns: The target's node name. Raises: Exception: If the target cannot be found. """ for target in self.list_targets(): if target['nodename'] and address in target['addresses']: ssh_address = FfxTarget.from_target_list_json(target).get_ssh_address() if ssh_address and ssh_address[1] == port: return target['nodename'] raise Exception('Failed to determine node name for target at %s' % format_host_port(address, port)) def daemon_stop(self): """Stops the ffx daemon.""" self.run_ffx(['daemon', 'stop'], check=False, suppress_repair=True) class FfxTarget(): """A helper to run `ffx` commands for a specific target.""" @classmethod def from_address(cls, ffx_runner, address, port=None): """Args: ffx_runner: The runner to use to run ffx. address: The target's address. port: The target's port, defaults to None in which case it will target the first device at the specified address """ return cls(ffx_runner, format_host_port(address, port) if port else address) @classmethod def from_node_name(cls, ffx_runner, node_name): """Args: ffx_runner: The runner to use to run ffx. node_name: The target's node name. """ return cls(ffx_runner, node_name) @classmethod def from_target_list_json(cls, ffx_runner, json_target): """Args: ffx_runner: The runner to use to run ffx. json_target: the json dict as returned from `ffx list targets` """ # Targets seen via `fx serve-remote` frequently have no name, so fall back # to using the first address. if json_target['nodename'].startswith('