<s> import collections <EOL> import functools <EOL> import json <EOL> import jinja2 <EOL> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> from subunit import v2 as subunit_v2 <EOL> import yaml <EOL> from shaker . engine import aggregators <EOL> from shaker . engine import config <EOL> from shaker . engine import sla <EOL> from shaker . engine import utils <EOL> from shaker . engine import writer <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def calculate_stats ( records , tests ) : <EOL> rec_map = collections . defaultdict ( <EOL> functools . partial ( collections . defaultdict , <EOL> functools . partial ( collections . defaultdict , list ) ) ) <EOL> for record in records . values ( ) : <EOL> if '<STR_LIT:test>' not in record : <EOL> continue <EOL> aggregator = aggregators . get_aggregator ( tests [ record [ '<STR_LIT:test>' ] ] ) <EOL> aggregator . record_summary ( record ) <EOL> rec_map [ record [ '<STR_LIT>' ] ] [ record [ '<STR_LIT:test>' ] ] [ <EOL> record [ '<STR_LIT>' ] ] . append ( record ) <EOL> for scenario , per_scenario in rec_map . items ( ) : <EOL> for test , per_test in per_scenario . items ( ) : <EOL> aggregator = aggregators . get_aggregator ( tests [ test ] ) <EOL> concurrency_aggregates = [ ] <EOL> for concurrency , per_concurrency in per_test . items ( ) : <EOL> summary = aggregator . concurrency_summary ( per_concurrency ) <EOL> if summary : <EOL> record_id = utils . make_record_id ( ) <EOL> summary . update ( dict ( id = record_id , <EOL> scenario = scenario , test = test , <EOL> concurrency = concurrency , <EOL> type = '<STR_LIT>' ) ) <EOL> records [ record_id ] = summary <EOL> concurrency_aggregates . append ( summary ) <EOL> per_test_summary = aggregator . test_summary ( concurrency_aggregates ) <EOL> if per_test_summary : <EOL> record_id = utils . make_record_id ( ) <EOL> per_test_summary . update ( dict ( id = record_id , <EOL> scenario = scenario , test = test , <EOL> type = '<STR_LIT:test>' ) ) <EOL> records [ record_id ] = per_test_summary <EOL> def verify_sla ( records , tests ) : <EOL> record_map = collections . defaultdict ( list ) <EOL> for r in records . values ( ) : <EOL> if ( '<STR_LIT:test>' in r ) and ( '<STR_LIT>' in tests [ r [ '<STR_LIT:test>' ] ] ) : <EOL> record_map [ r [ '<STR_LIT:test>' ] ] . append ( r ) <EOL> sla_records = [ ] <EOL> for test_id , records_per_test in record_map . items ( ) : <EOL> for sla_expr in tests [ test_id ] [ '<STR_LIT>' ] : <EOL> sla_records += sla . eval_expr ( sla_expr , records_per_test ) <EOL> return sla_records <EOL> def log_sla ( sla_records ) : <EOL> if sla_records : <EOL> LOG . info ( '<STR_LIT:*>' * <NUM_LIT> ) <EOL> for item in sla_records : <EOL> test_id = _get_location ( item . record ) + '<STR_LIT::>' + item . expression <EOL> LOG . info ( '<STR_LIT>' % ( test_id , '<STR_LIT>' % item . state ) ) <EOL> LOG . info ( '<STR_LIT:*>' * <NUM_LIT> ) <EOL> def output_sla ( sla_records ) : <EOL> log_sla ( sla_records ) <EOL> return [ dict ( record = item . record [ '<STR_LIT:id>' ] , state = item . state , <EOL> expression = item . expression ) for item in sla_records ] <EOL> def _get_location ( record ) : <EOL> return '<STR_LIT:.>' . join ( [ str ( record . get ( s ) ) <EOL> for s in [ '<STR_LIT>' , '<STR_LIT:test>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] if record . get ( s ) ] ) <EOL> def save_to_subunit ( sla_records , subunit_filename ) : <EOL> LOG . debug ( '<STR_LIT>' , subunit_filename ) <EOL> fd = None <EOL> state2subunit = { sla . STATE_TRUE : '<STR_LIT:success>' , <EOL> sla . STATE_FALSE : '<STR_LIT>' } <EOL> try : <EOL> fd = open ( subunit_filename , '<STR_LIT:w>' ) <EOL> output = subunit_v2 . StreamResultToBytes ( fd ) <EOL> for item in sla_records : <EOL> output . startTestRun ( ) <EOL> test_id = _get_location ( item . record ) + '<STR_LIT::>' + item . expression <EOL> if item . state != sla . STATE_TRUE : <EOL> output . status ( test_id = test_id , file_name = '<STR_LIT>' , <EOL> mime_type = '<STR_LIT>' , eof = True , <EOL> file_bytes = yaml . safe_dump ( <EOL> item . record , default_flow_style = False ) ) <EOL> output . status ( test_id = test_id , <EOL> test_status = state2subunit . get ( item . state , '<STR_LIT>' ) ) <EOL> output . stopTestRun ( ) <EOL> LOG . info ( '<STR_LIT>' , subunit_filename ) <EOL> except IOError as e : <EOL> LOG . error ( '<STR_LIT>' , e ) <EOL> finally : <EOL> if fd : <EOL> fd . close ( ) <EOL> def generate_report ( data , report_template , report_filename , subunit_filename , <EOL> book_folder = None ) : <EOL> LOG . debug ( '<STR_LIT>' , <EOL> report_template , report_filename or '<STR_LIT>' ) <EOL> calculate_stats ( data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] ) <EOL> sla_records = verify_sla ( data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] ) <EOL> data [ '<STR_LIT>' ] = output_sla ( sla_records ) <EOL> if subunit_filename : <EOL> save_to_subunit ( sla_records , subunit_filename ) <EOL> jinja_env = jinja2 . Environment ( variable_start_string = '<STR_LIT>' , <EOL> variable_end_string = '<STR_LIT>' , <EOL> comment_start_string = '<STR_LIT>' , <EOL> comment_end_string = '<STR_LIT>' ) <EOL> jinja_env . filters [ '<STR_LIT>' ] = json . dumps <EOL> jinja_env . filters [ '<STR_LIT>' ] = functools . partial ( yaml . safe_dump , indent = <NUM_LIT:2> , <EOL> default_flow_style = False ) <EOL> alias_mapper = lambda f : config . REPORT_TEMPLATES + '<STR_LIT>' % f <EOL> template = utils . read_file ( report_template , alias_mapper = alias_mapper ) <EOL> compiled_template = jinja_env . from_string ( template ) <EOL> rendered_template = compiled_template . render ( dict ( report = data ) ) <EOL> if report_filename : <EOL> LOG . debug ( '<STR_LIT>' , report_filename ) <EOL> try : <EOL> utils . write_file ( rendered_template , report_filename ) <EOL> LOG . info ( '<STR_LIT>' , report_filename ) <EOL> except IOError as e : <EOL> LOG . error ( '<STR_LIT>' , e ) <EOL> if book_folder : <EOL> writer . write_book ( book_folder , data ) <EOL> def main ( ) : <EOL> utils . init_config_and_logging ( config . REPORT_OPTS + config . INPUT_OPTS ) <EOL> LOG . debug ( '<STR_LIT>' , cfg . CONF . input ) <EOL> report_data = json . loads ( utils . read_file ( cfg . CONF . input ) ) <EOL> generate_report ( report_data , cfg . CONF . report_template , cfg . CONF . report , <EOL> cfg . CONF . subunit , cfg . CONF . book ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> main ( ) </s>
