<s> """<STR_LIT>""" <EOL> import time <EOL> from threading import Timer <EOL> from tests . unit import unittest <EOL> from boto . sqs . connection import SQSConnection <EOL> from boto . sqs . message import Message <EOL> from boto . sqs . message import MHMessage <EOL> from boto . exception import SQSError <EOL> class SQSConnectionTest ( unittest . TestCase ) : <EOL> sqs = True <EOL> def test_1_basic ( self ) : <EOL> print ( '<STR_LIT>' ) <EOL> c = SQSConnection ( ) <EOL> rs = c . get_all_queues ( ) <EOL> num_queues = <NUM_LIT:0> <EOL> for q in rs : <EOL> num_queues += <NUM_LIT:1> <EOL> try : <EOL> queue = c . create_queue ( '<STR_LIT>' ) <EOL> self . fail ( '<STR_LIT>' ) <EOL> except SQSError : <EOL> pass <EOL> queue_name = '<STR_LIT>' % int ( time . time ( ) ) <EOL> timeout = <NUM_LIT> <EOL> queue_1 = c . create_queue ( queue_name , timeout ) <EOL> self . addCleanup ( c . delete_queue , queue_1 , True ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> rs = c . get_all_queues ( ) <EOL> i = <NUM_LIT:0> <EOL> for q in rs : <EOL> i += <NUM_LIT:1> <EOL> assert i == num_queues + <NUM_LIT:1> <EOL> assert queue_1 . count_slow ( ) == <NUM_LIT:0> <EOL> t = queue_1 . get_timeout ( ) <EOL> assert t == timeout , '<STR_LIT>' % ( t , timeout ) <EOL> a = q . get_attributes ( ) <EOL> assert '<STR_LIT>' in a <EOL> assert '<STR_LIT>' in a <EOL> a = q . get_attributes ( '<STR_LIT>' ) <EOL> assert '<STR_LIT>' in a <EOL> assert '<STR_LIT>' not in a <EOL> a = q . get_attributes ( '<STR_LIT>' ) <EOL> assert '<STR_LIT>' not in a <EOL> assert '<STR_LIT>' in a <EOL> timeout = <NUM_LIT> <EOL> queue_1 . set_timeout ( timeout ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> t = queue_1 . get_timeout ( ) <EOL> assert t == timeout , '<STR_LIT>' % ( t , timeout ) <EOL> message_body = '<STR_LIT>' <EOL> message = queue_1 . new_message ( message_body ) <EOL> queue_1 . write ( message ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> assert queue_1 . count_slow ( ) == <NUM_LIT:1> <EOL> time . sleep ( <NUM_LIT> ) <EOL> message = queue_1 . read ( visibility_timeout = <NUM_LIT:10> ) <EOL> assert message <EOL> assert message . get_body ( ) == message_body <EOL> message = queue_1 . read ( ) <EOL> assert message == None <EOL> time . sleep ( <NUM_LIT:30> ) <EOL> message = queue_1 . read ( ) <EOL> assert message <EOL> queue_1 . delete_message ( message ) <EOL> time . sleep ( <NUM_LIT:30> ) <EOL> assert queue_1 . count_slow ( ) == <NUM_LIT:0> <EOL> num_msgs = <NUM_LIT:10> <EOL> msgs = [ ( i , '<STR_LIT>' % i , <NUM_LIT:0> ) for i in range ( num_msgs ) ] <EOL> queue_1 . write_batch ( msgs ) <EOL> deleted = <NUM_LIT:0> <EOL> while deleted < num_msgs : <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> msgs = queue_1 . get_messages ( num_msgs ) <EOL> if msgs : <EOL> br = queue_1 . delete_message_batch ( msgs ) <EOL> deleted += len ( br . results ) <EOL> num_msgs = <NUM_LIT:10> <EOL> attrs = { <EOL> '<STR_LIT:foo>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> } <EOL> msgs = [ ( i , '<STR_LIT>' % i , <NUM_LIT:0> , attrs ) for i in range ( num_msgs ) ] <EOL> queue_1 . write_batch ( msgs ) <EOL> queue_name = '<STR_LIT>' % int ( time . time ( ) ) <EOL> timeout = <NUM_LIT> <EOL> queue_2 = c . create_queue ( queue_name , timeout ) <EOL> self . addCleanup ( c . delete_queue , queue_2 , True ) <EOL> queue_2 . set_message_class ( MHMessage ) <EOL> time . sleep ( <NUM_LIT:30> ) <EOL> message = queue_2 . new_message ( ) <EOL> message [ '<STR_LIT:foo>' ] = '<STR_LIT:bar>' <EOL> queue_2 . write ( message ) <EOL> message_body = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:foo>' : '<STR_LIT:bar>' } <EOL> message = queue_2 . new_message ( body = message_body ) <EOL> queue_2 . write ( message ) <EOL> time . sleep ( <NUM_LIT:30> ) <EOL> m = queue_2 . read ( ) <EOL> assert m [ '<STR_LIT:foo>' ] == '<STR_LIT:bar>' <EOL> print ( '<STR_LIT>' ) <EOL> def test_sqs_timeout ( self ) : <EOL> c = SQSConnection ( ) <EOL> queue_name = '<STR_LIT>' % int ( time . time ( ) ) <EOL> queue = c . create_queue ( queue_name ) <EOL> self . addCleanup ( c . delete_queue , queue , True ) <EOL> start = time . time ( ) <EOL> poll_seconds = <NUM_LIT:2> <EOL> response = queue . read ( visibility_timeout = None , <EOL> wait_time_seconds = poll_seconds ) <EOL> total_time = time . time ( ) - start <EOL> self . assertTrue ( total_time > poll_seconds , <EOL> "<STR_LIT>" % <EOL> ( poll_seconds , total_time ) ) <EOL> self . assertIsNone ( response ) <EOL> c . send_message ( queue , '<STR_LIT>' ) <EOL> start = time . time ( ) <EOL> poll_seconds = <NUM_LIT:2> <EOL> message = c . receive_message ( <EOL> queue , number_messages = <NUM_LIT:1> , <EOL> visibility_timeout = None , attributes = None , <EOL> wait_time_seconds = poll_seconds ) [ <NUM_LIT:0> ] <EOL> total_time = time . time ( ) - start <EOL> self . assertTrue ( total_time < poll_seconds , <EOL> "<STR_LIT>" % <EOL> ( poll_seconds , total_time ) ) <EOL> self . assertEqual ( message . get_body ( ) , '<STR_LIT>' ) <EOL> attrs = c . get_queue_attributes ( queue , '<STR_LIT>' ) <EOL> self . assertEqual ( attrs [ '<STR_LIT>' ] , '<STR_LIT:0>' ) <EOL> def test_sqs_longpoll ( self ) : <EOL> c = SQSConnection ( ) <EOL> queue_name = '<STR_LIT>' % int ( time . time ( ) ) <EOL> queue = c . create_queue ( queue_name ) <EOL> self . addCleanup ( c . delete_queue , queue , True ) <EOL> messages = [ ] <EOL> def send_message ( ) : <EOL> messages . append ( <EOL> queue . write ( queue . new_message ( '<STR_LIT>' ) ) ) <EOL> t = Timer ( <NUM_LIT> , send_message ) <EOL> t . start ( ) <EOL> self . addCleanup ( t . join ) <EOL> start = time . time ( ) <EOL> response = queue . read ( wait_time_seconds = <NUM_LIT:10> ) <EOL> end = time . time ( ) <EOL> t . join ( ) <EOL> self . assertEqual ( response . id , messages [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( response . get_body ( ) , messages [ <NUM_LIT:0> ] . get_body ( ) ) <EOL> self . assertTrue ( <NUM_LIT> <= ( end - start ) <= <NUM_LIT> ) <EOL> def test_queue_deletion_affects_full_queues ( self ) : <EOL> conn = SQSConnection ( ) <EOL> initial_count = len ( conn . get_all_queues ( ) ) <EOL> empty = conn . create_queue ( '<STR_LIT>' % int ( time . time ( ) ) ) <EOL> full = conn . create_queue ( '<STR_LIT>' % int ( time . time ( ) ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( len ( conn . get_all_queues ( ) ) , initial_count + <NUM_LIT:2> ) <EOL> m1 = Message ( ) <EOL> m1 . set_body ( '<STR_LIT>' ) <EOL> full . write ( m1 ) <EOL> self . assertEqual ( full . count ( ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( conn . delete_queue ( empty ) ) <EOL> self . assertTrue ( conn . delete_queue ( full ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( len ( conn . get_all_queues ( ) ) , initial_count ) <EOL> def test_get_messages_attributes ( self ) : <EOL> conn = SQSConnection ( ) <EOL> current_timestamp = int ( time . time ( ) ) <EOL> test = self . create_temp_queue ( conn ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . put_queue_message ( test ) <EOL> self . assertEqual ( test . count ( ) , <NUM_LIT:1> ) <EOL> msgs = test . get_messages ( <EOL> num_messages = <NUM_LIT:1> , <EOL> attributes = '<STR_LIT>' <EOL> ) <EOL> for msg in msgs : <EOL> self . assertEqual ( msg . attributes [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> first_rec = msg . attributes [ '<STR_LIT>' ] <EOL> first_rec = int ( first_rec ) / <NUM_LIT:1000> <EOL> self . assertTrue ( first_rec >= current_timestamp ) <EOL> self . put_queue_message ( test ) <EOL> self . assertEqual ( test . count ( ) , <NUM_LIT:1> ) <EOL> msgs = test . get_messages ( <EOL> num_messages = <NUM_LIT:1> , <EOL> attributes = '<STR_LIT>' <EOL> ) <EOL> for msg in msgs : <EOL> self . assertEqual ( msg . attributes [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> with self . assertRaises ( KeyError ) : <EOL> msg . attributes [ '<STR_LIT>' ] <EOL> def test_queue_purge ( self ) : <EOL> conn = SQSConnection ( ) <EOL> test = self . create_temp_queue ( conn ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> for x in range ( <NUM_LIT:0> , <NUM_LIT:4> ) : <EOL> self . put_queue_message ( test ) <EOL> self . assertEqual ( test . count ( ) , <NUM_LIT:4> ) <EOL> conn . purge_queue ( test ) <EOL> self . assertEqual ( test . count ( ) , <NUM_LIT:0> ) <EOL> def create_temp_queue ( self , conn ) : <EOL> current_timestamp = int ( time . time ( ) ) <EOL> queue_name = '<STR_LIT>' % int ( time . time ( ) ) <EOL> test = conn . create_queue ( queue_name ) <EOL> self . addCleanup ( conn . delete_queue , test ) <EOL> return test <EOL> def put_queue_message ( self , queue ) : <EOL> m1 = Message ( ) <EOL> m1 . set_body ( '<STR_LIT>' ) <EOL> queue . write ( m1 ) </s>
