def cmd_run_smoke_test()

in sam/app-decompose-for-parallelism/ctl.py [0:0]

• 38 lines of code
• 6 McCabe index (conditional complexity)

def cmd_run_smoke_test(cli_args):

    iteration_count = 1
    try:
        iteration_count = int(cli_args[0])
    except:
        iteration_count = 1

    profile = get_parameter('profile')
    region = get_parameter('region')

    state_machine_arn = get_output_value('StateMachineMain')

    if state_machine_arn == '':
        print("Error: Failed to find statemachine arn in output: {}".format(get_stack_output()))
        exit()
    else:
        print("statemachine arn is {}".format(state_machine_arn))

    exec_input = {'iteration_count': iteration_count}
    exec_name = "smoketest_{}iterations_{}".format(iteration_count,int(time.time()))
    try:
        boto3.setup_default_session(profile_name=profile,region_name=region)
        client = boto3.client('stepfunctions')
        start_time = time.time()
        test_execution = client.start_execution(stateMachineArn=state_machine_arn,input=json.dumps(exec_input),name=exec_name)
        print("Started execution test execution {} at {} with input {}".format(test_execution['executionArn'],start_time,exec_input))
        print(".", end="",flush=True)

        # Poll for completion
        while 1==1:
            time.sleep(2)
            try:
                execution_description = client.describe_execution(executionArn=test_execution['executionArn'])
                if not execution_description['status'] == 'RUNNING':
                    # Then this finished
                    print("")
                    print("Test execution complete after {} seconds with result:".format(int(time.time() - start_time)))
                    print(json.dumps(process_execution_description(execution_description),indent=2,default=handler_datetime))
                    break
                else:
                    print(".", end="",flush=True)
            except:
                print("\nError: Failure describing execution for {}: {}".format(test_execution['executionArn'],sys.exc_info()))
    except:
        # Just pring an error and continue
        print("Error: Failed to execute {}: {}".format(test_execution['executionArn'],sys.exc_info()))