test_cooperator.py 文件源码

def testStoppedRejectsNewTasks(self):
        """
        Test that Cooperators refuse new tasks when they have been stopped.
        """
        def testwith(stuff):
            c = task.Cooperator()
            c.stop()
            d = c.coiterate(iter(()), stuff)
            d.addCallback(self.cbIter)
            d.addErrback(self.ebIter)
            return d.addCallback(lambda result:
                                 self.assertEquals(result, self.RESULT))
        return testwith(None).addCallback(lambda ign: testwith(defer.Deferred()))