| Index: tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/sqs/test_connection.py
|
| diff --git a/tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/sqs/test_connection.py b/tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/sqs/test_connection.py
|
| deleted file mode 100644
|
| index 5ab80924d39eec92a197c427f07a09f5ea818ae2..0000000000000000000000000000000000000000
|
| --- a/tools/telemetry/third_party/gsutilz/third_party/boto/tests/integration/sqs/test_connection.py
|
| +++ /dev/null
|
| @@ -1,304 +0,0 @@
|
| -# 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)
|
|
|