<s> from __future__ import absolute_import , unicode_literals <EOL> try : <EOL> import librabbitmq <EOL> except ImportError : <EOL> librabbitmq = None <EOL> else : <EOL> from kombu . transport import librabbitmq <EOL> from kombu . tests . case import Case , Mock , patch , skip <EOL> @ skip . unless_module ( '<STR_LIT>' ) <EOL> class lrmqCase ( Case ) : <EOL> pass <EOL> class test_Message ( lrmqCase ) : <EOL> def test_init ( self ) : <EOL> chan = Mock ( name = '<STR_LIT>' ) <EOL> message = librabbitmq . Message ( <EOL> chan , { '<STR_LIT>' : <NUM_LIT> } , { '<STR_LIT>' : <NUM_LIT> } , '<STR_LIT:body>' , <EOL> ) <EOL> self . assertEqual ( message . body , '<STR_LIT:body>' ) <EOL> self . assertEqual ( message . delivery_tag , <NUM_LIT> ) <EOL> self . assertEqual ( message . properties [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> class test_Channel ( lrmqCase ) : <EOL> def test_prepare_message ( self ) : <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> chan = librabbitmq . Channel ( conn , <NUM_LIT:1> ) <EOL> self . assertTrue ( chan ) <EOL> body = '<STR_LIT>' <EOL> properties = { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> body2 , props2 = chan . prepare_message ( <EOL> body , properties = properties , <EOL> priority = <NUM_LIT> , <EOL> content_type = '<STR_LIT>' , <EOL> content_encoding = '<STR_LIT>' , <EOL> headers = { '<STR_LIT:H>' : <NUM_LIT:2> } , <EOL> ) <EOL> self . assertEqual ( props2 [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( props2 [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> self . assertEqual ( props2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( props2 [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( props2 [ '<STR_LIT>' ] , { '<STR_LIT:H>' : <NUM_LIT:2> } ) <EOL> self . assertEqual ( body2 , body ) <EOL> body3 , props3 = chan . prepare_message ( body , priority = <NUM_LIT> ) <EOL> self . assertEqual ( props3 [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> self . assertEqual ( body3 , body ) <EOL> class test_Transport ( lrmqCase ) : <EOL> def setup ( self ) : <EOL> super ( test_Transport , self ) . setup ( ) <EOL> self . client = Mock ( name = '<STR_LIT>' ) <EOL> self . T = librabbitmq . Transport ( self . client ) <EOL> def test_driver_version ( self ) : <EOL> self . assertTrue ( self . T . driver_version ( ) ) <EOL> def test_create_channel ( self ) : <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> chan = self . T . create_channel ( conn ) <EOL> self . assertTrue ( chan ) <EOL> conn . channel . assert_called_with ( ) <EOL> def test_drain_events ( self ) : <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> self . T . drain_events ( conn , timeout = <NUM_LIT> ) <EOL> conn . drain_events . assert_called_with ( timeout = <NUM_LIT> ) <EOL> def test_establish_connection_SSL_not_supported ( self ) : <EOL> self . client . ssl = True <EOL> with self . assertRaises ( NotImplementedError ) : <EOL> self . T . establish_connection ( ) <EOL> def test_establish_connection ( self ) : <EOL> self . T . Connection = Mock ( name = '<STR_LIT>' ) <EOL> self . T . client . ssl = False <EOL> self . T . client . port = None <EOL> self . T . client . transport_options = { } <EOL> conn = self . T . establish_connection ( ) <EOL> self . assertEqual ( <EOL> self . T . client . port , <EOL> self . T . default_connection_params [ '<STR_LIT:port>' ] , <EOL> ) <EOL> self . assertEqual ( conn . client , self . T . client ) <EOL> self . assertEqual ( self . T . client . drain_events , conn . drain_events ) <EOL> def test_collect__no_conn ( self ) : <EOL> self . T . client . drain_events = <NUM_LIT> <EOL> self . T . _collect ( None ) <EOL> self . assertIsNone ( self . client . drain_events ) <EOL> self . assertIsNone ( self . T . client ) <EOL> def test_collect__with_conn ( self ) : <EOL> self . T . client . drain_events = <NUM_LIT> <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> chans = conn . channels = { <NUM_LIT:1> : Mock ( name = '<STR_LIT>' ) , <NUM_LIT:2> : Mock ( name = '<STR_LIT>' ) } <EOL> conn . callbacks = { '<STR_LIT:foo>' : Mock ( name = '<STR_LIT>' ) , '<STR_LIT:bar>' : Mock ( name = '<STR_LIT>' ) } <EOL> for i , chan in enumerate ( conn . channels . values ( ) ) : <EOL> chan . connection = i <EOL> with patch ( '<STR_LIT>' ) as close : <EOL> self . T . _collect ( conn ) <EOL> close . assert_called_with ( conn . fileno ( ) ) <EOL> self . assertFalse ( conn . channels ) <EOL> self . assertFalse ( conn . callbacks ) <EOL> for chan in chans . values ( ) : <EOL> self . assertIsNone ( chan . connection ) <EOL> self . assertIsNone ( self . client . drain_events ) <EOL> self . assertIsNone ( self . T . client ) <EOL> with patch ( '<STR_LIT>' ) as close : <EOL> self . T . client = self . client <EOL> close . side_effect = OSError ( ) <EOL> self . T . _collect ( conn ) <EOL> close . assert_called_with ( conn . fileno ( ) ) <EOL> def test_register_with_event_loop ( self ) : <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> loop = Mock ( name = '<STR_LIT>' ) <EOL> self . T . register_with_event_loop ( conn , loop ) <EOL> loop . add_reader . assert_called_with ( <EOL> conn . fileno ( ) , self . T . on_readable , conn , loop , <EOL> ) <EOL> def test_verify_connection ( self ) : <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> conn . connected = True <EOL> self . assertTrue ( self . T . verify_connection ( conn ) ) <EOL> def test_close_connection ( self ) : <EOL> conn = Mock ( name = '<STR_LIT>' ) <EOL> self . client . drain_events = <NUM_LIT> <EOL> self . T . close_connection ( conn ) <EOL> self . assertIsNone ( self . client . drain_events ) <EOL> conn . close . assert_called_with ( ) </s>
