<s> from sparts . tasks . periodic import PeriodicTask <EOL> from sparts . tests . base import SingleTaskTestCase <EOL> from sparts . timer import Timer <EOL> from sparts . vtask import TryLater <EOL> import time <EOL> import threading <EOL> class MyTask ( PeriodicTask ) : <EOL> INTERVAL = <NUM_LIT> <EOL> counter = <NUM_LIT:0> <EOL> fail_async = False <EOL> trylater = False <EOL> def initTask ( self ) : <EOL> super ( MyTask , self ) . initTask ( ) <EOL> self . visit_threads = set ( ) <EOL> def execute ( self ) : <EOL> if self . fail_async and self . has_pending ( ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if self . trylater : <EOL> raise TryLater ( str ( self . trylater ) , after = self . trylater ) <EOL> self . logger . debug ( "<STR_LIT>" , time . time ( ) ) <EOL> self . counter += <NUM_LIT:1> <EOL> self . visit_threads . add ( threading . current_thread ( ) . ident ) <EOL> return self . counter <EOL> class TestMyTask ( SingleTaskTestCase ) : <EOL> TASK = MyTask <EOL> def test_execute_happens ( self ) : <EOL> with Timer ( ) as t : <EOL> while self . task . counter <= <NUM_LIT:0> and t . elapsed < <NUM_LIT> : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertGreater ( self . task . counter , <NUM_LIT:0> ) <EOL> def test_execute_async ( self ) : <EOL> f = self . task . execute_async ( ) <EOL> res = f . result ( <NUM_LIT> ) <EOL> self . assertNotNone ( res ) <EOL> self . assertGreater ( res , <NUM_LIT:0> ) <EOL> self . task . fail_async = True <EOL> with self . assertRaises ( Exception ) as ctx : <EOL> self . task . execute_async ( ) . result ( <NUM_LIT:1.0> ) <EOL> self . task . execute_async ( ) . result ( <NUM_LIT:1.0> ) <EOL> self . assertEqual ( ctx . exception . args [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> with Timer ( ) as t : <EOL> while t . elapsed < <NUM_LIT> : <EOL> if not self . task . running : <EOL> break <EOL> self . assertFalse ( self . task . running ) <EOL> f = self . task . execute_async ( ) <EOL> with self . assertRaises ( Exception ) as ctx : <EOL> f . result ( ) <EOL> self . assertEqual ( ctx . exception . args [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> def test_trylater_after ( self ) : <EOL> t = self . task <EOL> t . _handle_try_later = self . mock . Mock ( wraps = t . _handle_try_later ) <EOL> t . stop_event = self . mock . Mock ( wraps = t . stop_event ) <EOL> t . execute_async ( ) . result ( ) <EOL> self . assertFalse ( t . _handle_try_later . called ) <EOL> t . trylater = <NUM_LIT> <EOL> t0 = time . time ( ) <EOL> while not t . _handle_try_later . called and time . time ( ) - t0 < <NUM_LIT> : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertTrue ( t . _handle_try_later ) <EOL> t . trylater = None <EOL> t . execute_async ( ) . result ( <NUM_LIT> ) <EOL> self . assertTrue ( t . _handle_try_later ) <EOL> self . task . stop_event . wait . assert_any_call ( <NUM_LIT> ) <EOL> class MyMultiTask ( MyTask ) : <EOL> workers = <NUM_LIT:5> <EOL> class TestMultiTask ( SingleTaskTestCase ) : <EOL> TASK = MyMultiTask <EOL> def test_multi_execute ( self ) : <EOL> with Timer ( ) as t : <EOL> while len ( self . task . visit_threads ) < <NUM_LIT:5> and t . elapsed < <NUM_LIT> : <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertGreaterEqual ( self . task . counter , <NUM_LIT:5> ) </s>
