<s> '''<STR_LIT>''' <EOL> from chai import Chai <EOL> from haigha . transports import event_transport <EOL> from haigha . transports . event_transport import * <EOL> class EventTransportTest ( Chai ) : <EOL> def setUp ( self ) : <EOL> super ( EventTransportTest , self ) . setUp ( ) <EOL> self . connection = mock ( ) <EOL> self . transport = EventTransport ( self . connection ) <EOL> self . transport . _host = '<STR_LIT>' <EOL> def test_sock_close_cb ( self ) : <EOL> expect ( self . connection . transport_closed ) . args ( <EOL> msg = '<STR_LIT>' ) <EOL> self . transport . _sock_close_cb ( '<STR_LIT>' ) <EOL> def test_sock_error_cb ( self ) : <EOL> expect ( self . connection . transport_closed ) . args ( <EOL> msg = '<STR_LIT>' ) <EOL> self . transport . _sock_error_cb ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_sock_read_cb ( self ) : <EOL> expect ( self . connection . read_frames ) <EOL> self . transport . _sock_read_cb ( '<STR_LIT>' ) <EOL> def test_connect ( self ) : <EOL> sock = mock ( ) <EOL> mock ( event_transport , '<STR_LIT>' ) <EOL> self . connection . _connect_timeout = <NUM_LIT> <EOL> self . connection . _sock_opts = { <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) : <NUM_LIT> , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) : '<STR_LIT>' <EOL> } <EOL> expect ( event_transport . EventSocket ) . args ( <EOL> read_cb = self . transport . _sock_read_cb , <EOL> close_cb = self . transport . _sock_close_cb , <EOL> error_cb = self . transport . _sock_error_cb , <EOL> debug = self . connection . debug , <EOL> logger = self . connection . logger , <EOL> ) . returns ( sock ) <EOL> expect ( sock . setsockopt ) . args ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) . any_order ( ) <EOL> expect ( sock . setsockopt ) . args ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) . any_order ( ) <EOL> expect ( sock . setblocking ) . args ( False ) <EOL> expect ( sock . connect ) . args ( ( '<STR_LIT:host>' , <NUM_LIT> ) , timeout = <NUM_LIT> ) <EOL> self . transport . connect ( ( '<STR_LIT:host>' , <NUM_LIT> ) ) <EOL> def test_read ( self ) : <EOL> self . transport . _heartbeat_timeout = None <EOL> self . transport . _sock = mock ( ) <EOL> expect ( self . transport . _sock . read ) . returns ( '<STR_LIT>' ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . read ( ) ) <EOL> def test_read_with_timeout_and_no_current_one ( self ) : <EOL> self . transport . _heartbeat_timeout = None <EOL> self . transport . _sock = mock ( ) <EOL> mock ( event_transport , '<STR_LIT>' ) <EOL> expect ( event_transport . event . timeout ) . args ( <EOL> '<STR_LIT>' , self . transport . _sock_read_cb , self . transport . _sock ) . returns ( <EOL> '<STR_LIT>' ) <EOL> expect ( self . transport . _sock . read ) . returns ( '<STR_LIT>' ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . read ( '<STR_LIT>' ) ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . _heartbeat_timeout ) <EOL> def test_read_with_timeout_and_current_one ( self ) : <EOL> self . transport . _heartbeat_timeout = mock ( ) <EOL> self . transport . _sock = mock ( ) <EOL> mock ( event_transport , '<STR_LIT>' ) <EOL> expect ( self . transport . _heartbeat_timeout . delete ) <EOL> expect ( event_transport . event . timeout ) . args ( <EOL> '<STR_LIT>' , self . transport . _sock_read_cb , self . transport . _sock ) . returns ( <EOL> '<STR_LIT>' ) <EOL> expect ( self . transport . _sock . read ) . returns ( '<STR_LIT>' ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . read ( '<STR_LIT>' ) ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . _heartbeat_timeout ) <EOL> def test_read_without_timeout_but_current_one ( self ) : <EOL> self . transport . _heartbeat_timeout = mock ( ) <EOL> self . transport . _sock = mock ( ) <EOL> mock ( event_transport , '<STR_LIT>' ) <EOL> expect ( self . transport . _heartbeat_timeout . delete ) <EOL> expect ( self . transport . _sock . read ) . returns ( '<STR_LIT>' ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . read ( ) ) <EOL> assert_equals ( None , self . transport . _heartbeat_timeout ) <EOL> def test_read_when_no_sock ( self ) : <EOL> self . transport . read ( ) <EOL> def test_buffer ( self ) : <EOL> self . transport . _sock = mock ( ) <EOL> expect ( self . transport . _sock . buffer ) . args ( '<STR_LIT>' ) <EOL> self . transport . buffer ( '<STR_LIT>' ) <EOL> def test_buffer_when_no_sock ( self ) : <EOL> self . transport . buffer ( '<STR_LIT>' ) <EOL> def test_write ( self ) : <EOL> self . transport . _sock = mock ( ) <EOL> expect ( self . transport . _sock . write ) . args ( '<STR_LIT>' ) <EOL> self . transport . write ( '<STR_LIT>' ) <EOL> def test_write_when_no_sock ( self ) : <EOL> self . transport . write ( '<STR_LIT>' ) <EOL> def test_disconnect ( self ) : <EOL> self . transport . _sock = mock ( ) <EOL> self . transport . _sock . close_cb = '<STR_LIT>' <EOL> expect ( self . transport . _sock . close ) <EOL> self . transport . disconnect ( ) <EOL> assert_equals ( None , self . transport . _sock . close_cb ) <EOL> def test_disconnect_when_no_sock ( self ) : <EOL> self . transport . disconnect ( ) </s>
