1# Copyright (c) 2014 The Chromium Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5from __future__ import print_function 6 7import argparse 8import copy 9import csv 10import logging 11import os 12import re 13import shutil 14 15CONNECT_FAIL = object() 16CONFIG_FAIL = object() 17RESULTS_DIR = '/tmp/chaos' 18 19 20class ChaosParser(object): 21 """Defines a parser for chaos test results""" 22 23 def __init__(self, results_dir, create_file, print_config_failures): 24 """ Constructs a parser interface. 25 26 @param results_dir: complete path to restuls directory for a chaos test. 27 @param create_file: True to create csv files; False otherwise. 28 @param print_config_failures: True to print the config info to stdout; 29 False otherwise. 30 31 """ 32 self._test_results_dir = results_dir 33 self._create_file = create_file 34 self._print_config_failures = print_config_failures 35 36 37 def convert_set_to_string(self, set_list): 38 """Converts a set to a single string. 39 40 @param set_list: a set to convert 41 42 @returns a string, which is all items separated by the word 'and' 43 44 """ 45 return_string = str() 46 for i in set_list: 47 return_string += str('%s and ' % i) 48 return return_string[:-5] 49 50 51 def create_csv(self, filename, data_list): 52 """Creates a file in .csv format. 53 54 @param filename: name for the csv file 55 @param data_list: a list of all the info to write to a file 56 57 """ 58 if not os.path.exists(RESULTS_DIR): 59 os.mkdir(RESULTS_DIR) 60 try: 61 path = os.path.join(RESULTS_DIR, filename + '.csv') 62 with open(path, 'wb') as f: 63 writer = csv.writer(f) 64 writer.writerow(data_list) 65 logging.info('Created CSV file %s', path) 66 except IOError as e: 67 logging.error('File operation failed with %s: %s', e.errno, 68 e.strerror) 69 return 70 71 72 def get_ap_name(self, line): 73 """Gets the router name from the string passed. 74 75 @param line: Test ERROR string from chaos status.log 76 77 @returns the router name or brand. 78 79 """ 80 router_info = re.search('Router name: ([\w\s]+)', line) 81 return router_info.group(1) 82 83 84 def get_ap_mode_chan_freq(self, ssid): 85 """Gets the AP band from ssid using channel. 86 87 @param ssid: A valid chaos test SSID as a string 88 89 @returns the AP band, mode, and channel. 90 91 """ 92 channel_security_info = ssid.split('_') 93 channel_info = channel_security_info[-2] 94 mode = channel_security_info[-3] 95 channel = int(re.split('(\d+)', channel_info)[1]) 96 # TODO Choose if we want to keep band, we never put it in the 97 # spreadsheet and is currently unused. 98 if channel in range(1, 15): 99 band = '2.4GHz' 100 else: 101 band = '5GHz' 102 return {'mode': mode.upper(), 'channel': channel, 103 'band': band} 104 105 106 def generate_percentage_string(self, passed_tests, total_tests): 107 """Creates a pass percentage string in the formation x/y (zz%) 108 109 @param passed_tests: int of passed tests 110 @param total_tests: int of total tests 111 112 @returns a formatted string as described above. 113 114 """ 115 percent = float(passed_tests)/float(total_tests) * 100 116 percent_string = str(int(round(percent))) + '%' 117 return str('%d/%d (%s)' % (passed_tests, total_tests, percent_string)) 118 119 120 def parse_keyval(self, filepath): 121 """Parses the 'keyvalue' file to get device details. 122 123 @param filepath: the complete path to the keyval file 124 125 @returns a board with device name and OS version. 126 127 """ 128 # Android information does not exist in the keyfile, add temporary 129 # information into the dictionary. crbug.com/570408 130 lsb_dict = {'board': 'unknown', 131 'version': 'unknown'} 132 f = open(filepath, 'r') 133 for line in f: 134 line = line.split('=') 135 if 'RELEASE_BOARD' in line[0]: 136 lsb_dict = {'board':line[1].rstrip()} 137 elif 'RELEASE_VERSION' in line[0]: 138 lsb_dict['version'] = line[1].rstrip() 139 else: 140 continue 141 f.close() 142 return lsb_dict 143 144 145 def parse_status_log(self, board, os_version, security, status_log_path): 146 """Parses the entire status.log file from chaos test for test failures. 147 and creates two CSV files for connect fail and configuration fail 148 respectively. 149 150 @param board: the board the test was run against as a string 151 @param os_version: the version of ChromeOS as a string 152 @param security: the security used during the test as a string 153 @param status_log_path: complete path to the status.log file 154 155 """ 156 # Items that can have multiple values 157 modes = list() 158 channels = list() 159 test_fail_aps = list() 160 static_config_failures = list() 161 dynamic_config_failures = list() 162 kernel_version = "" 163 fw_version = "" 164 f = open(status_log_path, 'r') 165 total = 0 166 for line in f: 167 line = line.strip() 168 if line.startswith('START\tnetwork_WiFi'): 169 # Do not count PDU failures in total tests run. 170 if 'PDU' in line: 171 continue 172 total += 1 173 elif 'kernel_version' in line: 174 kernel_version = re.search('[\d.]+', line).group(0) 175 elif 'firmware_version' in line: 176 fw_version = re.search('firmware_version\': \'([\w\s:().]+)', 177 line).group(1) 178 elif line.startswith('ERROR') or line.startswith('FAIL'): 179 title_info = line.split() 180 if 'reboot' in title_info: 181 continue 182 # Get the hostname for the AP that failed configuration. 183 if 'PDU' in title_info[1]: 184 continue 185 else: 186 # Get the router name, band for the AP that failed 187 # connect. 188 if 'Config' in title_info[1]: 189 failure_type = CONFIG_FAIL 190 else: 191 failure_type = CONNECT_FAIL 192 193 if (failure_type == CONFIG_FAIL and 194 'chromeos' in title_info[1]): 195 ssid = title_info[1].split('.')[1].split('_')[0] 196 else: 197 ssid_info = title_info[1].split('.') 198 ssid = ssid_info[1] 199 network_dict = self.get_ap_mode_chan_freq(ssid) 200 modes.append(network_dict['mode']) 201 channels.append(network_dict['channel']) 202 203 # Security mismatches and Ping failures are not connect 204 # failures. 205 if (('Ping command' in line or 'correct security' in line) 206 or failure_type == CONFIG_FAIL): 207 if 'StaticAPConfigurator' in line: 208 static_config_failures.append(ssid) 209 else: 210 dynamic_config_failures.append(ssid) 211 else: 212 test_fail_aps.append(ssid) 213 elif ('END GOOD' in line and ('ChaosConnectDisconnect' in line or 214 'ChaosLongConnect' in line)): 215 test_name = line.split()[2] 216 ssid = test_name.split('.')[1] 217 network_dict = self.get_ap_mode_chan_freq(ssid) 218 modes.append(network_dict['mode']) 219 channels.append(network_dict['channel']) 220 else: 221 continue 222 223 config_pass = total - (len(dynamic_config_failures) + 224 len(static_config_failures)) 225 config_pass_string = self.generate_percentage_string(config_pass, 226 total) 227 connect_pass = config_pass - len(test_fail_aps) 228 connect_pass_string = self.generate_percentage_string(connect_pass, 229 config_pass) 230 231 base_csv_list = [board, os_version, fw_version, kernel_version, 232 self.convert_set_to_string(set(modes)), 233 self.convert_set_to_string(set(channels)), 234 security] 235 236 static_config_csv_list = copy.deepcopy(base_csv_list) 237 static_config_csv_list.append(config_pass_string) 238 static_config_csv_list.extend(static_config_failures) 239 240 dynamic_config_csv_list = copy.deepcopy(base_csv_list) 241 dynamic_config_csv_list.append(config_pass_string) 242 dynamic_config_csv_list.extend(dynamic_config_failures) 243 244 connect_csv_list = copy.deepcopy(base_csv_list) 245 connect_csv_list.append(connect_pass_string) 246 connect_csv_list.extend(test_fail_aps) 247 248 print('Connect failure for security: %s' % security) 249 print(','.join(connect_csv_list)) 250 print('\n') 251 252 if self._print_config_failures: 253 config_files = [('Static', static_config_csv_list), 254 ('Dynamic', dynamic_config_csv_list)] 255 for config_data in config_files: 256 self.print_config_failures(config_data[0], security, 257 config_data[1]) 258 259 if self._create_file: 260 self.create_csv('chaos_WiFi_dynamic_config_fail.' + security, 261 dynamic_config_csv_list) 262 self.create_csv('chaos_WiFi_static_config_fail.' + security, 263 static_config_csv_list) 264 self.create_csv('chaos_WiFi_connect_fail.' + security, 265 connect_csv_list) 266 267 268 def print_config_failures(self, config_type, security, config_csv_list): 269 """Prints out the configuration failures. 270 271 @param config_type: string describing the configurator type 272 @param security: the security type as a string 273 @param config_csv_list: list of the configuration failures 274 275 """ 276 # 8 because that is the lenth of the base list 277 if len(config_csv_list) <= 8: 278 return 279 print('%s config failures for security: %s' % (config_type, security)) 280 print(','.join(config_csv_list)) 281 print('\n') 282 283 284 def traverse_results_dir(self, path): 285 """Walks through the results directory and get the pathnames for the 286 status.log and the keyval files. 287 288 @param path: complete path to a specific test result directory. 289 290 @returns a dict with absolute pathnames for the 'status.log' and 291 'keyfile' files. 292 293 """ 294 status = None 295 keyval = None 296 297 for root, dir_name, file_name in os.walk(path): 298 for name in file_name: 299 current_path = os.path.join(root, name) 300 if name == 'status.log' and not status: 301 status = current_path 302 elif name == 'keyval' and ('param-debug_info' in 303 open(current_path).read()): 304 # This is a keyval file for a single test and not a suite. 305 keyval = os.path.join(root, name) 306 break 307 else: 308 continue 309 if not keyval: 310 raise Exception('Did Chaos tests complete successfully? Rerun tests' 311 ' with missing results.') 312 return {'status_file': status, 'keyval_file': keyval} 313 314 315 def parse_results_dir(self): 316 """Parses each result directory. 317 318 For each results directory created by test_that, parse it and 319 create summary files. 320 321 """ 322 if os.path.exists(RESULTS_DIR): 323 shutil.rmtree(RESULTS_DIR) 324 test_processed = False 325 for results_dir in os.listdir(self._test_results_dir): 326 if 'results' in results_dir: 327 path = os.path.join(self._test_results_dir, results_dir) 328 test = results_dir.split('.')[1] 329 status_key_dict = self.traverse_results_dir(path) 330 status_log_path = status_key_dict['status_file'] 331 lsb_info = self.parse_keyval(status_key_dict['keyval_file']) 332 if test is not None: 333 self.parse_status_log(lsb_info['board'], 334 lsb_info['version'], 335 test, 336 status_log_path) 337 test_processed = True 338 if not test_processed: 339 raise RuntimeError('chaos_parse: Did not find any results directory' 340 'to process') 341 342 343def main(): 344 """Main function to call the parser.""" 345 logging.basicConfig(level=logging.INFO) 346 arg_parser = argparse.ArgumentParser() 347 arg_parser.add_argument('-d', '--directory', dest='dir_name', 348 help='Pathname to results generated by test_that', 349 required=True) 350 arg_parser.add_argument('--create_file', dest='create_file', 351 action='store_true', default=False) 352 arg_parser.add_argument('--print_config_failures', 353 dest='print_config_failures', 354 action='store_true', 355 default=False) 356 arguments = arg_parser.parse_args() 357 if not arguments.dir_name: 358 raise RuntimeError('chaos_parser: No directory name supplied. Use -h' 359 ' for help') 360 if not os.path.exists(arguments.dir_name): 361 raise RuntimeError('chaos_parser: Invalid directory name supplied.') 362 parser = ChaosParser(arguments.dir_name, arguments.create_file, 363 arguments.print_config_failures) 364 parser.parse_results_dir() 365 366 367if __name__ == '__main__': 368 main() 369