<s> from __future__ import unicode_literals <EOL> from collections import namedtuple <EOL> from bpython . curtsies import combined_events <EOL> from bpython . test import ( FixLanguageTestCase as TestCase , unittest ) <EOL> import curtsies . events <EOL> ScheduledEvent = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class EventGenerator ( object ) : <EOL> def __init__ ( self , initial_events = ( ) , scheduled_events = ( ) ) : <EOL> self . _events = [ ] <EOL> self . _current_tick = <NUM_LIT:0> <EOL> for e in initial_events : <EOL> self . schedule_event ( e , <NUM_LIT:0> ) <EOL> for e , w in scheduled_events : <EOL> self . schedule_event ( e , w ) <EOL> def schedule_event ( self , event , when ) : <EOL> self . _events . append ( ScheduledEvent ( when , event ) ) <EOL> self . _events . sort ( ) <EOL> def send ( self , timeout = None ) : <EOL> if timeout not in [ None , <NUM_LIT:0> ] : <EOL> raise ValueError ( '<STR_LIT>' % timeout ) <EOL> if not self . _events : <EOL> return None <EOL> if self . _events [ <NUM_LIT:0> ] . when <= self . _current_tick : <EOL> return self . _events . pop ( <NUM_LIT:0> ) . event <EOL> if timeout == <NUM_LIT:0> : <EOL> return None <EOL> elif timeout is None : <EOL> e = self . _events . pop ( <NUM_LIT:0> ) <EOL> self . _current_tick = e . when <EOL> return e . event <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % timeout ) <EOL> def tick ( self , dt = <NUM_LIT:1> ) : <EOL> self . _current_tick += dt <EOL> return self . _current_tick <EOL> class TestCurtsiesPasteDetection ( TestCase ) : <EOL> def test_paste_threshold ( self ) : <EOL> eg = EventGenerator ( list ( '<STR_LIT:abc>' ) ) <EOL> cb = combined_events ( eg , paste_threshold = <NUM_LIT:3> ) <EOL> e = next ( cb ) <EOL> self . assertIsInstance ( e , curtsies . events . PasteEvent ) <EOL> self . assertEqual ( e . events , list ( '<STR_LIT:abc>' ) ) <EOL> self . assertEqual ( next ( cb ) , None ) <EOL> eg = EventGenerator ( list ( '<STR_LIT:abc>' ) ) <EOL> cb = combined_events ( eg , paste_threshold = <NUM_LIT:4> ) <EOL> self . assertEqual ( next ( cb ) , '<STR_LIT:a>' ) <EOL> self . assertEqual ( next ( cb ) , '<STR_LIT:b>' ) <EOL> self . assertEqual ( next ( cb ) , '<STR_LIT:c>' ) <EOL> self . assertEqual ( next ( cb ) , None ) <EOL> def test_set_timeout ( self ) : <EOL> eg = EventGenerator ( '<STR_LIT:a>' , zip ( '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:3> , <NUM_LIT:3> , <NUM_LIT:4> ] ) ) <EOL> eg . schedule_event ( curtsies . events . SigIntEvent ( ) , <NUM_LIT:5> ) <EOL> eg . schedule_event ( '<STR_LIT:h>' , <NUM_LIT:6> ) <EOL> cb = combined_events ( eg , paste_threshold = <NUM_LIT:3> ) <EOL> self . assertEqual ( next ( cb ) , '<STR_LIT:a>' ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , None ) <EOL> self . assertEqual ( next ( cb ) , '<STR_LIT:b>' ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , None ) <EOL> eg . tick ( ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , '<STR_LIT:c>' ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , None ) <EOL> eg . tick ( ) <EOL> self . assertIsInstance ( cb . send ( <NUM_LIT:0> ) , curtsies . events . PasteEvent ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , None ) <EOL> self . assertEqual ( cb . send ( None ) , '<STR_LIT:g>' ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , None ) <EOL> eg . tick ( <NUM_LIT:1> ) <EOL> self . assertIsInstance ( cb . send ( <NUM_LIT:0> ) , curtsies . events . SigIntEvent ) <EOL> self . assertEqual ( cb . send ( <NUM_LIT:0> ) , None ) <EOL> self . assertEqual ( cb . send ( None ) , '<STR_LIT:h>' ) <EOL> self . assertEqual ( cb . send ( None ) , None ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
