<s> import logging <EOL> import time <EOL> import platform <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> import amqpstorm <EOL> from pamqp . specification import Connection <EOL> from pamqp . heartbeat import Heartbeat <EOL> from amqpstorm . channel0 import Channel0 <EOL> from amqpstorm import AMQPConnectionError <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 Channel0Tests ( unittest . TestCase ) : <EOL> def test_channel0_client_properties ( self ) : <EOL> channel = Channel0 ( FakeConnection ( ) ) <EOL> result = channel . _client_properties ( ) <EOL> information = '<STR_LIT>' <EOL> python_version = '<STR_LIT>' % platform . python_version ( ) <EOL> self . assertIsInstance ( result , dict ) <EOL> self . assertTrue ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( result [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( result [ '<STR_LIT>' ] , information ) <EOL> self . assertEqual ( result [ '<STR_LIT>' ] , python_version ) <EOL> def test_channel0_credentials ( self ) : <EOL> connection = FakeConnection ( ) <EOL> connection . parameters [ '<STR_LIT:username>' ] = '<STR_LIT>' <EOL> connection . parameters [ '<STR_LIT:password>' ] = '<STR_LIT:password>' <EOL> channel = Channel0 ( connection ) <EOL> credentials = channel . _plain_credentials ( ) <EOL> self . assertEqual ( credentials , '<STR_LIT>' ) <EOL> def test_channel0_close_connection ( self ) : <EOL> connection = FakeConnection ( ) <EOL> connection . set_state ( connection . OPEN ) <EOL> channel = Channel0 ( connection ) <EOL> self . assertTrue ( connection . is_open ) <EOL> channel . _close_connection ( <EOL> Connection . Close ( reply_text = b'<STR_LIT>' , <EOL> reply_code = <NUM_LIT:200> ) <EOL> ) <EOL> self . assertEqual ( connection . exceptions , [ ] ) <EOL> self . assertTrue ( connection . is_closed ) <EOL> def test_channel0_forcefully_closed_connection ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> connection . set_state ( connection . OPEN ) <EOL> channel = Channel0 ( connection ) <EOL> channel . _close_connection ( <EOL> Connection . Close ( reply_text = b'<STR_LIT>' , <EOL> reply_code = <NUM_LIT> ) <EOL> ) <EOL> self . assertTrue ( connection . is_closed ) <EOL> self . assertRaises ( AMQPConnectionError , connection . check_for_errors ) <EOL> def test_channel0_send_start_ok_frame ( self ) : <EOL> connection = FakeConnection ( ) <EOL> connection . parameters [ '<STR_LIT:username>' ] = '<STR_LIT>' <EOL> connection . parameters [ '<STR_LIT:password>' ] = '<STR_LIT:password>' <EOL> channel = Channel0 ( connection ) <EOL> channel . _send_start_ok_frame ( Connection . Start ( mechanisms = b'<STR_LIT>' ) ) <EOL> self . assertNotEqual ( connection . frames_out , [ ] ) <EOL> channel_id , frame_out = connection . frames_out . pop ( ) <EOL> self . assertEqual ( channel_id , <NUM_LIT:0> ) <EOL> self . assertIsInstance ( frame_out , Connection . StartOk ) <EOL> self . assertNotEqual ( frame_out . locale , '<STR_LIT>' ) <EOL> self . assertIsNotNone ( frame_out . locale ) <EOL> def test_channel0_send_tune_ok_frame ( self ) : <EOL> connection = FakeConnection ( ) <EOL> channel = Channel0 ( connection ) <EOL> channel . _send_tune_ok_frame ( ) <EOL> self . assertNotEqual ( connection . frames_out , [ ] ) <EOL> channel_id , frame_out = connection . frames_out . pop ( ) <EOL> self . assertEqual ( channel_id , <NUM_LIT:0> ) <EOL> self . assertIsInstance ( frame_out , Connection . TuneOk ) <EOL> def test_channel0_send_close_connection_frame ( self ) : <EOL> connection = FakeConnection ( ) <EOL> channel = Channel0 ( connection ) <EOL> channel . send_close_connection_frame ( ) <EOL> self . assertNotEqual ( connection . frames_out , [ ] ) <EOL> channel_id , frame_out = connection . frames_out . pop ( ) <EOL> self . assertEqual ( channel_id , <NUM_LIT:0> ) <EOL> self . assertIsInstance ( frame_out , Connection . Close ) <EOL> def test_channel0_invalid_authentication_mechanism ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel0 ( connection ) <EOL> channel . _send_start_ok_frame ( <EOL> Connection . Start ( mechanisms = '<STR_LIT>' ) ) <EOL> self . assertRaises ( AMQPConnectionError , connection . check_for_errors ) <EOL> class Channel0FrameTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . logging_handler = MockLoggingHandler ( ) <EOL> logging . root . addHandler ( self . logging_handler ) <EOL> def test_channel0_open_ok_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel0 ( connection ) <EOL> self . assertFalse ( connection . is_open ) <EOL> channel . on_frame ( Connection . OpenOk ( ) ) <EOL> self . assertTrue ( connection . is_open ) <EOL> def test_channel0_on_close_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> connection . set_state ( connection . OPEN ) <EOL> channel = Channel0 ( connection ) <EOL> self . assertFalse ( connection . exceptions ) <EOL> channel . on_frame ( Connection . Close ( ) ) <EOL> self . assertTrue ( connection . exceptions ) <EOL> self . assertTrue ( connection . is_closed ) <EOL> self . assertRaisesRegexp ( AMQPConnectionError , <EOL> '<STR_LIT>' , <EOL> connection . check_for_errors ) <EOL> def test_channel0_is_blocked ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel0 ( connection ) <EOL> self . assertFalse ( channel . is_blocked ) <EOL> channel . on_frame ( Connection . Blocked ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( channel . is_blocked ) <EOL> self . assertEqual ( self . logging_handler . messages [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> def test_channel0_unblocked ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel0 ( connection ) <EOL> channel . on_frame ( Connection . Blocked ( ) ) <EOL> self . assertTrue ( channel . is_blocked ) <EOL> channel . on_frame ( Connection . Unblocked ( ) ) <EOL> self . assertFalse ( channel . is_blocked ) <EOL> self . assertEqual ( self . logging_handler . messages [ '<STR_LIT:info>' ] [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' ) <EOL> def test_channel0_on_hearbeat_registers_heartbeat ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> last_heartbeat = connection . heartbeat . _last_heartbeat <EOL> start_time = time . time ( ) <EOL> channel = Channel0 ( connection ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> def fake ( * _ ) : <EOL> pass <EOL> channel . _write_frame = fake <EOL> self . assertEqual ( connection . heartbeat . _last_heartbeat , <NUM_LIT:0.0> ) <EOL> channel . on_frame ( Heartbeat ( ) ) <EOL> self . assertNotEqual ( connection . heartbeat . _last_heartbeat , <EOL> last_heartbeat ) <EOL> self . assertGreater ( connection . heartbeat . _last_heartbeat , start_time ) <EOL> def test_channel0_unhandled_frame ( self ) : <EOL> connection = amqpstorm . Connection ( '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> lazy = True ) <EOL> channel = Channel0 ( connection ) <EOL> channel . on_frame ( FakeFrame ( ) ) <EOL> self . assertEqual ( self . logging_handler . messages [ '<STR_LIT:error>' ] [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) </s>
