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