KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use create_api_gateway_and_deploy method in localstack

Best Python code snippet using localstack_python
test_api_gateway.py
Source:test_api_gateway.py
...748            self.assertEqual("NotFoundException", e.response["Error"]["Code"])749        # clean up750        client.delete_rest_api(restApiId=rest_api_id)751    def test_put_integration_dynamodb_proxy_validation_without_response_template(self):752        api_id = self.create_api_gateway_and_deploy({})753        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")754        response = requests.put(755            url,756            json.dumps({"id": "id1", "data": "foobar123"}),757        )758        self.assertEqual(404, response.status_code)759    def test_put_integration_dynamodb_proxy_validation_with_response_template(self):760        response_templates = {761            "application/json": json.dumps(762                {763                    "TableName": "MusicCollection",764                    "Item": {"id": "$.Id", "data": "$.data"},765                }766            )767        }768        api_id = self.create_api_gateway_and_deploy(response_templates)769        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")770        response = requests.put(771            url,772            json.dumps({"id": "id1", "data": "foobar123"}),773        )774        self.assertEqual(200, response.status_code)775        dynamo_client = aws_stack.connect_to_resource("dynamodb")776        table = dynamo_client.Table("MusicCollection")777        result = table.get_item(Key={"id": "id1"})778        self.assertEqual("foobar123", result["Item"]["data"])779    def test_api_key_required_for_methods(self):780        response_templates = {781            "application/json": json.dumps(782                {783                    "TableName": "MusicCollection",784                    "Item": {"id": "$.Id", "data": "$.data"},785                }786            )787        }788        api_id = self.create_api_gateway_and_deploy(response_templates, True)789        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")790        payload = {791            "name": "TEST-PLAN-2",792            "description": "Description",793            "quota": {"limit": 10, "period": "DAY", "offset": 0},794            "throttle": {"rateLimit": 2, "burstLimit": 1},795            "apiStages": [{"apiId": api_id, "stage": "staging"}],796            "tags": {"tag_key": "tag_value"},797        }798        client = aws_stack.connect_to_service("apigateway")799        usage_plan_id = client.create_usage_plan(**payload)["id"]800        key_name = "testApiKey"801        key_type = "API_KEY"802        api_key = client.create_api_key(name=key_name)803        payload = {804            "usagePlanId": usage_plan_id,805            "keyId": api_key["id"],806            "keyType": key_type,807        }808        client.create_usage_plan_key(**payload)809        response = requests.put(810            url,811            json.dumps({"id": "id1", "data": "foobar123"}),812        )813        # when the api key is not passed as part of the header814        self.assertEqual(403, response.status_code)815        response = requests.put(816            url,817            json.dumps({"id": "id1", "data": "foobar123"}),818            headers={"X-API-Key": api_key["value"]},819        )820        # when the api key is passed as part of the header821        self.assertEqual(200, response.status_code)822    def test_multiple_api_keys_validate(self):823        response_templates = {824            "application/json": json.dumps(825                {826                    "TableName": "MusicCollection",827                    "Item": {"id": "$.Id", "data": "$.data"},828                }829            )830        }831        api_id = self.create_api_gateway_and_deploy(response_templates, True)832        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")833        client = aws_stack.connect_to_service("apigateway")834        # Create multiple usage plans835        usage_plan_ids = []836        for i in range(2):837            payload = {838                "name": "APIKEYTEST-PLAN-{}".format(i),839                "description": "Description",840                "quota": {"limit": 10, "period": "DAY", "offset": 0},841                "throttle": {"rateLimit": 2, "burstLimit": 1},842                "apiStages": [{"apiId": api_id, "stage": "staging"}],843                "tags": {"tag_key": "tag_value"},844            }845            usage_plan_ids.append(client.create_usage_plan(**payload)["id"])846        api_keys = []847        key_type = "API_KEY"848        # Create multiple API Keys in each usage plan849        for usage_plan_id in usage_plan_ids:850            for i in range(2):851                api_key = client.create_api_key(name="testMultipleApiKeys{}".format(i))852                payload = {853                    "usagePlanId": usage_plan_id,854                    "keyId": api_key["id"],855                    "keyType": key_type,856                }857                client.create_usage_plan_key(**payload)858                api_keys.append(api_key["value"])859        response = requests.put(860            url,861            json.dumps({"id": "id1", "data": "foobar123"}),862        )863        # when the api key is not passed as part of the header864        self.assertEqual(403, response.status_code)865        # check that all API keys work866        for key in api_keys:867            response = requests.put(868                url,869                json.dumps({"id": "id1", "data": "foobar123"}),870                headers={"X-API-Key": key},871            )872            # when the api key is passed as part of the header873            self.assertEqual(200, response.status_code)874    def test_import_rest_api(self):875        rest_api_name = "restapi-%s" % short_uid()876        client = aws_stack.connect_to_service("apigateway")877        rest_api_id = client.create_rest_api(name=rest_api_name)["id"]878        spec_file = load_file(TEST_SWAGGER_FILE)879        rs = client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")880        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])881        rs = client.get_resources(restApiId=rest_api_id)882        self.assertEqual(883            2, len(rs["items"])884        )  # should contain 2 resources (including the root resource)885        resource = [res for res in rs["items"] if res["path"] == "/test"][0]886        self.assertIn("GET", resource["resourceMethods"])887        # clean up888        client.delete_rest_api(restApiId=rest_api_id)889        spec_file = load_file(TEST_IMPORT_REST_API_FILE)890        rs = client.import_rest_api(body=spec_file)891        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])892        rest_api_id = rs["id"]893        rs = client.get_resources(restApiId=rest_api_id)894        resources = rs["items"]895        self.assertEqual(3, len(resources))896        paths = [res["path"] for res in resources]897        self.assertIn("/", paths)898        self.assertIn("/pets", paths)899        self.assertIn("/pets/{petId}", paths)900        # clean up901        client.delete_rest_api(restApiId=rest_api_id)902    def test_step_function_integrations(self):903        client = aws_stack.connect_to_service("apigateway")904        sfn_client = aws_stack.connect_to_service("stepfunctions")905        lambda_client = aws_stack.connect_to_service("lambda")906        state_machine_name = "test"907        state_machine_def = {908            "Comment": "Hello World example",909            "StartAt": "step1",910            "States": {911                "step1": {"Type": "Task", "Resource": "__tbd__", "End": True},912            },913        }914        # create state machine915        fn_name = "test-stepfunctions-apigw"916        testutil.create_lambda_function(917            handler_file=TEST_LAMBDA_ECHO_FILE,918            func_name=fn_name,919            runtime=LAMBDA_RUNTIME_PYTHON36,920        )921        role_arn = aws_stack.role_arn("sfn_role")922        # create state machine definition923        definition = clone(state_machine_def)924        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)925        definition["States"]["step1"]["Resource"] = lambda_arn_1926        definition = json.dumps(definition)927        # create state machine928        result = sfn_client.create_state_machine(929            name=state_machine_name, definition=definition, roleArn=role_arn930        )931        sm_arn = result["stateMachineArn"]932        # create REST API and method933        rest_api = client.create_rest_api(name="test", description="test")934        resources = client.get_resources(restApiId=rest_api["id"])935        root_resource_id = resources["items"][0]["id"]936        client.put_method(937            restApiId=rest_api["id"],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")1036        api_id = api_rest["id"]1037        parent_response = client.get_resources(restApiId=api_id)1038        parent_id = parent_response["items"][0]["id"]1039        resource_1 = client.create_resource(restApiId=api_id, parentId=parent_id, pathPart="person")1040        resource_1_id = resource_1["id"]1041        resource_2 = client.create_resource(1042            restApiId=api_id, parentId=resource_1_id, pathPart="{id}"1043        )1044        resource_2_id = resource_2["id"]1045        client.put_method(1046            restApiId=api_id,1047            resourceId=resource_2_id,1048            httpMethod="GET",1049            authorizationType="NONE",1050            apiKeyRequired=False,1051            requestParameters={"method.request.path.id": True},1052        )1053        client.put_integration(1054            restApiId=api_id,1055            resourceId=resource_2_id,1056            httpMethod="GET",1057            integrationHttpMethod="GET",1058            type="HTTP",1059            uri=backend_url,1060            timeoutInMillis=3000,1061            contentHandling="CONVERT_TO_BINARY",1062            requestParameters={"integration.request.path.id": "method.request.path.id"},1063        )1064        client.create_deployment(restApiId=api_id, stageName="test")1065        def _test_invoke(url):1066            result = requests.get(url)1067            content = json.loads(to_str(result.content))1068            self.assertEqual(200, result.status_code)1069            self.assertRegex(1070                content["headers"].get(HEADER_LOCALSTACK_REQUEST_URL),1071                "http://.*localhost.*/person/123",1072            )1073        for use_hostname in [True, False]:1074            for use_ssl in [True, False] if use_hostname else [False]:1075                url = self._get_invoke_endpoint(1076                    api_id,1077                    stage="test",1078                    path="/person/123",1079                    use_hostname=use_hostname,1080                    use_ssl=use_ssl,1081                )1082                _test_invoke(url)1083        # clean up1084        client.delete_rest_api(restApiId=api_id)1085        proxy.stop()1086    def _get_invoke_endpoint(1087        self, api_id, stage="test", path="/", use_hostname=False, use_ssl=False1088    ):1089        path = path or "/"1090        path = path if path.startswith(path) else f"/{path}"1091        proto = "https" if use_ssl else "http"1092        if use_hostname:1093            return f"{proto}://{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}/{stage}{path}"1094        return (1095            f"{proto}://localhost:{config.EDGE_PORT}/restapis/{api_id}/{stage}/_user_request_{path}"1096        )1097    def test_api_gateway_s3_get_integration(self):1098        apigw_client = aws_stack.connect_to_service("apigateway")1099        s3_client = aws_stack.connect_to_service("s3")1100        bucket_name = "test-bucket"1101        object_name = "test.json"1102        object_content = '{ "success": "true" }'1103        object_content_type = "application/json"1104        api = apigw_client.create_rest_api(name="test")1105        api_id = api["id"]1106        s3_client.create_bucket(Bucket=bucket_name)1107        s3_client.put_object(1108            Bucket=bucket_name,1109            Key=object_name,1110            Body=object_content,1111            ContentType=object_content_type,1112        )1113        self.connect_api_gateway_to_s3(bucket_name, object_name, api_id, "GET")1114        apigw_client.create_deployment(restApiId=api_id, stageName="test")1115        url = gateway_request_url(api_id, "test", "/")1116        result = requests.get(url)1117        self.assertEqual(200, result.status_code)1118        self.assertEqual(object_content, result.text)1119        self.assertEqual(object_content_type, result.headers["content-type"])1120        # clean up1121        apigw_client.delete_rest_api(restApiId=api_id)1122        s3_client.delete_object(Bucket=bucket_name, Key=object_name)1123        s3_client.delete_bucket(Bucket=bucket_name)1124    def test_api_mock_integration_response_params(self):1125        # apigw_client = aws_stack.connect_to_service('apigateway')1126        resps = [1127            {1128                "statusCode": "204",1129                "httpMethod": "OPTIONS",1130                "responseParameters": {1131                    "method.response.header.Access-Control-Allow-Methods": "'POST,OPTIONS'",1132                    "method.response.header.Vary": "'Origin'",1133                },1134            }1135        ]1136        api_id = self.create_api_gateway_and_deploy(1137            integration_type="MOCK", integration_responses=resps1138        )1139        url = gateway_request_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/")1140        result = requests.options(url)1141        self.assertLess(result.status_code, 400)1142        self.assertEqual("Origin", result.headers.get("vary"))1143        self.assertEqual("POST,OPTIONS", result.headers.get("Access-Control-Allow-Methods"))1144    # =====================================================================1145    # Helper methods1146    # =====================================================================1147    def connect_api_gateway_to_s3(self, bucket_name, file_name, api_id, method):1148        """Connects the root resource of an api gateway to the given object of an s3 bucket."""1149        apigw_client = aws_stack.connect_to_service("apigateway")1150        s3_uri = "arn:aws:apigateway:{}:s3:path/{}/{}".format(1151            aws_stack.get_region(), bucket_name, file_name1152        )1153        test_role = "test-s3-role"1154        role_arn = aws_stack.role_arn(role_name=test_role)1155        resources = apigw_client.get_resources(restApiId=api_id)1156        # using the root resource '/' directly for this test1157        root_resource_id = resources["items"][0]["id"]1158        apigw_client.put_method(1159            restApiId=api_id,1160            resourceId=root_resource_id,1161            httpMethod=method,1162            authorizationType="NONE",1163            apiKeyRequired=False,1164            requestParameters={},1165        )1166        apigw_client.put_integration(1167            restApiId=api_id,1168            resourceId=root_resource_id,1169            httpMethod=method,1170            type="AWS",1171            integrationHttpMethod=method,1172            uri=s3_uri,1173            credentials=role_arn,1174        )1175    def connect_api_gateway_to_kinesis(self, gateway_name, kinesis_stream):1176        resources = {}1177        template = self.APIGATEWAY_DATA_INBOUND_TEMPLATE % kinesis_stream1178        resource_path = self.API_PATH_DATA_INBOUND.replace("/", "")1179        resources[resource_path] = [1180            {1181                "httpMethod": "POST",1182                "authorizationType": "NONE",1183                "integrations": [1184                    {1185                        "type": "AWS",1186                        "uri": "arn:aws:apigateway:%s:kinesis:action/PutRecords"1187                        % aws_stack.get_region(),1188                        "requestTemplates": {"application/json": template},1189                    }1190                ],1191            },1192            {1193                "httpMethod": "GET",1194                "authorizationType": "NONE",1195                "integrations": [1196                    {1197                        "type": "AWS",1198                        "uri": "arn:aws:apigateway:%s:kinesis:action/ListStreams"1199                        % aws_stack.get_region(),1200                        "requestTemplates": {"application/json": "{}"},1201                    }1202                ],1203            },1204        ]1205        return aws_stack.create_api_gateway(1206            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1207        )1208    def connect_api_gateway_to_http(1209        self, int_type, gateway_name, target_url, methods=[], path=None1210    ):1211        if not methods:1212            methods = ["GET", "POST"]1213        if not path:1214            path = "/"1215        resources = {}1216        resource_path = path.replace("/", "")1217        resources[resource_path] = []1218        req_templates = (1219            {"application/json": json.dumps({"foo": "bar"})} if int_type == "custom" else {}1220        )1221        for method in methods:1222            resources[resource_path].append(1223                {1224                    "httpMethod": method,1225                    "integrations": [1226                        {1227                            "type": "HTTP" if int_type == "custom" else "HTTP_PROXY",1228                            "uri": target_url,1229                            "requestTemplates": req_templates,1230                            "responseTemplates": {},1231                        }1232                    ],1233                }1234            )1235        return aws_stack.create_api_gateway(1236            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1237        )1238    @staticmethod1239    def create_lambda_function(fn_name):1240        testutil.create_lambda_function(1241            handler_file=TEST_LAMBDA_PYTHON, libs=TEST_LAMBDA_LIBS, func_name=fn_name1242        )1243    def test_apigw_test_invoke_method_api(self):1244        client = aws_stack.connect_to_service("apigateway")1245        lambda_client = aws_stack.connect_to_service("lambda")1246        # create test Lambda1247        fn_name = f"test-{short_uid()}"1248        testutil.create_lambda_function(1249            handler_file=TEST_LAMBDA_HANDLER_JS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X1250        )1251        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1252        # create REST API and test resource1253        rest_api = client.create_rest_api(name="test", description="test")1254        root_resource = client.get_resources(restApiId=rest_api["id"])1255        resource = client.create_resource(1256            restApiId=rest_api["id"], parentId=root_resource["items"][0]["id"], pathPart="foo"1257        )1258        # create method and integration1259        client.put_method(1260            restApiId=rest_api["id"],1261            resourceId=resource["id"],1262            httpMethod="GET",1263            authorizationType="NONE",1264        )1265        client.put_integration(1266            restApiId=rest_api["id"],1267            resourceId=resource["id"],1268            httpMethod="GET",1269            integrationHttpMethod="GET",1270            type="AWS",1271            uri="arn:aws:apigateway:{}:lambda:path//2015-03-31/functions/{}/invocations".format(1272                aws_stack.get_region(), lambda_arn_11273            ),1274        )1275        # run test_invoke_method API #11276        response = client.test_invoke_method(1277            restApiId=rest_api["id"],1278            resourceId=resource["id"],1279            httpMethod="GET",1280            pathWithQueryString="/foo",1281        )1282        self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])1283        self.assertEqual(200, response.get("status"))1284        self.assertIn("response from", response.get("body"))1285        # run test_invoke_method API #21286        response = client.test_invoke_method(1287            restApiId=rest_api["id"],1288            resourceId=resource["id"],1289            httpMethod="GET",1290            pathWithQueryString="/foo",1291            body='{"test": "val123"}',1292            headers={"content-type": "application/json"},1293        )1294        self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])1295        self.assertEqual(200, response.get("status"))1296        self.assertIn("response from", response.get("body"))1297        self.assertIn("val123", response.get("body"))1298        # Clean up1299        lambda_client.delete_function(FunctionName=fn_name)1300        client.delete_rest_api(restApiId=rest_api["id"])1301    @staticmethod1302    def start_http_backend(test_port):1303        # test listener for target HTTP backend1304        class TestListener(ProxyListener):1305            def forward_request(self, **kwargs):1306                response = Response()1307                response.status_code = 2001308                result = {1309                    "data": kwargs.get("data") or "{}",1310                    "headers": dict(kwargs.get("headers")),1311                }1312                response._content = json.dumps(json_safe(result))1313                return response1314        proxy = start_proxy(test_port, update_listener=TestListener())1315        return proxy1316    @staticmethod1317    def create_api_gateway_and_deploy(1318        response_templates=None,1319        is_api_key_required=False,1320        integration_type=None,1321        integration_responses=None,1322    ):1323        response_templates = response_templates or {}1324        integration_type = integration_type or "AWS_PROXY"1325        apigw_client = aws_stack.connect_to_service("apigateway")1326        response = apigw_client.create_rest_api(name="my_api", description="this is my api")1327        api_id = response["id"]1328        resources = apigw_client.get_resources(restApiId=api_id)1329        root_resources = [resource for resource in resources["items"] if resource["path"] == "/"]1330        root_id = root_resources[0]["id"]1331        kwargs = {}...
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.