Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_rest_api_method_response method in localstack

Best Python code snippet using localstack_python

test_apigateway.py

Source:test_apigateway.py

...388            requestTemplates={389                "application/json": '#set($allParams = $input.params())\n{\n"body-json" : $input.json("$"),\n"params" : {\n#foreach($type in $allParams.keySet())\n    #set($params = $allParams.get($type))\n"$type" : {\n    #foreach($paramName in $params.keySet())\n    "$paramName" : "$util.escapeJavaScript($params.get($paramName))"\n        #if($foreach.hasNext),#end\n    #end\n}\n    #if($foreach.hasNext),#end\n#end\n},\n"stage-variables" : {\n#foreach($key in $stageVariables.keySet())\n"$key" : "$util.escapeJavaScript($stageVariables.get($key))"\n    #if($foreach.hasNext),#end\n#end\n},\n"context" : {\n    "api-id" : "$context.apiId",\n    "api-key" : "$context.identity.apiKey",\n    "http-method" : "$context.httpMethod",\n    "stage" : "$context.stage",\n    "source-ip" : "$context.identity.sourceIp",\n    "user-agent" : "$context.identity.userAgent",\n    "request-id" : "$context.requestId",\n    "resource-id" : "$context.resourceId",\n    "resource-path" : "$context.resourcePath"\n    }\n}\n'390            },391        )392        create_rest_api_method_response(393            apigateway_client,394            restApiId=api_id,395            resourceId=resource_id,396            httpMethod="POST",397            statusCode="200",398            responseParameters={399                "method.response.header.Content-Type": False,400                "method.response.header.Access-Control-Allow-Origin": False,401            },402        )403        create_rest_api_integration_response(404            apigateway_client,405            restApiId=api_id,406            resourceId=resource_id,407            httpMethod="POST",408            statusCode="200",409            responseTemplates={"text/html": "$input.path('$')"},410            responseParameters={411                "method.response.header.Access-Control-Allow-Origin": "'*'",412                "method.response.header.Content-Type": "'text/html'",413            },414        )415        deployment_id, _ = create_rest_api_deployment(apigateway_client, restApiId=api_id)416        stage = create_rest_api_stage(417            apigateway_client, restApiId=api_id, stageName="local", deploymentId=deployment_id418        )419        aws_account_id = sts_client.get_caller_identity()["Account"]420        source_arn = f"arn:aws:execute-api:{region_name}:{aws_account_id}:{api_id}/*/*/test"421        lambda_client.add_permission(422            FunctionName=lambda_arn,423            StatementId=str(short_uid()),424            Action="lambda:InvokeFunction",425            Principal="apigateway.amazonaws.com",426            SourceArn=source_arn,427        )428        url = api_invoke_url(api_id, stage=stage, path="/test")429        response = requests.post(url, json={"test": "test"})430        assert response.headers["Content-Type"] == "text/html"431        assert response.headers["Access-Control-Allow-Origin"] == "*"432    @pytest.mark.parametrize("int_type", ["custom", "proxy"])433    def test_api_gateway_http_integrations(self, int_type, monkeypatch):434        monkeypatch.setattr(config, "DISABLE_CUSTOM_CORS_APIGATEWAY", False)435        test_port = get_free_tcp_port()436        backend_url = "http://localhost:%s%s" % (test_port, self.API_PATH_HTTP_BACKEND)437        # start test HTTP backend438        proxy = self.start_http_backend(test_port)439        # create API Gateway and connect it to the HTTP_PROXY/HTTP backend440        result = self.connect_api_gateway_to_http(441            int_type, "test_gateway2", backend_url, path=self.API_PATH_HTTP_BACKEND442        )443        url = path_based_url(444            api_id=result["id"],445            stage_name=self.TEST_STAGE_NAME,446            path=self.API_PATH_HTTP_BACKEND,447        )448        # make sure CORS headers are present449        origin = "localhost"450        result = requests.options(url, headers={"origin": origin})451        assert result.status_code == 200452        assert re.match(result.headers["Access-Control-Allow-Origin"].replace("*", ".*"), origin)453        assert "POST" in result.headers["Access-Control-Allow-Methods"]454        assert "PATCH" in result.headers["Access-Control-Allow-Methods"]455        custom_result = json.dumps({"foo": "bar"})456        # make test GET request to gateway457        result = requests.get(url)458        assert 200 == result.status_code459        expected = custom_result if int_type == "custom" else "{}"460        assert expected == json.loads(to_str(result.content))["data"]461        # make test POST request to gateway462        data = json.dumps({"data": 123})463        result = requests.post(url, data=data)464        assert 200 == result.status_code465        expected = custom_result if int_type == "custom" else data466        assert expected == json.loads(to_str(result.content))["data"]467        # make test POST request with non-JSON content type468        data = "test=123"469        ctype = "application/x-www-form-urlencoded"470        result = requests.post(url, data=data, headers={"content-type": ctype})471        assert 200 == result.status_code472        content = json.loads(to_str(result.content))473        headers = CaseInsensitiveDict(content["headers"])474        expected = custom_result if int_type == "custom" else data475        assert expected == content["data"]476        assert ctype == headers["content-type"]477        # clean up478        proxy.stop()479    def test_api_gateway_lambda_proxy_integration(self):480        self._test_api_gateway_lambda_proxy_integration(481            self.TEST_LAMBDA_PROXY_BACKEND, self.API_PATH_LAMBDA_PROXY_BACKEND482        )483    def test_api_gateway_lambda_proxy_integration_with_path_param(self):484        self._test_api_gateway_lambda_proxy_integration(485            self.TEST_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,486            self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,487        )488    def test_api_gateway_lambda_proxy_integration_with_is_base_64_encoded(self):489        # Test the case where `isBase64Encoded` is enabled.490        content = b"hello, please base64 encode me"491        def _mutate_data(data) -> None:492            data["return_is_base_64_encoded"] = True493            data["return_raw_body"] = base64.b64encode(content).decode("utf8")494        test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(495            self.TEST_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,496            self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,497            data_mutator_fn=_mutate_data,498        )499        # Ensure that `invoke_rest_api_integration_backend` correctly decodes the base64 content500        assert test_result.result.status_code == 203501        assert test_result.result.content == content502    def _test_api_gateway_lambda_proxy_integration_no_asserts(503        self,504        fn_name: str,505        path: str,506        data_mutator_fn: Optional[Callable] = None,507    ) -> ApiGatewayLambdaProxyIntegrationTestResult:508        """509        Perform the setup needed to do a POST against a Lambda Proxy Integration;510        then execute the POST.511        :param data_mutator_fn: a Callable[[Dict], None] that lets us mutate the512          data dictionary before sending it off to the lambda.513        """514        self.create_lambda_function(fn_name)515        # create API Gateway and connect it to the Lambda proxy backend516        lambda_uri = aws_stack.lambda_function_arn(fn_name)517        invocation_uri = "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations"518        target_uri = invocation_uri % (aws_stack.get_region(), lambda_uri)519        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(520            "test_gateway2", target_uri, path=path, stage_name=self.TEST_STAGE_NAME521        )522        api_id = result["id"]523        path_map = get_rest_api_paths(api_id)524        _, resource = get_resource_for_path(path, path_map)525        # make test request to gateway and check response526        path_with_replace = path.replace("{test_param1}", "foo1")527        path_with_params = path_with_replace + "?foo=foo&bar=bar&bar=baz"528        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path=path_with_params)529        # These values get read in `lambda_integration.py`530        data = {"return_status_code": 203, "return_headers": {"foo": "bar123"}}531        if data_mutator_fn:532            assert callable(data_mutator_fn)533            data_mutator_fn(data)534        result = requests.post(535            url,536            data=json.dumps(data),537            headers={"User-Agent": "python-requests/testing"},538        )539        return ApiGatewayLambdaProxyIntegrationTestResult(540            data=data,541            resource=resource,542            result=result,543            url=url,544            path_with_replace=path_with_replace,545        )546    def _test_api_gateway_lambda_proxy_integration(547        self,548        fn_name: str,549        path: str,550    ) -> None:551        test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(fn_name, path)552        data, resource, result, url, path_with_replace = test_result553        assert result.status_code == 203554        assert result.headers.get("foo") == "bar123"555        assert "set-cookie" in result.headers556        try:557            parsed_body = json.loads(to_str(result.content))558        except json.decoder.JSONDecodeError as e:559            raise Exception(560                "Couldn't json-decode content: {}".format(to_str(result.content))561            ) from e562        assert parsed_body.get("return_status_code") == 203563        assert parsed_body.get("return_headers") == {"foo": "bar123"}564        assert parsed_body.get("queryStringParameters") == {"foo": "foo", "bar": ["bar", "baz"]}565        request_context = parsed_body.get("requestContext")566        source_ip = request_context["identity"].pop("sourceIp")567        assert re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", source_ip)568        expected_path = "/" + self.TEST_STAGE_NAME + "/lambda/foo1"569        assert expected_path == request_context["path"]570        assert request_context.get("stageVariables") is None571        assert get_aws_account_id() == request_context["accountId"]572        assert resource.get("id") == request_context["resourceId"]573        assert self.TEST_STAGE_NAME == request_context["stage"]574        assert "python-requests/testing" == request_context["identity"]["userAgent"]575        assert "POST" == request_context["httpMethod"]576        assert "HTTP/1.1" == request_context["protocol"]577        assert "requestTimeEpoch" in request_context578        assert "requestTime" in request_context579        assert "requestId" in request_context580        # assert that header keys are lowercase (as in AWS)581        headers = parsed_body.get("headers") or {}582        header_names = list(headers.keys())583        assert "Host" in header_names584        assert "Content-Length" in header_names585        assert "User-Agent" in header_names586        result = requests.delete(url, data=json.dumps(data))587        assert 204 == result.status_code588        # send message with non-ASCII chars589        body_msg = "ð - åã"590        result = requests.post(url, data=json.dumps({"return_raw_body": body_msg}))591        assert body_msg == to_str(result.content)592        # send message with binary data593        binary_msg = b"\xff \xaa \x11"594        result = requests.post(url, data=binary_msg)595        result_content = json.loads(to_str(result.content))596        assert "/yCqIBE=" == result_content["body"]597        assert ["isBase64Encoded"]598    def test_api_gateway_lambda_proxy_integration_any_method(self):599        self._test_api_gateway_lambda_proxy_integration_any_method(600            self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,601            self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD,602        )603    def test_api_gateway_lambda_proxy_integration_any_method_with_path_param(self):604        self._test_api_gateway_lambda_proxy_integration_any_method(605            self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,606            self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,607        )608    def test_api_gateway_lambda_asynchronous_invocation(self, apigateway_client, create_rest_apigw):609        api_gateway_name = f"api_gateway_{short_uid()}"610        rest_api_id, _, _ = create_rest_apigw(name=api_gateway_name)611        fn_name = f"test-{short_uid()}"612        testutil.create_lambda_function(613            handler_file=TEST_LAMBDA_NODEJS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X614        )615        lambda_arn = aws_stack.lambda_function_arn(fn_name)616        spec_file = load_file(TEST_IMPORT_REST_API_ASYNC_LAMBDA)617        spec_file = spec_file.replace("${lambda_invocation_arn}", lambda_arn)618        apigateway_client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")619        url = path_based_url(api_id=rest_api_id, stage_name="latest", path="/wait/3")620        result = requests.get(url)621        assert result.status_code == 200622        assert result.content == b""623    def test_api_gateway_mock_integration(self, apigateway_client, create_rest_apigw):624        rest_api_name = f"apigw-{short_uid()}"625        rest_api_id, _, _ = create_rest_apigw(name=rest_api_name)626        spec_file = load_file(TEST_IMPORT_MOCK_INTEGRATION)627        apigateway_client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")628        url = path_based_url(api_id=rest_api_id, stage_name="latest", path="/echo/foobar")629        response = requests.get(url)630        assert response._content == b'{"echo": "foobar", "response": "mocked"}'631    def test_api_gateway_authorizer_crud(self):632        apig = aws_stack.create_external_boto_client("apigateway")633        authorizer = apig.create_authorizer(634            restApiId=self.TEST_API_GATEWAY_ID, **self.TEST_API_GATEWAY_AUTHORIZER635        )636        authorizer_id = authorizer.get("id")637        create_result = apig.get_authorizer(638            restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id639        )640        # ignore boto3 stuff641        del create_result["ResponseMetadata"]642        create_expected = clone(self.TEST_API_GATEWAY_AUTHORIZER)643        create_expected["id"] = authorizer_id644        assert create_expected == create_result645        apig.update_authorizer(646            restApiId=self.TEST_API_GATEWAY_ID,647            authorizerId=authorizer_id,648            patchOperations=self.TEST_API_GATEWAY_AUTHORIZER_OPS,649        )650        update_result = apig.get_authorizer(651            restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id652        )653        # ignore boto3 stuff654        del update_result["ResponseMetadata"]655        update_expected = apply_patch(create_expected, self.TEST_API_GATEWAY_AUTHORIZER_OPS)656        assert update_expected == update_result657        apig.delete_authorizer(restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id)658        with pytest.raises(Exception):659            apig.get_authorizer(self.TEST_API_GATEWAY_ID, authorizer_id)660    def test_apigateway_with_lambda_integration(self, apigateway_client, create_rest_apigw):661        # create Lambda function662        lambda_name = f"apigw-lambda-{short_uid()}"663        self.create_lambda_function(lambda_name)664        lambda_uri = aws_stack.lambda_function_arn(lambda_name)665        target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)666        # create REST API667        api_id, _, _ = create_rest_apigw(name="test-api", description="")668        root_res_id = apigateway_client.get_resources(restApiId=api_id)["items"][0]["id"]669        api_resource = apigateway_client.create_resource(670            restApiId=api_id, parentId=root_res_id, pathPart="test"671        )672        apigateway_client.put_method(673            restApiId=api_id,674            resourceId=api_resource["id"],675            httpMethod="GET",676            authorizationType="NONE",677        )678        rs = apigateway_client.put_integration(679            restApiId=api_id,680            resourceId=api_resource["id"],681            httpMethod="GET",682            integrationHttpMethod="POST",683            type="AWS",684            uri=target_uri,685            timeoutInMillis=3000,686            contentHandling="CONVERT_TO_BINARY",687            requestTemplates={"application/json": '{"param1": "$input.params(\'param1\')"}'},688        )689        integration_keys = [690            "httpMethod",691            "type",692            "cacheKeyParameters",693            "uri",694            "cacheNamespace",695            "timeoutInMillis",696            "contentHandling",697            "requestParameters",698        ]699        assert 201 == rs["ResponseMetadata"]["HTTPStatusCode"]700        for key in integration_keys:701            assert key in rs702        assert "responseTemplates" not in rs703        apigateway_client.create_deployment(restApiId=api_id, stageName=self.TEST_STAGE_NAME)704        rs = apigateway_client.get_integration(705            restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"706        )707        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]708        assert "AWS" == rs["type"]709        assert "POST" == rs["httpMethod"]710        assert target_uri == rs["uri"]711        # invoke the gateway endpoint712        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/test")713        response = requests.get(f"{url}?param1=foobar")714        assert response.status_code < 400715        content = response.json()716        assert {"param1": "foobar"} == content.get("event")717        # additional checks from https://github.com/localstack/localstack/issues/5041718        # pass Signature param719        response = requests.get(f"{url}?param1=foobar&Signature=1")720        assert response.status_code == 200721        content = response.json()722        assert {"param1": "foobar"} == content.get("event")723        # delete integration724        rs = apigateway_client.delete_integration(725            restApiId=api_id,726            resourceId=api_resource["id"],727            httpMethod="GET",728        )729        assert 204 == rs["ResponseMetadata"]["HTTPStatusCode"]730        with pytest.raises(ClientError) as ctx:731            # This call should not be successful as the integration is deleted732            apigateway_client.get_integration(733                restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"734            )735        assert ctx.value.response["Error"]["Code"] == "NotFoundException"736        # clean up737        lambda_client = aws_stack.create_external_boto_client("lambda")738        lambda_client.delete_function(FunctionName=lambda_name)739    def test_malformed_response_apigw_invocation(self, create_lambda_function, lambda_client):740        lambda_name = f"test_lambda_{short_uid()}"741        lambda_resource = "/api/v1/{proxy+}"742        lambda_path = "/api/v1/hello/world"743        create_lambda_function(744            func_name=lambda_name,745            zip_file=testutil.create_zip_file(TEST_LAMBDA_NODEJS_APIGW_502, get_content=True),746            runtime=LAMBDA_RUNTIME_NODEJS14X,747            handler="apigw_502.handler",748        )749        lambda_uri = aws_stack.lambda_function_arn(lambda_name)750        target_uri = f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path/2015-03-31/functions/{lambda_uri}/invocations"751        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(752            "test_gateway",753            target_uri,754            path=lambda_resource,755            stage_name="testing",756        )757        api_id = result["id"]758        url = path_based_url(api_id=api_id, stage_name="testing", path=lambda_path)759        result = requests.get(url)760        assert result.status_code == 502761        assert result.headers.get("Content-Type") == "application/json"762        assert json.loads(result.content)["message"] == "Internal server error"763    def test_api_gateway_handle_domain_name(self):764        domain_name = f"{short_uid()}.example.com"765        apigw_client = aws_stack.create_external_boto_client("apigateway")766        rs = apigw_client.create_domain_name(domainName=domain_name)767        assert 201 == rs["ResponseMetadata"]["HTTPStatusCode"]768        rs = apigw_client.get_domain_name(domainName=domain_name)769        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]770        assert domain_name == rs["domainName"]771        apigw_client.delete_domain_name(domainName=domain_name)772    def _test_api_gateway_lambda_proxy_integration_any_method(self, fn_name, path):773        self.create_lambda_function(fn_name)774        # create API Gateway and connect it to the Lambda proxy backend775        lambda_uri = aws_stack.lambda_function_arn(fn_name)776        target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)777        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(778            "test_gateway3",779            target_uri,780            methods=["ANY"],781            path=path,782            stage_name=self.TEST_STAGE_NAME,783        )784        # make test request to gateway and check response785        path = path.replace("{test_param1}", "foo1")786        url = path_based_url(api_id=result["id"], stage_name=self.TEST_STAGE_NAME, path=path)787        data = {}788        for method in ("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"):789            body = json.dumps(data) if method in ("POST", "PUT", "PATCH") else None790            result = getattr(requests, method.lower())(url, data=body)791            if method != "DELETE":792                assert 200 == result.status_code793                parsed_body = json.loads(to_str(result.content))794                assert method == parsed_body.get("httpMethod")795            else:796                assert 204 == result.status_code797    def test_apigateway_with_custom_authorization_method(798        self, apigateway_client, create_rest_apigw799    ):800        # create Lambda function801        lambda_name = f"apigw-lambda-{short_uid()}"802        self.create_lambda_function(lambda_name)803        lambda_uri = aws_stack.lambda_function_arn(lambda_name)804        # create REST API805        api_id, _, _ = create_rest_apigw(name="test-api", description="")806        root_res_id = apigateway_client.get_resources(restApiId=api_id)["items"][0]["id"]807        # create authorizer at root resource808        authorizer = apigateway_client.create_authorizer(809            restApiId=api_id,810            name="lambda_authorizer",811            type="TOKEN",812            authorizerUri="arn:aws:apigateway:us-east-1:lambda:path/ \813                2015-03-31/functions/{}/invocations".format(814                lambda_uri815            ),816            identitySource="method.request.header.Auth",817        )818        # create method with custom authorizer819        is_api_key_required = True820        method_response = apigateway_client.put_method(821            restApiId=api_id,822            resourceId=root_res_id,823            httpMethod="GET",824            authorizationType="CUSTOM",825            authorizerId=authorizer["id"],826            apiKeyRequired=is_api_key_required,827        )828        assert authorizer["id"] == method_response["authorizerId"]829        # clean up830        lambda_client = aws_stack.create_external_boto_client("lambda")831        lambda_client.delete_function(FunctionName=lambda_name)832    def test_create_model(self, create_rest_apigw, apigateway_client):833        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="this is my api")834        dummy_rest_api_id = "_non_existing_"835        model_name = "testModel"836        description = "test model"837        content_type = "application/json"838        # success case with valid params839        response = apigateway_client.create_model(840            restApiId=rest_api_id,841            name=model_name,842            description=description,843            contentType=content_type,844        )845        assert model_name == response["name"]846        assert description == response["description"]847        with pytest.raises(Exception) as ctx:848            apigateway_client.create_model(849                restApiId=dummy_rest_api_id,850                name=model_name,851                description=description,852                contentType=content_type,853            )854        assert "NotFoundException" == ctx.value.response["Error"]["Code"]855        assert "Invalid Rest API Id specified" == ctx.value.response["Error"]["Message"]856        with pytest.raises(Exception) as ctx:857            apigateway_client.create_model(858                restApiId=dummy_rest_api_id,859                name="",860                description=description,861                contentType=content_type,862            )863        assert "BadRequestException" == ctx.value.response["Error"]["Code"]864        assert "No Model Name specified" == ctx.value.response["Error"]["Message"]865    def test_get_api_models(self, apigateway_client, create_rest_apigw):866        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="this is my api")867        model_name = "testModel"868        description = "test model"869        content_type = "application/json"870        # when no models are present871        result = apigateway_client.get_models(restApiId=rest_api_id)872        assert [] == result["items"]873        # add a model874        apigateway_client.create_model(875            restApiId=rest_api_id,876            name=model_name,877            description=description,878            contentType=content_type,879        )880        # get models after adding881        result = apigateway_client.get_models(restApiId=rest_api_id)882        assert model_name == result["items"][0]["name"]883        assert description == result["items"][0]["description"]884    def test_request_validator(self, apigateway_client, create_rest_apigw):885        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="this is my api")886        # CREATE887        name = "validator123"888        result = apigateway_client.create_request_validator(restApiId=rest_api_id, name=name)889        assert 201 == result["ResponseMetadata"]["HTTPStatusCode"]890        validator_id = result["id"]891        # LIST892        result = apigateway_client.get_request_validators(restApiId=rest_api_id)893        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]894        assert [{"id": validator_id, "name": name}] == result["items"]895        # GET896        result = apigateway_client.get_request_validator(897            restApiId=rest_api_id, requestValidatorId=validator_id898        )899        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]900        assert select_attributes(result, ["id", "name"]) == {"id": validator_id, "name": name}901        # UPDATE902        result = apigateway_client.update_request_validator(903            restApiId=rest_api_id, requestValidatorId=validator_id, patchOperations=[]904        )905        # DELETE906        apigateway_client.delete_request_validator(907            restApiId=rest_api_id, requestValidatorId=validator_id908        )909        with pytest.raises(Exception):910            apigateway_client.get_request_validator(911                restApiId=rest_api_id, requestValidatorId=validator_id912            )913        with pytest.raises(Exception):914            apigateway_client.delete_request_validator(915                restApiId=rest_api_id, requestValidatorId=validator_id916            )917    def test_base_path_mapping(self, apigateway_client, create_rest_apigw):918        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="this is my api")919        # CREATE920        domain_name = "domain1.example.com"921        apigateway_client.create_domain_name(domainName=domain_name)922        root_res_id = apigateway_client.get_resources(restApiId=rest_api_id)["items"][0]["id"]923        res_id = apigateway_client.create_resource(924            restApiId=rest_api_id, parentId=root_res_id, pathPart="path"925        )["id"]926        apigateway_client.put_method(927            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"928        )929        apigateway_client.put_integration(930            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"931        )932        depl_id = apigateway_client.create_deployment(restApiId=rest_api_id)["id"]933        apigateway_client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")934        base_path = "foo"935        result = apigateway_client.create_base_path_mapping(936            domainName=domain_name,937            basePath=base_path,938            restApiId=rest_api_id,939            stage="dev",940        )941        assert result["ResponseMetadata"]["HTTPStatusCode"] in [200, 201]942        # LIST943        result = apigateway_client.get_base_path_mappings(domainName=domain_name)944        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]945        expected = {"basePath": base_path, "restApiId": rest_api_id, "stage": "dev"}946        assert [expected] == result["items"]947        # GET948        result = apigateway_client.get_base_path_mapping(domainName=domain_name, basePath=base_path)949        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]950        assert expected == select_attributes(result, ["basePath", "restApiId", "stage"])951        # UPDATE952        result = apigateway_client.update_base_path_mapping(953            domainName=domain_name, basePath=base_path, patchOperations=[]954        )955        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]956        # DELETE957        apigateway_client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)958        with pytest.raises(Exception):959            apigateway_client.get_base_path_mapping(domainName=domain_name, basePath=base_path)960        with pytest.raises(Exception):961            apigateway_client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)962    def test_base_path_mapping_root(self):963        client = aws_stack.create_external_boto_client("apigateway")964        response = client.create_rest_api(name="my_api2", description="this is my api")965        rest_api_id = response["id"]966        # CREATE967        domain_name = "domain2.example.com"968        client.create_domain_name(domainName=domain_name)969        root_res_id = client.get_resources(restApiId=rest_api_id)["items"][0]["id"]970        res_id = client.create_resource(971            restApiId=rest_api_id, parentId=root_res_id, pathPart="path"972        )["id"]973        client.put_method(974            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", authorizationType="NONE"975        )976        client.put_integration(977            restApiId=rest_api_id, resourceId=res_id, httpMethod="GET", type="MOCK"978        )979        depl_id = client.create_deployment(restApiId=rest_api_id)["id"]980        client.create_stage(restApiId=rest_api_id, deploymentId=depl_id, stageName="dev")981        result = client.create_base_path_mapping(982            domainName=domain_name,983            basePath="",984            restApiId=rest_api_id,985            stage="dev",986        )987        assert result["ResponseMetadata"]["HTTPStatusCode"] in [200, 201]988        base_path = "(none)"989        # LIST990        result = client.get_base_path_mappings(domainName=domain_name)991        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]992        expected = {"basePath": "(none)", "restApiId": rest_api_id, "stage": "dev"}993        assert [expected] == result["items"]994        # GET995        result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)996        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]997        assert expected == select_attributes(result, ["basePath", "restApiId", "stage"])998        # UPDATE999        result = client.update_base_path_mapping(1000            domainName=domain_name, basePath=base_path, patchOperations=[]1001        )1002        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]1003        # DELETE1004        client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)1005        with pytest.raises(Exception):1006            client.get_base_path_mapping(domainName=domain_name, basePath=base_path)1007        with pytest.raises(Exception):1008            client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)1009    def test_api_account(self, apigateway_client, create_rest_apigw):1010        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="test 123")1011        result = apigateway_client.get_account()1012        assert "UsagePlans" in result["features"]1013        result = apigateway_client.update_account(1014            patchOperations=[{"op": "add", "path": "/features/-", "value": "foobar"}]1015        )1016        assert "foobar" in result["features"]1017    def test_get_model_by_name(self, apigateway_client, create_rest_apigw):1018        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="this is my api")1019        dummy_rest_api_id = "_non_existing_"1020        model_name = "testModel"1021        description = "test model"1022        content_type = "application/json"1023        # add a model1024        apigateway_client.create_model(1025            restApiId=rest_api_id,1026            name=model_name,1027            description=description,1028            contentType=content_type,1029        )1030        # get models after adding1031        result = apigateway_client.get_model(restApiId=rest_api_id, modelName=model_name)1032        assert model_name == result["name"]1033        assert description == result["description"]1034        try:1035            apigateway_client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)1036            self.fail("This call should not be successful as the model is not created.")1037        except ClientError as e:1038            assert "NotFoundException" == e.response["Error"]["Code"]1039            assert "Invalid Rest API Id specified" == e.response["Error"]["Message"]1040    def test_get_model_with_invalid_name(self, apigateway_client, create_rest_apigw):1041        rest_api_id, _, _ = create_rest_apigw(name="my_api", description="this is my api")1042        # test with an invalid model name1043        try:1044            apigateway_client.get_model(restApiId=rest_api_id, modelName="fake")1045            self.fail("This call should not be successful as the model is not created.")1046        except ClientError as e:1047            assert "NotFoundException" == e.response["Error"]["Code"]1048    def test_put_integration_dynamodb_proxy_validation_without_response_template(self):1049        api_id = self.create_api_gateway_and_deploy({})1050        url = path_based_url(api_id=api_id, stage_name="staging", path="/")1051        response = requests.put(1052            url,1053            json.dumps({"id": "id1", "data": "foobar123"}),1054        )1055        assert 404 == response.status_code1056    def test_put_integration_dynamodb_proxy_validation_with_response_template(self):1057        response_templates = {1058            "application/json": json.dumps(1059                {1060                    "TableName": "MusicCollection",1061                    "Item": {"id": "$.Id", "data": "$.data"},1062                }1063            )1064        }1065        api_id = self.create_api_gateway_and_deploy(response_templates)1066        url = path_based_url(api_id=api_id, stage_name="staging", path="/")1067        response = requests.put(1068            url,1069            json.dumps({"id": "id1", "data": "foobar123"}),1070        )1071        assert 200 == response.status_code1072        dynamo_client = aws_stack.connect_to_resource("dynamodb")1073        table = dynamo_client.Table("MusicCollection")1074        result = table.get_item(Key={"id": "id1"})1075        assert "foobar123" == result["Item"]["data"]1076    def test_api_key_required_for_methods(self):1077        response_templates = {1078            "application/json": json.dumps(1079                {1080                    "TableName": "MusicCollection",1081                    "Item": {"id": "$.Id", "data": "$.data"},1082                }1083            )1084        }1085        api_id = self.create_api_gateway_and_deploy(response_templates, True)1086        url = path_based_url(api_id=api_id, stage_name="staging", path="/")1087        payload = {1088            "name": "TEST-PLAN-2",1089            "description": "Description",1090            "quota": {"limit": 10, "period": "DAY", "offset": 0},1091            "throttle": {"rateLimit": 2, "burstLimit": 1},1092            "apiStages": [{"apiId": api_id, "stage": "staging"}],1093            "tags": {"tag_key": "tag_value"},1094        }1095        client = aws_stack.create_external_boto_client("apigateway")1096        usage_plan_id = client.create_usage_plan(**payload)["id"]1097        key_name = "testApiKey"1098        key_type = "API_KEY"1099        api_key = client.create_api_key(name=key_name)1100        payload = {1101            "usagePlanId": usage_plan_id,1102            "keyId": api_key["id"],1103            "keyType": key_type,1104        }1105        client.create_usage_plan_key(**payload)1106        response = requests.put(1107            url,1108            json.dumps({"id": "id1", "data": "foobar123"}),1109        )1110        # when the api key is not passed as part of the header1111        assert 403 == response.status_code1112        response = requests.put(1113            url,1114            json.dumps({"id": "id1", "data": "foobar123"}),1115            headers={"X-API-Key": api_key["value"]},1116        )1117        # when the api key is passed as part of the header1118        assert 200 == response.status_code1119    def test_multiple_api_keys_validate(self, apigateway_client):1120        response_templates = {1121            "application/json": json.dumps(1122                {1123                    "TableName": "MusicCollection",1124                    "Item": {"id": "$.Id", "data": "$.data"},1125                }1126            )1127        }1128        api_id = self.create_api_gateway_and_deploy(response_templates, True)1129        url = path_based_url(api_id=api_id, stage_name="staging", path="/")1130        # Create multiple usage plans1131        usage_plan_ids = []1132        for i in range(2):1133            payload = {1134                "name": "APIKEYTEST-PLAN-{}".format(i),1135                "description": "Description",1136                "quota": {"limit": 10, "period": "DAY", "offset": 0},1137                "throttle": {"rateLimit": 2, "burstLimit": 1},1138                "apiStages": [{"apiId": api_id, "stage": "staging"}],1139                "tags": {"tag_key": "tag_value"},1140            }1141            usage_plan_ids.append(apigateway_client.create_usage_plan(**payload)["id"])1142        api_keys = []1143        key_type = "API_KEY"1144        # Create multiple API Keys in each usage plan1145        for usage_plan_id in usage_plan_ids:1146            for i in range(2):1147                api_key = apigateway_client.create_api_key(name="testMultipleApiKeys{}".format(i))1148                payload = {1149                    "usagePlanId": usage_plan_id,1150                    "keyId": api_key["id"],1151                    "keyType": key_type,1152                }1153                apigateway_client.create_usage_plan_key(**payload)1154                api_keys.append(api_key["value"])1155        response = requests.put(1156            url,1157            json.dumps({"id": "id1", "data": "foobar123"}),1158        )1159        # when the api key is not passed as part of the header1160        assert 403 == response.status_code1161        # check that all API keys work1162        for key in api_keys:1163            response = requests.put(1164                url,1165                json.dumps({"id": "id1", "data": "foobar123"}),1166                headers={"X-API-Key": key},1167            )1168            # when the api key is passed as part of the header1169            assert 200 == response.status_code1170    @pytest.mark.parametrize("base_path_type", ["ignore", "prepend", "split"])1171    def test_import_rest_api(1172        self,1173        base_path_type,1174        create_rest_apigw,1175        import_apigw,1176        apigateway_client,1177    ):1178        rest_api_name = f"restapi-{short_uid()}"1179        rest_api_id, _, _ = create_rest_apigw(name=rest_api_name)1180        spec_file = load_file(TEST_SWAGGER_FILE_JSON)1181        api_params = {"basepath": base_path_type}1182        rs = apigateway_client.put_rest_api(1183            restApiId=rest_api_id, body=spec_file, mode="overwrite", parameters=api_params1184        )1185        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]1186        resources = apigateway_client.get_resources(restApiId=rest_api_id)1187        for rv in resources.get("items"):1188            for method in rv.get("resourceMethods", {}).values():1189                assert method.get("authorizationType") == "request"1190                assert method.get("authorizerId") is not None1191        spec_file = load_file(TEST_SWAGGER_FILE_YAML)1192        rs = apigateway_client.put_rest_api(1193            restApiId=rest_api_id, body=spec_file, mode="overwrite", parameters=api_params1194        )1195        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]1196        rs = apigateway_client.get_resources(restApiId=rest_api_id)1197        expected_resources = 2 if base_path_type == "ignore" else 31198        assert len(rs["items"]) == expected_resources1199        abs_path = "/test" if base_path_type == "ignore" else "/base/test"1200        resource = [res for res in rs["items"] if res["path"] == abs_path][0]1201        assert "GET" in resource["resourceMethods"]1202        assert "requestParameters" in resource["resourceMethods"]["GET"]1203        assert {"integration.request.header.X-Amz-Invocation-Type": "'Event'"} == resource[1204            "resourceMethods"1205        ]["GET"]["requestParameters"]1206        url = path_based_url(api_id=rest_api_id, stage_name="dev", path=abs_path)1207        response = requests.get(url)1208        assert 200 == response.status_code1209        # clean up1210        spec_file = load_file(TEST_IMPORT_REST_API_FILE)1211        rest_api_id, _, _ = import_apigw(body=spec_file, parameters=api_params)1212        rs = apigateway_client.get_resources(restApiId=rest_api_id)1213        resources = rs["items"]1214        assert 3 == len(resources)1215        paths = [res["path"] for res in resources]1216        assert "/" in paths1217        assert "/pets" in paths1218        assert "/pets/{petId}" in paths1219    def test_step_function_integrations(self, create_rest_apigw, apigateway_client):1220        sfn_client = aws_stack.create_external_boto_client("stepfunctions")1221        lambda_client = aws_stack.create_external_boto_client("lambda")1222        state_machine_name = f"test-{short_uid()}"1223        state_machine_def = {1224            "Comment": "Hello World example",1225            "StartAt": "step1",1226            "States": {1227                "step1": {"Type": "Task", "Resource": "__tbd__", "End": True},1228            },1229        }1230        # create state machine1231        fn_name = f"sfn-apigw-{short_uid()}"1232        testutil.create_lambda_function(1233            handler_file=TEST_LAMBDA_PYTHON_ECHO,1234            func_name=fn_name,1235            runtime=LAMBDA_RUNTIME_PYTHON36,1236        )1237        role_arn = aws_stack.role_arn("sfn_role")1238        # create state machine definition1239        definition = clone(state_machine_def)1240        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1241        definition["States"]["step1"]["Resource"] = lambda_arn_11242        definition = json.dumps(definition)1243        # create state machine1244        result = sfn_client.create_state_machine(1245            name=state_machine_name, definition=definition, roleArn=role_arn1246        )1247        sm_arn = result["stateMachineArn"]1248        # create REST API and method1249        rest_api, _, _ = create_rest_apigw(name="test", description="test")1250        resources = apigateway_client.get_resources(restApiId=rest_api)1251        root_resource_id = resources["items"][0]["id"]1252        apigateway_client.put_method(1253            restApiId=rest_api,1254            resourceId=root_resource_id,1255            httpMethod="POST",1256            authorizationType="NONE",1257        )1258        def _prepare_method_integration(1259            integr_kwargs=None, resp_templates=None, action="StartExecution", overwrite=False1260        ):1261            if integr_kwargs is None:1262                integr_kwargs = {}1263            if resp_templates is None:1264                resp_templates = {}1265            if overwrite:1266                apigateway_client.delete_integration(1267                    restApiId=rest_api,1268                    resourceId=resources["items"][0]["id"],1269                    httpMethod="POST",1270                )1271            uri = f"arn:aws:apigateway:{aws_stack.get_region()}:states:action/{action}"1272            apigateway_client.put_integration(1273                restApiId=rest_api,1274                resourceId=root_resource_id,1275                httpMethod="POST",1276                integrationHttpMethod="POST",1277                type="AWS",1278                uri=uri,1279                **integr_kwargs,1280            )1281            if resp_templates:1282                apigateway_client.put_integration_response(1283                    restApiId=rest_api,1284                    resourceId=root_resource_id,1285                    selectionPattern="",1286                    responseTemplates=resp_templates,1287                    httpMethod="POST",1288                    statusCode="200",1289                )1290        # STEP 1: test integration with request template1291        _prepare_method_integration(1292            integr_kwargs={1293                "requestTemplates": {1294                    "application/json": """1295                    #set($data = $util.escapeJavaScript($input.json('$')))1296                    {"input": $data, "stateMachineArn": "%s"}1297                    """1298                    % sm_arn1299                }1300            }1301        )1302        # invoke stepfunction via API GW, assert results1303        apigateway_client.create_deployment(restApiId=rest_api, stageName="dev")1304        url = path_based_url(api_id=rest_api, stage_name="dev", path="/")1305        test_data = {"test": "test-value"}1306        resp = requests.post(url, data=json.dumps(test_data))1307        assert 200 == resp.status_code1308        assert "executionArn" in resp.content.decode()1309        assert "startDate" in resp.content.decode()1310        # STEP 2: test integration without request template1311        _prepare_method_integration(overwrite=True)1312        test_data_1 = {1313            "input": json.dumps(test_data),1314            "name": "MyExecution",1315            "stateMachineArn": sm_arn,1316        }1317        # invoke stepfunction via API GW, assert results1318        resp = requests.post(url, data=json.dumps(test_data_1))1319        assert 200 == resp.status_code1320        assert "executionArn" in resp.content.decode()1321        assert "startDate" in resp.content.decode()1322        # STEP 3: test integration with synchronous execution1323        _prepare_method_integration(overwrite=True, action="StartSyncExecution")1324        # invoke stepfunction via API GW, assert results1325        test_data_1["name"] += "1"1326        resp = requests.post(url, data=json.dumps(test_data_1))1327        assert 200 == resp.status_code1328        content = json.loads(to_str(resp.content.decode()))1329        assert "SUCCEEDED" == content.get("status")1330        assert test_data == json.loads(content.get("output"))1331        # STEP 4: test integration with synchronous execution and response templates1332        resp_templates = {APPLICATION_JSON: "$input.path('$.output')"}1333        _prepare_method_integration(1334            resp_templates=resp_templates, overwrite=True, action="StartSyncExecution"1335        )1336        # invoke stepfunction via API GW, assert results1337        test_data_1["name"] += "2"1338        resp = requests.post(url, data=json.dumps(test_data_1))1339        assert 200 == resp.status_code1340        assert test_data == json.loads(to_str(resp.content.decode()))1341        _prepare_method_integration(overwrite=True, action="DeleteStateMachine")1342        # Remove state machine with API GW1343        resp = requests.post(url, data=json.dumps({"stateMachineArn": sm_arn}))1344        assert 200 == resp.status_code1345        # Clean up1346        lambda_client.delete_function(FunctionName=fn_name)1347    def test_api_gateway_http_integration_with_path_request_parameter(1348        self, apigateway_client, create_rest_apigw1349    ):1350        test_port = get_free_tcp_port()1351        backend_url = "http://localhost:%s/person/{id}" % test_port1352        # start test HTTP backend1353        proxy = self.start_http_backend(test_port)1354        # create rest api1355        api_id, _, _ = create_rest_apigw(name="test")1356        parent_response = apigateway_client.get_resources(restApiId=api_id)1357        parent_id = parent_response["items"][0]["id"]1358        resource_1 = apigateway_client.create_resource(1359            restApiId=api_id, parentId=parent_id, pathPart="person"1360        )1361        resource_1_id = resource_1["id"]1362        resource_2 = apigateway_client.create_resource(1363            restApiId=api_id, parentId=resource_1_id, pathPart="{id}"1364        )1365        resource_2_id = resource_2["id"]1366        apigateway_client.put_method(1367            restApiId=api_id,1368            resourceId=resource_2_id,1369            httpMethod="GET",1370            authorizationType="NONE",1371            apiKeyRequired=False,1372            requestParameters={"method.request.path.id": True},1373        )1374        apigateway_client.put_integration(1375            restApiId=api_id,1376            resourceId=resource_2_id,1377            httpMethod="GET",1378            integrationHttpMethod="GET",1379            type="HTTP",1380            uri=backend_url,1381            timeoutInMillis=3000,1382            contentHandling="CONVERT_TO_BINARY",1383            requestParameters={"integration.request.path.id": "method.request.path.id"},1384        )1385        apigateway_client.create_deployment(restApiId=api_id, stageName="test")1386        def _test_invoke(url):1387            result = requests.get(url)1388            content = json.loads(to_str(result.content))1389            assert 200 == result.status_code1390            assert re.search(1391                "http://.*localhost.*/person/123",1392                content["headers"].get(HEADER_LOCALSTACK_REQUEST_URL),1393            )1394        for use_hostname in [True, False]:1395            for use_ssl in [True, False] if use_hostname else [False]:1396                url = self._get_invoke_endpoint(1397                    api_id,1398                    stage="test",1399                    path="/person/123",1400                    use_hostname=use_hostname,1401                    use_ssl=use_ssl,1402                )1403                _test_invoke(url)1404        proxy.stop()1405    def _get_invoke_endpoint(1406        self, api_id, stage="test", path="/", use_hostname=False, use_ssl=False1407    ):1408        path = path or "/"1409        path = path if path.startswith(path) else f"/{path}"1410        proto = "https" if use_ssl else "http"1411        if use_hostname:1412            return (1413                f"{proto}://{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}/"1414                f"{stage}{path}"1415            )1416        return (1417            f"{proto}://localhost:{config.EDGE_PORT}/restapis/{api_id}/{stage}/_user_request_{path}"1418        )1419    def test_api_gateway_s3_get_integration(self, apigateway_client, create_rest_apigw):1420        s3_client = aws_stack.create_external_boto_client("s3")1421        bucket_name = f"test-bucket-{short_uid()}"1422        apigateway_name = f"test-api-{short_uid()}"1423        object_name = "test.json"1424        object_content = '{ "success": "true" }'1425        object_content_type = "application/json"1426        api_id, _, _ = create_rest_apigw(name=apigateway_name)1427        try:1428            aws_stack.get_or_create_bucket(bucket_name)1429            s3_client.put_object(1430                Bucket=bucket_name,1431                Key=object_name,1432                Body=object_content,1433                ContentType=object_content_type,1434            )1435            self.connect_api_gateway_to_s3(bucket_name, object_name, api_id, "GET")1436            apigateway_client.create_deployment(restApiId=api_id, stageName="test")1437            url = path_based_url(api_id, "test", f"/{object_name}")1438            result = requests.get(url)1439            assert 200 == result.status_code1440            assert object_content == result.text1441            assert object_content_type == result.headers["content-type"]1442        finally:1443            s3_client.delete_object(Bucket=bucket_name, Key=object_name)1444            s3_client.delete_bucket(Bucket=bucket_name)1445    def test_api_mock_integration_response_params(self):1446        resps = [1447            {1448                "statusCode": "204",1449                "httpMethod": "OPTIONS",1450                "responseParameters": {1451                    "method.response.header.Access-Control-Allow-Methods": "'POST,OPTIONS'",1452                    "method.response.header.Vary": "'Origin'",1453                },1454            }1455        ]1456        api_id = self.create_api_gateway_and_deploy(1457            integration_type="MOCK", integration_responses=resps1458        )1459        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/")1460        result = requests.options(url)1461        assert result.status_code < 4001462        assert "Origin" == result.headers.get("vary")1463        assert "POST,OPTIONS" == result.headers.get("Access-Control-Allow-Methods")1464    def test_api_gateway_update_resource_path_part(self, apigateway_client, create_rest_apigw):1465        api_id, _, _ = create_rest_apigw(name="test-api", description="")1466        root_res_id = apigateway_client.get_resources(restApiId=api_id)["items"][0]["id"]1467        api_resource = apigateway_client.create_resource(1468            restApiId=api_id, parentId=root_res_id, pathPart="test"1469        )1470        response = apigateway_client.update_resource(1471            restApiId=api_id,1472            resourceId=api_resource.get("id"),1473            patchOperations=[1474                {"op": "replace", "path": "/pathPart", "value": "demo1"},1475            ],1476        )1477        assert response.get("pathPart") == "demo1"1478        response = apigateway_client.get_resource(1479            restApiId=api_id, resourceId=api_resource.get("id")1480        )1481        assert response.get("pathPart") == "demo1"1482    def test_response_headers_invocation_with_apigw(self, create_lambda_function, lambda_client):1483        lambda_name = f"test_lambda_{short_uid()}"1484        lambda_resource = "/api/v1/{proxy+}"1485        lambda_path = "/api/v1/hello/world"1486        create_lambda_function(1487            func_name=lambda_name,1488            zip_file=testutil.create_zip_file(1489                TEST_LAMBDA_NODEJS_APIGW_INTEGRATION, get_content=True1490            ),1491            runtime=LAMBDA_RUNTIME_NODEJS14X,1492            handler="apigw_integration.handler",1493        )1494        lambda_uri = aws_stack.lambda_function_arn(lambda_name)1495        target_uri = f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path/2015-03-31/functions/{lambda_uri}/invocations"1496        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(1497            "test_gateway",1498            target_uri,1499            path=lambda_resource,1500            stage_name=self.TEST_STAGE_NAME,1501        )1502        api_id = result["id"]1503        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path=lambda_path)1504        result = requests.get(url)1505        assert result.status_code == 3001506        assert result.headers["Content-Type"] == "application/xml"1507        body = xmltodict.parse(result.content)1508        assert body.get("message") == "completed"1509    # =====================================================================1510    # Helper methods1511    # =====================================================================1512    def connect_api_gateway_to_s3(self, bucket_name, file_name, api_id, method):1513        """Connects the root resource of an api gateway to the given object of an s3 bucket."""1514        apigw_client = aws_stack.create_external_boto_client("apigateway")1515        s3_uri = "arn:aws:apigateway:{}:s3:path/{}/{{proxy}}".format(1516            aws_stack.get_region(), bucket_name1517        )1518        test_role = "test-s3-role"1519        role_arn = aws_stack.role_arn(role_name=test_role)1520        resources = apigw_client.get_resources(restApiId=api_id)1521        # using the root resource '/' directly for this test1522        root_resource_id = resources["items"][0]["id"]1523        proxy_resource = apigw_client.create_resource(1524            restApiId=api_id, parentId=root_resource_id, pathPart="{proxy+}"1525        )1526        apigw_client.put_method(1527            restApiId=api_id,1528            resourceId=proxy_resource["id"],1529            httpMethod=method,1530            authorizationType="NONE",1531            apiKeyRequired=False,1532            requestParameters={},1533        )1534        apigw_client.put_integration(1535            restApiId=api_id,1536            resourceId=proxy_resource["id"],1537            httpMethod=method,1538            type="AWS",1539            integrationHttpMethod=method,1540            uri=s3_uri,1541            credentials=role_arn,1542            requestParameters={"integration.request.path.proxy": "method.request.path.proxy"},1543        )1544    def connect_api_gateway_to_kinesis(self, gateway_name, kinesis_stream):1545        template = self.APIGATEWAY_DATA_INBOUND_TEMPLATE % kinesis_stream1546        resource_path = self.API_PATH_DATA_INBOUND.replace("/", "")1547        resources = {1548            resource_path: [1549                {1550                    "httpMethod": "POST",1551                    "authorizationType": "NONE",1552                    "requestModels": {"application/json": "Empty"},1553                    "integrations": [1554                        {1555                            "type": "AWS",1556                            "uri": "arn:aws:apigateway:%s:kinesis:action/PutRecords"1557                            % aws_stack.get_region(),1558                            "requestTemplates": {"application/json": template},1559                        }1560                    ],1561                },1562                {1563                    "httpMethod": "GET",1564                    "authorizationType": "NONE",1565                    "requestModels": {"application/json": "Empty"},1566                    "integrations": [1567                        {1568                            "type": "AWS",1569                            "uri": "arn:aws:apigateway:%s:kinesis:action/ListStreams"1570                            % aws_stack.get_region(),1571                            "requestTemplates": {"application/json": "{}"},1572                        }1573                    ],1574                },1575            ]1576        }1577        return aws_stack.create_api_gateway(1578            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1579        )1580    def connect_api_gateway_to_http(1581        self, int_type, gateway_name, target_url, methods=None, path=None1582    ):1583        if methods is None:1584            methods = []1585        if not methods:1586            methods = ["GET", "POST"]1587        if not path:1588            path = "/"1589        resources = {}1590        resource_path = path.replace("/", "")1591        req_templates = (1592            {"application/json": json.dumps({"foo": "bar"})} if int_type == "custom" else {}1593        )1594        resources[resource_path] = [1595            {1596                "httpMethod": method,1597                "integrations": [1598                    {1599                        "type": "HTTP" if int_type == "custom" else "HTTP_PROXY",1600                        "uri": target_url,1601                        "requestTemplates": req_templates,1602                        "responseTemplates": {},1603                    }1604                ],1605            }1606            for method in methods1607        ]1608        return aws_stack.create_api_gateway(1609            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1610        )1611    @staticmethod1612    def create_lambda_function(fn_name):1613        testutil.create_lambda_function(1614            handler_file=TEST_LAMBDA_PYTHON, libs=TEST_LAMBDA_LIBS, func_name=fn_name1615        )1616    def test_apigw_test_invoke_method_api(self, apigateway_client, create_rest_apigw):1617        lambda_client = aws_stack.create_external_boto_client("lambda")1618        # create test Lambda1619        fn_name = f"test-{short_uid()}"1620        testutil.create_lambda_function(1621            handler_file=TEST_LAMBDA_NODEJS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X1622        )1623        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1624        # create REST API and test resource1625        rest_api_id, _, _ = create_rest_apigw(name="test", description="test")1626        root_resource = apigateway_client.get_resources(restApiId=rest_api_id)1627        resource = apigateway_client.create_resource(1628            restApiId=rest_api_id, parentId=root_resource["items"][0]["id"], pathPart="foo"1629        )1630        # create method and integration1631        apigateway_client.put_method(1632            restApiId=rest_api_id,1633            resourceId=resource["id"],1634            httpMethod="GET",1635            authorizationType="NONE",1636        )1637        apigateway_client.put_integration(1638            restApiId=rest_api_id,1639            resourceId=resource["id"],1640            httpMethod="GET",1641            integrationHttpMethod="GET",1642            type="AWS",1643            uri="arn:aws:apigateway:{}:lambda:path//2015-03-31/functions/{}/invocations".format(1644                aws_stack.get_region(), lambda_arn_11645            ),1646        )1647        # run test_invoke_method API #11648        response = apigateway_client.test_invoke_method(1649            restApiId=rest_api_id,1650            resourceId=resource["id"],1651            httpMethod="GET",1652            pathWithQueryString="/foo",1653        )1654        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]1655        assert 200 == response.get("status")1656        assert "response from" in json.loads(response.get("body")).get("body")1657        # run test_invoke_method API #21658        response = apigateway_client.test_invoke_method(1659            restApiId=rest_api_id,1660            resourceId=resource["id"],1661            httpMethod="GET",1662            pathWithQueryString="/foo",1663            body='{"test": "val123"}',1664            headers={"content-type": "application/json"},1665        )1666        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]1667        assert 200 == response.get("status")1668        assert "response from" in json.loads(response.get("body")).get("body")1669        assert "val123" in json.loads(response.get("body")).get("body")1670        # Clean up1671        lambda_client.delete_function(FunctionName=fn_name)1672    @staticmethod1673    def start_http_backend(test_port):1674        # test listener for target HTTP backend1675        class TestListener(ProxyListener):1676            def forward_request(self, **kwargs):1677                response = Response()1678                response.status_code = 2001679                result = {1680                    "data": kwargs.get("data") or "{}",1681                    "headers": dict(kwargs.get("headers")),1682                }1683                response._content = json.dumps(json_safe(result))1684                return response1685        proxy = start_proxy(test_port, update_listener=TestListener())1686        return proxy1687    @staticmethod1688    def create_api_gateway_and_deploy(1689        response_templates=None,1690        is_api_key_required=False,1691        integration_type=None,1692        integration_responses=None,1693    ):1694        response_templates = response_templates or {}1695        integration_type = integration_type or "AWS_PROXY"1696        apigw_client = aws_stack.create_external_boto_client("apigateway")1697        response = apigw_client.create_rest_api(name="my_api", description="this is my api")1698        api_id = response["id"]1699        resources = apigw_client.get_resources(restApiId=api_id)1700        root_resources = [resource for resource in resources["items"] if resource["path"] == "/"]1701        root_id = root_resources[0]["id"]1702        kwargs = {}1703        if integration_type == "AWS_PROXY":1704            aws_stack.create_dynamodb_table("MusicCollection", partition_key="id")1705            kwargs[1706                "uri"1707            ] = "arn:aws:apigateway:us-east-1:dynamodb:action/PutItem&Table=MusicCollection"1708        if not integration_responses:1709            integration_responses = [{"httpMethod": "PUT", "statusCode": "200"}]1710        for resp_details in integration_responses:1711            apigw_client.put_method(1712                restApiId=api_id,1713                resourceId=root_id,1714                httpMethod=resp_details["httpMethod"],1715                authorizationType="NONE",1716                apiKeyRequired=is_api_key_required,1717            )1718            apigw_client.put_method_response(1719                restApiId=api_id,1720                resourceId=root_id,1721                httpMethod=resp_details["httpMethod"],1722                statusCode="200",1723            )1724            apigw_client.put_integration(1725                restApiId=api_id,1726                resourceId=root_id,1727                httpMethod=resp_details["httpMethod"],1728                integrationHttpMethod=resp_details["httpMethod"],1729                type=integration_type,1730                **kwargs,1731            )1732            apigw_client.put_integration_response(1733                restApiId=api_id,1734                resourceId=root_id,1735                selectionPattern="",1736                responseTemplates=response_templates,1737                **resp_details,1738            )1739        apigw_client.create_deployment(restApiId=api_id, stageName="staging")1740        return api_id1741def test_import_swagger_api(apigateway_client):1742    apigateway_client.get_rest_apis()1743    api_spec = load_test_resource("openapi.swagger.json")1744    api_spec_dict = json.loads(api_spec)1745    backend = apigateway_backends[TEST_AWS_ACCOUNT_ID][TEST_AWS_REGION_NAME]1746    api_model = backend.create_rest_api(name="", description="")1747    imported_api = import_api_from_openapi_spec(api_model, api_spec_dict, {})1748    # test_cfn_handle_serverless_api_resource fails if we support title1749    # assert imported_api.name == api_spec_dict.get("info").get("title")1750    assert imported_api.description == api_spec_dict.get("info").get("description")1751    # assert that are no multiple authorizers1752    assert len(imported_api.authorizers) == 11753    paths = {v.path_part for k, v in imported_api.resources.items()}1754    assert paths == {"/", "pets", "{petId}"}1755    resource_methods = {v.path_part: v.resource_methods for k, v in imported_api.resources.items()}1756    methods = {kk[0] for k, v in resource_methods.items() for kk in v.items()}1757    assert methods == {"POST", "OPTIONS", "GET"}1758    assert resource_methods.get("/").get("GET")["methodResponses"] == {1759        "200": {1760            "statusCode": "200",1761            "responseModels": None,1762            "responseParameters": {"method.response.header.Content-Type": "'text/html'"},1763        }1764    }1765    assert resource_methods.get("pets").get("GET")["methodResponses"] == {1766        "200": {1767            "responseModels": {1768                "application/json": {1769                    "items": {1770                        "properties": {1771                            "id": {"type": "integer"},1772                            "price": {"type": "number"},1773                            "type": {"type": "string"},1774                        },1775                        "type": "object",1776                    },1777                    "type": "array",1778                }1779            },1780            "responseParameters": {"method.response.header.Access-Control-Allow-Origin": "'*'"},1781            "statusCode": "200",1782        }1783    }1784@pytest.mark.skipif(not use_docker(), reason="Rust lambdas cannot be executed in local executor")1785def test_apigateway_rust_lambda(1786    apigateway_client,1787    create_rest_apigw,1788    create_lambda_function,1789    create_iam_role_with_policy,1790):1791    function_name = f"test-rust-function-{short_uid()}"1792    api_gateway_name = f"api_gateway_{short_uid()}"1793    role_name = f"test_apigateway_role_{short_uid()}"1794    policy_name = f"test_apigateway_policy_{short_uid()}"1795    stage_name = "test"1796    first_name = f"test_name_{short_uid()}"1797    lambda_create_response = create_lambda_function(1798        func_name=function_name,1799        zip_file=load_file(TEST_LAMBDA_HTTP_RUST, mode="rb"),1800        handler="bootstrap.is.the.handler",1801        runtime="provided.al2",1802    )1803    role_arn = create_iam_role_with_policy(1804        RoleName=role_name,1805        PolicyName=policy_name,1806        RoleDefinition=APIGATEWAY_ASSUME_ROLE_POLICY,1807        PolicyDefinition=APIGATEWAY_LAMBDA_POLICY,1808    )1809    lambda_arn = lambda_create_response["CreateFunctionResponse"]["FunctionArn"]1810    rest_api_id, _, _ = create_rest_apigw(name=api_gateway_name)1811    root_resource_id = apigateway_client.get_resources(restApiId=rest_api_id)["items"][0]["id"]1812    apigateway_client.put_method(1813        restApiId=rest_api_id,1814        resourceId=root_resource_id,1815        httpMethod="GET",1816        authorizationType="NONE",1817    )1818    apigateway_client.put_method_response(1819        restApiId=rest_api_id, resourceId=root_resource_id, httpMethod="GET", statusCode="200"1820    )1821    lambda_target_uri = aws_stack.apigateway_invocations_arn(1822        lambda_uri=lambda_arn, region_name=apigateway_client.meta.region_name1823    )1824    apigateway_client.put_integration(1825        restApiId=rest_api_id,1826        resourceId=root_resource_id,1827        httpMethod="GET",1828        type="AWS_PROXY",1829        integrationHttpMethod="POST",1830        uri=lambda_target_uri,1831        credentials=role_arn,1832    )1833    apigateway_client.create_deployment(restApiId=rest_api_id, stageName=stage_name)1834    url = path_based_url(1835        api_id=rest_api_id, stage_name=stage_name, path=f"/?first_name={first_name}"1836    )1837    result = requests.get(url)1838    assert result.text == f"Hello, {first_name}!"1839def test_apigw_call_api_with_aws_endpoint_url():1840    headers = aws_stack.mock_aws_request_headers("apigateway")1841    headers["Host"] = "apigateway.us-east-2.amazonaws.com:4566"1842    url = f"{get_edge_url()}/apikeys?includeValues=true&name=test%40example.org"1843    response = requests.get(url, headers=headers)1844    assert response.ok1845    content = json.loads(to_str(response.content))1846    assert isinstance(content.get("item"), list)1847@pytest.mark.parametrize("method", ["GET", "ANY"])1848@pytest.mark.parametrize("url_function", [path_based_url, host_based_url])1849def test_rest_api_multi_region(method, url_function):1850    apigateway_client_eu = _client("apigateway", region_name="eu-west-1")1851    apigateway_client_us = _client("apigateway", region_name="us-west-1")1852    api_eu_id, _, root_resource_eu_id = create_rest_api(apigateway_client_eu, name="test-eu-region")1853    api_us_id, _, root_resource_us_id = create_rest_api(apigateway_client_us, name="test-us-region")1854    resource_eu_id, _ = create_rest_resource(1855        apigateway_client_eu, restApiId=api_eu_id, parentId=root_resource_eu_id, pathPart="demo"1856    )1857    resource_us_id, _ = create_rest_resource(1858        apigateway_client_us, restApiId=api_us_id, parentId=root_resource_us_id, pathPart="demo"1859    )1860    create_rest_resource_method(1861        apigateway_client_eu,1862        restApiId=api_eu_id,1863        resourceId=resource_eu_id,1864        httpMethod=method,1865        authorizationType="None",1866    )1867    create_rest_resource_method(1868        apigateway_client_us,1869        restApiId=api_us_id,1870        resourceId=resource_us_id,1871        httpMethod=method,1872        authorizationType="None",1873    )1874    lambda_name = f"lambda-{short_uid()}"1875    testutil.create_lambda_function(1876        handler_file=TEST_LAMBDA_NODEJS,1877        func_name=lambda_name,1878        runtime=LAMBDA_RUNTIME_NODEJS14X,1879        region_name="eu-west-1",1880    )1881    testutil.create_lambda_function(1882        handler_file=TEST_LAMBDA_NODEJS,1883        func_name=lambda_name,1884        runtime=LAMBDA_RUNTIME_NODEJS14X,1885        region_name="us-west-1",1886    )1887    lambda_eu_arn = aws_stack.lambda_function_arn(lambda_name, region_name="eu-west-1")1888    uri_eu = aws_stack.apigateway_invocations_arn(lambda_eu_arn, region_name="eu-west-1")1889    integration_uri, _ = create_rest_api_integration(1890        apigateway_client_eu,1891        restApiId=api_eu_id,1892        resourceId=resource_eu_id,1893        httpMethod=method,1894        type="AWS_PROXY",1895        integrationHttpMethod=method,1896        uri=uri_eu,1897    )1898    lambda_us_arn = aws_stack.lambda_function_arn(lambda_name, region_name="us-west-1")1899    uri_us = aws_stack.apigateway_invocations_arn(lambda_us_arn, region_name="us-west-1")1900    integration_uri, _ = create_rest_api_integration(1901        apigateway_client_us,1902        restApiId=api_us_id,1903        resourceId=resource_us_id,1904        httpMethod=method,1905        type="AWS_PROXY",1906        integrationHttpMethod=method,1907        uri=uri_us,1908    )1909    # test valid authorization using bearer token1910    endpoint = url_function(api_eu_id, stage_name="local", path="/demo")1911    result = requests.get(endpoint, headers={}, verify=False)1912    assert result.status_code == 2001913    endpoint = url_function(api_us_id, stage_name="local", path="/demo")1914    result = requests.get(endpoint, headers={}, verify=False)1915    assert result.status_code == 2001916    delete_rest_api(apigateway_client_eu, restApiId=api_eu_id)1917    delete_rest_api(apigateway_client_us, restApiId=api_us_id)1918    testutil.delete_lambda_function(name=lambda_name, region_name="eu-west-1")1919    testutil.delete_lambda_function(name=lambda_name, region_name="us-west-1")1920@pytest.mark.parametrize("method", ["GET", "POST"])1921@pytest.mark.parametrize("url_function", [path_based_url, host_based_url])1922@pytest.mark.parametrize("passthrough_behaviour", ["WHEN_NO_MATCH", "NEVER", "WHEN_NO_TEMPLATES"])1923def test_mock_integration_response(1924    apigateway_client, method, url_function, passthrough_behaviour, create_rest_apigw1925):1926    api_id, _, root_resource_id = create_rest_apigw(name="mock-api")1927    resource_id, _ = create_rest_resource(1928        apigateway_client, restApiId=api_id, parentId=root_resource_id, pathPart="{id}"1929    )1930    create_rest_resource_method(1931        apigateway_client,1932        restApiId=api_id,1933        resourceId=resource_id,1934        httpMethod=method,1935        authorizationType="NONE",1936    )1937    integration_uri, _ = create_rest_api_integration(1938        apigateway_client,1939        restApiId=api_id,1940        resourceId=resource_id,1941        httpMethod=method,1942        type="MOCK",1943        integrationHttpMethod=method,1944        passthroughBehavior=passthrough_behaviour,1945        requestTemplates={"application/json": '{"statusCode":200}'},1946    )1947    status_code = create_rest_api_method_response(1948        apigateway_client,1949        restApiId=api_id,1950        resourceId=resource_id,1951        httpMethod=method,1952        statusCode="200",1953        responseModels={"application/json": "Empty"},1954    )1955    create_rest_api_integration_response(1956        apigateway_client,1957        restApiId=api_id,1958        resourceId=resource_id,1959        httpMethod=method,1960        statusCode=status_code,1961        responseTemplates={...

apigateway_fixtures.py

Source:apigateway_fixtures.py

...64    assert_response_is_200(response)65def get_rest_api_integration(apigateway_client, **kwargs):66    response = apigateway_client.get_integration(**kwargs)67    assert_response_is_200(response)68def create_rest_api_method_response(apigateway_client, **kwargs):69    response = apigateway_client.put_method_response(**kwargs)70    assert_response_is_200(response)71    return response.get("statusCode")72def create_rest_api_integration_response(apigateway_client, **kwargs):73    response = apigateway_client.put_integration_response(**kwargs)74    assert_response_is_200(response)75    return response.get("statusCode")76def create_domain_name(apigateway_client, **kwargs):77    response = apigateway_client.create_domain_name(**kwargs)78    assert_response_is_200(response)79def create_base_path_mapping(apigateway_client, **kwargs):80    response = apigateway_client.create_base_path_mapping(**kwargs)81    assert_response_is_200(response)82    return response.get("basePath"), response.get("stage")...

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.