test_cooperator.py 文件源码

def testStopRunning(self):
        """
        Test that a running iterator will not run to completion when the
        cooperator is stopped.
        """
        c = task.Cooperator()
        def myiter():
            for myiter.value in range(3):
                yield myiter.value
        myiter.value = -1
        d = c.coiterate(myiter())
        d.addCallback(self.cbIter)
        d.addErrback(self.ebIter)
        c.stop()
        def doasserts(result):
            self.assertEquals(result, self.RESULT)
            self.assertEquals(myiter.value, -1)
        d.addCallback(doasserts)
        return d