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

How to use get_log_events method in localstack

Best Python code snippet using localstack_python

logs.py

Source:logs.py

...182        Checks the default stream if no stream_name is given."""183        if stream_name is None:184            stream_name = self._get_stream_name()185        client = boto3.client("logs", region_name=self.region)186        response = client.get_log_events(187            logGroupName=self.name,188            logStreamName=stream_name,189            # startTime=int(datetime.datetime(2021, 8, 19, 0, 0).strftime('%s'))*1000,190            # endTime=int(datetime.datetime(2021, 8, 20, 0, 0).strftime('%s'))*1000,191            limit=n,192            startFromHead=False,193        )194        return TailLogResponse(**response)195    def get_log_events(196        self,197        stream_name: Optional[str] = None,198        limit: int = 10,199        startTime: Optional[int] = None,200        endTime: Optional[int] = None,201        start_from_head: bool = True,202    ) -> Events:203        """Retrieve the entire log and returns it.204        Checks the default stream if no stream_name is given.205        Returns all the events for the log.206        The last item in the list is the newest event.207        Notes to self:208          - when startFromHead is set to True, it reads the oldest logs first.209          - when startFromHead is set to False, it reads the newest logs first.210          - using 'nextForwardToken' is moving from old to new logs.211          - using 'nextBackwardToken' is moving from new to old logs.212        """213        if stream_name is None:214            stream_name = self._get_stream_name()215        client = boto3.client("logs", region_name=self.region)216        events = Events()217        get_log_events_kwargs = {218            "logGroupName": self.name,219            "logStreamName": stream_name,220            "limit": limit,221            "startFromHead": start_from_head,222        }223        if startTime:224            get_log_events_kwargs["startTime"] = startTime225        if endTime:226            get_log_events_kwargs["endTime"] = endTime227        response = client.get_log_events(**get_log_events_kwargs)228        response = TailLogResponse(**response)229        for event in response.events:230            events.events.append(event)231        # print(response.nextBackwardToken)232        # print(response.nextForwardToken)233        next_token = response.nextForwardToken234        while True:235            get_log_events_kwargs["nextToken"] = next_token236            response = client.get_log_events(**get_log_events_kwargs)237            response = TailLogResponse(**response)238            for event in response.events:239                events.events.append(event)240                print(event)241            # The log is depleted when AWS starts returning242            # the same token over and over.243            if next_token == response.nextForwardToken:244                break245            else:246                next_token = response.nextForwardToken247        return events248    def get_log_events_last_seconds(self, seconds: int) -> Events:249        epoch_seconds = epoch_seconds_ago(seconds)250        events = self.get_log_events(251            limit=5000, startTime=epoch_seconds, start_from_head=True252        )253        return events254    def get_log_events_last_minutes(self, minutes: int) -> Events:255        epoch_minutes = epoch_minutes_ago(minutes)256        events = self.get_log_events(257            limit=5000, startTime=epoch_minutes, start_from_head=True258        )259        return events260    def get_log_events_last_hours(self, hours: int) -> Events:261        epoch_hours = epoch_hours_ago(hours)262        events = self.get_log_events(263            limit=5000, startTime=epoch_hours, start_from_head=True264        )265        return events266    def get_log_events_last_days(self, days: int) -> Events:267        epoch_days = epoch_days_ago(days)268        events = self.get_log_events(269            limit=5000, startTime=epoch_days, start_from_head=True270        )...

combine_logs.py

Source:combine_logs.py

...31    log_events = read_logs(unknown_args[0])32    print_logs(log_events, color=args.color, html=args.html)33def read_logs(tmp_dir):34    """Reads log files.35    Delegates to generator function get_log_events() to provide individual log events36    for each of the input log files."""37    files = [("test", "%s/test_framework.log" % tmp_dir)]38    for i in itertools.count():39        logfile = "{}/node{}/regtest/debug.log".format(tmp_dir, i)40        if not os.path.isfile(logfile):41            break42        files.append(("node%d" % i, logfile))43    return heapq.merge(*[get_log_events(source, f) for source, f in files])44def get_log_events(source, logfile):45    """Generator function that returns individual log events.46    Log events may be split over multiple lines. We use the timestamp47    regex match as the marker for a new log event."""48    try:49        with open(logfile, 'r') as infile:50            event = ''51            timestamp = ''52            for line in infile:53                # skip blank lines54                if line == '\n':55                    continue56                # if this line has a timestamp, it's the start of a new log event.57                time_match = TIMESTAMP_PATTERN.match(line)58                if time_match:...

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.