<s> from __future__ import absolute_import , unicode_literals <EOL> import pickle <EOL> from kombu import Connection , Exchange , Producer , Queue , binding <EOL> from kombu . exceptions import NotBoundError <EOL> from kombu . serialization import registry <EOL> from . case import Case , Mock , call <EOL> from . mocks import Transport <EOL> def get_conn ( ) : <EOL> return Connection ( transport = Transport ) <EOL> class test_binding ( Case ) : <EOL> def test_constructor ( self ) : <EOL> x = binding ( <EOL> Exchange ( '<STR_LIT:foo>' ) , '<STR_LIT>' , <EOL> arguments = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> unbind_arguments = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> ) <EOL> self . assertEqual ( x . exchange , Exchange ( '<STR_LIT:foo>' ) ) <EOL> self . assertEqual ( x . routing_key , '<STR_LIT>' ) <EOL> self . assertDictEqual ( x . arguments , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertDictEqual ( x . unbind_arguments , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_declare ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> x = binding ( Exchange ( '<STR_LIT:foo>' ) , '<STR_LIT>' ) <EOL> x . declare ( chan ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_declare_no_exchange ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> x = binding ( ) <EOL> x . declare ( chan ) <EOL> self . assertNotIn ( '<STR_LIT>' , chan ) <EOL> def test_bind ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> x = binding ( Exchange ( '<STR_LIT:foo>' ) ) <EOL> x . bind ( Exchange ( '<STR_LIT:bar>' ) ( chan ) ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_unbind ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> x = binding ( Exchange ( '<STR_LIT:foo>' ) ) <EOL> x . unbind ( Exchange ( '<STR_LIT:bar>' ) ( chan ) ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_repr ( self ) : <EOL> b = binding ( Exchange ( '<STR_LIT:foo>' ) , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT:foo>' , repr ( b ) ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( b ) ) <EOL> class test_Exchange ( Case ) : <EOL> def test_bound ( self ) : <EOL> exchange = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . assertFalse ( exchange . is_bound ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( exchange ) ) <EOL> chan = get_conn ( ) . channel ( ) <EOL> bound = exchange . bind ( chan ) <EOL> self . assertTrue ( bound . is_bound ) <EOL> self . assertIs ( bound . channel , chan ) <EOL> self . assertIn ( '<STR_LIT>' % ( chan . channel_id , ) , <EOL> repr ( bound ) ) <EOL> def test_hash ( self ) : <EOL> self . assertEqual ( hash ( Exchange ( '<STR_LIT:a>' ) ) , hash ( Exchange ( '<STR_LIT:a>' ) ) ) <EOL> self . assertNotEqual ( hash ( Exchange ( '<STR_LIT:a>' ) ) , hash ( Exchange ( '<STR_LIT:b>' ) ) ) <EOL> def test_can_cache_declaration ( self ) : <EOL> self . assertTrue ( Exchange ( '<STR_LIT:a>' , durable = True ) . can_cache_declaration ) <EOL> self . assertTrue ( Exchange ( '<STR_LIT:a>' , durable = False ) . can_cache_declaration ) <EOL> self . assertFalse ( Exchange ( '<STR_LIT:a>' , auto_delete = True ) . can_cache_declaration ) <EOL> self . assertFalse ( <EOL> Exchange ( <EOL> '<STR_LIT:a>' , durable = True , auto_delete = True <EOL> ) . can_cache_declaration , <EOL> ) <EOL> def test_pickle ( self ) : <EOL> e1 = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> e2 = pickle . loads ( pickle . dumps ( e1 ) ) <EOL> self . assertEqual ( e1 , e2 ) <EOL> def test_eq ( self ) : <EOL> e1 = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> e2 = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . assertEqual ( e1 , e2 ) <EOL> e3 = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . assertNotEqual ( e1 , e3 ) <EOL> self . assertEqual ( e1 . __eq__ ( True ) , NotImplemented ) <EOL> def test_revive ( self ) : <EOL> exchange = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> conn = get_conn ( ) <EOL> chan = conn . channel ( ) <EOL> exchange . revive ( chan ) <EOL> self . assertFalse ( exchange . is_bound ) <EOL> self . assertIsNone ( exchange . _channel ) <EOL> bound = exchange . bind ( chan ) <EOL> self . assertTrue ( bound . is_bound ) <EOL> self . assertIs ( bound . channel , chan ) <EOL> chan2 = conn . channel ( ) <EOL> bound . revive ( chan2 ) <EOL> self . assertTrue ( bound . is_bound ) <EOL> self . assertIs ( bound . _channel , chan2 ) <EOL> def test_assert_is_bound ( self ) : <EOL> exchange = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> with self . assertRaises ( NotBoundError ) : <EOL> exchange . declare ( ) <EOL> conn = get_conn ( ) <EOL> chan = conn . channel ( ) <EOL> exchange . bind ( chan ) . declare ( ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_set_transient_delivery_mode ( self ) : <EOL> exc = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' , delivery_mode = '<STR_LIT>' ) <EOL> self . assertEqual ( exc . delivery_mode , Exchange . TRANSIENT_DELIVERY_MODE ) <EOL> def test_set_passive_mode ( self ) : <EOL> exc = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' , passive = True ) <EOL> self . assertTrue ( exc . passive ) <EOL> def test_set_persistent_delivery_mode ( self ) : <EOL> exc = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' , delivery_mode = '<STR_LIT>' ) <EOL> self . assertEqual ( exc . delivery_mode , Exchange . PERSISTENT_DELIVERY_MODE ) <EOL> def test_bind_at_instantiation ( self ) : <EOL> self . assertTrue ( Exchange ( '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) . is_bound ) <EOL> def test_create_message ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> Exchange ( '<STR_LIT:foo>' , channel = chan ) . Message ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_publish ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> Exchange ( '<STR_LIT:foo>' , channel = chan ) . publish ( '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_delete ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> Exchange ( '<STR_LIT:foo>' , channel = chan ) . delete ( ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test__repr__ ( self ) : <EOL> b = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( b ) ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( b ) ) <EOL> def test_bind_to ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> bar = Exchange ( '<STR_LIT:bar>' , '<STR_LIT>' ) <EOL> foo ( chan ) . bind_to ( bar ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_bind_to_by_name ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> foo ( chan ) . bind_to ( '<STR_LIT:bar>' ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_unbind_from ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> bar = Exchange ( '<STR_LIT:bar>' , '<STR_LIT>' ) <EOL> foo ( chan ) . unbind_from ( bar ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_unbind_from_by_name ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> foo ( chan ) . unbind_from ( '<STR_LIT:bar>' ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_declare__no_declare ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' , no_declare = True ) <EOL> foo ( chan ) . declare ( ) <EOL> self . assertNotIn ( '<STR_LIT>' , chan ) <EOL> def test_declare__internal_exchange ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> foo ( chan ) . declare ( ) <EOL> self . assertNotIn ( '<STR_LIT>' , chan ) <EOL> def test_declare ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> foo = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' , no_declare = False ) <EOL> foo ( chan ) . declare ( ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> class test_Queue ( Case ) : <EOL> def setup ( self ) : <EOL> self . exchange = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_hash ( self ) : <EOL> self . assertEqual ( hash ( Queue ( '<STR_LIT:a>' ) ) , hash ( Queue ( '<STR_LIT:a>' ) ) ) <EOL> self . assertNotEqual ( hash ( Queue ( '<STR_LIT:a>' ) ) , hash ( Queue ( '<STR_LIT:b>' ) ) ) <EOL> def test_repr_with_bindings ( self ) : <EOL> ex = Exchange ( '<STR_LIT:foo>' ) <EOL> x = Queue ( '<STR_LIT:foo>' , bindings = [ ex . binding ( '<STR_LIT:A>' ) , ex . binding ( '<STR_LIT:B>' ) ] ) <EOL> self . assertTrue ( repr ( x ) ) <EOL> def test_anonymous ( self ) : <EOL> chan = Mock ( ) <EOL> x = Queue ( bindings = [ binding ( Exchange ( '<STR_LIT:foo>' ) , '<STR_LIT>' ) ] ) <EOL> chan . queue_declare . return_value = '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> <EOL> xx = x ( chan ) <EOL> xx . declare ( ) <EOL> self . assertEqual ( xx . name , '<STR_LIT>' ) <EOL> def test_basic_get__accept_disallowed ( self ) : <EOL> conn = Connection ( '<STR_LIT>' ) <EOL> q = Queue ( '<STR_LIT:foo>' , exchange = self . exchange ) <EOL> p = Producer ( conn ) <EOL> p . publish ( <EOL> { '<STR_LIT>' : object ( ) } , <EOL> declare = [ q ] , exchange = self . exchange , serializer = '<STR_LIT>' , <EOL> ) <EOL> message = q ( conn ) . get ( no_ack = True ) <EOL> self . assertIsNotNone ( message ) <EOL> with self . assertRaises ( q . ContentDisallowed ) : <EOL> message . decode ( ) <EOL> def test_basic_get__accept_allowed ( self ) : <EOL> conn = Connection ( '<STR_LIT>' ) <EOL> q = Queue ( '<STR_LIT:foo>' , exchange = self . exchange ) <EOL> p = Producer ( conn ) <EOL> p . publish ( <EOL> { '<STR_LIT>' : object ( ) } , <EOL> declare = [ q ] , exchange = self . exchange , serializer = '<STR_LIT>' , <EOL> ) <EOL> message = q ( conn ) . get ( accept = [ '<STR_LIT>' ] , no_ack = True ) <EOL> self . assertIsNotNone ( message ) <EOL> payload = message . decode ( ) <EOL> self . assertTrue ( payload [ '<STR_LIT>' ] ) <EOL> def test_when_bound_but_no_exchange ( self ) : <EOL> q = Queue ( '<STR_LIT:a>' ) <EOL> q . exchange = None <EOL> self . assertIsNone ( q . when_bound ( ) ) <EOL> def test_declare_but_no_exchange ( self ) : <EOL> q = Queue ( '<STR_LIT:a>' ) <EOL> q . queue_declare = Mock ( ) <EOL> q . queue_bind = Mock ( ) <EOL> q . exchange = None <EOL> q . declare ( ) <EOL> q . queue_declare . assert_called_with ( False , passive = False ) <EOL> def test_declare__no_declare ( self ) : <EOL> q = Queue ( '<STR_LIT:a>' , no_declare = True ) <EOL> q . queue_declare = Mock ( ) <EOL> q . queue_bind = Mock ( ) <EOL> q . exchange = None <EOL> q . declare ( ) <EOL> q . queue_declare . assert_not_called ( ) <EOL> q . queue_bind . assert_not_called ( ) <EOL> def test_bind_to_when_name ( self ) : <EOL> chan = Mock ( ) <EOL> q = Queue ( '<STR_LIT:a>' ) <EOL> q ( chan ) . bind_to ( '<STR_LIT>' ) <EOL> chan . queue_bind . assert_called ( ) <EOL> def test_get_when_no_m2p ( self ) : <EOL> chan = Mock ( ) <EOL> q = Queue ( '<STR_LIT:a>' ) ( chan ) <EOL> chan . message_to_python = None <EOL> self . assertTrue ( q . get ( ) ) <EOL> def test_multiple_bindings ( self ) : <EOL> chan = Mock ( ) <EOL> q = Queue ( '<STR_LIT>' , [ <EOL> binding ( Exchange ( '<STR_LIT>' ) , '<STR_LIT>' ) , <EOL> binding ( Exchange ( '<STR_LIT>' ) , '<STR_LIT>' ) , <EOL> binding ( Exchange ( '<STR_LIT>' ) , '<STR_LIT>' ) , <EOL> ] ) <EOL> q ( chan ) . declare ( ) <EOL> self . assertIn ( <EOL> call ( <EOL> nowait = False , <EOL> exchange = '<STR_LIT>' , <EOL> auto_delete = False , <EOL> passive = False , <EOL> arguments = None , <EOL> type = '<STR_LIT>' , <EOL> durable = True , <EOL> ) , <EOL> chan . exchange_declare . call_args_list , <EOL> ) <EOL> def test_can_cache_declaration ( self ) : <EOL> self . assertTrue ( Queue ( '<STR_LIT:a>' , durable = True ) . can_cache_declaration ) <EOL> self . assertTrue ( Queue ( '<STR_LIT:a>' , durable = False ) . can_cache_declaration ) <EOL> def test_eq ( self ) : <EOL> q1 = Queue ( '<STR_LIT>' , Exchange ( '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> q2 = Queue ( '<STR_LIT>' , Exchange ( '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( q1 , q2 ) <EOL> self . assertEqual ( q1 . __eq__ ( True ) , NotImplemented ) <EOL> q3 = Queue ( '<STR_LIT>' , Exchange ( '<STR_LIT>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertNotEqual ( q1 , q3 ) <EOL> def test_exclusive_implies_auto_delete ( self ) : <EOL> self . assertTrue ( <EOL> Queue ( '<STR_LIT:foo>' , self . exchange , exclusive = True ) . auto_delete , <EOL> ) <EOL> def test_binds_at_instantiation ( self ) : <EOL> self . assertTrue ( Queue ( '<STR_LIT:foo>' , self . exchange , <EOL> channel = get_conn ( ) . channel ( ) ) . is_bound ) <EOL> def test_also_binds_exchange ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange ) <EOL> self . assertFalse ( b . is_bound ) <EOL> self . assertFalse ( b . exchange . is_bound ) <EOL> b = b . bind ( chan ) <EOL> self . assertTrue ( b . is_bound ) <EOL> self . assertTrue ( b . exchange . is_bound ) <EOL> self . assertIs ( b . channel , b . exchange . channel ) <EOL> self . assertIsNot ( b . exchange , self . exchange ) <EOL> def test_declare ( self ) : <EOL> chan = get_conn ( ) . channel ( ) <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = chan ) <EOL> self . assertTrue ( b . is_bound ) <EOL> b . declare ( ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> self . assertIn ( '<STR_LIT>' , chan ) <EOL> def test_get ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) <EOL> b . get ( ) <EOL> self . assertIn ( '<STR_LIT>' , b . channel ) <EOL> def test_purge ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) <EOL> b . purge ( ) <EOL> self . assertIn ( '<STR_LIT>' , b . channel ) <EOL> def test_consume ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) <EOL> b . consume ( '<STR_LIT>' , None ) <EOL> self . assertIn ( '<STR_LIT>' , b . channel ) <EOL> def test_cancel ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) <EOL> b . cancel ( '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , b . channel ) <EOL> def test_delete ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) <EOL> b . delete ( ) <EOL> self . assertIn ( '<STR_LIT>' , b . channel ) <EOL> def test_queue_unbind ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' , channel = get_conn ( ) . channel ( ) ) <EOL> b . queue_unbind ( ) <EOL> self . assertIn ( '<STR_LIT>' , b . channel ) <EOL> def test_as_dict ( self ) : <EOL> q = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT>' ) <EOL> d = q . as_dict ( recurse = True ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , self . exchange . name ) <EOL> def test_queue_dump ( self ) : <EOL> b = binding ( self . exchange , '<STR_LIT>' ) <EOL> q = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT>' , bindings = [ b ] ) <EOL> d = q . as_dict ( recurse = True ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> registry . dumps ( d ) <EOL> def test__repr__ ( self ) : <EOL> b = Queue ( '<STR_LIT:foo>' , self . exchange , '<STR_LIT:foo>' ) <EOL> self . assertIn ( '<STR_LIT:foo>' , repr ( b ) ) <EOL> self . assertIn ( '<STR_LIT>' , repr ( b ) ) </s>
