Implemented remaining Queue attributes. This closes spulec/moto#22

The following attributes were added:

  - ApproximateNumberOfMessagesDelayed
  - ApproximateNumberOfMessagesNotVisible
  - CreatedTimestamp
  - DelaySeconds
  - LastModifiedTimestamp
  - MaximumMessageSize
  - MessageRetentionPeriod
  - QueueArn
  - ReceiveMessageWaitTimeSeconds
This commit is contained in:
dilshod.tadjibaev 2013-04-23 23:53:30 -07:00
parent 6107658c65
commit c9fb6f1cc1
2 changed files with 55 additions and 1 deletions

View File

@ -1,4 +1,5 @@
import md5
import time
from moto.core import BaseBackend
from moto.core.utils import camelcase_to_underscores, get_random_message_id
@ -6,6 +7,7 @@ from .utils import generate_receipt_handle
class Message(object):
def __init__(self, message_id, body):
self.id = message_id
self.body = body
@ -19,13 +21,35 @@ class Message(object):
class Queue(object):
camelcase_attributes = ['VisibilityTimeout', 'ApproximateNumberOfMessages']
camelcase_attributes = ['ApproximateNumberOfMessages',
'ApproximateNumberOfMessagesDelayed',
'ApproximateNumberOfMessagesNotVisible',
'CreatedTimestamp',
'DelaySeconds',
'LastModifiedTimestamp',
'MaximumMessageSize',
'MessageRetentionPeriod',
'QueueArn',
'ReceiveMessageWaitTimeSeconds',
'VisibilityTimeout']
def __init__(self, name, visibility_timeout):
self.name = name
self.visibility_timeout = visibility_timeout or 30
self.messages = []
now = time.time()
self.approximate_number_of_messages_delayed = 0
self.approximate_number_of_messages_not_visible = 0
self.created_timestamp = now
self.delay_seconds = 0
self.last_modified_timestamp = now
self.maximum_message_size = 64 << 10
self.message_retention_period = 86400 * 4 # four days
self.queue_arn = 'arn:aws:sqs:sqs.us-east-1:123456789012:%s' % self.name
self.receive_message_wait_time_seconds = 0
@property
def attributes(self):
result = {}

View File

@ -140,3 +140,33 @@ def test_delete_batch_operation():
@mock_sqs
def test_sqs_method_not_implemented():
requests.post.when.called_with("https://sqs.amazonaws.com/?Action=[foobar]").should.throw(NotImplementedError)
@mock_sqs
def test_queue_attributes():
conn = boto.connect_sqs('the_key', 'the_secret')
queue_name = 'test-queue'
visibility_timeout = 60
queue = conn.create_queue(queue_name, visibility_timeout=visibility_timeout)
attributes = queue.get_attributes()
attributes['QueueArn'].should.look_like(
'arn:aws:sqs:sqs.us-east-1:123456789012:%s' % queue_name)
attributes['VisibilityTimeout'].should.look_like(str(visibility_timeout))
attribute_names = queue.get_attributes().keys()
attribute_names.should.contain('ApproximateNumberOfMessagesNotVisible')
attribute_names.should.contain('MessageRetentionPeriod')
attribute_names.should.contain('ApproximateNumberOfMessagesDelayed')
attribute_names.should.contain('MaximumMessageSize')
attribute_names.should.contain('CreatedTimestamp')
attribute_names.should.contain('ApproximateNumberOfMessages')
attribute_names.should.contain('ReceiveMessageWaitTimeSeconds')
attribute_names.should.contain('DelaySeconds')
attribute_names.should.contain('VisibilityTimeout')
attribute_names.should.contain('LastModifiedTimestamp')
attribute_names.should.contain('QueueArn')