<s> '''<STR_LIT>''' <EOL> from chai import Chai <EOL> import errno <EOL> import unittest <EOL> try : <EOL> import gevent <EOL> from gevent . coros import Semaphore <EOL> from gevent import socket <EOL> from gevent . pool import Pool <EOL> from haigha . transports import gevent_transport <EOL> from haigha . transports . gevent_transport import * <EOL> except ImportError : <EOL> import warnings <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> gevent = None <EOL> @ unittest . skipIf ( gevent is None , '<STR_LIT>' ) <EOL> class GeventTransportTest ( Chai ) : <EOL> def setUp ( self ) : <EOL> super ( GeventTransportTest , self ) . setUp ( ) <EOL> self . connection = mock ( ) <EOL> self . transport = GeventTransport ( self . connection ) <EOL> self . transport . _host = '<STR_LIT>' <EOL> def test_init ( self ) : <EOL> assert_equals ( bytearray ( ) , self . transport . _buffer ) <EOL> assert_true ( isinstance ( self . transport . _read_lock , Semaphore ) ) <EOL> assert_true ( isinstance ( self . transport . _write_lock , Semaphore ) ) <EOL> def test_connect ( self ) : <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . connect ) . args ( <EOL> ( '<STR_LIT:host>' , '<STR_LIT:port>' ) , klass = is_arg ( socket . socket ) ) . returns ( '<STR_LIT>' ) <EOL> self . transport . connect ( ( '<STR_LIT:host>' , '<STR_LIT:port>' ) ) <EOL> def test_read ( self ) : <EOL> expect ( self . transport . _read_lock . acquire ) <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . read ) . args ( timeout = None ) . returns ( '<STR_LIT>' ) <EOL> expect ( self . transport . _read_lock . release ) <EOL> expect ( self . transport . _read_wait . set ) <EOL> expect ( self . transport . _read_wait . clear ) <EOL> assert_equals ( '<STR_LIT>' , self . transport . read ( ) ) <EOL> def test_read_when_already_locked ( self ) : <EOL> expect ( self . transport . _read_lock . locked ) . returns ( True ) <EOL> stub ( self . transport . _read_lock . acquire ) <EOL> stub ( mock ( gevent_transport , '<STR_LIT>' ) ) <EOL> stub ( self . transport . _read_lock . release ) <EOL> expect ( self . transport . _read_wait . wait ) <EOL> assert_equals ( None , self . transport . read ( ) ) <EOL> def test_read_when_raises_exception ( self ) : <EOL> expect ( self . transport . _read_lock . acquire ) <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . read ) . args ( timeout = '<STR_LIT:5>' ) . raises ( Exception ( '<STR_LIT>' ) ) <EOL> expect ( self . transport . _read_lock . release ) <EOL> assert_raises ( Exception , self . transport . read , timeout = '<STR_LIT:5>' ) <EOL> def test_buffer ( self ) : <EOL> expect ( self . transport . _read_lock . acquire ) <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . buffer ) . args ( '<STR_LIT>' ) <EOL> expect ( self . transport . _read_lock . release ) <EOL> self . transport . buffer ( '<STR_LIT>' ) <EOL> def test_buffer_when_raises_exception ( self ) : <EOL> expect ( self . transport . _read_lock . acquire ) <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . buffer ) . args ( '<STR_LIT>' ) . raises ( Exception ( '<STR_LIT>' ) ) <EOL> expect ( self . transport . _read_lock . release ) <EOL> assert_raises ( Exception , self . transport . buffer , '<STR_LIT>' ) <EOL> def test_write ( self ) : <EOL> expect ( self . transport . _write_lock . acquire ) <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . write ) . args ( '<STR_LIT>' ) <EOL> expect ( self . transport . _write_lock . release ) <EOL> self . transport . write ( '<STR_LIT>' ) <EOL> def test_write_when_raises_an_exception ( self ) : <EOL> expect ( self . transport . _write_lock . acquire ) <EOL> with expect ( mock ( gevent_transport , '<STR_LIT>' ) ) . args ( is_arg ( GeventTransport ) , GeventTransport ) . returns ( mock ( ) ) as parent : <EOL> expect ( parent . write ) . args ( '<STR_LIT>' ) . raises ( Exception ( '<STR_LIT>' ) ) <EOL> expect ( self . transport . _write_lock . release ) <EOL> assert_raises ( Exception , self . transport . write , '<STR_LIT>' ) <EOL> @ unittest . skipIf ( gevent is None , '<STR_LIT>' ) <EOL> class GeventPoolTransportTest ( Chai ) : <EOL> def setUp ( self ) : <EOL> super ( GeventPoolTransportTest , self ) . setUp ( ) <EOL> self . connection = mock ( ) <EOL> self . transport = GeventPoolTransport ( self . connection ) <EOL> self . transport . _host = '<STR_LIT>' <EOL> def test_init ( self ) : <EOL> assert_equals ( bytearray ( ) , self . transport . _buffer ) <EOL> assert_true ( isinstance ( self . transport . _read_lock , Semaphore ) ) <EOL> assert_true ( isinstance ( self . transport . pool , Pool ) ) <EOL> trans = GeventPoolTransport ( self . connection , pool = '<STR_LIT>' ) <EOL> assert_equals ( '<STR_LIT>' , trans . pool ) <EOL> def test_process_channels ( self ) : <EOL> chs = [ mock ( ) , mock ( ) ] <EOL> self . transport . _pool = mock ( ) <EOL> expect ( self . transport . _pool . spawn ) . args ( chs [ <NUM_LIT:0> ] . process_frames ) <EOL> expect ( self . transport . _pool . spawn ) . args ( chs [ <NUM_LIT:1> ] . process_frames ) <EOL> self . transport . process_channels ( chs ) </s>
