Index: third_party/gsutil/boto/tests/integration/sqs/test_connection.py |
diff --git a/third_party/gsutil/boto/tests/integration/sqs/test_connection.py b/third_party/gsutil/boto/tests/integration/sqs/test_connection.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..611d5219acb555c1248eeab89142305900054c20 |
--- /dev/null |
+++ b/third_party/gsutil/boto/tests/integration/sqs/test_connection.py |
@@ -0,0 +1,217 @@ |
+# Copyright (c) 2006-2010 Mitch Garnaat http://garnaat.org/ |
+# Copyright (c) 2010, Eucalyptus Systems, Inc. |
+# All rights reserved. |
+# |
+# Permission is hereby granted, free of charge, to any person obtaining a |
+# copy of this software and associated documentation files (the |
+# "Software"), to deal in the Software without restriction, including |
+# without limitation the rights to use, copy, modify, merge, publish, dis- |
+# tribute, sublicense, and/or sell copies of the Software, and to permit |
+# persons to whom the Software is furnished to do so, subject to the fol- |
+# lowing conditions: |
+# |
+# The above copyright notice and this permission notice shall be included |
+# in all copies or substantial portions of the Software. |
+# |
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
+# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
+# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
+# IN THE SOFTWARE. |
+ |
+""" |
+Some unit tests for the SQSConnection |
+""" |
+ |
+import time |
+from threading import Timer |
+from tests.unit import unittest |
+ |
+from boto.sqs.connection import SQSConnection |
+from boto.sqs.message import MHMessage |
+from boto.exception import SQSError |
+ |
+ |
+class SQSConnectionTest(unittest.TestCase): |
+ |
+ sqs = True |
+ |
+ def test_1_basic(self): |
+ print '--- running SQSConnection tests ---' |
+ c = SQSConnection() |
+ rs = c.get_all_queues() |
+ num_queues = 0 |
+ for q in rs: |
+ num_queues += 1 |
+ |
+ # try illegal name |
+ try: |
+ queue = c.create_queue('bad*queue*name') |
+ self.fail('queue name should have been bad') |
+ except SQSError: |
+ pass |
+ |
+ # now create one that should work and should be unique (i.e. a new one) |
+ queue_name = 'test%d' % int(time.time()) |
+ timeout = 60 |
+ queue = c.create_queue(queue_name, timeout) |
+ time.sleep(60) |
+ rs = c.get_all_queues() |
+ i = 0 |
+ for q in rs: |
+ i += 1 |
+ assert i == num_queues + 1 |
+ assert queue.count_slow() == 0 |
+ |
+ # check the visibility timeout |
+ t = queue.get_timeout() |
+ assert t == timeout, '%d != %d' % (t, timeout) |
+ |
+ # now try to get queue attributes |
+ a = q.get_attributes() |
+ assert 'ApproximateNumberOfMessages' in a |
+ assert 'VisibilityTimeout' in a |
+ a = q.get_attributes('ApproximateNumberOfMessages') |
+ assert 'ApproximateNumberOfMessages' in a |
+ assert 'VisibilityTimeout' not in a |
+ a = q.get_attributes('VisibilityTimeout') |
+ assert 'ApproximateNumberOfMessages' not in a |
+ assert 'VisibilityTimeout' in a |
+ |
+ # now change the visibility timeout |
+ timeout = 45 |
+ queue.set_timeout(timeout) |
+ time.sleep(60) |
+ t = queue.get_timeout() |
+ assert t == timeout, '%d != %d' % (t, timeout) |
+ |
+ # now add a message |
+ message_body = 'This is a test\n' |
+ message = queue.new_message(message_body) |
+ queue.write(message) |
+ time.sleep(60) |
+ assert queue.count_slow() == 1 |
+ time.sleep(90) |
+ |
+ # now read the message from the queue with a 10 second timeout |
+ message = queue.read(visibility_timeout=10) |
+ assert message |
+ assert message.get_body() == message_body |
+ |
+ # now immediately try another read, shouldn't find anything |
+ message = queue.read() |
+ assert message == None |
+ |
+ # now wait 30 seconds and try again |
+ time.sleep(30) |
+ message = queue.read() |
+ assert message |
+ |
+ # now delete the message |
+ queue.delete_message(message) |
+ time.sleep(30) |
+ assert queue.count_slow() == 0 |
+ |
+ # try a batch write |
+ num_msgs = 10 |
+ msgs = [(i, 'This is message %d' % i, 0) for i in range(num_msgs)] |
+ queue.write_batch(msgs) |
+ |
+ # try to delete all of the messages using batch delete |
+ deleted = 0 |
+ while deleted < num_msgs: |
+ time.sleep(5) |
+ msgs = queue.get_messages(num_msgs) |
+ if msgs: |
+ br = queue.delete_message_batch(msgs) |
+ deleted += len(br.results) |
+ |
+ # create another queue so we can test force deletion |
+ # we will also test MHMessage with this queue |
+ queue_name = 'test%d' % int(time.time()) |
+ timeout = 60 |
+ queue = c.create_queue(queue_name, timeout) |
+ queue.set_message_class(MHMessage) |
+ time.sleep(30) |
+ |
+ # now add a couple of messages |
+ message = queue.new_message() |
+ message['foo'] = 'bar' |
+ queue.write(message) |
+ message_body = {'fie': 'baz', 'foo': 'bar'} |
+ message = queue.new_message(body=message_body) |
+ queue.write(message) |
+ time.sleep(30) |
+ |
+ m = queue.read() |
+ assert m['foo'] == 'bar' |
+ |
+ # now delete that queue and messages |
+ c.delete_queue(queue, True) |
+ |
+ print '--- tests completed ---' |
+ |
+ def test_sqs_timeout(self): |
+ c = SQSConnection() |
+ queue_name = 'test_sqs_timeout_%s' % int(time.time()) |
+ queue = c.create_queue(queue_name) |
+ self.addCleanup(c.delete_queue, queue, True) |
+ start = time.time() |
+ poll_seconds = 2 |
+ response = queue.read(visibility_timeout=None, |
+ wait_time_seconds=poll_seconds) |
+ total_time = time.time() - start |
+ self.assertTrue(total_time > poll_seconds, |
+ "SQS queue did not block for at least %s seconds: %s" % |
+ (poll_seconds, total_time)) |
+ self.assertIsNone(response) |
+ |
+ # Now that there's an element in the queue, we should not block for 2 |
+ # seconds. |
+ c.send_message(queue, 'test message') |
+ start = time.time() |
+ poll_seconds = 2 |
+ message = c.receive_message( |
+ queue, number_messages=1, |
+ visibility_timeout=None, attributes=None, |
+ wait_time_seconds=poll_seconds)[0] |
+ total_time = time.time() - start |
+ self.assertTrue(total_time < poll_seconds, |
+ "SQS queue blocked longer than %s seconds: %s" % |
+ (poll_seconds, total_time)) |
+ self.assertEqual(message.get_body(), 'test message') |
+ |
+ attrs = c.get_queue_attributes(queue, 'ReceiveMessageWaitTimeSeconds') |
+ self.assertEqual(attrs['ReceiveMessageWaitTimeSeconds'], '0') |
+ |
+ def test_sqs_longpoll(self): |
+ c = SQSConnection() |
+ queue_name = 'test_sqs_longpoll_%s' % int(time.time()) |
+ queue = c.create_queue(queue_name) |
+ self.addCleanup(c.delete_queue, queue, True) |
+ messages = [] |
+ |
+ # The basic idea is to spawn a timer thread that will put something |
+ # on the queue in 5 seconds and verify that our long polling client |
+ # sees the message after waiting for approximately that long. |
+ def send_message(): |
+ messages.append( |
+ queue.write(queue.new_message('this is a test message'))) |
+ |
+ t = Timer(5.0, send_message) |
+ t.start() |
+ self.addCleanup(t.join) |
+ |
+ start = time.time() |
+ response = queue.read(wait_time_seconds=10) |
+ end = time.time() |
+ |
+ t.join() |
+ self.assertEqual(response.id, messages[0].id) |
+ self.assertEqual(response.get_body(), messages[0].get_body()) |
+ # The timer thread should send the message in 5 seconds, so |
+ # we're giving +- .5 seconds for the total time the queue |
+ # was blocked on the read call. |
+ self.assertTrue(4.5 <= (end - start) <= 5.5) |