Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Implement SQS get_queue_url. Closes #18

Browse Source
This commit is contained in:
Steve Pulec 2013-04-18 23:07:24 -04:00
parent afc7a64a34
commit dbf2368aa6
2 changed files with 25 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
moto/sqs/responses.py
View File

@ -17,6 +17,12 @@ class QueuesResponse(BaseResponse):
template = Template(CREATE_QUEUE_RESPONSE) template = Template(CREATE_QUEUE_RESPONSE)
return template.render(queue=queue) return template.render(queue=queue)
def get_queue_url(self):
queue_name = self.querystring.get("QueueName")[0]
queue = sqs_backend.get_queue(queue_name)
template = Template(GET_QUEUE_URL_RESPONSE)
return template.render(queue=queue)
def list_queues(self): def list_queues(self):
queues = sqs_backend.list_queues() queues = sqs_backend.list_queues()
template = Template(LIST_QUEUES_RESPONSE) template = Template(LIST_QUEUES_RESPONSE)
@ -143,6 +149,15 @@ CREATE_QUEUE_RESPONSE = """<CreateQueueResponse>
</ResponseMetadata> </ResponseMetadata>
</CreateQueueResponse>""" </CreateQueueResponse>"""
GET_QUEUE_URL_RESPONSE = """<GetQueueUrlResponse>
<GetQueueUrlResult>
<QueueUrl>http://sqs.us-east-1.amazonaws.com/123456789012/{{ queue.name }}</QueueUrl>
</GetQueueUrlResult>
<ResponseMetadata>
<RequestId>470a6f13-2ed9-4181-ad8a-2fdea142988e</RequestId>
</ResponseMetadata>
</GetQueueUrlResponse>"""
LIST_QUEUES_RESPONSE = """<ListQueuesResponse> LIST_QUEUES_RESPONSE = """<ListQueuesResponse>
<ListQueuesResult> <ListQueuesResult>
{% for queue in queues %} {% for queue in queues %}

10
tests/test_sqs/test_sqs.py
View File

@ -18,6 +18,16 @@ def test_create_queue():
all_queues[0].get_timeout().should.equal(60) all_queues[0].get_timeout().should.equal(60)
@mock_sqs
def test_get_queue():
conn = boto.connect_sqs('the_key', 'the_secret')
conn.create_queue("test-queue", visibility_timeout=60)
queue = conn.get_queue("test-queue")
queue.name.should.equal("test-queue")
queue.get_timeout().should.equal(60)
@mock_sqs @mock_sqs
def test_delete_queue(): def test_delete_queue():
conn = boto.connect_sqs('the_key', 'the_secret') conn = boto.connect_sqs('the_key', 'the_secret')