/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: Martin Pool
  • Date: 2007-08-20 05:53:39 UTC
  • mfrom: (2727 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2730.
  • Revision ID: mbp@sourcefrog.net-20070820055339-uzei7f7i7jo6tugg
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
27
27
    errors,
28
28
    osutils,
29
29
    tests,
 
30
    transport,
30
31
    )
31
32
from bzrlib.errors import (
32
33
        LockBreakMismatch,
220
221
        One thread holds on a lock and then releases it; another 
221
222
        tries to lock it.
222
223
        """
 
224
        # This test sometimes fails like this:
 
225
        # Traceback (most recent call last):
 
226
 
 
227
        #   File "/home/pqm/bzr-pqm-workdir/home/+trunk/bzrlib/tests/
 
228
        # test_lockdir.py", line 247, in test_32_lock_wait_succeed
 
229
        #     self.assertEqual(1, len(self._logged_reports))
 
230
        # AssertionError: not equal:
 
231
        # a = 1
 
232
        # b = 0
 
233
        raise tests.TestSkipped("Test fails intermittently")
223
234
        t = self.get_transport()
224
235
        lf1 = LockDir(t, 'test_lock')
225
236
        lf1.create()
257
268
        self.assertEndsWith(args[3], ' ago')
258
269
        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
259
270
 
260
 
    def test_33_wait(self):
261
 
        """Succeed when waiting on a lock that gets released
262
 
 
263
 
        The difference from test_32_lock_wait_succeed is that the second 
264
 
        caller does not actually acquire the lock, but just waits for it
265
 
        to be released.  This is done over a readonly transport.
266
 
        """
267
 
        t = self.get_transport()
268
 
        lf1 = LockDir(t, 'test_lock')
269
 
        lf1.create()
270
 
        lf1.attempt_lock()
271
 
 
272
 
        def wait_and_unlock():
273
 
            time.sleep(0.1)
274
 
            lf1.unlock()
275
 
        unlocker = Thread(target=wait_and_unlock)
276
 
        unlocker.start()
277
 
        try:
278
 
            lf2 = LockDir(self.get_readonly_transport(), 'test_lock')
279
 
            before = time.time()
280
 
            # wait but don't lock
281
 
            lf2.wait(timeout=0.4, poll=0.1)
282
 
            after = time.time()
283
 
            self.assertTrue(after - before <= 1.0)
284
 
        finally:
285
 
            unlocker.join()
286
 
 
287
271
    def test_34_lock_write_waits(self):
288
272
        """LockDir.lock_write() will wait for the lock.""" 
289
273
        # the test suite sets the default to 0 to make deadlocks fail fast.
290
274
        # change it for this test, as we want to try a manual deadlock.
 
275
        raise tests.TestSkipped('Timing-sensitive test')
291
276
        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 300
292
277
        t = self.get_transport()
293
278
        lf1 = LockDir(t, 'test_lock')
410
395
        unlocker.start()
411
396
        try:
412
397
            # Wait and play against the other thread
413
 
            lf2.wait_lock(timeout=1.0, poll=0.01)
 
398
            lf2.wait_lock(timeout=20.0, poll=0.01)
414
399
        finally:
415
400
            unlocker.join()
416
401
        lf2.unlock()
615
600
        os.mkdir(lock_path)
616
601
        osutils.make_readonly(lock_path)
617
602
        self.assertRaises(errors.PermissionDenied, ld1.attempt_lock)
 
603
 
 
604
    def test_lock_by_token(self):
 
605
        ld1 = self.get_lock()
 
606
        token = ld1.lock_write()
 
607
        self.assertNotEqual(None, token)
 
608
        ld2 = self.get_lock()
 
609
        t2 = ld2.lock_write(token)
 
610
        self.assertEqual(token, t2)
 
611
 
 
612
    def test_lock_with_buggy_rename(self):
 
613
        # test that lock acquisition handles servers which pretend they
 
614
        # renamed correctly but that actually fail
 
615
        t = transport.get_transport('brokenrename+' + self.get_url())
 
616
        ld1 = LockDir(t, 'test_lock')
 
617
        ld1.create()
 
618
        ld1.attempt_lock()
 
619
        ld2 = LockDir(t, 'test_lock')
 
620
        # we should fail to lock
 
621
        e = self.assertRaises(errors.LockContention, ld2.attempt_lock)
 
622
        # now the original caller should succeed in unlocking
 
623
        ld1.unlock()
 
624
        # and there should be nothing left over
 
625
        self.assertEquals([], t.list_dir('test_lock'))
 
626
 
 
627
    def test_failed_lock_leaves_no_trash(self):
 
628
        # if we fail to acquire the lock, we don't leave pending directories
 
629
        # behind -- https://bugs.launchpad.net/bzr/+bug/109169
 
630
        ld1 = self.get_lock()
 
631
        ld2 = self.get_lock()
 
632
        # should be nothing before we start
 
633
        ld1.create()
 
634
        t = self.get_transport().clone('test_lock')
 
635
        def check_dir(a):
 
636
            self.assertEquals(a, t.list_dir('.'))
 
637
        check_dir([])
 
638
        # when held, that's all we see
 
639
        ld1.attempt_lock()
 
640
        check_dir(['held'])
 
641
        # second guy should fail
 
642
        self.assertRaises(errors.LockContention, ld2.attempt_lock)
 
643
        # no kibble
 
644
        check_dir(['held'])