KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use put_log_events method in localstack

Best Python code snippet using localstack_python

cloudwatch_client.py

Source:cloudwatch_client.py

...84        except Exception as ex:85            error_msg = "[CloudWatchClient] Exception: Unable to execute create_log_stream in CloudWatch: {}".format(ex)86            LOG.error(error_msg)87            raise Exception(error_msg)88    def put_log_events(self, log_group_name, log_stream_name, batch, sequence_token):89        """Put log events into the log stream in the log group.90        Args:91            log_group_name (str): The log group to put the log batch in.92            log_stream_name (str): The log stream within the log group to put the log batch in.93            batch (list): list of dictionaries of log events.94            sequence_token (str): the sequence_token to put the log in next.95        Raises:96            Exception: Exception other then DataAlreadyAcceptedException and InvalidSequenceTokenException97        Returns:98            str: the next sequence token to use.99        """100        retry_num = 0101        while retry_num < self._max_retry_attempts:102            retry_num += 1103            client = self._get_cw_client()104            try:105                if sequence_token:106                    response = client.put_log_events(logGroupName=log_group_name, logStreamName=log_stream_name,107                                                     logEvents=batch, sequenceToken=sequence_token)108                else:109                    response = client.put_log_events(logGroupName=log_group_name, logStreamName=log_stream_name,110                                                     logEvents=batch)111                return response["nextSequenceToken"]112            except client.exceptions.DataAlreadyAcceptedException as ex:113                error = "[CloudWatchClient] DataAlreadyAcceptedException put_log_events :{}".format(ex)114                LOG.info(error)115                sequence_token = self._parse_next_sequence_token_from_exception(ex)116                return sequence_token117            except client.exceptions.InvalidSequenceTokenException as ex:118                error = "[CloudWatchClient] InvalidSequenceTokenException put_log_events :{}".format(ex)119                LOG.info(error)120                sequence_token = self._parse_next_sequence_token_from_exception(ex)121            except Exception as ex:122                error = "[CloudWatchClient] Exception put_log_events :{}".format(ex)123                LOG.info(error)...

test_logbeam.py

Source:test_logbeam.py

1import logging2import mock3import pytest4import time5from cwlogs.push import LogEvent6from .. import (7    CloudWatchLogsHandler,8    BatchedCloudWatchSink,9    nextSequenceToken,10)11log_message_data = [12    {13        'timestamp': (time.time() * 1000) + i,14        'message': 'Message {}'.format(i),15    }16    for i in range(100)17]18first_token = 'original fake sequence token'19second_token = 'new fake sequence token'20def sink_data(sink, events):21    for event in events:22        sink.add_event(LogEvent(**event))23@pytest.fixture24def mock_cw():25    mock_cw = mock.MagicMock()26    mock_cw.describe_log_streams = mock.MagicMock(return_value={27        'logStreams': [{28            'logStreamName': 'log_stream_name',29            'uploadSequenceToken': first_token,30        }],31    })32    mock_cw.put_log_events = mock.MagicMock(return_value={33        'nextSequenceToken': second_token,34        'rejectedLogEventsInfo': {},35    })36    return mock_cw37class TestSink(object):38    @pytest.fixture39    def sink(self, mock_cw):40        sink = BatchedCloudWatchSink(41            mock_cw,42            'log_group_name',43            'log_stream_name',44            500,45            50,46            1024 * 100,47        )48        sink.start()49        return sink50    def test_get_sequence_token(self, mock_cw):51        assert (52            nextSequenceToken(mock_cw, 'log_group_name', 'log_stream_name') ==53            first_token54        )55    def test_sink_batch(self, sink):56        sink_data(sink, log_message_data[:2])57        sink.shutdown()58        sink.logs_service.put_log_events.called_once_with(59            logEvents=log_message_data[:2],60            logGroupName='log_group_name',61            logStreamName='log_stream_name',62            sequenceToken=first_token,63        )64    def test_sink_timeout(self, sink):65        sink_data(sink, log_message_data[:1])66        time.sleep(4)67        sink.logs_service.put_log_events.assert_called_with(68            logEvents=log_message_data[:1],69            logGroupName='log_group_name',70            logStreamName='log_stream_name',71            sequenceToken=first_token,72        )73        sink_data(sink, log_message_data[1:2])74        sink.shutdown()75        sink.logs_service.put_log_events.assert_called_with(76            logEvents=log_message_data[1:2],77            logGroupName='log_group_name',78            logStreamName='log_stream_name',79            sequenceToken=second_token,80        )81    def test_sink_batch_length(self, sink):82        sink_data(sink, log_message_data)83        sink.shutdown()84        sink.logs_service.put_log_events.assert_any_call(85            logEvents=log_message_data[:50],86            logGroupName='log_group_name',87            logStreamName='log_stream_name',88            sequenceToken=first_token,89        )90        sink.logs_service.put_log_events.assert_any_call(91            logEvents=log_message_data[50:],92            logGroupName='log_group_name',93            logStreamName='log_stream_name',94            sequenceToken=second_token,95        )96class TestHandler(object):97    def test_handler(self, mock_cw):98        logger = logging.getLogger(__name__)99        logger.setLevel(logging.INFO)100        logger.propagate = False101        handler = CloudWatchLogsHandler(102            logs_client=mock_cw,103            log_group_name='log_group_name',104            log_stream_name='log_stream_name',105        )106        logger.addHandler(handler)107        logger.info('Hello world')108        handler.close()109        call_args = mock_cw.put_log_events.call_args[1]110        assert call_args['sequenceToken'] == first_token111        assert len(call_args['logEvents']) == 1...

log_events.py

Source:log_events.py

...37            )38            logger.info(f'call create_log_group(): log group has been created {response}')39        except Exception as e:40            logger.error(f'{e}')41    def put_log_events(self, message: str = ''):42        try:43            sequence_token = _read_parameters_store(param_store_name='sfigiel-sequenceToken')[0]44            response = self.client.put_log_events(45                logGroupName=self.log_group_name,46                logStreamName=self.log_stream_name,47                logEvents=[48                    {49                        'timestamp': int(time.time() * 1000),50                        'message': message51                    },52                ],53                sequenceToken=sequence_token54            )55            logger.info(f'call put_log_events().Response: {response}')56            self.sequence_token = response['nextSequenceToken']57            _put_parameter_to_store(value=self.sequence_token)58        except Exception as e:59            logger.error(f'call put_log_events().Error {e}')60            response = self.client.put_log_events(61                logGroupName=self.log_group_name,62                logStreamName=self.log_stream_name,63                logEvents=[64                    {65                        'timestamp': int(time.time() * 1000),66                        'message': "init logs"67                    },68                ],69            )70            logger.info(f'call put_log_events().Exception response: {response}')71            self.sequence_token = response['nextSequenceToken']72            _put_parameter_to_store(value=self.sequence_token)73    def __repr__(self):74        return f"Logger([{self.client}, {self.user}, {self.env}, {self.log_group_name}])"75    def __str__(self):...

Automation Testing Tutorials

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.