KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _timestamp_rfc822 method in localstack

Best Python code snippet using localstack_python
serialize.py
Source:serialize.py
...110            timestamp_format = ISO8601111        return value.strftime(timestamp_format)112    def _timestamp_unixtimestamp(self, value):113        return int(calendar.timegm(value.timetuple()))114    def _timestamp_rfc822(self, value):115        return formatdate(value, usegmt=True)116    def _convert_timestamp_to_str(self, value):117        datetime_obj = parse_to_aware_datetime(value)118        converter = getattr(119            self, '_timestamp_%s' % self.TIMESTAMP_FORMAT.lower())120        final_value = converter(datetime_obj)121        return final_value122    def _get_serialized_name(self, shape, default_name):123        # Returns the serialized name for the shape if it exists.124        # Otherwise it will return the passed in default_name.125        return shape.serialization.get('name', default_name)126    def _get_base64(self, value):127        # Returns the base64-encoded version of value, handling128        # both strings and bytes. The returned value is a string129        # via the default encoding.130        if isinstance(value, six.text_type):131            value = value.encode(self.DEFAULT_ENCODING)132        return base64.b64encode(value).strip().decode(133            self.DEFAULT_ENCODING)134class QuerySerializer(Serializer):135    TIMESTAMP_FORMAT = 'iso8601'136    def serialize_to_request(self, parameters, operation_model):137        shape = operation_model.input_shape138        serialized = self._create_default_request()139        serialized['method'] = operation_model.http.get('method',140                                                        self.DEFAULT_METHOD)141        # The query serializer only deals with body params so142        # that's what we hand off the _serialize_* methods.143        body_params = self.MAP_TYPE()144        body_params['Action'] = operation_model.name145        body_params['Version'] = operation_model.metadata['apiVersion']146        if shape is not None:147            self._serialize(body_params, parameters, shape)148        serialized['body'] = body_params149        return serialized150    def _serialize(self, serialized, value, shape, prefix=''):151        # serialized: The dict that is incrementally added to with the152        #             final serialized parameters.153        # value: The current user input value.154        # shape: The shape object that describes the structure of the155        #        input.156        # prefix: The incrementally built up prefix for the serialized157        #         key (i.e Foo.bar.members.1).158        method = getattr(self, '_serialize_type_%s' % shape.type_name,159                         self._default_serialize)160        method(serialized, value, shape, prefix=prefix)161    def _serialize_type_structure(self, serialized, value, shape, prefix=''):162        members = shape.members163        for key, value in value.items():164            member_shape = members[key]165            member_prefix = self._get_serialized_name(member_shape, key)166            if prefix:167                member_prefix = '%s.%s' % (prefix, member_prefix)168            self._serialize(serialized, value, member_shape, member_prefix)169    def _serialize_type_list(self, serialized, value, shape, prefix=''):170        if not value:171            # The query protocol serializes empty lists.172            serialized[prefix] = ''173            return174        if self._is_shape_flattened(shape):175            list_prefix = prefix176            if shape.member.serialization.get('name'):177                name = self._get_serialized_name(shape.member, default_name='')178                # Replace '.Original' with '.{name}'.179                list_prefix = '.'.join(prefix.split('.')[:-1] + [name])180        else:181            list_name = shape.member.serialization.get('name', 'member')182            list_prefix = '%s.%s' % (prefix, list_name)183        for i, element in enumerate(value, 1):184            element_prefix = '%s.%s' % (list_prefix, i)185            element_shape = shape.member186            self._serialize(serialized, element, element_shape, element_prefix)187    def _serialize_type_map(self, serialized, value, shape, prefix=''):188        if self._is_shape_flattened(shape):189            full_prefix = prefix190        else:191            full_prefix = '%s.entry' % prefix192        template = full_prefix + '.{i}.{suffix}'193        key_shape = shape.key194        value_shape = shape.value195        key_suffix = self._get_serialized_name(key_shape, default_name='key')196        value_suffix = self._get_serialized_name(value_shape, 'value')197        for i, key in enumerate(value, 1):198            key_prefix = template.format(i=i, suffix=key_suffix)199            value_prefix = template.format(i=i, suffix=value_suffix)200            self._serialize(serialized, key, key_shape, key_prefix)201            self._serialize(serialized, value[key], value_shape, value_prefix)202    def _serialize_type_blob(self, serialized, value, shape, prefix=''):203        # Blob args must be base64 encoded.204        serialized[prefix] = self._get_base64(value)205    def _serialize_type_timestamp(self, serialized, value, shape, prefix=''):206        serialized[prefix] = self._convert_timestamp_to_str(value)207    def _serialize_type_boolean(self, serialized, value, shape, prefix=''):208        if value:209            serialized[prefix] = 'true'210        else:211            serialized[prefix] = 'false'212    def _default_serialize(self, serialized, value, shape, prefix=''):213        serialized[prefix] = value214    def _is_shape_flattened(self, shape):215        return shape.serialization.get('flattened')216class EC2Serializer(QuerySerializer):217    """EC2 specific customizations to the query protocol serializers.218    The EC2 model is almost, but not exactly, similar to the query protocol219    serializer.  This class encapsulates those differences.  The model220    will have be marked with a ``protocol`` of ``ec2``, so you don't need221    to worry about wiring this class up correctly.222    """223    def _get_serialized_name(self, shape, default_name):224        # Returns the serialized name for the shape if it exists.225        # Otherwise it will return the passed in default_name.226        if 'queryName' in shape.serialization:227            return shape.serialization['queryName']228        elif 'name' in shape.serialization:229            # A locationName is always capitalized230            # on input for the ec2 protocol.231            name = shape.serialization['name']232            return name[0].upper() + name[1:]233        else:234            return default_name235    def _serialize_type_list(self, serialized, value, shape, prefix=''):236        for i, element in enumerate(value, 1):237            element_prefix = '%s.%s' % (prefix, i)238            element_shape = shape.member239            self._serialize(serialized, element, element_shape, element_prefix)240class JSONSerializer(Serializer):241    TIMESTAMP_FORMAT = 'unixtimestamp'242    def serialize_to_request(self, parameters, operation_model):243        target = '%s.%s' % (operation_model.metadata['targetPrefix'],244                            operation_model.name)245        json_version = operation_model.metadata['jsonVersion']246        serialized = self._create_default_request()247        serialized['method'] = operation_model.http.get('method',248                                                        self.DEFAULT_METHOD)249        serialized['headers'] = {250            'X-Amz-Target': target,251            'Content-Type': 'application/x-amz-json-%s' % json_version,252        }253        body = {}254        input_shape = operation_model.input_shape255        if input_shape is not None:256            self._serialize(body, parameters, input_shape)257        serialized['body'] = json.dumps(body).encode(self.DEFAULT_ENCODING)258        return serialized259    def _serialize(self, serialized, value, shape, key=None):260        method = getattr(self, '_serialize_type_%s' % shape.type_name,261                         self._default_serialize)262        method(serialized, value, shape, key)263    def _serialize_type_structure(self, serialized, value, shape, key):264        if key is not None:265            # If a key is provided, this is a result of a recursive266            # call so we need to add a new child dict as the value267            # of the passed in serialized dict.  We'll then add268            # all the structure members as key/vals in the new serialized269            # dictionary we just created.270            new_serialized = self.MAP_TYPE()271            serialized[key] = new_serialized272            serialized = new_serialized273        members = shape.members274        for member_key, member_value in value.items():275            member_shape = members[member_key]276            if 'name' in member_shape.serialization:277                member_key = member_shape.serialization['name']278            self._serialize(serialized, member_value, member_shape, member_key)279    def _serialize_type_map(self, serialized, value, shape, key):280        map_obj = self.MAP_TYPE()281        serialized[key] = map_obj282        for sub_key, sub_value in value.items():283            self._serialize(map_obj, sub_value, shape.value, sub_key)284    def _serialize_type_list(self, serialized, value, shape, key):285        list_obj = []286        serialized[key] = list_obj287        for list_item in value:288            wrapper = {}289            # The JSON list serialization is the only case where we aren't290            # setting a key on a dict.  We handle this by using291            # a __current__ key on a wrapper dict to serialize each292            # list item before appending it to the serialized list.293            self._serialize(wrapper, list_item, shape.member, "__current__")294            list_obj.append(wrapper["__current__"])295    def _default_serialize(self, serialized, value, shape, key):296        serialized[key] = value297    def _serialize_type_timestamp(self, serialized, value, shape, key):298        serialized[key] = self._convert_timestamp_to_str(value)299    def _serialize_type_blob(self, serialized, value, shape, key):300        serialized[key] = self._get_base64(value)301class BaseRestSerializer(Serializer):302    """Base class for rest protocols.303    The only variance between the various rest protocols is the304    way that the body is serialized.  All other aspects (headers, uri, etc.)305    are the same and logic for serializing those aspects lives here.306    Subclasses must implement the ``_serialize_body_params`` method.307    """308    # This is a list of known values for the "location" key in the309    # serialization dict.  The location key tells us where on the request310    # to put the serialized value.311    KNOWN_LOCATIONS = ['uri', 'querystring', 'header', 'headers']312    def serialize_to_request(self, parameters, operation_model):313        serialized = self._create_default_request()314        serialized['method'] = operation_model.http.get('method',315                                                        self.DEFAULT_METHOD)316        shape = operation_model.input_shape317        if shape is None:318            serialized['url_path'] = operation_model.http['requestUri']319            return serialized320        shape_members = shape.members321        # While the ``serialized`` key holds the final serialized request322        # data, we need interim dicts for the various locations of the323        # request.  We need this for the uri_path_kwargs and the324        # query_string_kwargs because they are templated, so we need325        # to gather all the needed data for the string template,326        # then we render the template.  The body_kwargs is needed327        # because once we've collected them all, we run them through328        # _serialize_body_params, which for rest-json, creates JSON,329        # and for rest-xml, will create XML.  This is what the330        # ``partitioned`` dict below is for.331        partitioned = {332            'uri_path_kwargs': self.MAP_TYPE(),333            'query_string_kwargs': self.MAP_TYPE(),334            'body_kwargs': self.MAP_TYPE(),335            'headers': self.MAP_TYPE(),336        }337        for param_name, param_value in parameters.items():338            if param_value is None:339                # Don't serialize any parameter with a None value.340                continue341            self._partition_parameters(partitioned, param_name, param_value,342                                       shape_members)343        serialized['url_path'] = self._render_uri_template(344            operation_model.http['requestUri'],345            partitioned['uri_path_kwargs'])346        # Note that we lean on the http implementation to handle the case347        # where the requestUri path already has query parameters.348        # The bundled http client, requests, already supports this.349        serialized['query_string'] = partitioned['query_string_kwargs']350        if partitioned['headers']:351            serialized['headers'] = partitioned['headers']352        self._serialize_payload(partitioned, parameters,353                                serialized, shape, shape_members)354        return serialized355    def _render_uri_template(self, uri_template, params):356        # We need to handle two cases::357        #358        # /{Bucket}/foo359        # /{Key+}/bar360        # A label ending with '+' is greedy.  There can only361        # be one greedy key.362        encoded_params = {}363        for template_param in re.findall(r'{(.*?)}', uri_template):364            if template_param.endswith('+'):365                encoded_params[template_param] = percent_encode(366                    params[template_param[:-1]], safe='/~')367            else:368                encoded_params[template_param] = percent_encode(369                    params[template_param])370        return uri_template.format(**encoded_params)371    def _serialize_payload(self, partitioned, parameters,372                           serialized, shape, shape_members):373        # partitioned - The user input params partitioned by location.374        # parameters - The user input params.375        # serialized - The final serialized request dict.376        # shape - Describes the expected input shape377        # shape_members - The members of the input struct shape378        payload_member = shape.serialization.get('payload')379        if payload_member is not None and \380                shape_members[payload_member].type_name in ['blob', 'string']:381            # If it's streaming, then the body is just the382            # value of the payload.383            body_payload = parameters.get(payload_member, b'')384            body_payload = self._encode_payload(body_payload)385            serialized['body'] = body_payload386        elif payload_member is not None:387            # If there's a payload member, we serialized that388            # member to they body.389            body_params = parameters.get(payload_member)390            if body_params is not None:391                serialized['body'] = self._serialize_body_params(392                    body_params,393                    shape_members[payload_member])394        elif partitioned['body_kwargs']:395            serialized['body'] = self._serialize_body_params(396                partitioned['body_kwargs'], shape)397    def _encode_payload(self, body):398        if isinstance(body, six.text_type):399            return body.encode(self.DEFAULT_ENCODING)400        return body401    def _partition_parameters(self, partitioned, param_name,402                              param_value, shape_members):403        # This takes the user provided input parameter (``param``)404        # and figures out where they go in the request dict.405        # Some params are HTTP headers, some are used in the URI, some406        # are in the request body.  This method deals with this.407        member = shape_members[param_name]408        location = member.serialization.get('location')409        key_name = member.serialization.get('name', param_name)410        if location == 'uri':411            partitioned['uri_path_kwargs'][key_name] = param_value412        elif location == 'querystring':413            if isinstance(param_value, dict):414                partitioned['query_string_kwargs'].update(param_value)415            elif isinstance(param_value, bool):416                partitioned['query_string_kwargs'][417                    key_name] = str(param_value).lower()418            else:419                partitioned['query_string_kwargs'][key_name] = param_value420        elif location == 'header':421            shape = shape_members[param_name]422            value = self._convert_header_value(shape, param_value)423            partitioned['headers'][key_name] = str(value)424        elif location == 'headers':425            # 'headers' is a bit of an oddball.  The ``key_name``426            # is actually really a prefix for the header names:427            header_prefix = key_name428            # The value provided by the user is a dict so we'll be429            # creating multiple header key/val pairs.  The key430            # name to use for each header is the header_prefix (``key_name``)431            # plus the key provided by the user.432            self._do_serialize_header_map(header_prefix,433                                          partitioned['headers'],434                                          param_value)435        else:436            partitioned['body_kwargs'][param_name] = param_value437    def _do_serialize_header_map(self, header_prefix, headers, user_input):438        for key, val in user_input.items():439            full_key = header_prefix + key440            headers[full_key] = val441    def _serialize_body_params(self, params, shape):442        raise NotImplementedError('_serialize_body_params')443    def _convert_header_value(self, shape, value):444        if shape.type_name == 'timestamp':445            datetime_obj = parse_to_aware_datetime(value)446            timestamp = calendar.timegm(datetime_obj.utctimetuple())447            return self._timestamp_rfc822(timestamp)448        else:449            return value450class RestJSONSerializer(BaseRestSerializer, JSONSerializer):451    def _serialize_body_params(self, params, shape):452        serialized_body = self.MAP_TYPE()453        self._serialize(serialized_body, params, shape)454        return json.dumps(serialized_body).encode(self.DEFAULT_ENCODING)455class RestXMLSerializer(BaseRestSerializer):456    TIMESTAMP_FORMAT = 'iso8601'457    def _serialize_body_params(self, params, shape):458        root_name = shape.serialization['name']459        pseudo_root = ElementTree.Element('')460        self._serialize(shape, params, pseudo_root, root_name)461        real_root = list(pseudo_root)[0]...
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.