test_defer.py 文件源码

def testSemaphore(self):
        N = 13
        sem = defer.DeferredSemaphore(N)

        controlDeferred = defer.Deferred()
        def helper(self, arg):
            self.arg = arg
            return controlDeferred

        results = []
        uniqueObject = object()
        resultDeferred = sem.run(helper, self=self, arg=uniqueObject)
        resultDeferred.addCallback(results.append)
        resultDeferred.addCallback(self._incr)
        self.assertEqual(results, [])
        self.assertEqual(self.arg, uniqueObject)
        controlDeferred.callback(None)
        self.assertIsNone(results.pop())
        self.assertEqual(self.counter, 1)

        self.counter = 0
        for i in range(1, 1 + N):
            sem.acquire().addCallback(self._incr)
            self.assertEqual(self.counter, i)


        success = []
        def fail(r):
            success.append(False)
        def succeed(r):
            success.append(True)
        d = sem.acquire().addCallbacks(fail, succeed)
        d.cancel()
        self.assertEqual(success, [True])

        sem.acquire().addCallback(self._incr)
        self.assertEqual(self.counter, N)

        sem.release()
        self.assertEqual(self.counter, N + 1)

        for i in range(1, 1 + N):
            sem.release()
            self.assertEqual(self.counter, N + 1)