<s> import os <EOL> import sys <EOL> import unittest <EOL> import pickle <EOL> import time <EOL> from itertools import count <EOL> sys . path . insert ( <NUM_LIT:0> , os . pardir ) <EOL> sys . path . append ( os . getcwd ( ) ) <EOL> from carrot . messaging import Consumer , Publisher , ConsumerSet <EOL> from carrot import serialization <EOL> from tests . utils import establish_test_connection <EOL> class AdvancedDataType ( object ) : <EOL> def __init__ ( self , something ) : <EOL> self . data = something <EOL> def fetch_next_message ( consumer ) : <EOL> while True : <EOL> message = consumer . fetch ( ) <EOL> if message : <EOL> return message <EOL> class BackendMessagingCase ( unittest . TestCase ) : <EOL> nextq = count ( <NUM_LIT:1> ) . next <EOL> def setUp ( self ) : <EOL> self . conn = establish_test_connection ( ) <EOL> self . queue = TEST_QUEUE <EOL> self . exchange = TEST_EXCHANGE <EOL> self . routing_key = TEST_ROUTING_KEY <EOL> def create_consumer ( self , ** options ) : <EOL> queue = "<STR_LIT>" % ( self . queue , self . nextq ( ) ) <EOL> return Consumer ( connection = self . conn , <EOL> queue = queue , exchange = self . exchange , <EOL> routing_key = self . routing_key , ** options ) <EOL> def create_consumerset ( self , queues = { } , consumers = [ ] , ** options ) : <EOL> return ConsumerSet ( connection = self . conn , <EOL> from_dict = queues , consumers = consumers , ** options ) <EOL> def create_publisher ( self , exchange = None , routing_key = None , ** options ) : <EOL> exchange = exchange or self . exchange <EOL> routing_key = routing_key or self . routing_key <EOL> return Publisher ( connection = self . conn , <EOL> exchange = exchange , routing_key = routing_key , <EOL> ** options ) <EOL> def test_regression_implied_auto_delete ( self ) : <EOL> consumer = self . create_consumer ( exclusive = True , auto_declare = False ) <EOL> self . assertTrue ( consumer . auto_delete , "<STR_LIT>" ) <EOL> consumer . close ( ) <EOL> consumer = self . create_consumer ( durable = True , auto_delete = False , <EOL> auto_declare = False ) <EOL> self . assertFalse ( consumer . auto_delete , <EOL> """<STR_LIT>""" ) <EOL> consumer . close ( ) <EOL> def test_consumer_options ( self ) : <EOL> opposite_defaults = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> consumer = Consumer ( connection = self . conn , ** opposite_defaults ) <EOL> for opt_name , opt_value in opposite_defaults . items ( ) : <EOL> self . assertEquals ( getattr ( consumer , opt_name ) , opt_value ) <EOL> consumer . close ( ) <EOL> def test_consumer_backend ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> self . assertTrue ( consumer . backend . connection is self . conn ) <EOL> consumer . close ( ) <EOL> def test_consumer_queue_declared ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> self . assertTrue ( consumer . backend . queue_exists ( consumer . queue ) ) <EOL> consumer . close ( ) <EOL> def test_consumer_callbacks ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> self . assertRaises ( NotImplementedError , consumer . receive , { } , { } ) <EOL> callback1_scratchpad = { } <EOL> def callback1 ( message_data , message ) : <EOL> callback1_scratchpad [ "<STR_LIT>" ] = message_data <EOL> callback2_scratchpad = { } <EOL> def callback2 ( message_data , message ) : <EOL> callback2_scratchpad . update ( { "<STR_LIT>" : message . delivery_tag , <EOL> "<STR_LIT>" : message . body } ) <EOL> self . assertFalse ( consumer . callbacks , "<STR_LIT>" ) <EOL> consumer . register_callback ( callback1 ) <EOL> consumer . register_callback ( callback2 ) <EOL> self . assertEquals ( len ( consumer . callbacks ) , <NUM_LIT:2> , "<STR_LIT>" ) <EOL> self . assertTrue ( consumer . callbacks [ <NUM_LIT:0> ] is callback1 , <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( consumer . callbacks [ <NUM_LIT:1> ] is callback2 , <EOL> "<STR_LIT>" ) <EOL> body = { "<STR_LIT:foo>" : "<STR_LIT:bar>" } <EOL> message = self . create_raw_message ( publisher , body , "<STR_LIT>" ) <EOL> consumer . _receive_callback ( message ) <EOL> self . assertEquals ( callback1_scratchpad . get ( "<STR_LIT>" ) , body , <EOL> "<STR_LIT>" ) <EOL> self . assertEquals ( callback2_scratchpad . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def create_raw_message ( self , publisher , body , delivery_tag ) : <EOL> raw_message = publisher . create_message ( body ) <EOL> raw_message . delivery_tag = delivery_tag <EOL> return raw_message <EOL> def test_empty_queue_returns_None ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> consumer . discard_all ( ) <EOL> self . assertFalse ( consumer . fetch ( ) ) <EOL> consumer . close ( ) <EOL> def test_custom_serialization_scheme ( self ) : <EOL> serialization . registry . register ( '<STR_LIT>' , <EOL> pickle . dumps , pickle . loads , <EOL> content_type = '<STR_LIT>' , <EOL> content_encoding = '<STR_LIT>' ) <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> data = { "<STR_LIT:string>" : "<STR_LIT>" , <EOL> "<STR_LIT:int>" : <NUM_LIT:10> , <EOL> "<STR_LIT:float>" : <NUM_LIT> , <EOL> "<STR_LIT>" : u"<STR_LIT>" , <EOL> "<STR_LIT>" : AdvancedDataType ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> "<STR_LIT>" : Exception ( "<STR_LIT>" ) , <EOL> } <EOL> publisher . send ( data , serializer = '<STR_LIT>' ) <EOL> message = fetch_next_message ( consumer ) <EOL> backend = self . conn . create_backend ( ) <EOL> self . assertTrue ( isinstance ( message , backend . Message ) ) <EOL> self . assertEquals ( message . payload . get ( "<STR_LIT:int>" ) , <NUM_LIT:10> ) <EOL> self . assertEquals ( message . content_type , '<STR_LIT>' ) <EOL> self . assertEquals ( message . content_encoding , '<STR_LIT>' ) <EOL> decoded_data = message . decode ( ) <EOL> self . assertEquals ( decoded_data . get ( "<STR_LIT:string>" ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEquals ( decoded_data . get ( "<STR_LIT:int>" ) , <NUM_LIT:10> ) <EOL> self . assertEquals ( decoded_data . get ( "<STR_LIT:float>" ) , <NUM_LIT> ) <EOL> self . assertEquals ( decoded_data . get ( "<STR_LIT>" ) , <EOL> u"<STR_LIT>" ) <EOL> self . assertEquals ( decoded_data . get ( "<STR_LIT>" ) , <EOL> set ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) ) <EOL> self . assertTrue ( isinstance ( decoded_data . get ( "<STR_LIT>" ) , Exception ) ) <EOL> self . assertEquals ( decoded_data . get ( "<STR_LIT>" ) . args [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( isinstance ( decoded_data . get ( "<STR_LIT>" ) , <EOL> AdvancedDataType ) ) <EOL> self . assertEquals ( decoded_data [ "<STR_LIT>" ] . data , "<STR_LIT>" ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_consumer_fetch ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> data = { "<STR_LIT:string>" : "<STR_LIT>" , <EOL> "<STR_LIT:int>" : <NUM_LIT:10> , <EOL> "<STR_LIT:float>" : <NUM_LIT> , <EOL> "<STR_LIT>" : u"<STR_LIT>" , <EOL> } <EOL> publisher . send ( data ) <EOL> message = fetch_next_message ( consumer ) <EOL> backend = self . conn . create_backend ( ) <EOL> self . assertTrue ( isinstance ( message , backend . Message ) ) <EOL> self . assertEquals ( message . decode ( ) , data ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_consumer_process_next ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> scratchpad = { } <EOL> def callback ( message_data , message ) : <EOL> scratchpad [ "<STR_LIT>" ] = message . delivery_tag <EOL> consumer . register_callback ( callback ) <EOL> publisher . send ( { "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } } ) <EOL> while True : <EOL> message = consumer . fetch ( enable_callbacks = True ) <EOL> if message : <EOL> break <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT>" ) , <EOL> message . delivery_tag ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_consumer_discard_all ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> for i in xrange ( <NUM_LIT:100> ) : <EOL> publisher . send ( { "<STR_LIT:foo>" : "<STR_LIT:bar>" } ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> self . assertEquals ( consumer . discard_all ( ) , <NUM_LIT:100> ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_iterqueue ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> num = consumer . discard_all ( ) <EOL> it = consumer . iterqueue ( limit = <NUM_LIT:100> ) <EOL> consumer . register_callback ( lambda * args : args ) <EOL> for i in xrange ( <NUM_LIT:100> ) : <EOL> publisher . send ( { "<STR_LIT>" % i : "<STR_LIT>" % i } ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> for i in xrange ( <NUM_LIT:100> ) : <EOL> try : <EOL> message = it . next ( ) <EOL> data = message . decode ( ) <EOL> self . assertTrue ( "<STR_LIT>" % i in data , "<STR_LIT>" % i ) <EOL> self . assertEquals ( data . get ( "<STR_LIT>" % i ) , "<STR_LIT>" % i ) <EOL> except StopIteration : <EOL> self . assertTrue ( False , "<STR_LIT>" ) <EOL> self . assertRaises ( StopIteration , it . next ) <EOL> it = consumer . iterqueue ( ) <EOL> self . assertRaises ( StopIteration , it . next ) <EOL> it = consumer . iterqueue ( infinite = True ) <EOL> self . assertTrue ( it . next ( ) is None , <EOL> "<STR_LIT>" ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_publisher_message_priority ( self ) : <EOL> consumer = self . create_consumer ( ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> m = publisher . create_message ( "<STR_LIT:foo>" , priority = <NUM_LIT:9> ) <EOL> publisher . send ( { "<STR_LIT:foo>" : "<STR_LIT:bar>" } , routing_key = "<STR_LIT>" , priority = <NUM_LIT:9> , <EOL> mandatory = False , immediate = False ) <EOL> consumer . discard_all ( ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_backend_survives_channel_close_regr17 ( self ) : <EOL> """<STR_LIT>""" <EOL> backend = self . create_publisher ( ) . backend <EOL> assert not backend . queue_exists ( '<STR_LIT>' ) <EOL> assert not backend . queue_exists ( '<STR_LIT>' ) <EOL> def disabled_publisher_mandatory_flag_regr16 ( self ) : <EOL> """<STR_LIT>""" <EOL> routing_key = '<STR_LIT>' <EOL> assert self . conn . connection is not None <EOL> message = { '<STR_LIT:foo>' : '<STR_LIT>' } <EOL> assert not self . create_consumer ( ) . backend . queue_exists ( routing_key ) <EOL> publisher = self . create_publisher ( ) <EOL> publisher . send ( message , routing_key = routing_key , mandatory = False ) <EOL> self . assertRaises ( Exception , publisher . send , message , <EOL> routing_key = routing_key , mandatory = True ) <EOL> consumer = Consumer ( connection = self . conn , <EOL> queue = routing_key , exchange = self . exchange , <EOL> routing_key = routing_key , durable = False , <EOL> exclusive = True ) <EOL> assert self . create_consumer ( ) . backend . queue_exists ( routing_key ) <EOL> publisher . send ( message , routing_key = routing_key , mandatory = True ) <EOL> def test_consumer_auto_ack ( self ) : <EOL> consumer = self . create_consumer ( auto_ack = True ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> publisher . send ( { "<STR_LIT:foo>" : "<STR_LIT>" } ) <EOL> message = fetch_next_message ( consumer ) <EOL> self . assertEquals ( message . _state , "<STR_LIT>" ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer = self . create_consumer ( auto_ack = False ) <EOL> publisher . send ( { "<STR_LIT:foo>" : "<STR_LIT>" } ) <EOL> message = fetch_next_message ( consumer ) <EOL> self . assertEquals ( message . _state , "<STR_LIT>" ) <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_consumer_consume ( self ) : <EOL> consumer = self . create_consumer ( auto_ack = True ) <EOL> publisher = self . create_publisher ( ) <EOL> consumer . discard_all ( ) <EOL> data = { "<STR_LIT:foo>" : "<STR_LIT>" } <EOL> publisher . send ( data ) <EOL> try : <EOL> data2 = { "<STR_LIT>" : "<STR_LIT>" } <EOL> publisher . send ( data2 ) <EOL> scratchpad = { } <EOL> def callback ( message_data , message ) : <EOL> scratchpad [ "<STR_LIT:data>" ] = message_data <EOL> consumer . register_callback ( callback ) <EOL> it = consumer . iterconsume ( ) <EOL> it . next ( ) <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , data ) <EOL> it . next ( ) <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , data2 ) <EOL> consumer . close ( ) <EOL> consumer = self . create_consumer ( auto_ack = True ) <EOL> consumer . register_callback ( callback ) <EOL> consumer . discard_all ( ) <EOL> scratchpad = { } <EOL> it = consumer . iterconsume ( limit = <NUM_LIT:4> ) <EOL> publisher . send ( data ) <EOL> publisher . send ( data2 ) <EOL> publisher . send ( data ) <EOL> publisher . send ( data2 ) <EOL> publisher . send ( data ) <EOL> it . next ( ) <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , data ) <EOL> it . next ( ) <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , data2 ) <EOL> it . next ( ) <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , data ) <EOL> it . next ( ) <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , data2 ) <EOL> self . assertRaises ( StopIteration , it . next ) <EOL> finally : <EOL> consumer . close ( ) <EOL> publisher . close ( ) <EOL> def test_consumerset_iterconsume ( self ) : <EOL> consumerset = self . create_consumerset ( queues = { <EOL> "<STR_LIT:bar>" : { <EOL> "<STR_LIT>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT:foo>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } } ) <EOL> publisher = self . create_publisher ( exchange = "<STR_LIT:foo>" ) <EOL> consumerset . discard_all ( ) <EOL> scratchpad = { } <EOL> def callback ( message_data , message ) : <EOL> scratchpad [ "<STR_LIT:data>" ] = message_data <EOL> def assertDataIs ( what ) : <EOL> self . assertEquals ( scratchpad . get ( "<STR_LIT:data>" ) , what ) <EOL> try : <EOL> consumerset . register_callback ( callback ) <EOL> it = consumerset . iterconsume ( ) <EOL> publisher . send ( { "<STR_LIT>" : "<STR_LIT>" } , routing_key = "<STR_LIT>" ) <EOL> it . next ( ) <EOL> assertDataIs ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> publisher . send ( { "<STR_LIT>" : "<STR_LIT>" } , routing_key = "<STR_LIT>" ) <EOL> publisher . send ( { "<STR_LIT>" : "<STR_LIT>" } , routing_key = "<STR_LIT>" ) <EOL> publisher . send ( { "<STR_LIT>" : "<STR_LIT>" } , routing_key = "<STR_LIT>" ) <EOL> publisher . send ( { "<STR_LIT>" : "<STR_LIT>" } , routing_key = "<STR_LIT>" ) <EOL> it . next ( ) <EOL> assertDataIs ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> it . next ( ) <EOL> assertDataIs ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> it . next ( ) <EOL> assertDataIs ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> it . next ( ) <EOL> assertDataIs ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> finally : <EOL> consumerset . close ( ) <EOL> publisher . close ( ) </s>
