<s> import unittest <EOL> from datetime import datetime , timedelta <EOL> from time import sleep <EOL> from rx . concurrency import NewThreadScheduler <EOL> class TestNewThreadScheduler ( unittest . TestCase ) : <EOL> def test_new_thread_now ( self ) : <EOL> scheduler = NewThreadScheduler ( ) <EOL> res = scheduler . now ( ) - datetime . utcnow ( ) <EOL> assert res < timedelta ( microseconds = <NUM_LIT:1000> ) <EOL> def test_new_thread_schedule_action ( self ) : <EOL> scheduler = NewThreadScheduler ( ) <EOL> ran = [ False ] <EOL> def action ( scheduler , state ) : <EOL> ran [ <NUM_LIT:0> ] = True <EOL> scheduler . schedule ( action ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> assert ( ran [ <NUM_LIT:0> ] == True ) <EOL> def test_new_thread_schedule_action_due ( self ) : <EOL> scheduler = NewThreadScheduler ( ) <EOL> starttime = datetime . utcnow ( ) <EOL> endtime = [ None ] <EOL> def action ( scheduler , state ) : <EOL> endtime [ <NUM_LIT:0> ] = datetime . utcnow ( ) <EOL> scheduler . schedule_relative ( timedelta ( milliseconds = <NUM_LIT:200> ) , action ) <EOL> sleep ( <NUM_LIT> ) <EOL> diff = endtime [ <NUM_LIT:0> ] - starttime <EOL> assert ( diff > timedelta ( milliseconds = <NUM_LIT> ) ) <EOL> def test_new_thread_schedule_action_cancel ( self ) : <EOL> ran = [ False ] <EOL> scheduler = NewThreadScheduler ( ) <EOL> def action ( scheduler , state ) : <EOL> ran [ <NUM_LIT:0> ] = True <EOL> d = scheduler . schedule_relative ( timedelta ( milliseconds = <NUM_LIT:1> ) , action ) <EOL> d . dispose ( ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> assert ( not ran [ <NUM_LIT:0> ] ) </s>
