test_cooperator.py 文件源码

def setUp(self):
        """
        Create a cooperator with a fake scheduler and a termination predicate
        that ensures only one unit of work will take place per tick.
        """
        self._doDeferNext = False
        self._doStopNext = False
        self._doDieNext = False
        self.work = []
        self.scheduler = FakeScheduler()
        self.cooperator = task.Cooperator(
            scheduler=self.scheduler,
            # Always stop after one iteration of work (return a function which
            # returns a function which always returns True)
            terminationPredicateFactory=lambda: lambda: True)
        self.task = self.cooperator.cooperate(self.worker())
        self.cooperator.start()