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

How to use _fix_endpoint method in localstack

Best Python code snippet using localstack_python

sqs_client.py

Source:sqs_client.py

...38    def __init__(self, credentials, url=None, proxyinfo=None):39        if not url:40            endpoint = SQSClient.endpointForRegion('us-east-1')41        else:42            endpoint = self._fix_endpoint(url)43        super(SQSClient, self).__init__(credentials, False, endpoint, xmlns='http://queue.amazonaws.com/doc/%s/' % SQSClient._apiVersion, proxyinfo=proxyinfo)44        log.debug("SQS client initialized with endpoint %s", endpoint)45    # SQS SSL certificates have CNs based on queue.amazonaws.com46    # Python2.6 will fail to verify the hostname of the certificate47    # Due to http://bugs.python.org/issue13034 only being fixed in 2.7 and 3.248    def _fix_endpoint(self, url):49        m = re.match(r'^https://sqs\.(.*?)\.amazonaws\.com(.*)$', url)50        if m:51            if m.group(1) == 'us-east-1':52                return 'https://queue.amazonaws.com%s' % m.group(2)53            return 'https://%s.queue.amazonaws.com%s' % m.group(1, 2)54        return url55    @classmethod56    def endpointForRegion(cls, region):57        if region == 'us-east-1':58            return 'https://queue.amazonaws.com'59        return 'https://%s.queue.amazonaws.com' % region60    @retry_on_failure(http_error_extractor=aws_client.Client._get_xml_extractor(_xmlns))61    def receive_message(self, queue_url, attributes=None, max_messages=1, visibility_timeout=None,62                              request_credentials=None, wait_time=None):63        """64        Calls ReceiveMessage and returns a list of Message objects65        Throws an IOError on failure.66        """67        if not attributes: attributes = ['All']68        queue_url = self._fix_endpoint(queue_url)69        log.debug("Receiving messages for queue %s", queue_url)70        params = { "Action" : "ReceiveMessage", "Version" : SQSClient._apiVersion, "MaxNumberOfMessages" : str(max_messages) }71        for i in range(len(attributes)):72            params['AttributeName.%s' % (i + 1)]=attributes[i]73        if visibility_timeout:74            params['VisibilityTimeout'] = str(visibility_timeout)75        if wait_time:76            params['WaitTimeSeconds'] = str(wait_time)77        response_content = self._call(params, queue_url, request_credentials,78                                      timeout=wait_time + 3 if wait_time else None).content79        return Message._parse_list(StringIO.StringIO(response_content), self._xmlns)80    @retry_on_failure(http_error_extractor=aws_client.Client._get_xml_extractor(_xmlns))81    def send_message(self, queue_url, message_body, delay_seconds=None, request_credentials=None):82        """83        Calls SendMessage and returns a tuple of (MessageId, MD5OfMessageBody)84        Throws an IOError on failure.85        """86        queue_url = self._fix_endpoint(queue_url)87        log.debug("Sending %s to queue %s", message_body, queue_url)88        params = { "Action" : "SendMessage", "Version" : SQSClient._apiVersion, "MessageBody" : message_body}89        if delay_seconds:90            params["DelaySeconds"] = delay_seconds91        root = ElementTree.ElementTree(file=StringIO.StringIO(self._call(params, queue_url, request_credentials, verb='POST').content))92        message_id = root.findtext('{%s}SendMessageResult/{%s}MessageId' % (self._xmlns, self._xmlns))93        md5_of_body = root.findtext('{%s}SendMessageResult/{%s}MD5OfMessageBody' % (self._xmlns, self._xmlns))94        return (message_id, md5_of_body)95    @retry_on_failure(http_error_extractor=aws_client.Client._get_xml_extractor(_xmlns))96    def delete_message(self, queue_url, receipt_handle, request_credentials=None):97        """98        Calls DeleteMessage on a specified receipt handle99        Throws an IOError on failure.100        """101        queue_url = self._fix_endpoint(queue_url)102        log.debug("Deleting %s from queue %s", receipt_handle, queue_url)103        params = { "Action" : "DeleteMessage", "Version" : SQSClient._apiVersion, "ReceiptHandle" : receipt_handle}104        self._call(params, queue_url, request_credentials)105class Message(object):106    """A message off of an SQS queue"""107    @classmethod108    def _parse_list(cls, data, xmlns):109        if not data:110            return []111        root = ElementTree.ElementTree(file=data)112        msgs = root.findall('{%s}ReceiveMessageResult/{%s}Message' % (xmlns, xmlns))113        return [cls._from_elem(elem, xmlns) for elem in msgs]114    @classmethod115    def _from_elem(cls, elem, xmlns):...

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.