<s> from o_test import test , tests <EOL> from contextlib import contextmanager <EOL> from monocle import _o <EOL> from monocle . callback import Callback <EOL> from monocle . stack import eventloop , network <EOL> from monocle . util import sleep <EOL> EOL = '<STR_LIT:\r\n>' <EOL> class StackConnection ( object ) : <EOL> def __init__ ( self ) : <EOL> self . buffer = "<STR_LIT>" <EOL> self . is_closed = False <EOL> self . is_reading = False <EOL> self . read_cb = Callback ( ) <EOL> self . connect_cb = Callback ( ) <EOL> self . out = [ ] <EOL> self . disconnect_called = <NUM_LIT:0> <EOL> self . resume_called = <NUM_LIT:0> <EOL> def disconnect ( self ) : <EOL> self . disconnect_called += <NUM_LIT:1> <EOL> def closed ( self ) : <EOL> return self . is_closed <EOL> def reading ( self ) : <EOL> return self . is_reading <EOL> def resume ( self ) : <EOL> self . resume_called += <NUM_LIT:1> <EOL> def write ( self , data ) : <EOL> self . out . append ( data ) <EOL> @ tests <EOL> class ConnectionTestCase ( object ) : <EOL> def __init__ ( self ) : <EOL> self . stack_conn = StackConnection ( ) <EOL> self . conn = network . Connection ( stack_conn = self . stack_conn ) <EOL> @ property <EOL> def buffer ( self ) : <EOL> return self . stack_conn . buffer <EOL> @ buffer . setter <EOL> def buffer ( self , value ) : <EOL> self . stack_conn . buffer = value <EOL> @ _o <EOL> def test ( self ) : <EOL> data = '<STR_LIT>' <EOL> self . buffer = data <EOL> r = yield self . conn . read ( <NUM_LIT:2> ) <EOL> assert r == data <EOL> assert self . buffer == '<STR_LIT>' <EOL> assert self . stack_conn . resume_called == <NUM_LIT:0> <EOL> @ _o <EOL> def test_read_delay ( self ) : <EOL> data = '<STR_LIT>' <EOL> self . buffer = '<STR_LIT:o>' <EOL> self . conn . timeout = <NUM_LIT> <EOL> def populate ( ) : <EOL> self . buffer += '<STR_LIT:k>' <EOL> self . stack_conn . read_cb ( None ) <EOL> eventloop . queue_task ( <NUM_LIT:0.1> , populate ) <EOL> r = yield self . conn . read ( <NUM_LIT:2> ) <EOL> assert r == data <EOL> assert self . stack_conn . resume_called == <NUM_LIT:1> <EOL> yield sleep ( <NUM_LIT> ) <EOL> @ _o <EOL> def test_read_timeout ( self ) : <EOL> self . conn . timeout = <NUM_LIT:0.1> <EOL> try : <EOL> yield self . conn . read ( <NUM_LIT:10> ) <EOL> except network . ConnectionLost : <EOL> pass <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> assert self . stack_conn . resume_called == <NUM_LIT:1> <EOL> @ _o <EOL> def test_read_some ( self ) : <EOL> data = '<STR_LIT>' <EOL> self . buffer = data <EOL> r = yield self . conn . read_some ( ) <EOL> assert r == data <EOL> assert self . buffer == '<STR_LIT>' <EOL> assert self . stack_conn . resume_called == <NUM_LIT:0> <EOL> @ _o <EOL> def test_read_some_delay ( self ) : <EOL> data = '<STR_LIT>' <EOL> self . conn . timeout = <NUM_LIT> <EOL> def populate ( ) : <EOL> self . buffer = data <EOL> self . stack_conn . read_cb ( None ) <EOL> eventloop . queue_task ( <NUM_LIT:0.1> , populate ) <EOL> r = yield self . conn . read_some ( ) <EOL> assert r == data <EOL> assert self . stack_conn . resume_called == <NUM_LIT:1> <EOL> yield sleep ( <NUM_LIT> ) <EOL> @ _o <EOL> def test_read_some_timeout ( self ) : <EOL> self . conn . timeout = <NUM_LIT:0.1> <EOL> try : <EOL> yield self . conn . read_some ( ) <EOL> except network . ConnectionLost : <EOL> pass <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> assert self . stack_conn . resume_called == <NUM_LIT:1> <EOL> @ _o <EOL> def test_read_util ( self ) : <EOL> data = '<STR_LIT>' <EOL> self . buffer = '<STR_LIT>' <EOL> r = yield self . conn . read_until ( '<STR_LIT:.>' ) <EOL> assert r == data <EOL> assert self . buffer == '<STR_LIT>' <EOL> assert self . stack_conn . resume_called == <NUM_LIT:0> <EOL> @ _o <EOL> def test_readline ( self ) : <EOL> data = '<STR_LIT>' <EOL> self . buffer = '<STR_LIT>' <EOL> r = yield self . conn . readline ( ) <EOL> assert r == data <EOL> assert self . buffer == '<STR_LIT>' <EOL> assert self . stack_conn . resume_called == <NUM_LIT:0> <EOL> @ contextmanager <EOL> def network_server_running ( addr , port ) : <EOL> @ _o <EOL> def handler ( conn ) : <EOL> while True : <EOL> try : <EOL> msg = yield conn . read_until ( EOL ) <EOL> except network . ConnectionLost : <EOL> break <EOL> yield conn . write ( '<STR_LIT>' + msg . strip ( ) + EOL ) <EOL> service = network . Service ( handler , bindaddr = addr , port = port ) <EOL> network . add_service ( service ) <EOL> try : <EOL> yield <EOL> finally : <EOL> service . stop ( ) <EOL> @ contextmanager <EOL> def network_client ( ) : <EOL> client = network . Client ( ) <EOL> try : <EOL> yield client <EOL> finally : <EOL> client . close ( ) <EOL> @ test <EOL> @ _o <EOL> def test_client ( ) : <EOL> addr = '<STR_LIT:127.0.0.1>' <EOL> port = <NUM_LIT> <EOL> with network_server_running ( addr , port ) : <EOL> with network_client ( ) as client : <EOL> msg = '<STR_LIT>' <EOL> yield client . connect ( addr , port ) <EOL> yield client . write ( msg + EOL ) <EOL> result = yield client . read_until ( EOL ) <EOL> assert result == '<STR_LIT>' + msg + EOL </s>
