Unlock 30% off on Manual Testing Annual Plans this Holiday Season.
Copied to Clipboard!
How to use is_test_invoke_method method in localstack

Best Python code snippet using localstack_python
apigateway_listener.py
Source:apigateway_listener.py
...174        if re.match(PATH_REGEX_VALIDATORS, path):175            return handle_validators(method, path, data, headers)176        if re.match(PATH_REGEX_RESPONSES, path):177            return handle_gateway_responses(method, path, data, headers)178        if is_test_invoke_method(method, path):179            # if call is from test_invoke_api then use http_method to find the integration,180            #   as test_invoke_api makes a POST call to request the test invocation181            match = re.match(PATH_REGEX_TEST_INVOKE_API, path)182            invocation_context.method = match[3]183            if data:184                orig_data = data185                path_with_query_string = orig_data.get("pathWithQueryString", None)186                if path_with_query_string:187                    invocation_context.path_with_query_string = path_with_query_string188                invocation_context.data = data.get("body")189                invocation_context.headers = orig_data.get("headers", {})190            result = invoke_rest_api_from_request(invocation_context)191            result = {192                "status": result.status_code,193                "body": to_str(result.content),194                "headers": dict(result.headers),195            }196            return result197        return True198    def return_response(self, method, path, data, headers, response):199        # fix backend issue (missing support for API documentation)200        if re.match(r"/restapis/[^/]+/documentation/versions", path):201            if response.status_code == 404:202                return requests_response({"position": "1", "items": []})203        # add missing implementations204        if response.status_code == 404:205            data = data and json.loads(to_str(data))206            result = None207            if path == "/account":208                result = handle_accounts(method, path, data, headers)209            elif path.startswith("/vpclinks"):210                result = handle_vpc_links(method, path, data, headers)211            elif re.match(PATH_REGEX_PATH_MAPPINGS, path):212                result = handle_base_path_mappings(method, path, data, headers)213            elif re.match(PATH_REGEX_CLIENT_CERTS, path):214                result = handle_client_certificates(method, path, data, headers)215            if result is not None:216                response.status_code = 200217                aws_responses.set_response_content(response, result, getattr(result, "headers", {}))218        # keep track of API regions for faster lookup later on219        if method == "POST" and path == "/restapis":220            content = json.loads(to_str(response.content))221            api_id = content["id"]222            region = aws_stack.extract_region_from_auth_header(headers)223            API_REGIONS[api_id] = region224        # publish event225        if method == "POST" and path == "/restapis":226            content = json.loads(to_str(response.content))227            event_publisher.fire_event(228                event_publisher.EVENT_APIGW_CREATE_API,229                payload={"a": event_publisher.get_hash(content["id"])},230            )231        api_regex = r"^/restapis/([a-zA-Z0-9\-]+)$"232        if method == "DELETE" and re.match(api_regex, path):233            api_id = re.sub(api_regex, r"\1", path)234            event_publisher.fire_event(235                event_publisher.EVENT_APIGW_DELETE_API,236                payload={"a": event_publisher.get_hash(api_id)},237            )238# ------------239# API METHODS240# ------------241def run_authorizer(invocation_context: ApiInvocationContext, authorizer: Dict):242    # TODO implement authorizers243    pass244def authorize_invocation(invocation_context: ApiInvocationContext):245    client = aws_stack.connect_to_service("apigateway")246    authorizers = client.get_authorizers(restApiId=invocation_context.api_id, limit=100).get(247        "items", []248    )249    for authorizer in authorizers:250        run_authorizer(invocation_context, authorizer)251def validate_api_key(api_key: str, stage: str):252    usage_plan_ids = []253    client = aws_stack.connect_to_service("apigateway")254    usage_plans = client.get_usage_plans()255    for item in usage_plans.get("items", []):256        api_stages = item.get("apiStages", [])257        for api_stage in api_stages:258            if api_stage.get("stage") == stage:259                usage_plan_ids.append(item.get("id"))260    for usage_plan_id in usage_plan_ids:261        usage_plan_keys = client.get_usage_plan_keys(usagePlanId=usage_plan_id)262        for key in usage_plan_keys.get("items", []):263            if key.get("value") == api_key:264                return True265    return False266def is_api_key_valid(is_api_key_required: bool, headers: Dict[str, str], stage: str):267    if not is_api_key_required:268        return True269    api_key = headers.get("X-API-Key")270    if not api_key:271        return False272    return validate_api_key(api_key, stage)273def update_content_length(response: Response):274    if response and response.content is not None:275        response.headers["Content-Length"] = str(len(response.content))276def apply_request_parameter(uri: str, integration: Dict[str, Any], path_params: Dict[str, str]):277    request_parameters = integration.get("requestParameters", None)278    uri = uri or integration.get("uri") or integration.get("integrationUri") or ""279    if request_parameters:280        for key in path_params:281            # check if path_params is present in the integration request parameters282            request_param_key = f"integration.request.path.{key}"283            request_param_value = f"method.request.path.{key}"284            if request_parameters.get(request_param_key, None) == request_param_value:285                uri = uri.replace(f"{{{key}}}", path_params[key])286    return uri287def apply_template(288    integration: Dict[str, Any],289    req_res_type: str,290    data: InvocationPayload,291    path_params={},292    query_params={},293    headers={},294    context={},295):296    integration_type = integration.get("type") or integration.get("integrationType")297    if integration_type in ["HTTP", "AWS"]:298        # apply custom request template299        content_type = APPLICATION_JSON  # TODO: make configurable!300        template = integration.get("%sTemplates" % req_res_type, {}).get(content_type)301        if template:302            variables = {"context": context or {}}303            input_ctx = {"body": data}304            def _params(name=None):305                # See https://docs.aws.amazon.com/apigateway/latest/developerguide/306                #    api-gateway-mapping-template-reference.html#input-variable-reference307                # Returns "request parameter from the path, query string, or header value (searched in that order)"308                combined = {}309                combined.update(path_params or {})310                combined.update(query_params or {})311                combined.update(headers or {})312                return combined if not name else combined.get(name)313            input_ctx["params"] = _params314            data = aws_stack.render_velocity_template(template, input_ctx, variables=variables)315    return data316def apply_response_parameters(invocation_context: ApiInvocationContext):317    response = invocation_context.response318    integration = invocation_context.integration319    int_responses = integration.get("integrationResponses") or {}320    if not int_responses:321        return response322    entries = list(int_responses.keys())323    return_code = str(response.status_code)324    if return_code not in entries:325        if len(entries) > 1:326            LOG.info("Found multiple integration response status codes: %s" % entries)327            return response328        return_code = entries[0]329    response_params = int_responses[return_code].get("responseParameters", {})330    for key, value in response_params.items():331        # TODO: add support for method.response.body, etc ...332        if str(key).lower().startswith("method.response.header."):333            header_name = key[len("method.response.header.") :]334            response.headers[header_name] = value.strip("'")335    return response336def get_api_id_stage_invocation_path(337    invocation_context: ApiInvocationContext,338) -> Tuple[str, str, str]:339    path = invocation_context.path340    headers = invocation_context.headers341    path_match = re.search(PATH_REGEX_USER_REQUEST, path)342    host_header = headers.get(HEADER_LOCALSTACK_EDGE_URL, "") or headers.get("Host") or ""343    host_match = re.search(HOST_REGEX_EXECUTE_API, host_header)344    test_invoke_match = re.search(PATH_REGEX_TEST_INVOKE_API, path)345    if path_match:346        api_id = path_match.group(1)347        stage = path_match.group(2)348        relative_path_w_query_params = "/%s" % path_match.group(3)349    elif host_match:350        api_id = extract_api_id_from_hostname_in_url(host_header)351        stage = path.strip("/").split("/")[0]352        relative_path_w_query_params = "/%s" % path.lstrip("/").partition("/")[2]353    elif test_invoke_match:354        # special case: fetch the resource details for TestInvokeApi invocations355        stage = None356        region_name = invocation_context.region_name357        api_id = test_invoke_match.group(1)358        resource_id = test_invoke_match.group(2)359        query_string = test_invoke_match.group(4) or ""360        apigateway = aws_stack.connect_to_service(361            service_name="apigateway", region_name=region_name362        )363        resource = apigateway.get_resource(restApiId=api_id, resourceId=resource_id)364        resource_path = resource.get("path")365        relative_path_w_query_params = f"{resource_path}{query_string}"366    else:367        raise Exception(f"Unable to extract API Gateway details from request: {path} {headers}")368    if api_id:369        # set current region in request thread local, to ensure aws_stack.get_region() works properly370        if getattr(THREAD_LOCAL, "request_context", None) is not None:371            THREAD_LOCAL.request_context.headers[MARKER_APIGW_REQUEST_REGION] = API_REGIONS.get(372                api_id, ""373            )374    return api_id, stage, relative_path_w_query_params375def extract_api_id_from_hostname_in_url(hostname: str) -> str:376    """Extract API ID 'id123' from URLs like https://id123.execute-api.localhost.localstack.cloud:4566"""377    match = re.match(HOST_REGEX_EXECUTE_API, hostname)378    api_id = match.group(1)379    return api_id380def invoke_rest_api_from_request(invocation_context: ApiInvocationContext):381    api_id, stage, relative_path_w_query_params = get_api_id_stage_invocation_path(382        invocation_context383    )384    invocation_context.api_id = api_id385    invocation_context.stage = stage386    invocation_context.path_with_query_string = relative_path_w_query_params387    try:388        return invoke_rest_api(invocation_context)389    except AuthorizationError as e:390        return make_error_response("Not authorized to invoke REST API %s: %s" % (api_id, e), 403)391def invoke_rest_api(invocation_context: ApiInvocationContext):392    invocation_path = invocation_context.path_with_query_string393    raw_path = invocation_context.path or invocation_path394    method = invocation_context.method395    headers = invocation_context.headers396    relative_path, query_string_params = extract_query_string_params(path=invocation_path)397    # run gateway authorizers for this request398    authorize_invocation(invocation_context)399    path_map = helpers.get_rest_api_paths(rest_api_id=invocation_context.api_id)400    try:401        extracted_path, resource = get_resource_for_path(path=relative_path, path_map=path_map)402    except Exception:403        return make_error_response("Unable to find path %s" % raw_path, 404)404    api_key_required = resource.get("resourceMethods", {}).get(method, {}).get("apiKeyRequired")405    if not is_api_key_valid(api_key_required, headers, invocation_context.stage):406        return make_error_response("Access denied - invalid API key", 403)407    integrations = resource.get("resourceMethods", {})408    integration = integrations.get(method, {})409    if not integration:410        integration = integrations.get("ANY", {})411    integration = integration.get("methodIntegration")412    if not integration:413        if method == "OPTIONS" and "Origin" in headers:414            # default to returning CORS headers if this is an OPTIONS request415            return get_cors_response(headers)416        return make_error_response("Unable to find integration for path %s" % raw_path, 404)417    res_methods = path_map.get(relative_path, {}).get("resourceMethods", {})418    meth_integration = res_methods.get(method, {}).get("methodIntegration", {})419    int_responses = meth_integration.get("integrationResponses", {})420    response_templates = int_responses.get("200", {}).get("responseTemplates", {})421    # update fields in invocation context, then forward request to next handler422    invocation_context.resource = resource423    invocation_context.resource_path = extracted_path424    invocation_context.response_templates = response_templates425    invocation_context.integration = integration426    result = invoke_rest_api_integration(invocation_context)427    return result428def invoke_rest_api_integration(invocation_context: ApiInvocationContext):429    try:430        response = invoke_rest_api_integration_backend(431            invocation_context, invocation_context.integration432        )433        invocation_context.response = response434        response = apply_response_parameters(invocation_context)435        return response436    except Exception as e:437        msg = f"Error invoking integration for API Gateway ID '{invocation_context.api_id}': {e}"438        LOG.exception(msg)439        return make_error_response(msg, 400)440def invoke_rest_api_integration_backend(441    invocation_context: ApiInvocationContext, integration: Dict442):443    # define local aliases from invocation context444    invocation_path = invocation_context.path_with_query_string445    method = invocation_context.method446    path = invocation_context.path447    data = invocation_context.data448    headers = invocation_context.headers449    api_id = invocation_context.api_id450    stage = invocation_context.stage451    resource_path = invocation_context.resource_path452    response_templates = invocation_context.response_templates453    # extract integration type and path parameters454    relative_path, query_string_params = extract_query_string_params(path=invocation_path)455    integration_type_orig = integration.get("type") or integration.get("integrationType") or ""456    integration_type = integration_type_orig.upper()457    uri = integration.get("uri") or integration.get("integrationUri") or ""458    try:459        path_params = extract_path_params(path=relative_path, extracted_path=resource_path)460    except Exception:461        path_params = {}462    if (uri.startswith("arn:aws:apigateway:") and ":lambda:path" in uri) or uri.startswith(463        "arn:aws:lambda"464    ):465        if integration_type in ["AWS", "AWS_PROXY"]:466            func_arn = uri467            if ":lambda:path" in uri:468                func_arn = (469                    uri.split(":lambda:path")[1].split("functions/")[1].split("/invocations")[0]470                )471            # apply custom request template472            data_str = data473            is_base64_encoded = False474            try:475                data_str = json.dumps(data) if isinstance(data, (dict, list)) else to_str(data)476                data_str = apply_template(477                    integration,478                    "request",479                    data_str,480                    path_params=path_params,481                    query_params=query_string_params,482                    headers=headers,483                )484            except UnicodeDecodeError:485                data_str = base64.b64encode(data_str)486                is_base64_encoded = True487            except Exception as e:488                LOG.warning("Unable to convert API Gateway payload to str: %s" % (e))489                pass490            # Sample request context:491            # https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-create-api-as-simple-proxy-for-lambda.html#api-gateway-create-api-as-simple-proxy-for-lambda-test492            request_context = get_lambda_event_request_context(invocation_context)493            stage_variables = (494                get_stage_variables(api_id, stage)495                if not is_test_invoke_method(method, path)496                else None497            )498            # TODO: change this signature to InvocationContext as well!499            result = lambda_api.process_apigateway_invocation(500                func_arn,501                relative_path,502                data_str,503                stage,504                api_id,505                headers,506                is_base64_encoded=is_base64_encoded,507                path_params=path_params,508                query_string_params=query_string_params,509                method=method,510                resource_path=resource_path,511                request_context=request_context,512                event_context=invocation_context.context,513                stage_variables=stage_variables,514            )515            if isinstance(result, FlaskResponse):516                response = flask_to_requests_response(result)517            elif isinstance(result, Response):518                response = result519            else:520                response = LambdaResponse()521                parsed_result = (522                    result if isinstance(result, dict) else json.loads(str(result or "{}"))523                )524                parsed_result = common.json_safe(parsed_result)525                parsed_result = {} if parsed_result is None else parsed_result526                response.status_code = int(parsed_result.get("statusCode", 200))527                parsed_headers = parsed_result.get("headers", {})528                if parsed_headers is not None:529                    response.headers.update(parsed_headers)530                try:531                    result_body = parsed_result.get("body")532                    if isinstance(result_body, dict):533                        response._content = json.dumps(result_body)534                    else:535                        body_bytes = to_bytes(to_str(result_body or ""))536                        if parsed_result.get("isBase64Encoded", False):537                            body_bytes = base64.b64decode(body_bytes)538                        response._content = body_bytes539                except Exception as e:540                    LOG.warning("Couldn't set Lambda response content: %s" % e)541                    response._content = "{}"542                update_content_length(response)543                response.multi_value_headers = parsed_result.get("multiValueHeaders") or {}544            # apply custom response template545            response._content = apply_template(integration, "response", response._content)546            response.headers["Content-Length"] = str(len(response.content or ""))547            return response548        raise Exception(549            'API Gateway integration type "%s", action "%s", method "%s" invalid or not yet implemented'550            % (integration_type, uri, method)551        )552    elif integration_type == "AWS":553        if "kinesis:action/" in uri:554            if uri.endswith("kinesis:action/PutRecord"):555                target = kinesis_listener.ACTION_PUT_RECORD556            elif uri.endswith("kinesis:action/PutRecords"):557                target = kinesis_listener.ACTION_PUT_RECORDS558            elif uri.endswith("kinesis:action/ListStreams"):559                target = kinesis_listener.ACTION_LIST_STREAMS560            else:561                LOG.info(562                    "Unexpected API Gateway integration URI '%s' for integration type %s",563                    uri,564                    integration_type,565                )566                target = ""567            # apply request templates568            new_data = apply_request_response_templates(569                data, integration.get("requestTemplates"), content_type=APPLICATION_JSON570            )571            # forward records to target kinesis stream572            headers = aws_stack.mock_aws_request_headers(service="kinesis")573            headers["X-Amz-Target"] = target574            result = common.make_http_request(575                url=config.TEST_KINESIS_URL, method="POST", data=new_data, headers=headers576            )577            # apply response template578            result = apply_request_response_templates(579                result, response_templates, content_type=APPLICATION_JSON580            )581            return result582        elif "states:action/" in uri:583            action = uri.split("/")[-1]584            payload = {}585            if APPLICATION_JSON in integration.get("requestTemplates", {}):586                payload = apply_request_response_templates(587                    data,588                    integration.get("requestTemplates"),589                    content_type=APPLICATION_JSON,590                    as_json=True,591                )592            else:593                payload = json.loads(data.decode("utf-8"))594            client = aws_stack.connect_to_service("stepfunctions")595            # Hot fix since step functions local package responses: Unsupported Operation: 'StartSyncExecution'596            method_name = (597                camel_to_snake_case(action) if action != "StartSyncExecution" else "start_execution"598            )599            try:600                method = getattr(client, method_name)601            except AttributeError:602                msg = "Invalid step function action: %s" % method_name603                LOG.error(msg)604                return make_error_response(msg, 400)605            result = method(606                **payload,607            )608            result = json_safe({k: result[k] for k in result if k not in "ResponseMetadata"})609            response = requests_response(610                content=result,611                headers=aws_stack.mock_aws_request_headers(),612            )613            if action == "StartSyncExecution":614                # poll for the execution result and return it615                result = await_sfn_execution_result(result["executionArn"])616                result_status = result.get("status")617                if result_status != "SUCCEEDED":618                    return make_error_response(619                        "StepFunctions execution %s failed with status '%s'"620                        % (result["executionArn"], result_status),621                        500,622                    )623                result = json_safe(result)624                response = requests_response(content=result)625            # apply response templates626            response = apply_request_response_templates(627                response, response_templates, content_type=APPLICATION_JSON628            )629            return response630        elif "s3:path/" in uri and method == "GET":631            s3 = aws_stack.connect_to_service("s3")632            uri_match = re.match(TARGET_REGEX_S3_URI, uri)633            if uri_match:634                bucket, object_key = uri_match.group("bucket", "object")635                LOG.debug("Getting request for bucket %s object %s", bucket, object_key)636                try:637                    object = s3.get_object(Bucket=bucket, Key=object_key)638                except s3.exceptions.NoSuchKey:639                    msg = "Object %s not found" % object_key640                    LOG.debug(msg)641                    return make_error_response(msg, 404)642                headers = aws_stack.mock_aws_request_headers(service="s3")643                if object.get("ContentType"):644                    headers["Content-Type"] = object["ContentType"]645                # stream used so large files do not fill memory646                response = request_response_stream(stream=object["Body"], headers=headers)647                return response648            else:649                msg = "Request URI does not match s3 specifications"650                LOG.warning(msg)651                return make_error_response(msg, 400)652        if method == "POST":653            if uri.startswith("arn:aws:apigateway:") and ":sqs:path" in uri:654                template = integration["requestTemplates"][APPLICATION_JSON]655                account_id, queue = uri.split("/")[-2:]656                region_name = uri.split(":")[3]657                new_request = "%s&QueueName=%s" % (658                    aws_stack.render_velocity_template(template, data),659                    queue,660                )661                headers = aws_stack.mock_aws_request_headers(service="sqs", region_name=region_name)662                url = urljoin(config.TEST_SQS_URL, "%s/%s" % (TEST_AWS_ACCOUNT_ID, queue))663                result = common.make_http_request(664                    url, method="POST", headers=headers, data=new_request665                )666                return result667        raise Exception(668            'API Gateway AWS integration action URI "%s", method "%s" not yet implemented'669            % (uri, method)670        )671    elif integration_type == "AWS_PROXY":672        if uri.startswith("arn:aws:apigateway:") and ":dynamodb:action" in uri:673            # arn:aws:apigateway:us-east-1:dynamodb:action/PutItem&Table=MusicCollection674            table_name = uri.split(":dynamodb:action")[1].split("&Table=")[1]675            action = uri.split(":dynamodb:action")[1].split("&Table=")[0]676            if "PutItem" in action and method == "PUT":677                response_template = response_templates.get("application/json")678                if response_template is None:679                    msg = "Invalid response template defined in integration response."680                    LOG.info("%s Existing: %s" % (msg, response_templates))681                    return make_error_response(msg, 404)682                response_template = json.loads(response_template)683                if response_template["TableName"] != table_name:684                    msg = "Invalid table name specified in integration response template."685                    return make_error_response(msg, 404)686                dynamo_client = aws_stack.connect_to_resource("dynamodb")687                table = dynamo_client.Table(table_name)688                event_data = {}689                data_dict = json.loads(data)690                for key, _ in response_template["Item"].items():691                    event_data[key] = data_dict[key]692                table.put_item(Item=event_data)693                response = requests_response(event_data)694                return response695        else:696            raise Exception(697                'API Gateway action uri "%s", integration type %s not yet implemented'698                % (uri, integration_type)699            )700    elif integration_type in ["HTTP_PROXY", "HTTP"]:701        if ":servicediscovery:" in uri:702            # check if this is a servicediscovery integration URI703            client = aws_stack.connect_to_service("servicediscovery")704            service_id = uri.split("/")[-1]705            instances = client.list_instances(ServiceId=service_id)["Instances"]706            instance = (instances or [None])[0]707            if instance and instance.get("Id"):708                uri = "http://%s/%s" % (instance["Id"], invocation_path.lstrip("/"))709        # apply custom request template710        data = apply_template(integration, "request", data)711        if isinstance(data, dict):712            data = json.dumps(data)713        uri = apply_request_parameter(uri, integration=integration, path_params=path_params)714        result = requests.request(method=method, url=uri, data=data, headers=headers)715        # apply custom response template716        result = apply_template(integration, "response", result)717        return result718    elif integration_type == "MOCK":719        # return empty response - details filled in via responseParameters above...720        return requests_response({})721    if method == "OPTIONS":722        # fall back to returning CORS headers if this is an OPTIONS request723        return get_cors_response(headers)724    raise Exception(725        'API Gateway integration type "%s", method "%s", URI "%s" not yet implemented'726        % (integration_type, method, uri)727    )728def get_stage_variables(api_id: str, stage: str) -> Dict[str, str]:729    if not stage:730        return731    region_name = [name for name, region in apigateway_backends.items() if api_id in region.apis][0]732    api_gateway_client = aws_stack.connect_to_service("apigateway", region_name=region_name)733    response = api_gateway_client.get_stage(restApiId=api_id, stageName=stage)734    return response.get("variables")735def get_lambda_event_request_context(736    invocation_context: ApiInvocationContext,737):738    method = invocation_context.method739    path = invocation_context.path740    headers = invocation_context.headers741    integration_uri = invocation_context.integration_uri742    resource_path = invocation_context.resource_path743    resource_id = invocation_context.resource_id744    auth_context = invocation_context.auth_context745    api_id, stage, relative_path_w_query_params = get_api_id_stage_invocation_path(746        invocation_context747    )748    relative_path, query_string_params = extract_query_string_params(749        path=relative_path_w_query_params750    )751    source_ip = headers.get("X-Forwarded-For", ",").split(",")[-2].strip()752    integration_uri = integration_uri or ""753    account_id = integration_uri.split(":lambda:path")[-1].split(":function:")[0].split(":")[-1]754    account_id = account_id or TEST_AWS_ACCOUNT_ID755    domain_name = f"{api_id}.execute-api.{LOCALHOST_HOSTNAME}"756    request_context = {757        "resourcePath": resource_path or relative_path,758        "apiId": api_id,759        "domainPrefix": api_id,760        "domainName": domain_name,761        "accountId": account_id,762        "resourceId": resource_id,763        "requestId": long_uid(),764        "identity": {765            "accountId": account_id,766            "sourceIp": source_ip,767            "userAgent": headers.get("User-Agent"),768        },769        "httpMethod": method,770        "protocol": "HTTP/1.1",771        "requestTime": datetime.datetime.utcnow(),772        "requestTimeEpoch": int(time.time() * 1000),773    }774    if auth_context:775        request_context["authorizer"] = auth_context776    if not is_test_invoke_method(method, path):777        request_context["path"] = (f"/{stage}" if stage else "") + relative_path778        request_context["stage"] = stage779    return request_context780def apply_request_response_templates(781    data: Union[Response, bytes],782    templates: Dict[str, str],783    content_type: str = None,784    as_json: bool = False,785):786    """Apply the matching request/response template (if it exists) to the payload data and return the result"""787    content_type = content_type or APPLICATION_JSON788    is_response = isinstance(data, Response)789    templates = templates or {}790    template = templates.get(content_type)791    if not template:792        return data793    content = (data.content if is_response else data) or ""794    result = aws_stack.render_velocity_template(template, content, as_json=as_json)795    if is_response:796        data._content = result797        update_content_length(data)798        return data799    return result800def is_test_invoke_method(method, path):801    return method == "POST" and bool(re.match(PATH_REGEX_TEST_INVOKE_API, path))802# instantiate listener...
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.