← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: John Arbash Meinel
  • Date: 2006-08-25 20:51:33 UTC
  • mto: This revision was merged to the branch mainline in revision 2050.
  • Revision ID: john@arbash-meinel.com-20060825205133-402530a133d623a0
Switch the default from instantly aborting, to waiting as long as 1 minute (down from 5 minutes)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
26
26
        LockContention, LockError, UnlockableTransport,
27
27
        LockNotHeld, LockBroken
28
28
        )
29
 
from bzrlib.lockdir import LockDir
 
29
from bzrlib.lockdir import LockDir, _DEFAULT_TIMEOUT_SECONDS
30
30
from bzrlib.tests import TestCaseWithTransport
31
31
 
32
32
# These tests sometimes use threads to test the behaviour of lock files with
260
260
        finally:
261
261
            unlocker.join()
262
262
 
 
263
    def test_34_lock_write_waits(self):
 
264
        """LockDir.lock_write() will wait for the lock.""" 
 
265
        t = self.get_transport()
 
266
        lf1 = LockDir(t, 'test_lock')
 
267
        lf1.create()
 
268
        lf1.attempt_lock()
 
269
 
 
270
        def wait_and_unlock():
 
271
            time.sleep(0.1)
 
272
            lf1.unlock()
 
273
        unlocker = Thread(target=wait_and_unlock)
 
274
        unlocker.start()
 
275
        try:
 
276
            lf2 = LockDir(t, 'test_lock')
 
277
            self.setup_log_reporter(lf2)
 
278
            before = time.time()
 
279
            # wait and then lock
 
280
            lf2.lock_write()
 
281
            after = time.time()
 
282
            self.assertTrue(after - before <= 1.0)
 
283
        finally:
 
284
            unlocker.join()
 
285
 
 
286
        # There should be only 1 report, even though it should have to
 
287
        # wait for a while
 
288
        lock_base = lf2.transport.abspath(lf2.path)
 
289
        self.assertEqual([('Unable to obtain lock on %s\n'
 
290
                           'Will continue to try for %s seconds\n',
 
291
                           (lock_base, _DEFAULT_TIMEOUT_SECONDS)),
 
292
                         ], self._logged_reports)
 
293
 
 
294
 
263
295
    def test_40_confirm_easy(self):
264
296
        """Confirm a lock that's already held"""
265
297
        t = self.get_transport()