[catapult] - Copy Telemetry's gsutilz over to third_party.

third_party/gsutil/third_party/boto/tests/integration/sqs/test_connection.py

Index: third_party/gsutil/third_party/boto/tests/integration/sqs/test_connection.py
diff --git a/third_party/gsutil/third_party/boto/tests/integration/sqs/test_connection.py b/third_party/gsutil/third_party/boto/tests/integration/sqs/test_connection.py
new file mode 100644
index 0000000000000000000000000000000000000000..5ab80924d39eec92a197c427f07a09f5ea818ae2
--- /dev/null
+++ b/third_party/gsutil/third_party/boto/tests/integration/sqs/test_connection.py
@@ -0,0 +1,304 @@
+# 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 Message
+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_1 = c.create_queue(queue_name, timeout)
+        self.addCleanup(c.delete_queue, queue_1, True)
+        time.sleep(60)
+        rs = c.get_all_queues()
+        i = 0
+        for q in rs:
+            i += 1
+        assert i == num_queues + 1
+        assert queue_1.count_slow() == 0
+
+        # check the visibility timeout
+        t = queue_1.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_1.set_timeout(timeout)
+        time.sleep(60)
+        t = queue_1.get_timeout()
+        assert t == timeout, '%d != %d' % (t, timeout)
+
+        # now add a message
+        message_body = 'This is a test\n'
+        message = queue_1.new_message(message_body)
+        queue_1.write(message)
+        time.sleep(60)
+        assert queue_1.count_slow() == 1
+        time.sleep(90)
+
+        # now read the message from the queue with a 10 second timeout
+        message = queue_1.read(visibility_timeout=10)
+        assert message
+        assert message.get_body() == message_body
+
+        # now immediately try another read, shouldn't find anything
+        message = queue_1.read()
+        assert message == None
+
+        # now wait 30 seconds and try again
+        time.sleep(30)
+        message = queue_1.read()
+        assert message
+
+        # now delete the message
+        queue_1.delete_message(message)
+        time.sleep(30)
+        assert queue_1.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_1.write_batch(msgs)
+
+        # try to delete all of the messages using batch delete
+        deleted = 0
+        while deleted < num_msgs:
+            time.sleep(5)
+            msgs = queue_1.get_messages(num_msgs)
+            if msgs:
+                br = queue_1.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_2 = c.create_queue(queue_name, timeout)
+        self.addCleanup(c.delete_queue, queue_2, True)
+        queue_2.set_message_class(MHMessage)
+        time.sleep(30)
+
+        # now add a couple of messages
+        message = queue_2.new_message()
+        message['foo'] = 'bar'
+        queue_2.write(message)
+        message_body = {'fie': 'baz', 'foo': 'bar'}
+        message = queue_2.new_message(body=message_body)
+        queue_2.write(message)
+        time.sleep(30)
+
+        m = queue_2.read()
+        assert m['foo'] == 'bar'
+
+        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 +- 1 second for the total time the queue
+        # was blocked on the read call.
+        self.assertTrue(4.0 <= (end - start) <= 6.0)
+
+    def test_queue_deletion_affects_full_queues(self):
+        conn = SQSConnection()
+        initial_count = len(conn.get_all_queues())
+
+        empty = conn.create_queue('empty%d' % int(time.time()))
+        full = conn.create_queue('full%d' % int(time.time()))
+        time.sleep(60)
+        # Make sure they're both around.
+        self.assertEqual(len(conn.get_all_queues()), initial_count + 2)
+
+        # Put a message in the full queue.
+        m1 = Message()
+        m1.set_body('This is a test message.')
+        full.write(m1)
+        self.assertEqual(full.count(), 1)
+
+        self.assertTrue(conn.delete_queue(empty))
+        # Here's the regression for the docs. SQS will delete a queue with
+        # messages in it, no ``force_deletion`` needed.
+        self.assertTrue(conn.delete_queue(full))
+        # Wait long enough for SQS to finally remove the queues.
+        time.sleep(90)
+        self.assertEqual(len(conn.get_all_queues()), initial_count)
+
+    def test_get_messages_attributes(self):
+        conn = SQSConnection()
+        current_timestamp = int(time.time())
+        test = self.create_temp_queue(conn)
+        time.sleep(65)
+
+        # Put a message in the queue.
+        self.put_queue_message(test)
+        self.assertEqual(test.count(), 1)
+
+        # Check all attributes.
+        msgs = test.get_messages(
+            num_messages=1,
+            attributes='All'
+        )
+        for msg in msgs:
+            self.assertEqual(msg.attributes['ApproximateReceiveCount'], '1')
+            first_rec = msg.attributes['ApproximateFirstReceiveTimestamp']
+            first_rec = int(first_rec) / 1000
+            self.assertTrue(first_rec >= current_timestamp)
+
+        # Put another message in the queue.
+        self.put_queue_message(test)
+        self.assertEqual(test.count(), 1)
+
+        # Check a specific attribute.
+        msgs = test.get_messages(
+            num_messages=1,
+            attributes='ApproximateReceiveCount'
+        )
+        for msg in msgs:
+            self.assertEqual(msg.attributes['ApproximateReceiveCount'], '1')
+            with self.assertRaises(KeyError):
+                msg.attributes['ApproximateFirstReceiveTimestamp']
+
+    def test_queue_purge(self):
+        conn = SQSConnection()
+        test = self.create_temp_queue(conn)
+        time.sleep(65)
+
+        # Put some messages in the queue.
+        for x in range(0, 4):
+            self.put_queue_message(test)
+        self.assertEqual(test.count(), 4)
+
+        # Now purge the queue
+        conn.purge_queue(test)
+
+        # Now assert queue count is 0
+        self.assertEqual(test.count(), 0)
+
+    def create_temp_queue(self, conn):
+        current_timestamp = int(time.time())
+        queue_name = 'test%d' % int(time.time())
+        test = conn.create_queue(queue_name)
+        self.addCleanup(conn.delete_queue, test)
+
+        return test
+
+    def put_queue_message(self, queue):
+        m1 = Message()
+        m1.set_body('This is a test message.')
+        queue.write(m1)