<s> """<STR_LIT>""" <EOL> import unittest <EOL> import os <EOL> import time <EOL> import logging <EOL> from decimal import Decimal <EOL> from event_loop import Timer , TimerManager , EventLoop <EOL> from packet_reader import PacketReader <EOL> class TestingTimeFunction ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . time = <NUM_LIT:0> <EOL> def __call__ ( self ) : <EOL> return self . time + <NUM_LIT> <EOL> class MockCallback ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . nb_calls = <NUM_LIT:0> <EOL> def __call__ ( self ) : <EOL> self . nb_calls += <NUM_LIT:1> <EOL> class FileLike ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fd ) : <EOL> self . fd = fd <EOL> def fileno ( self ) : <EOL> return self . fd <EOL> class TestTimer ( unittest . TestCase ) : <EOL> def test_non_repeating ( self ) : <EOL> self . _non_repeating ( decorate = False ) <EOL> def test_decorate ( self ) : <EOL> self . _non_repeating ( decorate = False ) <EOL> def _non_repeating ( self , decorate ) : <EOL> for delay in ( <NUM_LIT:0> , <NUM_LIT:1> ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> if decorate : <EOL> timer = Timer . decorate ( <NUM_LIT:10> , _time_function = time ) ( callback ) <EOL> else : <EOL> timer = Timer ( callback , <NUM_LIT:10> , _time_function = time ) <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> time . time = <NUM_LIT:9> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:1> <EOL> time . time = <NUM_LIT:10> + delay <EOL> assert timer . sleep_time ( ) == <NUM_LIT:0> <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> time . time = <NUM_LIT:15> <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> def test_cancel ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> timer = Timer ( callback , <NUM_LIT:10> , _time_function = time ) <EOL> time . time = <NUM_LIT:9> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert timer . sleep_time ( ) == <NUM_LIT:1> <EOL> timer . cancel ( ) <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> def test_reset_early ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> timer = Timer ( callback , <NUM_LIT:10> , _time_function = time ) <EOL> time . time = <NUM_LIT:7> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:3> <EOL> timer . reset ( ) <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> time . time = <NUM_LIT:16> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:1> <EOL> time . time = <NUM_LIT> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:0> <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> def test_reset_late ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> timer = Timer ( callback , <NUM_LIT:10> , _time_function = time ) <EOL> time . time = <NUM_LIT> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:0> <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> timer . reset ( ) <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> time . time = <NUM_LIT> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:2> <EOL> time . time = <NUM_LIT:30> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:0> <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:2> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> def test_repeating ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> timer = Timer ( callback , <NUM_LIT:10> , repeat = True , _time_function = time ) <EOL> time . time = <NUM_LIT:7> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:3> <EOL> time . time = <NUM_LIT> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:6> <EOL> time . time = <NUM_LIT> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:2> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> def test_cancel_repeating ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> timer = Timer ( callback , <NUM_LIT:10> , repeat = True , _time_function = time ) <EOL> time . time = <NUM_LIT:10> <EOL> timer . run ( ) <EOL> time . time = <NUM_LIT> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:2> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:6> <EOL> time . time = <NUM_LIT> <EOL> timer . cancel ( ) <EOL> for time . time in ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) : <EOL> still_alive = timer . run ( ) <EOL> assert not still_alive <EOL> assert callback . nb_calls == <NUM_LIT:2> <EOL> assert timer . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> def test_reset_repeating ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> callback = MockCallback ( ) <EOL> timer = Timer ( callback , <NUM_LIT:10> , repeat = True , _time_function = time ) <EOL> time . time = <NUM_LIT:12> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:8> <EOL> timer . reset ( ) <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> time . time = <NUM_LIT> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:0> <EOL> timer . reset ( ) <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> still_alive = timer . run ( ) <EOL> assert still_alive <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert timer . sleep_time ( ) == <NUM_LIT:10> <EOL> class TestTimerManager ( unittest . TestCase ) : <EOL> def test_empty_timer_list ( self ) : <EOL> manager = TimerManager ( ) <EOL> assert manager . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> def test_invalid_timeouts ( self ) : <EOL> manager = TimerManager ( ) <EOL> self . assertRaises ( AssertionError , manager . add_timer , <NUM_LIT:0> , None ) <EOL> self . assertRaises ( AssertionError , manager . add_timer , - <NUM_LIT:1> , None ) <EOL> def test_single_timer ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> manager = TimerManager ( _time_function = time ) <EOL> callback = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:30> , callback ) <EOL> assert manager . sleep_time ( ) == <NUM_LIT:30> <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> time . time = <NUM_LIT> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:8> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> time . time = <NUM_LIT:30> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:0> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert manager . sleep_time ( ) == Decimal ( '<STR_LIT>' ) <EOL> time . time = <NUM_LIT:100> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> def test_single_repeating_timer ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> manager = TimerManager ( _time_function = time ) <EOL> callback = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:30> , callback , repeat = True ) <EOL> assert manager . sleep_time ( ) == <NUM_LIT:30> <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> time . time = <NUM_LIT> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:8> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:0> <EOL> time . time = <NUM_LIT:30> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:0> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:1> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:30> <EOL> time . time = <NUM_LIT> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:0> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:2> <EOL> assert manager . sleep_time ( ) == <NUM_LIT> <EOL> time . time = <NUM_LIT:200> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:0> <EOL> manager . run ( ) <EOL> assert callback . nb_calls == <NUM_LIT:3> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:10> <EOL> def test_long_callback ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> manager = TimerManager ( _time_function = time ) <EOL> def loose_time ( ) : <EOL> time . time += <NUM_LIT:11> <EOL> manager . add_timer ( <NUM_LIT:2> , loose_time , repeat = True ) <EOL> manager . run ( ) <EOL> assert time . time == <NUM_LIT:0> <EOL> time . time = <NUM_LIT:2> <EOL> manager . run ( ) <EOL> assert time . time == <NUM_LIT> <EOL> assert manager . sleep_time ( ) == <NUM_LIT:0> <EOL> def test_many ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> manager = TimerManager ( _time_function = time ) <EOL> c5 = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:5> , c5 , repeat = True ) <EOL> c7 = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:7> , c7 , repeat = True ) <EOL> c13 = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT> , c13 , repeat = True ) <EOL> nb_sleeps = <NUM_LIT:0> <EOL> while <NUM_LIT:1> : <EOL> time . time += manager . sleep_time ( ) <EOL> if time . time >= <NUM_LIT:100> : <EOL> break <EOL> nb_sleeps += <NUM_LIT:1> <EOL> manager . run ( ) <EOL> assert c5 . nb_calls == <NUM_LIT> <EOL> assert c7 . nb_calls == <NUM_LIT> <EOL> assert c13 . nb_calls == <NUM_LIT:7> <EOL> assert nb_sleeps == ( c5 . nb_calls + c7 . nb_calls + c13 . nb_calls - <NUM_LIT:4> ) <EOL> def test_contsant_sleep ( self ) : <EOL> time = TestingTimeFunction ( ) <EOL> manager = TimerManager ( _time_function = time ) <EOL> c5 = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:5> , c5 , repeat = True ) <EOL> c7 = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:7> , c7 , repeat = True ) <EOL> c13 = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT> , c13 , repeat = True ) <EOL> for time . time in xrange ( <NUM_LIT:100> ) : <EOL> manager . run ( ) <EOL> if time . time == <NUM_LIT> : <EOL> def check_time ( ) : <EOL> assert time . time == <NUM_LIT> <EOL> one_shot = MockCallback ( ) <EOL> manager . add_timer ( <NUM_LIT:6> , check_time ) <EOL> manager . add_timer ( <NUM_LIT:6> , one_shot ) <EOL> assert c5 . nb_calls == <NUM_LIT> <EOL> assert c7 . nb_calls == <NUM_LIT> <EOL> assert c13 . nb_calls == <NUM_LIT:7> <EOL> assert one_shot . nb_calls == <NUM_LIT:1> <EOL> class TestEventLoop ( unittest . TestCase ) : <EOL> def _simple ( self , reader , writer ) : <EOL> loop = EventLoop ( ) <EOL> nb_reads = [ <NUM_LIT:0> ] <EOL> @ loop . block_reader ( reader ) <EOL> def incoming ( data ) : <EOL> assert data == '<STR_LIT:foo>' <EOL> nb_reads [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> loop . stop ( ) <EOL> assert os . write ( writer , '<STR_LIT:foo>' ) == <NUM_LIT:3> <EOL> loop . run ( ) <EOL> assert nb_reads [ <NUM_LIT:0> ] == <NUM_LIT:1> <EOL> def test_simple_pipe ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> self . _simple ( reader , writer ) <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> def test_filelike ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> self . _simple ( FileLike ( reader ) , writer ) <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> def test_pipe ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> loop = EventLoop ( ) <EOL> nb_reads = [ <NUM_LIT:0> ] <EOL> nb_writes = [ <NUM_LIT:0> ] <EOL> duration = <NUM_LIT> <EOL> interval = <NUM_LIT> <EOL> expected_nb = <NUM_LIT:10> <EOL> assert expected_nb == duration // interval <EOL> assert expected_nb < duration / interval <EOL> start = time . time ( ) <EOL> loop . add_timer ( duration ) ( loop . stop ) <EOL> @ loop . add_timer ( interval , repeat = True ) <EOL> def write_something ( ) : <EOL> assert os . write ( writer , '<STR_LIT:foo>' ) == <NUM_LIT:3> <EOL> nb_writes [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> @ loop . block_reader ( reader ) <EOL> def incoming ( data ) : <EOL> assert data == '<STR_LIT:foo>' <EOL> nb_reads [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> loop . run ( ) <EOL> assert round ( time . time ( ) - start , <NUM_LIT:2> ) == duration <EOL> assert nb_writes [ <NUM_LIT:0> ] == expected_nb , nb_writes <EOL> assert nb_reads [ <NUM_LIT:0> ] == expected_nb , nb_reads <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> class TestLineReader ( unittest . TestCase ) : <EOL> def test_line_reader ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> loop = EventLoop ( ) <EOL> data = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> data = data [ : : - <NUM_LIT:1> ] <EOL> start = time . time ( ) <EOL> @ loop . add_timer ( <NUM_LIT> , repeat = True ) <EOL> def slow_write ( ) : <EOL> if data : <EOL> d = data . pop ( ) <EOL> assert os . write ( writer , d ) == len ( d ) <EOL> else : <EOL> loop . stop ( ) <EOL> lines = [ ] <EOL> loop . line_reader ( reader , max_block_size = <NUM_LIT:5> ) ( lines . append ) <EOL> loop . run ( ) <EOL> assert lines == [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> def test_timing ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> loop = EventLoop ( ) <EOL> data = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> data = data [ : : - <NUM_LIT:1> ] <EOL> @ loop . add_timer ( <NUM_LIT> , repeat = True ) <EOL> def slow_write ( ) : <EOL> if data : <EOL> os . write ( writer , data . pop ( ) ) <EOL> else : <EOL> loop . stop ( ) <EOL> lines = [ ] <EOL> @ loop . line_reader ( reader ) <EOL> def new_line ( line ) : <EOL> lines . append ( line ) <EOL> expected_time = { <EOL> '<STR_LIT:L>' : <NUM_LIT> , '<STR_LIT:d>' : <NUM_LIT> , '<STR_LIT:s>' : <NUM_LIT> , '<STR_LIT:a>' : <NUM_LIT> } [ line [ <NUM_LIT:0> ] ] <EOL> assert round ( time . time ( ) - start , <NUM_LIT:2> ) == expected_time <EOL> time . sleep ( <NUM_LIT> ) <EOL> slow_write . reset ( ) <EOL> start = time . time ( ) <EOL> loop . run ( ) <EOL> assert lines == [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> class TestPushBack ( unittest . TestCase ) : <EOL> def test_push_back ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> data = '<STR_LIT>' <EOL> assert os . write ( writer , data ) == len ( data ) <EOL> loop = EventLoop ( ) <EOL> state = [ <NUM_LIT:1> ] <EOL> @ loop . push_back_reader ( reader , max_block_size = <NUM_LIT:5> ) <EOL> def new_block ( data , push_back ) : <EOL> if state [ <NUM_LIT:0> ] == <NUM_LIT:1> : <EOL> assert data == '<STR_LIT>' <EOL> elif state [ <NUM_LIT:0> ] == <NUM_LIT:2> : <EOL> assert data == '<STR_LIT>' <EOL> push_back ( data ) <EOL> elif state [ <NUM_LIT:0> ] == <NUM_LIT:3> : <EOL> assert data == '<STR_LIT>' <EOL> push_back ( '<STR_LIT:d>' ) <EOL> push_back ( '<STR_LIT>' ) <EOL> elif state [ <NUM_LIT:0> ] == <NUM_LIT:4> : <EOL> assert data == '<STR_LIT>' <EOL> elif state [ <NUM_LIT:0> ] == <NUM_LIT:5> : <EOL> assert data == '<STR_LIT>' <EOL> loop . stop ( ) <EOL> else : <EOL> assert False <EOL> state [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> loop . run ( ) <EOL> assert state [ <NUM_LIT:0> ] == <NUM_LIT:6> <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> class TestPacketReader ( unittest . TestCase ) : <EOL> def test_packets ( self ) : <EOL> reader , writer = os . pipe ( ) <EOL> try : <EOL> loop = EventLoop ( ) <EOL> original_packets = [ <EOL> '<STR_LIT:foo>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> packets = [ ] <EOL> def callback ( packet ) : <EOL> packets . append ( packet ) <EOL> assert packet_reader . dropped_bytes == { <EOL> <NUM_LIT:1> : <NUM_LIT:0> , <NUM_LIT:2> : <NUM_LIT:1> , <NUM_LIT:3> : <NUM_LIT:3> , <NUM_LIT:4> : <NUM_LIT:6> <EOL> } [ len ( packets ) ] <EOL> if len ( packets ) == len ( original_packets ) : <EOL> loop . stop ( ) <EOL> def write ( data ) : <EOL> assert os . write ( writer , data ) == len ( data ) <EOL> for i , packet in enumerate ( original_packets ) : <EOL> write ( '<STR_LIT>' [ : i ] ) <EOL> write ( PacketReader . PACKET_DELIMITER ) <EOL> write ( chr ( len ( packet ) + <NUM_LIT:1> ) ) <EOL> write ( packet ) <EOL> packet_reader = PacketReader ( loop , reader , callback , <EOL> max_block_size = <NUM_LIT:3> ) <EOL> loop . run ( ) <EOL> assert packets == original_packets <EOL> finally : <EOL> os . close ( reader ) <EOL> os . close ( writer ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
