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

How to use list_executions method in localstack

Best Python code snippet using localstack_python

aws_stepfunctions_info.py

Source:aws_stepfunctions_info.py

...110                    stateMachineArn=module.params['arn'],111                    statusFilter=module.params['status_filter']112                ), True113            else:114                return client.list_executions(115                    stateMachineArn=module.params['arn'],116                    statusFilter=module.params['status_filter']117                ), False118        elif module.params['list_state_machines']:119            if client.can_paginate('list_state_machines'):120                paginator = client.get_paginator('list_state_machines')121                return paginator.paginate(), True122            else:123                return client.list_state_machines(), False124        elif module.params['describe_state_machine']:125            if client.can_paginate('describe_state_machine'):126                paginator = client.get_paginator('describe_state_machine')127                return paginator.paginate(128                    stateMachineArn=module.params['arn'],...

autoscaleTaskHandler.py

Source:autoscaleTaskHandler.py

...12def start_state_function(state_machine_arn, data):13    logger.info('starting state machine with dict {}'.format(data))14    sfnConnection = boto3.client('stepfunctions')15    success_count = len(16        sfnConnection.list_executions(stateMachineArn=state_machine_arn, statusFilter='SUCCEEDED')['executions'])17    running_count = len(18        sfnConnection.list_executions(stateMachineArn=state_machine_arn, statusFilter='RUNNING')['executions'])19    logger.info('State machine running count is {} and success count is {}'.format(running_count, success_count))20    step_function_arns = []21    time.sleep(30)22    logger.info('Starting step function with input {}'.format(data))23    try:24        result = sfnConnection.start_execution(stateMachineArn=state_machine_arn, input=data)25    except execution as e:26        logger.info('Got exception {} starting step function'.format(e))27    logger.info('Got result from start execution '.format(result))28    logger.info('State maching ARN is {}'.format(result.get('executionArn')))29    step_function_arns.append(result.get('executionArn'))30    logger.info("Started StateMachine")31    time.sleep(30)32    success_count = len(33        sfnConnection.list_executions(stateMachineArn=state_machine_arn, statusFilter='SUCCEEDED')['executions'])34    running_count = len(35        sfnConnection.list_executions(stateMachineArn=state_machine_arn, statusFilter='RUNNING')['executions'])36    failed_count = len(37        sfnConnection.list_executions(stateMachineArn=state_machine_arn, statusFilter='FAILED')['executions'])38    logger.info('State machine running count is {}, failed count is {} and success count is {}'39                .format(running_count, failed_count, success_count))40    if running_count == 0 and failed_count > 0:41        logger.info('Problems starting the step function')42def lambda_handler(event, context):43    logger.info("Got event: {}".format(event))44    data = event['Records'][0]['Sns']['Message']45    logger.info("Got message: {}".format(data))46    """47    {48        "LifecycleHookName": "vpn-outASG-life-cycle-launch",49        "AccountId": "106808901653",50        "RequestId": "b575b9fb-d658-c309-04ed-f4bd537d5305",51        "LifecycleTransition": "autoscaling:EC2_INSTANCE_LAUNCHING",...

test_list_executions.py

Source:test_list_executions.py

...12            'logs', 'simple', is_end_to_end_path=True)13        context, first_user_execution_id2 = run_example(14            'logs', 'simple', context=context, is_end_to_end_path=True)15        first_user = context.configuration.user16        infos = context.controller.list_executions(first_user,17                                                   list_for_all_users=False)18        self.assertSetEqual(19            {first_user_execution_id1, first_user_execution_id2},20            {i['execution_id']21             for i in infos})22        second_user = first_user + '_second'23        context.configuration.user = second_user24        # Nothing for the second user yet25        infos = context.controller.list_executions(second_user,26                                                   list_for_all_users=False)27        self.assertListEqual([], infos)28        context, second_user_execution_id = run_example(29            'logs', 'simple', context=context, is_end_to_end_path=True)30        # Check that the execution of the second user appears31        infos = context.controller.list_executions(second_user,32                                                   list_for_all_users=False)33        self.assertListEqual([second_user_execution_id],34                             [i['execution_id'] for i in infos])35        # Check that the executions of the first user are still the same36        infos = context.controller.list_executions(first_user,37                                                   list_for_all_users=False)38        self.assertSetEqual(39            {first_user_execution_id1, first_user_execution_id2},40            {i['execution_id']41             for i in infos})42        # Check that when requesting to list for all users, we get all of them43        infos = context.controller.list_executions(first_user,44                                                   list_for_all_users=True)45        self.assertSetEqual(46            {47                first_user_execution_id1, first_user_execution_id2,48                second_user_execution_id49            }, {i['execution_id']50                for i in infos})51        # ...also for the second user52        infos = context.controller.list_executions(second_user,53                                                   list_for_all_users=True)54        self.assertSetEqual(55            {56                first_user_execution_id1, first_user_execution_id2,57                second_user_execution_id58            }, {i['execution_id']...

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.