Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use get_machine_arn method in localstack

Best Python code snippet using localstack_python

test_stepfunctions.py

Source:test_stepfunctions.py

...206        )207        # assert that the SM has been created208        self.assert_machine_created(state_machines_before)209        # run state machine210        sm_arn = self.get_machine_arn(sm_name)211        input = {"x": "1", "y": "2"}212        result = self.sfn_client.start_execution(stateMachineArn=sm_arn, input=json.dumps(input))213        self.assertTrue(result.get("executionArn"))214        # define expected output215        test_output = {**input, "added": {"Hello": TEST_RESULT_VALUE}}216        def check_result():217            result = self._get_execution_results(sm_arn)218            self.assertEqual(test_output, result)219        # assert that the result is correct220        retry(check_result, sleep=2, retries=10)221        # clean up222        self.cleanup(sm_arn, state_machines_before)223    def test_create_run_map_state_machine(self):224        names = ["Bob", "Meg", "Joe"]225        test_input = [{"map": name} for name in names]226        test_output = [{"Hello": name} for name in names]227        state_machines_before = self.sfn_client.list_state_machines()["stateMachines"]228        role_arn = aws_stack.role_arn("sfn_role")229        definition = clone(STATE_MACHINE_MAP)230        lambda_arn_3 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_3)231        definition["States"]["ExampleMapState"]["Iterator"]["States"]["CallLambda"][232            "Resource"233        ] = lambda_arn_3234        definition = json.dumps(definition)235        sm_name = "map-%s" % short_uid()236        _ = self.sfn_client.create_state_machine(237            name=sm_name, definition=definition, roleArn=role_arn238        )239        # assert that the SM has been created240        self.assert_machine_created(state_machines_before)241        # run state machine242        sm_arn = self.get_machine_arn(sm_name)243        lambda_api.LAMBDA_EXECUTOR.function_invoke_times.clear()244        result = self.sfn_client.start_execution(245            stateMachineArn=sm_arn, input=json.dumps(test_input)246        )247        self.assertTrue(result.get("executionArn"))248        def check_invocations():249            self.assertIn(lambda_arn_3, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)250            # assert that the result is correct251            result = self._get_execution_results(sm_arn)252            self.assertEqual(test_output, result)253        # assert that the lambda has been invoked by the SM execution254        retry(check_invocations, sleep=1, retries=10)255        # clean up256        self.cleanup(sm_arn, state_machines_before)257    def test_create_run_state_machine(self):258        state_machines_before = self.sfn_client.list_state_machines()["stateMachines"]259        # create state machine260        role_arn = aws_stack.role_arn("sfn_role")261        definition = clone(STATE_MACHINE_BASIC)262        lambda_arn_1 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_1)263        lambda_arn_2 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_2)264        definition["States"]["step1"]["Resource"] = lambda_arn_1265        definition["States"]["step2"]["Resource"] = lambda_arn_2266        definition = json.dumps(definition)267        sm_name = "basic-%s" % short_uid()268        result = self.sfn_client.create_state_machine(269            name=sm_name, definition=definition, roleArn=role_arn270        )271        # assert that the SM has been created272        self.assert_machine_created(state_machines_before)273        # run state machine274        sm_arn = self.get_machine_arn(sm_name)275        lambda_api.LAMBDA_EXECUTOR.function_invoke_times.clear()276        result = self.sfn_client.start_execution(stateMachineArn=sm_arn)277        self.assertTrue(result.get("executionArn"))278        def check_invocations():279            self.assertIn(lambda_arn_1, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)280            self.assertIn(lambda_arn_2, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)281            # assert that the result is correct282            result = self._get_execution_results(sm_arn)283            self.assertEqual({"Hello": TEST_RESULT_VALUE}, result["result_value"])284        # assert that the lambda has been invoked by the SM execution285        retry(check_invocations, sleep=0.7, retries=25)286        # clean up287        self.cleanup(sm_arn, state_machines_before)288    def test_try_catch_state_machine(self):289        state_machines_before = self.sfn_client.list_state_machines()["stateMachines"]290        # create state machine291        role_arn = aws_stack.role_arn("sfn_role")292        definition = clone(STATE_MACHINE_CATCH)293        lambda_arn_1 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_1)294        lambda_arn_2 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_2)295        definition["States"]["Start"]["Parameters"]["FunctionName"] = lambda_arn_1296        definition["States"]["ErrorHandler"]["Resource"] = lambda_arn_2297        definition["States"]["Final"]["Resource"] = lambda_arn_2298        definition = json.dumps(definition)299        sm_name = "catch-%s" % short_uid()300        result = self.sfn_client.create_state_machine(301            name=sm_name, definition=definition, roleArn=role_arn302        )303        # run state machine304        sm_arn = self.get_machine_arn(sm_name)305        lambda_api.LAMBDA_EXECUTOR.function_invoke_times.clear()306        result = self.sfn_client.start_execution(stateMachineArn=sm_arn)307        self.assertTrue(result.get("executionArn"))308        def check_invocations():309            self.assertIn(lambda_arn_1, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)310            self.assertIn(lambda_arn_2, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)311            # assert that the result is correct312            result = self._get_execution_results(sm_arn)313            self.assertEqual({"Hello": TEST_RESULT_VALUE}, result.get("handled"))314        # assert that the lambda has been invoked by the SM execution315        retry(check_invocations, sleep=1, retries=10)316        # clean up317        self.cleanup(sm_arn, state_machines_before)318    def test_intrinsic_functions(self):319        state_machines_before = self.sfn_client.list_state_machines()["stateMachines"]320        # create state machine321        role_arn = aws_stack.role_arn("sfn_role")322        definition = clone(STATE_MACHINE_INTRINSIC_FUNCS)323        lambda_arn_1 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_5)324        lambda_arn_2 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_5)325        if isinstance(definition["States"]["state1"].get("Parameters"), dict):326            definition["States"]["state1"]["Parameters"]["lambda_params"][327                "FunctionName"328            ] = lambda_arn_1329            definition["States"]["state3"]["Resource"] = lambda_arn_2330        definition = json.dumps(definition)331        sm_name = "intrinsic-%s" % short_uid()332        self.sfn_client.create_state_machine(name=sm_name, definition=definition, roleArn=role_arn)333        # run state machine334        sm_arn = self.get_machine_arn(sm_name)335        lambda_api.LAMBDA_EXECUTOR.function_invoke_times.clear()336        input = {}337        result = self.sfn_client.start_execution(stateMachineArn=sm_arn, input=json.dumps(input))338        self.assertTrue(result.get("executionArn"))339        def check_invocations():340            self.assertIn(lambda_arn_1, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)341            self.assertIn(lambda_arn_2, lambda_api.LAMBDA_EXECUTOR.function_invoke_times)342            # assert that the result is correct343            result = self._get_execution_results(sm_arn)344            self.assertEqual({"payload": {"values": [1, "v2"]}}, result.get("result_value"))345        # assert that the lambda has been invoked by the SM execution346        retry(check_invocations, sleep=1, retries=10)347        # clean up348        self.cleanup(sm_arn, state_machines_before)349    def test_events_state_machine(self):350        events = aws_stack.connect_to_service("events")351        state_machines_before = self.sfn_client.list_state_machines()["stateMachines"]352        # create event bus353        bus_name = f"bus-{short_uid()}"354        events.create_event_bus(Name=bus_name)355        # create state machine356        definition = clone(STATE_MACHINE_EVENTS)357        definition["States"]["step1"]["Parameters"]["Entries"][0]["EventBusName"] = bus_name358        definition = json.dumps(definition)359        sm_name = "events-%s" % short_uid()360        role_arn = aws_stack.role_arn("sfn_role")361        self.sfn_client.create_state_machine(name=sm_name, definition=definition, roleArn=role_arn)362        # run state machine363        events_before = len(TEST_EVENTS_CACHE)364        sm_arn = self.get_machine_arn(sm_name)365        result = self.sfn_client.start_execution(stateMachineArn=sm_arn)366        self.assertTrue(result.get("executionArn"))367        def check_invocations():368            # assert that the event is received369            self.assertEqual(events_before + 1, len(TEST_EVENTS_CACHE))370            last_event = TEST_EVENTS_CACHE[-1]371            self.assertEqual(bus_name, last_event["EventBusName"])372            self.assertEqual("TestSource", last_event["Source"])373            self.assertEqual("TestMessage", last_event["DetailType"])374            self.assertEqual(375                {"Message": "Hello from Step Functions!"}, json.loads(last_event["Detail"])376            )377        # assert that the event bus has received an event from the SM execution378        retry(check_invocations, sleep=1, retries=10)379        # clean up380        self.cleanup(sm_arn, state_machines_before)381        events.delete_event_bus(Name=bus_name)382    # -----------------383    # HELPER FUNCTIONS384    # -----------------385    def get_machine_arn(self, sm_name):386        state_machines = self.sfn_client.list_state_machines()["stateMachines"]387        return [m["stateMachineArn"] for m in state_machines if m["name"] == sm_name][0]388    def assert_machine_created(self, state_machines_before):389        return self._assert_machine_instances(len(state_machines_before) + 1)390    def assert_machine_deleted(self, state_machines_before):391        return self._assert_machine_instances(len(state_machines_before))392    def cleanup(self, sm_arn, state_machines_before):393        self.sfn_client.delete_state_machine(stateMachineArn=sm_arn)394        self.assert_machine_deleted(state_machines_before)395    def _assert_machine_instances(self, expected_instances):396        def check():397            state_machines_after = self.sfn_client.list_state_machines()["stateMachines"]398            self.assertEqual(expected_instances, len(state_machines_after))399            return state_machines_after...

step_functions.py

Source:step_functions.py

...14    def client(self) -> Session.client:15        client = boto3.client("stepfunctions")16        return client1718    def get_machine_arn(self) -> str:19        state_machine_name = os.getenv("STEP_MACHINE_NAME")20        list_all_machines = self.client.list_state_machines()21        state_machine = [22            machine23            for machine in list_all_machines["stateMachines"]24            if machine["name"] == state_machine_name25        ][0]26        return state_machine["stateMachineArn"]2728    def start_execution(self, payload: Dict[Any, Any]) -> Dict[Any, Any]:29        logger.debug(f"Start Send Notification Machine with input: {payload}")30        arn = self.get_machine_arn()31        response = self.client.start_execution(32            stateMachineArn=arn,33            input=json.dumps(payload),34        )35        logger.debug(f"Step Machine response is: {response}")
...

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.