# Copyright 2023 The Pigweed Authors # # Licensed under the Apache License, Version 2.0 (the "License"); you may not # use this file except in compliance with the License. You may obtain a copy of # the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations under # the License. """Log decoder tests.""" from dataclasses import dataclass import logging from random import randint from typing import Any from unittest import TestCase, main from pw_log.log_decoder import ( Log, LogStreamDecoder, log_decoded_log, pw_status_code_to_name, timestamp_parser_ns_since_boot, ) from pw_log.proto import log_pb2 import pw_tokenizer _MESSAGE_NO_FILENAME = 'World' _MESSAGE_AND_ARGS_NO_FILENAME = f'■msg♦{_MESSAGE_NO_FILENAME}' _MESSAGE_TOKEN_NO_FILENAME = pw_tokenizer.tokens.pw_tokenizer_65599_hash( _MESSAGE_AND_ARGS_NO_FILENAME ) # Creating a database with tokenized information for the core to detokenize # tokenized log entries. _TOKEN_DATABASE = pw_tokenizer.tokens.Database( [ pw_tokenizer.tokens.TokenizedStringEntry(0x01148A48, 'total_dropped'), pw_tokenizer.tokens.TokenizedStringEntry( 0x03796798, 'min_queue_remaining' ), pw_tokenizer.tokens.TokenizedStringEntry(0x2E668CD6, 'Jello, world!'), pw_tokenizer.tokens.TokenizedStringEntry(0x329481A2, 'parser_errors'), pw_tokenizer.tokens.TokenizedStringEntry(0x7F35A9A5, 'TestName'), pw_tokenizer.tokens.TokenizedStringEntry(0xCC6D3131, 'Jello?'), pw_tokenizer.tokens.TokenizedStringEntry( 0x144C501D, '■msg♦SampleMessage■module♦MODULE■file♦file/path.cc' ), pw_tokenizer.tokens.TokenizedStringEntry(0x0000106A, 'ModuleOrMessage'), pw_tokenizer.tokens.TokenizedStringEntry( pw_tokenizer.tokens.pw_tokenizer_65599_hash( '■msg♦World■module♦wifi■file♦/path/to/file.cc' ), '■msg♦World■module♦wifi■file♦/path/to/file.cc', ), pw_tokenizer.tokens.TokenizedStringEntry( _MESSAGE_TOKEN_NO_FILENAME, _MESSAGE_AND_ARGS_NO_FILENAME ), ] ) _DETOKENIZER = pw_tokenizer.Detokenizer(_TOKEN_DATABASE) def _create_log_entry_with_tokenized_fields( message: str, module: str, file: str, thread: str, line: int, level: int ) -> log_pb2.LogEntry: """Tokenizing tokenizable LogEntry fields to become a detoknized log.""" tokenized_message = pw_tokenizer.encode.encode_token_and_args( pw_tokenizer.tokens.pw_tokenizer_65599_hash(message) ) tokenized_module = pw_tokenizer.encode.encode_token_and_args( pw_tokenizer.tokens.pw_tokenizer_65599_hash(module) ) tokenized_file = pw_tokenizer.encode.encode_token_and_args( pw_tokenizer.tokens.pw_tokenizer_65599_hash(file) ) tokenized_thread = pw_tokenizer.encode.encode_token_and_args( pw_tokenizer.tokens.pw_tokenizer_65599_hash(thread) ) return log_pb2.LogEntry( message=tokenized_message, module=tokenized_module, file=tokenized_file, line_level=Log.pack_line_level(line, level), thread=tokenized_thread, ) def _create_random_log_entry() -> log_pb2.LogEntry: return log_pb2.LogEntry( message=bytes(f'message {randint(1,100)}'.encode('utf-8')), line_level=Log.pack_line_level( randint(0, 2000), randint(logging.DEBUG, logging.CRITICAL) ), file=b'main.cc', thread=bytes(f'thread {randint(1,5)}'.encode('utf-8')), ) def _create_drop_count_message_log_entry( drop_count: int, reason: str = '' ) -> log_pb2.LogEntry: log_entry = log_pb2.LogEntry(dropped=drop_count) if reason: log_entry.message = bytes(reason.encode('utf-8')) return log_entry class TestLogStreamDecoderBase(TestCase): """Base Test class for LogStreamDecoder.""" def setUp(self) -> None: """Set up logs decoder.""" def parse_pw_status(msg: str) -> str: return pw_status_code_to_name(msg) self.captured_logs: list[Log] = [] def decoded_log_handler(log: Log) -> None: self.captured_logs.append(log) self.decoder = LogStreamDecoder( decoded_log_handler=decoded_log_handler, detokenizer=_DETOKENIZER, source_name='source', timestamp_parser=timestamp_parser_ns_since_boot, message_parser=parse_pw_status, ) def _captured_logs_as_str(self) -> str: return '\n'.join(map(str, self.captured_logs)) class TestLogStreamDecoderDecodingFunctionality(TestLogStreamDecoderBase): """Tests LogStreamDecoder decoding functionality.""" def test_parse_log_entry_valid_non_tokenized(self) -> None: """Test that valid LogEntry protos are parsed correctly.""" expected_log = Log( message='Hello', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Hello', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual(result, expected_log) def test_parse_log_entry_valid_packed_message(self) -> None: """Test that valid LogEntry protos are parsed correctly.""" log_with_metadata_in_message = Log( message='World', file_and_line='/path/to/file.cc', level=logging.DEBUG, source_name=self.decoder.source_name, module_name='wifi', timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=bytes( '■msg♦World■module♦wifi■file♦/path/to/file.cc'.encode( 'utf-8' ) ), line_level=Log.pack_line_level(0, logging.DEBUG), timestamp=100, ) ) self.assertEqual(result, log_with_metadata_in_message) def test_parse_log_entry_valid_logs_drop_message(self) -> None: """Test that valid LogEntry protos are parsed correctly.""" dropped_message = Log( message='Dropped 30 logs due to buffer too small', level=logging.WARNING, source_name=self.decoder.source_name, ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry(message=b'buffer too small', dropped=30) ) self.assertEqual(result, dropped_message) def test_parse_log_entry_valid_tokenized(self) -> None: """Test that tokenized LogEntry protos are parsed correctly.""" message = 'Jello, world!' module_name = 'TestName' file = 'parser_errors' thread_name = 'Jello?' line = 123 level = logging.INFO expected_log = Log( message=message, module_name=module_name, file_and_line=file + ':123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', thread_name=thread_name, ) log_entry = _create_log_entry_with_tokenized_fields( message, module_name, file, thread_name, line, level ) result = self.decoder.parse_log_entry_proto(log_entry) self.assertEqual(result, expected_log, msg='Log was not detokenized') def test_tokenized_contents_not_detokenized(self): """Test fields with tokens not in the database are not decrypted.""" # The following strings do not have tokens in the device token db. message_not_in_db = 'device is shutting down.' module_name_not_in_db = 'Battery' file_not_in_db = 'charger.cc' thread_name_not_in_db = 'BatteryStatus' line = 123 level = logging.INFO log_entry = _create_log_entry_with_tokenized_fields( message_not_in_db, module_name_not_in_db, file_not_in_db, thread_name_not_in_db, line, level, ) message = pw_tokenizer.proto.decode_optionally_tokenized( _DETOKENIZER, log_entry.message ) module = pw_tokenizer.proto.decode_optionally_tokenized( _DETOKENIZER, log_entry.module ) file = pw_tokenizer.proto.decode_optionally_tokenized( _DETOKENIZER, log_entry.file ) thread = pw_tokenizer.proto.decode_optionally_tokenized( _DETOKENIZER, log_entry.thread ) expected_log = Log( message=message, module_name=module, file_and_line=file + ':123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', thread_name=thread, ) result = self.decoder.parse_log_entry_proto(log_entry) self.assertEqual( result, expected_log, msg='Log was unexpectedly detokenized' ) def test_extracting_log_entry_fields_from_tokenized_metadata(self): """Test that tokenized metadata can be used to extract other fields.""" metadata = '■msg♦World■module♦wifi■file♦/path/to/file.cc' thread_name = 'M0Log' log_entry = log_pb2.LogEntry( message=pw_tokenizer.encode.encode_token_and_args( pw_tokenizer.tokens.pw_tokenizer_65599_hash(metadata) ), line_level=Log.pack_line_level(0, logging.DEBUG), thread=bytes(thread_name.encode('utf-8')), ) log_with_metadata_in_message = Log( message='World', file_and_line='/path/to/file.cc', level=logging.DEBUG, source_name=self.decoder.source_name, module_name='wifi', timestamp='0:00', thread_name=thread_name, ) result = self.decoder.parse_log_entry_proto(log_entry) self.assertEqual( result, log_with_metadata_in_message, msg='Log was detokenized.' ) def test_extracting_status_argument_from_log_message(self): """Test extract status from log message.""" expected_log = Log( message='Could not start flux capacitor: PERMISSION_DENIED', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Status=7', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) expected_log = Log( message='Error connecting to server: UNAVAILABLE', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Error connecting to server: pw::Status=14', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_extracting_status_with_improper_spacing(self): """Test spaces before pw::Status are ignored.""" expected_log = Log( message='Error connecting to server:UNAVAILABLE', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Error connecting to server:pw::Status=14', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) expected_log = Log( message='Error connecting to server: UNAVAILABLE', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Error connecting to server: pw::Status=14', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_not_extracting_status_extra_space_before_code(self): """Test spaces after pw::Status are not allowed.""" expected_log = Log( message='Could not start flux capacitor: pw::Status= 7', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Status= 7', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_not_extracting_status_new_line_before_code(self): """Test new line characters after pw::Status are not allowed.""" expected_log = Log( message='Could not start flux capacitor: pw::Status=\n7', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Status=\n7', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_not_extracting_status_from_log_message_with_improper_format(self): """Test status not extracted from log message with incorrect format.""" expected_log = Log( message='Could not start flux capacitor: pw::Status12', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Status12', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_status_code_in_message_does_not_exist(self): """Test status does not exist in pw_status.""" expected_log = Log( message='Could not start flux capacitor: pw::Status=17', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Status=17', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_status_code_in_message_is_negative(self): """Test status code is negative.""" expected_log = Log( message='Could not start flux capacitor: pw::Status=-1', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Status=-1', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_status_code_is_name(self): """Test if the status code format includes the name instead.""" expected_log = Log( message='Cannot use flux capacitor: pw::Status=PERMISSION_DENIED', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=( b'Cannot use flux capacitor: pw::Status=PERMISSION_DENIED' ), file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_spelling_mistakes_with_status_keyword(self): """Test spelling mistakes with status keyword.""" expected_log = Log( message='Could not start flux capacitor: pw::Rtatus=12', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::Rtatus=12', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_spelling_mistakes_with_status_keyword_lowercase_s(self): """Test spelling mistakes with status keyword.""" expected_log = Log( message='Could not start flux capacitor: pw::status=13', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Could not start flux capacitor: pw::status=13', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_status_code_at_beginning_of_message(self): """Test embedded status argument is found.""" expected_log = Log( message='UNAVAILABLE to connect to server.', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'pw::Status=14 to connect to server.', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_status_code_in_the_middle_of_message(self): """Test embedded status argument is found.""" expected_log = Log( message='Connection error: UNAVAILABLE connecting to server.', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=( b'Connection error: pw::Status=14 connecting to server.' ), file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_status_code_with_no_surrounding_spaces(self): """Test embedded status argument is found.""" expected_log = Log( message='Connection error:UNAVAILABLEconnecting to server.', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Connection error:pw::Status=14connecting to server.', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_multiple_status_arguments_in_log_message(self): """Test replacement of multiple status arguments into status string.""" expected_log = Log( message='Connection error: UNAVAILABLE and PERMISSION_DENIED.', file_and_line='my/path/file.cc:123', level=logging.INFO, source_name=self.decoder.source_name, timestamp='0:00', ) result = self.decoder.parse_log_entry_proto( log_pb2.LogEntry( message=b'Connection error: pw::Status=14 and pw::Status=7.', file=b'my/path/file.cc', line_level=Log.pack_line_level(123, logging.INFO), ) ) self.assertEqual( result, expected_log, msg='Status was not extracted from log message.', ) def test_no_filename_in_message_parses_successfully(self): """Test that if the file name is not present the log entry is parsed.""" thread_name = 'thread' log_entry = log_pb2.LogEntry( message=pw_tokenizer.encode.encode_token_and_args( _MESSAGE_TOKEN_NO_FILENAME ), line_level=Log.pack_line_level(123, logging.DEBUG), thread=bytes(thread_name.encode('utf-8')), ) expected_log = Log( message=_MESSAGE_NO_FILENAME, file_and_line=':123', level=logging.DEBUG, source_name=self.decoder.source_name, timestamp='0:00', thread_name=thread_name, ) result = self.decoder.parse_log_entry_proto(log_entry) self.assertEqual(result, expected_log) def test_log_decoded_log(self): """Test that the logger correctly formats a decoded log.""" test_log = Log( message="SampleMessage", level=logging.DEBUG, timestamp='1:30', module_name="MyModule", source_name="MySource", thread_name="MyThread", file_and_line='my_file.cc:123', metadata_fields={'field1': 432, 'field2': 'value'}, ) class CapturingLogger(logging.Logger): """Captures values passed to log(). Tests that calls to log() have the correct level, extra arguments and the message format string and arguments match. """ @dataclass(frozen=True) class LoggerLog: """Represents a process log() call.""" level: int message: str kwargs: Any def __init__(self): super().__init__(name="CapturingLogger") self.log_calls: list[CapturingLogger.LoggerLog] = [] def log(self, level, msg, *args, **kwargs) -> None: log = CapturingLogger.LoggerLog( level=level, message=msg % args, kwargs=kwargs ) self.log_calls.append(log) test_logger = CapturingLogger() log_decoded_log(test_log, test_logger) self.assertEqual(len(test_logger.log_calls), 1) self.assertEqual(test_logger.log_calls[0].level, test_log.level) self.assertEqual( test_logger.log_calls[0].message, '[%s] %s %s %s %s' % ( test_log.source_name, test_log.module_name, test_log.timestamp, test_log.message, test_log.file_and_line, ), ) self.assertEqual( test_logger.log_calls[0].kwargs['extra']['extra_metadata_fields'], test_log.metadata_fields, ) class TestLogStreamDecoderLogDropDetectionFunctionality( TestLogStreamDecoderBase ): """Tests LogStreamDecoder log drop detection functionality.""" def test_log_drops_transport_error(self): """Tests log drops at transport.""" log_entry_in_log_entries_1 = 3 log_entries_1 = log_pb2.LogEntries( first_entry_sequence_id=0, entries=[ _create_random_log_entry() for _ in range(log_entry_in_log_entries_1) ], ) self.decoder.parse_log_entries_proto(log_entries_1) # Assume a second LogEntries was dropped with 5 log entries. I.e. a # log_entries_2 with sequence_ie = 4 and 5 log entries. log_entry_in_log_entries_2 = 5 log_entry_in_log_entries_3 = 3 log_entries_3 = log_pb2.LogEntries( first_entry_sequence_id=log_entry_in_log_entries_2 + log_entry_in_log_entries_3, entries=[ _create_random_log_entry() for _ in range(log_entry_in_log_entries_3) ], ) self.decoder.parse_log_entries_proto(log_entries_3) # The drop message is placed where the dropped logs were detected. self.assertEqual( len(self.captured_logs), log_entry_in_log_entries_1 + 1 + log_entry_in_log_entries_3, msg=( 'Unexpected number of messages received: ' f'{self._captured_logs_as_str()}' ), ) self.assertEqual( ( f'Dropped {log_entry_in_log_entries_2} logs due to ' f'{LogStreamDecoder.DROP_REASON_LOSS_AT_TRANSPORT}' ), self.captured_logs[log_entry_in_log_entries_1].message, ) def test_log_drops_source_not_connected(self): """Tests log drops when source of the logs was not connected.""" log_entry_in_log_entries = 4 drop_count = 7 log_entries = log_pb2.LogEntries( first_entry_sequence_id=drop_count, entries=[ _create_random_log_entry() for _ in range(log_entry_in_log_entries) ], ) self.decoder.parse_log_entries_proto(log_entries) # The drop message is placed where the log drops was detected. self.assertEqual( len(self.captured_logs), 1 + log_entry_in_log_entries, msg=( 'Unexpected number of messages received: ' f'{self._captured_logs_as_str()}' ), ) self.assertEqual( ( f'Dropped {drop_count} logs due to ' f'{LogStreamDecoder.DROP_REASON_SOURCE_NOT_CONNECTED}' ), self.captured_logs[0].message, ) def test_log_drops_source_enqueue_failure_no_message(self): """Tests log drops when source reports log drops.""" drop_count = 5 log_entries = log_pb2.LogEntries( first_entry_sequence_id=0, entries=[ _create_random_log_entry(), _create_random_log_entry(), _create_drop_count_message_log_entry(drop_count), _create_random_log_entry(), ], ) self.decoder.parse_log_entries_proto(log_entries) # The drop message is placed where the log drops was detected. self.assertEqual( len(self.captured_logs), 4, msg=( 'Unexpected number of messages received: ' f'{self._captured_logs_as_str()}' ), ) self.assertEqual( ( f'Dropped {drop_count} logs due to ' f'{LogStreamDecoder.DROP_REASON_SOURCE_ENQUEUE_FAILURE}' ), self.captured_logs[2].message, ) def test_log_drops_source_enqueue_failure_with_message(self): """Tests log drops when source reports log drops.""" drop_count = 8 reason = 'Flux Capacitor exploded' log_entries = log_pb2.LogEntries( first_entry_sequence_id=0, entries=[ _create_random_log_entry(), _create_random_log_entry(), _create_drop_count_message_log_entry(drop_count, reason), _create_random_log_entry(), ], ) self.decoder.parse_log_entries_proto(log_entries) # The drop message is placed where the log drops was detected. self.assertEqual( len(self.captured_logs), 4, msg=( 'Unexpected number of messages received: ' f'{self._captured_logs_as_str()}' ), ) self.assertEqual( f'Dropped {drop_count} logs due to {reason.lower()}', self.captured_logs[2].message, ) if __name__ == '__main__': main()