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

Best Python code snippet using localstack_python
test_api_gateway.py
Source:test_api_gateway.py
...218        backend_url = "http://localhost:%s%s" % (test_port, self.API_PATH_HTTP_BACKEND)219        old_config = config.DISABLE_CUSTOM_CORS_APIGATEWAY220        config.DISABLE_CUSTOM_CORS_APIGATEWAY = False221        # start test HTTP backend222        proxy = self.start_http_backend(test_port)223        # create API Gateway and connect it to the HTTP_PROXY/HTTP backend224        result = self.connect_api_gateway_to_http(225            int_type, "test_gateway2", backend_url, path=self.API_PATH_HTTP_BACKEND226        )227        url = gateway_request_url(228            api_id=result["id"],229            stage_name=self.TEST_STAGE_NAME,230            path=self.API_PATH_HTTP_BACKEND,231        )232        # make sure CORS headers are present233        origin = "localhost"234        result = requests.options(url, headers={"origin": origin})235        self.assertEqual(result.status_code, 200)236        self.assertTrue(237            re.match(result.headers["Access-Control-Allow-Origin"].replace("*", ".*"), origin)238        )239        self.assertIn("POST", result.headers["Access-Control-Allow-Methods"])240        self.assertIn("PATCH", result.headers["Access-Control-Allow-Methods"])241        custom_result = json.dumps({"foo": "bar"})242        # make test GET request to gateway243        result = requests.get(url)244        self.assertEqual(200, result.status_code)245        expected = custom_result if int_type == "custom" else "{}"246        self.assertEqual(expected, json.loads(to_str(result.content))["data"])247        # make test POST request to gateway248        data = json.dumps({"data": 123})249        result = requests.post(url, data=data)250        self.assertEqual(200, result.status_code)251        expected = custom_result if int_type == "custom" else data252        self.assertEqual(expected, json.loads(to_str(result.content))["data"])253        # make test POST request with non-JSON content type254        data = "test=123"255        ctype = "application/x-www-form-urlencoded"256        result = requests.post(url, data=data, headers={"content-type": ctype})257        self.assertEqual(200, result.status_code)258        content = json.loads(to_str(result.content))259        headers = CaseInsensitiveDict(content["headers"])260        expected = custom_result if int_type == "custom" else data261        self.assertEqual(expected, content["data"])262        self.assertEqual(ctype, headers["content-type"])263        # clean up264        config.DISABLE_CUSTOM_CORS_APIGATEWAY = old_config265        proxy.stop()266    def test_api_gateway_lambda_proxy_integration(self):267        self._test_api_gateway_lambda_proxy_integration(268            self.TEST_LAMBDA_PROXY_BACKEND, self.API_PATH_LAMBDA_PROXY_BACKEND269        )270    def test_api_gateway_lambda_proxy_integration_with_path_param(self):271        self._test_api_gateway_lambda_proxy_integration(272            self.TEST_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,273            self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_PATH_PARAM,274        )275    def test_api_gateway_lambda_proxy_integration_with_is_base_64_encoded(self):276        # Test the case where `isBase64Encoded` is enabled.277        content = b"hello, please base64 encode me"278        def _mutate_data(data) -> None:279            data["return_is_base_64_encoded"] = True280            data["return_raw_body"] = base64.b64encode(content).decode("utf8")281        test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(282            self.TEST_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,283            self.API_PATH_LAMBDA_PROXY_BACKEND_WITH_IS_BASE64,284            data_mutator_fn=_mutate_data,285        )286        # Ensure that `invoke_rest_api_integration_backend` correctly decodes the base64 content287        self.assertEqual(test_result.result.status_code, 203)288        self.assertEqual(test_result.result.content, content)289    def _test_api_gateway_lambda_proxy_integration_no_asserts(290        self,291        fn_name: str,292        path: str,293        data_mutator_fn: Optional[Callable] = None,294    ) -> ApiGatewayLambdaProxyIntegrationTestResult:295        """296        Perform the setup needed to do a POST against a Lambda Proxy Integration;297        then execute the POST.298        :param data_mutator_fn: a Callable[[Dict], None] that lets us mutate the299          data dictionary before sending it off to the lambda.300        """301        self.create_lambda_function(fn_name)302        # create API Gateway and connect it to the Lambda proxy backend303        lambda_uri = aws_stack.lambda_function_arn(fn_name)304        invocation_uri = "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations"305        target_uri = invocation_uri % (aws_stack.get_region(), lambda_uri)306        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(307            "test_gateway2", target_uri, path=path, stage_name=self.TEST_STAGE_NAME308        )309        api_id = result["id"]310        path_map = get_rest_api_paths(api_id)311        _, resource = get_resource_for_path(path, path_map)312        # make test request to gateway and check response313        path_with_replace = path.replace("{test_param1}", "foo1")314        path_with_params = path_with_replace + "?foo=foo&bar=bar&bar=baz"315        url = gateway_request_url(316            api_id=api_id, stage_name=self.TEST_STAGE_NAME, path=path_with_params317        )318        # These values get read in `lambda_integration.py`319        data = {"return_status_code": 203, "return_headers": {"foo": "bar123"}}320        if data_mutator_fn:321            assert callable(data_mutator_fn)322            data_mutator_fn(data)323        result = requests.post(324            url,325            data=json.dumps(data),326            headers={"User-Agent": "python-requests/testing"},327        )328        return ApiGatewayLambdaProxyIntegrationTestResult(329            data=data,330            resource=resource,331            result=result,332            url=url,333            path_with_replace=path_with_replace,334        )335    def _test_api_gateway_lambda_proxy_integration(336        self,337        fn_name: str,338        path: str,339    ) -> None:340        test_result = self._test_api_gateway_lambda_proxy_integration_no_asserts(fn_name, path)341        data, resource, result, url, path_with_replace = test_result342        self.assertEqual(result.status_code, 203)343        self.assertEqual(result.headers.get("foo"), "bar123")344        self.assertIn("set-cookie", result.headers)345        try:346            parsed_body = json.loads(to_str(result.content))347        except json.decoder.JSONDecodeError as e:348            raise Exception(349                "Couldn't json-decode content: {}".format(to_str(result.content))350            ) from e351        self.assertEqual(parsed_body.get("return_status_code"), 203)352        self.assertDictEqual(parsed_body.get("return_headers"), {"foo": "bar123"})353        self.assertDictEqual(354            parsed_body.get("queryStringParameters"),355            {"foo": "foo", "bar": ["bar", "baz"]},356        )357        request_context = parsed_body.get("requestContext")358        source_ip = request_context["identity"].pop("sourceIp")359        self.assertTrue(re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", source_ip))360        expected_path = "/" + self.TEST_STAGE_NAME + "/lambda/foo1"361        self.assertEqual(expected_path, request_context["path"])362        self.assertIsNone(request_context.get("stageVariables"))363        self.assertEqual(TEST_AWS_ACCOUNT_ID, request_context["accountId"])364        self.assertEqual(resource.get("id"), request_context["resourceId"])365        self.assertEqual(self.TEST_STAGE_NAME, request_context["stage"])366        self.assertEqual("python-requests/testing", request_context["identity"]["userAgent"])367        self.assertEqual("POST", request_context["httpMethod"])368        self.assertEqual("HTTP/1.1", request_context["protocol"])369        self.assertIn("requestTimeEpoch", request_context)370        self.assertIn("requestTime", request_context)371        self.assertIn("requestId", request_context)372        # assert that header keys are lowercase (as in AWS)373        headers = parsed_body.get("headers") or {}374        header_names = list(headers.keys())375        self.assertIn("host", header_names)376        self.assertIn("content-length", header_names)377        self.assertIn("user-agent", header_names)378        result = requests.delete(url, data=json.dumps(data))379        self.assertEqual(204, result.status_code)380        # send message with non-ASCII chars381        body_msg = "ð - åã"382        result = requests.post(url, data=json.dumps({"return_raw_body": body_msg}))383        self.assertEqual(body_msg, to_str(result.content))384    def test_api_gateway_lambda_proxy_integration_any_method(self):385        self._test_api_gateway_lambda_proxy_integration_any_method(386            self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,387            self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD,388        )389    def test_api_gateway_lambda_proxy_integration_any_method_with_path_param(self):390        self._test_api_gateway_lambda_proxy_integration_any_method(391            self.TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,392            self.API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD_WITH_PATH_PARAM,393        )394    def test_api_gateway_authorizer_crud(self):395        apig = aws_stack.connect_to_service("apigateway")396        authorizer = apig.create_authorizer(397            restApiId=self.TEST_API_GATEWAY_ID, **self.TEST_API_GATEWAY_AUTHORIZER398        )399        authorizer_id = authorizer.get("id")400        create_result = apig.get_authorizer(401            restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id402        )403        # ignore boto3 stuff404        del create_result["ResponseMetadata"]405        create_expected = clone(self.TEST_API_GATEWAY_AUTHORIZER)406        create_expected["id"] = authorizer_id407        self.assertDictEqual(create_expected, create_result)408        apig.update_authorizer(409            restApiId=self.TEST_API_GATEWAY_ID,410            authorizerId=authorizer_id,411            patchOperations=self.TEST_API_GATEWAY_AUTHORIZER_OPS,412        )413        update_result = apig.get_authorizer(414            restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id415        )416        # ignore boto3 stuff417        del update_result["ResponseMetadata"]418        update_expected = apply_patch(create_expected, self.TEST_API_GATEWAY_AUTHORIZER_OPS)419        self.assertDictEqual(update_expected, update_result)420        apig.delete_authorizer(restApiId=self.TEST_API_GATEWAY_ID, authorizerId=authorizer_id)421        self.assertRaises(Exception, apig.get_authorizer, self.TEST_API_GATEWAY_ID, authorizer_id)422    def test_apigateway_with_lambda_integration(self):423        apigw_client = aws_stack.connect_to_service("apigateway")424        # create Lambda function425        lambda_name = "apigw-lambda-%s" % short_uid()426        self.create_lambda_function(lambda_name)427        lambda_uri = aws_stack.lambda_function_arn(lambda_name)428        target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)429        # create REST API430        api = apigw_client.create_rest_api(name="test-api", description="")431        api_id = api["id"]432        root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]433        api_resource = apigw_client.create_resource(434            restApiId=api_id, parentId=root_res_id, pathPart="test"435        )436        apigw_client.put_method(437            restApiId=api_id,438            resourceId=api_resource["id"],439            httpMethod="GET",440            authorizationType="NONE",441        )442        rs = apigw_client.put_integration(443            restApiId=api_id,444            resourceId=api_resource["id"],445            httpMethod="GET",446            integrationHttpMethod="POST",447            type="AWS",448            uri=target_uri,449            timeoutInMillis=3000,450            contentHandling="CONVERT_TO_BINARY",451            requestTemplates={"application/json": '{"param1": "$input.params(\'param1\')"}'},452        )453        integration_keys = [454            "httpMethod",455            "type",456            "passthroughBehavior",457            "cacheKeyParameters",458            "uri",459            "cacheNamespace",460            "timeoutInMillis",461            "contentHandling",462            "requestParameters",463        ]464        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])465        for key in integration_keys:466            self.assertIn(key, rs)467        self.assertNotIn("responseTemplates", rs)468        apigw_client.create_deployment(restApiId=api_id, stageName=self.TEST_STAGE_NAME)469        rs = apigw_client.get_integration(470            restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"471        )472        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])473        self.assertEqual("AWS", rs["type"])474        self.assertEqual("POST", rs["httpMethod"])475        self.assertEqual(target_uri, rs["uri"])476        # invoke the gateway endpoint477        url = gateway_request_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/test")478        response = requests.get("%s?param1=foobar" % url)479        self.assertLess(response.status_code, 400)480        content = json.loads(to_str(response.content))481        self.assertEqual("GET", content.get("httpMethod"))482        self.assertEqual(api_resource["id"], content.get("requestContext", {}).get("resourceId"))483        self.assertEqual(self.TEST_STAGE_NAME, content.get("requestContext", {}).get("stage"))484        self.assertEqual('{"param1": "foobar"}', content.get("body"))485        # delete integration486        rs = apigw_client.delete_integration(487            restApiId=api_id,488            resourceId=api_resource["id"],489            httpMethod="GET",490        )491        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])492        with self.assertRaises(ClientError) as ctx:493            # This call should not be successful as the integration is deleted494            apigw_client.get_integration(495                restApiId=api_id, resourceId=api_resource["id"], httpMethod="GET"496            )497        self.assertEqual(ctx.exception.response["Error"]["Code"], "NotFoundException")498        # clean up499        lambda_client = aws_stack.connect_to_service("lambda")500        lambda_client.delete_function(FunctionName=lambda_name)501        apigw_client.delete_rest_api(restApiId=api_id)502    def test_api_gateway_handle_domain_name(self):503        domain_name = "%s.example.com" % short_uid()504        apigw_client = aws_stack.connect_to_service("apigateway")505        rs = apigw_client.create_domain_name(domainName=domain_name)506        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])507        rs = apigw_client.get_domain_name(domainName=domain_name)508        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])509        self.assertEqual(domain_name, rs["domainName"])510        # clean up511        apigw_client.delete_domain_name(domainName=domain_name)512    def _test_api_gateway_lambda_proxy_integration_any_method(self, fn_name, path):513        self.create_lambda_function(fn_name)514        # create API Gateway and connect it to the Lambda proxy backend515        lambda_uri = aws_stack.lambda_function_arn(fn_name)516        target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)517        result = testutil.connect_api_gateway_to_http_with_lambda_proxy(518            "test_gateway3",519            target_uri,520            methods=["ANY"],521            path=path,522            stage_name=self.TEST_STAGE_NAME,523        )524        # make test request to gateway and check response525        path = path.replace("{test_param1}", "foo1")526        url = gateway_request_url(api_id=result["id"], stage_name=self.TEST_STAGE_NAME, path=path)527        data = {}528        for method in ("GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"):529            body = json.dumps(data) if method in ("POST", "PUT", "PATCH") else None530            result = getattr(requests, method.lower())(url, data=body)531            if method != "DELETE":532                self.assertEqual(200, result.status_code)533                parsed_body = json.loads(to_str(result.content))534                self.assertEqual(method, parsed_body.get("httpMethod"))535            else:536                self.assertEqual(204, result.status_code)537    def test_apigateway_with_custom_authorization_method(self):538        apigw_client = aws_stack.connect_to_service("apigateway")539        # create Lambda function540        lambda_name = "apigw-lambda-%s" % short_uid()541        self.create_lambda_function(lambda_name)542        lambda_uri = aws_stack.lambda_function_arn(lambda_name)543        # create REST API544        api = apigw_client.create_rest_api(name="test-api", description="")545        api_id = api["id"]546        root_res_id = apigw_client.get_resources(restApiId=api_id)["items"][0]["id"]547        # create authorizer at root resource548        authorizer = apigw_client.create_authorizer(549            restApiId=api_id,550            name="lambda_authorizer",551            type="TOKEN",552            authorizerUri="arn:aws:apigateway:us-east-1:lambda:path/ \553                2015-03-31/functions/{}/invocations".format(554                lambda_uri555            ),556            identitySource="method.request.header.Auth",557        )558        # create method with custom authorizer559        is_api_key_required = True560        method_response = apigw_client.put_method(561            restApiId=api_id,562            resourceId=root_res_id,563            httpMethod="GET",564            authorizationType="CUSTOM",565            authorizerId=authorizer["id"],566            apiKeyRequired=is_api_key_required,567        )568        self.assertEqual(authorizer["id"], method_response["authorizerId"])569        # clean up570        lambda_client = aws_stack.connect_to_service("lambda")571        lambda_client.delete_function(FunctionName=lambda_name)572        apigw_client.delete_rest_api(restApiId=api_id)573    def test_create_model(self):574        client = aws_stack.connect_to_service("apigateway")575        response = client.create_rest_api(name="my_api", description="this is my api")576        rest_api_id = response["id"]577        dummy_rest_api_id = "_non_existing_"578        model_name = "testModel"579        description = "test model"580        content_type = "application/json"581        # success case with valid params582        response = client.create_model(583            restApiId=rest_api_id,584            name=model_name,585            description=description,586            contentType=content_type,587        )588        self.assertEqual(model_name, response["name"])589        self.assertEqual(description, response["description"])590        with self.assertRaises(Exception) as ctx:591            client.create_model(592                restApiId=dummy_rest_api_id,593                name=model_name,594                description=description,595                contentType=content_type,596            )597        self.assertEqual("NotFoundException", ctx.exception.response["Error"]["Code"])598        self.assertEqual(599            "Invalid Rest API Id specified", ctx.exception.response["Error"]["Message"]600        )601        with self.assertRaises(Exception) as ctx:602            client.create_model(603                restApiId=dummy_rest_api_id,604                name="",605                description=description,606                contentType=content_type,607            )608        self.assertEqual("BadRequestException", ctx.exception.response["Error"]["Code"])609        self.assertEqual("No Model Name specified", ctx.exception.response["Error"]["Message"])610        # clean up611        client.delete_rest_api(restApiId=rest_api_id)612    def test_get_api_models(self):613        client = aws_stack.connect_to_service("apigateway")614        response = client.create_rest_api(name="my_api", description="this is my api")615        rest_api_id = response["id"]616        model_name = "testModel"617        description = "test model"618        content_type = "application/json"619        # when no models are present620        result = client.get_models(restApiId=rest_api_id)621        self.assertEqual([], result["items"])622        # add a model623        client.create_model(624            restApiId=rest_api_id,625            name=model_name,626            description=description,627            contentType=content_type,628        )629        # get models after adding630        result = client.get_models(restApiId=rest_api_id)631        self.assertEqual(model_name, result["items"][0]["name"])632        self.assertEqual(description, result["items"][0]["description"])633        # clean up634        client.delete_rest_api(restApiId=rest_api_id)635    def test_request_validator(self):636        client = aws_stack.connect_to_service("apigateway")637        response = client.create_rest_api(name="my_api", description="this is my api")638        rest_api_id = response["id"]639        # CREATE640        name = "validator123"641        result = client.create_request_validator(restApiId=rest_api_id, name=name)642        self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])643        validator_id = result["id"]644        # LIST645        result = client.get_request_validators(restApiId=rest_api_id)646        self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])647        self.assertEqual([{"id": validator_id, "name": name}], result["items"])648        # GET649        result = client.get_request_validator(650            restApiId=rest_api_id, requestValidatorId=validator_id651        )652        self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])653        self.assertEqual(654            select_attributes(result, ["id", "name"]),655            {"id": validator_id, "name": name},656        )657        # UPDATE658        result = client.update_request_validator(659            restApiId=rest_api_id, requestValidatorId=validator_id, patchOperations=[]660        )661        # DELETE662        client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)663        with self.assertRaises(Exception):664            client.get_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)665        with self.assertRaises(Exception):666            client.delete_request_validator(restApiId=rest_api_id, requestValidatorId=validator_id)667        # clean up668        client.delete_rest_api(restApiId=rest_api_id)669    def test_base_path_mapping(self):670        client = aws_stack.connect_to_service("apigateway")671        response = client.create_rest_api(name="my_api", description="this is my api")672        rest_api_id = response["id"]673        # CREATE674        domain_name = "domain1.example.com"675        base_path = "/foo"676        result = client.create_base_path_mapping(677            domainName=domain_name,678            basePath=base_path,679            restApiId=rest_api_id,680            stage="dev",681        )682        self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])683        # LIST684        result = client.get_base_path_mappings(domainName=domain_name)685        self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])686        expected = {"basePath": base_path, "restApiId": rest_api_id, "stage": "dev"}687        self.assertEqual([expected], result["items"])688        # GET689        result = client.get_base_path_mapping(domainName=domain_name, basePath=base_path)690        self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])691        self.assertEqual(expected, select_attributes(result, ["basePath", "restApiId", "stage"]))692        # UPDATE693        result = client.update_base_path_mapping(694            domainName=domain_name, basePath=base_path, patchOperations=[]695        )696        # DELETE697        client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)698        with self.assertRaises(Exception):699            client.get_base_path_mapping(domainName=domain_name, basePath=base_path)700        with self.assertRaises(Exception):701            client.delete_base_path_mapping(domainName=domain_name, basePath=base_path)702    def test_api_account(self):703        client = aws_stack.connect_to_service("apigateway")704        response = client.create_rest_api(name="my_api", description="test 123")705        rest_api_id = response["id"]706        result = client.get_account()707        self.assertIn("UsagePlans", result["features"])708        result = client.update_account(709            patchOperations=[{"op": "add", "path": "/features/-", "value": "foobar"}]710        )711        self.assertIn("foobar", result["features"])712        # clean up713        client.delete_rest_api(restApiId=rest_api_id)714    def test_get_model_by_name(self):715        client = aws_stack.connect_to_service("apigateway")716        response = client.create_rest_api(name="my_api", description="this is my api")717        rest_api_id = response["id"]718        dummy_rest_api_id = "_non_existing_"719        model_name = "testModel"720        description = "test model"721        content_type = "application/json"722        # add a model723        client.create_model(724            restApiId=rest_api_id,725            name=model_name,726            description=description,727            contentType=content_type,728        )729        # get models after adding730        result = client.get_model(restApiId=rest_api_id, modelName=model_name)731        self.assertEqual(model_name, result["name"])732        self.assertEqual(description, result["description"])733        try:734            client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)735            self.fail("This call should not be successful as the model is not created.")736        except ClientError as e:737            self.assertEqual("NotFoundException", e.response["Error"]["Code"])738            self.assertEqual("Invalid Rest API Id specified", e.response["Error"]["Message"])739    def test_get_model_with_invalid_name(self):740        client = aws_stack.connect_to_service("apigateway")741        response = client.create_rest_api(name="my_api", description="this is my api")742        rest_api_id = response["id"]743        # test with an invalid model name744        try:745            client.get_model(restApiId=rest_api_id, modelName="fake")746            self.fail("This call should not be successful as the model is not created.")747        except ClientError as e:748            self.assertEqual("NotFoundException", e.response["Error"]["Code"])749        # clean up750        client.delete_rest_api(restApiId=rest_api_id)751    def test_put_integration_dynamodb_proxy_validation_without_response_template(self):752        api_id = self.create_api_gateway_and_deploy({})753        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")754        response = requests.put(755            url,756            json.dumps({"id": "id1", "data": "foobar123"}),757        )758        self.assertEqual(404, response.status_code)759    def test_put_integration_dynamodb_proxy_validation_with_response_template(self):760        response_templates = {761            "application/json": json.dumps(762                {763                    "TableName": "MusicCollection",764                    "Item": {"id": "$.Id", "data": "$.data"},765                }766            )767        }768        api_id = self.create_api_gateway_and_deploy(response_templates)769        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")770        response = requests.put(771            url,772            json.dumps({"id": "id1", "data": "foobar123"}),773        )774        self.assertEqual(200, response.status_code)775        dynamo_client = aws_stack.connect_to_resource("dynamodb")776        table = dynamo_client.Table("MusicCollection")777        result = table.get_item(Key={"id": "id1"})778        self.assertEqual("foobar123", result["Item"]["data"])779    def test_api_key_required_for_methods(self):780        response_templates = {781            "application/json": json.dumps(782                {783                    "TableName": "MusicCollection",784                    "Item": {"id": "$.Id", "data": "$.data"},785                }786            )787        }788        api_id = self.create_api_gateway_and_deploy(response_templates, True)789        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")790        payload = {791            "name": "TEST-PLAN-2",792            "description": "Description",793            "quota": {"limit": 10, "period": "DAY", "offset": 0},794            "throttle": {"rateLimit": 2, "burstLimit": 1},795            "apiStages": [{"apiId": api_id, "stage": "staging"}],796            "tags": {"tag_key": "tag_value"},797        }798        client = aws_stack.connect_to_service("apigateway")799        usage_plan_id = client.create_usage_plan(**payload)["id"]800        key_name = "testApiKey"801        key_type = "API_KEY"802        api_key = client.create_api_key(name=key_name)803        payload = {804            "usagePlanId": usage_plan_id,805            "keyId": api_key["id"],806            "keyType": key_type,807        }808        client.create_usage_plan_key(**payload)809        response = requests.put(810            url,811            json.dumps({"id": "id1", "data": "foobar123"}),812        )813        # when the api key is not passed as part of the header814        self.assertEqual(403, response.status_code)815        response = requests.put(816            url,817            json.dumps({"id": "id1", "data": "foobar123"}),818            headers={"X-API-Key": api_key["value"]},819        )820        # when the api key is passed as part of the header821        self.assertEqual(200, response.status_code)822    def test_multiple_api_keys_validate(self):823        response_templates = {824            "application/json": json.dumps(825                {826                    "TableName": "MusicCollection",827                    "Item": {"id": "$.Id", "data": "$.data"},828                }829            )830        }831        api_id = self.create_api_gateway_and_deploy(response_templates, True)832        url = gateway_request_url(api_id=api_id, stage_name="staging", path="/")833        client = aws_stack.connect_to_service("apigateway")834        # Create multiple usage plans835        usage_plan_ids = []836        for i in range(2):837            payload = {838                "name": "APIKEYTEST-PLAN-{}".format(i),839                "description": "Description",840                "quota": {"limit": 10, "period": "DAY", "offset": 0},841                "throttle": {"rateLimit": 2, "burstLimit": 1},842                "apiStages": [{"apiId": api_id, "stage": "staging"}],843                "tags": {"tag_key": "tag_value"},844            }845            usage_plan_ids.append(client.create_usage_plan(**payload)["id"])846        api_keys = []847        key_type = "API_KEY"848        # Create multiple API Keys in each usage plan849        for usage_plan_id in usage_plan_ids:850            for i in range(2):851                api_key = client.create_api_key(name="testMultipleApiKeys{}".format(i))852                payload = {853                    "usagePlanId": usage_plan_id,854                    "keyId": api_key["id"],855                    "keyType": key_type,856                }857                client.create_usage_plan_key(**payload)858                api_keys.append(api_key["value"])859        response = requests.put(860            url,861            json.dumps({"id": "id1", "data": "foobar123"}),862        )863        # when the api key is not passed as part of the header864        self.assertEqual(403, response.status_code)865        # check that all API keys work866        for key in api_keys:867            response = requests.put(868                url,869                json.dumps({"id": "id1", "data": "foobar123"}),870                headers={"X-API-Key": key},871            )872            # when the api key is passed as part of the header873            self.assertEqual(200, response.status_code)874    def test_import_rest_api(self):875        rest_api_name = "restapi-%s" % short_uid()876        client = aws_stack.connect_to_service("apigateway")877        rest_api_id = client.create_rest_api(name=rest_api_name)["id"]878        spec_file = load_file(TEST_SWAGGER_FILE)879        rs = client.put_rest_api(restApiId=rest_api_id, body=spec_file, mode="overwrite")880        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])881        rs = client.get_resources(restApiId=rest_api_id)882        self.assertEqual(883            2, len(rs["items"])884        )  # should contain 2 resources (including the root resource)885        resource = [res for res in rs["items"] if res["path"] == "/test"][0]886        self.assertIn("GET", resource["resourceMethods"])887        # clean up888        client.delete_rest_api(restApiId=rest_api_id)889        spec_file = load_file(TEST_IMPORT_REST_API_FILE)890        rs = client.import_rest_api(body=spec_file)891        self.assertEqual(200, rs["ResponseMetadata"]["HTTPStatusCode"])892        rest_api_id = rs["id"]893        rs = client.get_resources(restApiId=rest_api_id)894        resources = rs["items"]895        self.assertEqual(3, len(resources))896        paths = [res["path"] for res in resources]897        self.assertIn("/", paths)898        self.assertIn("/pets", paths)899        self.assertIn("/pets/{petId}", paths)900        # clean up901        client.delete_rest_api(restApiId=rest_api_id)902    def test_step_function_integrations(self):903        client = aws_stack.connect_to_service("apigateway")904        sfn_client = aws_stack.connect_to_service("stepfunctions")905        lambda_client = aws_stack.connect_to_service("lambda")906        state_machine_name = "test"907        state_machine_def = {908            "Comment": "Hello World example",909            "StartAt": "step1",910            "States": {911                "step1": {"Type": "Task", "Resource": "__tbd__", "End": True},912            },913        }914        # create state machine915        fn_name = "test-stepfunctions-apigw"916        testutil.create_lambda_function(917            handler_file=TEST_LAMBDA_ECHO_FILE,918            func_name=fn_name,919            runtime=LAMBDA_RUNTIME_PYTHON36,920        )921        role_arn = aws_stack.role_arn("sfn_role")922        # create state machine definition923        definition = clone(state_machine_def)924        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)925        definition["States"]["step1"]["Resource"] = lambda_arn_1926        definition = json.dumps(definition)927        # create state machine928        result = sfn_client.create_state_machine(929            name=state_machine_name, definition=definition, roleArn=role_arn930        )931        sm_arn = result["stateMachineArn"]932        # create REST API and method933        rest_api = client.create_rest_api(name="test", description="test")934        resources = client.get_resources(restApiId=rest_api["id"])935        root_resource_id = resources["items"][0]["id"]936        client.put_method(937            restApiId=rest_api["id"],938            resourceId=root_resource_id,939            httpMethod="POST",940            authorizationType="NONE",941        )942        def _prepare_method_integration(943            integr_kwargs={}, resp_templates={}, action="StartExecution", overwrite=False944        ):945            if overwrite:946                client.delete_integration(947                    restApiId=rest_api["id"],948                    resourceId=resources["items"][0]["id"],949                    httpMethod="POST",950                )951            uri = f"arn:aws:apigateway:{aws_stack.get_region()}:states:action/{action}"952            client.put_integration(953                restApiId=rest_api["id"],954                resourceId=root_resource_id,955                httpMethod="POST",956                integrationHttpMethod="POST",957                type="AWS",958                uri=uri,959                **integr_kwargs,960            )961            if resp_templates:962                client.put_integration_response(963                    restApiId=rest_api["id"],964                    resourceId=root_resource_id,965                    selectionPattern="",966                    responseTemplates=resp_templates,967                    httpMethod="POST",968                    statusCode="200",969                )970        # STEP 1: test integration with request template971        _prepare_method_integration(972            integr_kwargs={973                "requestTemplates": {974                    "application/json": """975                    #set($data = $util.escapeJavaScript($input.json('$')))976                    {"input": "$data","stateMachineArn": "%s"}977                    """978                    % sm_arn979                }980            }981        )982        # invoke stepfunction via API GW, assert results983        client.create_deployment(restApiId=rest_api["id"], stageName="dev")984        url = gateway_request_url(api_id=rest_api["id"], stage_name="dev", path="/")985        test_data = {"test": "test-value"}986        resp = requests.post(url, data=json.dumps(test_data))987        self.assertEqual(200, resp.status_code)988        self.assertIn("executionArn", resp.content.decode())989        self.assertIn("startDate", resp.content.decode())990        # STEP 2: test integration without request template991        _prepare_method_integration(overwrite=True)992        test_data_1 = {993            "input": json.dumps(test_data),994            "name": "MyExecution",995            "stateMachineArn": sm_arn,996        }997        # invoke stepfunction via API GW, assert results998        resp = requests.post(url, data=json.dumps(test_data_1))999        self.assertEqual(200, resp.status_code)1000        self.assertIn("executionArn", resp.content.decode())1001        self.assertIn("startDate", resp.content.decode())1002        # STEP 3: test integration with synchronous execution1003        _prepare_method_integration(overwrite=True, action="StartSyncExecution")1004        # invoke stepfunction via API GW, assert results1005        test_data_1["name"] += "1"1006        resp = requests.post(url, data=json.dumps(test_data_1))1007        self.assertEqual(200, resp.status_code)1008        content = json.loads(to_str(resp.content.decode()))1009        self.assertEqual("SUCCEEDED", content.get("status"))1010        self.assertEqual(test_data, json.loads(content.get("output")))1011        # STEP 4: test integration with synchronous execution and response templates1012        resp_templates = {APPLICATION_JSON: "$input.path('$.output')"}1013        _prepare_method_integration(1014            resp_templates=resp_templates, overwrite=True, action="StartSyncExecution"1015        )1016        # invoke stepfunction via API GW, assert results1017        test_data_1["name"] += "2"1018        resp = requests.post(url, data=json.dumps(test_data_1))1019        self.assertEqual(200, resp.status_code)1020        self.assertEqual(test_data, json.loads(to_str(resp.content.decode())))1021        _prepare_method_integration(overwrite=True, action="DeleteStateMachine")1022        # Remove state machine with API GW1023        resp = requests.post(url, data=json.dumps({"stateMachineArn": sm_arn}))1024        self.assertEqual(200, resp.status_code)1025        # Clean up1026        lambda_client.delete_function(FunctionName=fn_name)1027        client.delete_rest_api(restApiId=rest_api["id"])1028    def test_api_gateway_http_integration_with_path_request_parameter(self):1029        client = aws_stack.connect_to_service("apigateway")1030        test_port = get_free_tcp_port()1031        backend_url = "http://localhost:%s/person/{id}" % (test_port)1032        # start test HTTP backend1033        proxy = self.start_http_backend(test_port)1034        # create rest api1035        api_rest = client.create_rest_api(name="test")1036        api_id = api_rest["id"]1037        parent_response = client.get_resources(restApiId=api_id)1038        parent_id = parent_response["items"][0]["id"]1039        resource_1 = client.create_resource(restApiId=api_id, parentId=parent_id, pathPart="person")1040        resource_1_id = resource_1["id"]1041        resource_2 = client.create_resource(1042            restApiId=api_id, parentId=resource_1_id, pathPart="{id}"1043        )1044        resource_2_id = resource_2["id"]1045        client.put_method(1046            restApiId=api_id,1047            resourceId=resource_2_id,1048            httpMethod="GET",1049            authorizationType="NONE",1050            apiKeyRequired=False,1051            requestParameters={"method.request.path.id": True},1052        )1053        client.put_integration(1054            restApiId=api_id,1055            resourceId=resource_2_id,1056            httpMethod="GET",1057            integrationHttpMethod="GET",1058            type="HTTP",1059            uri=backend_url,1060            timeoutInMillis=3000,1061            contentHandling="CONVERT_TO_BINARY",1062            requestParameters={"integration.request.path.id": "method.request.path.id"},1063        )1064        client.create_deployment(restApiId=api_id, stageName="test")1065        def _test_invoke(url):1066            result = requests.get(url)1067            content = json.loads(to_str(result.content))1068            self.assertEqual(200, result.status_code)1069            self.assertRegex(1070                content["headers"].get(HEADER_LOCALSTACK_REQUEST_URL),1071                "http://.*localhost.*/person/123",1072            )1073        for use_hostname in [True, False]:1074            for use_ssl in [True, False] if use_hostname else [False]:1075                url = self._get_invoke_endpoint(1076                    api_id,1077                    stage="test",1078                    path="/person/123",1079                    use_hostname=use_hostname,1080                    use_ssl=use_ssl,1081                )1082                _test_invoke(url)1083        # clean up1084        client.delete_rest_api(restApiId=api_id)1085        proxy.stop()1086    def _get_invoke_endpoint(1087        self, api_id, stage="test", path="/", use_hostname=False, use_ssl=False1088    ):1089        path = path or "/"1090        path = path if path.startswith(path) else f"/{path}"1091        proto = "https" if use_ssl else "http"1092        if use_hostname:1093            return f"{proto}://{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}/{stage}{path}"1094        return (1095            f"{proto}://localhost:{config.EDGE_PORT}/restapis/{api_id}/{stage}/_user_request_{path}"1096        )1097    def test_api_gateway_s3_get_integration(self):1098        apigw_client = aws_stack.connect_to_service("apigateway")1099        s3_client = aws_stack.connect_to_service("s3")1100        bucket_name = "test-bucket"1101        object_name = "test.json"1102        object_content = '{ "success": "true" }'1103        object_content_type = "application/json"1104        api = apigw_client.create_rest_api(name="test")1105        api_id = api["id"]1106        s3_client.create_bucket(Bucket=bucket_name)1107        s3_client.put_object(1108            Bucket=bucket_name,1109            Key=object_name,1110            Body=object_content,1111            ContentType=object_content_type,1112        )1113        self.connect_api_gateway_to_s3(bucket_name, object_name, api_id, "GET")1114        apigw_client.create_deployment(restApiId=api_id, stageName="test")1115        url = gateway_request_url(api_id, "test", "/")1116        result = requests.get(url)1117        self.assertEqual(200, result.status_code)1118        self.assertEqual(object_content, result.text)1119        self.assertEqual(object_content_type, result.headers["content-type"])1120        # clean up1121        apigw_client.delete_rest_api(restApiId=api_id)1122        s3_client.delete_object(Bucket=bucket_name, Key=object_name)1123        s3_client.delete_bucket(Bucket=bucket_name)1124    def test_api_mock_integration_response_params(self):1125        # apigw_client = aws_stack.connect_to_service('apigateway')1126        resps = [1127            {1128                "statusCode": "204",1129                "httpMethod": "OPTIONS",1130                "responseParameters": {1131                    "method.response.header.Access-Control-Allow-Methods": "'POST,OPTIONS'",1132                    "method.response.header.Vary": "'Origin'",1133                },1134            }1135        ]1136        api_id = self.create_api_gateway_and_deploy(1137            integration_type="MOCK", integration_responses=resps1138        )1139        url = gateway_request_url(api_id=api_id, stage_name=self.TEST_STAGE_NAME, path="/")1140        result = requests.options(url)1141        self.assertLess(result.status_code, 400)1142        self.assertEqual("Origin", result.headers.get("vary"))1143        self.assertEqual("POST,OPTIONS", result.headers.get("Access-Control-Allow-Methods"))1144    # =====================================================================1145    # Helper methods1146    # =====================================================================1147    def connect_api_gateway_to_s3(self, bucket_name, file_name, api_id, method):1148        """Connects the root resource of an api gateway to the given object of an s3 bucket."""1149        apigw_client = aws_stack.connect_to_service("apigateway")1150        s3_uri = "arn:aws:apigateway:{}:s3:path/{}/{}".format(1151            aws_stack.get_region(), bucket_name, file_name1152        )1153        test_role = "test-s3-role"1154        role_arn = aws_stack.role_arn(role_name=test_role)1155        resources = apigw_client.get_resources(restApiId=api_id)1156        # using the root resource '/' directly for this test1157        root_resource_id = resources["items"][0]["id"]1158        apigw_client.put_method(1159            restApiId=api_id,1160            resourceId=root_resource_id,1161            httpMethod=method,1162            authorizationType="NONE",1163            apiKeyRequired=False,1164            requestParameters={},1165        )1166        apigw_client.put_integration(1167            restApiId=api_id,1168            resourceId=root_resource_id,1169            httpMethod=method,1170            type="AWS",1171            integrationHttpMethod=method,1172            uri=s3_uri,1173            credentials=role_arn,1174        )1175    def connect_api_gateway_to_kinesis(self, gateway_name, kinesis_stream):1176        resources = {}1177        template = self.APIGATEWAY_DATA_INBOUND_TEMPLATE % kinesis_stream1178        resource_path = self.API_PATH_DATA_INBOUND.replace("/", "")1179        resources[resource_path] = [1180            {1181                "httpMethod": "POST",1182                "authorizationType": "NONE",1183                "integrations": [1184                    {1185                        "type": "AWS",1186                        "uri": "arn:aws:apigateway:%s:kinesis:action/PutRecords"1187                        % aws_stack.get_region(),1188                        "requestTemplates": {"application/json": template},1189                    }1190                ],1191            },1192            {1193                "httpMethod": "GET",1194                "authorizationType": "NONE",1195                "integrations": [1196                    {1197                        "type": "AWS",1198                        "uri": "arn:aws:apigateway:%s:kinesis:action/ListStreams"1199                        % aws_stack.get_region(),1200                        "requestTemplates": {"application/json": "{}"},1201                    }1202                ],1203            },1204        ]1205        return aws_stack.create_api_gateway(1206            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1207        )1208    def connect_api_gateway_to_http(1209        self, int_type, gateway_name, target_url, methods=[], path=None1210    ):1211        if not methods:1212            methods = ["GET", "POST"]1213        if not path:1214            path = "/"1215        resources = {}1216        resource_path = path.replace("/", "")1217        resources[resource_path] = []1218        req_templates = (1219            {"application/json": json.dumps({"foo": "bar"})} if int_type == "custom" else {}1220        )1221        for method in methods:1222            resources[resource_path].append(1223                {1224                    "httpMethod": method,1225                    "integrations": [1226                        {1227                            "type": "HTTP" if int_type == "custom" else "HTTP_PROXY",1228                            "uri": target_url,1229                            "requestTemplates": req_templates,1230                            "responseTemplates": {},1231                        }1232                    ],1233                }1234            )1235        return aws_stack.create_api_gateway(1236            name=gateway_name, resources=resources, stage_name=self.TEST_STAGE_NAME1237        )1238    @staticmethod1239    def create_lambda_function(fn_name):1240        testutil.create_lambda_function(1241            handler_file=TEST_LAMBDA_PYTHON, libs=TEST_LAMBDA_LIBS, func_name=fn_name1242        )1243    def test_apigw_test_invoke_method_api(self):1244        client = aws_stack.connect_to_service("apigateway")1245        lambda_client = aws_stack.connect_to_service("lambda")1246        # create test Lambda1247        fn_name = f"test-{short_uid()}"1248        testutil.create_lambda_function(1249            handler_file=TEST_LAMBDA_HANDLER_JS, func_name=fn_name, runtime=LAMBDA_RUNTIME_NODEJS12X1250        )1251        lambda_arn_1 = aws_stack.lambda_function_arn(fn_name)1252        # create REST API and test resource1253        rest_api = client.create_rest_api(name="test", description="test")1254        root_resource = client.get_resources(restApiId=rest_api["id"])1255        resource = client.create_resource(1256            restApiId=rest_api["id"], parentId=root_resource["items"][0]["id"], pathPart="foo"1257        )1258        # create method and integration1259        client.put_method(1260            restApiId=rest_api["id"],1261            resourceId=resource["id"],1262            httpMethod="GET",1263            authorizationType="NONE",1264        )1265        client.put_integration(1266            restApiId=rest_api["id"],1267            resourceId=resource["id"],1268            httpMethod="GET",1269            integrationHttpMethod="GET",1270            type="AWS",1271            uri="arn:aws:apigateway:{}:lambda:path//2015-03-31/functions/{}/invocations".format(1272                aws_stack.get_region(), lambda_arn_11273            ),1274        )1275        # run test_invoke_method API #11276        response = client.test_invoke_method(1277            restApiId=rest_api["id"],1278            resourceId=resource["id"],1279            httpMethod="GET",1280            pathWithQueryString="/foo",1281        )1282        self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])1283        self.assertEqual(200, response.get("status"))1284        self.assertIn("response from", response.get("body"))1285        # run test_invoke_method API #21286        response = client.test_invoke_method(1287            restApiId=rest_api["id"],1288            resourceId=resource["id"],1289            httpMethod="GET",1290            pathWithQueryString="/foo",1291            body='{"test": "val123"}',1292            headers={"content-type": "application/json"},1293        )1294        self.assertEqual(200, response["ResponseMetadata"]["HTTPStatusCode"])1295        self.assertEqual(200, response.get("status"))1296        self.assertIn("response from", response.get("body"))1297        self.assertIn("val123", response.get("body"))1298        # Clean up1299        lambda_client.delete_function(FunctionName=fn_name)1300        client.delete_rest_api(restApiId=rest_api["id"])1301    @staticmethod1302    def start_http_backend(test_port):1303        # test listener for target HTTP backend1304        class TestListener(ProxyListener):1305            def forward_request(self, **kwargs):1306                response = Response()1307                response.status_code = 2001308                result = {1309                    "data": kwargs.get("data") or "{}",1310                    "headers": dict(kwargs.get("headers")),1311                }1312                response._content = json.dumps(json_safe(result))1313                return response1314        proxy = start_proxy(test_port, update_listener=TestListener())1315        return proxy1316    @staticmethod...
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.