<s> """<STR_LIT>""" <EOL> import copy <EOL> import csv <EOL> import os <EOL> import time <EOL> from collections import OrderedDict <EOL> from datetime import datetime <EOL> from bzt . engine import Reporter <EOL> from bzt . modules . aggregator import DataPoint , KPISet , AggregatorListener , ResultsProvider <EOL> from bzt . modules . blazemeter import BlazeMeterUploader , CloudProvisioning <EOL> from bzt . modules . passfail import PassFailStatus <EOL> from bzt . six import etree , iteritems , string_types <EOL> from bzt . utils import get_full_path <EOL> class FinalStatus ( Reporter , AggregatorListener ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( FinalStatus , self ) . __init__ ( ) <EOL> self . last_sec = None <EOL> self . start_time = time . time ( ) <EOL> self . end_time = None <EOL> def startup ( self ) : <EOL> self . start_time = time . time ( ) <EOL> def prepare ( self ) : <EOL> super ( FinalStatus , self ) . prepare ( ) <EOL> if isinstance ( self . engine . aggregator , ResultsProvider ) : <EOL> self . engine . aggregator . add_listener ( self ) <EOL> def aggregated_second ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . last_sec = data <EOL> def post_process ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( FinalStatus , self ) . post_process ( ) <EOL> self . end_time = time . time ( ) <EOL> if self . parameters . get ( "<STR_LIT>" , True ) : <EOL> self . __report_duration ( ) <EOL> if self . last_sec : <EOL> summary_kpi = self . last_sec [ DataPoint . CUMULATIVE ] [ "<STR_LIT>" ] <EOL> if self . parameters . get ( "<STR_LIT>" , True ) : <EOL> self . __report_samples_count ( summary_kpi ) <EOL> if self . parameters . get ( "<STR_LIT>" , True ) : <EOL> self . __report_percentiles ( summary_kpi ) <EOL> if self . parameters . get ( "<STR_LIT>" , False ) : <EOL> self . __report_failed_labels ( self . last_sec [ DataPoint . CUMULATIVE ] ) <EOL> if self . parameters . get ( "<STR_LIT>" , None ) : <EOL> self . __dump_xml ( self . parameters . get ( "<STR_LIT>" ) ) <EOL> if self . parameters . get ( "<STR_LIT>" , None ) : <EOL> self . __dump_csv ( self . parameters . get ( "<STR_LIT>" ) ) <EOL> def __report_samples_count ( self , summary_kpi_set ) : <EOL> """<STR_LIT>""" <EOL> if summary_kpi_set [ KPISet . SAMPLE_COUNT ] : <EOL> err_rate = <NUM_LIT:100> * summary_kpi_set [ KPISet . FAILURES ] / float ( summary_kpi_set [ KPISet . SAMPLE_COUNT ] ) <EOL> self . log . info ( "<STR_LIT>" , summary_kpi_set [ KPISet . SAMPLE_COUNT ] , err_rate ) <EOL> def __report_percentiles ( self , summary_kpi_set ) : <EOL> """<STR_LIT>""" <EOL> fmt = "<STR_LIT>" <EOL> self . log . info ( fmt , summary_kpi_set [ KPISet . AVG_RESP_TIME ] , summary_kpi_set [ KPISet . AVG_LATENCY ] , <EOL> summary_kpi_set [ KPISet . AVG_CONN_TIME ] ) <EOL> for key in sorted ( summary_kpi_set [ KPISet . PERCENTILES ] . keys ( ) , key = float ) : <EOL> self . log . info ( "<STR_LIT>" , float ( key ) , summary_kpi_set [ KPISet . PERCENTILES ] [ key ] ) <EOL> def __report_failed_labels ( self , cumulative ) : <EOL> """<STR_LIT>""" <EOL> report_template = "<STR_LIT>" <EOL> sorted_labels = sorted ( cumulative . keys ( ) ) <EOL> for sample_label in sorted_labels : <EOL> if sample_label != "<STR_LIT>" : <EOL> failed_samples_count = cumulative [ sample_label ] [ '<STR_LIT>' ] <EOL> if failed_samples_count : <EOL> self . log . info ( report_template , failed_samples_count , sample_label ) <EOL> def __report_duration ( self ) : <EOL> """<STR_LIT>""" <EOL> date_start = datetime . fromtimestamp ( int ( self . start_time ) ) <EOL> date_end = datetime . fromtimestamp ( int ( self . end_time ) ) <EOL> self . log . info ( "<STR_LIT>" , date_end - date_start ) <EOL> def __dump_xml ( self , filename ) : <EOL> self . log . info ( "<STR_LIT>" , filename ) <EOL> root = etree . Element ( "<STR_LIT>" ) <EOL> if self . last_sec : <EOL> for label , kpiset in iteritems ( self . last_sec [ DataPoint . CUMULATIVE ] ) : <EOL> root . append ( self . __get_xml_summary ( label , kpiset ) ) <EOL> with open ( get_full_path ( filename ) , '<STR_LIT:wb>' ) as fhd : <EOL> tree = etree . ElementTree ( root ) <EOL> tree . write ( fhd , pretty_print = True , encoding = "<STR_LIT>" , xml_declaration = True ) <EOL> def __get_xml_summary ( self , label , kpiset ) : <EOL> elem = etree . Element ( "<STR_LIT>" , label = label ) <EOL> for kpi_name , kpi_val in iteritems ( kpiset ) : <EOL> if kpi_name in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> continue <EOL> if isinstance ( kpi_val , dict ) : <EOL> for param_name , param_val in iteritems ( kpi_val ) : <EOL> elem . append ( self . __get_kpi_xml ( kpi_name , param_val , param_name ) ) <EOL> else : <EOL> elem . append ( self . __get_kpi_xml ( kpi_name , kpi_val ) ) <EOL> return elem <EOL> def __get_kpi_xml ( self , kpi_name , kpi_val , param = None ) : <EOL> kpi = etree . Element ( kpi_name ) <EOL> kpi . attrib [ '<STR_LIT:value>' ] = self . __val_to_str ( kpi_val ) <EOL> elm_name = etree . Element ( "<STR_LIT:name>" ) <EOL> elm_name . text = kpi_name <EOL> if param is not None : <EOL> kpi . attrib [ '<STR_LIT>' ] = self . __val_to_str ( param ) <EOL> elm_name . text += "<STR_LIT:/>" + param <EOL> kpi . append ( elm_name ) <EOL> elm_value = etree . Element ( "<STR_LIT:value>" ) <EOL> elm_value . text = self . __val_to_str ( kpi_val ) <EOL> kpi . append ( elm_value ) <EOL> return kpi <EOL> def __val_to_str ( self , kpi_val ) : <EOL> if isinstance ( kpi_val , float ) : <EOL> return '<STR_LIT>' % kpi_val <EOL> elif isinstance ( kpi_val , int ) : <EOL> return '<STR_LIT>' % kpi_val <EOL> elif isinstance ( kpi_val , string_types ) : <EOL> return kpi_val <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % type ( kpi_val ) ) <EOL> def __dump_csv ( self , filename ) : <EOL> self . log . info ( "<STR_LIT>" , filename ) <EOL> with open ( get_full_path ( filename ) , '<STR_LIT>' ) as fhd : <EOL> writer = csv . DictWriter ( fhd , self . __get_csv_dict ( '<STR_LIT>' , self . last_sec [ DataPoint . CUMULATIVE ] [ '<STR_LIT>' ] ) . keys ( ) ) <EOL> writer . writeheader ( ) <EOL> for label , kpiset in iteritems ( self . last_sec [ DataPoint . CUMULATIVE ] ) : <EOL> writer . writerow ( self . __get_csv_dict ( label , kpiset ) ) <EOL> def __get_csv_dict ( self , label , kpiset ) : <EOL> kpi_copy = copy . deepcopy ( kpiset ) <EOL> res = OrderedDict ( ) <EOL> for key in sorted ( kpi_copy . keys ( ) ) : <EOL> res [ key ] = kpi_copy [ key ] <EOL> for level , val in iteritems ( kpiset [ KPISet . PERCENTILES ] ) : <EOL> res [ '<STR_LIT>' % level ] = val <EOL> for rcd , val in iteritems ( kpiset [ KPISet . RESP_CODES ] ) : <EOL> res [ '<STR_LIT>' % rcd ] = val <EOL> for key in res : <EOL> if isinstance ( res [ key ] , float ) : <EOL> res [ key ] = "<STR_LIT>" % res [ key ] <EOL> del res [ '<STR_LIT>' ] <EOL> del res [ '<STR_LIT>' ] <EOL> del res [ '<STR_LIT>' ] <EOL> del res [ '<STR_LIT>' ] <EOL> res [ '<STR_LIT:label>' ] = label <EOL> return res <EOL> class JUnitXMLReporter ( Reporter , AggregatorListener ) : <EOL> """<STR_LIT>""" <EOL> REPORT_FILE_NAME = "<STR_LIT>" <EOL> REPORT_FILE_EXT = "<STR_LIT>" <EOL> def __init__ ( self ) : <EOL> super ( JUnitXMLReporter , self ) . __init__ ( ) <EOL> self . report_file_path = "<STR_LIT>" <EOL> self . last_second = None <EOL> def prepare ( self ) : <EOL> """<STR_LIT>""" <EOL> filename = self . parameters . get ( "<STR_LIT:filename>" , None ) <EOL> if filename : <EOL> self . report_file_path = filename <EOL> else : <EOL> self . report_file_path = self . engine . create_artifact ( JUnitXMLReporter . REPORT_FILE_NAME , <EOL> JUnitXMLReporter . REPORT_FILE_EXT ) <EOL> self . parameters [ "<STR_LIT:filename>" ] = self . report_file_path <EOL> if isinstance ( self . engine . aggregator , ResultsProvider ) : <EOL> self . engine . aggregator . add_listener ( self ) <EOL> def aggregated_second ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . last_second = data <EOL> def post_process ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( JUnitXMLReporter , self ) . post_process ( ) <EOL> test_data_source = self . parameters . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if test_data_source == "<STR_LIT>" : <EOL> if self . last_second : <EOL> root_element = self . process_sample_labels ( ) <EOL> self . save_report ( root_element ) <EOL> else : <EOL> self . log . warning ( "<STR_LIT>" ) <EOL> elif test_data_source == "<STR_LIT>" : <EOL> root_element = self . process_pass_fail ( ) <EOL> self . save_report ( root_element ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % test_data_source ) <EOL> def process_sample_labels ( self ) : <EOL> """<STR_LIT>""" <EOL> _kpiset = self . last_second [ DataPoint . CUMULATIVE ] <EOL> root_xml_element = etree . Element ( "<STR_LIT>" , name = "<STR_LIT>" , package = "<STR_LIT>" ) <EOL> bza_report_info = self . get_bza_report_info ( ) <EOL> class_name = bza_report_info [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] if bza_report_info else "<STR_LIT>" + str ( self . __hash__ ( ) ) <EOL> report_urls = [ info_item [ <NUM_LIT:0> ] for info_item in bza_report_info ] <EOL> for key in sorted ( _kpiset . keys ( ) ) : <EOL> if key == "<STR_LIT>" : <EOL> continue <EOL> test_case_etree_element = etree . Element ( "<STR_LIT>" , classname = class_name , name = key ) <EOL> if report_urls : <EOL> system_out_etree = etree . SubElement ( test_case_etree_element , "<STR_LIT>" ) <EOL> system_out_etree . text = "<STR_LIT>" . join ( report_urls ) <EOL> if _kpiset [ key ] [ KPISet . ERRORS ] : <EOL> for er_dict in _kpiset [ key ] [ KPISet . ERRORS ] : <EOL> err_message = str ( er_dict [ "<STR_LIT>" ] ) <EOL> err_type = str ( er_dict [ "<STR_LIT>" ] ) <EOL> err_desc = "<STR_LIT>" + str ( er_dict [ "<STR_LIT>" ] ) <EOL> err_etree_element = etree . Element ( "<STR_LIT:error>" , message = err_message , type = err_type ) <EOL> err_etree_element . text = err_desc <EOL> test_case_etree_element . append ( err_etree_element ) <EOL> root_xml_element . append ( test_case_etree_element ) <EOL> return root_xml_element <EOL> def get_bza_report_info ( self ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> if isinstance ( self . engine . provisioning , CloudProvisioning ) : <EOL> cloud_prov = self . engine . provisioning <EOL> report_url = "<STR_LIT>" % cloud_prov . client . results_url <EOL> test_name = cloud_prov . settings . get ( '<STR_LIT:test>' , None ) <EOL> result . append ( ( report_url , test_name if test_name is not None else report_url ) ) <EOL> else : <EOL> bza_reporters = [ _x for _x in self . engine . reporters if isinstance ( _x , BlazeMeterUploader ) ] <EOL> for bza_reporter in bza_reporters : <EOL> if bza_reporter . client . results_url : <EOL> report_url = "<STR_LIT>" % bza_reporter . client . results_url <EOL> test_name = bza_reporter . parameters . get ( "<STR_LIT:test>" , None ) <EOL> result . append ( ( report_url , test_name if test_name is not None else report_url ) ) <EOL> if len ( result ) > <NUM_LIT:1> : <EOL> self . log . warning ( "<STR_LIT>" ) <EOL> return result <EOL> def save_report ( self , root_node ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if os . path . exists ( self . report_file_path ) : <EOL> self . log . warning ( "<STR_LIT>" , self . report_file_path ) <EOL> else : <EOL> dirname = os . path . dirname ( self . report_file_path ) <EOL> if dirname and not os . path . exists ( dirname ) : <EOL> os . makedirs ( dirname ) <EOL> etree_obj = etree . ElementTree ( root_node ) <EOL> self . log . info ( "<STR_LIT>" , self . report_file_path ) <EOL> with open ( get_full_path ( self . report_file_path ) , '<STR_LIT:wb>' ) as _fds : <EOL> etree_obj . write ( _fds , xml_declaration = True , encoding = "<STR_LIT>" , pretty_print = True ) <EOL> except BaseException : <EOL> self . log . error ( "<STR_LIT>" , self . report_file_path ) <EOL> raise <EOL> def process_pass_fail ( self ) : <EOL> """<STR_LIT>""" <EOL> mods = self . engine . reporters + self . engine . services <EOL> pass_fail_objects = [ _x for _x in mods if isinstance ( _x , PassFailStatus ) ] <EOL> self . log . debug ( "<STR_LIT>" , pass_fail_objects ) <EOL> fail_criterias = [ ] <EOL> for pf_obj in pass_fail_objects : <EOL> if pf_obj . criterias : <EOL> for _fc in pf_obj . criterias : <EOL> fail_criterias . append ( _fc ) <EOL> root_xml_element = etree . Element ( "<STR_LIT>" , name = '<STR_LIT>' , package = "<STR_LIT>" ) <EOL> bza_report_info = self . get_bza_report_info ( ) <EOL> classname = bza_report_info [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] if bza_report_info else "<STR_LIT>" + str ( self . __hash__ ( ) ) <EOL> report_urls = [ info_item [ <NUM_LIT:0> ] for info_item in bza_report_info ] <EOL> for fc_obj in fail_criterias : <EOL> testcase_etree = self . __process_criteria ( classname , fc_obj , report_urls ) <EOL> root_xml_element . append ( testcase_etree ) <EOL> return root_xml_element <EOL> def __process_criteria ( self , classname , fc_obj , report_urls ) : <EOL> if fc_obj . config [ '<STR_LIT:label>' ] : <EOL> data = ( fc_obj . config [ '<STR_LIT>' ] , fc_obj . config [ '<STR_LIT:label>' ] , <EOL> fc_obj . config [ '<STR_LIT>' ] , fc_obj . config [ '<STR_LIT>' ] ) <EOL> tpl = "<STR_LIT>" <EOL> else : <EOL> data = ( fc_obj . config [ '<STR_LIT>' ] , fc_obj . config [ '<STR_LIT>' ] , <EOL> fc_obj . config [ '<STR_LIT>' ] ) <EOL> tpl = "<STR_LIT>" <EOL> if fc_obj . config [ '<STR_LIT>' ] : <EOL> tpl += "<STR_LIT>" <EOL> data += ( fc_obj . config [ '<STR_LIT>' ] , ) <EOL> disp_name = tpl % data <EOL> testcase_etree = etree . Element ( "<STR_LIT>" , classname = classname , name = disp_name ) <EOL> if report_urls : <EOL> system_out_etree = etree . SubElement ( testcase_etree , "<STR_LIT>" ) <EOL> system_out_etree . text = "<STR_LIT>" . join ( report_urls ) <EOL> if fc_obj . is_triggered and fc_obj . fail : <EOL> etree . SubElement ( testcase_etree , "<STR_LIT:error>" , type = "<STR_LIT>" , message = "<STR_LIT>" ) <EOL> return testcase_etree </s>
