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

How to use apply_response_parameters method in localstack

Best Python code snippet using localstack_python

invocations.py

Source:invocations.py

...165            request_param_val = f"method.request.querystring.{key}"166            if request_parameters.get(request_query_key, None) != request_param_val:167                query_params.pop(key)168    return add_query_params_to_url(uri, query_params)169def apply_response_parameters(invocation_context: ApiInvocationContext):170    response = invocation_context.response171    integration = invocation_context.integration172    int_responses = integration.get("integrationResponses") or {}173    if not int_responses:174        return response175    entries = list(int_responses.keys())176    return_code = str(response.status_code)177    if return_code not in entries:178        if len(entries) > 1:179            LOG.info("Found multiple integration response status codes: %s", entries)180            return response181        return_code = entries[0]182    response_params = int_responses[return_code].get("responseParameters", {})183    for key, value in response_params.items():184        # TODO: add support for method.response.body, etc ...185        if str(key).lower().startswith("method.response.header."):186            header_name = key[len("method.response.header.") :]187            response.headers[header_name] = value.strip("'")188    return response189def invoke_rest_api_from_request(invocation_context: ApiInvocationContext):190    helpers.set_api_id_stage_invocation_path(invocation_context)191    try:192        return invoke_rest_api(invocation_context)193    except AuthorizationError as e:194        api_id = invocation_context.api_id195        return make_error_response("Not authorized to invoke REST API %s: %s" % (api_id, e), 403)196def invoke_rest_api(invocation_context: ApiInvocationContext):197    invocation_path = invocation_context.path_with_query_string198    raw_path = invocation_context.path or invocation_path199    method = invocation_context.method200    headers = invocation_context.headers201    # run gateway authorizers for this request202    authorize_invocation(invocation_context)203    extracted_path, resource = helpers.get_target_resource_details(invocation_context)204    if not resource:205        return make_error_response("Unable to find path %s" % invocation_context.path, 404)206    # validate request207    validator = RequestValidator(invocation_context, aws_stack.connect_to_service("apigateway"))208    if not validator.is_request_valid():209        return make_error_response("Invalid request body", 400)210    api_key_required = resource.get("resourceMethods", {}).get(method, {}).get("apiKeyRequired")211    if not is_api_key_valid(api_key_required, headers, invocation_context.stage):212        return make_error_response("Access denied - invalid API key", 403)213    integrations = resource.get("resourceMethods", {})214    integration = integrations.get(method, {})215    if not integration:216        # HttpMethod: '*'217        # ResourcePath: '/*' - produces 'X-AMAZON-APIGATEWAY-ANY-METHOD'218        integration = integrations.get("ANY", {}) or integrations.get(219            "X-AMAZON-APIGATEWAY-ANY-METHOD", {}220        )221    integration = integration.get("methodIntegration")222    if not integration:223        if method == "OPTIONS" and "Origin" in headers:224            # default to returning CORS headers if this is an OPTIONS request225            return get_cors_response(headers)226        return make_error_response(227            "Unable to find integration for: %s %s (%s)" % (method, invocation_path, raw_path),228            404,229        )230    res_methods = resource.get("resourceMethods", {})231    meth_integration = res_methods.get(method, {}).get("methodIntegration", {})232    int_responses = meth_integration.get("integrationResponses", {})233    response_templates = int_responses.get("200", {}).get("responseTemplates", {})234    # update fields in invocation context, then forward request to next handler235    invocation_context.resource = resource236    invocation_context.resource_path = extracted_path237    invocation_context.response_templates = response_templates238    invocation_context.integration = integration239    return invoke_rest_api_integration(invocation_context)240def invoke_rest_api_integration(invocation_context: ApiInvocationContext):241    try:242        response = invoke_rest_api_integration_backend(invocation_context)243        # TODO remove this setter once all the integrations are migrated to the new response244        #  handling245        invocation_context.response = response246        response = apply_response_parameters(invocation_context)247        return response248    except Exception as e:249        msg = f"Error invoking integration for API Gateway ID '{invocation_context.api_id}': {e}"250        LOG.exception(msg)251        return make_error_response(msg, 400)252# TODO: refactor this to have a class per integration type to make it easy to253# test the encapsulated logic254def invoke_rest_api_integration_backend(invocation_context: ApiInvocationContext):255    # define local aliases from invocation context256    invocation_path = invocation_context.path_with_query_string257    method = invocation_context.method258    data = invocation_context.data259    headers = invocation_context.headers260    api_id = invocation_context.api_id...

integration.py

Source:integration.py

...74        response.headers = headers75        response._content = data76        return response77    @classmethod78    def apply_response_parameters(79        cls, invocation_context: ApiInvocationContext, response: Response80    ):81        integration = invocation_context.integration82        integration_responses = integration.get("integrationResponses") or {}83        if not integration_responses:84            return response85        entries = list(integration_responses.keys())86        return_code = str(response.status_code)87        if return_code not in entries:88            if len(entries) > 1:89                LOG.info("Found multiple integration response status codes: %s", entries)90                return response91            return_code = entries[0]92        response_params = integration_responses[return_code].get("responseParameters", {})93        for key, value in response_params.items():94            # TODO: add support for method.response.body, etc ...95            if str(key).lower().startswith("method.response.header."):96                header_name = key[len("method.response.header.") :]97                response.headers[header_name] = value.strip("'")98        return response99class SnsIntegration(BackendIntegration):100    def invoke(self, invocation_context: ApiInvocationContext) -> Response:101        invocation_context.context = get_event_request_context(invocation_context)102        try:103            payload = self.request_templates.render(invocation_context)104        except Exception as e:105            LOG.warning("Failed to apply template for SNS integration", e)106            raise107        uri = (108            invocation_context.integration.get("uri")109            or invocation_context.integration.get("integrationUri")110            or ""111        )112        region_name = uri.split(":")[3]113        headers = aws_stack.mock_aws_request_headers(service="sns", region_name=region_name)114        return make_http_request(115            config.service_url("sns"), method="POST", headers=headers, data=payload116        )117class LambdaProxyIntegration(BackendIntegration):118    @classmethod119    def update_content_length(cls, response: Response):120        if response and response.content is not None:121            response.headers["Content-Length"] = str(len(response.content))122    def invoke(self, invocation_context: ApiInvocationContext):123        uri = (124            invocation_context.integration.get("uri")125            or invocation_context.integration.get("integrationUri")126            or ""127        )128        relative_path, query_string_params = extract_query_string_params(129            path=invocation_context.path_with_query_string130        )131        api_id = invocation_context.api_id132        stage = invocation_context.stage133        headers = invocation_context.headers134        resource_path = invocation_context.resource_path135        invocation_context.context = get_event_request_context(invocation_context)136        try:137            path_params = extract_path_params(path=relative_path, extracted_path=resource_path)138            invocation_context.path_params = path_params139        except Exception:140            path_params = {}141        func_arn = uri142        if ":lambda:path" in uri:143            func_arn = uri.split(":lambda:path")[1].split("functions/")[1].split("/invocations")[0]144        if invocation_context.authorizer_type:145            invocation_context.context["authorizer"] = invocation_context.auth_context146        payload = self.request_templates.render(invocation_context)147        # TODO: change this signature to InvocationContext as well!148        result = lambda_api.process_apigateway_invocation(149            func_arn,150            relative_path,151            payload,152            stage,153            api_id,154            headers,155            is_base64_encoded=invocation_context.is_data_base64_encoded,156            path_params=path_params,157            query_string_params=query_string_params,158            method=invocation_context.method,159            resource_path=resource_path,160            request_context=invocation_context.context,161            stage_variables=invocation_context.stage_variables,162        )163        if isinstance(result, FlaskResponse):164            response = flask_to_requests_response(result)165        elif isinstance(result, Response):166            response = result167        else:168            response = LambdaResponse()169            response.headers.update({"content-type": "application/json"})170            parsed_result = result if isinstance(result, dict) else json.loads(str(result or "{}"))171            parsed_result = common.json_safe(parsed_result)172            parsed_result = {} if parsed_result is None else parsed_result173            keys = parsed_result.keys()174            if not ("statusCode" in keys and "body" in keys):175                LOG.warning(176                    'Lambda output should follow the next JSON format: { "isBase64Encoded": true|false, "statusCode": httpStatusCode, "headers": { "headerName": "headerValue", ... },"body": "..."}'177                )178                response.status_code = 502179                response._content = json.dumps({"message": "Internal server error"})180                return response181            response.status_code = int(parsed_result.get("statusCode", 200))182            parsed_headers = parsed_result.get("headers", {})183            if parsed_headers is not None:184                response.headers.update(parsed_headers)185            try:186                result_body = parsed_result.get("body")187                if isinstance(result_body, dict):188                    response._content = json.dumps(result_body)189                else:190                    body_bytes = to_bytes(to_str(result_body or ""))191                    if parsed_result.get("isBase64Encoded", False):192                        body_bytes = base64.b64decode(body_bytes)193                    response._content = body_bytes194            except Exception as e:195                LOG.warning("Couldn't set Lambda response content: %s", e)196                response._content = "{}"197            response.multi_value_headers = parsed_result.get("multiValueHeaders") or {}198        # apply custom response template199        self.update_content_length(response)200        invocation_context.response = response201        self.response_templates.render(invocation_context)202        return invocation_context.response203class MockIntegration(BackendIntegration):204    @classmethod205    def check_passthrough_behavior(cls, passthrough_behavior: str, request_template: str):206        return MappingTemplates(passthrough_behavior).check_passthrough_behavior(request_template)207    def invoke(self, invocation_context: ApiInvocationContext) -> Response:208        passthrough_behavior = invocation_context.integration.get("passthroughBehavior") or ""209        request_template = invocation_context.integration.get("requestTemplates", {}).get(210            invocation_context.headers.get(HEADER_CONTENT_TYPE)211        )212        # based on the configured passthrough behavior and the existence of template or not,213        # we proceed calling the integration or raise an exception.214        try:215            self.check_passthrough_behavior(passthrough_behavior, request_template)216        except MappingTemplates.UnsupportedMediaType:217            http_status = HTTPStatus(415)218            return MockIntegration._create_response(219                http_status.value,220                headers={"Content-Type": APPLICATION_JSON},221                data=json.dumps({"message": f"{http_status.phrase}"}),222            )223        # request template rendering224        request_payload = self.request_templates.render(invocation_context)225        # mapping is done based on "statusCode" field226        status_code = 200227        if invocation_context.headers.get(HEADER_CONTENT_TYPE) == APPLICATION_JSON:228            try:229                mock_response = json.loads(request_payload)230                status_code = mock_response.get("statusCode", status_code)231            except Exception as e:232                LOG.warning("failed to deserialize request payload after transformation: %s", e)233                http_status = HTTPStatus(500)234                return MockIntegration._create_response(235                    http_status.value,236                    headers={"Content-Type": APPLICATION_JSON},237                    data=json.dumps({"message": f"{http_status.phrase}"}),238                )239        # response template240        response = MockIntegration._create_response(241            status_code, invocation_context.headers, data=request_payload242        )243        response._content = self.response_templates.render(invocation_context, response=response)244        # apply response parameters245        response = self.apply_response_parameters(invocation_context, response)246        if not invocation_context.headers.get(HEADER_CONTENT_TYPE):247            invocation_context.headers.update({HEADER_CONTENT_TYPE: APPLICATION_JSON})248        return response249class VelocityUtilApiGateway(VelocityUtil):250    """251    Simple class to mimic the behavior of variable '$util' in AWS API Gateway integration252    velocity templates.253    See: http://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html254    """255    def base64Encode(self, s):256        if not isinstance(s, str):257            s = json.dumps(s)258        encoded_str = s.encode(config.DEFAULT_ENCODING)259        encoded_b64_str = base64.b64encode(encoded_str)...

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.