1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 This module contains tests for L{twisted.internet.task.Cooperator} and
9 from twisted.internet import reactor, defer, task
10 from twisted.trial import unittest
14 class FakeDelayedCall(object):
16 Fake delayed call which lets us simulate the scheduler.
18 def __init__(self, func):
20 A function to run, later.
23 self.cancelled = False
28 Don't run my function later.
34 class FakeScheduler(object):
36 A fake scheduler for testing against.
40 Create a fake scheduler with a list of work to do.
45 def __call__(self, thunk):
47 Schedule a unit of work to be done later.
49 unit = FakeDelayedCall(thunk)
50 self.work.append(unit)
56 Do all of the work that is currently available to be done.
58 work, self.work = self.work, []
60 if not unit.cancelled:
65 class TestCooperator(unittest.TestCase):
68 def ebIter(self, err):
69 err.trap(task.SchedulerStopped)
73 def cbIter(self, ign):
77 def testStoppedRejectsNewTasks(self):
79 Test that Cooperators refuse new tasks when they have been stopped.
84 d = c.coiterate(iter(()), stuff)
85 d.addCallback(self.cbIter)
86 d.addErrback(self.ebIter)
87 return d.addCallback(lambda result:
88 self.assertEqual(result, self.RESULT))
89 return testwith(None).addCallback(lambda ign: testwith(defer.Deferred()))
92 def testStopRunning(self):
94 Test that a running iterator will not run to completion when the
95 cooperator is stopped.
99 for myiter.value in range(3):
102 d = c.coiterate(myiter())
103 d.addCallback(self.cbIter)
104 d.addErrback(self.ebIter)
106 def doasserts(result):
107 self.assertEqual(result, self.RESULT)
108 self.assertEqual(myiter.value, -1)
109 d.addCallback(doasserts)
113 def testStopOutstanding(self):
115 An iterator run with L{Cooperator.coiterate} paused on a L{Deferred}
116 yielded by that iterator will fire its own L{Deferred} (the one
117 returned by C{coiterate}) when L{Cooperator.stop} is called.
119 testControlD = defer.Deferred()
120 outstandingD = defer.Deferred()
122 reactor.callLater(0, testControlD.callback, None)
125 c = task.Cooperator()
126 d = c.coiterate(myiter())
129 outstandingD.callback('arglebargle')
131 testControlD.addCallback(stopAndGo)
132 d.addCallback(self.cbIter)
133 d.addErrback(self.ebIter)
135 return d.addCallback(
136 lambda result: self.assertEqual(result, self.RESULT))
139 def testUnexpectedError(self):
140 c = task.Cooperator()
146 d = c.coiterate(myiter())
147 return self.assertFailure(d, RuntimeError)
150 def testUnexpectedErrorActuallyLater(self):
153 reactor.callLater(0, D.errback, RuntimeError())
156 c = task.Cooperator()
157 d = c.coiterate(myiter())
158 return self.assertFailure(d, RuntimeError)
161 def testUnexpectedErrorNotActuallyLater(self):
163 yield defer.fail(RuntimeError())
165 c = task.Cooperator()
166 d = c.coiterate(myiter())
167 return self.assertFailure(d, RuntimeError)
170 def testCooperation(self):
177 groupsOfThings = ['abc', (1, 2, 3), 'def', (4, 5, 6)]
179 c = task.Cooperator()
181 for stuff in groupsOfThings:
182 tasks.append(c.coiterate(myiter(stuff)))
184 return defer.DeferredList(tasks).addCallback(
185 lambda ign: self.assertEqual(tuple(L), sum(zip(*groupsOfThings), ())))
188 def testResourceExhaustion(self):
202 c = task.Cooperator(terminationPredicateFactory=_TPF)
203 c.coiterate(myiter()).addErrback(self.ebIter)
204 c._delayedCall.cancel()
205 # testing a private method because only the test case will ever care
206 # about this, so we have to carefully clean up after ourselves.
209 self.failUnless(_TPF.stopped)
210 self.assertEqual(output, range(10))
213 def testCallbackReCoiterate(self):
215 If a callback to a deferred returned by coiterate calls coiterate on
216 the same Cooperator, we should make sure to only do the minimal amount
217 of scheduling work. (This test was added to demonstrate a specific bug
218 that was found while writing the scheduler.)
223 def __init__(self, func):
227 return '<FakeCall %r>' % (self.func,)
230 self.failIf(calls, repr(calls))
231 calls.append(FakeCall(f))
234 c = task.Cooperator(scheduler=sched, terminationPredicateFactory=lambda: lambda: True)
235 d = c.coiterate(iter(()))
238 def anotherTask(ign):
239 c.coiterate(iter(())).addBoth(done.append)
241 d.addCallback(anotherTask)
250 self.fail("Cooperator took too long")
253 def test_removingLastTaskStopsScheduledCall(self):
255 If the last task in a Cooperator is removed, the scheduled call for
256 the next tick is cancelled, since it is no longer necessary.
258 This behavior is useful for tests that want to assert they have left
259 no reactor state behind when they're done.
263 calls[0] = FakeDelayedCall(f)
265 coop = task.Cooperator(scheduler=sched)
267 # Add two task; this should schedule the tick:
268 task1 = coop.cooperate(iter([1, 2]))
269 task2 = coop.cooperate(iter([1, 2]))
270 self.assertEqual(calls[0].func, coop._tick)
272 # Remove first task; scheduled call should still be going:
274 self.assertEqual(calls[0].cancelled, False)
275 self.assertEqual(coop._delayedCall, calls[0])
277 # Remove second task; scheduled call should be cancelled:
279 self.assertEqual(calls[0].cancelled, True)
280 self.assertEqual(coop._delayedCall, None)
282 # Add another task; scheduled call will be recreated:
283 task3 = coop.cooperate(iter([1, 2]))
284 self.assertEqual(calls[0].cancelled, False)
285 self.assertEqual(coop._delayedCall, calls[0])
289 class UnhandledException(Exception):
291 An exception that should go unhandled.
296 class AliasTests(unittest.TestCase):
298 Integration test to verify that the global singleton aliases do what
302 def test_cooperate(self):
304 L{twisted.internet.task.cooperate} ought to run the generator that it is
313 theTask = task.cooperate(it)
314 self.assertIn(theTask, task._theCooperator._tasks)
319 class RunStateTests(unittest.TestCase):
321 Tests to verify the behavior of L{CooperativeTask.pause},
322 L{CooperativeTask.resume}, L{CooperativeTask.stop}, exhausting the
323 underlying iterator, and their interactions with each other.
328 Create a cooperator with a fake scheduler and a termination predicate
329 that ensures only one unit of work will take place per tick.
331 self._doDeferNext = False
332 self._doStopNext = False
333 self._doDieNext = False
335 self.scheduler = FakeScheduler()
336 self.cooperator = task.Cooperator(
337 scheduler=self.scheduler,
338 # Always stop after one iteration of work (return a function which
339 # returns a function which always returns True)
340 terminationPredicateFactory=lambda: lambda: True)
341 self.task = self.cooperator.cooperate(self.worker())
342 self.cooperator.start()
347 This is a sample generator which yields Deferreds when we are testing
348 deferral and an ascending integer count otherwise.
353 if self._doDeferNext:
354 self._doDeferNext = False
358 elif self._doStopNext:
360 elif self._doDieNext:
361 raise UnhandledException()
369 Drop references to interesting parts of the fixture to allow Deferred
370 errors to be noticed when things start failing.
378 Defer the next result from my worker iterator.
380 self._doDeferNext = True
385 Make the next result from my worker iterator be completion (raising
388 self._doStopNext = True
393 Make the next result from my worker iterator be raising an
394 L{UnhandledException}.
396 def ignoreUnhandled(failure):
397 failure.trap(UnhandledException)
399 self._doDieNext = True
402 def test_pauseResume(self):
404 Cooperators should stop running their tasks when they're paused, and
405 start again when they're resumed.
407 # first, sanity check
408 self.scheduler.pump()
409 self.assertEqual(self.work, [1])
410 self.scheduler.pump()
411 self.assertEqual(self.work, [1, 2])
415 self.scheduler.pump()
416 self.assertEqual(self.work, [1, 2])
418 # Resuming itself shoult not do any work
419 self.assertEqual(self.work, [1, 2])
420 self.scheduler.pump()
421 # But when the scheduler rolls around again...
422 self.assertEqual(self.work, [1, 2, 3])
425 def test_resumeNotPaused(self):
427 L{CooperativeTask.resume} should raise a L{TaskNotPaused} exception if
428 it was not paused; e.g. if L{CooperativeTask.pause} was not invoked
429 more times than L{CooperativeTask.resume} on that object.
431 self.assertRaises(task.NotPaused, self.task.resume)
434 self.assertRaises(task.NotPaused, self.task.resume)
437 def test_pauseTwice(self):
439 Pauses on tasks should behave like a stack. If a task is paused twice,
440 it needs to be resumed twice.
444 self.scheduler.pump()
445 self.assertEqual(self.work, [])
448 self.scheduler.pump()
449 self.assertEqual(self.work, [])
450 # resume once (it shouldn't)
452 self.scheduler.pump()
453 self.assertEqual(self.work, [])
454 # resume twice (now it should go)
456 self.scheduler.pump()
457 self.assertEqual(self.work, [1])
460 def test_pauseWhileDeferred(self):
462 C{pause()}ing a task while it is waiting on an outstanding
463 L{defer.Deferred} should put the task into a state where the
464 outstanding L{defer.Deferred} must be called back I{and} the task is
465 C{resume}d before it will continue processing.
468 self.scheduler.pump()
469 self.assertEqual(len(self.work), 1)
470 self.failUnless(isinstance(self.work[0], defer.Deferred))
471 self.scheduler.pump()
472 self.assertEqual(len(self.work), 1)
474 self.scheduler.pump()
475 self.assertEqual(len(self.work), 1)
477 self.scheduler.pump()
478 self.assertEqual(len(self.work), 1)
479 self.work[0].callback("STUFF!")
480 self.scheduler.pump()
481 self.assertEqual(len(self.work), 2)
482 self.assertEqual(self.work[1], 2)
485 def test_whenDone(self):
487 L{CooperativeTask.whenDone} returns a Deferred which fires when the
488 Cooperator's iterator is exhausted. It returns a new Deferred each
489 time it is called; callbacks added to other invocations will not modify
490 the value that subsequent invocations will fire with.
493 deferred1 = self.task.whenDone()
494 deferred2 = self.task.whenDone()
500 def callbackOne(result):
501 results1.append(result)
504 def callbackTwo(result):
505 results2.append(result)
508 deferred1.addCallback(callbackOne)
509 deferred2.addCallback(callbackTwo)
511 deferred1.addCallback(final1.append)
512 deferred2.addCallback(final2.append)
514 # exhaust the task iterator
517 self.scheduler.pump()
519 self.assertEqual(len(results1), 1)
520 self.assertEqual(len(results2), 1)
522 self.assertIdentical(results1[0], self.task._iterator)
523 self.assertIdentical(results2[0], self.task._iterator)
525 self.assertEqual(final1, [1])
526 self.assertEqual(final2, [2])
529 def test_whenDoneError(self):
531 L{CooperativeTask.whenDone} returns a L{defer.Deferred} that will fail
532 when the iterable's C{next} method raises an exception, with that
535 deferred1 = self.task.whenDone()
537 deferred1.addErrback(results.append)
539 self.scheduler.pump()
540 self.assertEqual(len(results), 1)
541 self.assertEqual(results[0].check(UnhandledException), UnhandledException)
544 def test_whenDoneStop(self):
546 L{CooperativeTask.whenDone} returns a L{defer.Deferred} that fails with
547 L{TaskStopped} when the C{stop} method is called on that
550 deferred1 = self.task.whenDone()
552 deferred1.addErrback(errors.append)
554 self.assertEqual(len(errors), 1)
555 self.assertEqual(errors[0].check(task.TaskStopped), task.TaskStopped)
558 def test_whenDoneAlreadyDone(self):
560 L{CooperativeTask.whenDone} will return a L{defer.Deferred} that will
561 succeed immediately if its iterator has already completed.
564 self.scheduler.pump()
566 self.task.whenDone().addCallback(results.append)
567 self.assertEqual(results, [self.task._iterator])
570 def test_stopStops(self):
572 C{stop()}ping a task should cause it to be removed from the run just as
573 C{pause()}ing, with the distinction that C{resume()} will raise a
574 L{TaskStopped} exception.
577 self.scheduler.pump()
578 self.assertEqual(len(self.work), 0)
579 self.assertRaises(task.TaskStopped, self.task.stop)
580 self.assertRaises(task.TaskStopped, self.task.pause)
581 # Sanity check - it's still not scheduled, is it?
582 self.scheduler.pump()
583 self.assertEqual(self.work, [])
586 def test_pauseStopResume(self):
588 C{resume()}ing a paused, stopped task should be a no-op; it should not
589 raise an exception, because it's paused, but neither should it actually
590 do more work from the task.
595 self.scheduler.pump()
596 self.assertEqual(self.work, [])
599 def test_stopDeferred(self):
601 As a corrolary of the interaction of C{pause()} and C{unpause()},
602 C{stop()}ping a task which is waiting on a L{Deferred} should cause the
603 task to gracefully shut down, meaning that it should not be unpaused
604 when the deferred fires.
607 self.scheduler.pump()
609 self.assertEqual(self.task._pauseCount, 1)
611 d.addBoth(results.append)
612 self.scheduler.pump()
614 self.scheduler.pump()
616 self.scheduler.pump()
617 # Let's make sure that Deferred doesn't come out fried with an
618 # unhandled error that will be logged. The value is None, rather than
619 # our test value, 7, because this Deferred is returned to and consumed
620 # by the cooperator code. Its callback therefore has no contract.
621 self.assertEqual(results, [None])
622 # But more importantly, no further work should have happened.
623 self.assertEqual(self.work, [])
626 def test_stopExhausted(self):
628 C{stop()}ping a L{CooperativeTask} whose iterator has been exhausted
629 should raise L{TaskDone}.
632 self.scheduler.pump()
633 self.assertRaises(task.TaskDone, self.task.stop)
636 def test_stopErrored(self):
638 C{stop()}ping a L{CooperativeTask} whose iterator has encountered an
639 error should raise L{TaskFailed}.
642 self.scheduler.pump()
643 self.assertRaises(task.TaskFailed, self.task.stop)
646 def test_stopCooperatorReentrancy(self):
648 If a callback of a L{Deferred} from L{CooperativeTask.whenDone} calls
649 C{Cooperator.stop} on its L{CooperativeTask._cooperator}, the
650 L{Cooperator} will stop, but the L{CooperativeTask} whose callback is
651 calling C{stop} should already be considered 'stopped' by the time the
652 callback is running, and therefore removed from the
657 callbackPhases.append(result)
658 self.cooperator.stop()
659 # "done" here is a sanity check to make sure that we get all the
660 # way through the callback; i.e. stop() shouldn't be raising an
661 # exception due to the stopped-ness of our main task.
662 callbackPhases.append("done")
663 self.task.whenDone().addCallback(stopit)
665 self.scheduler.pump()
666 self.assertEqual(callbackPhases, [self.task._iterator, "done"])