<s> from twisted . python . compat import set <EOL> from twisted . trial import unittest <EOL> from twisted . internet import interfaces , task , reactor , defer , error <EOL> Clock = task . Clock <EOL> from twisted . python import failure <EOL> class TestableLoopingCall ( task . LoopingCall ) : <EOL> def __init__ ( self , clock , * a , ** kw ) : <EOL> super ( TestableLoopingCall , self ) . __init__ ( * a , ** kw ) <EOL> self . clock = clock <EOL> class TestException ( Exception ) : <EOL> pass <EOL> class ClockTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testSeconds ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> self . assertEqual ( c . seconds ( ) , <NUM_LIT:0> ) <EOL> def testCallLater ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:1> , lambda a , b : None , <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> self . failUnless ( interfaces . IDelayedCall . providedBy ( call ) ) <EOL> self . assertEqual ( call . getTime ( ) , <NUM_LIT:1> ) <EOL> self . failUnless ( call . active ( ) ) <EOL> def testCallLaterCancelled ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:1> , lambda a , b : None , <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> call . cancel ( ) <EOL> self . failIf ( call . active ( ) ) <EOL> def test_callLaterOrdering ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> call1 = c . callLater ( <NUM_LIT:10> , lambda a , b : None , <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> call2 = c . callLater ( <NUM_LIT:1> , lambda a , b : None , <NUM_LIT:3> , b = <NUM_LIT:4> ) <EOL> self . failIf ( call1 is call2 ) <EOL> def testAdvance ( self ) : <EOL> """<STR_LIT>""" <EOL> events = [ ] <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:2> , lambda : events . append ( None ) ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( events , [ ] ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( events , [ None ] ) <EOL> self . failIf ( call . active ( ) ) <EOL> def testAdvanceCancel ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> def cb ( ) : <EOL> self . assertRaises ( error . AlreadyCalled , call . cancel ) <EOL> call = c . callLater ( <NUM_LIT:1> , cb ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> def testCallLaterDelayed ( self ) : <EOL> """<STR_LIT>""" <EOL> events = [ ] <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:1> , lambda a , b : events . append ( ( a , b ) ) , <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> call . delay ( <NUM_LIT:1> ) <EOL> self . assertEqual ( call . getTime ( ) , <NUM_LIT:2> ) <EOL> c . advance ( <NUM_LIT> ) <EOL> self . assertEqual ( events , [ ] ) <EOL> c . advance ( <NUM_LIT:1.0> ) <EOL> self . assertEqual ( events , [ ( <NUM_LIT:1> , <NUM_LIT:2> ) ] ) <EOL> def testCallLaterResetLater ( self ) : <EOL> """<STR_LIT>""" <EOL> events = [ ] <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:2> , lambda a , b : events . append ( ( a , b ) ) , <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> call . reset ( <NUM_LIT:3> ) <EOL> self . assertEqual ( call . getTime ( ) , <NUM_LIT:4> ) <EOL> c . advance ( <NUM_LIT:2> ) <EOL> self . assertEqual ( events , [ ] ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( events , [ ( <NUM_LIT:1> , <NUM_LIT:2> ) ] ) <EOL> def testCallLaterResetSooner ( self ) : <EOL> """<STR_LIT>""" <EOL> events = [ ] <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:4> , lambda a , b : events . append ( ( a , b ) ) , <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> call . reset ( <NUM_LIT:3> ) <EOL> self . assertEqual ( call . getTime ( ) , <NUM_LIT:3> ) <EOL> c . advance ( <NUM_LIT:3> ) <EOL> self . assertEqual ( events , [ ( <NUM_LIT:1> , <NUM_LIT:2> ) ] ) <EOL> def test_getDelayedCalls ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> call = c . callLater ( <NUM_LIT:1> , lambda x : None ) <EOL> call2 = c . callLater ( <NUM_LIT:2> , lambda x : None ) <EOL> calls = c . getDelayedCalls ( ) <EOL> self . assertEqual ( set ( [ call , call2 ] ) , set ( calls ) ) <EOL> def test_getDelayedCallsEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> c = task . Clock ( ) <EOL> self . assertEqual ( c . getDelayedCalls ( ) , [ ] ) <EOL> def test_providesIReactorTime ( self ) : <EOL> c = task . Clock ( ) <EOL> self . failUnless ( interfaces . IReactorTime . providedBy ( c ) , <EOL> "<STR_LIT>" ) <EOL> def test_callLaterKeepsCallsOrdered ( self ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> expected = [ ( '<STR_LIT:b>' , <NUM_LIT> ) , ( '<STR_LIT:a>' , <NUM_LIT> ) ] <EOL> clock = task . Clock ( ) <EOL> logtime = lambda n : result . append ( ( n , clock . seconds ( ) ) ) <EOL> call_a = clock . callLater ( <NUM_LIT:1.0> , logtime , "<STR_LIT:a>" ) <EOL> call_a . reset ( <NUM_LIT> ) <EOL> clock . callLater ( <NUM_LIT> , logtime , "<STR_LIT:b>" ) <EOL> clock . pump ( [ <NUM_LIT:1> ] * <NUM_LIT:3> ) <EOL> self . assertEqual ( result , expected ) <EOL> def test_callLaterResetKeepsCallsOrdered ( self ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> expected = [ ( '<STR_LIT:b>' , <NUM_LIT> ) , ( '<STR_LIT:a>' , <NUM_LIT> ) ] <EOL> clock = task . Clock ( ) <EOL> logtime = lambda n : result . append ( ( n , clock . seconds ( ) ) ) <EOL> call_a = clock . callLater ( <NUM_LIT:1.0> , logtime , "<STR_LIT:a>" ) <EOL> clock . callLater ( <NUM_LIT> , logtime , "<STR_LIT:b>" ) <EOL> call_a . reset ( <NUM_LIT> ) <EOL> clock . pump ( [ <NUM_LIT:1> ] * <NUM_LIT:3> ) <EOL> self . assertEqual ( result , expected ) <EOL> def test_callLaterResetInsideCallKeepsCallsOrdered ( self ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> expected = [ ( '<STR_LIT:c>' , <NUM_LIT> ) , ( '<STR_LIT:b>' , <NUM_LIT> ) ] <EOL> clock = task . Clock ( ) <EOL> logtime = lambda n : result . append ( ( n , clock . seconds ( ) ) ) <EOL> call_b = clock . callLater ( <NUM_LIT> , logtime , "<STR_LIT:b>" ) <EOL> def a ( ) : <EOL> call_b . reset ( <NUM_LIT> ) <EOL> clock . callLater ( <NUM_LIT:1.0> , a ) <EOL> clock . callLater ( <NUM_LIT> , logtime , "<STR_LIT:c>" ) <EOL> clock . pump ( [ <NUM_LIT:0.5> ] * <NUM_LIT:10> ) <EOL> self . assertEqual ( result , expected ) <EOL> class LoopTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_defaultClock ( self ) : <EOL> """<STR_LIT>""" <EOL> call = task . LoopingCall ( lambda : None ) <EOL> self . assertEqual ( call . clock , reactor ) <EOL> def test_callbackTimeSkips ( self ) : <EOL> """<STR_LIT>""" <EOL> times = [ ] <EOL> callDuration = None <EOL> clock = task . Clock ( ) <EOL> def aCallback ( ) : <EOL> times . append ( clock . seconds ( ) ) <EOL> clock . advance ( callDuration ) <EOL> call = task . LoopingCall ( aCallback ) <EOL> call . clock = clock <EOL> callDuration = <NUM_LIT:2> <EOL> call . start ( <NUM_LIT:0.5> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( clock . seconds ( ) , callDuration ) <EOL> clock . advance ( <NUM_LIT:0> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> ] ) <EOL> callDuration = <NUM_LIT:1> <EOL> clock . advance ( <NUM_LIT:0.5> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> , <NUM_LIT> ] ) <EOL> self . assertEqual ( clock . seconds ( ) , <NUM_LIT> ) <EOL> clock . advance ( <NUM_LIT:0> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> , <NUM_LIT> ] ) <EOL> callDuration = <NUM_LIT:0> <EOL> clock . advance ( <NUM_LIT:0.5> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:4> ] ) <EOL> self . assertEqual ( clock . seconds ( ) , <NUM_LIT:4> ) <EOL> def test_reactorTimeSkips ( self ) : <EOL> """<STR_LIT>""" <EOL> times = [ ] <EOL> clock = task . Clock ( ) <EOL> def aCallback ( ) : <EOL> times . append ( clock . seconds ( ) ) <EOL> call = task . LoopingCall ( aCallback ) <EOL> call . clock = clock <EOL> call . start ( <NUM_LIT:0.5> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> ] ) <EOL> clock . advance ( <NUM_LIT:2> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> , <NUM_LIT:2> ] ) <EOL> clock . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> clock . advance ( <NUM_LIT:0> ) <EOL> self . assertEqual ( times , [ <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> def test_reactorTimeCountSkips ( self ) : <EOL> """<STR_LIT>""" <EOL> times = [ ] <EOL> clock = task . Clock ( ) <EOL> def aCallback ( numCalls ) : <EOL> times . append ( ( clock . seconds ( ) , numCalls ) ) <EOL> call = task . LoopingCall . withCount ( aCallback ) <EOL> call . clock = clock <EOL> INTERVAL = <NUM_LIT:0.5> <EOL> REALISTIC_DELAY = <NUM_LIT> <EOL> call . start ( INTERVAL ) <EOL> self . assertEqual ( times , [ ( <NUM_LIT:0> , <NUM_LIT:1> ) ] ) <EOL> clock . advance ( INTERVAL + REALISTIC_DELAY ) <EOL> self . assertEqual ( times , [ ( <NUM_LIT:0> , <NUM_LIT:1> ) , ( INTERVAL + REALISTIC_DELAY , <NUM_LIT:1> ) ] ) <EOL> clock . advance ( ( <NUM_LIT:3> * INTERVAL ) + REALISTIC_DELAY ) <EOL> self . assertEqual ( times , <EOL> [ ( <NUM_LIT:0> , <NUM_LIT:1> ) , ( INTERVAL + REALISTIC_DELAY , <NUM_LIT:1> ) , <EOL> ( ( <NUM_LIT:4> * INTERVAL ) + ( <NUM_LIT:2> * REALISTIC_DELAY ) , <NUM_LIT:3> ) ] ) <EOL> clock . advance ( <NUM_LIT:0> ) <EOL> self . assertEqual ( times , <EOL> [ ( <NUM_LIT:0> , <NUM_LIT:1> ) , ( INTERVAL + REALISTIC_DELAY , <NUM_LIT:1> ) , <EOL> ( ( <NUM_LIT:4> * INTERVAL ) + ( <NUM_LIT:2> * REALISTIC_DELAY ) , <NUM_LIT:3> ) ] ) <EOL> def test_countLengthyIntervalCounts ( self ) : <EOL> """<STR_LIT>""" <EOL> times = [ ] <EOL> clock = task . Clock ( ) <EOL> def aCallback ( count ) : <EOL> times . append ( ( clock . seconds ( ) , count ) ) <EOL> call = task . LoopingCall . withCount ( aCallback ) <EOL> call . clock = clock <EOL> INTERVAL = <NUM_LIT:0.5> <EOL> REALISTIC_DELAY = <NUM_LIT> <EOL> call . start ( INTERVAL ) <EOL> self . assertEqual ( times . pop ( ) , ( <NUM_LIT:0> , <NUM_LIT:1> ) ) <EOL> clock . advance ( INTERVAL + REALISTIC_DELAY ) <EOL> self . assertEqual ( times . pop ( ) , ( INTERVAL + REALISTIC_DELAY , <NUM_LIT:1> ) ) <EOL> clock . advance ( INTERVAL * <NUM_LIT> ) <EOL> self . assertEqual ( times . pop ( ) , ( ( <NUM_LIT> * INTERVAL ) + REALISTIC_DELAY , <NUM_LIT:1> ) ) <EOL> clock . advance ( INTERVAL * <NUM_LIT> ) <EOL> self . assertEqual ( times . pop ( ) , ( ( <NUM_LIT> * INTERVAL ) + REALISTIC_DELAY , <NUM_LIT:1> ) ) <EOL> def testBasicFunction ( self ) : <EOL> timings = [ <NUM_LIT> , <NUM_LIT:0.1> , <NUM_LIT:0.1> ] <EOL> clock = task . Clock ( ) <EOL> L = [ ] <EOL> def foo ( a , b , c = None , d = None ) : <EOL> L . append ( ( a , b , c , d ) ) <EOL> lc = TestableLoopingCall ( clock , foo , "<STR_LIT:a>" , "<STR_LIT:b>" , d = "<STR_LIT:d>" ) <EOL> D = lc . start ( <NUM_LIT:0.1> ) <EOL> theResult = [ ] <EOL> def saveResult ( result ) : <EOL> theResult . append ( result ) <EOL> D . addCallback ( saveResult ) <EOL> clock . pump ( timings ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:3> , <EOL> "<STR_LIT>" % ( len ( L ) , ) ) <EOL> for ( a , b , c , d ) in L : <EOL> self . assertEqual ( a , "<STR_LIT:a>" ) <EOL> self . assertEqual ( b , "<STR_LIT:b>" ) <EOL> self . assertEqual ( c , None ) <EOL> self . assertEqual ( d , "<STR_LIT:d>" ) <EOL> lc . stop ( ) <EOL> self . assertIdentical ( theResult [ <NUM_LIT:0> ] , lc ) <EOL> self . failIf ( clock . calls ) <EOL> def testDelayedStart ( self ) : <EOL> timings = [ <NUM_LIT> , <NUM_LIT:0.1> , <NUM_LIT:0.1> ] <EOL> clock = task . Clock ( ) <EOL> L = [ ] <EOL> lc = TestableLoopingCall ( clock , L . append , None ) <EOL> d = lc . start ( <NUM_LIT:0.1> , now = False ) <EOL> theResult = [ ] <EOL> def saveResult ( result ) : <EOL> theResult . append ( result ) <EOL> d . addCallback ( saveResult ) <EOL> clock . pump ( timings ) <EOL> self . assertEqual ( len ( L ) , <NUM_LIT:2> , <EOL> "<STR_LIT>" % ( len ( L ) , ) ) <EOL> lc . stop ( ) <EOL> self . assertIdentical ( theResult [ <NUM_LIT:0> ] , lc ) <EOL> self . failIf ( clock . calls ) <EOL> def testBadDelay ( self ) : <EOL> lc = task . LoopingCall ( lambda : None ) <EOL> self . assertRaises ( ValueError , lc . start , - <NUM_LIT:1> ) <EOL> def _stoppingTest ( self , delay ) : <EOL> ran = [ ] <EOL> def foo ( ) : <EOL> ran . append ( None ) <EOL> clock = task . Clock ( ) <EOL> lc = TestableLoopingCall ( clock , foo ) <EOL> lc . start ( delay , now = False ) <EOL> lc . stop ( ) <EOL> self . failIf ( ran ) <EOL> self . failIf ( clock . calls ) <EOL> def testStopAtOnce ( self ) : <EOL> return self . _stoppingTest ( <NUM_LIT:0> ) <EOL> def testStoppingBeforeDelayedStart ( self ) : <EOL> return self . _stoppingTest ( <NUM_LIT:10> ) <EOL> def test_reset ( self ) : <EOL> """<STR_LIT>""" <EOL> ran = [ ] <EOL> def foo ( ) : <EOL> ran . append ( None ) <EOL> c = task . Clock ( ) <EOL> lc = TestableLoopingCall ( c , foo ) <EOL> lc . start ( <NUM_LIT:2> , now = False ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> lc . reset ( ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( ran , [ ] ) <EOL> c . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( ran , [ None ] ) <EOL> class ReactorLoopTestCase ( unittest . TestCase ) : <EOL> def testFailure ( self ) : <EOL> def foo ( x ) : <EOL> raise TestException ( x ) <EOL> lc = task . LoopingCall ( foo , "<STR_LIT:bar>" ) <EOL> return self . assertFailure ( lc . start ( <NUM_LIT:0.1> ) , TestException ) <EOL> def testFailAndStop ( self ) : <EOL> def foo ( x ) : <EOL> lc . stop ( ) <EOL> raise TestException ( x ) <EOL> lc = task . LoopingCall ( foo , "<STR_LIT:bar>" ) <EOL> return self . assertFailure ( lc . start ( <NUM_LIT:0.1> ) , TestException ) <EOL> def testEveryIteration ( self ) : <EOL> ran = [ ] <EOL> def foo ( ) : <EOL> ran . append ( None ) <EOL> if len ( ran ) > <NUM_LIT:5> : <EOL> lc . stop ( ) <EOL> lc = task . LoopingCall ( foo ) <EOL> d = lc . start ( <NUM_LIT:0> ) <EOL> def stopped ( ign ) : <EOL> self . assertEqual ( len ( ran ) , <NUM_LIT:6> ) <EOL> return d . addCallback ( stopped ) <EOL> def testStopAtOnceLater ( self ) : <EOL> d = defer . Deferred ( ) <EOL> def foo ( ) : <EOL> d . errback ( failure . DefaultException ( <EOL> "<STR_LIT>" ) ) <EOL> self . _lc = task . LoopingCall ( foo ) <EOL> self . _lc . start ( <NUM_LIT:1> , now = False ) <EOL> reactor . callLater ( <NUM_LIT:0> , self . _callback_for_testStopAtOnceLater , d ) <EOL> return d <EOL> def _callback_for_testStopAtOnceLater ( self , d ) : <EOL> self . _lc . stop ( ) <EOL> reactor . callLater ( <NUM_LIT:0> , d . callback , "<STR_LIT:success>" ) <EOL> def testWaitDeferred ( self ) : <EOL> timings = [ <NUM_LIT> , <NUM_LIT> ] <EOL> clock = task . Clock ( ) <EOL> def foo ( ) : <EOL> d = defer . Deferred ( ) <EOL> d . addCallback ( lambda _ : lc . stop ( ) ) <EOL> clock . callLater ( <NUM_LIT:1> , d . callback , None ) <EOL> return d <EOL> lc = TestableLoopingCall ( clock , foo ) <EOL> lc . start ( <NUM_LIT> ) <EOL> clock . pump ( timings ) <EOL> self . failIf ( clock . calls ) <EOL> def testFailurePropagation ( self ) : <EOL> timings = [ <NUM_LIT> ] <EOL> clock = task . Clock ( ) <EOL> def foo ( ) : <EOL> d = defer . Deferred ( ) <EOL> clock . callLater ( <NUM_LIT> , d . errback , TestException ( ) ) <EOL> return d <EOL> lc = TestableLoopingCall ( clock , foo ) <EOL> d = lc . start ( <NUM_LIT:1> ) <EOL> self . assertFailure ( d , TestException ) <EOL> clock . pump ( timings ) <EOL> self . failIf ( clock . calls ) <EOL> return d <EOL> def test_deferredWithCount ( self ) : <EOL> """<STR_LIT>""" <EOL> testClock = task . Clock ( ) <EOL> d = defer . Deferred ( ) <EOL> deferredCounts = [ ] <EOL> def countTracker ( possibleCount ) : <EOL> deferredCounts . append ( possibleCount ) <EOL> if len ( deferredCounts ) == <NUM_LIT:1> : <EOL> return d <EOL> else : <EOL> return None <EOL> lc = task . LoopingCall . withCount ( countTracker ) <EOL> lc . clock = testClock <EOL> d = lc . start ( <NUM_LIT> , now = False ) <EOL> self . assertEqual ( deferredCounts , [ ] ) <EOL> testClock . pump ( [ <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertEqual ( len ( deferredCounts ) , <NUM_LIT:1> ) <EOL> d . callback ( None ) <EOL> testClock . pump ( [ <NUM_LIT> ] ) <EOL> self . assertEqual ( len ( deferredCounts ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( deferredCounts , [ <NUM_LIT:1> , <NUM_LIT:3> ] ) <EOL> class DeferLaterTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_callback ( self ) : <EOL> """<STR_LIT>""" <EOL> results = [ ] <EOL> flag = object ( ) <EOL> def callable ( foo , bar ) : <EOL> results . append ( ( foo , bar ) ) <EOL> return flag <EOL> clock = task . Clock ( ) <EOL> d = task . deferLater ( clock , <NUM_LIT:3> , callable , '<STR_LIT:foo>' , bar = '<STR_LIT:bar>' ) <EOL> d . addCallback ( self . assertIdentical , flag ) <EOL> clock . advance ( <NUM_LIT:2> ) <EOL> self . assertEqual ( results , [ ] ) <EOL> clock . advance ( <NUM_LIT:1> ) <EOL> self . assertEqual ( results , [ ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ] ) <EOL> return d <EOL> def test_errback ( self ) : <EOL> """<STR_LIT>""" <EOL> def callable ( ) : <EOL> raise TestException ( ) <EOL> clock = task . Clock ( ) <EOL> d = task . deferLater ( clock , <NUM_LIT:1> , callable ) <EOL> clock . advance ( <NUM_LIT:1> ) <EOL> return self . assertFailure ( d , TestException ) <EOL> def test_cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> called = [ ] <EOL> clock = task . Clock ( ) <EOL> d = task . deferLater ( clock , <NUM_LIT:1> , called . append , None ) <EOL> d . cancel ( ) <EOL> def cbCancelled ( ignored ) : <EOL> self . assertEqual ( [ ] , clock . getDelayedCalls ( ) ) <EOL> self . assertFalse ( called ) <EOL> self . assertFailure ( d , defer . CancelledError ) <EOL> d . addCallback ( cbCancelled ) <EOL> return d </s>
