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

How to use get_queue_attributes method in localstack

Best Python code snippet using localstack_python

test_queue.py

Source:test_queue.py

1# -*- coding: utf-8 -*-2from __future__ import absolute_import, unicode_literals3import pytest4from case import Mock5from kombu.asynchronous.aws.sqs.message import AsyncMessage6from kombu.asynchronous.aws.sqs.queue import AsyncQueue7from t.mocks import PromiseMock8from ..case import AWSCase9class test_AsyncQueue(AWSCase):10    def setup(self):11        self.conn = Mock(name='connection')12        self.x = AsyncQueue(self.conn, '/url')13        self.callback = PromiseMock(name='callback')14    def test_message_class(self):15        assert issubclass(self.x.message_class, AsyncMessage)16    def test_get_attributes(self):17        self.x.get_attributes(attributes='QueueSize', callback=self.callback)18        self.x.connection.get_queue_attributes.assert_called_with(19            self.x, 'QueueSize', self.callback,20        )21    def test_set_attribute(self):22        self.x.set_attribute('key', 'value', callback=self.callback)23        self.x.connection.set_queue_attribute.assert_called_with(24            self.x, 'key', 'value', self.callback,25        )26    def test_get_timeout(self):27        self.x.get_timeout(callback=self.callback)28        self.x.connection.get_queue_attributes.assert_called()29        on_ready = self.x.connection.get_queue_attributes.call_args[0][2]30        self.x.connection.get_queue_attributes.assert_called_with(31            self.x, 'VisibilityTimeout', on_ready,32        )33        on_ready({'VisibilityTimeout': '303'})34        self.callback.assert_called_with(303)35    def test_set_timeout(self):36        self.x.set_timeout(808, callback=self.callback)37        self.x.connection.set_queue_attribute.assert_called()38        on_ready = self.x.connection.set_queue_attribute.call_args[0][3]39        self.x.connection.set_queue_attribute.assert_called_with(40            self.x, 'VisibilityTimeout', 808, on_ready,41        )42        on_ready(808)43        self.callback.assert_called_with(808)44        assert self.x.visibility_timeout == 80845        on_ready(None)46        assert self.x.visibility_timeout == 80847    def test_add_permission(self):48        self.x.add_permission(49            'label', 'accid', 'action', callback=self.callback,50        )51        self.x.connection.add_permission.assert_called_with(52            self.x, 'label', 'accid', 'action', self.callback,53        )54    def test_remove_permission(self):55        self.x.remove_permission('label', callback=self.callback)56        self.x.connection.remove_permission.assert_called_with(57            self.x, 'label', self.callback,58        )59    def test_read(self):60        self.x.read(visibility_timeout=909, callback=self.callback)61        self.x.connection.receive_message.assert_called()62        on_ready = self.x.connection.receive_message.call_args[1]['callback']63        self.x.connection.receive_message.assert_called_with(64            self.x, number_messages=1, visibility_timeout=909,65            attributes=None, wait_time_seconds=None, callback=on_ready,66        )67        messages = [Mock(name='message1')]68        on_ready(messages)69        self.callback.assert_called_with(messages[0])70    def MockMessage(self, id, md5):71        m = Mock(name='Message-{0}'.format(id))72        m.id = id73        m.md5 = md574        return m75    def test_write(self):76        message = self.MockMessage('id1', 'digest1')77        self.x.write(message, delay_seconds=303, callback=self.callback)78        self.x.connection.send_message.assert_called()79        on_ready = self.x.connection.send_message.call_args[1]['callback']80        self.x.connection.send_message.assert_called_with(81            self.x, message.get_body_encoded(), 303,82            callback=on_ready,83        )84        new_message = self.MockMessage('id2', 'digest2')85        on_ready(new_message)86        assert message.id == 'id2'87        assert message.md5 == 'digest2'88    def test_write_batch(self):89        messages = [('id1', 'A', 0), ('id2', 'B', 303)]90        self.x.write_batch(messages, callback=self.callback)91        self.x.connection.send_message_batch.assert_called_with(92            self.x, messages, callback=self.callback,93        )94    def test_delete_message(self):95        message = self.MockMessage('id1', 'digest1')96        self.x.delete_message(message, callback=self.callback)97        self.x.connection.delete_message.assert_called_with(98            self.x, message, self.callback,99        )100    def test_delete_message_batch(self):101        messages = [102            self.MockMessage('id1', 'r1'),103            self.MockMessage('id2', 'r2'),104        ]105        self.x.delete_message_batch(messages, callback=self.callback)106        self.x.connection.delete_message_batch.assert_called_with(107            self.x, messages, callback=self.callback,108        )109    def test_change_message_visibility_batch(self):110        messages = [111            (self.MockMessage('id1', 'r1'), 303),112            (self.MockMessage('id2', 'r2'), 909),113        ]114        self.x.change_message_visibility_batch(115            messages, callback=self.callback,116        )117        self.x.connection.change_message_visibility_batch.assert_called_with(118            self.x, messages, callback=self.callback,119        )120    def test_delete(self):121        self.x.delete(callback=self.callback)122        self.x.connection.delete_queue.assert_called_with(123            self.x, callback=self.callback,124        )125    def test_count(self):126        self.x.count(callback=self.callback)127        self.x.connection.get_queue_attributes.assert_called()128        on_ready = self.x.connection.get_queue_attributes.call_args[0][2]129        self.x.connection.get_queue_attributes.assert_called_with(130            self.x, 'ApproximateNumberOfMessages', on_ready,131        )132        on_ready({'ApproximateNumberOfMessages': '909'})133        self.callback.assert_called_with(909)134    def test_interface__count_slow(self):135        with pytest.raises(NotImplementedError):136            self.x.count_slow()137    def test_interface__dump(self):138        with pytest.raises(NotImplementedError):139            self.x.dump()140    def test_interface__save_to_file(self):141        with pytest.raises(NotImplementedError):142            self.x.save_to_file()143    def test_interface__save_to_filename(self):144        with pytest.raises(NotImplementedError):145            self.x.save_to_filename()146    def test_interface__save(self):147        with pytest.raises(NotImplementedError):148            self.x.save()149    def test_interface__save_to_s3(self):150        with pytest.raises(NotImplementedError):151            self.x.save_to_s3()152    def test_interface__load_from_s3(self):153        with pytest.raises(NotImplementedError):154            self.x.load_from_s3()155    def test_interface__load_from_file(self):156        with pytest.raises(NotImplementedError):157            self.x.load_from_file()158    def test_interface__load_from_filename(self):159        with pytest.raises(NotImplementedError):160            self.x.load_from_filename()161    def test_interface__load(self):162        with pytest.raises(NotImplementedError):163            self.x.load()164    def test_interface__clear(self):165        with pytest.raises(NotImplementedError):...

sqs_barrel.py

Source:sqs_barrel.py

...32            items[region] = []33            response = client.list_queues()34            items[region].extend(response['QueueUrls'])35        return items36    def get_queue_attributes(self):37        if self.cache.get('get_queue_attributes'):38            return self.cache['get_queue_attributes']39        items = {}40        for region, client in self.clients.items():41            items[region] = {}42            for queue in self.tap('list_queues')[region]:43                response = client.get_queue_attributes(44                    QueueUrl=queue45                )46                items[region][queue] = response['Attributes']...

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.