test_defer.py 文件源码

def testLock(self):
        lock = defer.DeferredLock()
        lock.acquire().addCallback(self._incr)
        self.assertTrue(lock.locked)
        self.assertEqual(self.counter, 1)

        lock.acquire().addCallback(self._incr)
        self.assertTrue(lock.locked)
        self.assertEqual(self.counter, 1)

        lock.release()
        self.assertTrue(lock.locked)
        self.assertEqual(self.counter, 2)

        lock.release()
        self.assertFalse(lock.locked)
        self.assertEqual(self.counter, 2)

        self.assertRaises(TypeError, lock.run)

        firstUnique = object()
        secondUnique = object()

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

        resultDeferred = lock.run(helper, self=self, b=firstUnique)
        self.assertTrue(lock.locked)
        self.assertEqual(self.b, firstUnique)

        resultDeferred.addCallback(lambda x: setattr(self, 'result', x))

        lock.acquire().addCallback(self._incr)
        self.assertTrue(lock.locked)
        self.assertEqual(self.counter, 2)

        controlDeferred.callback(secondUnique)
        self.assertEqual(self.result, secondUnique)
        self.assertTrue(lock.locked)
        self.assertEqual(self.counter, 3)

        d = lock.acquire().addBoth(lambda x: setattr(self, 'result', x))
        d.cancel()
        self.assertEqual(self.result.type, defer.CancelledError)

        lock.release()
        self.assertFalse(lock.locked)