240
248
times.append(clock.seconds())
250
# Start a LoopingCall that tracks the time passed, with a 0.5 second
242
252
call = task.LoopingCall(aCallback)
243
253
call.clock = clock
256
# Initially, no time should have passed!
246
257
self.assertEqual(times, [0])
259
# Advance the clock by 2 seconds (2 seconds should have passed)
249
261
self.assertEqual(times, [0, 2])
263
# Advance the clock by 1 second (3 total should have passed)
252
265
self.assertEqual(times, [0, 2, 3])
267
# Advance the clock by 0 seconds (this should have no effect!)
255
269
self.assertEqual(times, [0, 2, 3])
272
def test_reactorTimeCountSkips(self):
274
When L{LoopingCall} schedules itself to run again, if more than the
275
specified interval has passed, it should schedule the next call for the
276
next interval which is still in the future. If it was created
277
using L{LoopingCall.withCount}, a positional argument will be
278
inserted at the beginning of the argument list, indicating the number
279
of calls that should have been made.
283
def aCallback(numCalls):
284
times.append((clock.seconds(), numCalls))
286
# Start a LoopingCall that tracks the time passed, and the number of
287
# skips, with a 0.5 second increment.
288
call = task.LoopingCall.withCount(aCallback)
291
REALISTIC_DELAY = 0.01
294
# Initially, no seconds should have passed, and one calls should have
296
self.assertEqual(times, [(0, 1)])
298
# After the interval (plus a small delay, to account for the time that
299
# the reactor takes to wake up and process the LoopingCall), we should
300
# still have only made one call.
301
clock.advance(INTERVAL + REALISTIC_DELAY)
302
self.assertEqual(times, [(0, 1), (INTERVAL + REALISTIC_DELAY, 1)])
304
# After advancing the clock by three intervals (plus a small delay to
305
# account for the reactor), we should have skipped two calls; one less
306
# than the number of intervals which have completely elapsed. Along
307
# with the call we did actually make, the final number of calls is 3.
308
clock.advance((3 * INTERVAL) + REALISTIC_DELAY)
309
self.assertEqual(times,
310
[(0, 1), (INTERVAL + REALISTIC_DELAY, 1),
311
((4 * INTERVAL) + (2 * REALISTIC_DELAY), 3)])
313
# Advancing the clock by 0 seconds should not cause any changes!
315
self.assertEqual(times,
316
[(0, 1), (INTERVAL + REALISTIC_DELAY, 1),
317
((4 * INTERVAL) + (2 * REALISTIC_DELAY), 3)])
320
def test_countLengthyIntervalCounts(self):
322
L{LoopingCall.withCount} counts only calls that were expected to be
323
made. So, if more than one, but less than two intervals pass between
324
invocations, it won't increase the count above 1. For example, a
325
L{LoopingCall} with interval T expects to be invoked at T, 2T, 3T, etc.
326
However, the reactor takes some time to get around to calling it, so in
327
practice it will be called at T+something, 2T+something, 3T+something;
328
and due to other things going on in the reactor, "something" is
329
variable. It won't increase the count unless "something" is greater
330
than T. So if the L{LoopingCall} is invoked at T, 2.75T, and 3T,
331
the count has not increased, even though the distance between
332
invocation 1 and invocation 2 is 1.75T.
336
def aCallback(count):
337
times.append((clock.seconds(), count))
339
# Start a LoopingCall that tracks the time passed, and the number of
340
# calls, with a 0.5 second increment.
341
call = task.LoopingCall.withCount(aCallback)
344
REALISTIC_DELAY = 0.01
346
self.assertEqual(times.pop(), (0, 1))
348
# About one interval... So far, so good
349
clock.advance(INTERVAL + REALISTIC_DELAY)
350
self.assertEqual(times.pop(), (INTERVAL + REALISTIC_DELAY, 1))
352
# Oh no, something delayed us for a while.
353
clock.advance(INTERVAL * 1.75)
354
self.assertEqual(times.pop(), ((2.75 * INTERVAL) + REALISTIC_DELAY, 1))
356
# Back on track! We got invoked when we expected this time.
357
clock.advance(INTERVAL * 0.25)
358
self.assertEqual(times.pop(), ((3.0 * INTERVAL) + REALISTIC_DELAY, 1))
258
361
def testBasicFunction(self):
259
362
# Arrange to have time advanced enough so that our function is
260
363
# called a few times.
545
def test_deferredWithCount(self):
547
In the case that the function passed to L{LoopingCall.withCount}
548
returns a deferred, which does not fire before the next interval
549
elapses, the function should not be run again. And if a function call
550
is skipped in this fashion, the appropriate count should be
553
testClock = task.Clock()
557
def countTracker(possibleCount):
558
# Keep a list of call counts
559
deferredCounts.append(possibleCount)
560
# Return a deferred, but only on the first request
561
if len(deferredCounts) == 1:
566
# Start a looping call for our countTracker function
567
# Set the increment to 0.2, and do not call the function on startup.
568
lc = task.LoopingCall.withCount(countTracker)
570
d = lc.start(0.2, now=False)
572
# Confirm that nothing has happened yet.
573
self.assertEquals(deferredCounts, [])
575
# Advance the clock by 0.2 and then 0.4;
576
testClock.pump([0.2, 0.4])
577
# We should now have exactly one count (of 1 call)
578
self.assertEquals(len(deferredCounts), 1)
580
# Fire the deferred, and advance the clock by another 0.2
582
testClock.pump([0.2])
583
# We should now have exactly 2 counts...
584
self.assertEquals(len(deferredCounts), 2)
585
# The first count should be 1 (one call)
586
# The second count should be 3 (calls were missed at about 0.6 and 0.8)
587
self.assertEquals(deferredCounts, [1, 3])
443
591
class DeferLaterTests(unittest.TestCase):
added
removed
twisted/test/test_task.py
