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

How to use check_passthrough_behavior method in localstack

Best Python code snippet using localstack_python

integration.py

Source:integration.py

...43    class UnsupportedMediaType(Exception):44        pass45    def __init__(self, passthrough_behaviour: str):46        self.passthrough_behavior = self.get_passthrough_behavior(passthrough_behaviour)47    def check_passthrough_behavior(self, request_template):48        """49        Specifies how the method request body of an unmapped content type will be passed through50        the integration request to the back end without transformation.51        A content type is unmapped if no mapping template is defined in the integration or the52        content type does not match any of the mapped content types, as specified in requestTemplates53        """54        if not request_template and self.passthrough_behavior in {55            PassthroughBehavior.NEVER,56            PassthroughBehavior.WHEN_NO_TEMPLATES,57        }:58            raise MappingTemplates.UnsupportedMediaType()59    @staticmethod60    def get_passthrough_behavior(passthrough_behaviour: str):61        return getattr(PassthroughBehavior, passthrough_behaviour, None)62class BackendIntegration(ABC):63    """Abstract base class representing a backend integration"""64    def __init__(self):65        self.request_templates = RequestTemplates()66        self.response_templates = ResponseTemplates()67    @abstractmethod68    def invoke(self, invocation_context: ApiInvocationContext):69        pass70    @classmethod71    def _create_response(cls, status_code, headers, data=""):72        response = Response()73        response.status_code = status_code74        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)...

templates.py

Source:templates.py

...29    class UnsupportedMediaType(Exception):30        pass31    def __init__(self, passthrough_behaviour: str):32        self.passthrough_behavior = self.get_passthrough_behavior(passthrough_behaviour)33    def check_passthrough_behavior(self, request_template):34        """35        Specifies how the method request body of an unmapped content type will be passed through36        the integration request to the back end without transformation.37        A content type is unmapped if no mapping template is defined in the integration or the38        content type does not match any of the mapped content types, as specified in requestTemplates39        """40        if not request_template and self.passthrough_behavior in {41            PassthroughBehavior.NEVER,42            PassthroughBehavior.WHEN_NO_TEMPLATES,43        }:44            raise MappingTemplates.UnsupportedMediaType()45    @staticmethod46    def get_passthrough_behavior(passthrough_behaviour: str):47        return getattr(PassthroughBehavior, passthrough_behaviour, None)...

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.