test_task.py 文件源码

def test_asynchronousStop(self):
        """
        L{task.react} handles when the reactor is stopped and the
        returned L{Deferred} doesn't fire.
        """
        def main(reactor):
            reactor.callLater(1, reactor.stop)
            return defer.Deferred()
        r = _FakeReactor()
        exitError = self.assertRaises(
            SystemExit, task.react, main, [], _reactor=r)
        self.assertEqual(0, exitError.code)