| #!/usr/bin/env python3.8 |
| # Copyright 2021 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| #### CATEGORY=Test |
| ### Query and calculate test stats. |
| |
| import argparse |
| import sys |
| import os |
| import subprocess |
| import json |
| from collections import defaultdict |
| import random |
| import time |
| |
| |
| # Retrieve stats on test execution. |
| # |
| # Example: fx test-stats --print |
| # (Prints a categorization of all tests in the current build.) |
| # |
| # Example: fx test-stats --print --os fuchsia |
| # (Prints a categorization of Fuchsia tests in the current build.) |
| # |
| # Example: fx test-stats --run --os fuchsia |
| # (Run all Fuchsia tests on a device.) |
| # |
| # Example: fx test-stats --run --dimension device_type=AEMU |
| # (Run only tests with device_type AEMU.) |
| # |
| # Example: fx test-stats --run --type v2 |
| # (Run only components v2 tests on a device.) |
| # |
| # Example: fx test-stats --run --type v2 --parallel 4 |
| # (Run components v2 tests with up to 4 cases running concurrently.) |
| # |
| # Example: fx test-stats --run --concurrent 4 |
| # (Run all tests with up to 4 test suites running concurrently.) |
| # |
| # Example: fx test-stats --run --timeout 5 |
| # (Run all tests with a maximum allowed running time of 5 seconds.) |
| def main(): |
| parser = argparse.ArgumentParser('Retrieve stats on test execution') |
| action_group = parser.add_mutually_exclusive_group(required=True) |
| action_group.add_argument( |
| '--print', |
| action='store_true', |
| help='Print a count of matching tests by their various properties') |
| action_group.add_argument( |
| '--run', action='store_true', help='Run all of the matching tests once') |
| filter_group = parser.add_argument_group('filter') |
| filter_group.add_argument( |
| '--dimension', |
| action='append', |
| help='Include only tests matching given dimensions; can be specified multiple times. ' |
| 'Example: --dimension os=Linux --dimension cpu=x64') |
| filter_group.add_argument( |
| '--os', |
| action='store', |
| help='Include only tests matching the given os. Example: --os fuchsia') |
| filter_group.add_argument( |
| '--type', |
| action='store', |
| help='Include only tests of this type. Example: --type v2') |
| run_group = parser.add_argument_group('run') |
| run_group.add_argument( |
| '-c', |
| '--concurrent', |
| action='store', |
| type=int, |
| default=1, |
| help='Number of tests to run concurrently. Default is 1.', |
| ) |
| run_group.add_argument( |
| '-t', |
| '--timeout', |
| action='store', |
| type=float, |
| default=120, |
| help='Timeout for tests, in seconds. Default is 120.', |
| ) |
| run_group.add_argument( |
| '-p', |
| '--parallel', |
| action='store', |
| type=int, |
| default=None, |
| help='Number of test cases per v2 suite to run in parallel. Uses runner default if not set.', |
| ) |
| run_group.add_argument( |
| '--shuffle', |
| action='store', |
| type=int, |
| default=None, |
| help='Toggle shuffling input. If set to 0, the current timestamp is used as the seed. If set no a non-zero number, that value will be used as the shuffle seed', |
| ) |
| run_group.add_argument( |
| '--count', |
| action='store', |
| type=int, |
| help='If set, only run this number of tests. The first tests in order following any shuffling will be used' |
| ) |
| args = parser.parse_args() |
| |
| test_tuples = get_tests_tuples(args) |
| |
| if args.print: |
| print_stats(test_tuples) |
| elif args.run: |
| run_tests( |
| test_tuples, |
| timeout_seconds=args.timeout, |
| max_running_tests=args.concurrent, |
| parallel=args.parallel) |
| else: |
| print('Unknown mode.') |
| parser.print_help() |
| return 1 |
| |
| return 0 |
| |
| |
| # Loads test definitions, parses them, and returns only those matching the argument filters. |
| # |
| # Returns a list of tuples (parsed_test, original_test_json) |
| def get_tests_tuples(args): |
| json_data = json.loads( |
| subprocess.check_output(['fx', 'test', '--printtests'])) |
| |
| dimension_pairs = None |
| if args.dimension: |
| dimension_pairs = set([ |
| (s[0], s[1]) |
| for s in map(lambda x: x.split('='), args.dimension) |
| if len(s) > 1 |
| ]) |
| |
| def to_include(val): |
| (parsed_test, test_json) = val |
| if parsed_test is None: |
| return False |
| |
| ret = True |
| if dimension_pairs is not None: |
| # Only include this test if one of its set of dimensions is completely included in the specified dimensions. |
| if not any( |
| all( |
| map(lambda v: v in dimension_pairs, |
| e['dimensions'].items())) |
| for e in test_json['environments']): |
| ret = False |
| if args.os is not None: |
| if test_json['test']['os'] != args.os: |
| ret = False |
| if args.type is not None: |
| if parsed_test.test_type != args.type: |
| ret = False |
| |
| return ret |
| |
| test_list = list( |
| filter(to_include, map(lambda x: (parse_test(x), x), json_data))) |
| |
| if args.shuffle is not None: |
| shuffle = args.shuffle |
| if shuffle == 0: |
| shuffle = None |
| random.seed(shuffle) |
| random.shuffle(test_list) |
| if args.count is not None: |
| test_list = test_list[:args.count] |
| |
| return test_list |
| |
| |
| # Wraps a test parsed from `fx test --printtests` |
| class Test: |
| |
| def __init__(self, test_type=None, path=None, package_url=None): |
| self.test_type = test_type |
| self.path = path |
| self.package_url = package_url |
| |
| # Get a unique key for this test |
| def key(self): |
| return self.package_url or self.path |
| |
| def __str__(self): |
| return '{} test "{}"'.format(self.test_type, self.package_url or |
| self.path) |
| |
| |
| # Parse a single JSON dict into a Test class. |
| def parse_test(test): |
| test_val = test['test'] |
| if 'package_url' in test_val: |
| suffix = test_val['package_url'][-3:] |
| if suffix == 'cmx': |
| return Test(test_type='v1', package_url=test_val['package_url']) |
| elif suffix == '.cm': |
| return Test(test_type='v2', package_url=test_val['package_url']) |
| else: |
| return 'unknown package' |
| elif 'path' in test_val: |
| return Test(test_type='host', path=test_val['path']) |
| else: |
| return Test(test_type='unknown') |
| |
| |
| # Implementation for --print mode. |
| def print_stats(test_tuples): |
| dimension_counts = defaultdict(int) |
| type_counts = defaultdict(int) |
| for (parsed_test, test_json) in test_tuples: |
| for env in test_json['environments']: |
| lst = [] |
| for name, value in env['dimensions'].items(): |
| lst.append('{}: {}'.format(name, value)) |
| lst.sort() |
| dimension_counts['({})'.format(', '.join(lst))] += 1 |
| type_counts[parsed_test.test_type] += 1 |
| |
| print('Dimensions:') |
| for name, count in dimension_counts.items(): |
| print(' {}: {}'.format(name, count)) |
| print('Types:') |
| for name, count in type_counts.items(): |
| print(' {}: {}'.format(name, count)) |
| |
| |
| # Wrapper for a test that is currently running on a device. |
| # |
| # This class keeps track of the start and end times of the test run. |
| class StartedTest: |
| |
| # Start executing the given command line and wrap it in a StartedTest. |
| @staticmethod |
| def create(command_line): |
| return StartedTest( |
| command_line, time.time(), |
| subprocess.Popen( |
| command_line, |
| stdout=subprocess.DEVNULL, |
| stderr=subprocess.DEVNULL)) |
| |
| def __init__(self, command_line, start_time, running_process): |
| self._command_line = command_line |
| self._start_time = start_time |
| self._running_process = running_process |
| self._end_time = None |
| |
| # Check if the test is done, updating internal state. |
| # |
| # This function must be called periodically to determine when the test is complete. |
| # |
| # Returns True if the test is done, False otherwise. |
| def poll_done(self): |
| if self._end_time is not None: |
| return True |
| |
| if self._running_process.poll() is not None: |
| self._end_time = time.time() |
| self._running_process.communicate() # drain pipes |
| return True |
| return False |
| |
| # Returns the return code for the test, or None if it is still running. |
| def return_code(self): |
| return self._running_process.returncode |
| |
| # Returns the original command line for the test. |
| def command_line(self): |
| return str(self._command_line) |
| |
| # Returns the runtime for the test. |
| # |
| # If the test is currently running, this returns currently elapsed |
| # time. Otherwise it returns the runtime from start to end of the |
| # wrapped test. |
| def runtime(self): |
| return (self._end_time - self._start_time if self._end_time is not None |
| else time.time() - self._start_time) |
| |
| # Force the test to terminate if it is currently running. |
| def terminate(self): |
| self._running_process.terminate() |
| |
| |
| # Start an individual Test. |
| # |
| # Returns a StartedTest if the test could be started, and None otherwise. |
| def start_test(test_object, parallel=None, timeout=None): |
| if test_object.test_type == 'v1': |
| command_line = ['fx', 'shell', 'run-test-component'] |
| if timeout: |
| command_line.append(f'--timeout={int(timeout)}') |
| command_line.append(test_object.package_url) |
| return StartedTest.create(command_line) |
| if test_object.test_type == 'v2': |
| command_line = [ |
| 'fx', 'shell', 'run-test-suite', test_object.package_url |
| ] |
| if parallel: |
| command_line.append('--parallel') |
| command_line.append(f'{parallel}') |
| if timeout: |
| command_line.append('--timeout') |
| command_line.append(f'{int(timeout)}') |
| return StartedTest.create(command_line) |
| else: |
| return None |
| |
| |
| # Implementation for --run mode. |
| def run_tests(to_run, |
| timeout_seconds=None, |
| max_running_tests=None, |
| parallel=None): |
| |
| test_iter = iter(to_run) |
| running_tests = [] |
| outcomes = defaultdict(list) |
| start_time = time.time() |
| |
| # Internal function to poll and update individual tests. |
| def process_running_test(test): |
| if test.poll_done(): |
| mode = '' |
| if test.return_code() == 0: |
| mode = 'SUCCESS' |
| else: |
| mode = f'code {test.return_code()}' |
| outcomes[mode].append(test) |
| print(f'{mode} [{test.runtime()}]: {test.command_line()}') |
| return False |
| elif test.runtime() > timeout_seconds: |
| test.terminate() |
| |
| return True |
| |
| # Internal function to print status of the run. |
| def print_status(): |
| skipped = len(outcomes['SKIPPED']) if 'SKIPPED' in outcomes else 0 |
| total_done = sum([len(v) for v in outcomes.values()]) - skipped |
| print( |
| f'Status: {total_done}/{len(to_run)} {len(running_tests)} running {skipped} skipped' |
| ) |
| for (k, v) in sorted(outcomes.items()): |
| if k == 'SKIPPED': |
| continue |
| runtimes = list( |
| map(lambda x: x.runtime() if hasattr(x, 'runtime') else 0, v)) |
| total_time = sum(runtimes) |
| average_time = total_time / len(runtimes) |
| print( |
| f' {k:10s}: {len(runtimes):7d} {total_time:9.3f} {average_time:9.3f} avg.' |
| ) |
| overall_time = time.time() - start_time |
| avg_overall_time = overall_time / total_done if total_done != 0 else 0 |
| print( |
| f' TOTAL : {total_done:7d} {overall_time:9.3f} {avg_overall_time:9.3f} avg.' |
| ) |
| |
| next_test = next(test_iter, None) |
| |
| while running_tests or next_test is not None: |
| start_set = {t.command_line() for t in running_tests} |
| |
| running_tests = list(filter(process_running_test, running_tests)) |
| |
| # Continueally start tests so long as there is a test to be |
| # started and we have not yet reached the maximum. |
| while len(running_tests) < max_running_tests and next_test is not None: |
| started_test = start_test(next_test[0], parallel, timeout_seconds) |
| |
| if started_test is not None: |
| print(f'Started {next_test[0].key()}') |
| running_tests.append(started_test) |
| else: |
| outcomes['SKIPPED'].append(next_test) |
| print(f'Skipped {next_test[0].key()}') |
| |
| next_test = next(test_iter, None) |
| |
| # Print output if we have changed the set of running tests since the start of this iteration. |
| if {t.command_line() for t in running_tests} != start_set: |
| print_status() |
| |
| # Sleep for 10ms before continuing loop to reduce CPU load. |
| time.sleep(.01) |
| |
| print_status() |
| |
| |
| if __name__ == '__main__': |
| sys.exit(main()) |
