<s> import pytest <EOL> import time <EOL> from serfclient import connection <EOL> class TestSerfConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> @ pytest . fixture <EOL> def rpc ( self ) : <EOL> return connection . SerfConnection ( ) <EOL> def test_has_a_default_host_port_and_timeout ( self , rpc ) : <EOL> assert rpc . host == '<STR_LIT:localhost>' <EOL> assert rpc . port == <NUM_LIT> <EOL> assert rpc . timeout == <NUM_LIT:3> <EOL> def test_allows_passing_host_port_and_timeout ( self ) : <EOL> rpc = connection . SerfConnection ( host = '<STR_LIT:foo>' , port = <NUM_LIT> , timeout = <NUM_LIT> ) <EOL> assert rpc . host == '<STR_LIT:foo>' <EOL> assert rpc . port == <NUM_LIT> <EOL> assert rpc . timeout == <NUM_LIT> <EOL> def test_representation ( self , rpc ) : <EOL> assert str ( rpc ) == '<STR_LIT>' <EOL> def test_connection_to_bad_socket_throws_exception ( self ) : <EOL> rpc = connection . SerfConnection ( port = <NUM_LIT> ) <EOL> with pytest . raises ( connection . SerfConnectionError ) as exceptionInfo : <EOL> rpc . handshake ( ) <EOL> assert '<STR_LIT>' in str ( exceptionInfo ) <EOL> def test_handshake_to_serf_agent ( self , rpc ) : <EOL> assert rpc . handshake ( ) . head == { b'<STR_LIT>' : <NUM_LIT:0> , b'<STR_LIT>' : b'<STR_LIT>' } <EOL> def test_call_throws_exception_if_socket_none ( self , rpc ) : <EOL> with pytest . raises ( connection . SerfConnectionError ) as exceptionInfo : <EOL> rpc . call ( '<STR_LIT>' ) <EOL> assert '<STR_LIT>' in str ( exceptionInfo ) <EOL> def test_handshake_and_call_increments_counter ( self , rpc ) : <EOL> assert '<STR_LIT>' in str ( rpc ) <EOL> rpc . handshake ( ) <EOL> assert '<STR_LIT>' in str ( rpc ) <EOL> rpc . call ( '<STR_LIT>' , <EOL> { "<STR_LIT:Name>" : "<STR_LIT:foo>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : True } , <EOL> expect_body = False ) <EOL> assert '<STR_LIT>' in str ( rpc ) <EOL> def test_msgpack_object_stream_decode ( self , rpc ) : <EOL> rpc . handshake ( ) <EOL> result = rpc . call ( '<STR_LIT>' ) <EOL> assert result . head == { b'<STR_LIT>' : b'<STR_LIT>' , b'<STR_LIT>' : <NUM_LIT:1> } <EOL> assert b'<STR_LIT>' in result . body . keys ( ) <EOL> def test_small_socket_recv_size ( self , rpc ) : <EOL> rpc . handshake ( ) <EOL> rpc . _socket_recv_size = <NUM_LIT:7> <EOL> result = rpc . call ( '<STR_LIT>' ) <EOL> assert result . head == { b'<STR_LIT>' : b'<STR_LIT>' , b'<STR_LIT>' : <NUM_LIT:1> } <EOL> assert b'<STR_LIT>' in result . body . keys ( ) <EOL> def test_rpc_timeout ( self , rpc ) : <EOL> rpc . timeout = <NUM_LIT:0.1> <EOL> rpc . handshake ( ) <EOL> with pytest . raises ( connection . SerfTimeout ) : <EOL> rpc . call ( '<STR_LIT>' , <EOL> { "<STR_LIT:Name>" : "<STR_LIT:foo>" , "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True } , <EOL> expect_body = True ) <EOL> def test_rxing_too_many_messages ( self , rpc ) : <EOL> rpc . handshake ( ) <EOL> for index in range ( <NUM_LIT:2> ) : <EOL> rpc . _socket . sendall ( <EOL> connection . msgpack . packb ( { "<STR_LIT>" : rpc . _counter ( ) , <EOL> "<STR_LIT>" : "<STR_LIT>" } ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> with pytest . raises ( connection . SerfProtocolError ) : <EOL> rpc . call ( '<STR_LIT>' ) <EOL> def test_connection_closed ( self , rpc ) : <EOL> rpc . handshake ( ) <EOL> class MockSocket : <EOL> def sendall ( self , content ) : <EOL> pass <EOL> def recv ( self , size ) : <EOL> return "<STR_LIT>" <EOL> rpc . _socket = MockSocket ( ) <EOL> with pytest . raises ( connection . SerfConnectionError ) : <EOL> rpc . handshake ( ) </s>
