Best Python code snippet using autotest_python
results_parser.py
Source:results_parser.py
1import os2import os.path as osp3import logging4import shutil5import re6from .logger import LogTaskStatus7from . import utillib8from . import confreader9from .utillib import FileNotFoundException10def just_parse(input_root_dir, output_root_dir):11 results_conf_file = osp.join(input_root_dir, 'results.conf')12 results_conf = confreader.read_conf_into_dict(results_conf_file)13 if int(results_conf['exit-code']) != 0:14 return int(results_conf['exit-code'])15 results_archive = osp.join(input_root_dir, results_conf['results-archive'])16 cwd = os.getcwd()17 with LogTaskStatus('results-unarchive'):18 status = utillib.unpack_archive(results_archive, cwd)19 if status != 0:20 return status21 results_root_dir = osp.join(cwd, results_conf['results-dir'])22 assessment_summary_file = osp.join(results_root_dir,23 results_conf['assessment-summary-file'])24 return parse_results(input_root_dir,25 assessment_summary_file,26 results_root_dir,27 output_root_dir)28def _get_results_parser(input_dir):29 with LogTaskStatus('resultparser-unarchive'):30 parser_dir = osp.join(os.getcwd(), 'result-parser')31 if not osp.isdir(parser_dir):32 os.mkdir(parser_dir)33 parser_conf_file = osp.join(input_dir, 'resultparser.conf')34 if not osp.isfile(parser_conf_file):35 raise FileNotFoundException(parser_conf_file)36 parser_attr = confreader.read_conf_into_dict(parser_conf_file)37 logging.info('RESULTS PARSER CONF: ' + str(parser_attr))38 parser_archive = osp.join(input_dir, parser_attr['result-parser-archive'])39 utillib.unpack_archive(parser_archive, parser_dir)40 parser_dir = osp.join(parser_dir, parser_attr['result-parser-dir'])41 parser_exe_file = osp.join(parser_dir, parser_attr['result-parser-cmd'])42 return parser_exe_file43def read_task_info_file(weakness_count_file):44 # STATUS_DICT = {'PASS', 'FAIL', 'SKIP', 'NOTE'}45 short_msg = ''46 status = 'PASS'47 long_msg = ''48 with open(weakness_count_file) as fobj:49 short_msg = fobj.readline().strip()50 if short_msg:51 regex_sep = re.compile(r'^-+$')52 next_line = fobj.readline().strip()53 if next_line:54 if regex_sep.match(next_line) is None:55 status = next_line56 next_line = fobj.readline().strip()57 else:58 status = 'PASS'59 if next_line:60 if regex_sep.match(next_line):61 long_msg = ''.join([line for line in fobj])62 else:63 raise Exception("Invalid long message separator '%s' in '%s'" %64 (next_line, weakness_count_file))65 return (short_msg, status, long_msg)66def parse_results(input_dir, assessment_summary_file, results_dir, output_dir):67 command_template = '{EXECUTABLE}\68 --summary_file={PATH_TO_SUMMARY_FILE}\69 --input_dir={PATH_TO_RESULTS_DIR}\70 --output_file={OUTPUT_FILENAME}\71 --weakness_count_file={WEAKNESS_COUNT_FILENAME}\72 --parsed_results_data_conf_file={PARSED_RESULTS_DATA_CONF_FILE}'73 if not osp.isfile(assessment_summary_file):74 raise FileNotFoundException(assessment_summary_file)75 parser_exe_file = _get_results_parser(input_dir)76 services_conf_file = osp.join(input_dir, 'services.conf')77 if osp.isfile(services_conf_file):78 command_template += ' --services_conf_file={SERVICES_CONF_FILE}'79 parse_results_dir = osp.join(os.getcwd(), 'parsed_results')80 if not osp.isdir(parse_results_dir):81 os.mkdir(parse_results_dir)82 parsed_results_data_conf_file = osp.join(parse_results_dir, 'parsed_results_data.conf')83 try:84 parse_results_logfile = osp.join(parse_results_dir, 'resultparser.log')85 parse_results_output_file = osp.join(parse_results_dir, 'parsed_results.xml')86 parse_weakness_count_file = osp.join(parse_results_dir, 'weakness_count.out')87 stdout_filename = 'resultparser_stdout.out'88 stderr_filename = 'resultparser_stderr.out'89 resultparser_stdout_file = osp.join(parse_results_dir, stdout_filename)90 resultparser_stderr_file = osp.join(parse_results_dir, stderr_filename)91 with LogTaskStatus('parse-results') as status_dot_out:92 if 'PERL5LIB' in os.environ:93 os.environ['PERL5LIB'] = '${0}:{1}'.format(os.environ['PERL5LIB'],94 osp.dirname(parser_exe_file))95 else:96 os.environ['PERL5LIB'] = osp.dirname(parser_exe_file)97 command = command_template.format(EXECUTABLE=parser_exe_file,98 PATH_TO_SUMMARY_FILE=assessment_summary_file,99 PATH_TO_RESULTS_DIR=results_dir,100 PATH_TO_OUTPUT_DIR=parse_results_dir,101 OUTPUT_FILENAME=parse_results_output_file,102 WEAKNESS_COUNT_FILENAME=parse_weakness_count_file,103 SERVICES_CONF_FILE=services_conf_file,104 PARSED_RESULTS_DATA_CONF_FILE=parsed_results_data_conf_file,105 LOGFILE=parse_results_logfile)106 exit_code, _ = utillib.run_cmd(command,107 cwd=osp.dirname(parser_exe_file),108 description='PARSE RESULTS',109 outfile=resultparser_stdout_file,110 errfile=resultparser_stderr_file)111 short_msg = ''112 status = 'PASS'113 long_msg = ''114 if osp.isfile(parse_weakness_count_file):115 short_msg, status, long_msg = read_task_info_file(parse_weakness_count_file)116 else:117 status = 'FAIL'118 long_msg = "weakness count file ({0}) not found".format(parse_weakness_count_file)119 if status == 'SKIP':120 status_dot_out.skip_task(short_msg, long_msg)121 else:122 if (exit_code != 0):123 if long_msg:124 long_msg += "\n"125 long_msg += "Result Parser exit code {0}".format(exit_code)126 elif status == 'FAIL':127 exit_code = 1128 status_dot_out.update_task_status(exit_code, short_msg, long_msg)129 # NOTE: Not sure which part of the code will throw an exception. Handling exception map not be required130 except Exception as err:131 logging.exception(err)132 exit_code = 1133 finally:134 with LogTaskStatus('parsed-results-archive'):135 shutil.make_archive(osp.join(output_dir,136 osp.basename(parse_results_dir)),137 'gztar',138 osp.dirname(parse_results_dir),139 osp.basename(parse_results_dir))140 fileFound = osp.isfile(parsed_results_data_conf_file)141 if fileFound:142 parsed_results_conf = confreader.read_conf_into_dict(parsed_results_data_conf_file)143 parsed_results_conf['parsed-results-dir'] = osp.basename(parse_results_dir)144 parsed_results_conf['parsed-results-archive'] = '{0}.tar.gz'.format(osp.basename(parse_results_dir))145 parsed_results_conf['resultparser-stdout-file'] = stdout_filename146 parsed_results_conf['resultparser-stderr-file'] = stderr_filename147 utillib.write_to_file(osp.join(output_dir, 'parsed_results.conf'),148 parsed_results_conf)149 if not fileFound and exit_code == 0:150 raise Exception('parsed_results_data.conf file not found at {0}'.format(parsed_results_data_conf_file))151 if exit_code != 0:152 raise Exception('Result Parser Exit Code {0}'.format(exit_code))153 else:...
results-to-csv.py
Source:results-to-csv.py
...10else:11 OUTPUT = DIRECTORY + ".csv"12def print_line(version, w, s, i, time):13 return "%s,%s,%s,%s,%s\n" % (version, w, s, i, time)14def parse_results_dir(dir_name):15 out = "version,w,s,i,time (ms)\n"16 errors = []17 for f_name in os.listdir(dir_name):18 if f_name == "info.txt":19 continue20 match = re.search(r'(.+)-w(\d+)(?:-s(\d+))?-i(\d+).txt', f_name)21 if match is None:22 errors.append("Ignoring %s: wrong file name." % f_name)23 continue24 version, w, s, i = match.groups()25 with open(os.path.join(dir_name, f_name)) as f:26 contents = f.read()27 matches = re.findall(r'Total execution time: ([\d.]+) ms', contents)28 if len(matches) != 1:29 errors.append("File %s has %i time measurements, expected 1." % \30 (f_name, len(matches)))31 continue32 time = matches[0]33 out += print_line(version, w, s, i, time)34 return out, errors35out, errors = parse_results_dir(DIRECTORY)36with open(OUTPUT, "x") as f:37 f.write(out)38if len(errors) != 0:39 print("ERRORS:")...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!