KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _parse_shape method in localstack

Best Python code snippet using localstack_python

parsers.py

Source:parsers.py

...246            "%s._do_error_parse" % self.__class__.__name__)247    def _do_modeled_error_parse(self, response, shape, parsed):248        raise NotImplementedError(249            "%s._do_modeled_error_parse" % self.__class__.__name__)250    def _parse_shape(self, shape, node):251        handler = getattr(self, '_handle_%s' % shape.type_name,252                          self._default_handle)253        return handler(shape, node)254    def _handle_list(self, shape, node):255        # Enough implementations share list serialization that it's moved256        # up here in the base class.257        parsed = []258        member_shape = shape.member259        for item in node:260            parsed.append(self._parse_shape(member_shape, item))261        return parsed262    def _default_handle(self, shape, value):263        return value264    def _create_event_stream(self, response, shape):265        parser = self._event_stream_parser266        name = response['context'].get('operation_name')267        return EventStream(response['body'], shape, parser, name)268    def _get_first_key(self, value):269        return list(value)[0]270    def _has_unknown_tagged_union_member(self, shape, value):271        if shape.is_tagged_union:272            if len(value) != 1:273                error_msg = (274                    "Invalid service response: %s must have one and only "275                    "one member set."276                )277                raise ResponseParserError(error_msg % shape.name)278            tag = self._get_first_key(value)279            if tag not in shape.members:280                msg = (281                    "Received a tagged union response with member "282                    "unknown to client: %s. Please upgrade SDK for full "283                    "response support."284                )285                LOG.info(msg % tag)286                return True287        return False288    def _handle_unknown_tagged_union_member(self, tag):289        return {'SDK_UNKNOWN_MEMBER': {'name': tag}}290class BaseXMLResponseParser(ResponseParser):291    def __init__(self, timestamp_parser=None, blob_parser=None):292        super(BaseXMLResponseParser, self).__init__(timestamp_parser,293                                                    blob_parser)294        self._namespace_re = re.compile('{.*}')295    def _handle_map(self, shape, node):296        parsed = {}297        key_shape = shape.key298        value_shape = shape.value299        key_location_name = key_shape.serialization.get('name') or 'key'300        value_location_name = value_shape.serialization.get('name') or 'value'301        if shape.serialization.get('flattened') and not isinstance(node, list):302            node = [node]303        for keyval_node in node:304            for single_pair in keyval_node:305                # Within each <entry> there's a <key> and a <value>306                tag_name = self._node_tag(single_pair)307                if tag_name == key_location_name:308                    key_name = self._parse_shape(key_shape, single_pair)309                elif tag_name == value_location_name:310                    val_name = self._parse_shape(value_shape, single_pair)311                else:312                    raise ResponseParserError("Unknown tag: %s" % tag_name)313            parsed[key_name] = val_name314        return parsed315    def _node_tag(self, node):316        return self._namespace_re.sub('', node.tag)317    def _handle_list(self, shape, node):318        # When we use _build_name_to_xml_node, repeated elements are aggregated319        # into a list.  However, we can't tell the difference between a scalar320        # value and a single element flattened list.  So before calling the321        # real _handle_list, we know that "node" should actually be a list if322        # it's flattened, and if it's not, then we make it a one element list.323        if shape.serialization.get('flattened') and not isinstance(node, list):324            node = [node]325        return super(BaseXMLResponseParser, self)._handle_list(shape, node)326    def _handle_structure(self, shape, node):327        parsed = {}328        members = shape.members329        if shape.metadata.get('exception', False):330            node = self._get_error_root(node)331        xml_dict = self._build_name_to_xml_node(node)332        if self._has_unknown_tagged_union_member(shape, xml_dict):333            tag = self._get_first_key(xml_dict)334            return self._handle_unknown_tagged_union_member(tag)335        for member_name in members:336            member_shape = members[member_name]337            if 'location' in member_shape.serialization or \338               member_shape.serialization.get('eventheader'):339                # All members with locations have already been handled,340                # so we don't need to parse these members.341                continue342            xml_name = self._member_key_name(member_shape, member_name)343            member_node = xml_dict.get(xml_name)344            if member_node is not None:345                parsed[member_name] = self._parse_shape(346                    member_shape, member_node)347            elif member_shape.serialization.get('xmlAttribute'):348                attribs = {}349                location_name = member_shape.serialization['name']350                for key, value in node.attrib.items():351                    new_key = self._namespace_re.sub(352                        location_name.split(':')[0] + ':', key)353                    attribs[new_key] = value354                if location_name in attribs:355                    parsed[member_name] = attribs[location_name]356        return parsed357    def _get_error_root(self, original_root):358        if self._node_tag(original_root) == 'ErrorResponse':359            for child in original_root:360                if self._node_tag(child) == 'Error':361                    return child362        return original_root363    def _member_key_name(self, shape, member_name):364        # This method is needed because we have to special case flattened list365        # with a serialization name.  If this is the case we use the366        # locationName from the list's member shape as the key name for the367        # surrounding structure.368        if shape.type_name == 'list' and shape.serialization.get('flattened'):369            list_member_serialized_name = shape.member.serialization.get(370                'name')371            if list_member_serialized_name is not None:372                return list_member_serialized_name373        serialized_name = shape.serialization.get('name')374        if serialized_name is not None:375            return serialized_name376        return member_name377    def _build_name_to_xml_node(self, parent_node):378        # If the parent node is actually a list. We should not be trying379        # to serialize it to a dictionary. Instead, return the first element380        # in the list.381        if isinstance(parent_node, list):382            return self._build_name_to_xml_node(parent_node[0])383        xml_dict = {}384        for item in parent_node:385            key = self._node_tag(item)386            if key in xml_dict:387                # If the key already exists, the most natural388                # way to handle this is to aggregate repeated389                # keys into a single list.390                # <foo>1</foo><foo>2</foo> -> {'foo': [Node(1), Node(2)]}391                if isinstance(xml_dict[key], list):392                    xml_dict[key].append(item)393                else:394                    # Convert from a scalar to a list.395                    xml_dict[key] = [xml_dict[key], item]396            else:397                xml_dict[key] = item398        return xml_dict399    def _parse_xml_string_to_dom(self, xml_string):400        try:401            parser = ETree.XMLParser(402                target=ETree.TreeBuilder(),403                encoding=self.DEFAULT_ENCODING)404            parser.feed(xml_string)405            root = parser.close()406        except XMLParseError as e:407            raise ResponseParserError(408                "Unable to parse response (%s), "409                "invalid XML received. Further retries may succeed:\n%s" %410                (e, xml_string))411        return root412    def _replace_nodes(self, parsed):413        for key, value in parsed.items():414            if list(value):415                sub_dict = self._build_name_to_xml_node(value)416                parsed[key] = self._replace_nodes(sub_dict)417            else:418                parsed[key] = value.text419        return parsed420    @_text_content421    def _handle_boolean(self, shape, text):422        if text == 'true':423            return True424        else:425            return False426    @_text_content427    def _handle_float(self, shape, text):428        return float(text)429    @_text_content430    def _handle_timestamp(self, shape, text):431        return self._timestamp_parser(text)432    @_text_content433    def _handle_integer(self, shape, text):434        return int(text)435    @_text_content436    def _handle_string(self, shape, text):437        return text438    @_text_content439    def _handle_blob(self, shape, text):440        return self._blob_parser(text)441    _handle_character = _handle_string442    _handle_double = _handle_float443    _handle_long = _handle_integer444class QueryParser(BaseXMLResponseParser):445    def _do_error_parse(self, response, shape):446        xml_contents = response['body']447        root = self._parse_xml_string_to_dom(xml_contents)448        parsed = self._build_name_to_xml_node(root)449        self._replace_nodes(parsed)450        # Once we've converted xml->dict, we need to make one or two451        # more adjustments to extract nested errors and to be consistent452        # with ResponseMetadata for non-error responses:453        # 1. {"Errors": {"Error": {...}}} -> {"Error": {...}}454        # 2. {"RequestId": "id"} -> {"ResponseMetadata": {"RequestId": "id"}}455        if 'Errors' in parsed:456            parsed.update(parsed.pop('Errors'))457        if 'RequestId' in parsed:458            parsed['ResponseMetadata'] = {'RequestId': parsed.pop('RequestId')}459        return parsed460    def _do_modeled_error_parse(self, response, shape):461        return self._parse_body_as_xml(response, shape, inject_metadata=False)462    def _do_parse(self, response, shape):463        return self._parse_body_as_xml(response, shape, inject_metadata=True)464    def _parse_body_as_xml(self, response, shape, inject_metadata=True):465        xml_contents = response['body']466        root = self._parse_xml_string_to_dom(xml_contents)467        parsed = {}468        if shape is not None:469            start = root470            if 'resultWrapper' in shape.serialization:471                start = self._find_result_wrapped_shape(472                    shape.serialization['resultWrapper'],473                    root)474            parsed = self._parse_shape(shape, start)475        if inject_metadata:476            self._inject_response_metadata(root, parsed)477        return parsed478    def _find_result_wrapped_shape(self, element_name, xml_root_node):479        mapping = self._build_name_to_xml_node(xml_root_node)480        return mapping[element_name]481    def _inject_response_metadata(self, node, inject_into):482        mapping = self._build_name_to_xml_node(node)483        child_node = mapping.get('ResponseMetadata')484        if child_node is not None:485            sub_mapping = self._build_name_to_xml_node(child_node)486            for key, value in sub_mapping.items():487                sub_mapping[key] = value.text488            inject_into['ResponseMetadata'] = sub_mapping489class EC2QueryParser(QueryParser):490    def _inject_response_metadata(self, node, inject_into):491        mapping = self._build_name_to_xml_node(node)492        child_node = mapping.get('requestId')493        if child_node is not None:494            inject_into['ResponseMetadata'] = {'RequestId': child_node.text}495    def _do_error_parse(self, response, shape):496        # EC2 errors look like:497        # <Response>498        #   <Errors>499        #     <Error>500        #       <Code>InvalidInstanceID.Malformed</Code>501        #       <Message>Invalid id: "1343124"</Message>502        #     </Error>503        #   </Errors>504        #   <RequestID>12345</RequestID>505        # </Response>506        # This is different from QueryParser in that it's RequestID,507        # not RequestId508        original = super(EC2QueryParser, self)._do_error_parse(response, shape)509        if 'RequestID' in original:510            original['ResponseMetadata'] = {511                'RequestId': original.pop('RequestID')512            }513        return original514    def _get_error_root(self, original_root):515        for child in original_root:516            if self._node_tag(child) == 'Errors':517                for errors_child in child:518                    if self._node_tag(errors_child) == 'Error':519                        return errors_child520        return original_root521class BaseJSONParser(ResponseParser):522    def _handle_structure(self, shape, value):523        final_parsed = {}524        if shape.is_document_type:525            final_parsed = value526        else:527            member_shapes = shape.members528            if value is None:529                # If the comes across the wire as "null" (None in python),530                # we should be returning this unchanged, instead of as an531                # empty dict.532                return None533            final_parsed = {}534            if self._has_unknown_tagged_union_member(shape, value):535                tag = self._get_first_key(value)536                return self._handle_unknown_tagged_union_member(tag)537            for member_name in member_shapes:538                member_shape = member_shapes[member_name]539                json_name = member_shape.serialization.get('name', member_name)540                raw_value = value.get(json_name)541                if raw_value is not None:542                    final_parsed[member_name] = self._parse_shape(543                        member_shapes[member_name],544                        raw_value)545        return final_parsed546    def _handle_map(self, shape, value):547        parsed = {}548        key_shape = shape.key549        value_shape = shape.value550        for key, value in value.items():551            actual_key = self._parse_shape(key_shape, key)552            actual_value = self._parse_shape(value_shape, value)553            parsed[actual_key] = actual_value554        return parsed555    def _handle_blob(self, shape, value):556        return self._blob_parser(value)557    def _handle_timestamp(self, shape, value):558        return self._timestamp_parser(value)559    def _do_error_parse(self, response, shape):560        body = self._parse_body_as_json(response['body'])561        error = {"Error": {"Message": '', "Code": ''}, "ResponseMetadata": {}}562        # Error responses can have slightly different structures for json.563        # The basic structure is:564        #565        # {"__type":"ConnectClientException",566        #  "message":"The error message."}567        # The error message can either come in the 'message' or 'Message' key568        # so we need to check for both.569        error['Error']['Message'] = body.get('message',570                                             body.get('Message', ''))571        # if the message did not contain an error code572        # include the response status code573        response_code = response.get('status_code')574        code = body.get('__type', response_code and str(response_code))575        if code is not None:576            # code has a couple forms as well:577            # * "com.aws.dynamodb.vAPI#ProvisionedThroughputExceededException"578            # * "ResourceNotFoundException"579            if '#' in code:580                code = code.rsplit('#', 1)[1]581            error['Error']['Code'] = code582        self._inject_response_metadata(error, response['headers'])583        return error584    def _inject_response_metadata(self, parsed, headers):585        if 'x-amzn-requestid' in headers:586            parsed.setdefault('ResponseMetadata', {})['RequestId'] = (587                headers['x-amzn-requestid'])588    def _parse_body_as_json(self, body_contents):589        if not body_contents:590            return {}591        body = body_contents.decode(self.DEFAULT_ENCODING)592        try:593            original_parsed = json.loads(body)594            return original_parsed595        except ValueError:596            # if the body cannot be parsed, include597            # the literal string as the message598            return {'message': body}599class BaseEventStreamParser(ResponseParser):600    def _do_parse(self, response, shape):601        final_parsed = {}602        if shape.serialization.get('eventstream'):603            event_type = response['headers'].get(':event-type')604            event_shape = shape.members.get(event_type)605            if event_shape:606                final_parsed[event_type] = self._do_parse(response, event_shape)607        else:608            self._parse_non_payload_attrs(response, shape,609                                          shape.members, final_parsed)610            self._parse_payload(response, shape, shape.members, final_parsed)611        return final_parsed612    def _do_error_parse(self, response, shape):613        exception_type = response['headers'].get(':exception-type')614        exception_shape = shape.members.get(exception_type)615        if exception_shape is not None:616            original_parsed = self._initial_body_parse(response['body'])617            body = self._parse_shape(exception_shape, original_parsed)618            error = {619                'Error': {620                    'Code': exception_type,621                    'Message': body.get('Message', body.get('message', ''))622                }623            }624        else:625            error = {626                'Error': {627                    'Code': response['headers'].get(':error-code', ''),628                    'Message': response['headers'].get(':error-message', ''),629                }630            }631        return error632    def _parse_payload(self, response, shape, member_shapes, final_parsed):633        if shape.serialization.get('event'):634            for name in member_shapes:635                member_shape = member_shapes[name]636                if member_shape.serialization.get('eventpayload'):637                    body = response['body']638                    if member_shape.type_name == 'blob':639                        parsed_body = body640                    elif member_shape.type_name == 'string':641                        parsed_body = body.decode(self.DEFAULT_ENCODING)642                    else:643                        raw_parse = self._initial_body_parse(body)644                        parsed_body = self._parse_shape(member_shape, raw_parse)645                    final_parsed[name] = parsed_body646                    return647            # If we didn't find an explicit payload, use the current shape648            original_parsed = self._initial_body_parse(response['body'])649            body_parsed = self._parse_shape(shape, original_parsed)650            final_parsed.update(body_parsed)651    def _parse_non_payload_attrs(self, response, shape,652                                 member_shapes, final_parsed):653        headers = response['headers']654        for name in member_shapes:655            member_shape = member_shapes[name]656            if member_shape.serialization.get('eventheader'):657                if name in headers:658                    value = headers[name]659                    if member_shape.type_name == 'timestamp':660                        # Event stream timestamps are an in milleseconds so we661                        # divide by 1000 to convert to seconds.662                        value = self._timestamp_parser(value / 1000.0)663                    final_parsed[name] = value664    def _initial_body_parse(self, body_contents):665        # This method should do the initial xml/json parsing of the666        # body.  We we still need to walk the parsed body in order667        # to convert types, but this method will do the first round668        # of parsing.669        raise NotImplementedError("_initial_body_parse")670class EventStreamJSONParser(BaseEventStreamParser, BaseJSONParser):671    def _initial_body_parse(self, body_contents):672        return self._parse_body_as_json(body_contents)673class EventStreamXMLParser(BaseEventStreamParser, BaseXMLResponseParser):674    def _initial_body_parse(self, xml_string):675        if not xml_string:676            return ETree.Element('')677        return self._parse_xml_string_to_dom(xml_string)678class JSONParser(BaseJSONParser):679    EVENT_STREAM_PARSER_CLS = EventStreamJSONParser680    """Response parser for the "json" protocol."""681    def _do_parse(self, response, shape):682        parsed = {}683        if shape is not None:684            event_name = shape.event_stream_name685            if event_name:686                parsed = self._handle_event_stream(response, shape, event_name)687            else:688                parsed = self._handle_json_body(response['body'], shape)689        self._inject_response_metadata(parsed, response['headers'])690        return parsed691    def _do_modeled_error_parse(self, response, shape):692        return self._handle_json_body(response['body'], shape)693    def _handle_event_stream(self, response, shape, event_name):694        event_stream_shape = shape.members[event_name]695        event_stream = self._create_event_stream(response, event_stream_shape)696        try:697            event = event_stream.get_initial_response()698        except NoInitialResponseError:699            error_msg = 'First event was not of type initial-response'700            raise ResponseParserError(error_msg)701        parsed = self._handle_json_body(event.payload, shape)702        parsed[event_name] = event_stream703        return parsed704    def _handle_json_body(self, raw_body, shape):705        # The json.loads() gives us the primitive JSON types,706        # but we need to traverse the parsed JSON data to convert707        # to richer types (blobs, timestamps, etc.708        parsed_json = self._parse_body_as_json(raw_body)709        return self._parse_shape(shape, parsed_json)710class BaseRestParser(ResponseParser):711    def _do_parse(self, response, shape):712        final_parsed = {}713        final_parsed['ResponseMetadata'] = self._populate_response_metadata(714            response)715        self._add_modeled_parse(response, shape, final_parsed)716        return final_parsed717    def _add_modeled_parse(self, response, shape, final_parsed):718        if shape is None:719            return final_parsed720        member_shapes = shape.members721        self._parse_non_payload_attrs(response, shape,722                                      member_shapes, final_parsed)723        self._parse_payload(response, shape, member_shapes, final_parsed)724    def _do_modeled_error_parse(self, response, shape):725        final_parsed = {}726        self._add_modeled_parse(response, shape, final_parsed)727        return final_parsed728    def _populate_response_metadata(self, response):729        metadata = {}730        headers = response['headers']731        if 'x-amzn-requestid' in headers:732            metadata['RequestId'] = headers['x-amzn-requestid']733        elif 'x-amz-request-id' in headers:734            metadata['RequestId'] = headers['x-amz-request-id']735            # HostId is what it's called whenever this value is returned736            # in an XML response body, so to be consistent, we'll always737            # call is HostId.738            metadata['HostId'] = headers.get('x-amz-id-2', '')739        return metadata740    def _parse_payload(self, response, shape, member_shapes, final_parsed):741        if 'payload' in shape.serialization:742            # If a payload is specified in the output shape, then only that743            # shape is used for the body payload.744            payload_member_name = shape.serialization['payload']745            body_shape = member_shapes[payload_member_name]746            if body_shape.serialization.get('eventstream'):747                body = self._create_event_stream(response, body_shape)748                final_parsed[payload_member_name] = body749            elif body_shape.type_name in ['string', 'blob']:750                # This is a stream751                body = response['body']752                if isinstance(body, bytes):753                    body = body.decode(self.DEFAULT_ENCODING)754                final_parsed[payload_member_name] = body755            else:756                original_parsed = self._initial_body_parse(response['body'])757                final_parsed[payload_member_name] = self._parse_shape(758                    body_shape, original_parsed)759        else:760            original_parsed = self._initial_body_parse(response['body'])761            body_parsed = self._parse_shape(shape, original_parsed)762            final_parsed.update(body_parsed)763    def _parse_non_payload_attrs(self, response, shape,764                                 member_shapes, final_parsed):765        headers = response['headers']766        for name in member_shapes:767            member_shape = member_shapes[name]768            location = member_shape.serialization.get('location')769            if location is None:770                continue771            elif location == 'statusCode':772                final_parsed[name] = self._parse_shape(773                    member_shape, response['status_code'])774            elif location == 'headers':775                final_parsed[name] = self._parse_header_map(member_shape,776                                                            headers)777            elif location == 'header':778                header_name = member_shape.serialization.get('name', name)779                if header_name in headers:780                    final_parsed[name] = self._parse_shape(781                        member_shape, headers[header_name])782    def _parse_header_map(self, shape, headers):783        # Note that headers are case insensitive, so we .lower()784        # all header names and header prefixes.785        parsed = {}786        prefix = shape.serialization.get('name', '').lower()787        for header_name in headers:788            if header_name.lower().startswith(prefix):789                # The key name inserted into the parsed hash790                # strips off the prefix.791                name = header_name[len(prefix):]792                parsed[name] = headers[header_name]793        return parsed794    def _initial_body_parse(self, body_contents):...

shape_parser.py

Source:shape_parser.py

...8    fake response (dummy response) JSON object.9    """10    def parse(self, shape):11        """Start the parsing process on received shape."""12        dummy_response = self._parse_shape(shape)13        return dummy_response14    def _parse_shape(self, shape):15        shape_type = shape.type_name16        if shape_type == 'list':17            return self._handle_list(shape)18        if shape_type == 'structure':19            return self._handle_structure(shape)20        if shape_type == 'map':21            return self._handle_map(shape)22        return self._default_handle(shape)23    def _default_handle(self, shape):24        # Return shape type as dummy value of scalar type.25        return '%s:%s' % (shape.name, shape.type_name)26    def _handle_list(self, shape):27        member_shape = shape.member28        parsed = []29        parsed.append(self._parse_shape(member_shape))30        return parsed31    def _handle_structure(self, shape):32        parsed = {}33        member_shapes = shape.members34        for name, member in member_shapes.items():35            parsed[name] = self._parse_shape(member)36        return parsed37    def _handle_map(self, shape):38        parsed = {}39        key_shape = shape.key40        value_shape = shape.value41        key = self._parse_shape(key_shape)42        value = self._parse_shape(value_shape)43        parsed[key] = value...

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.