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!!
