How to use host_based_url method in localstack

Best Python code snippet using localstack_python

test_apigateway.py

Source:test_apigateway.py

1# -*- coding: utf-8 -*-2import base643import json4import os5import re6from collections import namedtuple7from typing import Callable, Optional8import pytest9import xmltodict10from botocore.exceptions import ClientError11from jsonpatch import apply_patch12from moto.apigateway import apigateway_backends13from requests.models import Response14from requests.structures import CaseInsensitiveDict15from localstack import config16from localstack.aws.accounts import get_aws_account_id17from localstack.config import get_edge_url18from localstack.constants import APPLICATION_JSON, HEADER_LOCALSTACK_REQUEST_URL, LOCALHOST_HOSTNAME19from localstack.services.apigateway.helpers import (20    TAG_KEY_CUSTOM_ID,21    connect_api_gateway_to_sqs,22    get_resource_for_path,23    get_rest_api_paths,24    host_based_url,25    import_api_from_openapi_spec,26    path_based_url,27)28from localstack.services.awslambda.lambda_api import add_event_source, use_docker29from localstack.services.awslambda.lambda_utils import (30    LAMBDA_RUNTIME_NODEJS12X,31    LAMBDA_RUNTIME_NODEJS14X,32    LAMBDA_RUNTIME_PYTHON36,33    LAMBDA_RUNTIME_PYTHON39,34)35from localstack.services.generic_proxy import ProxyListener36from localstack.services.infra import start_proxy37from localstack.utils import testutil38from localstack.utils.aws import aws_stack39from localstack.utils.common import clone, get_free_tcp_port, json_safe, load_file40from localstack.utils.common import safe_requests as requests41from localstack.utils.common import select_attributes, short_uid, to_str42from tests.integration.apigateway_fixtures import (43    _client,44    api_invoke_url,45    create_rest_api,46    create_rest_api_integration,47    create_rest_api_integration_response,48    create_rest_api_method_response,49    create_rest_resource,50    create_rest_resource_method,51    delete_rest_api,52)53from tests.integration.awslambda.test_lambda_integration import TEST_STAGE_NAME54from ..unit.test_apigateway import load_test_resource55from .awslambda.test_lambda import (56    TEST_LAMBDA_HTTP_RUST,57    TEST_LAMBDA_LIBS,58    TEST_LAMBDA_NODEJS,59    TEST_LAMBDA_NODEJS_APIGW_INTEGRATION,60    TEST_LAMBDA_PYTHON,61    TEST_LAMBDA_PYTHON_ECHO,62)63APIGATEWAY_ASSUME_ROLE_POLICY = {64    "Statement": {65        "Sid": "",66        "Effect": "Allow",67        "Principal": {"Service": "apigateway.amazonaws.com"},68        "Action": "sts:AssumeRole",69    }70}71APIGATEWAY_LAMBDA_POLICY = {72    "Version": "2012-10-17",73    "Statement": [{"Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": "*"}],74}75THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))76TEST_SWAGGER_FILE_JSON = os.path.join(THIS_FOLDER, "files", "swagger.json")77TEST_SWAGGER_FILE_YAML = os.path.join(THIS_FOLDER, "files", "swagger.yaml")78TEST_IMPORT_REST_API_FILE = os.path.join(THIS_FOLDER, "files", "pets.json")79TEST_IMPORT_PETSTORE_SWAGGER = os.path.join(THIS_FOLDER, "files", "petstore-swagger.json")80TEST_IMPORT_MOCK_INTEGRATION = os.path.join(THIS_FOLDER, "files", "openapi-mock.json")81TEST_IMPORT_REST_API_ASYNC_LAMBDA = os.path.join(THIS_FOLDER, "files", "api_definition.yaml")82ApiGatewayLambdaProxyIntegrationTestResult = namedtuple(83    "ApiGatewayLambdaProxyIntegrationTestResult",84    [85        "data",86        "resource",87        "result",88        "url",89        "path_with_replace",90    ],91)92class TestAPIGateway:93    # template used to transform incoming requests at the API Gateway (stream name to be filled in later)94    APIGATEWAY_DATA_INBOUND_TEMPLATE = """{95        "StreamName": "%s",96        "Records": [97            #set( $numRecords = $input.path('$.records').size() )98            #if($numRecords > 0)99            #set( $maxIndex = $numRecords - 1 )100            #foreach( $idx in [0..$maxIndex] )101            #set( $elem = $input.path("$.records[${idx}]") )102            #set( $elemJsonB64 = $util.base64Encode($elem.data) )103            {104                "Data": "$elemJsonB64",105                "PartitionKey": #if( $elem.partitionKey != '')"$elem.partitionKey"106                                #else"$elemJsonB64.length()"#end107            }#if($foreach.hasNext),#end108            #end109            #end110        ]111    }"""112    # endpoint paths113    API_PATH_DATA_INBOUND = "/data"114    API_PATH_HTTP_BACKEND = "/hello_world"115    API_PATH_LAMBDA_PROXY_BACKEND = "/lambda/foo1"116    API_PATH_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM = "/lambda/{test_param1}"117    API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD = "/lambda-any-method/foo1"118    API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM = "/lambda-any-method/{test_param1}"119    API_PATH_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64 = "/lambda-is-base64/foo1"120    # name of Kinesis stream connected to API Gateway121    TEST_STREAM_KINESIS_API_GW = "test-stream-api-gw"122    TEST_STAGE_NAME = "testing"123    TEST_LAMBDA_PROXY_BACKEND = "test_lambda_apigw_backend"124    TEST_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM = "test_lambda_apigw_backend_path_param"125    TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD = "test_lambda_apigw_backend_any_method"126    TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM = (127        "test_lambda_apigw_backend_any_method_path_param"128    )129    TEST_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64 = "test_lambda_apigw_backend_with_is_base64"130    TEST_LAMBDA_SQS_HANDLER_NAME = "lambda_sqs_handler"131    TEST_LAMBDA_AUTHORIZER_HANDLER_NAME = "lambda_authorizer_handler"132    TEST_API_GATEWAY_ID = "fugvjdxtri"133    TEST_API_GATEWAY_AUTHORIZER = {134        "name": "test",135        "type": "TOKEN",136        "providerARNs": ["arn:aws:cognito-idp:us-east-1:123412341234:userpool/us-east-1_123412341"],137        "authType": "custom",138        "authorizerUri": "arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/"139        + "arn:aws:lambda:us-east-1:123456789012:function:myApiAuthorizer/invocations",140        "authorizerCredentials": "arn:aws:iam::123456789012:role/apigAwsProxyRole",141        "identitySource": "method.request.header.Authorization",142        "identityValidationExpression": ".*",143        "authorizerResultTtlInSeconds": 300,144    }145    TEST_API_GATEWAY_AUTHORIZER_OPS = [{"op": "replace", "path": "/name", "value": "test1"}]146    @pytest.mark.aws_validated147    def test_delete_rest_api_with_invalid_id(self, apigateway_client):148        with pytest.raises(ClientError) as e:149            apigateway_client.delete_rest_api(restApiId="foobar")150        assert e.value.response["Error"]["Code"] == "NotFoundException"151        assert "Invalid API identifier specified" in e.value.response["Error"]["Message"]152        assert "foobar" in e.value.response["Error"]["Message"]153    def test_create_rest_api_with_custom_id(self):154        client = aws_stack.create_external_boto_client("apigateway")155        apigw_name = f"gw-{short_uid()}"156        test_id = "testId123"157        result = client.create_rest_api(name=apigw_name, tags={TAG_KEY_CUSTOM_ID: test_id})158        assert test_id == result["id"]159        assert apigw_name == result["name"]160        result = client.get_rest_api(restApiId=test_id)161        assert test_id == result["id"]162        assert apigw_name == result["name"]163    def test_api_gateway_kinesis_integration(self):164        # create target Kinesis stream165        stream = aws_stack.create_kinesis_stream(self.TEST_STREAM_KINESIS_API_GW)166        stream.wait_for()167        # create API Gateway and connect it to the target stream168        result = self.connect_api_gateway_to_kinesis(169            "test_gateway1", self.TEST_STREAM_KINESIS_API_GW170        )171        # generate test data172        test_data = {173            "records": [174                {"data": '{"foo": "bar1"}'},175                {"data": '{"foo": "bar2"}'},176                {"data": '{"foo": "bar3"}'},177            ]178        }179        url = path_based_url(180            api_id=result["id"],181            stage_name=self.TEST_STAGE_NAME,182            path=self.API_PATH_DATA_INBOUND,183        )184        # list Kinesis streams via API Gateway185        result = requests.get(url)186        result = json.loads(to_str(result.content))187        assert "StreamNames" in result188        # post test data to Kinesis via API Gateway189        result = requests.post(url, data=json.dumps(test_data))190        result = json.loads(to_str(result.content))191        assert 0 == result["FailedRecordCount"]192        assert len(test_data["records"]) == len(result["Records"])193        # clean up194        kinesis = aws_stack.create_external_boto_client("kinesis")195        kinesis.delete_stream(StreamName=self.TEST_STREAM_KINESIS_API_GW)196    def test_api_gateway_sqs_integration_with_event_source(self):197        # create target SQS stream198        queue_name = f"queue-{short_uid()}"199        queue_url = aws_stack.create_sqs_queue(queue_name)["QueueUrl"]200        # create API Gateway and connect it to the target queue201        result = connect_api_gateway_to_sqs(202            "test_gateway4",203            stage_name=self.TEST_STAGE_NAME,204            queue_arn=queue_name,205            path=self.API_PATH_DATA_INBOUND,206        )207        # create event source for sqs lambda processor208        self.create_lambda_function(self.TEST_LAMBDA_SQS_HANDLER_NAME)209        event_source_data = {210            "FunctionName": self.TEST_LAMBDA_SQS_HANDLER_NAME,211            "EventSourceArn": aws_stack.sqs_queue_arn(queue_name),212            "Enabled": True,213        }214        add_event_source(event_source_data)215        # generate test data216        test_data = {"spam": "eggs & beans"}217        url = path_based_url(218            api_id=result["id"],219            stage_name=self.TEST_STAGE_NAME,220            path=self.API_PATH_DATA_INBOUND,221        )222        result = requests.post(url, data=json.dumps(test_data))223        assert 200 == result.status_code224        parsed_json = xmltodict.parse(result.content)225        result = parsed_json["SendMessageResponse"]["SendMessageResult"]226        body_md5 = result["MD5OfMessageBody"]227        assert "b639f52308afd65866c86f274c59033f" == body_md5228        # clean up229        sqs_client = aws_stack.create_external_boto_client("sqs")230        sqs_client.delete_queue(QueueUrl=queue_url)231        lambda_client = aws_stack.create_external_boto_client("lambda")232        lambda_client.delete_function(FunctionName=self.TEST_LAMBDA_SQS_HANDLER_NAME)233    def test_api_gateway_sqs_integration(self):234        # create target SQS stream235        queue_name = f"queue-{short_uid()}"236        aws_stack.create_sqs_queue(queue_name)237        # create API Gateway and connect it to the target queue238        result = connect_api_gateway_to_sqs(239            "test_gateway4",240            stage_name=self.TEST_STAGE_NAME,241            queue_arn=queue_name,242            path=self.API_PATH_DATA_INBOUND,243        )244        # generate test data245        test_data = {"spam": "eggs"}246        url = path_based_url(247            api_id=result["id"],248            stage_name=self.TEST_STAGE_NAME,249            path=self.API_PATH_DATA_INBOUND,250        )251        result = requests.post(url, data=json.dumps(test_data))252        assert 200 == result.status_code253        messages = aws_stack.sqs_receive_message(queue_name)["Messages"]254        assert 1 == len(messages)255        assert test_data == json.loads(base64.b64decode(messages[0]["Body"]))256    def test_api_gateway_lambda_integration(self, apigateway_client, create_lambda_function):257        """258        API gateway to lambda integration test returns a response with the same body as the lambda function input.259        """260        fn_name = f"test-{short_uid()}"261        create_lambda_function(262            func_name=fn_name,263            handler_file=TEST_LAMBDA_PYTHON_ECHO,264            runtime=LAMBDA_RUNTIME_PYTHON39,265        )266        lambda_arn = aws_stack.lambda_function_arn(fn_name)267        api_id, _, root = create_rest_api(apigateway_client, name="aws lambda api")268        resource_id, _ = create_rest_resource(269            apigateway_client, restApiId=api_id, parentId=root, pathPart="test"270        )271        # create method and integration272        create_rest_resource_method(273            apigateway_client,274            restApiId=api_id,275            resourceId=resource_id,276            httpMethod="GET",277            authorizationType="NONE",278        )279        create_rest_api_integration(280            apigateway_client,281            restApiId=api_id,282            resourceId=resource_id,283            httpMethod="GET",284            integrationHttpMethod="GET",285            type="AWS",286            uri=f"arn:aws:apigateway:{aws_stack.get_region()}:lambda:path//2015-03-31/functions/{lambda_arn}/invocations",287        )288        url = api_invoke_url(api_id=api_id, stage="local", path="/test")289        response = requests.get(url)290        body = response.json()291        # authorizer contains an object that does not contain the authorizer type ('lambda', 'sns')292        assert body.get("requestContext").get("authorizer") == {"context": {}, "identity": {}}293    @pytest.mark.parametrize("int_type", ["custom", "proxy"])294    def test_api_gateway_http_integrations(self, int_type, monkeypatch):295        monkeypatch.setattr(config, "DISABLE_CUSTOM_CORS_APIGATEWAY", False)296        test_port = get_free_tcp_port()297        backend_url = "http://localhost:%s%s" % (test_port, self.API_PATH_HTTP_BACKEND)298        # start test HTTP backend299        proxy = self.start_http_backend(test_port)300        # create API Gateway and connect it to the HTTP_PROXY/HTTP backend301        result = self.connect_api_gateway_to_http(302            int_type, "test_gateway2", backend_url, path=self.API_PATH_HTTP_BACKEND303        )304        url = path_based_url(305            api_id=result["id"],306            stage_name=self.TEST_STAGE_NAME,307            path=self.API_PATH_HTTP_BACKEND,308        )309        # make sure CORS headers are present310        origin = "localhost"311        result = requests.options(url, headers={"origin": origin})312        assert result.status_code == 200313        assert re.match(result.headers["Access-Control-Allow-Origin"].replace("*", ".*"), origin)314        assert "POST" in result.headers["Access-Control-Allow-Methods"]315        assert "PATCH" in result.headers["Access-Control-Allow-Methods"]316        custom_result = json.dumps({"foo": "bar"})317        # make test GET request to gateway318        result = requests.get(url)319        assert 200 == result.status_code320        expected = custom_result if int_type == "custom" else "{}"321        assert expected == json.loads(to_str(result.content))["data"]322        # make test POST request to gateway323        data = json.dumps({"data": 123})324        result = requests.post(url, data=data)325        assert 200 == result.status_code326        expected = custom_result if int_type == "custom" else data327        assert expected == json.loads(to_str(result.content))["data"]328        # make test POST request with non-JSON content type329        data = "test=123"330        ctype = "application/x-www-form-urlencoded"331        result = requests.post(url, data=data, headers={"content-type": ctype})332        assert 200 == result.status_code333        content = json.loads(to_str(result.content))334        headers = CaseInsensitiveDict(content["headers"])335        expected = custom_result if int_type == "custom" else data336        assert expected == content["data"]337        assert ctype == headers["content-type"]338        # clean up339        proxy.stop()340    def test_api_gateway_lambda_proxy_integration(self):341        self._test_api_gateway_lambda_proxy_integration(342            self.TEST_LAMBDA_PROXY_BACKEND, self.API_PATH_LAMBDA_PROXY_BACKEND343        )344    def test_api_gateway_lambda_proxy_integration_with_path_param(self):345        self._test_api_gateway_lambda_proxy_integration(346            self.TEST_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,347            self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,348        )349    def test_api_gateway_lambda_proxy_integration_with_is_base_64_encoded(self):350        # Test the case where `isBase64Encoded` is enabled.351        content = b"hello, please base64 encode me"352        def _mutate_data(data) -> None:353            data["return_is_base_64_encoded"] = True354            data["return_raw_body"] = base64.b64encode(content).decode("utf8")355        test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(356            self.TEST_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,357            self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,358            data_mutator_fn=_mutate_data,359        )360        # Ensure that `invoke_rest_api_integration_backend` correctly decodes the base64 content361        assert test_result.result.status_code == 203362        assert test_result.result.content == content363    def _test_api_gateway_lambda_proxy_integration_no_asserts(364        self,365        fn_name: str,366        path: str,367        data_mutator_fn: Optional[Callable] = None,368    ) -> ApiGatewayLambdaProxyIntegrationTestResult:369        """370        Perform the setup needed to do a POST against a Lambda Proxy Integration;371        then execute the POST.372        :param data_mutator_fn: a Callable[[Dict], None] that lets us mutate the373          data dictionary before sending it off to the lambda.374        """375        self.create_lambda_function(fn_name)376        # create API Gateway and connect it to the Lambda proxy backend377        lambda_uri = aws_stack.lambda_function_arn(fn_name)378        invocation_uri = "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations"379        target_uri = invocation_uri % (aws_stack.get_region(), lambda_uri)380        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(381            "test_gateway2", target_uri, path=path, stage_name=self.TEST_STAGE_NAME382        )383        api_id = result["id"]384        path_map = get_rest_api_paths(api_id)385        _, resource = get_resource_for_path(path, path_map)386        # make test request to gateway and check response387        path_with_replace = path.replace("{test_param1}", "foo1")388        path_with_params = path_with_replace + "?foo=foo&bar=bar&bar=baz"389        url = path_based_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path=path_with_params)390        # These values get read in `lambda_integration.py`391        data = {"return_status_code": 203, "return_headers": {"foo": "bar123"}}392        if data_mutator_fn:393            assert callable(data_mutator_fn)394            data_mutator_fn(data)395        result = requests.post(396            url,397            data=json.dumps(data),398            headers={"User-Agent": "python-requests/testing"},399        )400        return ApiGatewayLambdaProxyIntegrationTestResult(401            data=data,402            resource=resource,403            result=result,404            url=url,405            path_with_replace=path_with_replace,406        )407    def _test_api_gateway_lambda_proxy_integration(408        self,409        fn_name: str,410        path: str,411    ) -> None:412        test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(fn_name, path)413        data, resource, result, url, path_with_replace = test_result414        assert result.status_code == 203415        assert result.headers.get("foo") == "bar123"416        assert "set-cookie" in result.headers417        try:418            parsed_body = json.loads(to_str(result.content))419        except json.decoder.JSONDecodeError as e:420            raise Exception(421                "Couldn't json-decode content: {}".format(to_str(result.content))422            ) from e423        assert parsed_body.get("return_status_code") == 203424        assert parsed_body.get("return_headers") == {"foo": "bar123"}425        assert parsed_body.get("queryStringParameters") == {"foo": "foo", "bar": ["bar", "baz"]}426        request_context = parsed_body.get("requestContext")427        source_ip = request_context["identity"].pop("sourceIp")428        assert re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", source_ip)429        expected_path = "/" + self.TEST_STAGE_NAME + "/lambda/foo1"430        assert expected_path == request_context["path"]431        assert request_context.get("stageVariables") is None432        assert get_aws_account_id() == request_context["accountId"]433        assert resource.get("id") == request_context["resourceId"]434        assert self.TEST_STAGE_NAME == request_context["stage"]435        assert "python-requests/testing" == request_context["identity"]["userAgent"]436        assert "POST" == request_context["httpMethod"]437        assert "HTTP/1.1" == request_context["protocol"]438        assert "requestTimeEpoch" in request_context439        assert "requestTime" in request_context440        assert "requestId" in request_context441        # assert that header keys are lowercase (as in AWS)442        headers = parsed_body.get("headers") or {}443        header_names = list(headers.keys())444        assert "Host" in header_names445        assert "Content-Length" in header_names446        assert "User-Agent" in header_names447        result = requests.delete(url, data=json.dumps(data))448        assert 204 == result.status_code449        # send message with non-ASCII chars450        body_msg = "ð - åã"451        result = requests.post(url, data=json.dumps({"return_raw_body": body_msg}))452        assert body_msg == to_str(result.content)453        # send message with binary data454        binary_msg = b"\xff \xaa \x11"455        result = requests.post(url, data=binary_msg)456        result_content = json.loads(to_str(result.content))457        assert "/yCqIBE=" == result_content["body"]458        assert ["isBase64Encoded"]459    def test_api_gateway_lambda_proxy_integration_any_method(self):460        self._test_api_gateway_lambda_proxy_integration_any_method(461            self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,462            self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD,463        )464    def test_api_gateway_lambda_proxy_integration_any_method_with_path_param(self):465        self._test_api_gateway_lambda_proxy_integration_any_method(466            self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,467            self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,468        )469    def test_api_gateway_lambda_asynchronous_invocation(self, apigateway_client):470        api_gateway_name = f"api_gateway_{short_uid()}"471        rest_api_id = apigateway_client.create_rest_api(name=api_gateway_name)["id"]472        fn_name = f"test-{short_uid()}"473        testutil.create_lambda_function(474            handler_file=TEST_LAMBDA_NODEJS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X475        )476        lambda_arn = aws_stack.lambda_function_arn(fn_name)477        spec_file = load_file(TEST_IMPORT_REST_API_ASYNC_LAMBDA)478        spec_file = spec_file.replace("${lambda_invocation_arn}", lambda_arn)479        apigateway_client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")480        url = path_based_url(api_id=rest_api_id, stage_name="latest", path="/wait/3")481        result = requests.get(url)482        assert result.status_code == 200483        assert result.content == b""484    def test_api_gateway_mock_integration(self, apigateway_client):485        rest_api_name = f"apigw-{short_uid()}"486        rest_api_id = apigateway_client.create_rest_api(name=rest_api_name)["id"]487        spec_file = load_file(TEST_IMPORT_MOCK_INTEGRATION)488        apigateway_client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")489        url = path_based_url(api_id=rest_api_id, stage_name="latest", path="/echo/foobar")490        response = requests.get(url)491        assert response._content == b'{"echo": "foobar", "response": "mocked"}'492    def test_api_gateway_authorizer_crud(self):493        apig = aws_stack.create_external_boto_client("apigateway")494        authorizer = apig.create_authorizer(495            restApiId=self.TEST_API_GATEWAY_ID, **self.TEST_API_GATEWAY_AUTHORIZER496        )497        authorizer_id = authorizer.get("id")498        create_result = apig.get_authorizer(499            restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id500        )501        # ignore boto3 stuff502        del create_result["ResponseMetadata"]503        create_expected = clone(self.TEST_API_GATEWAY_AUTHORIZER)504        create_expected["id"] = authorizer_id505        assert create_expected == create_result506        apig.update_authorizer(507            restApiId=self.TEST_API_GATEWAY_ID,508            authorizerId=authorizer_id,509            patchOperations=self.TEST_API_GATEWAY_AUTHORIZER_OPS,510        )511        update_result = apig.get_authorizer(512            restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id513        )514        # ignore boto3 stuff515        del update_result["ResponseMetadata"]516        update_expected = apply_patch(create_expected, self.TEST_API_GATEWAY_AUTHORIZER_OPS)517        assert update_expected == update_result518        apig.delete_authorizer(restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id)519        with pytest.raises(Exception):520            apig.get_authorizer(self.TEST_API_GATEWAY_ID, authorizer_id)521    def test_apigateway_with_lambda_integration(self):522        apigw_client = aws_stack.create_external_boto_client("apigateway")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")1785    result = requests.get(endpoint, headers={}, verify=False)1786    assert result.status_code == 2001787    delete_rest_api(apigateway_client_eu, restApiId=api_eu_id)1788    delete_rest_api(apigateway_client_us, restApiId=api_us_id)1789    testutil.delete_lambda_function(name=lambda_name, region_name="eu-west-1")1790    testutil.delete_lambda_function(name=lambda_name, region_name="us-west-1")1791@pytest.mark.parametrize("method", ["GET", "POST"])1792@pytest.mark.parametrize("url_function", [path_based_url, host_based_url])1793@pytest.mark.parametrize("passthrough_behaviour", ["WHEN_NO_MATCH", "NEVER", "WHEN_NO_TEMPLATES"])1794def test_mock_integration_response(apigateway_client, method, url_function, passthrough_behaviour):1795    api_id, _, root_resource_id = create_rest_api(apigateway_client, name="mock-api")1796    resource_id, _ = create_rest_resource(1797        apigateway_client, restApiId=api_id, parentId=root_resource_id, pathPart="{id}"1798    )1799    create_rest_resource_method(1800        apigateway_client,1801        restApiId=api_id,1802        resourceId=resource_id,1803        httpMethod=method,1804        authorizationType="NONE",1805    )1806    integration_uri, _ = create_rest_api_integration(1807        apigateway_client,1808        restApiId=api_id,1809        resourceId=resource_id,1810        httpMethod=method,1811        type="MOCK",1812        integrationHttpMethod=method,1813        passthroughBehavior=passthrough_behaviour,1814        requestTemplates={"application/json": '{"statusCode":200}'},1815    )1816    status_code = create_rest_api_method_response(1817        apigateway_client,1818        restApiId=api_id,1819        resourceId=resource_id,1820        httpMethod=method,1821        statusCode="200",1822        responseModels={"application/json": "Empty"},1823    )1824    create_rest_api_integration_response(1825        apigateway_client,1826        restApiId=api_id,1827        resourceId=resource_id,1828        httpMethod=method,1829        statusCode=status_code,1830        responseTemplates={1831            "application/json": '{"statusCode": 200, "id": $input.params().path.id}'1832        },1833    )1834    endpoint = url_function(api_id, stage_name="local", path="/42")1835    result = requests.request(1836        method,1837        endpoint,1838        headers={"Content-Type": "application/json"},1839        verify=False,1840    )1841    assert result.status_code == 2001842    assert to_str(result.content) == '{"statusCode": 200, "id": 42}'1843    delete_rest_api(apigateway_client, restApiId=api_id)1844def test_tag_api():1845    client = aws_stack.connect_to_service("apigateway")1846    api_name = f"api-{short_uid()}"1847    tags = {"foo": "bar"}1848    # add resource tags1849    result = client.create_rest_api(name=api_name)1850    api_id = result["id"]1851    api_arn = aws_stack.apigateway_restapi_arn(api_id=api_id)1852    client.tag_resource(resourceArn=api_arn, tags=tags)1853    # receive and assert tags1854    tags_saved = client.get_tags(resourceArn=api_arn)["tags"]1855    assert tags == tags_saved1856    # clean up...

apigateway_fixtures.py

Source:apigateway_fixtures.py

...122    if is_aws_cloud():123        stage = f"/{stage}" if stage else ""124        return f"https://{api_id}.execute-api.{aws_stack.get_boto3_region()}.amazonaws.com{stage}{path}"125    if url_type == UrlType.HOST_BASED:126        return host_based_url(api_id, stage_name=stage, path=path)...

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.