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

How to use create_log_stream method in localstack

Best Python code snippet using localstack_python

test_log_group_client.py

Source:test_log_group_client.py

...36        ''' creating log groups propagates other errors '''37        with patch.object(self.cwl, 'create_log_group', side_effect=client_error('other error')):38            with self.assertRaises(ClientError):39                self.client.create_log_group()40    def test_create_log_stream(self):41        ''' it creates log streams '''42        self.client.create_log_stream(self.STREAM)43        self.cwl.create_log_stream.assert_called_once_with(logGroupName=self.GROUP, logStreamName=self.STREAM)44    def test_create_log_stream_exists(self):45        ''' if log stream exists it ignores the error '''46        with patch.object(self.cwl, 'create_log_stream', side_effect=client_error('ResourceAlreadyExistsException')):47            self.client.create_log_stream(self.STREAM)48    def test_create_log_stream_other_error(self):49        ''' creating log streams propagates other errors '''50        with patch.object(self.cwl, 'create_log_stream', side_effect=client_error('other error')):51            with self.assertRaises(ClientError):52                self.client.create_log_stream(self.STREAM)53    def test_get_new_seq_token_empty(self):54        ''' no log streams found, it should create the log stream '''55        value = dict(logStreams=[])56        self.cwl.describe_log_streams.return_value = value57        self.client.get_new_seq_token(self.STREAM)58        self.assertEqual(self.cwl.mock_calls, [59            call.describe_log_streams(limit=1, logGroupName=self.GROUP, logStreamNamePrefix=self.STREAM),60            call.create_log_stream(logGroupName=self.GROUP, logStreamName=self.STREAM),61        ])62    def test_get_new_seq_token_no_match(self):63        ''' no matching log streams found, it should create the log stream '''64        value = dict(logStreams=[dict(logStreamName='blargh')])65        self.cwl.describe_log_streams.return_value = value66        self.client.get_new_seq_token(self.STREAM)67        self.assertEqual(self.cwl.mock_calls, [68            call.describe_log_streams(limit=1, logGroupName=self.GROUP, logStreamNamePrefix=self.STREAM),69            call.create_log_stream(logGroupName=self.GROUP, logStreamName=self.STREAM),70        ])71    def test_get_new_seq_token(self):72        ''' return the seq token '''73        value = dict(logStreams=[dict(logStreamName=self.STREAM, uploadSequenceToken=sentinel.token)])74        self.cwl.describe_log_streams.return_value = value75        self.assertIs( self.client.get_new_seq_token(self.STREAM), sentinel.token )76    def test_get_seq_token(self):77        ''' it should get a new seq token '''78        with patch.object(self.client, 'get_new_seq_token', return_value=sentinel.token):79            self.assertIs( self.client.get_seq_token(self.STREAM), sentinel.token )80    def test_get_seq_token_cached(self):81        ''' it should return a cached seq token '''82        self.client.tokens[self.STREAM] = sentinel.token83        self.assertIs( self.client.get_seq_token(self.STREAM), sentinel.token )...

test_cloudwatch_logs.py

Source:test_cloudwatch_logs.py

...46    expected_params = {"logGroupName": log_group, "logStreamName": log_stream}47    spy_logger = mocker.spy(cloudwatch_logs.logger, "debug")48    client_stubber.add_response("create_log_stream", response, expected_params)49    client_stubber.activate()50    cloudwatch_logs.create_log_stream(log_group, log_stream)51    assert spy_logger.call_count == 152def test_create_log_stream_exception(mocker):53    cloudwatch_logs = CloudWatchLogs()54    client_stubber = Stubber(cloudwatch_logs.cw_logs)55    log_stream = "test_stream"56    log_group = "test_log_group"57    spy_logger = mocker.spy(cloudwatch_logs.logger, "warning")58    client_stubber.add_client_error("create_log_stream", "Invalid_request")59    client_stubber.activate()60    cloudwatch_logs.create_log_stream(log_group, log_stream)61    assert spy_logger.call_count == 262def test_log(mocker):63    cloudwatch_logs = CloudWatchLogs()64    log_group = "test_log_group"65    message = "Test Message"66    mocker.patch.object(cloudwatch_logs, "put_log_events")67    mocker.patch.object(cloudwatch_logs, "create_log_stream")68    spy_put_log_events = mocker.spy(cloudwatch_logs, "put_log_events")69    spy_create_log_stream = mocker.spy(cloudwatch_logs, "create_log_stream")70    cloudwatch_logs.log(log_group, message)71    spy_put_log_events.assert_called_once()...

redshift.py

Source:redshift.py

...33            },34        }35        variables.update(additional)36        return variables37    def create_log_stream(self):38        t = self.template39        super(DeliveryStream, self).create_log_stream()40        self.redshift_log_stream = t.add_resource(41            logs.LogStream(42                REDSHIFT_LOG_STREAM,43                LogGroupName=Ref(self.log_group),44                DependsOn=self.log_group.title45            )46        )47        t.add_output(48            Output(49                "RedshiftLogStreamName",50                Value=Ref(self.redshift_log_stream)51            )52        )53    def create_delivery_stream(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.