rendered paste body def testWaitUntilLockedWithTimeoutLocked(self): lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) self.failUnless(lock.lock()) clock = Clock() d = lock.deferUntilLocked(timeout=5, interval=1, clock=clock) clock.advance(10) self.assertFailure(d, lockfile.TimeoutError) return d def testWaitUntilLockedWithTimeoutUnlocked(self): def onTimeout(f): f.trap(lockfile.TimeoutError) self.fail("Should not have timed out") lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) self.failUnless(lock.lock()) clock = Clock() clock.callLater(1, lock.unlock) d = lock.deferUntilLocked(timeout=10, interval=1, clock=clock) d.addErrback(onTimeout) clock.advance(10) return d