← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

merge robert, debugging

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2006 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
27
27
    errors,
28
28
    osutils,
29
29
    tests,
30
 
    transport,
31
30
    )
32
31
from bzrlib.errors import (
33
 
    LockBreakMismatch,
34
 
    LockBroken,
35
 
    LockContention,
36
 
    LockError,
37
 
    LockFailed,
38
 
    LockNotHeld,
39
 
    )
40
 
from bzrlib.lockdir import LockDir
 
32
        LockBreakMismatch,
 
33
        LockContention, LockError, UnlockableTransport,
 
34
        LockNotHeld, LockBroken
 
35
        )
 
36
from bzrlib.lockdir import LockDir, _DEFAULT_TIMEOUT_SECONDS
41
37
from bzrlib.tests import TestCaseWithTransport
42
38
from bzrlib.trace import note
43
39
 
108
104
        """Fail to create lock on readonly transport"""
109
105
        t = self.get_readonly_transport()
110
106
        lf = LockDir(t, 'test_lock')
111
 
        self.assertRaises(LockFailed, lf.create)
 
107
        self.assertRaises(UnlockableTransport, lf.create)
112
108
 
113
109
    def test_12_lock_readonly_transport(self):
114
110
        """Fail to lock on readonly transport"""
115
111
        lf = LockDir(self.get_transport(), 'test_lock')
116
112
        lf.create()
117
113
        lf = LockDir(self.get_readonly_transport(), 'test_lock')
118
 
        self.assertRaises(LockFailed, lf.attempt_lock)
 
114
        self.assertRaises(UnlockableTransport, lf.attempt_lock)
119
115
 
120
116
    def test_20_lock_contested(self):
121
117
        """Contention to get a lock"""
224
220
        One thread holds on a lock and then releases it; another 
225
221
        tries to lock it.
226
222
        """
227
 
        # This test sometimes fails like this:
228
 
        # Traceback (most recent call last):
229
 
 
230
 
        #   File "/home/pqm/bzr-pqm-workdir/home/+trunk/bzrlib/tests/
231
 
        # test_lockdir.py", line 247, in test_32_lock_wait_succeed
232
 
        #     self.assertEqual(1, len(self._logged_reports))
233
 
        # AssertionError: not equal:
234
 
        # a = 1
235
 
        # b = 0
236
 
        raise tests.TestSkipped("Test fails intermittently")
237
223
        t = self.get_transport()
238
224
        lf1 = LockDir(t, 'test_lock')
239
225
        lf1.create()
271
257
        self.assertEndsWith(args[3], ' ago')
272
258
        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
273
259
 
 
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
 
274
287
    def test_34_lock_write_waits(self):
275
288
        """LockDir.lock_write() will wait for the lock.""" 
276
 
        # the test suite sets the default to 0 to make deadlocks fail fast.
277
 
        # change it for this test, as we want to try a manual deadlock.
278
 
        raise tests.TestSkipped('Timing-sensitive test')
279
 
        bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 300
280
289
        t = self.get_transport()
281
290
        lf1 = LockDir(t, 'test_lock')
282
291
        lf1.create()
398
407
        unlocker.start()
399
408
        try:
400
409
            # Wait and play against the other thread
401
 
            lf2.wait_lock(timeout=20.0, poll=0.01)
 
410
            lf2.wait_lock(timeout=1.0, poll=0.01)
402
411
        finally:
403
412
            unlocker.join()
404
413
        lf2.unlock()
602
611
        lock_path = ld1.transport.local_abspath('test_lock')
603
612
        os.mkdir(lock_path)
604
613
        osutils.make_readonly(lock_path)
605
 
        self.assertRaises(errors.LockFailed, ld1.attempt_lock)
606
 
 
607
 
    def test_lock_by_token(self):
608
 
        ld1 = self.get_lock()
609
 
        token = ld1.lock_write()
610
 
        self.assertNotEqual(None, token)
611
 
        ld2 = self.get_lock()
612
 
        t2 = ld2.lock_write(token)
613
 
        self.assertEqual(token, t2)
614
 
 
615
 
    def test_lock_with_buggy_rename(self):
616
 
        # test that lock acquisition handles servers which pretend they
617
 
        # renamed correctly but that actually fail
618
 
        t = transport.get_transport('brokenrename+' + self.get_url())
619
 
        ld1 = LockDir(t, 'test_lock')
620
 
        ld1.create()
621
 
        ld1.attempt_lock()
622
 
        ld2 = LockDir(t, 'test_lock')
623
 
        # we should fail to lock
624
 
        e = self.assertRaises(errors.LockContention, ld2.attempt_lock)
625
 
        # now the original caller should succeed in unlocking
626
 
        ld1.unlock()
627
 
        # and there should be nothing left over
628
 
        self.assertEquals([], t.list_dir('test_lock'))
629
 
 
630
 
    def test_failed_lock_leaves_no_trash(self):
631
 
        # if we fail to acquire the lock, we don't leave pending directories
632
 
        # behind -- https://bugs.launchpad.net/bzr/+bug/109169
633
 
        ld1 = self.get_lock()
634
 
        ld2 = self.get_lock()
635
 
        # should be nothing before we start
636
 
        ld1.create()
637
 
        t = self.get_transport().clone('test_lock')
638
 
        def check_dir(a):
639
 
            self.assertEquals(a, t.list_dir('.'))
640
 
        check_dir([])
641
 
        # when held, that's all we see
642
 
        ld1.attempt_lock()
643
 
        check_dir(['held'])
644
 
        # second guy should fail
645
 
        self.assertRaises(errors.LockContention, ld2.attempt_lock)
646
 
        # no kibble
647
 
        check_dir(['held'])
 
614
        self.assertRaises(errors.PermissionDenied, ld1.attempt_lock)