1# Copyright 2019, The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14""" 15ATest execution info generator. 16""" 17 18from __future__ import print_function 19 20import glob 21import logging 22import json 23import os 24import sys 25 26import atest_utils as au 27import constants 28 29from metrics import metrics_utils 30 31_ARGS_KEY = 'args' 32_STATUS_PASSED_KEY = 'PASSED' 33_STATUS_FAILED_KEY = 'FAILED' 34_STATUS_IGNORED_KEY = 'IGNORED' 35_SUMMARY_KEY = 'summary' 36_TOTAL_SUMMARY_KEY = 'total_summary' 37_TEST_RUNNER_KEY = 'test_runner' 38_TEST_NAME_KEY = 'test_name' 39_TEST_TIME_KEY = 'test_time' 40_TEST_DETAILS_KEY = 'details' 41_TEST_RESULT_NAME = 'test_result' 42_EXIT_CODE_ATTR = 'EXIT_CODE' 43_MAIN_MODULE_KEY = '__main__' 44_UUID_LEN = 30 45_RESULT_LEN = 35 46_COMMAND_LEN = 50 47_LOGCAT_FMT = '{}/log/invocation_*/{}*logcat-on-failure*' 48 49_SUMMARY_MAP_TEMPLATE = {_STATUS_PASSED_KEY : 0, 50 _STATUS_FAILED_KEY : 0, 51 _STATUS_IGNORED_KEY : 0,} 52 53PREPARE_END_TIME = None 54 55 56def preparation_time(start_time): 57 """Return the preparation time. 58 59 Args: 60 start_time: The time. 61 62 Returns: 63 The preparation time if PREPARE_END_TIME is set, None otherwise. 64 """ 65 return PREPARE_END_TIME - start_time if PREPARE_END_TIME else None 66 67 68def symlink_latest_result(test_result_dir): 69 """Make the symbolic link to latest result. 70 71 Args: 72 test_result_dir: A string of the dir path. 73 """ 74 symlink = os.path.join(constants.ATEST_RESULT_ROOT, 'LATEST') 75 if os.path.exists(symlink) or os.path.islink(symlink): 76 os.remove(symlink) 77 os.symlink(test_result_dir, symlink) 78 79 80def print_test_result(root, history_arg): 81 """Make a list of latest n test result. 82 83 Args: 84 root: A string of the test result root path. 85 history_arg: A string of an integer or uuid. If it's an integer string, 86 the number of lines of test result will be given; else it 87 will be treated a uuid and print test result accordingly 88 in detail. 89 """ 90 if not history_arg.isdigit(): 91 path = os.path.join(constants.ATEST_RESULT_ROOT, history_arg, 92 'test_result') 93 print_test_result_by_path(path) 94 return 95 target = '%s/20*_*_*' % root 96 paths = glob.glob(target) 97 paths.sort(reverse=True) 98 print('{:-^{uuid_len}} {:-^{result_len}} {:-^{command_len}}' 99 .format('uuid', 'result', 'command', 100 uuid_len=_UUID_LEN, 101 result_len=_RESULT_LEN, 102 command_len=_COMMAND_LEN)) 103 for path in paths[0: int(history_arg)+1]: 104 result_path = os.path.join(path, 'test_result') 105 if os.path.isfile(result_path): 106 try: 107 with open(result_path) as json_file: 108 result = json.load(json_file) 109 total_summary = result.get(_TOTAL_SUMMARY_KEY, {}) 110 summary_str = ', '.join([k+':'+str(v) 111 for k, v in total_summary.items()]) 112 print('{:<{uuid_len}} {:<{result_len}} {:<{command_len}}' 113 .format(os.path.basename(path), 114 summary_str, 115 'atest '+result.get(_ARGS_KEY, ''), 116 uuid_len=_UUID_LEN, 117 result_len=_RESULT_LEN, 118 command_len=_COMMAND_LEN)) 119 except ValueError: 120 pass 121 122 123def print_test_result_by_path(path): 124 """Print latest test result. 125 126 Args: 127 path: A string of test result path. 128 """ 129 if os.path.isfile(path): 130 with open(path) as json_file: 131 result = json.load(json_file) 132 print("\natest {}".format(result.get(_ARGS_KEY, ''))) 133 print('\nTotal Summary:\n--------------') 134 total_summary = result.get(_TOTAL_SUMMARY_KEY, {}) 135 print(', '.join([(k+':'+str(v)) 136 for k, v in total_summary.items()])) 137 fail_num = total_summary.get(_STATUS_FAILED_KEY) 138 if fail_num > 0: 139 message = '%d test failed' % fail_num 140 print('\n') 141 print(au.colorize(message, constants.RED)) 142 print('-' * len(message)) 143 test_runner = result.get(_TEST_RUNNER_KEY, {}) 144 for runner_name in test_runner.keys(): 145 test_dict = test_runner.get(runner_name, {}) 146 for test_name in test_dict: 147 test_details = test_dict.get(test_name, {}) 148 for fail in test_details.get(_STATUS_FAILED_KEY): 149 print(au.colorize('{}'.format( 150 fail.get(_TEST_NAME_KEY)), constants.RED)) 151 failure_files = glob.glob(_LOGCAT_FMT.format( 152 os.path.dirname(path), fail.get(_TEST_NAME_KEY) 153 )) 154 if failure_files: 155 print('{} {}'.format( 156 au.colorize('LOGCAT-ON-FAILURES:', 157 constants.CYAN), 158 failure_files[0])) 159 print('{} {}'.format( 160 au.colorize('STACKTRACE:\n', constants.CYAN), 161 fail.get(_TEST_DETAILS_KEY))) 162 163 164def has_non_test_options(args): 165 """ 166 check whether non-test option in the args. 167 168 Args: 169 args: An argspace.Namespace class instance holding parsed args. 170 171 Returns: 172 True, if args has at least one non-test option. 173 False, otherwise. 174 """ 175 return (args.collect_tests_only 176 or args.dry_run 177 or args.help 178 or args.history 179 or args.info 180 or args.version 181 or args.latest_result) 182 183 184class AtestExecutionInfo(object): 185 """Class that stores the whole test progress information in JSON format. 186 187 ---- 188 For example, running command 189 atest hello_world_test HelloWorldTest 190 191 will result in storing the execution detail in JSON: 192 { 193 "args": "hello_world_test HelloWorldTest", 194 "test_runner": { 195 "AtestTradefedTestRunner": { 196 "hello_world_test": { 197 "FAILED": [ 198 {"test_time": "(5ms)", 199 "details": "Hello, Wor...", 200 "test_name": "HelloWorldTest#PrintHelloWorld"} 201 ], 202 "summary": {"FAILED": 1, "PASSED": 0, "IGNORED": 0} 203 }, 204 "HelloWorldTests": { 205 "PASSED": [ 206 {"test_time": "(27ms)", 207 "details": null, 208 "test_name": "...HelloWorldTest#testHalloWelt"}, 209 {"test_time": "(1ms)", 210 "details": null, 211 "test_name": "....HelloWorldTest#testHelloWorld"} 212 ], 213 "summary": {"FAILED": 0, "PASSED": 2, "IGNORED": 0} 214 } 215 } 216 }, 217 "total_summary": {"FAILED": 1, "PASSED": 2, "IGNORED": 0} 218 } 219 """ 220 221 result_reporters = [] 222 223 def __init__(self, args, work_dir, args_ns): 224 """Initialise an AtestExecutionInfo instance. 225 226 Args: 227 args: Command line parameters. 228 work_dir: The directory for saving information. 229 args_ns: An argspace.Namespace class instance holding parsed args. 230 231 Returns: 232 A json format string. 233 """ 234 self.args = args 235 self.work_dir = work_dir 236 self.result_file = None 237 self.args_ns = args_ns 238 239 def __enter__(self): 240 """Create and return information file object.""" 241 full_file_name = os.path.join(self.work_dir, _TEST_RESULT_NAME) 242 try: 243 self.result_file = open(full_file_name, 'w') 244 except IOError: 245 logging.error('Cannot open file %s', full_file_name) 246 return self.result_file 247 248 def __exit__(self, exit_type, value, traceback): 249 """Write execution information and close information file.""" 250 if self.result_file: 251 self.result_file.write(AtestExecutionInfo. 252 _generate_execution_detail(self.args)) 253 self.result_file.close() 254 if not has_non_test_options(self.args_ns): 255 symlink_latest_result(self.work_dir) 256 main_module = sys.modules.get(_MAIN_MODULE_KEY) 257 main_exit_code = getattr(main_module, _EXIT_CODE_ATTR, 258 constants.EXIT_CODE_ERROR) 259 if main_exit_code == constants.EXIT_CODE_SUCCESS: 260 metrics_utils.send_exit_event(main_exit_code) 261 else: 262 metrics_utils.handle_exc_and_send_exit_event(main_exit_code) 263 264 @staticmethod 265 def _generate_execution_detail(args): 266 """Generate execution detail. 267 268 Args: 269 args: Command line parameters that you want to save. 270 271 Returns: 272 A json format string. 273 """ 274 info_dict = {_ARGS_KEY: ' '.join(args)} 275 try: 276 AtestExecutionInfo._arrange_test_result( 277 info_dict, 278 AtestExecutionInfo.result_reporters) 279 return json.dumps(info_dict) 280 except ValueError as err: 281 logging.warn('Parsing test result failed due to : %s', err) 282 283 @staticmethod 284 def _arrange_test_result(info_dict, reporters): 285 """Append test result information in given dict. 286 287 Arrange test information to below 288 "test_runner": { 289 "test runner name": { 290 "test name": { 291 "FAILED": [ 292 {"test time": "", 293 "details": "", 294 "test name": ""} 295 ], 296 "summary": {"FAILED": 0, "PASSED": 0, "IGNORED": 0} 297 }, 298 }, 299 "total_summary": {"FAILED": 0, "PASSED": 0, "IGNORED": 0} 300 301 Args: 302 info_dict: A dict you want to add result information in. 303 reporters: A list of result_reporter. 304 305 Returns: 306 A dict contains test result information data. 307 """ 308 info_dict[_TEST_RUNNER_KEY] = {} 309 for reporter in reporters: 310 for test in reporter.all_test_results: 311 runner = info_dict[_TEST_RUNNER_KEY].setdefault(test.runner_name, {}) 312 group = runner.setdefault(test.group_name, {}) 313 result_dict = {_TEST_NAME_KEY : test.test_name, 314 _TEST_TIME_KEY : test.test_time, 315 _TEST_DETAILS_KEY : test.details} 316 group.setdefault(test.status, []).append(result_dict) 317 318 total_test_group_summary = _SUMMARY_MAP_TEMPLATE.copy() 319 for runner in info_dict[_TEST_RUNNER_KEY]: 320 for group in info_dict[_TEST_RUNNER_KEY][runner]: 321 group_summary = _SUMMARY_MAP_TEMPLATE.copy() 322 for status in info_dict[_TEST_RUNNER_KEY][runner][group]: 323 count = len(info_dict[_TEST_RUNNER_KEY][runner][group][status]) 324 if _SUMMARY_MAP_TEMPLATE.has_key(status): 325 group_summary[status] = count 326 total_test_group_summary[status] += count 327 info_dict[_TEST_RUNNER_KEY][runner][group][_SUMMARY_KEY] = group_summary 328 info_dict[_TOTAL_SUMMARY_KEY] = total_test_group_summary 329 return info_dict 330