How to use _prepare_method_integration method in localstack

Best Python code snippet using localstack_python

test_api_gateway.py

Source:test_api_gateway.py

...938            resourceId=root_resource_id,939            httpMethod="POST",940            authorizationType="NONE",941        )942        def _prepare_method_integration(943            integr_kwargs={}, resp_templates={}, action="StartExecution", overwrite=False944        ):945            if overwrite:946                client.delete_integration(947                    restApiId=rest_api["id"],948                    resourceId=resources["items"][0]["id"],949                    httpMethod="POST",950                )951            uri = f"arn:aws:apigateway:{aws_stack.get_region()}:states:action/{action}"952            client.put_integration(953                restApiId=rest_api["id"],954                resourceId=root_resource_id,955                httpMethod="POST",956                integrationHttpMethod="POST",957                type="AWS",958                uri=uri,959                **integr_kwargs,960            )961            if resp_templates:962                client.put_integration_response(963                    restApiId=rest_api["id"],964                    resourceId=root_resource_id,965                    selectionPattern="",966                    responseTemplates=resp_templates,967                    httpMethod="POST",968                    statusCode="200",969                )970        # STEP 1: test integration with request template971        _prepare_method_integration(972            integr_kwargs={973                "requestTemplates": {974                    "application/json": """975                    #set($data = $util.escapeJavaScript($input.json('$')))976                    {"input": "$data","stateMachineArn": "%s"}977                    """978                    % sm_arn979                }980            }981        )982        # invoke stepfunction via API GW, assert results983        client.create_deployment(restApiId=rest_api["id"], stageName="dev")984        url = gateway_request_url(api_id=rest_api["id"], stage_name="dev", path="/")985        test_data = {"test": "test-value"}986        resp = requests.post(url, data=json.dumps(test_data))987        self.assertEqual(200, resp.status_code)988        self.assertIn("executionArn", resp.content.decode())989        self.assertIn("startDate", resp.content.decode())990        # STEP 2: test integration without request template991        _prepare_method_integration(overwrite=True)992        test_data_1 = {993            "input": json.dumps(test_data),994            "name": "MyExecution",995            "stateMachineArn": sm_arn,996        }997        # invoke stepfunction via API GW, assert results998        resp = requests.post(url, data=json.dumps(test_data_1))999        self.assertEqual(200, resp.status_code)1000        self.assertIn("executionArn", resp.content.decode())1001        self.assertIn("startDate", resp.content.decode())1002        # STEP 3: test integration with synchronous execution1003        _prepare_method_integration(overwrite=True, action="StartSyncExecution")1004        # invoke stepfunction via API GW, assert results1005        test_data_1["name"] += "1"1006        resp = requests.post(url, data=json.dumps(test_data_1))1007        self.assertEqual(200, resp.status_code)1008        content = json.loads(to_str(resp.content.decode()))1009        self.assertEqual("SUCCEEDED", content.get("status"))1010        self.assertEqual(test_data, json.loads(content.get("output")))1011        # STEP 4: test integration with synchronous execution and response templates1012        resp_templates = {APPLICATION_JSON: "$input.path('$.output')"}1013        _prepare_method_integration(1014            resp_templates=resp_templates, overwrite=True, action="StartSyncExecution"1015        )1016        # invoke stepfunction via API GW, assert results1017        test_data_1["name"] += "2"1018        resp = requests.post(url, data=json.dumps(test_data_1))1019        self.assertEqual(200, resp.status_code)1020        self.assertEqual(test_data, json.loads(to_str(resp.content.decode())))1021        _prepare_method_integration(overwrite=True, action="DeleteStateMachine")1022        # Remove state machine with API GW1023        resp = requests.post(url, data=json.dumps({"stateMachineArn": sm_arn}))1024        self.assertEqual(200, resp.status_code)1025        # Clean up1026        lambda_client.delete_function(FunctionName=fn_name)1027        client.delete_rest_api(restApiId=rest_api["id"])1028    def test_api_gateway_http_integration_with_path_request_parameter(self):1029        client = aws_stack.connect_to_service("apigateway")1030        test_port = get_free_tcp_port()1031        backend_url = "http://localhost:%s/person/{id}" % (test_port)1032        # start test HTTP backend1033        proxy = self.start_http_backend(test_port)1034        # create rest api1035        api_rest = client.create_rest_api(name="test")...

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.