Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use is_sdk_v2_request method in localstack

Best Python code snippet using localstack_python

kinesis_listener.py

Source:kinesis_listener.py

...152            response.encoding = "UTF-8"153            response._content = json.dumps(content)154            return response155        elif action == "GetRecords":156            sdk_v2 = self.is_sdk_v2_request(headers)157            results, encoding_type = self.decode_content(response.content, True)158            records = results.get("Records", [])159            if not records:160                return response161            for record in records:162                if sdk_v2:163                    record["ApproximateArrivalTimestamp"] = int(164                        record["ApproximateArrivalTimestamp"]165                    )166                tmp = record["Data"]167                # "tmp" is either a base64-encoded string, or a dict {"data": <data_bytes...>}168                is_base64_encoded = isinstance(tmp, str)169                if isinstance(tmp, dict):170                    tmp = bytearray(tmp.get("data", []))171                if is_base64_encoded:172                    tmp = base64.b64decode(tmp)173                if encoding_type == APPLICATION_JSON:174                    # Remove double quotes from data written in regular JSON encoding (not for CBOR!)175                    # https://github.com/localstack/localstack/issues/3588176                    if len(tmp) >= 2 and tmp[0] == tmp[-1] == b'"'[0]:177                        tmp = tmp[1:-1]178                if encoding_type == APPLICATION_CBOR:179                    # Note: AWS Java SDK requires bytes embedded in CBOR encoding, not base64 strings!180                    tmp = to_bytes(tmp)181                if encoding_type == APPLICATION_JSON:182                    if is_base64_encoded:183                        tmp = base64.b64encode(tmp)184                    tmp = to_str(tmp)185                record["Data"] = tmp186            response._content = encode_data(results, encoding_type)187            return response188        if response.status_code >= 400:189            response_body = self.decode_content(response.content)190            if (191                response_body192                and response_body.get("__type")193                and not headers.get(HEADER_AMZN_ERROR_TYPE)194            ):195                response.headers[HEADER_AMZN_ERROR_TYPE] = str(response_body.get("__type"))196    def is_sdk_v2_request(self, headers):197        user_agent = headers.get("User-Agent", "").split(" ")[0]198        if re.search(r"\/2.\d+.\d+", user_agent):199            return True200        return False201    def replace_in_encoded(self, data):202        if not data:203            return ""204        def _replace(_data):205            return re.sub(206                r"arn:aws:kinesis:[^:]+:",207                "arn:aws:kinesis:%s:" % aws_stack.get_region(),208                _data,209            )210        decoded, type_encoding = self.decode_content(data, True)...

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.