1# Copyright 2014 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 5 6import collections 7import itertools 8import json 9import logging 10import time 11 12import six 13 14from pylib.base import base_test_result 15 16def GenerateResultsDict(test_run_results, global_tags=None): 17 """Create a results dict from |test_run_results| suitable for writing to JSON. 18 Args: 19 test_run_results: a list of base_test_result.TestRunResults objects. 20 Returns: 21 A results dict that mirrors the one generated by 22 base/test/launcher/test_results_tracker.cc:SaveSummaryAsJSON. 23 """ 24 # Example json output. 25 # { 26 # "global_tags": [], 27 # "all_tests": [ 28 # "test1", 29 # "test2", 30 # ], 31 # "disabled_tests": [], 32 # "per_iteration_data": [ 33 # { 34 # "test1": [ 35 # { 36 # "status": "SUCCESS", 37 # "elapsed_time_ms": 1, 38 # "output_snippet": "", 39 # "output_snippet_base64": "", 40 # "losless_snippet": "", 41 # }, 42 # ... 43 # ], 44 # "test2": [ 45 # { 46 # "status": "FAILURE", 47 # "elapsed_time_ms": 12, 48 # "output_snippet": "", 49 # "output_snippet_base64": "", 50 # "losless_snippet": "", 51 # }, 52 # ... 53 # ], 54 # }, 55 # { 56 # "test1": [ 57 # { 58 # "status": "SUCCESS", 59 # "elapsed_time_ms": 1, 60 # "output_snippet": "", 61 # "output_snippet_base64": "", 62 # "losless_snippet": "", 63 # }, 64 # ], 65 # "test2": [ 66 # { 67 # "status": "FAILURE", 68 # "elapsed_time_ms": 12, 69 # "output_snippet": "", 70 # "output_snippet_base64": "", 71 # "losless_snippet": "", 72 # }, 73 # ], 74 # }, 75 # ... 76 # ], 77 # } 78 79 all_tests = set() 80 per_iteration_data = [] 81 test_run_links = {} 82 83 for test_run_result in test_run_results: 84 iteration_data = collections.defaultdict(list) 85 if isinstance(test_run_result, list): 86 results_iterable = itertools.chain(*(t.GetAll() for t in test_run_result)) 87 for tr in test_run_result: 88 test_run_links.update(tr.GetLinks()) 89 90 else: 91 results_iterable = test_run_result.GetAll() 92 test_run_links.update(test_run_result.GetLinks()) 93 94 for r in results_iterable: 95 result_dict = { 96 'status': r.GetType(), 97 'elapsed_time_ms': r.GetDuration(), 98 'output_snippet': six.ensure_text(r.GetLog(), errors='replace'), 99 'losless_snippet': True, 100 'output_snippet_base64': '', 101 'links': r.GetLinks(), 102 } 103 iteration_data[r.GetName()].append(result_dict) 104 105 all_tests = all_tests.union(set(six.iterkeys(iteration_data))) 106 per_iteration_data.append(iteration_data) 107 108 return { 109 'global_tags': global_tags or [], 110 'all_tests': sorted(list(all_tests)), 111 # TODO(jbudorick): Add support for disabled tests within base_test_result. 112 'disabled_tests': [], 113 'per_iteration_data': per_iteration_data, 114 'links': test_run_links, 115 } 116 117 118def GenerateJsonTestResultFormatDict(test_run_results, interrupted): 119 """Create a results dict from |test_run_results| suitable for writing to JSON. 120 121 Args: 122 test_run_results: a list of base_test_result.TestRunResults objects. 123 interrupted: True if tests were interrupted, e.g. timeout listing tests 124 Returns: 125 A results dict that mirrors the standard JSON Test Results Format. 126 """ 127 128 tests = {} 129 counts = {'PASS': 0, 'FAIL': 0, 'SKIP': 0, 'CRASH': 0, 'TIMEOUT': 0} 130 131 for test_run_result in test_run_results: 132 if isinstance(test_run_result, list): 133 results_iterable = itertools.chain(*(t.GetAll() for t in test_run_result)) 134 else: 135 results_iterable = test_run_result.GetAll() 136 137 for r in results_iterable: 138 element = tests 139 for key in r.GetName().split('.'): 140 if key not in element: 141 element[key] = {} 142 element = element[key] 143 144 element['expected'] = 'PASS' 145 146 if r.GetType() == base_test_result.ResultType.PASS: 147 result = 'PASS' 148 elif r.GetType() == base_test_result.ResultType.SKIP: 149 result = 'SKIP' 150 elif r.GetType() == base_test_result.ResultType.CRASH: 151 result = 'CRASH' 152 elif r.GetType() == base_test_result.ResultType.TIMEOUT: 153 result = 'TIMEOUT' 154 else: 155 result = 'FAIL' 156 157 if 'actual' in element: 158 element['actual'] += ' ' + result 159 else: 160 counts[result] += 1 161 element['actual'] = result 162 if result == 'FAIL': 163 element['is_unexpected'] = True 164 165 if r.GetDuration() != 0: 166 element['time'] = r.GetDuration() 167 168 # Fill in required fields. 169 return { 170 'interrupted': interrupted, 171 'num_failures_by_type': counts, 172 'path_delimiter': '.', 173 'seconds_since_epoch': time.time(), 174 'tests': tests, 175 'version': 3, 176 } 177 178 179def GenerateJsonResultsFile(test_run_result, file_path, global_tags=None, 180 **kwargs): 181 """Write |test_run_result| to JSON. 182 183 This emulates the format of the JSON emitted by 184 base/test/launcher/test_results_tracker.cc:SaveSummaryAsJSON. 185 186 Args: 187 test_run_result: a base_test_result.TestRunResults object. 188 file_path: The path to the JSON file to write. 189 """ 190 with open(file_path, 'w') as json_result_file: 191 json_result_file.write(json.dumps( 192 GenerateResultsDict(test_run_result, global_tags=global_tags), 193 **kwargs)) 194 logging.info('Generated json results file at %s', file_path) 195 196 197def GenerateJsonTestResultFormatFile(test_run_result, interrupted, file_path, 198 **kwargs): 199 """Write |test_run_result| to JSON. 200 201 This uses the official Chromium Test Results Format. 202 203 Args: 204 test_run_result: a base_test_result.TestRunResults object. 205 interrupted: True if tests were interrupted, e.g. timeout listing tests 206 file_path: The path to the JSON file to write. 207 """ 208 with open(file_path, 'w') as json_result_file: 209 json_result_file.write( 210 json.dumps( 211 GenerateJsonTestResultFormatDict(test_run_result, interrupted), 212 **kwargs)) 213 logging.info('Generated json results file at %s', file_path) 214 215 216def ParseResultsFromJson(json_results): 217 """Creates a list of BaseTestResult objects from JSON. 218 219 Args: 220 json_results: A JSON dict in the format created by 221 GenerateJsonResultsFile. 222 """ 223 224 def string_as_status(s): 225 if s in base_test_result.ResultType.GetTypes(): 226 return s 227 return base_test_result.ResultType.UNKNOWN 228 229 results_list = [] 230 testsuite_runs = json_results['per_iteration_data'] 231 for testsuite_run in testsuite_runs: 232 for test, test_runs in six.iteritems(testsuite_run): 233 results_list.extend( 234 [base_test_result.BaseTestResult(test, 235 string_as_status(tr['status']), 236 duration=tr['elapsed_time_ms'], 237 log=tr.get('output_snippet')) 238 for tr in test_runs]) 239 return results_list 240