<s> import logging <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> from pamqp import specification <EOL> from pamqp . body import ContentBody <EOL> from pamqp . header import ContentHeader <EOL> import amqpstorm <EOL> from amqpstorm import exception <EOL> from amqpstorm import Message <EOL> from amqpstorm import Channel <EOL> from amqpstorm . basic import Basic <EOL> from amqpstorm . exchange import Exchange <EOL> from amqpstorm . queue import Queue <EOL> from amqpstorm . exception import * <EOL> from amqpstorm . tests . utility import FakeConnection <EOL> from amqpstorm . tests . utility import FakeFrame <EOL> from amqpstorm . tests . utility import MockLoggingHandler <EOL> logging . basicConfig ( level = logging . DEBUG ) <EOL> class ChannelTests ( unittest . TestCase ) : <EOL> def test_channel_id ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> self . assertEqual ( int ( channel ) , <NUM_LIT:0> ) <EOL> channel = Channel ( <NUM_LIT> , None , <NUM_LIT> ) <EOL> self . assertEqual ( int ( channel ) , <NUM_LIT> ) <EOL> def test_channel_close ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> channel . _inbound = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> channel . set_state ( channel . OPEN ) <EOL> channel . _consumer_tags = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> channel . _close_channel ( specification . Channel . Close ( reply_text = '<STR_LIT>' ) ) <EOL> self . assertEqual ( channel . _inbound , [ ] ) <EOL> self . assertEqual ( channel . _consumer_tags , [ ] ) <EOL> self . assertEqual ( channel . _state , channel . CLOSED ) <EOL> def test_channel_basic_handler ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> self . assertIsNone ( channel . _basic ) <EOL> self . assertIsInstance ( channel . basic , Basic ) <EOL> self . assertIsNotNone ( channel . _basic ) <EOL> def test_channel_exchange_handler ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> self . assertIsNone ( channel . _exchange ) <EOL> self . assertIsInstance ( channel . exchange , Exchange ) <EOL> self . assertIsNotNone ( channel . _exchange ) <EOL> def test_channel_queue_handler ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> self . assertIsNone ( channel . _queue ) <EOL> self . assertIsInstance ( channel . queue , Queue ) <EOL> self . assertIsNotNone ( channel . _queue ) <EOL> class ChannelExceptionTests ( unittest . TestCase ) : <EOL> def test_chanel_invalid_close_parameter ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> self . assertRaisesRegexp ( AMQPInvalidArgument , <EOL> '<STR_LIT>' , <EOL> channel . close , '<STR_LIT>' , '<STR_LIT:error>' ) <EOL> self . assertRaisesRegexp ( AMQPInvalidArgument , <EOL> '<STR_LIT>' , <EOL> channel . close , <NUM_LIT:200> , <NUM_LIT:200> ) <EOL> def test_chanel_callback_not_set ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> self . assertRaisesRegexp ( AMQPChannelError , <EOL> '<STR_LIT>' , <EOL> channel . process_data_events ) <EOL> def test_channel_throw_exception_check_for_error ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( ) , <NUM_LIT> ) <EOL> channel . set_state ( channel . OPEN ) <EOL> channel . exceptions . append ( AMQPConnectionError ( '<STR_LIT>' ) ) <EOL> self . assertRaises ( AMQPConnectionError , channel . check_for_errors ) <EOL> def test_channel_check_error_no_exception ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( ) , <NUM_LIT> ) <EOL> channel . set_state ( Channel . OPEN ) <EOL> channel . check_for_errors ( ) <EOL> def test_channel_check_error_when_closed ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( ) , <NUM_LIT> ) <EOL> self . assertRaises ( exception . AMQPChannelError , channel . check_for_errors ) <EOL> def test_channel_check_error_connection_closed ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( FakeConnection . CLOSED ) , <NUM_LIT> ) <EOL> self . assertRaises ( exception . AMQPConnectionError , <EOL> channel . check_for_errors ) <EOL> def test_channel_raises_when_closed ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( FakeConnection . OPEN ) , <NUM_LIT> ) <EOL> channel . set_state ( channel . CLOSED ) <EOL> self . assertFalse ( channel . is_open ) <EOL> self . assertRaisesRegexp ( exception . AMQPChannelError , <EOL> '<STR_LIT>' , <EOL> channel . check_for_errors ) <EOL> self . assertTrue ( channel . is_closed ) <EOL> def test_channel_closed_after_connection_closed ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( FakeConnection . CLOSED ) , <NUM_LIT> ) <EOL> channel . set_state ( channel . OPEN ) <EOL> self . assertTrue ( channel . is_open ) <EOL> self . assertRaisesRegexp ( exception . AMQPConnectionError , <EOL> '<STR_LIT>' , <EOL> channel . check_for_errors ) <EOL> self . assertTrue ( channel . is_closed ) <EOL> def test_channel_closed_after_connection_exception ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , <NUM_LIT> ) <EOL> connection . exceptions . append ( AMQPConnectionError ( '<STR_LIT:error>' ) ) <EOL> channel . set_state ( channel . OPEN ) <EOL> self . assertTrue ( connection . is_closed ) <EOL> self . assertTrue ( channel . is_open ) <EOL> self . assertRaisesRegexp ( exception . AMQPConnectionError , '<STR_LIT:error>' , <EOL> channel . check_for_errors ) <EOL> self . assertTrue ( channel . is_closed ) <EOL> def test_channel_consume_exception_when_recoverable ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> connection . set_state ( connection . OPEN ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , <NUM_LIT> ) <EOL> channel . set_state ( channel . OPEN ) <EOL> channel . exceptions . append ( AMQPChannelError ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( connection . is_open ) <EOL> self . assertTrue ( channel . is_open ) <EOL> self . assertRaisesRegexp ( exception . AMQPChannelError , '<STR_LIT>' , <EOL> channel . check_for_errors ) <EOL> self . assertTrue ( channel . is_open ) <EOL> channel . check_for_errors ( ) <EOL> class ChannelBuildMessageTests ( unittest . TestCase ) : <EOL> def test_channel_build_message ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> message = b'<STR_LIT>' <EOL> message_len = len ( message ) <EOL> deliver = specification . Basic . Deliver ( ) <EOL> header = ContentHeader ( body_size = message_len ) <EOL> body = ContentBody ( value = message ) <EOL> channel . _inbound = [ deliver , header , body ] <EOL> result = channel . _build_message ( ) <EOL> self . assertEqual ( result . _body , message ) <EOL> def test_channel_build_out_of_order_message ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> message = b'<STR_LIT>' <EOL> message_len = len ( message ) <EOL> deliver = specification . Basic . Deliver ( ) <EOL> header = ContentHeader ( body_size = message_len ) <EOL> body = ContentBody ( value = message ) <EOL> channel . _inbound = [ deliver , deliver , header , body ] <EOL> result = channel . _build_message ( ) <EOL> self . assertEqual ( result , None ) <EOL> def test_channel_build_message_body ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> message = b'<STR_LIT>' <EOL> message_len = len ( message ) <EOL> body = ContentBody ( value = message ) <EOL> channel . _inbound = [ body ] <EOL> result = channel . _build_message_body ( message_len ) <EOL> self . assertEqual ( message , result ) <EOL> def test_channel_build_empty_inbound_messages ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( ) , <NUM_LIT> ) <EOL> channel . set_state ( Channel . OPEN ) <EOL> result = None <EOL> for message in channel . build_inbound_messages ( break_on_empty = True ) : <EOL> result = message <EOL> self . assertIsNone ( result ) <EOL> def test_channel_build_inbound_messages ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( ) , <NUM_LIT> ) <EOL> channel . set_state ( Channel . OPEN ) <EOL> message = b'<STR_LIT>' <EOL> message_len = len ( message ) <EOL> deliver = specification . Basic . Deliver ( ) <EOL> header = ContentHeader ( body_size = message_len ) <EOL> body = ContentBody ( value = message ) <EOL> channel . _inbound = [ deliver , header , body ] <EOL> for message in channel . build_inbound_messages ( break_on_empty = True ) : <EOL> self . assertIsInstance ( message , Message ) <EOL> def test_channel_build_multiple_inbound_messages ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , FakeConnection ( ) , <NUM_LIT> ) <EOL> channel . set_state ( Channel . OPEN ) <EOL> message = b'<STR_LIT>' <EOL> message_len = len ( message ) <EOL> deliver = specification . Basic . Deliver ( ) <EOL> header = ContentHeader ( body_size = message_len ) <EOL> body = ContentBody ( value = message ) <EOL> channel . _inbound = [ deliver , header , body , deliver , header , body , <EOL> deliver , header , body , deliver , header , body ] <EOL> index = <NUM_LIT:0> <EOL> for message in channel . build_inbound_messages ( break_on_empty = True ) : <EOL> self . assertIsInstance ( message , Message ) <EOL> index += <NUM_LIT:1> <EOL> self . assertEqual ( index , <NUM_LIT:4> ) <EOL> class ChannelFrameTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . logging_handler = MockLoggingHandler ( ) <EOL> logging . root . addHandler ( self . logging_handler ) <EOL> def test_channel_consume_ok_frame ( self ) : <EOL> tag = '<STR_LIT>' <EOL> channel = Channel ( <NUM_LIT:0> , None , rpc_timeout = <NUM_LIT> ) <EOL> channel . on_frame ( specification . Basic . ConsumeOk ( tag ) ) <EOL> self . assertEqual ( channel . consumer_tags [ <NUM_LIT:0> ] , tag ) <EOL> def test_channel_cancel_ok_frame ( self ) : <EOL> tag = '<STR_LIT>' <EOL> channel = Channel ( <NUM_LIT:0> , None , rpc_timeout = <NUM_LIT> ) <EOL> channel . add_consumer_tag ( tag ) <EOL> channel . on_frame ( specification . Basic . CancelOk ( tag ) ) <EOL> self . assertFalse ( channel . consumer_tags ) <EOL> def test_channel_flow_frame ( self ) : <EOL> connection = FakeConnection ( ) <EOL> connection . set_state ( connection . OPEN ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , rpc_timeout = <NUM_LIT> ) <EOL> channel . set_state ( channel . OPEN ) <EOL> channel . on_frame ( specification . Channel . Flow ( ) ) <EOL> self . assertIsInstance ( connection . frames_out [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , <EOL> specification . Channel . FlowOk ) <EOL> def test_channel_basic_cancel_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , rpc_timeout = <NUM_LIT> ) <EOL> channel . on_frame ( specification . Basic . Cancel ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( self . logging_handler . messages [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> def test_channel_basic_return_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , rpc_timeout = <NUM_LIT> ) <EOL> channel . on_frame ( specification . Basic . Return ( reply_code = <NUM_LIT> , <EOL> reply_text = '<STR_LIT:test>' , <EOL> exchange = '<STR_LIT>' , <EOL> routing_key = '<STR_LIT>' ) ) <EOL> self . assertEqual ( str ( channel . exceptions [ <NUM_LIT:0> ] ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_channel_close_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , rpc_timeout = <NUM_LIT> ) <EOL> channel . on_frame ( specification . Channel . Close ( reply_code = <NUM_LIT> , <EOL> reply_text = '<STR_LIT:test>' ) ) <EOL> self . assertEqual ( str ( channel . exceptions [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' ) <EOL> def test_channel_basic_return_raises_when_500 ( self ) : <EOL> channel = Channel ( <NUM_LIT:0> , None , <NUM_LIT> ) <EOL> basic_return = specification . Basic . Return ( reply_code = <NUM_LIT> , <EOL> reply_text = '<STR_LIT>' ) <EOL> channel . _basic_return ( basic_return ) <EOL> self . assertEqual ( len ( channel . exceptions ) , <NUM_LIT:1> ) <EOL> why = channel . exceptions . pop ( <NUM_LIT:0> ) <EOL> self . assertIsInstance ( why , AMQPMessageError ) <EOL> self . assertEqual ( str ( why ) , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_channel_unhandled_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel ( <NUM_LIT:0> , connection , rpc_timeout = <NUM_LIT> ) <EOL> channel . on_frame ( FakeFrame ( ) ) <EOL> self . assertEqual ( self . logging_handler . messages [ '<STR_LIT:error>' ] [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) </s>
