Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use test_cfn_handle_serverless_api_resource method in localstack

Best Python code snippet using localstack_python
helpers.py
Source:helpers.py
1import contextlib2import datetime3import json4import logging5import re6import time7from typing import Any, Dict, List, Optional, Tuple, Union8from urllib import parse as urlparse9import pytz10from botocore.utils import InvalidArnException11from jsonpatch import apply_patch12from jsonpointer import JsonPointerException13from moto.apigateway import models as apigateway_models14from moto.apigateway.utils import create_id as create_resource_id15from requests.models import Response16from localstack import config17from localstack.constants import (18    APPLICATION_JSON,19    HEADER_LOCALSTACK_EDGE_URL,20    LOCALHOST_HOSTNAME,21    PATH_USER_REQUEST,22    TEST_AWS_ACCOUNT_ID,23)24from localstack.services.apigateway.context import ApiInvocationContext25from localstack.services.generic_proxy import RegionBackend26from localstack.utils import common27from localstack.utils.aws import aws_stack28from localstack.utils.aws.aws_responses import requests_error_response_json, requests_response29from localstack.utils.aws.aws_stack import parse_arn30from localstack.utils.aws.request_context import MARKER_APIGW_REQUEST_REGION, THREAD_LOCAL31from localstack.utils.strings import long_uid32LOG = logging.getLogger(__name__)33REQUEST_TIME_DATE_FORMAT = "%d/%b/%Y:%H:%M:%S %z"34# regex path pattern for user requests, handles stages like $default35PATH_REGEX_USER_REQUEST = (36    r"^/restapis/([A-Za-z0-9_\\-]+)(?:/([A-Za-z0-9\_($|%%24)\\-]+))?/%s/(.*)$" % PATH_USER_REQUEST37)38# URL pattern for invocations39HOST_REGEX_EXECUTE_API = r"(?:.*://)?([a-zA-Z0-9-]+)\.execute-api\.(localhost.localstack.cloud|([^\.]+)\.amazonaws\.com)(.*)"40# regex path patterns41PATH_REGEX_MAIN = r"^/restapis/([A-Za-z0-9_\-]+)/[a-z]+(\?.*)?"42PATH_REGEX_SUB = r"^/restapis/([A-Za-z0-9_\-]+)/[a-z]+/([A-Za-z0-9_\-]+)/.*"43# path regex patterns44PATH_REGEX_AUTHORIZERS = r"^/restapis/([A-Za-z0-9_\-]+)/authorizers/?([^?/]+)?(\?.*)?"45PATH_REGEX_VALIDATORS = r"^/restapis/([A-Za-z0-9_\-]+)/requestvalidators/?([^?/]+)?(\?.*)?"46PATH_REGEX_RESPONSES = r"^/restapis/([A-Za-z0-9_\-]+)/gatewayresponses(/[A-Za-z0-9_\-]+)?(\?.*)?"47PATH_REGEX_DOC_PARTS = r"^/restapis/([A-Za-z0-9_\-]+)/documentation/parts/?([^?/]+)?(\?.*)?"48PATH_REGEX_PATH_MAPPINGS = r"/domainnames/([^/]+)/basepathmappings/?(.*)"49PATH_REGEX_CLIENT_CERTS = r"/clientcertificates/?([^/]+)?$"50PATH_REGEX_VPC_LINKS = r"/vpclinks/([^/]+)?(.*)"51PATH_REGEX_TEST_INVOKE_API = r"^\/restapis\/([A-Za-z0-9_\-]+)\/resources\/([A-Za-z0-9_\-]+)\/methods\/([A-Za-z0-9_\-]+)/?(\?.*)?"52# template for SQS inbound data53APIGATEWAY_SQS_DATA_INBOUND_TEMPLATE = (54    "Action=SendMessage&MessageBody=$util.base64Encode($input.json('$'))"55)56# special tag name to allow specifying a custom ID for new REST APIs57TAG_KEY_CUSTOM_ID = "_custom_id_"58# map API IDs to region names - TODO remove and replace with in-memory lookup59API_REGIONS = {}60# TODO: make the CRUD operations in this file generic for the different model types (authorizes, validators, ...)61class APIGatewayRegion(RegionBackend):62    # TODO: introduce a RestAPI class to encapsulate the variables below63    # maps (API id) -> [authorizers]64    authorizers: Dict[str, List[Dict]]65    # maps (API id) -> [validators]66    validators: Dict[str, List[Dict]]67    # maps (API id) -> [documentation_parts]68    documentation_parts: Dict[str, List[Dict]]69    # maps (API id) -> [gateway_responses]70    gateway_responses: Dict[str, List[Dict]]71    # account details72    account: Dict[str, Any]73    # maps (domain_name) -> [path_mappings]74    base_path_mappings: Dict[str, List[Dict]]75    # maps ID to VPC link details76    vpc_links: Dict[str, Dict]77    # maps cert ID to client certificate details78    client_certificates: Dict[str, Dict]79    # maps resource ARN to tags80    TAGS: Dict[str, Dict[str, str]] = {}81    def __init__(self):82        self.authorizers = {}83        self.validators = {}84        self.documentation_parts = {}85        self.gateway_responses = {}86        self.account = {87            "cloudwatchRoleArn": aws_stack.role_arn("api-gw-cw-role"),88            "throttleSettings": {"burstLimit": 1000, "rateLimit": 500},89            "features": ["UsagePlans"],90            "apiKeyVersion": "1",91        }92        self.base_path_mappings = {}93        self.vpc_links = {}94        self.client_certificates = {}95class Resolver:96    def __init__(self, document: dict, allow_recursive=True):97        self.document = document98        self.allow_recursive = allow_recursive99        # cache which maps known refs to part of the document100        self._cache = {}101        self._refpaths = ["#"]102    def _is_ref(self, item) -> bool:103        return isinstance(item, dict) and "$ref" in item104    def _is_internal_ref(self, refpath) -> bool:105        return str(refpath).startswith("#/")106    @property107    def current_path(self):108        return self._refpaths[-1]109    @contextlib.contextmanager110    def _pathctx(self, refpath: str):111        if not self._is_internal_ref(refpath):112            refpath = "/".join((self.current_path, refpath))113        self._refpaths.append(refpath)114        yield115        self._refpaths.pop()116    def _resolve_refpath(self, refpath: str) -> dict:117        if refpath in self._refpaths and not self.allow_recursive:118            raise Exception("recursion detected with allow_recursive=False")119        if refpath in self._cache:120            return self._cache.get(refpath)121        with self._pathctx(refpath):122            if self._is_internal_ref(self.current_path):123                cur = self.document124            else:125                raise NotImplementedError("External references not yet supported.")126            for step in self.current_path.split("/")[1:]:127                cur = cur.get(step)128            self._cache[self.current_path] = cur129            return cur130    def _namespaced_resolution(self, namespace: str, data: Union[dict, list]) -> Union[dict, list]:131        with self._pathctx(namespace):132            return self._resolve_references(data)133    def _resolve_references(self, data) -> Union[dict, list]:134        if self._is_ref(data):135            return self._resolve_refpath(data["$ref"])136        if isinstance(data, dict):137            for k, v in data.items():138                data[k] = self._namespaced_resolution(k, v)139        elif isinstance(data, list):140            for i, v in enumerate(data):141                data[i] = self._namespaced_resolution(str(i), v)142        return data143    def resolve_references(self) -> dict:144        return self._resolve_references(self.document)145def resolve_references(data: dict, allow_recursive=True) -> dict:146    resolver = Resolver(data, allow_recursive=allow_recursive)147    return resolver.resolve_references()148def make_json_response(message):149    return requests_response(json.dumps(message), headers={"Content-Type": APPLICATION_JSON})150def make_error_response(message, code=400, error_type=None):151    if code == 404 and not error_type:152        error_type = "NotFoundException"153    error_type = error_type or "InvalidRequest"154    return requests_error_response_json(message, code=code, error_type=error_type)155def make_accepted_response():156    response = Response()157    response.status_code = 202158    return response159def get_api_id_from_path(path):160    if match := re.match(PATH_REGEX_SUB, path):161        return match.group(1)162    return re.match(PATH_REGEX_MAIN, path).group(1)163def is_test_invoke_method(method, path):164    return method == "POST" and bool(re.match(PATH_REGEX_TEST_INVOKE_API, path))165def get_stage_variables(context: ApiInvocationContext) -> Optional[Dict[str, str]]:166    if is_test_invoke_method(context.method, context.path):167        return None168    if not context.stage:169        return {}170    region_name = [171        name172        for name, region in apigateway_models.apigateway_backends.items()173        if context.api_id in region.apis174    ][0]175    api_gateway_client = aws_stack.connect_to_service("apigateway", region_name=region_name)176    try:177        response = api_gateway_client.get_stage(restApiId=context.api_id, stageName=context.stage)178        return response.get("variables")179    except Exception:180        LOG.info("Failed to get stage %s for API id %s", context.stage, context.api_id)181        return {}182# -----------------------183# GATEWAY RESPONSES APIs184# -----------------------185# TODO: merge with to_response_json(..) above186def gateway_response_to_response_json(item, api_id):187    base_path = "/restapis/%s/gatewayresponses" % api_id188    item["_links"] = {189        "self": {"href": "%s/%s" % (base_path, item["responseType"])},190        "gatewayresponse:put": {191            "href": "%s/{response_type}" % base_path,192            "templated": True,193        },194        "gatewayresponse:update": {"href": "%s/%s" % (base_path, item["responseType"])},195    }196    item["responseParameters"] = item.get("responseParameters", {})197    item["responseTemplates"] = item.get("responseTemplates", {})198    return item199def get_gateway_responses(api_id):200    region_details = APIGatewayRegion.get()201    result = region_details.gateway_responses.get(api_id, [])202    href = "http://docs.aws.amazon.com/apigateway/latest/developerguide/restapi-gatewayresponse-{rel}.html"203    base_path = "/restapis/%s/gatewayresponses" % api_id204    result = {205        "_links": {206            "curies": {"href": href, "name": "gatewayresponse", "templated": True},207            "self": {"href": base_path},208            "first": {"href": base_path},209            "gatewayresponse:by-type": {210                "href": "%s/{response_type}" % base_path,211                "templated": True,212            },213            "item": [{"href": "%s/%s" % (base_path, r["responseType"])} for r in result],214        },215        "_embedded": {"item": [gateway_response_to_response_json(i, api_id) for i in result]},216        # Note: Looks like the format required by aws CLI ("item" at top level) differs from the docs:217        # https://docs.aws.amazon.com/apigateway/api-reference/resource/gateway-responses/218        "item": [gateway_response_to_response_json(i, api_id) for i in result],219    }220    return result221def get_gateway_response(api_id, response_type):222    region_details = APIGatewayRegion.get()223    responses = region_details.gateway_responses.get(api_id, [])224    result = [r for r in responses if r["responseType"] == response_type]225    if result:226        return result[0]227    return make_error_response(228        "Gateway response %s for API Gateway %s not found" % (response_type, api_id),229        code=404,230    )231def put_gateway_response(api_id, response_type, data):232    region_details = APIGatewayRegion.get()233    responses = region_details.gateway_responses.setdefault(api_id, [])234    existing = ([r for r in responses if r["responseType"] == response_type] or [None])[0]235    if existing:236        existing.update(data)237    else:238        data["responseType"] = response_type239        responses.append(data)240    return data241def delete_gateway_response(api_id, response_type):242    region_details = APIGatewayRegion.get()243    responses = region_details.gateway_responses.get(api_id) or []244    region_details.gateway_responses[api_id] = [245        r for r in responses if r["responseType"] != response_type246    ]247    return make_accepted_response()248def update_gateway_response(api_id, response_type, data):249    region_details = APIGatewayRegion.get()250    responses = region_details.gateway_responses.setdefault(api_id, [])251    existing = ([r for r in responses if r["responseType"] == response_type] or [None])[0]252    if existing is None:253        return make_error_response(254            "Gateway response %s for API Gateway %s not found" % (response_type, api_id),255            code=404,256        )257    result = apply_json_patch_safe(existing, data["patchOperations"])258    return result259def handle_gateway_responses(method, path, data, headers):260    search_match = re.search(PATH_REGEX_RESPONSES, path)261    api_id = search_match.group(1)262    response_type = (search_match.group(2) or "").lstrip("/")263    if method == "GET":264        if response_type:265            return get_gateway_response(api_id, response_type)266        return get_gateway_responses(api_id)267    if method == "PUT":268        return put_gateway_response(api_id, response_type, data)269    if method == "PATCH":270        return update_gateway_response(api_id, response_type, data)271    if method == "DELETE":272        return delete_gateway_response(api_id, response_type)273    return make_error_response(274        "Not implemented for API Gateway gateway responses: %s" % method, code=404275    )276# ---------------277# UTIL FUNCTIONS278# ---------------279def find_api_subentity_by_id(api_id, entity_id, map_name):280    region_details = APIGatewayRegion.get()281    auth_list = getattr(region_details, map_name).get(api_id) or []282    entity = ([a for a in auth_list if a["id"] == entity_id] or [None])[0]283    return entity284def path_based_url(api_id, stage_name, path):285    """Return URL for inbound API gateway for given API ID, stage name, and path"""286    pattern = "%s/restapis/{api_id}/{stage_name}/%s{path}" % (287        config.service_url("apigateway"),288        PATH_USER_REQUEST,289    )290    return pattern.format(api_id=api_id, stage_name=stage_name, path=path)291def host_based_url(rest_api_id: str, path: str, stage_name: str = None):292    """Return URL for inbound API gateway for given API ID, stage name, and path with custom dns293    format"""294    pattern = "http://{endpoint}{stage}{path}"295    stage = stage_name and f"/{stage_name}" or ""296    return pattern.format(endpoint=get_execute_api_endpoint(rest_api_id), stage=stage, path=path)297def get_execute_api_endpoint(api_id: str, protocol: str = "") -> str:298    port = config.get_edge_port_http()299    return f"{protocol}{api_id}.execute-api.{LOCALHOST_HOSTNAME}:{port}"300def tokenize_path(path):301    return path.lstrip("/").split("/")302def extract_path_params(path: str, extracted_path: str) -> Dict[str, str]:303    tokenized_extracted_path = tokenize_path(extracted_path)304    # Looks for '{' in the tokenized extracted path305    path_params_list = [(i, v) for i, v in enumerate(tokenized_extracted_path) if "{" in v]306    tokenized_path = tokenize_path(path)307    path_params = {}308    for param in path_params_list:309        path_param_name = param[1][1:-1]310        path_param_position = param[0]311        if path_param_name.endswith("+"):312            path_params[path_param_name.rstrip("+")] = "/".join(313                tokenized_path[path_param_position:]314            )315        else:316            path_params[path_param_name] = tokenized_path[path_param_position]317    path_params = common.json_safe(path_params)318    return path_params319def extract_query_string_params(path: str) -> Tuple[str, Dict[str, str]]:320    parsed_path = urlparse.urlparse(path)321    path = parsed_path.path322    parsed_query_string_params = urlparse.parse_qs(parsed_path.query)323    query_string_params = {}324    for query_param_name, query_param_values in parsed_query_string_params.items():325        if len(query_param_values) == 1:326            query_string_params[query_param_name] = query_param_values[0]327        else:328            query_string_params[query_param_name] = query_param_values329    # strip trailing slashes from path to fix downstream lookups330    path = path.rstrip("/") or "/"331    return [path, query_string_params]332def get_cors_response(headers):333    # TODO: for now we simply return "allow-all" CORS headers, but in the future334    # we should implement custom headers for CORS rules, as supported by API Gateway:335    # http://docs.aws.amazon.com/apigateway/latest/developerguide/how-to-cors.html336    response = Response()337    response.status_code = 200338    response.headers["Access-Control-Allow-Origin"] = "*"339    response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, PATCH"340    response.headers["Access-Control-Allow-Headers"] = "*"341    response._content = ""342    return response343def get_rest_api_paths(rest_api_id, region_name=None):344    apigateway = aws_stack.connect_to_service(service_name="apigateway", region_name=region_name)345    resources = apigateway.get_resources(restApiId=rest_api_id, limit=100)346    resource_map = {}347    for resource in resources["items"]:348        path = resource.get("path")349        # TODO: check if this is still required in the general case (can we rely on "path" being350        #  present?)351        path = path or aws_stack.get_apigateway_path_for_resource(352            rest_api_id, resource["id"], region_name=region_name353        )354        resource_map[path] = resource355    return resource_map356# TODO: Extract this to a set of rules that have precedence and easy to test individually.357#358#  https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-method-settings359#  -method-request.html360#  https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api-develop-routes.html361def get_resource_for_path(path: str, path_map: Dict[str, Dict]) -> Optional[Tuple[str, dict]]:362    matches = []363    # creates a regex from the input path if there are parameters, e.g /foo/{bar}/baz -> /foo/[364    # ^\]+/baz, otherwise is a direct match.365    for api_path, details in path_map.items():366        api_path_regex = re.sub(r"{[^+]+\+}", r"[^\?#]+", api_path)367        api_path_regex = re.sub(r"{[^}]+}", r"[^/]+", api_path_regex)368        if re.match(r"^%s$" % api_path_regex, path):369            matches.append((api_path, details))370    # if there are no matches, it's not worth to proceed, bail here!371    if not matches:372        return None373    # so we have matches and perhaps more than one, e.g374    # /{proxy+} and /api/{proxy+} for inputs like /api/foo/bar375    # /foo/{param1}/baz and /foo/{param1}/{param2} for inputs like /for/bar/baz376    if len(matches) > 1:377        # check if we have an exact match (exact matches take precedence)378        for match in matches:379            if match[0] == path:380                return match381        # not an exact match but parameters can fit in382        for match in matches:383            if path_matches_pattern(path, match[0]):384                return match385        # at this stage, we have more than one match but we have an eager example like386        # /{proxy+} or /api/{proxy+}, so we pick the best match by sorting by length387        sorted_matches = sorted(matches, key=lambda x: len(x[0]), reverse=True)388        return sorted_matches[0]389    return matches[0]390def path_matches_pattern(path, api_path):391    api_paths = api_path.split("/")392    paths = path.split("/")393    reg_check = re.compile(r"{(.*)}")394    if len(api_paths) != len(paths):395        return False396    results = [397        part == paths[indx]398        for indx, part in enumerate(api_paths)399        if reg_check.match(part) is None and part400    ]401    return len(results) > 0 and all(results)402def connect_api_gateway_to_sqs(gateway_name, stage_name, queue_arn, path, region_name=None):403    resources = {}404    template = APIGATEWAY_SQS_DATA_INBOUND_TEMPLATE405    resource_path = path.replace("/", "")406    region_name = region_name or aws_stack.get_region()407    try:408        arn = parse_arn(queue_arn)409        queue_name = arn["resource"]410        sqs_region = arn["region"]411    except InvalidArnException:412        queue_name = queue_arn413        sqs_region = region_name414    resources[resource_path] = [415        {416            "httpMethod": "POST",417            "authorizationType": "NONE",418            "integrations": [419                {420                    "type": "AWS",421                    "uri": "arn:aws:apigateway:%s:sqs:path/%s/%s"422                    % (sqs_region, TEST_AWS_ACCOUNT_ID, queue_name),423                    "requestTemplates": {"application/json": template},424                }425            ],426        }427    ]428    return aws_stack.create_api_gateway(429        name=gateway_name,430        resources=resources,431        stage_name=stage_name,432        region_name=region_name,433    )434def apply_json_patch_safe(subject, patch_operations, in_place=True, return_list=False):435    """Apply JSONPatch operations, using some customizations for compatibility with API GW436    resources."""437    results = []438    patch_operations = (439        [patch_operations] if isinstance(patch_operations, dict) else patch_operations440    )441    for operation in patch_operations:442        try:443            # special case: for "replace" operations, assume "" as the default value444            if operation["op"] == "replace" and operation.get("value") is None:445                operation["value"] = ""446            if operation["op"] != "remove" and operation.get("value") is None:447                LOG.info('Missing "value" in JSONPatch operation for %s: %s', subject, operation)448                continue449            if operation["op"] == "add":450                path = operation["path"]451                target = subject.get(path.strip("/"))452                target = target or common.extract_from_jsonpointer_path(subject, path)453                if not isinstance(target, list):454                    # for "add" operations, we should ensure that the path target is a list instance455                    value = [] if target is None else [target]456                    common.assign_to_path(subject, path, value=value, delimiter="/")457                target = common.extract_from_jsonpointer_path(subject, path)458                if isinstance(target, list) and not path.endswith("/-"):459                    # if "path" is an attribute name pointing to an array in "subject", and we're running460                    # an "add" operation, then we should use the standard-compliant notation "/path/-"461                    operation["path"] = "%s/-" % path462            result = apply_patch(subject, [operation], in_place=in_place)463            if not in_place:464                subject = result465            results.append(result)466        except JsonPointerException:467            pass  # path cannot be found - ignore468        except Exception as e:469            if "non-existent object" in str(e):470                if operation["op"] == "replace":471                    # fall back to an ADD operation if the REPLACE fails472                    operation["op"] = "add"473                    result = apply_patch(subject, [operation], in_place=in_place)474                    results.append(result)475                    continue476                if operation["op"] == "remove" and isinstance(subject, dict):477                    result = subject.pop(operation["path"], None)478                    results.append(result)479                    continue480            raise481    if return_list:482        return results483    return (results or [subject])[-1]484def import_api_from_openapi_spec(485    rest_api: apigateway_models.RestAPI, body: Dict, query_params: Dict486) -> apigateway_models.RestAPI:487    """Import an API from an OpenAPI spec document"""488    resolved_schema = resolve_references(body)489    # XXX for some reason this makes cf tests fail that's why is commented.490    # test_cfn_handle_serverless_api_resource491    # rest_api.name = resolved_schema.get("info", {}).get("title")492    rest_api.description = resolved_schema.get("info", {}).get("description")493    # Remove default root, then add paths from API spec494    rest_api.resources = {}495    def get_or_create_path(path):496        parts = path.rstrip("/").replace("//", "/").split("/")497        parent_id = ""498        if len(parts) > 1:499            parent_path = "/".join(parts[:-1])500            parent = get_or_create_path(parent_path)501            parent_id = parent.id502        if existing := [503            r504            for r in rest_api.resources.values()505            if r.path_part == (parts[-1] or "/") and (r.parent_id or "") == (parent_id or "")506        ]:507            return existing[0]508        return add_path(path, parts, parent_id=parent_id)509    def add_path(path, parts, parent_id=""):510        child_id = create_resource_id()511        path = path or "/"512        child = apigateway_models.Resource(513            resource_id=child_id,514            region_name=rest_api.region_name,515            api_id=rest_api.id,516            path_part=parts[-1] or "/",517            parent_id=parent_id,518        )519        for method, method_schema in resolved_schema["paths"].get(path, {}).items():520            method = method.upper()521            method_resource = child.add_method(method, None, None)522            method_integration = method_schema.get("x-amazon-apigateway-integration", {})523            responses = method_schema.get("responses", {})524            for status_code in responses:525                response_model = None526                if model_schema := responses.get(status_code, {}).get("schema", {}):527                    response_model = {APPLICATION_JSON: model_schema}528                response_parameters = (529                    method_integration.get("responses", {})530                    .get("default", {})531                    .get("responseParameters")532                )533                method_resource.create_response(534                    status_code,535                    response_model,536                    response_parameters,537                )538            integration = apigateway_models.Integration(539                http_method=method,540                uri=method_integration.get("uri"),541                integration_type=method_integration["type"],542                passthrough_behavior=method_integration.get("passthroughBehavior"),543                request_templates=method_integration.get("requestTemplates") or {},544            )545            integration.create_integration_response(546                status_code=method_integration.get("default", {}).get("statusCode", 200),547                selection_pattern=None,548                response_templates=method_integration.get("default", {}).get(549                    "responseTemplates", None550                ),551                content_handling=None,552            )553            child.resource_methods[method]["methodIntegration"] = integration554        rest_api.resources[child_id] = child555        return child556    if definitions := resolved_schema.get("definitions", {}):557        for name, model in definitions.items():558            rest_api.add_model(name=name, schema=model, content_type=APPLICATION_JSON)559    basepath_mode = (query_params.get("basepath") or ["prepend"])[0]560    base_path = (resolved_schema.get("basePath") or "") if basepath_mode == "prepend" else ""561    for path in resolved_schema.get("paths", {}):562        get_or_create_path(base_path + path)563    policy = resolved_schema.get("x-amazon-apigateway-policy")564    if policy:565        policy = json.dumps(policy) if isinstance(policy, dict) else str(policy)566        rest_api.policy = policy567    minimum_compression_size = resolved_schema.get("x-amazon-apigateway-minimum-compression-size")568    if minimum_compression_size is not None:569        rest_api.minimum_compression_size = int(minimum_compression_size)570    endpoint_config = resolved_schema.get("x-amazon-apigateway-endpoint-configuration")571    if endpoint_config:572        if endpoint_config.get("vpcEndpointIds"):573            endpoint_config.setdefault("types", ["PRIVATE"])574        rest_api.endpoint_configuration = endpoint_config575    return rest_api576def get_target_resource_details(invocation_context: ApiInvocationContext) -> Tuple[str, Dict]:577    """Look up and return the API GW resource (path pattern + resource dict) for the given invocation context."""578    path_map = get_rest_api_paths(579        rest_api_id=invocation_context.api_id, region_name=invocation_context.region_name580    )581    relative_path = invocation_context.invocation_path582    try:583        extracted_path, resource = get_resource_for_path(path=relative_path, path_map=path_map)584        invocation_context.resource = resource585        return extracted_path, resource586    except Exception:587        return None, None588def get_target_resource_method(invocation_context: ApiInvocationContext) -> Optional[Dict]:589    """Look up and return the API GW resource method for the given invocation context."""590    _, resource = get_target_resource_details(invocation_context)591    if not resource:592        return None593    methods = resource.get("resourceMethods") or {}594    method_name = invocation_context.method.upper()595    return methods.get(method_name) or methods.get("ANY")596def get_event_request_context(invocation_context: ApiInvocationContext):597    method = invocation_context.method598    path = invocation_context.path599    headers = invocation_context.headers600    integration_uri = invocation_context.integration_uri601    resource_path = invocation_context.resource_path602    resource_id = invocation_context.resource_id603    set_api_id_stage_invocation_path(invocation_context)604    relative_path, query_string_params = extract_query_string_params(605        path=invocation_context.path_with_query_string606    )607    api_id = invocation_context.api_id608    stage = invocation_context.stage609    source_ip = headers.get("X-Forwarded-For", ",").split(",")[-2].strip()610    integration_uri = integration_uri or ""611    account_id = integration_uri.split(":lambda:path")[-1].split(":function:")[0].split(":")[-1]612    account_id = account_id or TEST_AWS_ACCOUNT_ID613    request_context = {614        "accountId": account_id,615        "apiId": api_id,616        "resourcePath": resource_path or relative_path,617        "domainPrefix": invocation_context.domain_prefix,618        "domainName": invocation_context.domain_name,619        "resourceId": resource_id,620        "requestId": long_uid(),621        "identity": {622            "accountId": account_id,623            "sourceIp": source_ip,624            "userAgent": headers.get("User-Agent"),625        },626        "httpMethod": method,627        "protocol": "HTTP/1.1",628        "requestTime": pytz.utc.localize(datetime.datetime.utcnow()).strftime(629            REQUEST_TIME_DATE_FORMAT630        ),631        "requestTimeEpoch": int(time.time() * 1000),632        "authorizer": {},633    }634    # set "authorizer" and "identity" event attributes from request context635    auth_context = invocation_context.auth_context636    if auth_context:637        request_context["authorizer"] = auth_context638    request_context["identity"].update(invocation_context.auth_identity or {})639    if not is_test_invoke_method(method, path):640        request_context["path"] = (f"/{stage}" if stage else "") + relative_path641        request_context["stage"] = stage642    return request_context643def set_api_id_stage_invocation_path(644    invocation_context: ApiInvocationContext,645) -> ApiInvocationContext:646    # skip if all details are already available647    values = (648        invocation_context.api_id,649        invocation_context.stage,650        invocation_context.path_with_query_string,651    )652    if all(values):653        return invocation_context654    # skip if this is a websocket request655    if invocation_context.is_websocket_request():656        return invocation_context657    path = invocation_context.path658    headers = invocation_context.headers659    path_match = re.search(PATH_REGEX_USER_REQUEST, path)660    host_header = headers.get(HEADER_LOCALSTACK_EDGE_URL, "") or headers.get("Host") or ""661    host_match = re.search(HOST_REGEX_EXECUTE_API, host_header)662    test_invoke_match = re.search(PATH_REGEX_TEST_INVOKE_API, path)663    if path_match:664        api_id = path_match.group(1)665        stage = path_match.group(2)666        relative_path_w_query_params = "/%s" % path_match.group(3)667    elif host_match:668        api_id = extract_api_id_from_hostname_in_url(host_header)669        stage = path.strip("/").split("/")[0]670        relative_path_w_query_params = "/%s" % path.lstrip("/").partition("/")[2]671    elif test_invoke_match:672        # special case: fetch the resource details for TestInvokeApi invocations673        stage = None674        region_name = invocation_context.region_name675        api_id = test_invoke_match.group(1)676        resource_id = test_invoke_match.group(2)677        query_string = test_invoke_match.group(4) or ""678        apigateway = aws_stack.connect_to_service(679            service_name="apigateway", region_name=region_name680        )681        resource = apigateway.get_resource(restApiId=api_id, resourceId=resource_id)682        resource_path = resource.get("path")683        relative_path_w_query_params = f"{resource_path}{query_string}"684    else:685        raise Exception(686            f"Unable to extract API Gateway details from request: {path} {dict(headers)}"687        )688    if api_id:689        # set current region in request thread local, to ensure aws_stack.get_region() works properly690        # TODO: replace with RequestContextManager691        if getattr(THREAD_LOCAL, "request_context", None) is not None:692            api_region = API_REGIONS.get(api_id, "")693            THREAD_LOCAL.request_context.headers[MARKER_APIGW_REQUEST_REGION] = api_region694    # set details in invocation context695    invocation_context.api_id = api_id696    invocation_context.stage = stage697    invocation_context.path_with_query_string = relative_path_w_query_params698    return invocation_context699def extract_api_id_from_hostname_in_url(hostname: str) -> str:700    """Extract API ID 'id123' from URLs like https://id123.execute-api.localhost.localstack.cloud:4566"""701    match = re.match(HOST_REGEX_EXECUTE_API, hostname)702    api_id = match.group(1)...
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.