Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use moto_put_log_events method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...127        for stream in log_group.streams.values():128            stream.filter_pattern = filter_pattern129    log_group.put_subscription_filter(filter_name, filter_pattern, destination_arn, role_arn)130@patch(MotoLogStream.put_log_events, pass_target=False)131def moto_put_log_events(self, log_group_name, log_stream_name, log_events):132    # TODO: call/patch upstream method here, instead of duplicating the code!133    self.last_ingestion_time = int(unix_time_millis())134    self.stored_bytes += sum([len(log_event["message"]) for log_event in log_events])135    events = [logs_models.LogEvent(self.last_ingestion_time, log_event) for log_event in log_events]136    self.events += events137    self.upload_sequence_token += 1138    # apply filterpattern -> only forward what matches the pattern139    if self.filter_pattern:140        # TODO only patched in pro141        matches = get_pattern_matcher(self.filter_pattern)142        events = [143            logs_models.LogEvent(self.last_ingestion_time, event)144            for event in log_events145            if matches(self.filter_pattern, event)...

logs_listener.py

Source:logs_listener.py

...78                        LOG.info("Unable to put metric data for matching CloudWatch log events", e)79def get_pattern_matcher(pattern: str) -> Callable[[str, Dict], bool]:80    """Returns a pattern matcher. Can be patched by plugins to return a more sophisticated pattern matcher."""81    return lambda _pattern, _log_event: True82def moto_put_log_events(_, self, log_group_name, log_stream_name, log_events, sequence_token):83    # TODO: call/patch upstream method here, instead of duplicating the code!84    self.last_ingestion_time = int(unix_time_millis())85    self.stored_bytes += sum([len(log_event["message"]) for log_event in log_events])86    events = [logs_models.LogEvent(self.last_ingestion_time, log_event) for log_event in log_events]87    self.events += events88    self.upload_sequence_token += 189    log_events = [90        {91            "id": event.event_id,92            "timestamp": event.timestamp,93            "message": event.message,94        }95        for event in events96    ]...

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.