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

How to use replace_in_encoded method in localstack

Best Python code snippet using localstack_python

kinesis_listener.py

Source:kinesis_listener.py

...70        return True71    def return_response(self, method, path, data, headers, response):72        action = headers.get('X-Amz-Target')73        data = self.decode_content(data or '{}')74        response._content = self.replace_in_encoded(response.content or '')75        records = []76        if action in (ACTION_CREATE_STREAM, ACTION_DELETE_STREAM):77            event_type = (event_publisher.EVENT_KINESIS_CREATE_STREAM if action == ACTION_CREATE_STREAM78                          else event_publisher.EVENT_KINESIS_DELETE_STREAM)79            payload = {'n': event_publisher.get_hash(data.get('StreamName'))}80            if action == ACTION_CREATE_STREAM:81                payload['s'] = data.get('ShardCount')82            event_publisher.fire_event(event_type, payload=payload)83        elif action == ACTION_PUT_RECORD:84            response_body = self.decode_content(response.content)85            # Note: avoid adding 'encryptionType':'NONE' in the event_record, as this breaks .NET Lambdas86            event_record = {87                'approximateArrivalTimestamp': epoch_timestamp(),88                'data': data['Data'],89                'partitionKey': data['PartitionKey'],90                'sequenceNumber': response_body.get('SequenceNumber')91            }92            event_records = [event_record]93            stream_name = data['StreamName']94            lambda_api.process_kinesis_records(event_records, stream_name)95        elif action == ACTION_PUT_RECORDS:96            event_records = []97            response_body = self.decode_content(response.content)98            if 'Records' in response_body:99                response_records = response_body['Records']100                records = data['Records']101                for index in range(0, len(records)):102                    record = records[index]103                    # Note: avoid adding 'encryptionType':'NONE' in the event_record, as this breaks .NET Lambdas104                    event_record = {105                        'approximateArrivalTimestamp': epoch_timestamp(),106                        'data': record['Data'],107                        'partitionKey': record['PartitionKey'],108                        'sequenceNumber': response_records[index].get('SequenceNumber')109                    }110                    event_records.append(event_record)111                stream_name = data['StreamName']112                lambda_api.process_kinesis_records(event_records, stream_name)113        elif action == ACTION_UPDATE_SHARD_COUNT:114            # Currently kinesalite, which backs the Kinesis implementation for localstack, does115            # not support UpdateShardCount:116            # https://github.com/mhart/kinesalite/issues/61117            #118            # [Terraform](https://www.terraform.io) makes the call to UpdateShardCount when it119            # applies Kinesis resources. A Terraform run fails when this is not present.120            #121            # The code that follows just returns a successful response, bypassing the 400122            # response that kinesalite returns.123            #124            response = Response()125            response.status_code = 200126            content = {127                'CurrentShardCount': 1,128                'StreamName': data['StreamName'],129                'TargetShardCount': data['TargetShardCount']130            }131            response.encoding = 'UTF-8'132            response._content = json.dumps(content)133            return response134        elif action == ACTION_GET_RECORDS:135            sdk_v2 = self.sdk_is_v2(headers.get('User-Agent', '').split(' ')[0])136            results, encoding_type = self.decode_content(response.content, True)137            for record in results['Records']:138                if sdk_v2:139                    record['ApproximateArrivalTimestamp'] = int(record['ApproximateArrivalTimestamp'] * 1000)140                if not isinstance(record['Data'], str):141                    record['Data'] = base64.encodebytes(bytearray(record['Data']['data']))142            if encoding_type == APPLICATION_CBOR:143                response._content = cbor2.dumps(results)144            else:145                response._content = json.dumps(results)146            return response147    def sdk_is_v2(self, user_agent):148        if re.search(r'\/2.\d+.\d+', user_agent):149            return True150        return False151    def replace_in_encoded(self, data):152        if not data:153            return ''154        decoded, type_encoding = self.decode_content(data, True)155        if type_encoding == APPLICATION_JSON:156            return re.sub(r'arn:aws:kinesis:[^:]+:', 'arn:aws:kinesis:%s:' % aws_stack.get_region(),157            to_str(data))158        if type_encoding == APPLICATION_CBOR:159            replaced = re.sub(r'arn:aws:kinesis:[^:]+:', 'arn:aws:kinesis:%s:' % aws_stack.get_region(),160            json.dumps(decoded))161            return cbor2.dumps(json.loads(replaced))162    def decode_content(self, data, describe=False):163        content_type = ''164        try:165            decoded = json.loads(to_str(data))...

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.