<s> from __future__ import absolute_import , unicode_literals <EOL> import pickle <EOL> import sys <EOL> from collections import defaultdict <EOL> from kombu import Connection , Consumer , Producer , Exchange , Queue <EOL> from kombu . exceptions import MessageStateError <EOL> from kombu . utils import ChannelPromise <EOL> from kombu . utils import json <EOL> from . case import Case , Mock , patch <EOL> from . mocks import Transport <EOL> class test_Producer ( Case ) : <EOL> def setup ( self ) : <EOL> self . exchange = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . connection = Connection ( transport = Transport ) <EOL> self . connection . connect ( ) <EOL> self . assertTrue ( self . connection . connection . connected ) <EOL> self . assertFalse ( self . exchange . is_bound ) <EOL> def test_repr ( self ) : <EOL> p = Producer ( self . connection ) <EOL> self . assertTrue ( repr ( p ) ) <EOL> def test_pickle ( self ) : <EOL> chan = Mock ( ) <EOL> producer = Producer ( chan , serializer = '<STR_LIT>' ) <EOL> p2 = pickle . loads ( pickle . dumps ( producer ) ) <EOL> self . assertEqual ( p2 . serializer , producer . serializer ) <EOL> def test_no_channel ( self ) : <EOL> p = Producer ( None ) <EOL> self . assertFalse ( p . _channel ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_maybe_declare ( self , maybe_declare ) : <EOL> p = self . connection . Producer ( ) <EOL> q = Queue ( '<STR_LIT:foo>' ) <EOL> p . maybe_declare ( q ) <EOL> maybe_declare . assert_called_with ( q , p . channel , False ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_maybe_declare_when_entity_false ( self , maybe_declare ) : <EOL> p = self . connection . Producer ( ) <EOL> p . maybe_declare ( None ) <EOL> maybe_declare . assert_not_called ( ) <EOL> def test_auto_declare ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , auto_declare = True ) <EOL> self . assertIsNot ( p . exchange , self . exchange , <EOL> '<STR_LIT>' ) <EOL> self . assertTrue ( p . exchange . is_bound ) <EOL> self . assertIn ( '<STR_LIT>' , channel , <EOL> '<STR_LIT>' ) <EOL> def test_manual_declare ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , auto_declare = False ) <EOL> self . assertTrue ( p . exchange . is_bound ) <EOL> self . assertNotIn ( '<STR_LIT>' , channel , <EOL> '<STR_LIT>' ) <EOL> p . declare ( ) <EOL> self . assertIn ( '<STR_LIT>' , channel , <EOL> '<STR_LIT>' ) <EOL> def test_prepare ( self ) : <EOL> message = { '<STR_LIT>' : '<STR_LIT>' } <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , serializer = '<STR_LIT>' ) <EOL> m , ctype , cencoding = p . _prepare ( message , headers = { } ) <EOL> self . assertDictEqual ( message , json . loads ( m ) ) <EOL> self . assertEqual ( ctype , '<STR_LIT:application/json>' ) <EOL> self . assertEqual ( cencoding , '<STR_LIT:utf-8>' ) <EOL> def test_prepare_compression ( self ) : <EOL> message = { '<STR_LIT>' : '<STR_LIT>' } <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , serializer = '<STR_LIT>' ) <EOL> headers = { } <EOL> m , ctype , cencoding = p . _prepare ( message , compression = '<STR_LIT>' , <EOL> headers = headers ) <EOL> self . assertEqual ( ctype , '<STR_LIT:application/json>' ) <EOL> self . assertEqual ( cencoding , '<STR_LIT:utf-8>' ) <EOL> self . assertEqual ( headers [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> import zlib <EOL> self . assertEqual ( <EOL> json . loads ( zlib . decompress ( m ) . decode ( '<STR_LIT:utf-8>' ) ) , <EOL> message , <EOL> ) <EOL> def test_prepare_custom_content_type ( self ) : <EOL> message = '<STR_LIT>' . encode ( '<STR_LIT:utf-8>' ) <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , serializer = '<STR_LIT>' ) <EOL> m , ctype , cencoding = p . _prepare ( message , content_type = '<STR_LIT>' ) <EOL> self . assertEqual ( m , message ) <EOL> self . assertEqual ( ctype , '<STR_LIT>' ) <EOL> self . assertEqual ( cencoding , '<STR_LIT>' ) <EOL> m , ctype , cencoding = p . _prepare ( message , content_type = '<STR_LIT>' , <EOL> content_encoding = '<STR_LIT>' ) <EOL> self . assertEqual ( m , message ) <EOL> self . assertEqual ( ctype , '<STR_LIT>' ) <EOL> self . assertEqual ( cencoding , '<STR_LIT>' ) <EOL> def test_prepare_is_already_unicode ( self ) : <EOL> message = '<STR_LIT>' <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , serializer = '<STR_LIT>' ) <EOL> m , ctype , cencoding = p . _prepare ( message , content_type = '<STR_LIT>' ) <EOL> self . assertEqual ( m , message . encode ( '<STR_LIT:utf-8>' ) ) <EOL> self . assertEqual ( ctype , '<STR_LIT>' ) <EOL> self . assertEqual ( cencoding , '<STR_LIT:utf-8>' ) <EOL> m , ctype , cencoding = p . _prepare ( message , content_type = '<STR_LIT>' , <EOL> content_encoding = '<STR_LIT:utf-8>' ) <EOL> self . assertEqual ( m , message . encode ( '<STR_LIT:utf-8>' ) ) <EOL> self . assertEqual ( ctype , '<STR_LIT>' ) <EOL> self . assertEqual ( cencoding , '<STR_LIT:utf-8>' ) <EOL> def test_publish_with_Exchange_instance ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . channel = Mock ( ) <EOL> p . publish ( '<STR_LIT:hello>' , exchange = Exchange ( '<STR_LIT:foo>' ) , delivery_mode = '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> p . _channel . basic_publish . call_args [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , '<STR_LIT:foo>' , <EOL> ) <EOL> def test_publish_with_expiration ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . channel = Mock ( ) <EOL> p . publish ( '<STR_LIT:hello>' , exchange = Exchange ( '<STR_LIT:foo>' ) , expiration = <NUM_LIT:10> ) <EOL> properties = p . _channel . prepare_message . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:5> ] <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_publish_with_reply_to ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . channel = Mock ( ) <EOL> p . publish ( '<STR_LIT:hello>' , exchange = Exchange ( '<STR_LIT:foo>' ) , reply_to = Queue ( '<STR_LIT:foo>' ) ) <EOL> properties = p . _channel . prepare_message . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:5> ] <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , '<STR_LIT:foo>' ) <EOL> def test_set_on_return ( self ) : <EOL> chan = Mock ( ) <EOL> chan . events = defaultdict ( Mock ) <EOL> p = Producer ( ChannelPromise ( lambda : chan ) , on_return = '<STR_LIT>' ) <EOL> p . channel <EOL> chan . events [ '<STR_LIT>' ] . add . assert_called_with ( '<STR_LIT>' ) <EOL> def test_publish_retry_calls_ensure ( self ) : <EOL> p = Producer ( Mock ( ) ) <EOL> p . _connection = Mock ( ) <EOL> ensure = p . connection . ensure = Mock ( ) <EOL> p . publish ( '<STR_LIT:foo>' , exchange = '<STR_LIT:foo>' , retry = True ) <EOL> ensure . assert_called ( ) <EOL> def test_publish_retry_with_declare ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . maybe_declare = Mock ( ) <EOL> p . connection . ensure = Mock ( ) <EOL> ex = Exchange ( '<STR_LIT:foo>' ) <EOL> p . _publish ( '<STR_LIT:hello>' , <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT>' , { } , { } , '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> , ex , declare = [ ex ] ) <EOL> p . maybe_declare . assert_called_with ( ex ) <EOL> def test_revive_when_channel_is_connection ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . exchange = Mock ( ) <EOL> new_conn = Connection ( '<STR_LIT>' ) <EOL> defchan = new_conn . default_channel <EOL> p . revive ( new_conn ) <EOL> self . assertIs ( p . channel , defchan ) <EOL> p . exchange . revive . assert_called_with ( defchan ) <EOL> def test_enter_exit ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . release = Mock ( ) <EOL> self . assertIs ( p . __enter__ ( ) , p ) <EOL> p . __exit__ ( ) <EOL> p . release . assert_called_with ( ) <EOL> def test_connection_property_handles_AttributeError ( self ) : <EOL> p = self . connection . Producer ( ) <EOL> p . channel = object ( ) <EOL> p . __connection__ = None <EOL> self . assertIsNone ( p . connection ) <EOL> def test_publish ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> p = Producer ( channel , self . exchange , serializer = '<STR_LIT>' ) <EOL> message = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ret = p . publish ( message , routing_key = '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> m , exc , rkey = ret <EOL> self . assertDictEqual ( message , json . loads ( m [ '<STR_LIT:body>' ] ) ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf-8>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } , m ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : <NUM_LIT:2> } , m [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( exc , p . exchange . name ) <EOL> self . assertEqual ( rkey , '<STR_LIT>' ) <EOL> def test_no_exchange ( self ) : <EOL> chan = self . connection . channel ( ) <EOL> p = Producer ( chan ) <EOL> self . assertFalse ( p . exchange . name ) <EOL> def test_revive ( self ) : <EOL> chan = self . connection . channel ( ) <EOL> p = Producer ( chan ) <EOL> chan2 = self . connection . channel ( ) <EOL> p . revive ( chan2 ) <EOL> self . assertIs ( p . channel , chan2 ) <EOL> self . assertIs ( p . exchange . channel , chan2 ) <EOL> def test_on_return ( self ) : <EOL> chan = self . connection . channel ( ) <EOL> def on_return ( exception , exchange , routing_key , message ) : <EOL> pass <EOL> p = Producer ( chan , on_return = on_return ) <EOL> self . assertIn ( on_return , chan . events [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( p . on_return ) <EOL> class test_Consumer ( Case ) : <EOL> def setup ( self ) : <EOL> self . connection = Connection ( transport = Transport ) <EOL> self . connection . connect ( ) <EOL> self . assertTrue ( self . connection . connection . connected ) <EOL> self . exchange = Exchange ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> def test_accept ( self ) : <EOL> a = Consumer ( self . connection ) <EOL> self . assertIsNone ( a . accept ) <EOL> b = Consumer ( self . connection , accept = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertSetEqual ( <EOL> b . accept , <EOL> { '<STR_LIT:application/json>' , '<STR_LIT>' } , <EOL> ) <EOL> c = Consumer ( self . connection , accept = b . accept ) <EOL> self . assertSetEqual ( b . accept , c . accept ) <EOL> def test_enter_exit_cancel_raises ( self ) : <EOL> c = Consumer ( self . connection ) <EOL> c . cancel = Mock ( name = '<STR_LIT>' ) <EOL> c . cancel . side_effect = KeyError ( '<STR_LIT:foo>' ) <EOL> with c : <EOL> pass <EOL> c . cancel . assert_called_with ( ) <EOL> def test_receive_callback_accept ( self ) : <EOL> message = Mock ( name = '<STR_LIT>' ) <EOL> message . errors = [ ] <EOL> callback = Mock ( name = '<STR_LIT>' ) <EOL> c = Consumer ( self . connection , accept = [ '<STR_LIT>' ] , on_message = callback ) <EOL> c . on_decode_error = None <EOL> c . channel = Mock ( name = '<STR_LIT>' ) <EOL> c . channel . message_to_python = None <EOL> c . _receive_callback ( message ) <EOL> callback . assert_called_with ( message ) <EOL> self . assertSetEqual ( message . accept , c . accept ) <EOL> def test_accept__content_disallowed ( self ) : <EOL> conn = Connection ( '<STR_LIT>' ) <EOL> q = Queue ( '<STR_LIT:foo>' , exchange = self . exchange ) <EOL> p = conn . Producer ( ) <EOL> p . publish ( <EOL> { '<STR_LIT>' : object ( ) } , <EOL> declare = [ q ] , exchange = self . exchange , serializer = '<STR_LIT>' , <EOL> ) <EOL> callback = Mock ( name = '<STR_LIT>' ) <EOL> with conn . Consumer ( queues = [ q ] , callbacks = [ callback ] ) as consumer : <EOL> with self . assertRaises ( consumer . ContentDisallowed ) : <EOL> conn . drain_events ( timeout = <NUM_LIT:1> ) <EOL> callback . assert_not_called ( ) <EOL> def test_accept__content_allowed ( self ) : <EOL> conn = Connection ( '<STR_LIT>' ) <EOL> q = Queue ( '<STR_LIT:foo>' , exchange = self . exchange ) <EOL> p = conn . Producer ( ) <EOL> p . publish ( <EOL> { '<STR_LIT>' : object ( ) } , <EOL> declare = [ q ] , exchange = self . exchange , serializer = '<STR_LIT>' , <EOL> ) <EOL> callback = Mock ( name = '<STR_LIT>' ) <EOL> with conn . Consumer ( queues = [ q ] , accept = [ '<STR_LIT>' ] , <EOL> callbacks = [ callback ] ) : <EOL> conn . drain_events ( timeout = <NUM_LIT:1> ) <EOL> callback . assert_called ( ) <EOL> body , message = callback . call_args [ <NUM_LIT:0> ] <EOL> self . assertTrue ( body [ '<STR_LIT>' ] ) <EOL> def test_set_no_channel ( self ) : <EOL> c = Consumer ( None ) <EOL> self . assertIsNone ( c . channel ) <EOL> c . revive ( Mock ( ) ) <EOL> self . assertTrue ( c . channel ) <EOL> def test_set_no_ack ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = True , no_ack = True ) <EOL> self . assertTrue ( consumer . no_ack ) <EOL> def test_add_queue_when_auto_declare ( self ) : <EOL> consumer = self . connection . Consumer ( auto_declare = True ) <EOL> q = Mock ( ) <EOL> q . return_value = q <EOL> consumer . add_queue ( q ) <EOL> self . assertIn ( q , consumer . queues ) <EOL> q . declare . assert_called_with ( ) <EOL> def test_add_queue_when_not_auto_declare ( self ) : <EOL> consumer = self . connection . Consumer ( auto_declare = False ) <EOL> q = Mock ( ) <EOL> q . return_value = q <EOL> consumer . add_queue ( q ) <EOL> self . assertIn ( q , consumer . queues ) <EOL> self . assertFalse ( q . declare . call_count ) <EOL> def test_consume_without_queues_returns ( self ) : <EOL> consumer = self . connection . Consumer ( ) <EOL> consumer . queues [ : ] = [ ] <EOL> self . assertIsNone ( consumer . consume ( ) ) <EOL> def test_consuming_from ( self ) : <EOL> consumer = self . connection . Consumer ( ) <EOL> consumer . queues [ : ] = [ Queue ( '<STR_LIT:a>' ) , Queue ( '<STR_LIT:b>' ) , Queue ( '<STR_LIT:d>' ) ] <EOL> consumer . _active_tags = { '<STR_LIT:a>' : <NUM_LIT:1> , '<STR_LIT:b>' : <NUM_LIT:2> } <EOL> self . assertFalse ( consumer . consuming_from ( Queue ( '<STR_LIT:c>' ) ) ) <EOL> self . assertFalse ( consumer . consuming_from ( '<STR_LIT:c>' ) ) <EOL> self . assertFalse ( consumer . consuming_from ( Queue ( '<STR_LIT:d>' ) ) ) <EOL> self . assertFalse ( consumer . consuming_from ( '<STR_LIT:d>' ) ) <EOL> self . assertTrue ( consumer . consuming_from ( Queue ( '<STR_LIT:a>' ) ) ) <EOL> self . assertTrue ( consumer . consuming_from ( Queue ( '<STR_LIT:b>' ) ) ) <EOL> self . assertTrue ( consumer . consuming_from ( '<STR_LIT:b>' ) ) <EOL> def test_receive_callback_without_m2p ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> c = channel . Consumer ( ) <EOL> m2p = getattr ( channel , '<STR_LIT>' ) <EOL> channel . message_to_python = None <EOL> try : <EOL> message = Mock ( ) <EOL> message . errors = [ ] <EOL> message . decode . return_value = '<STR_LIT>' <EOL> recv = c . receive = Mock ( ) <EOL> c . _receive_callback ( message ) <EOL> recv . assert_called_with ( '<STR_LIT>' , message ) <EOL> finally : <EOL> channel . message_to_python = m2p <EOL> def test_receive_callback__message_errors ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> channel . message_to_python = None <EOL> c = channel . Consumer ( ) <EOL> message = Mock ( ) <EOL> try : <EOL> raise KeyError ( '<STR_LIT:foo>' ) <EOL> except KeyError : <EOL> message . errors = [ sys . exc_info ( ) ] <EOL> message . _reraise_error . side_effect = KeyError ( ) <EOL> with self . assertRaises ( KeyError ) : <EOL> c . _receive_callback ( message ) <EOL> def test_set_callbacks ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> callbacks = [ lambda x , y : x , <EOL> lambda x , y : x ] <EOL> consumer = Consumer ( channel , queue , auto_declare = True , <EOL> callbacks = callbacks ) <EOL> self . assertEqual ( consumer . callbacks , callbacks ) <EOL> def test_auto_declare ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = True ) <EOL> consumer . consume ( ) <EOL> consumer . consume ( ) <EOL> self . assertIsNot ( consumer . queues [ <NUM_LIT:0> ] , queue ) <EOL> self . assertTrue ( consumer . queues [ <NUM_LIT:0> ] . is_bound ) <EOL> self . assertTrue ( consumer . queues [ <NUM_LIT:0> ] . exchange . is_bound ) <EOL> self . assertIsNot ( consumer . queues [ <NUM_LIT:0> ] . exchange , self . exchange ) <EOL> for meth in ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> self . assertIn ( meth , channel ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( consumer . _active_tags ) <EOL> consumer . cancel_by_queue ( queue . name ) <EOL> consumer . cancel_by_queue ( queue . name ) <EOL> self . assertFalse ( consumer . _active_tags ) <EOL> def test_consumer_tag_prefix ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , tag_prefix = '<STR_LIT>' ) <EOL> consumer . consume ( ) <EOL> self . assertTrue ( <EOL> consumer . _active_tags [ queue . name ] . startswith ( '<STR_LIT>' ) , <EOL> ) <EOL> def test_manual_declare ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = False ) <EOL> self . assertIsNot ( consumer . queues [ <NUM_LIT:0> ] , queue ) <EOL> self . assertTrue ( consumer . queues [ <NUM_LIT:0> ] . is_bound ) <EOL> self . assertTrue ( consumer . queues [ <NUM_LIT:0> ] . exchange . is_bound ) <EOL> self . assertIsNot ( consumer . queues [ <NUM_LIT:0> ] . exchange , self . exchange ) <EOL> for meth in ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> self . assertNotIn ( meth , channel ) <EOL> consumer . declare ( ) <EOL> for meth in ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> self . assertIn ( meth , channel ) <EOL> self . assertNotIn ( '<STR_LIT>' , channel ) <EOL> consumer . consume ( ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_consume__cancel ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = True ) <EOL> consumer . consume ( ) <EOL> consumer . cancel ( ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> self . assertFalse ( consumer . _active_tags ) <EOL> def test___enter____exit__ ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = True ) <EOL> context = consumer . __enter__ ( ) <EOL> self . assertIs ( context , consumer ) <EOL> self . assertTrue ( consumer . _active_tags ) <EOL> res = consumer . __exit__ ( None , None , None ) <EOL> self . assertFalse ( res ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> self . assertFalse ( consumer . _active_tags ) <EOL> def test_flow ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = True ) <EOL> consumer . flow ( False ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_qos ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> queue = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , queue , auto_declare = True ) <EOL> consumer . qos ( <NUM_LIT:30> , <NUM_LIT:10> , False ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_purge ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> b2 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> b3 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> b4 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 , b2 , b3 , b4 ] , auto_declare = True ) <EOL> consumer . purge ( ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> def test_multiple_queues ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> b2 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> b3 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> b4 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 , b2 , b3 , b4 ] ) <EOL> consumer . consume ( ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( len ( consumer . _active_tags ) , <NUM_LIT:4> ) <EOL> consumer . cancel ( ) <EOL> self . assertEqual ( channel . called . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> self . assertFalse ( len ( consumer . _active_tags ) ) <EOL> def test_receive_callback ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> received = [ ] <EOL> def callback ( message_data , message ) : <EOL> received . append ( message_data ) <EOL> message . ack ( ) <EOL> message . payload <EOL> consumer . register_callback ( callback ) <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> self . assertEqual ( received [ <NUM_LIT:0> ] , { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> def test_basic_ack_twice ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> def callback ( message_data , message ) : <EOL> message . ack ( ) <EOL> message . ack ( ) <EOL> consumer . register_callback ( callback ) <EOL> with self . assertRaises ( MessageStateError ) : <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> def test_basic_reject ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> def callback ( message_data , message ) : <EOL> message . reject ( ) <EOL> consumer . register_callback ( callback ) <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_basic_reject_twice ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> def callback ( message_data , message ) : <EOL> message . reject ( ) <EOL> message . reject ( ) <EOL> consumer . register_callback ( callback ) <EOL> with self . assertRaises ( MessageStateError ) : <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_basic_reject__requeue ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> def callback ( message_data , message ) : <EOL> message . requeue ( ) <EOL> consumer . register_callback ( callback ) <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_basic_reject__requeue_twice ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> def callback ( message_data , message ) : <EOL> message . requeue ( ) <EOL> message . requeue ( ) <EOL> consumer . register_callback ( callback ) <EOL> with self . assertRaises ( MessageStateError ) : <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_receive_without_callbacks_raises ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> with self . assertRaises ( NotImplementedError ) : <EOL> consumer . receive ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> def test_decode_error ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> consumer . channel . throw_decode_error = True <EOL> with self . assertRaises ( ValueError ) : <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> def test_on_decode_error_callback ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> thrown = [ ] <EOL> def on_decode_error ( msg , exc ) : <EOL> thrown . append ( ( msg . body , exc ) ) <EOL> consumer = Consumer ( channel , [ b1 ] , on_decode_error = on_decode_error ) <EOL> consumer . channel . throw_decode_error = True <EOL> consumer . _receive_callback ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertTrue ( thrown ) <EOL> m , exc = thrown [ <NUM_LIT:0> ] <EOL> self . assertEqual ( json . loads ( m ) , { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertIsInstance ( exc , ValueError ) <EOL> def test_recover ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> consumer . recover ( ) <EOL> self . assertIn ( '<STR_LIT>' , channel ) <EOL> def test_revive ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> consumer = Consumer ( channel , [ b1 ] ) <EOL> channel2 = self . connection . channel ( ) <EOL> consumer . revive ( channel2 ) <EOL> self . assertIs ( consumer . channel , channel2 ) <EOL> self . assertIs ( consumer . queues [ <NUM_LIT:0> ] . channel , channel2 ) <EOL> self . assertIs ( consumer . queues [ <NUM_LIT:0> ] . exchange . channel , channel2 ) <EOL> def test_revive__with_prefetch_count ( self ) : <EOL> channel = Mock ( name = '<STR_LIT>' ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> Consumer ( channel , [ b1 ] , prefetch_count = <NUM_LIT> ) <EOL> channel . basic_qos . assert_called_with ( <NUM_LIT:0> , <NUM_LIT> , False ) <EOL> def test__repr__ ( self ) : <EOL> channel = self . connection . channel ( ) <EOL> b1 = Queue ( '<STR_LIT>' , self . exchange , '<STR_LIT>' ) <EOL> self . assertTrue ( repr ( Consumer ( channel , [ b1 ] ) ) ) <EOL> def test_connection_property_handles_AttributeError ( self ) : <EOL> p = self . connection . Consumer ( ) <EOL> p . channel = object ( ) <EOL> self . assertIsNone ( p . connection ) </s>
