<s> import sys <EOL> import time <EOL> import unittest <EOL> import pyev <EOL> from whizzer . defer import Deferred , CancelledError , AlreadyCalledError , TimeoutError <EOL> from common import loop <EOL> def throw_always ( result ) : <EOL> raise Exception ( "<STR_LIT:success>" ) <EOL> def one_always ( result ) : <EOL> return <NUM_LIT:1> <EOL> def add ( a , b ) : <EOL> return a + b <EOL> class TestDeferred ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . deferred = Deferred ( loop ) <EOL> self . result = None <EOL> def tearDown ( self ) : <EOL> self . deferred = None <EOL> self . result = None <EOL> def set_result ( self , result ) : <EOL> self . result = result <EOL> def set_exception ( self , exception ) : <EOL> self . exception = exception <EOL> def call_later ( self , delay , func , * args , ** kwargs ) : <EOL> timer = pyev . Timer ( delay , <NUM_LIT:0.0> , loop , self . _do_later , ( func , args , kwargs ) ) <EOL> timer . start ( ) <EOL> return timer <EOL> def _do_later ( self , watcher , events ) : <EOL> ( func , args , kwargs ) = watcher . data <EOL> func ( * args , ** kwargs ) <EOL> watcher . stop ( ) <EOL> def test_callback ( self ) : <EOL> self . deferred . add_callback ( self . set_result ) <EOL> self . deferred . callback ( <NUM_LIT:5> ) <EOL> self . assertTrue ( self . result == <NUM_LIT:5> ) <EOL> def test_callback_chain ( self ) : <EOL> d = self . deferred <EOL> d . add_callback ( add , <NUM_LIT:1> ) <EOL> d . add_callback ( self . set_result ) <EOL> self . deferred . callback ( <NUM_LIT:5> ) <EOL> self . assertTrue ( self . result == <NUM_LIT:6> ) <EOL> def test_log_error ( self ) : <EOL> """<STR_LIT>""" <EOL> self . deferred . add_callback ( throw_always ) <EOL> self . deferred . callback ( None ) <EOL> self . deferred = None <EOL> def test_errback ( self ) : <EOL> self . deferred . add_errback ( self . set_result ) <EOL> self . deferred . errback ( Exception ( ) ) <EOL> self . assertTrue ( isinstance ( self . result , Exception ) ) <EOL> def test_callback_skips ( self ) : <EOL> """<STR_LIT>""" <EOL> self . deferred . add_callback ( throw_always ) <EOL> self . deferred . add_callback ( one_always ) <EOL> self . deferred . add_callback ( add , <NUM_LIT:2> ) <EOL> self . deferred . add_errback ( one_always ) <EOL> self . deferred . add_callback ( self . set_result ) <EOL> self . deferred . callback ( None ) <EOL> self . assertTrue ( self . result == <NUM_LIT:1> ) <EOL> def test_errback_reraised ( self ) : <EOL> """<STR_LIT>""" <EOL> self . deferred . add_errback ( throw_always ) <EOL> self . deferred . add_errback ( self . set_result ) <EOL> self . deferred . errback ( Exception ( ) ) <EOL> self . assertTrue ( isinstance ( self . result , Exception ) ) <EOL> def test_cancelled ( self ) : <EOL> self . deferred . cancel ( ) <EOL> self . assertRaises ( CancelledError , self . deferred . errback , Exception ( "<STR_LIT>" ) ) <EOL> self . assertRaises ( CancelledError , self . deferred . callback , None ) <EOL> self . assertRaises ( CancelledError , self . deferred . result ) <EOL> def test_already_called ( self ) : <EOL> self . deferred . callback ( None ) <EOL> self . assertRaises ( AlreadyCalledError , self . deferred . errback , Exception ( "<STR_LIT>" ) ) <EOL> self . assertRaises ( AlreadyCalledError , self . deferred . callback , None ) <EOL> self . assertRaises ( AlreadyCalledError , self . deferred . cancel ) <EOL> def test_cancel_callback ( self ) : <EOL> self . deferred = Deferred ( loop , cancelled_cb = self . set_result ) <EOL> self . deferred . cancel ( ) <EOL> self . assertTrue ( self . result == self . deferred ) <EOL> def test_result_chain ( self ) : <EOL> self . deferred . callback ( <NUM_LIT:5> ) <EOL> self . assertTrue ( self . deferred . result ( ) == <NUM_LIT:5> ) <EOL> self . deferred . add_callback ( add , <NUM_LIT:2> ) <EOL> self . assertTrue ( self . deferred . result ( ) == <NUM_LIT:7> ) <EOL> self . deferred . add_callback ( throw_always ) <EOL> self . assertRaises ( Exception , self . deferred . result ) <EOL> def test_result ( self ) : <EOL> self . deferred . callback ( <NUM_LIT:5> ) <EOL> self . assertTrue ( self . deferred . result ( ) == <NUM_LIT:5> ) <EOL> def test_result_exceptioned ( self ) : <EOL> self . deferred . errback ( Exception ( "<STR_LIT>" ) ) <EOL> self . assertRaises ( Exception , self . deferred . result ) <EOL> def test_delayed_result ( self ) : <EOL> now = time . time ( ) <EOL> t1 = self . call_later ( <NUM_LIT:0.5> , self . deferred . callback , <NUM_LIT:5> ) <EOL> self . assertTrue ( self . deferred . result ( ) == <NUM_LIT:5> ) <EOL> self . assertTrue ( time . time ( ) - now > <NUM_LIT> ) <EOL> def test_delayed_result_chained ( self ) : <EOL> now = time . time ( ) <EOL> t1 = self . call_later ( <NUM_LIT:0.5> , self . deferred . callback , <NUM_LIT:5> ) <EOL> self . deferred . add_callback ( add , <NUM_LIT:4> ) <EOL> self . assertTrue ( self . deferred . result ( ) == <NUM_LIT:9> ) <EOL> self . assertTrue ( time . time ( ) - now > <NUM_LIT> ) <EOL> def test_delayed_result_timeout ( self ) : <EOL> t1 = self . call_later ( <NUM_LIT:0.5> , self . deferred . callback , <NUM_LIT:5> ) <EOL> self . assertRaises ( TimeoutError , self . deferred . result , <NUM_LIT:0.1> ) <EOL> def test_delayed_result_cancelled ( self ) : <EOL> t1 = self . call_later ( <NUM_LIT:0.5> , self . deferred . callback , <NUM_LIT:5> ) <EOL> t2 = self . call_later ( <NUM_LIT> , self . deferred . cancel ) <EOL> self . assertRaises ( CancelledError , self . deferred . result , <NUM_LIT> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
