<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from behave import given , when , then , step <EOL> from behave4cmd0 . command_steps import step_file_should_contain_multiline_text , step_file_should_not_contain_multiline_text <EOL> from behave . configuration import LogLevel <EOL> from behave . log_capture import LoggingCapture <EOL> import logging <EOL> def make_log_record ( category , level , message ) : <EOL> if category in ( "<STR_LIT:root>" , "<STR_LIT>" ) : <EOL> category = None <EOL> logger = logging . getLogger ( category ) <EOL> logger . log ( level , message ) <EOL> def make_log_record_output ( category , level , message , <EOL> format = None , datefmt = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not category or ( category == "<STR_LIT>" ) : <EOL> category = "<STR_LIT:root>" <EOL> levelname = logging . getLevelName ( level ) <EOL> record_data = dict ( name = category , levelname = levelname , msg = message ) <EOL> record_data . update ( kwargs ) <EOL> record = logging . makeLogRecord ( record_data ) <EOL> formatter = logging . Formatter ( format , datefmt = datefmt ) <EOL> return formatter . format ( record ) <EOL> class LogRecordTable ( object ) : <EOL> @ classmethod <EOL> def make_output_for_row ( cls , row , format = None , datefmt = None , ** kwargs ) : <EOL> category = row . get ( "<STR_LIT>" , None ) <EOL> level = LogLevel . parse_type ( row . get ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> message = row . get ( "<STR_LIT:message>" , "<STR_LIT>" ) <EOL> return make_log_record_output ( category , level , message , <EOL> format , datefmt , ** kwargs ) <EOL> @ staticmethod <EOL> def annotate_with_row_schema ( table , row_schema ) : <EOL> """<STR_LIT>""" <EOL> for column , value in row_schema . items ( ) : <EOL> if column not in table . headings : <EOL> table . add_column ( column , default_value = value ) <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_I_create_logrecords_with_table ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> for row in context . table . rows : <EOL> category = row [ "<STR_LIT>" ] <EOL> if category == "<STR_LIT>" : <EOL> category = None <EOL> level = LogLevel . parse_type ( row [ "<STR_LIT>" ] ) <EOL> message = row [ "<STR_LIT:message>" ] <EOL> make_log_record ( category , level , message ) <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_I_create_logrecord_with_table ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> assert len ( context . table . rows ) == <NUM_LIT:1> , "<STR_LIT>" <EOL> step_I_create_logrecords_with_table ( context ) <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_I_define_logrecord_schema_with_table ( context ) : <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> assert len ( context . table . rows ) == <NUM_LIT:1> , "<STR_LIT>" % ( len ( context . table . rows ) ) <EOL> row = context . table . rows [ <NUM_LIT:0> ] <EOL> row_schema = dict ( category = row [ "<STR_LIT>" ] , level = row [ "<STR_LIT>" ] , <EOL> message = row [ "<STR_LIT:message>" ] ) <EOL> context . log_record_row_schema = row_schema <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_command_output_should_contain_log_records ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> format = getattr ( context , "<STR_LIT>" , context . config . logging_format ) <EOL> for row in context . table . rows : <EOL> output = LogRecordTable . make_output_for_row ( row , format ) <EOL> context . execute_steps ( u'''<STR_LIT>''' . format ( expected_output = output ) ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_command_output_should_not_contain_log_records ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> format = getattr ( context , "<STR_LIT>" , context . config . logging_format ) <EOL> for row in context . table . rows : <EOL> output = LogRecordTable . make_output_for_row ( row , format ) <EOL> context . execute_steps ( u'''<STR_LIT>''' . format ( expected_output = output ) ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_command_output_should_contain_log_record ( context ) : <EOL> assert context . table , "<STR_LIT>" <EOL> assert len ( context . table . rows ) == <NUM_LIT:1> , "<STR_LIT>" <EOL> step_command_output_should_contain_log_records ( context ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_command_output_should_not_contain_log_record ( context ) : <EOL> assert context . table , "<STR_LIT>" <EOL> assert len ( context . table . rows ) == <NUM_LIT:1> , "<STR_LIT>" <EOL> step_command_output_should_not_contain_log_records ( context ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_command_output_should_contain_log_records_from_categories ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_column ( "<STR_LIT>" ) <EOL> record_schema = context . log_record_row_schema <EOL> LogRecordTable . annotate_with_row_schema ( context . table , record_schema ) <EOL> step_command_output_should_contain_log_records ( context ) <EOL> context . table . remove_columns ( [ "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_command_output_should_not_contain_log_records_from_categories ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_column ( "<STR_LIT>" ) <EOL> record_schema = context . log_record_row_schema <EOL> LogRecordTable . annotate_with_row_schema ( context . table , record_schema ) <EOL> step_command_output_should_not_contain_log_records ( context ) <EOL> context . table . remove_columns ( [ "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_file_should_contain_log_records ( context , filename ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> format = getattr ( context , "<STR_LIT>" , context . config . logging_format ) <EOL> for row in context . table . rows : <EOL> output = LogRecordTable . make_output_for_row ( row , format ) <EOL> context . text = output <EOL> step_file_should_contain_multiline_text ( context , filename ) <EOL> @ then ( '<STR_LIT>' ) <EOL> def step_file_should_not_contain_log_records ( context , filename ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:message>" ] ) <EOL> format = getattr ( context , "<STR_LIT>" , context . config . logging_format ) <EOL> for row in context . table . rows : <EOL> output = LogRecordTable . make_output_for_row ( row , format ) <EOL> context . text = output <EOL> step_file_should_not_contain_multiline_text ( context , filename ) <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_use_log_record_format_text ( context , log_record_format ) : <EOL> context . log_record_format = log_record_format <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_use_log_record_configuration ( context ) : <EOL> """<STR_LIT>""" <EOL> assert context . table , "<STR_LIT>" <EOL> context . table . require_columns ( [ "<STR_LIT>" , "<STR_LIT:value>" ] ) <EOL> for row in context . table . rows : <EOL> property_name = row [ "<STR_LIT>" ] <EOL> value = row [ "<STR_LIT:value>" ] <EOL> if property_name == "<STR_LIT>" : <EOL> context . log_record_format = value <EOL> elif property_name == "<STR_LIT>" : <EOL> context . log_record_datefmt = value <EOL> else : <EOL> raise KeyError ( "<STR_LIT>" % property_name ) <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_I_capture_logrecords ( context , level ) : <EOL> raise NotImplementedError ( ) <EOL> @ step ( '<STR_LIT>' ) <EOL> def step_I_capture_logrecords ( context ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> logcapture = getattr ( context , "<STR_LIT>" , None ) <EOL> if not logcapture : <EOL> context . logcapture = LoggingCapture ( ) </s>
