How to use _add_additional_error_tags method in localstack

Best Python code snippet using localstack_python
serializer.py
Source:serializer.py
...429        error_tag = ETree.SubElement(root, "Error")430        self._add_error_tags(error, error_tag)431        request_id = ETree.SubElement(root, "RequestId")432        request_id.text = gen_amzn_requestid_long()433        self._add_additional_error_tags(error, root, shape)434        response.set_response(self._encode_payload(self._xml_to_string(root)))435    def _add_error_tags(self, error: ServiceException, error_tag: ETree.Element) -> None:436        code_tag = ETree.SubElement(error_tag, "Code")437        code_tag.text = error.code438        message = self._get_error_message(error)439        if message:440            self._default_serialize(error_tag, message, None, "Message")441        if error.sender_fault:442            # The sender fault is either not set or "Sender"443            self._default_serialize(error_tag, "Sender", None, "Type")444    def _add_additional_error_tags(445        self, error: ServiceException, node: ETree, shape: StructureShape446    ):447        if shape:448            params = {}449            # TODO add a possibility to serialize simple non-modelled errors (like S3 NoSuchBucket#BucketName)450            for member in shape.members:451                # XML protocols do not add modeled default fields to the root node452                # (tested for cloudfront, route53, cloudwatch, iam)453                if member.lower() not in ["code", "message"] and hasattr(error, member):454                    params[member] = getattr(error, member)455            # If there is an error shape with members which should be set, they need to be added to the node456            if params:457                # Serialize the remaining params458                root_name = shape.serialization.get("name", shape.name)459                pseudo_root = ETree.Element("")460                self._serialize(shape, params, pseudo_root, root_name)461                real_root = list(pseudo_root)[0]462                # Add the child elements to the already created root error element463                for child in list(real_root):464                    node.append(child)465    def _serialize_body_params(466        self, params: dict, shape: Shape, operation_model: OperationModel467    ) -> Optional[str]:468        root = self._serialize_body_params_to_xml(params, shape, operation_model)469        self._prepare_additional_traits_in_xml(root)470        return self._xml_to_string(root)471    def _serialize_body_params_to_xml(472        self, params: dict, shape: Shape, operation_model: OperationModel473    ) -> Optional[ETree.Element]:474        if shape is None:475            return476        # The botocore serializer expects `shape.serialization["name"]`, but this isn't always present for responses477        root_name = shape.serialization.get("name", shape.name)478        pseudo_root = ETree.Element("")479        self._serialize(shape, params, pseudo_root, root_name)480        real_root = list(pseudo_root)[0]481        return real_root482    def _serialize(self, shape: Shape, params: Any, xmlnode: ETree.Element, name: str) -> None:483        """This method dynamically invokes the correct `_serialize_type_*` method for each shape type."""484        if shape is None:485            return486        # Some output shapes define a `resultWrapper` in their serialization spec.487        # While the name would imply that the result is _wrapped_, it is actually renamed.488        if shape.serialization.get("resultWrapper"):489            name = shape.serialization.get("resultWrapper")490        try:491            method = getattr(self, "_serialize_type_%s" % shape.type_name, self._default_serialize)492            method(xmlnode, params, shape, name)493        except (TypeError, ValueError, AttributeError) as e:494            raise ProtocolSerializerError(495                f"Invalid type when serializing {shape.name}: '{xmlnode}' cannot be parsed to {shape.type_name}."496            ) from e497    def _serialize_type_structure(498        self, xmlnode: ETree.Element, params: dict, shape: StructureShape, name: str499    ) -> None:500        structure_node = ETree.SubElement(xmlnode, name)501        if "xmlNamespace" in shape.serialization:502            namespace_metadata = shape.serialization["xmlNamespace"]503            attribute_name = "xmlns"504            if namespace_metadata.get("prefix"):505                attribute_name += ":%s" % namespace_metadata["prefix"]506            structure_node.attrib[attribute_name] = namespace_metadata["uri"]507        for key, value in params.items():508            if value is None:509                # Don't serialize any param whose value is None.510                continue511            try:512                member_shape = shape.members[key]513            except KeyError:514                LOG.warning(515                    "Response object %s contains a member which is not specified: %s",516                    shape.name,517                    key,518                )519                continue520            member_name = member_shape.serialization.get("name", key)521            # We need to special case member shapes that are marked as an xmlAttribute.522            # Rather than serializing into an XML child node, we instead serialize the shape to523            # an XML attribute of the *current* node.524            if member_shape.serialization.get("xmlAttribute"):525                # xmlAttributes must have a serialization name.526                xml_attribute_name = member_shape.serialization["name"]527                structure_node.attrib[xml_attribute_name] = value528                continue529            self._serialize(member_shape, value, structure_node, member_name)530    def _serialize_type_list(531        self, xmlnode: ETree.Element, params: list, shape: ListShape, name: str532    ) -> None:533        if params is None:534            # Don't serialize any param whose value is None.535            return536        member_shape = shape.member537        if shape.serialization.get("flattened"):538            # If the list is flattened, either take the member's "name" or the name of the usual name for the parent539            # element for the children.540            element_name = self._get_serialized_name(member_shape, name)541            list_node = xmlnode542        else:543            element_name = self._get_serialized_name(member_shape, "member")544            list_node = ETree.SubElement(xmlnode, name)545        for item in params:546            # Don't serialize any item which is None547            if item is not None:548                self._serialize(member_shape, item, list_node, element_name)549    def _serialize_type_map(550        self, xmlnode: ETree.Element, params: dict, shape: MapShape, name: str551    ) -> None:552        """553        Given the ``name`` of MyMap, an input of {"key1": "val1", "key2": "val2"}, and the ``flattened: False``554        we serialize this as:555          <MyMap>556            <entry>557              <key>key1</key>558              <value>val1</value>559            </entry>560            <entry>561              <key>key2</key>562              <value>val2</value>563            </entry>564          </MyMap>565        If it is flattened, it is serialized as follows:566          <MyMap>567            <key>key1</key>568            <value>val1</value>569          </MyMap>570          <MyMap>571            <key>key2</key>572            <value>val2</value>573          </MyMap>574        """575        if params is None:576            # Don't serialize a non-existing map577            return578        if shape.serialization.get("flattened"):579            entries_node = xmlnode580            entry_node_name = name581        else:582            entries_node = ETree.SubElement(xmlnode, name)583            entry_node_name = "entry"584        for key, value in params.items():585            if value is None:586                # Don't serialize any param whose value is None.587                continue588            entry_node = ETree.SubElement(entries_node, entry_node_name)589            key_name = self._get_serialized_name(shape.key, default_name="key")590            val_name = self._get_serialized_name(shape.value, default_name="value")591            self._serialize(shape.key, key, entry_node, key_name)592            self._serialize(shape.value, value, entry_node, val_name)593    @staticmethod594    def _serialize_type_boolean(xmlnode: ETree.Element, params: bool, _, name: str) -> None:595        """596        For scalar types, the 'params' attr is actually just a scalar value representing the data597        we need to serialize as a boolean. It will either be 'true' or 'false'598        """599        node = ETree.SubElement(xmlnode, name)600        if params:601            str_value = "true"602        else:603            str_value = "false"604        node.text = str_value605    def _serialize_type_blob(606        self, xmlnode: ETree.Element, params: Union[str, bytes], _, name: str607    ) -> None:608        node = ETree.SubElement(xmlnode, name)609        node.text = self._get_base64(params)610    def _serialize_type_timestamp(611        self, xmlnode: ETree.Element, params: str, shape: Shape, name: str612    ) -> None:613        node = ETree.SubElement(xmlnode, name)614        node.text = self._convert_timestamp_to_str(615            params, shape.serialization.get("timestampFormat")616        )617    def _default_serialize(self, xmlnode: ETree.Element, params: str, _, name: str) -> None:618        node = ETree.SubElement(xmlnode, name)619        node.text = str(params)620    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):621        """622        Prepares the XML root node before being serialized with additional traits (like the Response ID in the Query623        protocol).624        For some protocols (like rest-xml), the root can be None.625        """626        pass627    def _create_default_response(self, operation_model: OperationModel) -> HttpResponse:628        response = super()._create_default_response(operation_model)629        response.headers["Content-Type"] = "text/xml"630        return response631    def _xml_to_string(self, root: Optional[ETree.Element]) -> Optional[str]:632        """Generates the string representation of the given XML element."""633        if root is not None:634            return ETree.tostring(635                element=root, encoding=self.DEFAULT_ENCODING, xml_declaration=True636            )637class BaseRestResponseSerializer(ResponseSerializer, ABC):638    """639    The BaseRestResponseSerializer performs the basic logic for the ReST response serialization.640    In our case it basically only adds the request metadata to the HTTP header.641    """642    HEADER_TIMESTAMP_FORMAT = "rfc822"643    def _serialize_response(644        self,645        parameters: dict,646        response: HttpResponse,647        shape: Optional[Shape],648        shape_members: dict,649        operation_model: OperationModel,650    ) -> None:651        header_params, payload_params = self._partition_members(parameters, shape)652        self._process_header_members(header_params, response, shape)653        # "HEAD" responses are basically "GET" responses without the actual body.654        # Do not process the body payload in this case (setting a body could also manipulate the headers)655        if operation_model.http.get("method") != "HEAD":656            self._serialize_payload(payload_params, response, shape, shape_members, operation_model)657        self._serialize_content_type(response, shape, shape_members)658        self._prepare_additional_traits_in_response(response, operation_model)659    def _serialize_payload(660        self,661        parameters: dict,662        response: HttpResponse,663        shape: Optional[Shape],664        shape_members: dict,665        operation_model: OperationModel,666    ) -> None:667        """668        Serializes the given payload.669        :param parameters: The user input params670        :param response: The final serialized HttpResponse671        :param shape: Describes the expected output shape (can be None in case of an "empty" response)672        :param shape_members: The members of the output struct shape673        :param operation_model: The specification of the operation of which the response is serialized here674        :return: None - the given `serialized` dict is modified675        """676        if shape is None:677            return678        payload_member = shape.serialization.get("payload")679        if payload_member is not None and shape_members[payload_member].type_name in [680            "blob",681            "string",682        ]:683            # If it's streaming, then the body is just the value of the payload.684            body_payload = parameters.get(payload_member, b"")685            body_payload = self._encode_payload(body_payload)686            response.set_response(body_payload)687        elif payload_member is not None:688            # If there's a payload member, we serialized that member to the body.689            body_params = parameters.get(payload_member)690            if body_params is not None:691                response.set_response(692                    self._encode_payload(693                        self._serialize_body_params(694                            body_params, shape_members[payload_member], operation_model695                        )696                    )697                )698        else:699            # Otherwise, we use the "traditional" way of serializing the whole parameters dict recursively.700            response.set_response(701                self._encode_payload(702                    self._serialize_body_params(parameters, shape, operation_model)703                )704            )705    def _serialize_content_type(self, serialized: HttpResponse, shape: Shape, shape_members: dict):706        """707        Some protocols require varied Content-Type headers708        depending on user input. This allows subclasses to apply709        this conditionally.710        """711        pass712    def _has_streaming_payload(self, payload: Optional[str], shape_members):713        """Determine if payload is streaming (a blob or string)."""714        return payload is not None and shape_members[payload].type_name in ["blob", "string"]715    def _prepare_additional_traits_in_response(716        self, response: HttpResponse, operation_model: OperationModel717    ):718        """Adds the request ID to the headers (in contrast to the body - as in the Query protocol)."""719        response = super()._prepare_additional_traits_in_response(response, operation_model)720        response.headers["x-amz-request-id"] = gen_amzn_requestid_long()721        return response722    def _process_header_members(self, parameters: dict, response: HttpResponse, shape: Shape):723        shape_members = shape.members if isinstance(shape, StructureShape) else []724        for name in shape_members:725            member_shape = shape_members[name]726            location = member_shape.serialization.get("location")727            if not location:728                continue729            if name not in parameters:730                # ignores optional keys731                continue732            key = member_shape.serialization.get("name", name)733            value = parameters[name]734            if value is None:735                continue736            if location == "header":737                response.headers[key] = self._serialize_header_value(member_shape, value)738            elif location == "headers":739                header_prefix = key740                self._serialize_header_map(header_prefix, response, value)741            elif location == "statusCode":742                response.status_code = int(value)743    def _serialize_header_map(self, prefix: str, response: HttpResponse, params: dict) -> None:744        """Serializes the header map for the location trait "headers"."""745        for key, val in params.items():746            actual_key = prefix + key747            response.headers[actual_key] = val748    def _serialize_header_value(self, shape: Shape, value: Any):749        """Serializes a value for the location trait "header"."""750        if shape.type_name == "timestamp":751            datetime_obj = parse_to_aware_datetime(value)752            timestamp_format = shape.serialization.get(753                "timestampFormat", self.HEADER_TIMESTAMP_FORMAT754            )755            return self._convert_timestamp_to_str(datetime_obj, timestamp_format)756        elif shape.type_name == "list":757            converted_value = [758                self._serialize_header_value(shape.member, v) for v in value if v is not None759            ]760            return ",".join(converted_value)761        elif shape.type_name == "boolean":762            # Set the header value to "true" if the given value is truthy, otherwise set the header value to "false".763            return "true" if value else "false"764        elif is_json_value_header(shape):765            # Serialize with no spaces after separators to save space in766            # the header.767            return self._get_base64(json.dumps(value, separators=(",", ":")))768        else:769            return value770    def _partition_members(self, parameters: dict, shape: Optional[Shape]) -> Tuple[dict, dict]:771        """Separates the top-level keys in the given parameters dict into header- and payload-located params."""772        if not isinstance(shape, StructureShape):773            # If the shape isn't a structure, we default to the whole response being parsed in the body.774            # Non-payload members are only loated in the top-level hierarchy and those are always structures.775            return {}, parameters776        header_params = {}777        payload_params = {}778        shape_members = shape.members779        for name in shape_members:780            member_shape = shape_members[name]781            if name not in parameters:782                continue783            location = member_shape.serialization.get("location")784            if location:785                header_params[name] = parameters[name]786            else:787                payload_params[name] = parameters[name]788        return header_params, payload_params789class RestXMLResponseSerializer(BaseRestResponseSerializer, BaseXMLResponseSerializer):790    """791    The ``RestXMLResponseSerializer`` is responsible for the serialization of responses from services with the792    ``rest-xml`` protocol.793    It combines the ``BaseRestResponseSerializer`` (for the ReST specific logic) with the ``BaseXMLResponseSerializer``794    (for the XML body response serialization).795    """796    pass797class QueryResponseSerializer(BaseXMLResponseSerializer):798    """799    The ``QueryResponseSerializer`` is responsible for the serialization of responses from services which use the800    ``query`` protocol. The responses of these services also use XML. It is basically a subset of the features, since it801    does not allow any payload or location traits.802    """803    def _serialize_response(804        self,805        parameters: dict,806        response: HttpResponse,807        shape: Optional[Shape],808        shape_members: dict,809        operation_model: OperationModel,810    ) -> None:811        """812        Serializes the given parameters as XML for the query protocol.813        :param parameters: The user input params814        :param response: The final serialized HttpResponse815        :param shape: Describes the expected output shape (can be None in case of an "empty" response)816        :param shape_members: The members of the output struct shape817        :param operation_model: The specification of the operation of which the response is serialized here818        :return: None - the given `serialized` dict is modified819        """820        response.set_response(821            self._encode_payload(self._serialize_body_params(parameters, shape, operation_model))822        )823    def _serialize_body_params_to_xml(824        self, params: dict, shape: Shape, operation_model: OperationModel825    ) -> ETree.Element:826        # The Query protocol responses have a root element which is not contained in the specification file.827        # Therefore, we first call the super function to perform the normal XML serialization, and afterwards wrap the828        # result in a root element based on the operation name.829        node = super()._serialize_body_params_to_xml(params, shape, operation_model)830        # Check if we need to add a namespace831        attr = (832            {"xmlns": operation_model.metadata.get("xmlNamespace")}833            if "xmlNamespace" in operation_model.metadata834            else None835        )836        # Create the root element and add the result of the XML serializer as a child node837        root = ETree.Element(f"{operation_model.name}Response", attr)838        if node is not None:839            root.append(node)840        return root841    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):842        # Add the response metadata here (it's not defined in the specs)843        # For the ec2 and the query protocol, the root cannot be None at this time.844        response_metadata = ETree.SubElement(root, "ResponseMetadata")845        request_id = ETree.SubElement(response_metadata, "RequestId")846        request_id.text = gen_amzn_requestid_long()847class EC2ResponseSerializer(QueryResponseSerializer):848    """849    The ``EC2ResponseSerializer`` is responsible for the serialization of responses from services which use the850    ``ec2`` protocol (basically the EC2 service). This protocol is basically equal to the ``query`` protocol with only851    a few subtle differences.852    """853    def _serialize_error(854        self,855        error: ServiceException,856        response: HttpResponse,857        shape: StructureShape,858        operation_model: OperationModel,859    ) -> None:860        # EC2 errors look like:861        # <Response>862        #   <Errors>863        #     <Error>864        #       <Code>InvalidInstanceID.Malformed</Code>865        #       <Message>Invalid id: "1343124"</Message>866        #     </Error>867        #   </Errors>868        #   <RequestID>12345</RequestID>869        # </Response>870        # This is different from QueryParser in that it's RequestID, not RequestId871        # and that the Error tag is in an enclosing Errors tag.872        attr = (873            {"xmlns": operation_model.metadata.get("xmlNamespace")}874            if "xmlNamespace" in operation_model.metadata875            else None876        )877        root = ETree.Element("Response", attr)878        errors_tag = ETree.SubElement(root, "Errors")879        error_tag = ETree.SubElement(errors_tag, "Error")880        self._add_error_tags(error, error_tag)881        request_id = ETree.SubElement(root, "RequestID")882        request_id.text = gen_amzn_requestid_long()883        response.set_response(self._encode_payload(self._xml_to_string(root)))884    def _prepare_additional_traits_in_xml(self, root: Optional[ETree.Element]):885        # The EC2 protocol does not use the root output shape, therefore we need to remove the hierarchy level886        # below the root level887        if len(root) > 0:888            output_node = root[0]889            for child in output_node:890                root.append(child)891            root.remove(output_node)892        # Add the requestId here (it's not defined in the specs)893        # For the ec2 and the query protocol, the root cannot be None at this time.894        request_id = ETree.SubElement(root, "requestId")895        request_id.text = gen_amzn_requestid_long()896class JSONResponseSerializer(ResponseSerializer):897    """898    The ``JSONResponseSerializer`` is responsible for the serialization of responses from services with the ``json``899    protocol. It implements the JSON response body serialization, which is also used by the900    ``RestJSONResponseSerializer``.901    """902    TIMESTAMP_FORMAT = "unixtimestamp"903    def _serialize_error(904        self,905        error: ServiceException,906        response: HttpResponse,907        shape: StructureShape,908        operation_model: OperationModel,909    ) -> None:910        body = {}911        # TODO implement different service-specific serializer configurations912        #   - currently we set both, the `__type` member as well as the `X-Amzn-Errortype` header913        #   - the specification defines that it's either the __type field OR the header914        response.headers["X-Amzn-Errortype"] = error.code915        body["__type"] = error.code916        if shape:917            remaining_params = {}918            # TODO add a possibility to serialize simple non-modelled errors (like S3 NoSuchBucket#BucketName)919            for member in shape.members:920                if hasattr(error, member):921                    remaining_params[member] = getattr(error, member)922                # Default error message fields can sometimes have different casing in the specs923                elif member.lower() in ["code", "message"] and hasattr(error, member.lower()):924                    remaining_params[member] = getattr(error, member.lower())925            self._serialize(body, remaining_params, shape)926        # Only set the message if it has not been set with the shape members927        if "message" not in body and "Message" not in body:928            message = self._get_error_message(error)929            if message is not None:930                body["message"] = message931        response.set_json(body)932    def _serialize_response(933        self,934        parameters: dict,935        response: HttpResponse,936        shape: Optional[Shape],937        shape_members: dict,938        operation_model: OperationModel,939    ) -> None:940        json_version = operation_model.metadata.get("jsonVersion")941        if json_version is not None:942            response.headers["Content-Type"] = "application/x-amz-json-%s" % json_version943        response.set_response(self._serialize_body_params(parameters, shape, operation_model))944    def _serialize_body_params(945        self, params: dict, shape: Shape, operation_model: OperationModel946    ) -> Optional[str]:947        body = {}948        if shape is not None:949            self._serialize(body, params, shape)950        return json.dumps(body)951    def _serialize(self, body: dict, value: Any, shape, key: Optional[str] = None):952        """This method dynamically invokes the correct `_serialize_type_*` method for each shape type."""953        try:954            method = getattr(self, "_serialize_type_%s" % shape.type_name, self._default_serialize)955            method(body, value, shape, key)956        except (TypeError, ValueError, AttributeError) as e:957            raise ProtocolSerializerError(958                f"Invalid type when serializing {shape.name}: '{value}' cannot be parsed to {shape.type_name}."959            ) from e960    def _serialize_type_structure(self, body: dict, value: dict, shape: StructureShape, key: str):961        if value is None:962            return963        if shape.is_document_type:964            body[key] = value965        else:966            if key is not None:967                # If a key is provided, this is a result of a recursive968                # call so we need to add a new child dict as the value969                # of the passed in serialized dict.  We'll then add970                # all the structure members as key/vals in the new serialized971                # dictionary we just created.972                new_serialized = {}973                body[key] = new_serialized974                body = new_serialized975            members = shape.members976            for member_key, member_value in value.items():977                if member_value is None:978                    continue979                try:980                    member_shape = members[member_key]981                except KeyError:982                    LOG.warning(983                        "Response object %s contains a member which is not specified: %s",984                        shape.name,985                        member_key,986                    )987                    continue988                if "name" in member_shape.serialization:989                    member_key = member_shape.serialization["name"]990                self._serialize(body, member_value, member_shape, member_key)991    def _serialize_type_map(self, body: dict, value: dict, shape: MapShape, key: str):992        if value is None:993            return994        map_obj = {}995        body[key] = map_obj996        for sub_key, sub_value in value.items():997            if sub_value is not None:998                self._serialize(map_obj, sub_value, shape.value, sub_key)999    def _serialize_type_list(self, body: dict, value: list, shape: ListShape, key: str):1000        if value is None:1001            return1002        list_obj = []1003        body[key] = list_obj1004        for list_item in value:1005            if list_item is not None:1006                wrapper = {}1007                # The JSON list serialization is the only case where we aren't1008                # setting a key on a dict.  We handle this by using1009                # a __current__ key on a wrapper dict to serialize each1010                # list item before appending it to the serialized list.1011                self._serialize(wrapper, list_item, shape.member, "__current__")1012                list_obj.append(wrapper["__current__"])1013    def _default_serialize(self, body: dict, value: Any, _, key: str):1014        body[key] = value1015    def _serialize_type_timestamp(self, body: dict, value: Any, shape: Shape, key: str):1016        body[key] = self._convert_timestamp_to_str(1017            value, shape.serialization.get("timestampFormat")1018        )1019    def _serialize_type_blob(self, body: dict, value: Union[str, bytes], _, key: str):1020        body[key] = self._get_base64(value)1021    def _prepare_additional_traits_in_response(1022        self, response: HttpResponse, operation_model: OperationModel1023    ):1024        response.headers["x-amzn-requestid"] = gen_amzn_requestid_long()1025        response = super()._prepare_additional_traits_in_response(response, operation_model)1026        return response1027class RestJSONResponseSerializer(BaseRestResponseSerializer, JSONResponseSerializer):1028    """1029    The ``RestJSONResponseSerializer`` is responsible for the serialization of responses from services with the1030    ``rest-json`` protocol.1031    It combines the ``BaseRestResponseSerializer`` (for the ReST specific logic) with the ``JSONResponseSerializer``1032    (for the JSOn body response serialization).1033    """1034    def _serialize_content_type(self, serialized: HttpResponse, shape: Shape, shape_members: dict):1035        """Set Content-Type to application/json for all structured bodies."""1036        payload = shape.serialization.get("payload") if shape is not None else None1037        if self._has_streaming_payload(payload, shape_members):1038            # Don't apply content-type to streaming bodies1039            return1040        has_body = serialized.data != b""1041        has_content_type = self._has_header("Content-Type", serialized.headers)1042        if has_body and not has_content_type:1043            serialized.headers["Content-Type"] = "application/json"1044class S3ResponseSerializer(RestXMLResponseSerializer):1045    """1046    The ``S3ResponseSerializer`` adds some minor logic to handle S3 specific peculiarities with the error response1047    serialization.1048    """1049    def _serialize_error(1050        self,1051        error: ServiceException,1052        response: HttpResponse,1053        shape: StructureShape,1054        operation_model: OperationModel,1055    ) -> None:1056        attr = (1057            {"xmlns": operation_model.metadata.get("xmlNamespace")}1058            if "xmlNamespace" in operation_model.metadata1059            else {}1060        )1061        root = ETree.Element("Error", attr)1062        self._add_error_tags(error, root)1063        request_id = ETree.SubElement(root, "RequestId")1064        request_id.text = gen_amzn_requestid_long()1065        self._add_additional_error_tags(error, root, shape)1066        response.set_response(self._encode_payload(self._xml_to_string(root)))1067class SqsResponseSerializer(QueryResponseSerializer):1068    """1069    Unfortunately, SQS uses a rare interpretation of the XML protocol: It uses HTML entities within XML tag text nodes.1070    For example:1071    - Normal XML serializers: <Message>No need to escape quotes (like this: ") with HTML entities in XML.</Message>1072    - SQS XML serializer: <Message>No need to escape quotes (like this: &quot;) with HTML entities in XML.</Message>1073    None of the prominent XML frameworks for python allow HTML entity escapes when serializing XML.1074    This serializer implements the following workaround:1075    - Escape quotes and \r with their HTML entities (&quot; and &#xD;).1076    - Since & is (correctly) escaped in XML, the serialized string contains &amp;quot; and &amp;#xD;1077    - These double-escapes are corrected by replacing such strings with their original.1078    """1079    def _default_serialize(self, xmlnode: ETree.Element, params: str, _, name: str) -> 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.