| # Copyright 2019, The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| """ATest execution info generator.""" |
| |
| # pylint: disable=line-too-long |
| |
| from __future__ import print_function |
| |
| import glob |
| import logging |
| import json |
| import os |
| import sys |
| |
| import atest_utils as au |
| import constants |
| |
| from metrics import metrics_utils |
| |
| _ARGS_KEY = 'args' |
| _STATUS_PASSED_KEY = 'PASSED' |
| _STATUS_FAILED_KEY = 'FAILED' |
| _STATUS_IGNORED_KEY = 'IGNORED' |
| _SUMMARY_KEY = 'summary' |
| _TOTAL_SUMMARY_KEY = 'total_summary' |
| _TEST_RUNNER_KEY = 'test_runner' |
| _TEST_NAME_KEY = 'test_name' |
| _TEST_TIME_KEY = 'test_time' |
| _TEST_DETAILS_KEY = 'details' |
| _TEST_RESULT_NAME = 'test_result' |
| _EXIT_CODE_ATTR = 'EXIT_CODE' |
| _MAIN_MODULE_KEY = '__main__' |
| _UUID_LEN = 30 |
| _RESULT_LEN = 35 |
| _COMMAND_LEN = 50 |
| _LOGCAT_FMT = '{}/log/invocation_*/{}*logcat-on-failure*' |
| |
| _SUMMARY_MAP_TEMPLATE = {_STATUS_PASSED_KEY : 0, |
| _STATUS_FAILED_KEY : 0, |
| _STATUS_IGNORED_KEY : 0,} |
| |
| PREPARE_END_TIME = None |
| |
| |
| def preparation_time(start_time): |
| """Return the preparation time. |
| |
| Args: |
| start_time: The time. |
| |
| Returns: |
| The preparation time if PREPARE_END_TIME is set, None otherwise. |
| """ |
| return PREPARE_END_TIME - start_time if PREPARE_END_TIME else None |
| |
| |
| def symlink_latest_result(test_result_dir): |
| """Make the symbolic link to latest result. |
| |
| Args: |
| test_result_dir: A string of the dir path. |
| """ |
| symlink = os.path.join(constants.ATEST_RESULT_ROOT, 'LATEST') |
| if os.path.exists(symlink) or os.path.islink(symlink): |
| os.remove(symlink) |
| os.symlink(test_result_dir, symlink) |
| |
| |
| def print_test_result(root, history_arg): |
| """Make a list of latest n test result. |
| |
| Args: |
| root: A string of the test result root path. |
| history_arg: A string of an integer or uuid. If it's an integer string, |
| the number of lines of test result will be given; else it |
| will be treated a uuid and print test result accordingly |
| in detail. |
| """ |
| if not history_arg.isdigit(): |
| path = os.path.join(constants.ATEST_RESULT_ROOT, history_arg, |
| 'test_result') |
| print_test_result_by_path(path) |
| return |
| target = '%s/20*_*_*' % root |
| paths = glob.glob(target) |
| paths.sort(reverse=True) |
| print('{:-^{uuid_len}} {:-^{result_len}} {:-^{command_len}}' |
| .format('uuid', 'result', 'command', |
| uuid_len=_UUID_LEN, |
| result_len=_RESULT_LEN, |
| command_len=_COMMAND_LEN)) |
| for path in paths[0: int(history_arg)+1]: |
| result_path = os.path.join(path, 'test_result') |
| if os.path.isfile(result_path): |
| try: |
| with open(result_path) as json_file: |
| result = json.load(json_file) |
| total_summary = result.get(_TOTAL_SUMMARY_KEY, {}) |
| summary_str = ', '.join([k+':'+str(v) |
| for k, v in total_summary.items()]) |
| print('{:<{uuid_len}} {:<{result_len}} atest {:<{command_len}}' |
| .format(os.path.basename(path), |
| summary_str, |
| result.get(_ARGS_KEY, ''), |
| uuid_len=_UUID_LEN, |
| result_len=_RESULT_LEN, |
| command_len=_COMMAND_LEN)) |
| except ValueError: |
| pass |
| |
| |
| def print_test_result_by_path(path): |
| """Print latest test result. |
| |
| Args: |
| path: A string of test result path. |
| """ |
| if os.path.isfile(path): |
| with open(path) as json_file: |
| result = json.load(json_file) |
| print("\natest {}".format(result.get(_ARGS_KEY, ''))) |
| print('\nTotal Summary:\n{}'.format(au.delimiter('-'))) |
| total_summary = result.get(_TOTAL_SUMMARY_KEY, {}) |
| print(', '.join([(k+':'+str(v)) |
| for k, v in total_summary.items()])) |
| fail_num = total_summary.get(_STATUS_FAILED_KEY) |
| if fail_num > 0: |
| message = '%d test failed' % fail_num |
| print('\n') |
| print(au.colorize(message, constants.RED)) |
| print('-' * len(message)) |
| test_runner = result.get(_TEST_RUNNER_KEY, {}) |
| for runner_name in test_runner.keys(): |
| test_dict = test_runner.get(runner_name, {}) |
| for test_name in test_dict: |
| test_details = test_dict.get(test_name, {}) |
| for fail in test_details.get(_STATUS_FAILED_KEY): |
| print(au.colorize('{}'.format( |
| fail.get(_TEST_NAME_KEY)), constants.RED)) |
| failure_files = glob.glob(_LOGCAT_FMT.format( |
| os.path.dirname(path), fail.get(_TEST_NAME_KEY) |
| )) |
| if failure_files: |
| print('{} {}'.format( |
| au.colorize('LOGCAT-ON-FAILURES:', |
| constants.CYAN), |
| failure_files[0])) |
| print('{} {}'.format( |
| au.colorize('STACKTRACE:\n', constants.CYAN), |
| fail.get(_TEST_DETAILS_KEY))) |
| |
| |
| def has_non_test_options(args): |
| """ |
| check whether non-test option in the args. |
| |
| Args: |
| args: An argspace.Namespace class instance holding parsed args. |
| |
| Returns: |
| True, if args has at least one non-test option. |
| False, otherwise. |
| """ |
| return (args.collect_tests_only |
| or args.dry_run |
| or args.help |
| or args.history |
| or args.info |
| or args.version |
| or args.latest_result) |
| |
| |
| class AtestExecutionInfo: |
| """Class that stores the whole test progress information in JSON format. |
| |
| ---- |
| For example, running command |
| atest hello_world_test HelloWorldTest |
| |
| will result in storing the execution detail in JSON: |
| { |
| "args": "hello_world_test HelloWorldTest", |
| "test_runner": { |
| "AtestTradefedTestRunner": { |
| "hello_world_test": { |
| "FAILED": [ |
| {"test_time": "(5ms)", |
| "details": "Hello, Wor...", |
| "test_name": "HelloWorldTest#PrintHelloWorld"} |
| ], |
| "summary": {"FAILED": 1, "PASSED": 0, "IGNORED": 0} |
| }, |
| "HelloWorldTests": { |
| "PASSED": [ |
| {"test_time": "(27ms)", |
| "details": null, |
| "test_name": "...HelloWorldTest#testHalloWelt"}, |
| {"test_time": "(1ms)", |
| "details": null, |
| "test_name": "....HelloWorldTest#testHelloWorld"} |
| ], |
| "summary": {"FAILED": 0, "PASSED": 2, "IGNORED": 0} |
| } |
| } |
| }, |
| "total_summary": {"FAILED": 1, "PASSED": 2, "IGNORED": 0} |
| } |
| """ |
| |
| result_reporters = [] |
| |
| def __init__(self, args, work_dir, args_ns): |
| """Initialise an AtestExecutionInfo instance. |
| |
| Args: |
| args: Command line parameters. |
| work_dir: The directory for saving information. |
| args_ns: An argspace.Namespace class instance holding parsed args. |
| |
| Returns: |
| A json format string. |
| """ |
| self.args = args |
| self.work_dir = work_dir |
| self.result_file = None |
| self.args_ns = args_ns |
| |
| def __enter__(self): |
| """Create and return information file object.""" |
| full_file_name = os.path.join(self.work_dir, _TEST_RESULT_NAME) |
| try: |
| self.result_file = open(full_file_name, 'w') |
| except IOError: |
| logging.error('Cannot open file %s', full_file_name) |
| return self.result_file |
| |
| def __exit__(self, exit_type, value, traceback): |
| """Write execution information and close information file.""" |
| if self.result_file: |
| self.result_file.write(AtestExecutionInfo. |
| _generate_execution_detail(self.args)) |
| self.result_file.close() |
| if not has_non_test_options(self.args_ns): |
| symlink_latest_result(self.work_dir) |
| main_module = sys.modules.get(_MAIN_MODULE_KEY) |
| main_exit_code = getattr(main_module, _EXIT_CODE_ATTR, |
| constants.EXIT_CODE_ERROR) |
| if main_exit_code == constants.EXIT_CODE_SUCCESS: |
| metrics_utils.send_exit_event(main_exit_code) |
| else: |
| metrics_utils.handle_exc_and_send_exit_event(main_exit_code) |
| |
| @staticmethod |
| def _generate_execution_detail(args): |
| """Generate execution detail. |
| |
| Args: |
| args: Command line parameters that you want to save. |
| |
| Returns: |
| A json format string. |
| """ |
| info_dict = {_ARGS_KEY: ' '.join(args)} |
| try: |
| AtestExecutionInfo._arrange_test_result( |
| info_dict, |
| AtestExecutionInfo.result_reporters) |
| return json.dumps(info_dict) |
| except ValueError as err: |
| logging.warning('Parsing test result failed due to : %s', err) |
| |
| @staticmethod |
| def _arrange_test_result(info_dict, reporters): |
| """Append test result information in given dict. |
| |
| Arrange test information to below |
| "test_runner": { |
| "test runner name": { |
| "test name": { |
| "FAILED": [ |
| {"test time": "", |
| "details": "", |
| "test name": ""} |
| ], |
| "summary": {"FAILED": 0, "PASSED": 0, "IGNORED": 0} |
| }, |
| }, |
| "total_summary": {"FAILED": 0, "PASSED": 0, "IGNORED": 0} |
| |
| Args: |
| info_dict: A dict you want to add result information in. |
| reporters: A list of result_reporter. |
| |
| Returns: |
| A dict contains test result information data. |
| """ |
| info_dict[_TEST_RUNNER_KEY] = {} |
| for reporter in reporters: |
| for test in reporter.all_test_results: |
| runner = info_dict[_TEST_RUNNER_KEY].setdefault( |
| test.runner_name, {}) |
| group = runner.setdefault(test.group_name, {}) |
| result_dict = {_TEST_NAME_KEY : test.test_name, |
| _TEST_TIME_KEY : test.test_time, |
| _TEST_DETAILS_KEY : test.details} |
| group.setdefault(test.status, []).append(result_dict) |
| |
| total_test_group_summary = _SUMMARY_MAP_TEMPLATE.copy() |
| for runner in info_dict[_TEST_RUNNER_KEY]: |
| for group in info_dict[_TEST_RUNNER_KEY][runner]: |
| group_summary = _SUMMARY_MAP_TEMPLATE.copy() |
| for status in info_dict[_TEST_RUNNER_KEY][runner][group]: |
| count = len(info_dict[_TEST_RUNNER_KEY][runner][group][status]) |
| if status in _SUMMARY_MAP_TEMPLATE: |
| group_summary[status] = count |
| total_test_group_summary[status] += count |
| info_dict[_TEST_RUNNER_KEY][runner][group][_SUMMARY_KEY] = group_summary |
| info_dict[_TOTAL_SUMMARY_KEY] = total_test_group_summary |
| return info_dict |
