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

How to use _add_modeled_error_fields method in localstack

Best Python code snippet using localstack_python

endpoint.py

Source:endpoint.py

...241        # they currently have no reference to the operation or service model242        # The parsers should probably take the operation model instead of243        # output shape but we can't change that now244        if http_response.status_code >= 300:245            self._add_modeled_error_fields(246                response_dict, parsed_response,247                operation_model, parser,248            )249        history_recorder.record('PARSED_RESPONSE', parsed_response)250        return (http_response, parsed_response), None251    def _add_modeled_error_fields(252            self, response_dict, parsed_response,253            operation_model, parser,254    ):255        error_code = parsed_response.get("Error", {}).get("Code")256        if error_code is None:257            return258        service_model = operation_model.service_model259        error_shape = service_model.shape_for_error_code(error_code)260        if error_shape is None:261            return262        modeled_parse = parser.parse(response_dict, error_shape)263        # TODO: avoid naming conflicts with ResponseMetadata and Error264        parsed_response.update(modeled_parse)265    def _needs_retry(self, attempts, operation_model, request_dict,...

client.py

Source:client.py

...45    def stream(self) -> Iterable[bytes]:46        return self.iterator47    def close(self):48        pass49def _add_modeled_error_fields(50    response_dict: Dict,51    parsed_response: Dict,52    operation_model: OperationModel,53    parser: ResponseParser,54):55    """56    This function adds additional error shape members (other than message, code, and type) to an already parsed error57    response dict.58    Port of botocore's Endpoint#_add_modeled_error_fields.59    """60    error_code = parsed_response.get("Error", {}).get("Code")61    if error_code is None:62        return63    service_model = operation_model.service_model64    error_shape = service_model.shape_for_error_code(error_code)65    if error_shape is None:66        return67    modeled_parse = parser.parse(response_dict, error_shape)68    parsed_response.update(modeled_parse)69def _cbor_timestamp_parser(value):70    return datetime.fromtimestamp(value / 1000)71def _cbor_blob_parser(value):72    return bytes(value)73@hooks.on_infra_start()74def _patch_botocore_json_parser():75    from botocore.parsers import BaseJSONParser76    @patch(BaseJSONParser._parse_body_as_json)77    def _parse_body_as_json(fn, self, body_contents):78        """79        botocore does not support CBOR encoded response parsing. Since we use the botocore parsers80        to parse responses from external backends (like kinesalite), we need to patch botocore to81        try CBOR decoding in case the JSON decoding fails.82        """83        try:84            return fn(self, body_contents)85        except UnicodeDecodeError as json_exception:86            import cbor287            try:88                LOG.debug("botocore failed decoding JSON. Trying to decode as CBOR.")89                return cbor2.loads(body_contents)90            except Exception as cbor_exception:91                LOG.debug("CBOR fallback decoding failed.")92                raise cbor_exception from json_exception93def parse_response(94    operation: OperationModel, response: Response, include_response_metadata: bool = True95) -> ServiceResponse:96    """97    Parses an HTTP Response object into an AWS response object using botocore. It does this by adapting the98    procedure of ``botocore.endpoint.convert_to_response_dict`` to work with Werkzeug's server-side response object.99    :param operation: the operation of the original request100    :param response: the HTTP response object containing the response of the operation101    :param include_response_metadata: True if the ResponseMetadata (typical for boto response dicts) should be included102    :return: a parsed dictionary as it is returned by botocore103    """104    # this is what botocore.endpoint.convert_to_response_dict normally does105    response_dict = {106        "headers": dict(response.headers.items()),  # boto doesn't like werkzeug headers107        "status_code": response.status_code,108        "context": {109            "operation_name": operation.name,110        },111    }112    if response_dict["status_code"] >= 301:113        response_dict["body"] = response.data114    elif operation.has_event_stream_output:115        # TODO test this116        response_dict["body"] = _RawStream(response)117    elif operation.has_streaming_output:118        # for s3.GetObject for example, the Body attribute is actually a stream, not the raw bytes value119        response_dict["body"] = _ResponseStream(response)120    else:121        response_dict["body"] = response.data122    factory = ResponseParserFactory()123    if response.content_type and response.content_type.startswith("application/x-amz-cbor"):124        # botocore cannot handle CBOR encoded responses (because it never sends them), we need to modify the parser125        factory.set_parser_defaults(126            timestamp_parser=_cbor_timestamp_parser, blob_parser=_cbor_blob_parser127        )128    parser = factory.create_parser(operation.service_model.protocol)129    parsed_response = parser.parse(response_dict, operation.output_shape)130    if response.status_code >= 301:131        # Add possible additional error shape members132        _add_modeled_error_fields(response_dict, parsed_response, operation, parser)133    if not include_response_metadata:134        parsed_response.pop("ResponseMetadata", None)135    return parsed_response136def parse_service_exception(137    response: Response, parsed_response: Dict138) -> Optional[ServiceException]:139    """140    Creates a ServiceException (one ASF can handle) from a parsed response (one that botocore would return).141    It does not automatically raise the exception (see #raise_service_exception).142    :param response: Un-parsed response143    :param parsed_response: Parsed response144    :return: ServiceException or None (if it's not an error response)145    """146    if response.status_code < 301 or "Error" not in parsed_response:...

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.