<s> """<STR_LIT>""" <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> import mock <EOL> from pamqp import specification <EOL> from rabbitpy import PYPY <EOL> from rabbitpy import amqp_queue <EOL> from rabbitpy import channel <EOL> from rabbitpy import exceptions <EOL> from rabbitpy import utils <EOL> class QueueInitializationTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . chan = channel . Channel ( <NUM_LIT:1> , { } , None , None , None , None , <NUM_LIT> , None ) <EOL> def test_empty_queue_name ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertEqual ( queue . name , '<STR_LIT>' ) <EOL> def test_invalid_queue_name ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , self . chan , None ) <EOL> def test_auto_delete_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertFalse ( queue . auto_delete ) <EOL> def test_auto_delete_true ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , auto_delete = True ) <EOL> self . assertTrue ( queue . auto_delete ) <EOL> def test_auto_delete_false ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , auto_delete = False ) <EOL> self . assertFalse ( queue . auto_delete ) <EOL> def test_auto_delete_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , <EOL> self . chan , None , None , None , <NUM_LIT:10> ) <EOL> def test_durable_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertFalse ( queue . durable ) <EOL> def test_durable_true ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , durable = True ) <EOL> self . assertTrue ( queue . durable ) <EOL> def test_durable_false ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , durable = False ) <EOL> self . assertFalse ( queue . durable ) <EOL> def test_durable_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , <EOL> self . chan , None , '<STR_LIT>' ) <EOL> def test_exclusive_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertFalse ( queue . exclusive ) <EOL> def test_exclusive_true ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , exclusive = True ) <EOL> self . assertTrue ( queue . exclusive ) <EOL> def test_exclusive_false ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , exclusive = False ) <EOL> self . assertFalse ( queue . exclusive ) <EOL> def test_exclusive_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , <EOL> self . chan , None , None , '<STR_LIT>' ) <EOL> def test_expires_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertIsNone ( queue . expires ) <EOL> def test_expires_named_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , expires = <NUM_LIT:10> ) <EOL> self . assertEqual ( queue . expires , <NUM_LIT:10> ) <EOL> self . assertIsInstance ( queue . expires , int ) <EOL> def test_expires_positional_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , '<STR_LIT>' , True , False , True , <EOL> None , None , <NUM_LIT:10> ) <EOL> self . assertEqual ( queue . expires , <NUM_LIT:10> ) <EOL> self . assertIsInstance ( queue . expires , int ) <EOL> def test_expires_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , <EOL> self . chan , '<STR_LIT>' , True , False , True , None , None , '<STR_LIT>' ) <EOL> def test_max_length_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertIsNone ( queue . max_length ) <EOL> def test_max_length_named_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , max_length = <NUM_LIT:10> ) <EOL> self . assertEqual ( queue . max_length , <NUM_LIT:10> ) <EOL> self . assertIsInstance ( queue . max_length , int ) <EOL> def test_max_length_positional_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , '<STR_LIT>' , True , False , True , <NUM_LIT:10> ) <EOL> self . assertEqual ( queue . max_length , <NUM_LIT:10> ) <EOL> self . assertIsInstance ( queue . max_length , int ) <EOL> def test_max_length_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , <EOL> self . chan , '<STR_LIT>' , True , False , True , '<STR_LIT>' ) <EOL> def test_message_ttl_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertIsNone ( queue . message_ttl ) <EOL> def test_message_ttl_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , message_ttl = <NUM_LIT:10> ) <EOL> self . assertEqual ( queue . message_ttl , <NUM_LIT:10> ) <EOL> self . assertIsInstance ( queue . message_ttl , int ) <EOL> def test_message_ttl_positional_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , '<STR_LIT>' , True , False , True , None , <NUM_LIT:10> ) <EOL> self . assertEqual ( queue . message_ttl , <NUM_LIT:10> ) <EOL> self . assertIsInstance ( queue . message_ttl , int ) <EOL> def test_message_ttl_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , <EOL> self . chan , '<STR_LIT>' , True , False , True , None , '<STR_LIT>' ) <EOL> def test_dlx_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertIsNone ( queue . dead_letter_exchange ) <EOL> def test_dlx_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , dead_letter_exchange = '<STR_LIT>' ) <EOL> self . assertEqual ( queue . dead_letter_exchange , '<STR_LIT>' ) <EOL> def test_dlx_bytes ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , dead_letter_exchange = b'<STR_LIT>' ) <EOL> self . assertIsInstance ( queue . dead_letter_exchange , bytes ) <EOL> def test_dlx_str ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , dead_letter_exchange = '<STR_LIT>' ) <EOL> self . assertIsInstance ( queue . dead_letter_exchange , str ) <EOL> @ unittest . skipIf ( utils . PYTHON3 , '<STR_LIT>' ) <EOL> def test_dlx_unicode ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , <EOL> dead_letter_exchange = unicode ( '<STR_LIT>' ) ) <EOL> self . assertIsInstance ( queue . dead_letter_exchange , unicode ) <EOL> def test_message_dlx_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , self . chan , '<STR_LIT>' , True , <EOL> False , True , None , None , None , True ) <EOL> def test_dlr_default ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertIsNone ( queue . dead_letter_routing_key ) <EOL> def test_dlr_value ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , <EOL> dead_letter_routing_key = '<STR_LIT>' ) <EOL> self . assertEqual ( queue . dead_letter_routing_key , '<STR_LIT>' ) <EOL> def test_dlr_bytes ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , <EOL> dead_letter_routing_key = b'<STR_LIT>' ) <EOL> self . assertIsInstance ( queue . dead_letter_routing_key , bytes ) <EOL> def test_dlr_str ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan , <EOL> dead_letter_routing_key = '<STR_LIT>' ) <EOL> self . assertIsInstance ( queue . dead_letter_routing_key , str ) <EOL> @ unittest . skipIf ( utils . PYTHON3 , '<STR_LIT>' ) <EOL> def test_dlr_unicode ( self ) : <EOL> routing_key = unicode ( '<STR_LIT>' ) <EOL> queue = amqp_queue . Queue ( self . chan , <EOL> dead_letter_routing_key = routing_key ) <EOL> self . assertIsInstance ( queue . dead_letter_routing_key , unicode ) <EOL> def test_dlr_validation ( self ) : <EOL> self . assertRaises ( ValueError , amqp_queue . Queue , self . chan , '<STR_LIT>' , True , <EOL> False , True , None , None , None , None , True ) <EOL> @ unittest . skipIf ( PYPY , '<STR_LIT>' ) <EOL> def test_stop_consuming_raises_exception ( self ) : <EOL> queue = amqp_queue . Queue ( self . chan ) <EOL> self . assertRaises ( exceptions . NotConsumingError , queue . stop_consuming ) <EOL> class QueueDeclareTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . chan = channel . Channel ( <NUM_LIT:1> , { } , None , None , None , None , <NUM_LIT> , None ) <EOL> def test_default_declare ( self ) : <EOL> obj = amqp_queue . Queue ( self . chan ) <EOL> expectation = { '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> self . assertDictEqual ( dict ( obj . _declare ( False ) ) , expectation ) <EOL> def test_default_declare_passive ( self ) : <EOL> obj = amqp_queue . Queue ( self . chan ) <EOL> expectation = { '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> self . assertDictEqual ( dict ( obj . _declare ( True ) ) , expectation ) <EOL> def test_queue_name ( self ) : <EOL> obj = amqp_queue . Queue ( self . chan , '<STR_LIT>' ) <EOL> expectation = { '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> self . assertDictEqual ( dict ( obj . _declare ( False ) ) , expectation ) <EOL> def test_non_defaults ( self ) : <EOL> obj = amqp_queue . Queue ( self . chan , '<STR_LIT>' , False , True , True , <EOL> <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' ) <EOL> expectation = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:100> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> self . assertDictEqual ( dict ( obj . _declare ( False ) ) , expectation ) <EOL> class QueueAssignmentTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . chan = channel . Channel ( <NUM_LIT:1> , { } , None , None , None , None , <NUM_LIT> , None ) <EOL> self . queue = amqp_queue . Queue ( self . chan ) <EOL> def test_auto_delete_assign_true ( self ) : <EOL> self . queue . auto_delete = True <EOL> self . assertTrue ( self . queue . auto_delete ) <EOL> def test_auto_delete_assign_false ( self ) : <EOL> self . queue . auto_delete = False <EOL> self . assertFalse ( self . queue . auto_delete ) <EOL> def test_auto_delete_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . auto_delete = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_durable_assign_true ( self ) : <EOL> self . queue . durable = True <EOL> self . assertTrue ( self . queue . durable ) <EOL> def test_durable_assign_false ( self ) : <EOL> self . queue . durable = False <EOL> self . assertFalse ( self . queue . durable ) <EOL> def test_durable_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . durable = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_exclusive_assign_true ( self ) : <EOL> self . queue . exclusive = True <EOL> self . assertTrue ( self . queue . exclusive ) <EOL> def test_exclusive_assign_false ( self ) : <EOL> self . queue . exclusive = False <EOL> self . assertFalse ( self . queue . exclusive ) <EOL> def test_exclusive_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . exclusive = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_expires_assign_value ( self ) : <EOL> self . queue . expires = <NUM_LIT:100> <EOL> self . assertEqual ( self . queue . expires , <NUM_LIT:100> ) <EOL> def test_expires_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . expires = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_max_length_assign_value ( self ) : <EOL> self . queue . max_length = <NUM_LIT:100> <EOL> self . assertEqual ( self . queue . max_length , <NUM_LIT:100> ) <EOL> def test_max_length_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . max_length = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_message_ttl_assign_value ( self ) : <EOL> self . queue . message_ttl = <NUM_LIT:100> <EOL> self . assertEqual ( self . queue . message_ttl , <NUM_LIT:100> ) <EOL> def test_message_ttl_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . message_ttl = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_dead_letter_exchange_assign_value ( self ) : <EOL> self . queue . dead_letter_exchange = '<STR_LIT>' <EOL> self . assertEqual ( self . queue . dead_letter_exchange , '<STR_LIT>' ) <EOL> def test_dead_letter_exchange_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . dead_letter_exchange = <NUM_LIT> <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_dead_letter_routing_key_assign_value ( self ) : <EOL> self . queue . dead_letter_routing_key = '<STR_LIT>' <EOL> self . assertEqual ( self . queue . dead_letter_routing_key , '<STR_LIT>' ) <EOL> def test_dead_letter_routing_key_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . dead_letter_routing_key = <NUM_LIT> <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> def test_arguments_assign_value ( self ) : <EOL> self . queue . arguments = { '<STR_LIT:foo>' : '<STR_LIT:bar>' } <EOL> self . assertDictEqual ( self . queue . arguments , { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> def test_arguments_assign_raises_type_error ( self ) : <EOL> def assign_value ( ) : <EOL> self . queue . arguments = <NUM_LIT> <EOL> self . assertRaises ( ValueError , assign_value ) <EOL> class WriteFrameTests ( unittest . TestCase ) : <EOL> NAME = '<STR_LIT:test>' <EOL> DURABLE = True <EOL> EXCLUSIVE = True <EOL> AUTO_DELETE = False <EOL> MAX_LENGTH = <NUM_LIT:200> <EOL> MESSAGE_TTL = <NUM_LIT> <EOL> EXPIRES = <NUM_LIT:10> <EOL> DEAD_LETTER_EXCHANGE = '<STR_LIT>' <EOL> DEAD_LETTER_ROUTING_KEY = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> self . chan = channel . Channel ( <NUM_LIT:1> , { } , None , None , None , None , <NUM_LIT> , None ) <EOL> self . queue = amqp_queue . Queue ( self . chan , self . NAME , self . DURABLE , <EOL> self . EXCLUSIVE , self . AUTO_DELETE , <EOL> self . MAX_LENGTH , self . MESSAGE_TTL , <EOL> self . EXPIRES , self . DEAD_LETTER_EXCHANGE , <EOL> self . DEAD_LETTER_ROUTING_KEY ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_declare_invokes_write_frame_with_queue_declare ( self , rpc ) : <EOL> self . queue . declare ( ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Declare ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_ha_declare_invokes_write_frame_with_queue_declare ( self , rpc ) : <EOL> self . queue . ha_declare ( ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Declare ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_ha_declare_list_invokes_write_frame_with_queue_declare ( self , rpc ) : <EOL> self . queue . ha_declare ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Declare ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_ha_declare_list_sets_proper_attributes ( self , rpc ) : <EOL> self . queue . ha_declare ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> self . assertListEqual ( self . queue . arguments [ '<STR_LIT>' ] , <EOL> [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> self . assertEqual ( self . queue . arguments [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_ha_declare_list_invokes_write_frame_with_queue_declare ( self , rpc ) : <EOL> self . queue . arguments [ '<STR_LIT>' ] = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] <EOL> self . queue . ha_declare ( ) <EOL> self . assertNotIn ( '<STR_LIT>' , self . queue . arguments ) <EOL> self . assertEqual ( self . queue . arguments [ '<STR_LIT>' ] , '<STR_LIT:all>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_bind_invokes_write_frame_with_queue_bind ( self , rpc ) : <EOL> self . queue . bind ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Bind ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_unbind_invokes_write_frame_with_queue_declare ( self , rpc ) : <EOL> self . queue . unbind ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Unbind ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_unbind_with_obj_invokes_write_frame_with_queue_declare ( self , rpc ) : <EOL> exchange = mock . Mock ( ) <EOL> exchange . name = '<STR_LIT:foo>' <EOL> self . queue . unbind ( exchange , '<STR_LIT:bar>' ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Unbind ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_unbind_invokes_write_frame_with_queue_delete ( self , rpc ) : <EOL> self . queue . delete ( ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Delete ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_purge_invokes_write_frame_with_queue_purge ( self , rpc ) : <EOL> self . queue . purge ( ) <EOL> self . assertIsInstance ( rpc . mock_calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] , <EOL> specification . Queue . Purge ) </s>
