<s> from iron_mq import * <EOL> import unittest <EOL> import random <EOL> import time <EOL> class TestIronMQ ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mq = IronMQ ( ) <EOL> self . random_number = str ( int ( random . random ( ) * <NUM_LIT:10> ** <NUM_LIT:10> ) ) <EOL> def test_postMessage ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> old_size = q . size ( ) <EOL> q . post ( "<STR_LIT>" ) <EOL> self . assertEqual ( old_size , q . size ( ) - <NUM_LIT:1> ) <EOL> def test_addAlerts ( self ) : <EOL> queue_name = "<STR_LIT>" % time . time ( ) <EOL> self . mq . create_queue ( queue_name ) <EOL> q = self . mq . queue ( queue_name ) <EOL> fixed_alert = [ { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1000> , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> response = q . add_alerts ( * fixed_alert ) <EOL> self . assertTrue ( "<STR_LIT>" in response ) <EOL> def test_infoShouldReturnAlerts ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> fixed_alert = [ { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1000> , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> q . add_alerts ( * fixed_alert ) <EOL> info = q . info ( ) <EOL> self . assertTrue ( '<STR_LIT>' in info ) <EOL> self . assertEqual ( len ( info [ '<STR_LIT>' ] ) , <NUM_LIT:1> ) <EOL> def test_removeAlerts ( self ) : <EOL> queue_name = "<STR_LIT>" % time . time ( ) <EOL> self . mq . create_queue ( queue_name ) <EOL> q = self . mq . queue ( queue_name ) <EOL> q . add_alerts ( { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1000> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> alerts = [ alert [ '<STR_LIT:id>' ] for alert in q . info ( ) [ '<STR_LIT>' ] ] [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> last_alert = q . info ( ) [ '<STR_LIT>' ] [ <NUM_LIT:2> ] <EOL> result = q . remove_alerts ( * alerts ) <EOL> self . assertEqual ( result [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( q . info ( ) [ '<STR_LIT>' ] ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( q . info ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] , last_alert [ '<STR_LIT:id>' ] ) <EOL> def test_touchMessage ( self ) : <EOL> msg = "<STR_LIT>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> id = q . post ( msg ) <EOL> message = q . reserve ( ) <EOL> response = q . touch ( message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] , message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( "<STR_LIT>" , response [ "<STR_LIT>" ] ) <EOL> def test_touchMessageTwice ( self ) : <EOL> msg = "<STR_LIT>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> id = q . post ( msg ) <EOL> message = q . reserve ( ) <EOL> response = q . touch ( message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] , message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) <EOL> response = q . touch ( message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] , response [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( "<STR_LIT>" , response [ "<STR_LIT>" ] ) <EOL> def test_releaseMessage ( self ) : <EOL> msg = "<STR_LIT>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> id = q . post ( msg ) <EOL> message = q . reserve ( ) <EOL> response = q . release ( message_id = message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] , reservation_id = message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( "<STR_LIT>" , response [ "<STR_LIT>" ] ) <EOL> def test_addSubscribers ( self ) : <EOL> queue_name = "<STR_LIT>" % time . time ( ) <EOL> subscribers = [ { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:url>' : '<STR_LIT>' } ] <EOL> self . mq . create_queue ( queue_name , { '<STR_LIT>' : { '<STR_LIT>' : subscribers } } ) <EOL> q = self . mq . queue ( queue_name ) <EOL> response = q . add_subscribers ( * [ { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:url>' : '<STR_LIT>' } ] ) <EOL> self . assertEqual ( response [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> info = q . info ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( info [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) ) <EOL> def test_removeSubscribers ( self ) : <EOL> queue_name = "<STR_LIT>" % time . time ( ) <EOL> subscribers = [ { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' } ] <EOL> self . mq . create_queue ( queue_name , { '<STR_LIT>' : { '<STR_LIT>' : subscribers } } ) <EOL> q = self . mq . queue ( queue_name ) <EOL> response = q . remove_subscribers ( * [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( response [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> info = q . info ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( info [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) ) <EOL> def test_getMessage ( self ) : <EOL> msg = "<STR_LIT:%s>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> q . post ( msg ) <EOL> message = q . get ( ) <EOL> message = "<STR_LIT:%s>" % message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:body>" ] <EOL> self . assertEqual ( msg , message ) <EOL> def test_reserveMessages ( self ) : <EOL> msg = "<STR_LIT:%s>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> ids = q . post ( msg + str ( <NUM_LIT:0> ) , msg + str ( <NUM_LIT:1> ) , msg + str ( <NUM_LIT:2> ) ) <EOL> messages = q . reserve ( <NUM_LIT:3> ) <EOL> for idx , val in enumerate ( messages [ "<STR_LIT>" ] ) : <EOL> self . assertEqual ( msg + str ( idx ) , val [ "<STR_LIT:body>" ] ) <EOL> self . assertEqual ( ids [ "<STR_LIT>" ] [ idx ] , val [ "<STR_LIT:id>" ] ) <EOL> self . assertTrue ( len ( val [ "<STR_LIT>" ] ) > <NUM_LIT:0> ) <EOL> self . assertEqual ( val [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> def test_reserveMessage ( self ) : <EOL> msg = "<STR_LIT>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> id = q . post ( msg ) <EOL> message = q . reserve ( ) <EOL> self . assertEqual ( msg , message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:body>" ] ) <EOL> self . assertEqual ( id [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] ) <EOL> def test_getMessageTimeout ( self ) : <EOL> msg = "<STR_LIT:%s>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> q . post ( msg ) <EOL> message = q . get ( timeout = <NUM_LIT> ) <EOL> message = "<STR_LIT:%s>" % message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:body>" ] <EOL> self . assertEqual ( msg , message ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> m2 = q . get ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( m2 [ "<STR_LIT>" ] ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> m3 = q . get ( ) <EOL> self . assertEqual ( msg , message ) <EOL> def test_deleteMessage ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> size = q . size ( ) <EOL> msg = q . post ( "<STR_LIT:test>" ) <EOL> self . assertEqual ( size , q . size ( ) - <NUM_LIT:1> ) <EOL> q . delete ( msg [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( size , q . size ( ) ) <EOL> def test_deleteReservedMessage ( self ) : <EOL> msg = "<STR_LIT:%s>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> id = q . post ( msg ) <EOL> message = q . reserve ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> self . assertEqual ( <NUM_LIT:1> , q . size ( ) ) <EOL> reservation_id = message [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> q . delete ( id [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , reservation_id ) <EOL> self . assertEqual ( <NUM_LIT:0> , q . size ( ) ) <EOL> def test_clearQueue ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . post ( "<STR_LIT:%s>" % time . time ( ) ) <EOL> self . assertTrue ( q . size ( ) > <NUM_LIT:0> ) <EOL> q . clear ( ) <EOL> self . assertEqual ( q . size ( ) , <NUM_LIT:0> ) <EOL> def test_deprecated ( self ) : <EOL> self . assertEqual ( self . mq . queues ( ) , self . mq . getQueues ( ) ) <EOL> name = '<STR_LIT>' <EOL> q = self . mq . queue ( name ) <EOL> q . clear ( ) <EOL> self . assertEqual ( '<STR_LIT>' , self . mq . clearQueue ( name ) [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , q . size ( ) ) <EOL> info = self . mq . getQueueDetails ( name ) <EOL> self . assertEqual ( q . info ( ) , info ) <EOL> msg_id = self . mq . postMessage ( name , [ '<STR_LIT>' ] ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> resp = self . mq . getMessage ( name ) <EOL> self . assertEqual ( msg_id , resp [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] [ '<STR_LIT:id>' ] ) <EOL> def test_postAndDeleteMultipleMessages ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> old_size = q . size ( ) <EOL> ids = q . post ( "<STR_LIT>" , "<STR_LIT>" ) [ "<STR_LIT>" ] <EOL> self . assertEqual ( old_size , q . size ( ) - <NUM_LIT:2> ) <EOL> q . delete_multiple ( ids = ids ) <EOL> self . assertEqual ( old_size , q . size ( ) ) <EOL> def test_deleteMultipleReservedMessages ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> old_size = q . size ( ) <EOL> q . post ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> messages = q . reserve ( <NUM_LIT:2> , <NUM_LIT> ) <EOL> self . assertEqual ( old_size , q . size ( ) - <NUM_LIT:2> ) <EOL> q . delete_multiple ( messages = messages ) <EOL> self . assertEqual ( old_size , q . size ( ) ) <EOL> def test_getMessageById ( self ) : <EOL> body = "<STR_LIT:%s>" % time . time ( ) <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> response_post = q . post ( body ) <EOL> message = q . get_message_by_id ( response_post [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( body , message [ "<STR_LIT:body>" ] ) <EOL> def test_peekMessages ( self ) : <EOL> q = self . mq . queue ( "<STR_LIT>" ) <EOL> q . clear ( ) <EOL> q . post ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> response = q . peek ( <NUM_LIT:2> ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( response [ "<STR_LIT>" ] ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
