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

How to use apigateway_invocations_arn method in localstack

Best Python code snippet using localstack_python

test_apigateway.py

Source:test_apigateway.py

...523        # create Lambda function524        lambda_name = f"apigw-lambda-{short_uid()}"525        self.create_lambda_function(lambda_name)526        lambda_uri = aws_stack.lambda_function_arn(lambda_name)527        target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)528        # create REST API529        api = apigw_client.create_rest_api(name="test-api", description="")530        api_id = api["id"]531        root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]532        api_resource = apigw_client.create_resource(533            restApiId=api_id, parentId=root_res_id, pathPart="test"534        )535        apigw_client.put_method(536            restApiId=api_id,537            resourceId=api_resource["id"],538            httpMethod="GET",539            authorizationType="NONE",540        )541        rs = apigw_client.put_integration(542            restApiId=api_id,543            resourceId=api_resource["id"],544            httpMethod="GET",545            integrationHttpMethod="POST",546            type="AWS",547            uri=target_uri,548            timeoutInMillis=3000,549            contentHandling="CONVERT_TO_BINARY",550            requestTemplates={"application/json": '{"param1": "$input.params(\'param1\')"}'},551        )552        integration_keys = [553            "httpMethod",554            "type",555            "cacheKeyParameters",556            "uri",557            "cacheNamespace",558            "timeoutInMillis",559            "contentHandling",560            "requestParameters",561        ]562        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]563        for key in integration_keys:564            assert key in rs565        assert "responseTemplates" not in rs566        apigw_client.create_deployment(restApiId=api_id, stageName=self.TEST_STAGE_NAME)567        rs = apigw_client.get_integration(568            restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"569        )570        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]571        assert "AWS" == rs["type"]572        assert "POST" == rs["httpMethod"]573        assert target_uri == rs["uri"]574        # invoke the gateway endpoint575        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/test")576        response = requests.get(f"{url}?param1=foobar")577        assert response.status_code < 400578        content = response.json()579        assert '{"param1": "foobar"}' == content.get("body").get("body")580        # additional checks from https://github.com/localstack/localstack/issues/5041581        # pass Signature param582        response = requests.get(f"{url}?param1=foobar&Signature=1")583        assert response.status_code == 200584        content = response.json()585        assert '{"param1": "foobar"}' == content.get("body").get("body")586        assert {"Signature": "1", "param1": "foobar"} == content.get("body").get(587            "queryStringParameters"588        )589        # delete integration590        rs = apigw_client.delete_integration(591            restApiId=api_id,592            resourceId=api_resource["id"],593            httpMethod="GET",594        )595        assert 204 == rs["ResponseMetadata"]["HTTPStatusCode"]596        with pytest.raises(ClientError) as ctx:597            # This call should not be successful as the integration is deleted598            apigw_client.get_integration(599                restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"600            )601        assert ctx.value.response["Error"]["Code"] == "NotFoundException"602        # clean up603        lambda_client = aws_stack.create_external_boto_client("lambda")604        lambda_client.delete_function(FunctionName=lambda_name)605        apigw_client.delete_rest_api(restApiId=api_id)606    def test_api_gateway_handle_domain_name(self):607        domain_name = "%s.example.com" % short_uid()608        apigw_client = aws_stack.create_external_boto_client("apigateway")609        rs = apigw_client.create_domain_name(domainName=domain_name)610        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]611        rs = apigw_client.get_domain_name(domainName=domain_name)612        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]613        assert domain_name == rs["domainName"]614        # clean up615        apigw_client.delete_domain_name(domainName=domain_name)616    def _test_api_gateway_lambda_proxy_integration_any_method(self, fn_name, path):617        self.create_lambda_function(fn_name)618        # create API Gateway and connect it to the Lambda proxy backend619        lambda_uri = aws_stack.lambda_function_arn(fn_name)620        target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)621        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(622            "test_gateway3",623            target_uri,624            methods=["ANY"],625            path=path,626            stage_name=self.TEST_STAGE_NAME,627        )628        # make test request to gateway and check response629        path = path.replace("{test_param1}", "foo1")630        url = path_based_url(api_id=result["id"], stage_name=self.TEST_STAGE_NAME, path=path)631        data = {}632        for method in ("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"):633            body = json.dumps(data) if method in ("POST", "PUT", "PATCH") else None634            result = getattr(requests, method.lower())(url, data=body)635            if method != "DELETE":636                assert 200 == result.status_code637                parsed_body = json.loads(to_str(result.content))638                assert method == parsed_body.get("httpMethod")639            else:640                assert 204 == result.status_code641    def test_apigateway_with_custom_authorization_method(self):642        apigw_client = aws_stack.create_external_boto_client("apigateway")643        # create Lambda function644        lambda_name = "apigw-lambda-%s" % short_uid()645        self.create_lambda_function(lambda_name)646        lambda_uri = aws_stack.lambda_function_arn(lambda_name)647        # create REST API648        api = apigw_client.create_rest_api(name="test-api", description="")649        api_id = api["id"]650        root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]651        # create authorizer at root resource652        authorizer = apigw_client.create_authorizer(653            restApiId=api_id,654            name="lambda_authorizer",655            type="TOKEN",656            authorizerUri="arn:aws:apigateway:us-east-1:lambda:path/ \657                2015-03-31/functions/{}/invocations".format(658                lambda_uri659            ),660            identitySource="method.request.header.Auth",661        )662        # create method with custom authorizer663        is_api_key_required = True664        method_response = apigw_client.put_method(665            restApiId=api_id,666            resourceId=root_res_id,667            httpMethod="GET",668            authorizationType="CUSTOM",669            authorizerId=authorizer["id"],670            apiKeyRequired=is_api_key_required,671        )672        assert authorizer["id"] == method_response["authorizerId"]673        # clean up674        lambda_client = aws_stack.create_external_boto_client("lambda")675        lambda_client.delete_function(FunctionName=lambda_name)676        apigw_client.delete_rest_api(restApiId=api_id)677    def test_create_model(self):678        client = aws_stack.create_external_boto_client("apigateway")679        response = client.create_rest_api(name="my_api", description="this is my api")680        rest_api_id = response["id"]681        dummy_rest_api_id = "_non_existing_"682        model_name = "testModel"683        description = "test model"684        content_type = "application/json"685        # success case with valid params686        response = client.create_model(687            restApiId=rest_api_id,688            name=model_name,689            description=description,690            contentType=content_type,691        )692        assert model_name == response["name"]693        assert description == response["description"]694        with pytest.raises(Exception) as ctx:695            client.create_model(696                restApiId=dummy_rest_api_id,697                name=model_name,698                description=description,699                contentType=content_type,700            )701        assert "NotFoundException" == ctx.value.response["Error"]["Code"]702        assert "Invalid Rest API Id specified" == ctx.value.response["Error"]["Message"]703        with pytest.raises(Exception) as ctx:704            client.create_model(705                restApiId=dummy_rest_api_id,706                name="",707                description=description,708                contentType=content_type,709            )710        assert "BadRequestException" == ctx.value.response["Error"]["Code"]711        assert "No Model Name specified" == ctx.value.response["Error"]["Message"]712        # clean up713        client.delete_rest_api(restApiId=rest_api_id)714    def test_get_api_models(self):715        client = aws_stack.create_external_boto_client("apigateway")716        response = client.create_rest_api(name="my_api", description="this is my api")717        rest_api_id = response["id"]718        model_name = "testModel"719        description = "test model"720        content_type = "application/json"721        # when no models are present722        result = client.get_models(restApiId=rest_api_id)723        assert [] == result["items"]724        # add a model725        client.create_model(726            restApiId=rest_api_id,727            name=model_name,728            description=description,729            contentType=content_type,730        )731        # get models after adding732        result = client.get_models(restApiId=rest_api_id)733        assert model_name == result["items"][0]["name"]734        assert description == result["items"][0]["description"]735        # clean up736        client.delete_rest_api(restApiId=rest_api_id)737    def test_request_validator(self):738        client = aws_stack.create_external_boto_client("apigateway")739        response = client.create_rest_api(name="my_api", description="this is my api")740        rest_api_id = response["id"]741        # CREATE742        name = "validator123"743        result = client.create_request_validator(restApiId=rest_api_id, name=name)744        assert 201 == result["ResponseMetadata"]["HTTPStatusCode"]745        validator_id = result["id"]746        # LIST747        result = client.get_request_validators(restApiId=rest_api_id)748        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]749        assert [{"id": validator_id, "name": name}] == result["items"]750        # GET751        result = client.get_request_validator(752            restApiId=rest_api_id, requestValidatorId=validator_id753        )754        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]755        assert select_attributes(result, ["id", "name"]) == {"id": validator_id, "name": name}756        # UPDATE757        result = client.update_request_validator(758            restApiId=rest_api_id, requestValidatorId=validator_id, patchOperations=[]759        )760        # DELETE761        client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)762        with pytest.raises(Exception):763            client.get_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)764        with pytest.raises(Exception):765            client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)766        # clean up767        client.delete_rest_api(restApiId=rest_api_id)768    def test_base_path_mapping(self):769        client = aws_stack.create_external_boto_client("apigateway")770        response = client.create_rest_api(name="my_api", description="this is my api")771        rest_api_id = response["id"]772        # CREATE773        domain_name = "domain1.example.com"774        client.create_domain_name(domainName=domain_name)775        root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]776        res_id = client.create_resource(777            restApiId=rest_api_id, parentId=root_res_id, pathPart="path"778        )["id"]779        client.put_method(780            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"781        )782        client.put_integration(783            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"784        )785        depl_id = client.create_deployment(restApiId=rest_api_id)["id"]786        client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")787        base_path = "foo"788        result = client.create_base_path_mapping(789            domainName=domain_name,790            basePath=base_path,791            restApiId=rest_api_id,792            stage="dev",793        )794        assert result["ResponseMetadata"]["HTTPStatusCode"] in [200, 201]795        # LIST796        result = client.get_base_path_mappings(domainName=domain_name)797        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]798        expected = {"basePath": base_path, "restApiId": rest_api_id, "stage": "dev"}799        assert [expected] == result["items"]800        # GET801        result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)802        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]803        assert expected == select_attributes(result, ["basePath", "restApiId", "stage"])804        # UPDATE805        result = client.update_base_path_mapping(806            domainName=domain_name, basePath=base_path, patchOperations=[]807        )808        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]809        # DELETE810        client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)811        with pytest.raises(Exception):812            client.get_base_path_mapping(domainName=domain_name, basePath=base_path)813        with pytest.raises(Exception):814            client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)815    def test_base_path_mapping_root(self):816        client = aws_stack.create_external_boto_client("apigateway")817        response = client.create_rest_api(name="my_api2", description="this is my api")818        rest_api_id = response["id"]819        # CREATE820        domain_name = "domain2.example.com"821        client.create_domain_name(domainName=domain_name)822        root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]823        res_id = client.create_resource(824            restApiId=rest_api_id, parentId=root_res_id, pathPart="path"825        )["id"]826        client.put_method(827            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"828        )829        client.put_integration(830            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"831        )832        depl_id = client.create_deployment(restApiId=rest_api_id)["id"]833        client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")834        result = client.create_base_path_mapping(835            domainName=domain_name,836            basePath="",837            restApiId=rest_api_id,838            stage="dev",839        )840        assert result["ResponseMetadata"]["HTTPStatusCode"] in [200, 201]841        base_path = "(none)"842        # LIST843        result = client.get_base_path_mappings(domainName=domain_name)844        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]845        expected = {"basePath": "(none)", "restApiId": rest_api_id, "stage": "dev"}846        assert [expected] == result["items"]847        # GET848        result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)849        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]850        assert expected == select_attributes(result, ["basePath", "restApiId", "stage"])851        # UPDATE852        result = client.update_base_path_mapping(853            domainName=domain_name, basePath=base_path, patchOperations=[]854        )855        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]856        # DELETE857        client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)858        with pytest.raises(Exception):859            client.get_base_path_mapping(domainName=domain_name, basePath=base_path)860        with pytest.raises(Exception):861            client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)862    def test_api_account(self):863        client = aws_stack.create_external_boto_client("apigateway")864        response = client.create_rest_api(name="my_api", description="test 123")865        rest_api_id = response["id"]866        result = client.get_account()867        assert "UsagePlans" in result["features"]868        result = client.update_account(869            patchOperations=[{"op": "add", "path": "/features/-", "value": "foobar"}]870        )871        assert "foobar" in result["features"]872        # clean up873        client.delete_rest_api(restApiId=rest_api_id)874    def test_get_model_by_name(self):875        client = aws_stack.create_external_boto_client("apigateway")876        response = client.create_rest_api(name="my_api", description="this is my api")877        rest_api_id = response["id"]878        dummy_rest_api_id = "_non_existing_"879        model_name = "testModel"880        description = "test model"881        content_type = "application/json"882        # add a model883        client.create_model(884            restApiId=rest_api_id,885            name=model_name,886            description=description,887            contentType=content_type,888        )889        # get models after adding890        result = client.get_model(restApiId=rest_api_id, modelName=model_name)891        assert model_name == result["name"]892        assert description == result["description"]893        try:894            client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)895            self.fail("This call should not be successful as the model is not created.")896        except ClientError as e:897            assert "NotFoundException" == e.response["Error"]["Code"]898            assert "Invalid Rest API Id specified" == e.response["Error"]["Message"]899    def test_get_model_with_invalid_name(self):900        client = aws_stack.create_external_boto_client("apigateway")901        response = client.create_rest_api(name="my_api", description="this is my api")902        rest_api_id = response["id"]903        # test with an invalid model name904        try:905            client.get_model(restApiId=rest_api_id, modelName="fake")906            self.fail("This call should not be successful as the model is not created.")907        except ClientError as e:908            assert "NotFoundException" == e.response["Error"]["Code"]909        # clean up910        client.delete_rest_api(restApiId=rest_api_id)911    def test_put_integration_dynamodb_proxy_validation_without_response_template(self):912        api_id = self.create_api_gateway_and_deploy({})913        url = path_based_url(api_id=api_id, stage_name="staging", path="/")914        response = requests.put(915            url,916            json.dumps({"id": "id1", "data": "foobar123"}),917        )918        assert 404 == response.status_code919    def test_put_integration_dynamodb_proxy_validation_with_response_template(self):920        response_templates = {921            "application/json": json.dumps(922                {923                    "TableName": "MusicCollection",924                    "Item": {"id": "$.Id", "data": "$.data"},925                }926            )927        }928        api_id = self.create_api_gateway_and_deploy(response_templates)929        url = path_based_url(api_id=api_id, stage_name="staging", path="/")930        response = requests.put(931            url,932            json.dumps({"id": "id1", "data": "foobar123"}),933        )934        assert 200 == response.status_code935        dynamo_client = aws_stack.connect_to_resource("dynamodb")936        table = dynamo_client.Table("MusicCollection")937        result = table.get_item(Key={"id": "id1"})938        assert "foobar123" == result["Item"]["data"]939    def test_api_key_required_for_methods(self):940        response_templates = {941            "application/json": json.dumps(942                {943                    "TableName": "MusicCollection",944                    "Item": {"id": "$.Id", "data": "$.data"},945                }946            )947        }948        api_id = self.create_api_gateway_and_deploy(response_templates, True)949        url = path_based_url(api_id=api_id, stage_name="staging", path="/")950        payload = {951            "name": "TEST-PLAN-2",952            "description": "Description",953            "quota": {"limit": 10, "period": "DAY", "offset": 0},954            "throttle": {"rateLimit": 2, "burstLimit": 1},955            "apiStages": [{"apiId": api_id, "stage": "staging"}],956            "tags": {"tag_key": "tag_value"},957        }958        client = aws_stack.create_external_boto_client("apigateway")959        usage_plan_id = client.create_usage_plan(**payload)["id"]960        key_name = "testApiKey"961        key_type = "API_KEY"962        api_key = client.create_api_key(name=key_name)963        payload = {964            "usagePlanId": usage_plan_id,965            "keyId": api_key["id"],966            "keyType": key_type,967        }968        client.create_usage_plan_key(**payload)969        response = requests.put(970            url,971            json.dumps({"id": "id1", "data": "foobar123"}),972        )973        # when the api key is not passed as part of the header974        assert 403 == response.status_code975        response = requests.put(976            url,977            json.dumps({"id": "id1", "data": "foobar123"}),978            headers={"X-API-Key": api_key["value"]},979        )980        # when the api key is passed as part of the header981        assert 200 == response.status_code982    def test_multiple_api_keys_validate(self):983        response_templates = {984            "application/json": json.dumps(985                {986                    "TableName": "MusicCollection",987                    "Item": {"id": "$.Id", "data": "$.data"},988                }989            )990        }991        api_id = self.create_api_gateway_and_deploy(response_templates, True)992        url = path_based_url(api_id=api_id, stage_name="staging", path="/")993        client = aws_stack.create_external_boto_client("apigateway")994        # Create multiple usage plans995        usage_plan_ids = []996        for i in range(2):997            payload = {998                "name": "APIKEYTEST-PLAN-{}".format(i),999                "description": "Description",1000                "quota": {"limit": 10, "period": "DAY", "offset": 0},1001                "throttle": {"rateLimit": 2, "burstLimit": 1},1002                "apiStages": [{"apiId": api_id, "stage": "staging"}],1003                "tags": {"tag_key": "tag_value"},1004            }1005            usage_plan_ids.append(client.create_usage_plan(**payload)["id"])1006        api_keys = []1007        key_type = "API_KEY"1008        # Create multiple API Keys in each usage plan1009        for usage_plan_id in usage_plan_ids:1010            for i in range(2):1011                api_key = client.create_api_key(name="testMultipleApiKeys{}".format(i))1012                payload = {1013                    "usagePlanId": usage_plan_id,1014                    "keyId": api_key["id"],1015                    "keyType": key_type,1016                }1017                client.create_usage_plan_key(**payload)1018                api_keys.append(api_key["value"])1019        response = requests.put(1020            url,1021            json.dumps({"id": "id1", "data": "foobar123"}),1022        )1023        # when the api key is not passed as part of the header1024        assert 403 == response.status_code1025        # check that all API keys work1026        for key in api_keys:1027            response = requests.put(1028                url,1029                json.dumps({"id": "id1", "data": "foobar123"}),1030                headers={"X-API-Key": key},1031            )1032            # when the api key is passed as part of the header1033            assert 200 == response.status_code1034    @pytest.mark.parametrize("base_path_type", ["ignore", "prepend", "split"])1035    def test_import_rest_api(self, base_path_type):1036        rest_api_name = f"restapi-{short_uid()}"1037        client = aws_stack.create_external_boto_client("apigateway")1038        rest_api_id = client.create_rest_api(name=rest_api_name)["id"]1039        spec_file = load_file(TEST_SWAGGER_FILE_JSON)1040        api_params = {"basepath": base_path_type}1041        rs = client.put_rest_api(1042            restApiId=rest_api_id, body=spec_file, mode="overwrite", parameters=api_params1043        )1044        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]1045        resources = client.get_resources(restApiId=rest_api_id)1046        for rv in resources.get("items"):1047            for method in rv.get("resourceMethods", {}).values():1048                assert method.get("authorizationType") == "request"1049                assert method.get("authorizerId") is not None1050        spec_file = load_file(TEST_SWAGGER_FILE_YAML)1051        rs = client.put_rest_api(1052            restApiId=rest_api_id, body=spec_file, mode="overwrite", parameters=api_params1053        )1054        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]1055        rs = client.get_resources(restApiId=rest_api_id)1056        expected_resources = 2 if base_path_type == "ignore" else 31057        assert len(rs["items"]) == expected_resources1058        abs_path = "/test" if base_path_type == "ignore" else "/base/test"1059        resource = [res for res in rs["items"] if res["path"] == abs_path][0]1060        assert "GET" in resource["resourceMethods"]1061        assert "requestParameters" in resource["resourceMethods"]["GET"]1062        assert {"integration.request.header.X-Amz-Invocation-Type": "'Event'"} == resource[1063            "resourceMethods"1064        ]["GET"]["requestParameters"]1065        url = path_based_url(api_id=rest_api_id, stage_name="dev", path=abs_path)1066        response = requests.get(url)1067        assert 200 == response.status_code1068        # clean up1069        client.delete_rest_api(restApiId=rest_api_id)1070        spec_file = load_file(TEST_IMPORT_REST_API_FILE)1071        rs = client.import_rest_api(body=spec_file, parameters=api_params)1072        assert 201 == rs["ResponseMetadata"]["HTTPStatusCode"]1073        rest_api_id = rs["id"]1074        rs = client.get_resources(restApiId=rest_api_id)1075        resources = rs["items"]1076        assert 3 == len(resources)1077        paths = [res["path"] for res in resources]1078        assert "/" in paths1079        assert "/pets" in paths1080        assert "/pets/{petId}" in paths1081        # clean up1082        client.delete_rest_api(restApiId=rest_api_id)1083    def test_step_function_integrations(self):1084        client = aws_stack.create_external_boto_client("apigateway")1085        sfn_client = aws_stack.create_external_boto_client("stepfunctions")1086        lambda_client = aws_stack.create_external_boto_client("lambda")1087        state_machine_name = "test"1088        state_machine_def = {1089            "Comment": "Hello World example",1090            "StartAt": "step1",1091            "States": {1092                "step1": {"Type": "Task", "Resource": "__tbd__", "End": True},1093            },1094        }1095        # create state machine1096        fn_name = "test-stepfunctions-apigw"1097        testutil.create_lambda_function(1098            handler_file=TEST_LAMBDA_PYTHON_ECHO,1099            func_name=fn_name,1100            runtime=LAMBDA_RUNTIME_PYTHON36,1101        )1102        role_arn = aws_stack.role_arn("sfn_role")1103        # create state machine definition1104        definition = clone(state_machine_def)1105        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1106        definition["States"]["step1"]["Resource"] = lambda_arn_11107        definition = json.dumps(definition)1108        # create state machine1109        result = sfn_client.create_state_machine(1110            name=state_machine_name, definition=definition, roleArn=role_arn1111        )1112        sm_arn = result["stateMachineArn"]1113        # create REST API and method1114        rest_api = client.create_rest_api(name="test", description="test")1115        resources = client.get_resources(restApiId=rest_api["id"])1116        root_resource_id = resources["items"][0]["id"]1117        client.put_method(1118            restApiId=rest_api["id"],1119            resourceId=root_resource_id,1120            httpMethod="POST",1121            authorizationType="NONE",1122        )1123        def _prepare_method_integration(1124            integr_kwargs=None, resp_templates=None, action="StartExecution", overwrite=False1125        ):1126            if integr_kwargs is None:1127                integr_kwargs = {}1128            if resp_templates is None:1129                resp_templates = {}1130            if overwrite:1131                client.delete_integration(1132                    restApiId=rest_api["id"],1133                    resourceId=resources["items"][0]["id"],1134                    httpMethod="POST",1135                )1136            uri = f"arn:aws:apigateway:{aws_stack.get_region()}:states:action/{action}"1137            client.put_integration(1138                restApiId=rest_api["id"],1139                resourceId=root_resource_id,1140                httpMethod="POST",1141                integrationHttpMethod="POST",1142                type="AWS",1143                uri=uri,1144                **integr_kwargs,1145            )1146            if resp_templates:1147                client.put_integration_response(1148                    restApiId=rest_api["id"],1149                    resourceId=root_resource_id,1150                    selectionPattern="",1151                    responseTemplates=resp_templates,1152                    httpMethod="POST",1153                    statusCode="200",1154                )1155        # STEP 1: test integration with request template1156        _prepare_method_integration(1157            integr_kwargs={1158                "requestTemplates": {1159                    "application/json": """1160                    #set($data = $util.escapeJavaScript($input.json('$')))1161                    {"input": $data, "stateMachineArn": "%s"}1162                    """1163                    % sm_arn1164                }1165            }1166        )1167        # invoke stepfunction via API GW, assert results1168        client.create_deployment(restApiId=rest_api["id"], stageName="dev")1169        url = path_based_url(api_id=rest_api["id"], stage_name="dev", path="/")1170        test_data = {"test": "test-value"}1171        resp = requests.post(url, data=json.dumps(test_data))1172        assert 200 == resp.status_code1173        assert "executionArn" in resp.content.decode()1174        assert "startDate" in resp.content.decode()1175        # STEP 2: test integration without request template1176        _prepare_method_integration(overwrite=True)1177        test_data_1 = {1178            "input": json.dumps(test_data),1179            "name": "MyExecution",1180            "stateMachineArn": sm_arn,1181        }1182        # invoke stepfunction via API GW, assert results1183        resp = requests.post(url, data=json.dumps(test_data_1))1184        assert 200 == resp.status_code1185        assert "executionArn" in resp.content.decode()1186        assert "startDate" in resp.content.decode()1187        # STEP 3: test integration with synchronous execution1188        _prepare_method_integration(overwrite=True, action="StartSyncExecution")1189        # invoke stepfunction via API GW, assert results1190        test_data_1["name"] += "1"1191        resp = requests.post(url, data=json.dumps(test_data_1))1192        assert 200 == resp.status_code1193        content = json.loads(to_str(resp.content.decode()))1194        assert "SUCCEEDED" == content.get("status")1195        assert test_data == json.loads(content.get("output"))1196        # STEP 4: test integration with synchronous execution and response templates1197        resp_templates = {APPLICATION_JSON: "$input.path('$.output')"}1198        _prepare_method_integration(1199            resp_templates=resp_templates, overwrite=True, action="StartSyncExecution"1200        )1201        # invoke stepfunction via API GW, assert results1202        test_data_1["name"] += "2"1203        resp = requests.post(url, data=json.dumps(test_data_1))1204        assert 200 == resp.status_code1205        assert test_data == json.loads(to_str(resp.content.decode()))1206        _prepare_method_integration(overwrite=True, action="DeleteStateMachine")1207        # Remove state machine with API GW1208        resp = requests.post(url, data=json.dumps({"stateMachineArn": sm_arn}))1209        assert 200 == resp.status_code1210        # Clean up1211        lambda_client.delete_function(FunctionName=fn_name)1212        client.delete_rest_api(restApiId=rest_api["id"])1213    def test_api_gateway_http_integration_with_path_request_parameter(self):1214        client = aws_stack.create_external_boto_client("apigateway")1215        test_port = get_free_tcp_port()1216        backend_url = "http://localhost:%s/person/{id}" % test_port1217        # start test HTTP backend1218        proxy = self.start_http_backend(test_port)1219        # create rest api1220        api_rest = client.create_rest_api(name="test")1221        api_id = api_rest["id"]1222        parent_response = client.get_resources(restApiId=api_id)1223        parent_id = parent_response["items"][0]["id"]1224        resource_1 = client.create_resource(restApiId=api_id, parentId=parent_id, pathPart="person")1225        resource_1_id = resource_1["id"]1226        resource_2 = client.create_resource(1227            restApiId=api_id, parentId=resource_1_id, pathPart="{id}"1228        )1229        resource_2_id = resource_2["id"]1230        client.put_method(1231            restApiId=api_id,1232            resourceId=resource_2_id,1233            httpMethod="GET",1234            authorizationType="NONE",1235            apiKeyRequired=False,1236            requestParameters={"method.request.path.id": True},1237        )1238        client.put_integration(1239            restApiId=api_id,1240            resourceId=resource_2_id,1241            httpMethod="GET",1242            integrationHttpMethod="GET",1243            type="HTTP",1244            uri=backend_url,1245            timeoutInMillis=3000,1246            contentHandling="CONVERT_TO_BINARY",1247            requestParameters={"integration.request.path.id": "method.request.path.id"},1248        )1249        client.create_deployment(restApiId=api_id, stageName="test")1250        def _test_invoke(url):1251            result = requests.get(url)1252            content = json.loads(to_str(result.content))1253            assert 200 == result.status_code1254            assert re.search(1255                "http://.*localhost.*/person/123",1256                content["headers"].get(HEADER_LOCALSTACK_REQUEST_URL),1257            )1258        for use_hostname in [True, False]:1259            for use_ssl in [True, False] if use_hostname else [False]:1260                url = self._get_invoke_endpoint(1261                    api_id,1262                    stage="test",1263                    path="/person/123",1264                    use_hostname=use_hostname,1265                    use_ssl=use_ssl,1266                )1267                _test_invoke(url)1268        # clean up1269        client.delete_rest_api(restApiId=api_id)1270        proxy.stop()1271    def _get_invoke_endpoint(1272        self, api_id, stage="test", path="/", use_hostname=False, use_ssl=False1273    ):1274        path = path or "/"1275        path = path if path.startswith(path) else f"/{path}"1276        proto = "https" if use_ssl else "http"1277        if use_hostname:1278            return f"{proto}://{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}/{stage}{path}"1279        return (1280            f"{proto}://localhost:{config.EDGE_PORT}/restapis/{api_id}/{stage}/_user_request_{path}"1281        )1282    def test_api_gateway_s3_get_integration(self):1283        apigw_client = aws_stack.create_external_boto_client("apigateway")1284        s3_client = aws_stack.create_external_boto_client("s3")1285        bucket_name = f"test-bucket-{short_uid()}"1286        apigateway_name = f"test-api-{short_uid()}"1287        object_name = "test.json"1288        object_content = '{ "success": "true" }'1289        object_content_type = "application/json"1290        api = apigw_client.create_rest_api(name=apigateway_name)1291        api_id = api["id"]1292        try:1293            aws_stack.get_or_create_bucket(bucket_name)1294            s3_client.put_object(1295                Bucket=bucket_name,1296                Key=object_name,1297                Body=object_content,1298                ContentType=object_content_type,1299            )1300            self.connect_api_gateway_to_s3(bucket_name, object_name, api_id, "GET")1301            apigw_client.create_deployment(restApiId=api_id, stageName="test")1302            url = path_based_url(api_id, "test", f"/{object_name}")1303            result = requests.get(url)1304            assert 200 == result.status_code1305            assert object_content == result.text1306            assert object_content_type == result.headers["content-type"]1307        finally:1308            # clean up1309            apigw_client.delete_rest_api(restApiId=api_id)1310            s3_client.delete_object(Bucket=bucket_name, Key=object_name)1311            s3_client.delete_bucket(Bucket=bucket_name)1312    def test_api_mock_integration_response_params(self):1313        resps = [1314            {1315                "statusCode": "204",1316                "httpMethod": "OPTIONS",1317                "responseParameters": {1318                    "method.response.header.Access-Control-Allow-Methods": "'POST,OPTIONS'",1319                    "method.response.header.Vary": "'Origin'",1320                },1321            }1322        ]1323        api_id = self.create_api_gateway_and_deploy(1324            integration_type="MOCK", integration_responses=resps1325        )1326        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/")1327        result = requests.options(url)1328        assert result.status_code < 4001329        assert "Origin" == result.headers.get("vary")1330        assert "POST,OPTIONS" == result.headers.get("Access-Control-Allow-Methods")1331    def test_api_gateway_update_resource_path_part(self):1332        apigw_client = aws_stack.connect_to_service("apigateway")1333        api = apigw_client.create_rest_api(name="test-api", description="")1334        api_id = api["id"]1335        root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]1336        api_resource = apigw_client.create_resource(1337            restApiId=api_id, parentId=root_res_id, pathPart="test"1338        )1339        response = apigw_client.update_resource(1340            restApiId=api_id,1341            resourceId=api_resource.get("id"),1342            patchOperations=[1343                {"op": "replace", "path": "/pathPart", "value": "demo1"},1344            ],1345        )1346        assert response.get("pathPart") == "demo1"1347        response = apigw_client.get_resource(restApiId=api_id, resourceId=api_resource.get("id"))1348        assert response.get("pathPart") == "demo1"1349        # clean up1350        apigw_client.delete_rest_api(restApiId=api_id)1351    def test_response_headers_invocation_with_apigw(self, create_lambda_function, lambda_client):1352        lambda_name = f"test_lambda_{short_uid()}"1353        lambda_resource = "/api/v1/{proxy+}"1354        lambda_path = "/api/v1/hello/world"1355        create_lambda_function(1356            func_name=lambda_name,1357            zip_file=testutil.create_zip_file(1358                TEST_LAMBDA_NODEJS_APIGW_INTEGRATION, get_content=True1359            ),1360            runtime=LAMBDA_RUNTIME_NODEJS14X,1361            handler="apigw_integration.handler",1362        )1363        lambda_uri = aws_stack.lambda_function_arn(lambda_name)1364        target_uri = f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path/2015-03-31/functions/{lambda_uri}/invocations"1365        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(1366            "test_gateway",1367            target_uri,1368            path=lambda_resource,1369            stage_name=TEST_STAGE_NAME,1370        )1371        api_id = result["id"]1372        url = path_based_url(api_id=api_id, stage_name=TEST_STAGE_NAME, path=lambda_path)1373        result = requests.get(url)1374        assert result.status_code == 3001375        assert result.headers["Content-Type"] == "application/xml"1376        body = xmltodict.parse(result.content)1377        assert body.get("message") == "completed"1378    # =====================================================================1379    # Helper methods1380    # =====================================================================1381    def connect_api_gateway_to_s3(self, bucket_name, file_name, api_id, method):1382        """Connects the root resource of an api gateway to the given object of an s3 bucket."""1383        apigw_client = aws_stack.create_external_boto_client("apigateway")1384        s3_uri = "arn:aws:apigateway:{}:s3:path/{}/{{proxy}}".format(1385            aws_stack.get_region(), bucket_name1386        )1387        test_role = "test-s3-role"1388        role_arn = aws_stack.role_arn(role_name=test_role)1389        resources = apigw_client.get_resources(restApiId=api_id)1390        # using the root resource '/' directly for this test1391        root_resource_id = resources["items"][0]["id"]1392        proxy_resource = apigw_client.create_resource(1393            restApiId=api_id, parentId=root_resource_id, pathPart="{proxy+}"1394        )1395        apigw_client.put_method(1396            restApiId=api_id,1397            resourceId=proxy_resource["id"],1398            httpMethod=method,1399            authorizationType="NONE",1400            apiKeyRequired=False,1401            requestParameters={},1402        )1403        apigw_client.put_integration(1404            restApiId=api_id,1405            resourceId=proxy_resource["id"],1406            httpMethod=method,1407            type="AWS",1408            integrationHttpMethod=method,1409            uri=s3_uri,1410            credentials=role_arn,1411            requestParameters={"integration.request.path.proxy": "method.request.path.proxy"},1412        )1413    def connect_api_gateway_to_kinesis(self, gateway_name, kinesis_stream):1414        template = self.APIGATEWAY_DATA_INBOUND_TEMPLATE % kinesis_stream1415        resource_path = self.API_PATH_DATA_INBOUND.replace("/", "")1416        resources = {1417            resource_path: [1418                {1419                    "httpMethod": "POST",1420                    "authorizationType": "NONE",1421                    "requestModels": {"application/json": "Empty"},1422                    "integrations": [1423                        {1424                            "type": "AWS",1425                            "uri": "arn:aws:apigateway:%s:kinesis:action/PutRecords"1426                            % aws_stack.get_region(),1427                            "requestTemplates": {"application/json": template},1428                        }1429                    ],1430                },1431                {1432                    "httpMethod": "GET",1433                    "authorizationType": "NONE",1434                    "requestModels": {"application/json": "Empty"},1435                    "integrations": [1436                        {1437                            "type": "AWS",1438                            "uri": "arn:aws:apigateway:%s:kinesis:action/ListStreams"1439                            % aws_stack.get_region(),1440                            "requestTemplates": {"application/json": "{}"},1441                        }1442                    ],1443                },1444            ]1445        }1446        return aws_stack.create_api_gateway(1447            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1448        )1449    def connect_api_gateway_to_http(1450        self, int_type, gateway_name, target_url, methods=None, path=None1451    ):1452        if methods is None:1453            methods = []1454        if not methods:1455            methods = ["GET", "POST"]1456        if not path:1457            path = "/"1458        resources = {}1459        resource_path = path.replace("/", "")1460        req_templates = (1461            {"application/json": json.dumps({"foo": "bar"})} if int_type == "custom" else {}1462        )1463        resources[resource_path] = [1464            {1465                "httpMethod": method,1466                "integrations": [1467                    {1468                        "type": "HTTP" if int_type == "custom" else "HTTP_PROXY",1469                        "uri": target_url,1470                        "requestTemplates": req_templates,1471                        "responseTemplates": {},1472                    }1473                ],1474            }1475            for method in methods1476        ]1477        return aws_stack.create_api_gateway(1478            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1479        )1480    @staticmethod1481    def create_lambda_function(fn_name):1482        testutil.create_lambda_function(1483            handler_file=TEST_LAMBDA_PYTHON, libs=TEST_LAMBDA_LIBS, func_name=fn_name1484        )1485    def test_apigw_test_invoke_method_api(self):1486        client = aws_stack.create_external_boto_client("apigateway")1487        lambda_client = aws_stack.create_external_boto_client("lambda")1488        # create test Lambda1489        fn_name = f"test-{short_uid()}"1490        testutil.create_lambda_function(1491            handler_file=TEST_LAMBDA_NODEJS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X1492        )1493        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1494        # create REST API and test resource1495        rest_api = client.create_rest_api(name="test", description="test")1496        root_resource = client.get_resources(restApiId=rest_api["id"])1497        resource = client.create_resource(1498            restApiId=rest_api["id"], parentId=root_resource["items"][0]["id"], pathPart="foo"1499        )1500        # create method and integration1501        client.put_method(1502            restApiId=rest_api["id"],1503            resourceId=resource["id"],1504            httpMethod="GET",1505            authorizationType="NONE",1506        )1507        client.put_integration(1508            restApiId=rest_api["id"],1509            resourceId=resource["id"],1510            httpMethod="GET",1511            integrationHttpMethod="GET",1512            type="AWS",1513            uri="arn:aws:apigateway:{}:lambda:path//2015-03-31/functions/{}/invocations".format(1514                aws_stack.get_region(), lambda_arn_11515            ),1516        )1517        # run test_invoke_method API #11518        response = client.test_invoke_method(1519            restApiId=rest_api["id"],1520            resourceId=resource["id"],1521            httpMethod="GET",1522            pathWithQueryString="/foo",1523        )1524        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]1525        assert 200 == response.get("status")1526        assert "response from" in response.get("body").get("body")1527        # run test_invoke_method API #21528        response = client.test_invoke_method(1529            restApiId=rest_api["id"],1530            resourceId=resource["id"],1531            httpMethod="GET",1532            pathWithQueryString="/foo",1533            body='{"test": "val123"}',1534            headers={"content-type": "application/json"},1535        )1536        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]1537        assert 200 == response.get("status")1538        assert "response from" in response.get("body").get("body")1539        assert "val123" in response.get("body").get("body")1540        # Clean up1541        lambda_client.delete_function(FunctionName=fn_name)1542        client.delete_rest_api(restApiId=rest_api["id"])1543    @staticmethod1544    def start_http_backend(test_port):1545        # test listener for target HTTP backend1546        class TestListener(ProxyListener):1547            def forward_request(self, **kwargs):1548                response = Response()1549                response.status_code = 2001550                result = {1551                    "data": kwargs.get("data") or "{}",1552                    "headers": dict(kwargs.get("headers")),1553                }1554                response._content = json.dumps(json_safe(result))1555                return response1556        proxy = start_proxy(test_port, update_listener=TestListener())1557        return proxy1558    @staticmethod1559    def create_api_gateway_and_deploy(1560        response_templates=None,1561        is_api_key_required=False,1562        integration_type=None,1563        integration_responses=None,1564    ):1565        response_templates = response_templates or {}1566        integration_type = integration_type or "AWS_PROXY"1567        apigw_client = aws_stack.create_external_boto_client("apigateway")1568        response = apigw_client.create_rest_api(name="my_api", description="this is my api")1569        api_id = response["id"]1570        resources = apigw_client.get_resources(restApiId=api_id)1571        root_resources = [resource for resource in resources["items"] if resource["path"] == "/"]1572        root_id = root_resources[0]["id"]1573        kwargs = {}1574        if integration_type == "AWS_PROXY":1575            aws_stack.create_dynamodb_table("MusicCollection", partition_key="id")1576            kwargs[1577                "uri"1578            ] = "arn:aws:apigateway:us-east-1:dynamodb:action/PutItem&Table=MusicCollection"1579        if not integration_responses:1580            integration_responses = [{"httpMethod": "PUT", "statusCode": "200"}]1581        for resp_details in integration_responses:1582            apigw_client.put_method(1583                restApiId=api_id,1584                resourceId=root_id,1585                httpMethod=resp_details["httpMethod"],1586                authorizationType="NONE",1587                apiKeyRequired=is_api_key_required,1588            )1589            apigw_client.put_method_response(1590                restApiId=api_id,1591                resourceId=root_id,1592                httpMethod=resp_details["httpMethod"],1593                statusCode="200",1594            )1595            apigw_client.put_integration(1596                restApiId=api_id,1597                resourceId=root_id,1598                httpMethod=resp_details["httpMethod"],1599                integrationHttpMethod=resp_details["httpMethod"],1600                type=integration_type,1601                **kwargs,1602            )1603            apigw_client.put_integration_response(1604                restApiId=api_id,1605                resourceId=root_id,1606                selectionPattern="",1607                responseTemplates=response_templates,1608                **resp_details,1609            )1610        apigw_client.create_deployment(restApiId=api_id, stageName="staging")1611        return api_id1612def test_import_swagger_api(apigateway_client):1613    apigateway_client.get_rest_apis()1614    api_spec = load_test_resource("openapi.swagger.json")1615    api_spec_dict = json.loads(api_spec)1616    backend = apigateway_backends["eu-west-1"]1617    api_model = backend.create_rest_api(name="", description="")1618    imported_api = import_api_from_openapi_spec(api_model, api_spec_dict, {})1619    # test_cfn_handle_serverless_api_resource fails if we support title1620    # assert imported_api.name == api_spec_dict.get("info").get("title")1621    assert imported_api.description == api_spec_dict.get("info").get("description")1622    # assert that are no multiple authorizers1623    assert len(imported_api.authorizers) == 11624    paths = {v.path_part for k, v in imported_api.resources.items()}1625    assert paths == {"/", "pets", "{petId}"}1626    resource_methods = {v.path_part: v.resource_methods for k, v in imported_api.resources.items()}1627    methods = {kk[0] for k, v in resource_methods.items() for kk in v.items()}1628    assert methods == {"POST", "OPTIONS", "GET"}1629    assert resource_methods.get("/").get("GET")["methodResponses"] == {1630        "200": {1631            "statusCode": "200",1632            "responseModels": None,1633            "responseParameters": {"method.response.header.Content-Type": "'text/html'"},1634        }1635    }1636    assert resource_methods.get("pets").get("GET")["methodResponses"] == {1637        "200": {1638            "responseModels": {1639                "application/json": {1640                    "items": {1641                        "properties": {1642                            "id": {"type": "integer"},1643                            "price": {"type": "number"},1644                            "type": {"type": "string"},1645                        },1646                        "type": "object",1647                    },1648                    "type": "array",1649                }1650            },1651            "responseParameters": {"method.response.header.Access-Control-Allow-Origin": "'*'"},1652            "statusCode": "200",1653        }1654    }1655@pytest.mark.skipif(not use_docker(), reason="Rust lambdas cannot be executed in local executor")1656def test_apigateway_rust_lambda(1657    apigateway_client, create_lambda_function, create_iam_role_with_policy1658):1659    function_name = f"test-rust-function-{short_uid()}"1660    api_gateway_name = f"api_gateway_{short_uid()}"1661    role_name = f"test_apigateway_role_{short_uid()}"1662    policy_name = f"test_apigateway_policy_{short_uid()}"1663    stage_name = "test"1664    first_name = f"test_name_{short_uid()}"1665    lambda_create_response = create_lambda_function(1666        func_name=function_name,1667        zip_file=load_file(TEST_LAMBDA_HTTP_RUST, mode="rb"),1668        handler="bootstrap.is.the.handler",1669        runtime="provided.al2",1670    )1671    role_arn = create_iam_role_with_policy(1672        RoleName=role_name,1673        PolicyName=policy_name,1674        RoleDefinition=APIGATEWAY_ASSUME_ROLE_POLICY,1675        PolicyDefinition=APIGATEWAY_LAMBDA_POLICY,1676    )1677    lambda_arn = lambda_create_response["CreateFunctionResponse"]["FunctionArn"]1678    rest_api_id = apigateway_client.create_rest_api(name=api_gateway_name)["id"]1679    try:1680        root_resource_id = apigateway_client.get_resources(restApiId=rest_api_id)["items"][0]["id"]1681        apigateway_client.put_method(1682            restApiId=rest_api_id,1683            resourceId=root_resource_id,1684            httpMethod="GET",1685            authorizationType="NONE",1686        )1687        apigateway_client.put_method_response(1688            restApiId=rest_api_id, resourceId=root_resource_id, httpMethod="GET", statusCode="200"1689        )1690        lambda_target_uri = aws_stack.apigateway_invocations_arn(1691            lambda_uri=lambda_arn, region_name=apigateway_client.meta.region_name1692        )1693        apigateway_client.put_integration(1694            restApiId=rest_api_id,1695            resourceId=root_resource_id,1696            httpMethod="GET",1697            type="AWS_PROXY",1698            integrationHttpMethod="POST",1699            uri=lambda_target_uri,1700            credentials=role_arn,1701        )1702        apigateway_client.create_deployment(restApiId=rest_api_id, stageName=stage_name)1703        url = path_based_url(1704            api_id=rest_api_id, stage_name=stage_name, path=f"/?first_name={first_name}"1705        )1706        result = requests.get(url)1707        assert result.text == f"Hello, {first_name}!"1708    finally:1709        apigateway_client.delete_rest_api(restApiId=rest_api_id)1710def test_apigw_call_api_with_aws_endpoint_url():1711    headers = aws_stack.mock_aws_request_headers("apigateway")1712    headers["Host"] = "apigateway.us-east-2.amazonaws.com:4566"1713    url = f"{get_edge_url()}/apikeys?includeValues=true&name=test%40example.org"1714    response = requests.get(url, headers=headers)1715    assert response.ok1716    content = json.loads(to_str(response.content))1717    assert isinstance(content.get("item"), list)1718@pytest.mark.parametrize("method", ["GET", "ANY"])1719@pytest.mark.parametrize("url_function", [path_based_url, host_based_url])1720def test_rest_api_multi_region(method, url_function):1721    apigateway_client_eu = _client("apigateway", region_name="eu-west-1")1722    apigateway_client_us = _client("apigateway", region_name="us-west-1")1723    api_eu_id, _, root_resource_eu_id = create_rest_api(apigateway_client_eu, name="test-eu-region")1724    api_us_id, _, root_resource_us_id = create_rest_api(apigateway_client_us, name="test-us-region")1725    resource_eu_id, _ = create_rest_resource(1726        apigateway_client_eu, restApiId=api_eu_id, parentId=root_resource_eu_id, pathPart="demo"1727    )1728    resource_us_id, _ = create_rest_resource(1729        apigateway_client_us, restApiId=api_us_id, parentId=root_resource_us_id, pathPart="demo"1730    )1731    create_rest_resource_method(1732        apigateway_client_eu,1733        restApiId=api_eu_id,1734        resourceId=resource_eu_id,1735        httpMethod=method,1736        authorizationType="None",1737    )1738    create_rest_resource_method(1739        apigateway_client_us,1740        restApiId=api_us_id,1741        resourceId=resource_us_id,1742        httpMethod=method,1743        authorizationType="None",1744    )1745    lambda_name = f"lambda-{short_uid()}"1746    testutil.create_lambda_function(1747        handler_file=TEST_LAMBDA_NODEJS,1748        func_name=lambda_name,1749        runtime=LAMBDA_RUNTIME_NODEJS14X,1750        region_name="eu-west-1",1751    )1752    testutil.create_lambda_function(1753        handler_file=TEST_LAMBDA_NODEJS,1754        func_name=lambda_name,1755        runtime=LAMBDA_RUNTIME_NODEJS14X,1756        region_name="us-west-1",1757    )1758    lambda_eu_arn = aws_stack.lambda_function_arn(lambda_name, region_name="eu-west-1")1759    uri_eu = aws_stack.apigateway_invocations_arn(lambda_eu_arn, region_name="eu-west-1")1760    integration_uri, _ = create_rest_api_integration(1761        apigateway_client_eu,1762        restApiId=api_eu_id,1763        resourceId=resource_eu_id,1764        httpMethod=method,1765        type="AWS_PROXY",1766        integrationHttpMethod=method,1767        uri=uri_eu,1768    )1769    lambda_us_arn = aws_stack.lambda_function_arn(lambda_name, region_name="us-west-1")1770    uri_us = aws_stack.apigateway_invocations_arn(lambda_us_arn, region_name="us-west-1")1771    integration_uri, _ = create_rest_api_integration(1772        apigateway_client_us,1773        restApiId=api_us_id,1774        resourceId=resource_us_id,1775        httpMethod=method,1776        type="AWS_PROXY",1777        integrationHttpMethod=method,1778        uri=uri_us,1779    )1780    # test valid authorization using bearer token1781    endpoint = url_function(api_eu_id, stage_name="local", path="/demo")1782    result = requests.get(endpoint, headers={}, verify=False)1783    assert result.status_code == 2001784    endpoint = url_function(api_us_id, stage_name="local", path="/demo")...

test_api_gateway.py

Source:test_api_gateway.py

...172    testutil.create_lambda_function(func_name=TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,173        zip_file=zip_file, runtime=LAMBDA_RUNTIME_PYTHON27)174    # create API Gateway and connect it to the Lambda proxy backend175    lambda_uri = aws_stack.lambda_function_arn(TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD)176    target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)177    result = connect_api_gateway_to_http_with_lambda_proxy('test_gateway3', target_uri,178        methods=['ANY'],179        path=API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD)180    # make test request to gateway and check response181    path = API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD.replace('{test_param1}', 'foo1')182    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'], stage_name=TEST_STAGE_NAME, path=path)183    data = {}184    for method in ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'):185        body = json.dumps(data) if method in ('POST', 'PUT', 'PATCH') else None186        result = getattr(requests, method.lower())(url, data=body)187        assert result.status_code == 200188        parsed_body = json.loads(to_str(result.content))...

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.