def testStopAndCancelWithOneUnderway(self):
        """
        Start a dispatch queue of width 2, and send it 3 jobs. Verify that
        2 of the jobs are underway. Then stop it before they can complete,
        telling it to cancel the underway jobs. The two jobs that were
        underway should both be cancelled and returned by the stop method.
        The first 2 jobs returned should have state CANCELLED, and the
        final one should still be PENDING.
        """
        def ok(result):
            self.fail('Unexpected success!')

        def checkCancel(failure):
            self.assertEqual(failure.value.state, Job.CANCELLED)

        dq = ResizableDispatchQueue(self.slow, 2)
        dq.put(0).addCallbacks(ok, checkCancel)
        dq.put(1).addCallbacks(ok, checkCancel)
        dq.put(2)
        reactor.callLater(0.01, self._testUnderway, dq, set([0, 1]))
        pendingJobs = yield task.deferLater(
            reactor, 0.1, dq.stop, cancelUnderway=True)
        pendingArgs = [p.jobarg for p in pendingJobs]
        self.assertEqual([0, 1, 2], sorted(pendingArgs))
        self.assertEqual(pendingJobs[0].state, Job.CANCELLED)
        self.assertEqual(pendingJobs[1].state, Job.CANCELLED)
        self.assertEqual(pendingJobs[2].state, Job.PENDING)

 def testZeroOnExplicitCreation(self):
     """
     A new queue whose width is 10 should have no pending stops.
     """
     dq = ResizableDispatchQueue(None, 10)
     self.assertEqual(0, dq.pendingStops)
     return dq.stop()

 def testPauseResumeWithNonZeroWidth(self):
     """
     Check that it is possible to resume with a non-zero width.
     """
     dq = ResizableDispatchQueue(None, 1)
     yield dq.pause()
     dq.resume(10)
     self.assertEqual(dq.width, 10)

 def testZeroOnDefaultCreation(self):
     """
     A new queue whose width is unspecified should have no pending
     stops.
     """
     dq = ResizableDispatchQueue(None)
     self.assertEqual(0, dq.pendingStops)
     return dq.stop()

 def testReprioritizeNonPendingJob(self):
     """
     Reprioritizing a job that is not pending must raise
     C{RuntimeError}.
     """
     dq = ResizableDispatchQueue(lambda x: x, 1)
     job = yield dq.put('a')
     self.assertRaises(RuntimeError, dq.reprioritize, job, 10)

 def testJustOneWithZeroInitialWidth(self):
     # Same as testJustOne (above), except we initialize with zero width
     # and start the dispatcher by explicitly setting its width.
     dq = ResizableDispatchQueue(self.slow)
     map(dq.put, range(3))
     self.assertEqual(0, dq.width)
     dq.width = 1
     return self._stopAndTest(0.0001, dq, [1, 2])

 def testSimplePending(self):
     """
     Putting 3 jobs onto a queue with zero width should result in a
     number of pending jobs of 3.
     """
     dq = ResizableDispatchQueue(None)
     map(dq.put, range(3))
     self.assertEqual((0, 3), dq.size())

 def testPause(self):
     items = range(3)
     dq = ResizableDispatchQueue(self.slow, 1)
     map(dq.put, items)
     reactor.callLater(0.2, self._testPaused, dq)
     yield task.deferLater(reactor, 0.1, dq.pause)
     pendingJobs = dq.pending()
     self.assertEqual([p.jobarg for p in pendingJobs], [1, 2])

 def testPutRaisesWhenStopped(self):
     """
     Putting something onto a stopped queue should raise
     C{QueueStopped}.
     """
     dq = ResizableDispatchQueue(None)
     yield dq.stop()
     self.assertRaises(QueueStopped, dq.put, None)

 def testDeferredFromPutFiresWithCorrectResult(self):
     """
     Make sure the C{Deferred} we get back from a put to the queue fires
     with the expected result.
     """
     dq = ResizableDispatchQueue(lambda x: 2 * x, width=1)
     job = yield dq.put(6)
     self.assertEqual(job.result, 12)

 def testIncrementToVeryHighValue(self):
     """
     Create a queue of width 10 and then set its width to 1000.
     The number of pending stops will be 0.
     """
     dq = ResizableDispatchQueue(None, 10)
     dq.width = 1000
     self.assertEqual(dq.pendingStops, 0)
     return dq.stop()

 def testNarrow(self):
     """
     Create a queue of width 5 and narrow it to width 3. There should
     then be 2 pending stops.
     """
     dq = ResizableDispatchQueue(None, 5)
     dq.width = 3
     self.assertEqual(2, dq.pendingStops)
     return dq.stop()

 def testSetImmediatelyToZero(self):
     """
     Create a queue of width 10 and then set its width to 0.
     The number of pending stops will be 10.
     """
     dq = ResizableDispatchQueue(None, 10)
     dq.width = 0
     self.assertEqual(dq.pendingStops, 10)
     return dq.stop()

 def testPuttingMoreThanWidthOnlyDispatchesWidth(self):
     """
     Make sure that if we put 5 things onto a queue whose width is 3
     that shortly thereafter there are 3 jobs underway and 2 pending.
     """
     dq = ResizableDispatchQueue(self.slow, 3)
     for value in range(5):
         reactor.callLater(0.01, dq.put, value)
     yield task.deferLater(reactor, 0.1, self._testSize, dq, (3, 2))
     remaining = yield dq.stop()

    def testRaiseFailsWithJob(self):
        """
        Raising an exception in the job handler should result in a failure
        that contains a L{txrdq.job.Job} value.
        """
        def raiseException(jobarg):
            raise Exception()

        dq = ResizableDispatchQueue(raiseException, 1)
        d = dq.put('Some argument', 1)
        return self.assertFailure(d, Job)

 def testUnstarted(self):
     """
     Check basic facts about a fresh instance of a
     ResizableDispatchQueue.
     """
     dq = ResizableDispatchQueue(None)
     self.assertEqual(dq.pending(), [])
     self.assertEqual(dq.underway(), set())
     self.assertFalse(dq.stopped)
     self.assertFalse(dq.paused)
     self.assertEqual((0, 0), dq.size())
     return dq.stop()

 def testNarrowWiden(self):
     """
     Create a queue of width 10. Narrowing to 7 means there will be 3
     pending stops, then widening to 8 will decrease the number of
     pending stops to 2.
     """
     dq = ResizableDispatchQueue(None, 10)
     dq.width = 7
     self.assertEqual(dq.pendingStops, 3)
     dq.width = 8
     self.assertEqual(dq.pendingStops, 2)
     return dq.stop()

 def testSetWidthToZeroAfterInitiallyNonZero(self):
     """
     Make sure that a queue whose width is initially non-zero and which
     is then set to zero width does not then begin to process any added
     jobs.
     """
     dq = ResizableDispatchQueue(self.slow, 3)
     yield task.deferLater(reactor, 0.01, dq.setWidth, 0)
     reactor.callLater(0.01, dq.put, 'aaa')
     reactor.callLater(0.01, dq.put, 'bbb')
     yield task.deferLater(reactor, 0.05, self._testSize, dq, (0, 2))
     remaining = yield dq.stop()
     self.assertEqual(['aaa', 'bbb'], [job.jobarg for job in remaining])

    def testCancelRemovesFromQueue(self):
        """
        Cancelling a job should remove it from the queue.
        """
        def eb(failure):
            self.assertEqual(failure.value.state, Job.CANCELLED)

        dq = ResizableDispatchQueue(self.slow, 1)
        dq.put('a', 1)
        deferred = dq.put('b', 2)
        deferred.addErrback(eb)
        deferred.cancel()
        yield task.deferLater(reactor, 0.1, dq.pause)
        pendingJobs = dq.pending()
        self.assertEqual(pendingJobs, [])

    def testRaiseFailsWithCorrectJobArgument(self):
        """
        Raising an exception in the job handler should result in a failure
        that contains a job with the original job argument.
        """
        result = []

        def raiseException(jobarg):
            raise Exception()

        def extractJobarg(failure):
            result.append(failure.value.jobarg)

        dq = ResizableDispatchQueue(raiseException, 1)
        yield dq.put('Some argument', 1).addErrback(extractJobarg)
        self.assertEqual(['Some argument'], result)