1
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
2
# See LICENSE for details.
4
from twisted.trial import unittest
6
from twisted.internet import interfaces, task, reactor, defer, error
8
# Be compatible with any jerks who used our private stuff
11
from twisted.python import failure
13
class TestableLoopingCall(task.LoopingCall):
14
def __init__(self, clock, *a, **kw):
15
super(TestableLoopingCall, self).__init__(*a, **kw)
16
self._callLater = lambda delay: clock.callLater(delay, self)
17
self._seconds = clock.seconds
21
class TestException(Exception):
26
class ClockTestCase(unittest.TestCase):
28
Test the non-wallclock based clock implementation.
30
def testSeconds(self):
32
Test that the L{seconds} method of the fake clock returns fake time.
35
self.assertEquals(c.seconds(), 0)
38
def testCallLater(self):
40
Test that calls can be scheduled for later with the fake clock and
41
hands back an L{IDelayedCall}.
44
call = c.callLater(1, lambda a, b: None, 1, b=2)
45
self.failUnless(interfaces.IDelayedCall.providedBy(call))
46
self.assertEquals(call.getTime(), 1)
47
self.failUnless(call.active())
50
def testCallLaterCancelled(self):
52
Test that calls can be cancelled.
55
call = c.callLater(1, lambda a, b: None, 1, b=2)
57
self.failIf(call.active())
60
def testAdvance(self):
62
Test that advancing the clock will fire some calls.
66
call = c.callLater(2, lambda: events.append(None))
68
self.assertEquals(events, [])
70
self.assertEquals(events, [None])
71
self.failIf(call.active())
74
def testAdvanceCancel(self):
76
Test attemping to cancel the call in a callback.
78
AlreadyCalled should be raised, not for example a ValueError from
79
removing the call from Clock.calls. This requires call.called to be
80
set before the callback is called.
84
self.assertRaises(error.AlreadyCalled, call.cancel)
85
call = c.callLater(1, cb)
89
def testCallLaterDelayed(self):
91
Test that calls can be delayed.
95
call = c.callLater(1, lambda a, b: events.append((a, b)), 1, b=2)
97
self.assertEquals(call.getTime(), 2)
99
self.assertEquals(events, [])
101
self.assertEquals(events, [(1, 2)])
104
def testCallLaterResetLater(self):
106
Test that calls can have their time reset to a later time.
110
call = c.callLater(2, lambda a, b: events.append((a, b)), 1, b=2)
113
self.assertEquals(call.getTime(), 4)
115
self.assertEquals(events, [])
117
self.assertEquals(events, [(1, 2)])
120
def testCallLaterResetSooner(self):
122
Test that calls can have their time reset to an earlier time.
126
call = c.callLater(4, lambda a, b: events.append((a, b)), 1, b=2)
128
self.assertEquals(call.getTime(), 3)
130
self.assertEquals(events, [(1, 2)])
134
class LoopTestCase(unittest.TestCase):
135
def testBasicFunction(self):
136
# Arrange to have time advanced enough so that our function is
137
# called a few times.
138
# Only need to go to 2.5 to get 3 calls, since the first call
139
# happens before any time has elapsed.
140
timings = [0.05, 0.1, 0.1]
145
def foo(a, b, c=None, d=None):
146
L.append((a, b, c, d))
148
lc = TestableLoopingCall(clock, foo, "a", "b", d="d")
152
def saveResult(result):
153
theResult.append(result)
154
D.addCallback(saveResult)
158
self.assertEquals(len(L), 3,
159
"got %d iterations, not 3" % (len(L),))
161
for (a, b, c, d) in L:
162
self.assertEquals(a, "a")
163
self.assertEquals(b, "b")
164
self.assertEquals(c, None)
165
self.assertEquals(d, "d")
168
self.assertIdentical(theResult[0], lc)
170
# Make sure it isn't planning to do anything further.
171
self.failIf(clock.calls)
174
def testDelayedStart(self):
175
timings = [0.05, 0.1, 0.1]
180
lc = TestableLoopingCall(clock, L.append, None)
181
d = lc.start(0.1, now=False)
184
def saveResult(result):
185
theResult.append(result)
186
d.addCallback(saveResult)
190
self.assertEquals(len(L), 2,
191
"got %d iterations, not 2" % (len(L),))
193
self.assertIdentical(theResult[0], lc)
195
self.failIf(clock.calls)
198
def testBadDelay(self):
199
lc = task.LoopingCall(lambda: None)
200
self.assertRaises(ValueError, lc.start, -1)
203
# Make sure that LoopingCall.stop() prevents any subsequent calls.
204
def _stoppingTest(self, delay):
210
lc = TestableLoopingCall(clock, foo)
211
d = lc.start(delay, now=False)
214
self.failIf(clock.calls)
217
def testStopAtOnce(self):
218
return self._stoppingTest(0)
221
def testStoppingBeforeDelayedStart(self):
222
return self._stoppingTest(10)
226
class ReactorLoopTestCase(unittest.TestCase):
227
# Slightly inferior tests which exercise interactions with an actual
229
def testFailure(self):
231
raise TestException(x)
233
lc = task.LoopingCall(foo, "bar")
234
return self.assertFailure(lc.start(0.1), TestException)
237
def testFailAndStop(self):
240
raise TestException(x)
242
lc = task.LoopingCall(foo, "bar")
243
return self.assertFailure(lc.start(0.1), TestException)
246
def testEveryIteration(self):
254
lc = task.LoopingCall(foo)
257
self.assertEquals(len(ran), 6)
258
return d.addCallback(stopped)
261
def testStopAtOnceLater(self):
262
# Ensure that even when LoopingCall.stop() is called from a
263
# reactor callback, it still prevents any subsequent calls.
266
d.errback(failure.DefaultException(
267
"This task also should never get called."))
268
self._lc = task.LoopingCall(foo)
269
self._lc.start(1, now=False)
270
reactor.callLater(0, self._callback_for_testStopAtOnceLater, d)
274
def _callback_for_testStopAtOnceLater(self, d):
276
reactor.callLater(0, d.callback, "success")
278
def testWaitDeferred(self):
279
# Tests if the callable isn't scheduled again before the returned
280
# deferred has fired.
286
d.addCallback(lambda _: lc.stop())
287
clock.callLater(1, d.callback, None)
290
lc = TestableLoopingCall(clock, foo)
293
self.failIf(clock.calls)
295
def testFailurePropagation(self):
296
# Tests if the failure of the errback of the deferred returned by the
297
# callable is propagated to the lc errback.
299
# To make sure this test does not hang trial when LoopingCall does not
300
# wait for the callable's deferred, it also checks there are no
301
# calls in the clock's callLater queue.
307
clock.callLater(0.3, d.errback, TestException())
310
lc = TestableLoopingCall(clock, foo)
312
self.assertFailure(d, TestException)
315
self.failIf(clock.calls)
added
removed
src/twisted/test/test_task.py
