Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use collect_events method in localstack

Best Python code snippet using localstack_python

start_time.py

Source:start_time.py

...61        :return:62        """63        collected_start_time = self.collect_specific_text_from_logs('.*Node started.*:.*[stage=.')64        if collected_start_time:65            configure_sys_pool = self.collect_events('.*stage="Configure system pool" (\d+) ms', collected_start_time)66            start_managers = self.collect_events('.*stage="Start managers" (\d+) ms', collected_start_time)67            configure_binary_meta = self.collect_events('.*stage="Configure binary metadata" (\d+) ms',68                                                        collected_start_time)69            start_procs = self.collect_events('.*stage="Start processors" (\d+) ms', collected_start_time)70            gg_plugin = self.collect_events('.*stage="Start \'GridGain\' plugin" (\d+) ms', collected_start_time)71            init_regions = self.collect_events('.*stage="Init and start regions" (\d+) ms', collected_start_time)72            restore_binary = self.collect_events('.*stage="Restore binary memory" (\d+) ms', collected_start_time)73            restore_logical = self.collect_events('.*stage="Restore logical state" (\d+) ms', collected_start_time)74            finish_recovery = self.collect_events('.*stage="Finish recovery" (\d+) ms', collected_start_time)75            join_topology = self.collect_events('.*stage="Join topology" (\d+) ms', collected_start_time)76            transition = self.collect_events('.*stage="Await transition" (\d+) ms', collected_start_time)77            exchange = self.collect_events('.*stage="Await exchange" (\d+) ms', collected_start_time)78            total = self.collect_events('.*stage="Total time" (\d+) ms', collected_start_time)79            self.results[self.version] = {80                'system pool': configure_sys_pool,81                'start managers': start_managers,82                'configure binary meta': configure_binary_meta,83                'start processors': start_procs,84                'gg_plugin': gg_plugin,85                'init and start regions': init_regions,86                'restore binary memory': restore_binary,87                'restore logical state': restore_logical,88                'finish recovery': finish_recovery,89                'join topology': join_topology,90                'await transition': transition,91                'await exchange': exchange,92                'total': total,93            }94    def print_pre_259_start_times(self):95        stopped_in_cp = {}96        for node, logs_content in self.collect_specific_text_from_logs(97                '.*Ignite node stopped in the middle of checkpoint.').items():98            stopped_in_cp[node] = 'Ignite node stopped in the middle of checkpoint.' in logs_content99        binary_recovery_time = self.collect_events('.*performed in (\d+) ms',100                                                   self.collect_specific_text_from_logs('Binary recovery performed in'))101        part_states = self.collect_events('.*partitionsProcessed=(\d+), time=(\d+)ms]',102                                          self.collect_specific_text_from_logs(103                                              'Finished restoring partition state for local groups'))104        wal_changes = self.collect_events('.*updatesApplied=(\d+), time=(\d+) ms',105                                          self.collect_specific_text_from_logs('Finished applying WAL changes'))106        memory_changes = self.collect_events('.*changesApplied=(\d+), time=(\d+) ms',107                                             self.collect_specific_text_from_logs('Finished applying memory changes'))108        log_print(stopped_in_cp, color='yellow')109        log_print(binary_recovery_time, color='yellow')110        log_print(part_states, color='yellow')111        log_print(wal_changes, color='yellow')112        log_print(memory_changes, color='yellow')113        for node in stopped_in_cp.keys():114            log_print("Report for node %s %s" % (node, '(coordinator)' if node == 1 else ''), color='green')115            if stopped_in_cp[node]:116                log_print("Stopped in the middle of checkpoint", color='green')117            else:118                log_print("Stopped gracefully", color='green')119            log_print("Binary was restored in %s ms" % binary_recovery_time[node][0], color='green')120            part_stated_time = part_states[node][1]121            part_stated_speed = (part_states[node][1] / part_states[node][0]) if part_states[node][0] != 0 else 0122            log_print("Partitions recovery time = %s ms, (time / partitions) = %.2f" %123                      (part_stated_time, part_stated_speed), color='green')124            wal_changes_time = wal_changes[node][1]125            wal_changes_speed = wal_changes[node][0]126            log_print("Applying WAL changes time = %s ms, updates = %s" %127                      (wal_changes_time, wal_changes_speed), color='green')128            if node in memory_changes:129                memory_changes_time = memory_changes[node][1]130                memory_changes_speed = (memory_changes[node][1] / memory_changes[node][0]) \131                    if memory_changes[node][0] != 0 else 0132                log_print("Applying memory changes time = %s ms, (time / changes) = %.2f" %133                          (memory_changes_time, memory_changes_speed), color='green')134            else:135                log_print("There is no memory changes log message found", color='green')136    def collect_events(self, msg, vals):137        collected_cps = {}138        for node, part_state in vals.items():139            m = re.search(msg, part_state)140            if m:141                for group in m.groups():142                    if node in collected_cps:143                        collected_cps[node].append(int(group))144                    else:145                        collected_cps[node] = [int(group), ]146        return collected_cps147    def print_results(self):148        log_print('Results for checkpointLockHoldTime probe: {}'.format(self.results))149    def is_passed(self, **kwargs):150        """...

test_std_movements.py

Source:test_std_movements.py

...3from battlefield.Facing import Facing4from battlefield.Battlefield import Cell5import pytest6@pytest.fixture()7def collect_events(hero_only_game):8    events = []9    hero_only_game.events_platform.process_event = lambda x: events.append(x)10    yield events11facings = [Facing.NORTH, Facing.EAST, Facing.SOUTH, Facing.WEST]12def check_contains_right_action_and_cell(collect_events, action, cell):13    activated = False14    assert collect_events15    for e in collect_events:16        if isinstance(e, ActiveEvent):17            if e.active.name == action.name:18                activated = True19            assert e.targeting == cell20    assert activated21@pytest.mark.parametrize("facing", facings)...

functools.py

Source:functools.py

...13        self.assertEqual(d, {'c': 2})14    def test_return_locals_profile_events(self):15        '''Insure profile function gets events for function decorated with16        return_locals at least the same as for function itself.'''17        def collect_events(func):18            events = []19            def tracer(frame, event, args):20                events.append((frame.f_code, event))21            old_tracer = sys.getprofile()22            sys.setprofile(tracer)23            try:24                func()25            finally:26                sys.setprofile(old_tracer)27            return events28        def inner():29            return 230        def outer():31            a = 132            b = inner()33        events1 = set(collect_events(outer))34        events2 = set(collect_events(return_locals(outer)))35        self.assertTrue(events2.issuperset(events1))36    def test_return_locals_debugger_events(self):37        '''Insure debugging function gets events for function decorated with38        return_locals at least the same as for function itself.'''39        def collect_events(func):40            events = []41            def tracer(frame, event, args):42                events.append((frame.f_code, event))43                return tracer44            old_tracer = sys.gettrace()45            sys.settrace(tracer)46            try:47                func()48            finally:49                sys.settrace(old_tracer)50            return events51        def inner():52            return 253        def outer():54            a = 155            b = inner()56        events1 = set(collect_events(outer))57        events2 = set(collect_events(return_locals(outer)))58        self.assertTrue(events2.issuperset(events1))59    def test_return_locals_cProfile(self):60        '''Insure code using return_locals is profilable with cProfile.'''61        @return_locals62        def func():63            pass...

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.