<s> import queue <EOL> import sched <EOL> import time <EOL> import unittest <EOL> from test import support <EOL> try : <EOL> import threading <EOL> except ImportError : <EOL> threading = None <EOL> TIMEOUT = <NUM_LIT:10> <EOL> class Timer : <EOL> def __init__ ( self ) : <EOL> self . _cond = threading . Condition ( ) <EOL> self . _time = <NUM_LIT:0> <EOL> self . _stop = <NUM_LIT:0> <EOL> def time ( self ) : <EOL> with self . _cond : <EOL> return self . _time <EOL> def sleep ( self , t ) : <EOL> assert t >= <NUM_LIT:0> <EOL> with self . _cond : <EOL> t += self . _time <EOL> while self . _stop < t : <EOL> self . _time = self . _stop <EOL> self . _cond . wait ( ) <EOL> self . _time = t <EOL> def advance ( self , t ) : <EOL> assert t >= <NUM_LIT:0> <EOL> with self . _cond : <EOL> self . _stop += t <EOL> self . _cond . notify_all ( ) <EOL> class TestCase ( unittest . TestCase ) : <EOL> def test_enter ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> for x in [ <NUM_LIT:0.5> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0.1> ] : <EOL> z = scheduler . enter ( x , <NUM_LIT:1> , fun , ( x , ) ) <EOL> scheduler . run ( ) <EOL> self . assertEqual ( l , [ <NUM_LIT:0.1> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0.5> ] ) <EOL> def test_enterabs ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> for x in [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] : <EOL> z = scheduler . enterabs ( x , <NUM_LIT:1> , fun , ( x , ) ) <EOL> scheduler . run ( ) <EOL> self . assertEqual ( l , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> def test_enter_concurrent ( self ) : <EOL> q = queue . Queue ( ) <EOL> fun = q . put <EOL> timer = Timer ( ) <EOL> scheduler = sched . scheduler ( timer . time , timer . sleep ) <EOL> scheduler . enter ( <NUM_LIT:1> , <NUM_LIT:1> , fun , ( <NUM_LIT:1> , ) ) <EOL> scheduler . enter ( <NUM_LIT:3> , <NUM_LIT:1> , fun , ( <NUM_LIT:3> , ) ) <EOL> t = threading . Thread ( target = scheduler . run ) <EOL> t . start ( ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> for x in [ <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:2> ] : <EOL> z = scheduler . enter ( x - <NUM_LIT:1> , <NUM_LIT:1> , fun , ( x , ) ) <EOL> timer . advance ( <NUM_LIT:2> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:3> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:4> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:5> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> timer . advance ( <NUM_LIT:1000> ) <EOL> t . join ( timeout = TIMEOUT ) <EOL> self . assertFalse ( t . is_alive ( ) ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> self . assertEqual ( timer . time ( ) , <NUM_LIT:5> ) <EOL> def test_priority ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> for priority in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ] : <EOL> z = scheduler . enterabs ( <NUM_LIT> , priority , fun , ( priority , ) ) <EOL> scheduler . run ( ) <EOL> self . assertEqual ( l , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ] ) <EOL> def test_cancel ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> now = time . time ( ) <EOL> event1 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun , ( <NUM_LIT> , ) ) <EOL> event2 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun , ( <NUM_LIT> , ) ) <EOL> event3 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun , ( <NUM_LIT> , ) ) <EOL> event4 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun , ( <NUM_LIT> , ) ) <EOL> event5 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun , ( <NUM_LIT> , ) ) <EOL> scheduler . cancel ( event1 ) <EOL> scheduler . cancel ( event5 ) <EOL> scheduler . run ( ) <EOL> self . assertEqual ( l , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> def test_cancel_concurrent ( self ) : <EOL> q = queue . Queue ( ) <EOL> fun = q . put <EOL> timer = Timer ( ) <EOL> scheduler = sched . scheduler ( timer . time , timer . sleep ) <EOL> now = timer . time ( ) <EOL> event1 = scheduler . enterabs ( now + <NUM_LIT:1> , <NUM_LIT:1> , fun , ( <NUM_LIT:1> , ) ) <EOL> event2 = scheduler . enterabs ( now + <NUM_LIT:2> , <NUM_LIT:1> , fun , ( <NUM_LIT:2> , ) ) <EOL> event4 = scheduler . enterabs ( now + <NUM_LIT:4> , <NUM_LIT:1> , fun , ( <NUM_LIT:4> , ) ) <EOL> event5 = scheduler . enterabs ( now + <NUM_LIT:5> , <NUM_LIT:1> , fun , ( <NUM_LIT:5> , ) ) <EOL> event3 = scheduler . enterabs ( now + <NUM_LIT:3> , <NUM_LIT:1> , fun , ( <NUM_LIT:3> , ) ) <EOL> t = threading . Thread ( target = scheduler . run ) <EOL> t . start ( ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> scheduler . cancel ( event2 ) <EOL> scheduler . cancel ( event5 ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:3> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> timer . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( q . get ( timeout = TIMEOUT ) , <NUM_LIT:4> ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> timer . advance ( <NUM_LIT:1000> ) <EOL> t . join ( timeout = TIMEOUT ) <EOL> self . assertFalse ( t . is_alive ( ) ) <EOL> self . assertTrue ( q . empty ( ) ) <EOL> self . assertEqual ( timer . time ( ) , <NUM_LIT:4> ) <EOL> def test_empty ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> self . assertTrue ( scheduler . empty ( ) ) <EOL> for x in [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] : <EOL> z = scheduler . enterabs ( x , <NUM_LIT:1> , fun , ( x , ) ) <EOL> self . assertFalse ( scheduler . empty ( ) ) <EOL> scheduler . run ( ) <EOL> self . assertTrue ( scheduler . empty ( ) ) <EOL> def test_queue ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> now = time . time ( ) <EOL> e5 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun ) <EOL> e1 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun ) <EOL> e2 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun ) <EOL> e4 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun ) <EOL> e3 = scheduler . enterabs ( now + <NUM_LIT> , <NUM_LIT:1> , fun ) <EOL> self . assertEqual ( scheduler . queue , [ e1 , e2 , e3 , e4 , e5 ] ) <EOL> def test_args_kwargs ( self ) : <EOL> flag = [ ] <EOL> def fun ( * a , ** b ) : <EOL> flag . append ( None ) <EOL> self . assertEqual ( a , ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) ) <EOL> self . assertEqual ( b , { "<STR_LIT:foo>" : <NUM_LIT:1> } ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> z = scheduler . enterabs ( <NUM_LIT> , <NUM_LIT:1> , fun , argument = ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) , kwargs = { "<STR_LIT:foo>" : <NUM_LIT:1> } ) <EOL> scheduler . run ( ) <EOL> self . assertEqual ( flag , [ None ] ) <EOL> def test_run_non_blocking ( self ) : <EOL> l = [ ] <EOL> fun = lambda x : l . append ( x ) <EOL> scheduler = sched . scheduler ( time . time , time . sleep ) <EOL> for x in [ <NUM_LIT:10> , <NUM_LIT:9> , <NUM_LIT:8> , <NUM_LIT:7> , <NUM_LIT:6> ] : <EOL> scheduler . enter ( x , <NUM_LIT:1> , fun , ( x , ) ) <EOL> scheduler . run ( blocking = False ) <EOL> self . assertEqual ( l , [ ] ) <EOL> def test_main ( ) : <EOL> support . run_unittest ( TestCase ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) </s>
