← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/workingtree_implementations/test_locking.py

Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/workingtree_implementations/test_locking.py
16
16
 
17
17
"""Tests for the (un)lock interfaces on all working tree implemenations."""
18
18
 
19
 
import bzrlib.branch as branch
20
 
import bzrlib.errors as errors
 
19
from bzrlib import (
 
20
    branch,
 
21
    errors,
 
22
    lockdir,
 
23
    )
21
24
from bzrlib.tests.workingtree_implementations import TestCaseWithWorkingTree
22
25
 
23
26
 
53
56
        self.assertFalse(wt.is_locked())
54
57
        self.assertFalse(wt.branch.is_locked())
55
58
        
 
59
    def test_trivial_lock_tree_write_unlock(self):
 
60
        """Locking for tree write is ok when the branch is not locked."""
 
61
        wt = self.make_branch_and_tree('.')
 
62
 
 
63
        self.assertFalse(wt.is_locked())
 
64
        self.assertFalse(wt.branch.is_locked())
 
65
        wt.lock_tree_write()
 
66
        try:
 
67
            self.assertTrue(wt.is_locked())
 
68
            self.assertTrue(wt.branch.is_locked())
 
69
        finally:
 
70
            wt.unlock()
 
71
        self.assertFalse(wt.is_locked())
 
72
        self.assertFalse(wt.branch.is_locked())
 
73
        
 
74
    def test_trivial_lock_tree_write_branch_read_locked(self):
 
75
        """It is ok to lock_tree_write when the branch is read locked."""
 
76
        wt = self.make_branch_and_tree('.')
 
77
 
 
78
        self.assertFalse(wt.is_locked())
 
79
        self.assertFalse(wt.branch.is_locked())
 
80
        wt.branch.lock_read()
 
81
        try:
 
82
            wt.lock_tree_write()
 
83
        except errors.ReadOnlyError:
 
84
            # When ReadOnlyError is raised, it indicates that the 
 
85
            # workingtree shares its lock with the branch, which is what
 
86
            # the git/hg/bzr0.6 formats do.
 
87
            # in this case, no lock should have been taken - but the tree
 
88
            # will have been locked because they share a lock. Unlocking
 
89
            # just the branch should make everything match again correctly.
 
90
            wt.branch.unlock()
 
91
            self.assertFalse(wt.is_locked())
 
92
            self.assertFalse(wt.branch.is_locked())
 
93
            return
 
94
        try:
 
95
            self.assertTrue(wt.is_locked())
 
96
            self.assertTrue(wt.branch.is_locked())
 
97
        finally:
 
98
            wt.unlock()
 
99
        self.assertFalse(wt.is_locked())
 
100
        self.assertTrue(wt.branch.is_locked())
 
101
        wt.branch.unlock()
 
102
        
 
103
    def _test_unlock_with_lock_method(self, methodname):
 
104
        """Create a tree and then test its unlocking behaviour.
 
105
 
 
106
        :param methodname: The lock method to use to establish locks.
 
107
        """
 
108
        # when unlocking the last lock count from tree_write_lock,
 
109
        # the tree should do a flush().
 
110
        # we test that by changing the inventory using set_root_id
 
111
        tree = self.make_branch_and_tree('tree')
 
112
        # prepare for a series of changes that will modify the 
 
113
        # inventory
 
114
        getattr(tree, methodname)()
 
115
        # note that we dont have a try:finally here because of two reasons:
 
116
        # firstly there will only be errors reported if the test fails, and 
 
117
        # when it fails thats ok as long as the test suite cleanup still works,
 
118
        # which it will as the lock objects are released (thats where the 
 
119
        # warning comes from.  Secondly, it is hard in this test to be 
 
120
        # sure that we've got the right interactions between try:finally
 
121
        # and the lock/unlocks we are doing.
 
122
        getattr(tree, methodname)()
 
123
        # this should really do something within the public api
 
124
        # e.g. mkdir('foo') but all the mutating methods at the
 
125
        # moment trigger inventory writes and thus will not 
 
126
        # let us trigger a read-when-dirty situation.
 
127
        old_root = tree.get_root_id()
 
128
        tree.set_root_id('new-root')
 
129
        # to detect that the inventory is written by unlock, we
 
130
        # first check that it was not written yet.
 
131
        reference_tree = tree.bzrdir.open_workingtree()
 
132
        self.assertEqual(old_root, reference_tree.get_root_id())
 
133
        # now unlock the second held lock, which should do nothing.
 
134
        tree.unlock()
 
135
        reference_tree = tree.bzrdir.open_workingtree()
 
136
        self.assertEqual(old_root, reference_tree.get_root_id())
 
137
        # unlocking the first lock we took will now flush.
 
138
        tree.unlock()
 
139
        # and check it was written using another reference tree
 
140
        reference_tree = tree.bzrdir.open_workingtree()
 
141
        self.assertEqual('new-root', reference_tree.get_root_id())
 
142
 
 
143
    def test_unlock_from_tree_write_lock_flushes(self):
 
144
        self._test_unlock_with_lock_method("lock_tree_write")
 
145
        
 
146
    def test_unlock_from_write_lock_flushes(self):
 
147
        self._test_unlock_with_lock_method("lock_write")
 
148
        
56
149
    def test_unlock_branch_failures(self):
57
150
        """If the branch unlock fails the tree must still unlock."""
58
151
        # The public interface for WorkingTree requires a branch, but
135
228
        branch_copy.lock_write()
136
229
        try:
137
230
            try:
138
 
                self.assertRaises(errors.LockError, wt.lock_write)
 
231
                orig_default = lockdir._DEFAULT_TIMEOUT_SECONDS
 
232
                try:
 
233
                    lockdir._DEFAULT_TIMEOUT_SECONDS = 1
 
234
                    self.assertRaises(errors.LockError, wt.lock_write)
 
235
                finally:
 
236
                    lockdir._DEFAULT_TIMEOUT_SECONDS = orig_default
 
237
 
139
238
                self.assertFalse(wt.is_locked())
140
239
                self.assertFalse(wt.branch.is_locked())
141
240
            finally:
143
242
                    wt.unlock()
144
243
        finally:
145
244
            branch_copy.unlock()
 
245
 
 
246
    def test_failing_to_lock_tree_write_branch_does_not_lock(self):
 
247
        """If the branch cannot be read locked, dont lock the tree."""
 
248
        # Many implementations treat read-locks as non-blocking, but some
 
249
        # treat them as blocking with writes.. Accordingly we test this by
 
250
        # opening the branch twice, and locking the branch for write in the
 
251
        # second instance.  Our lock contract requires separate instances to
 
252
        # mutually exclude if a lock is exclusive at all: If we get no error
 
253
        # locking, the test still passes.
 
254
        wt = self.make_branch_and_tree('.')
 
255
        branch_copy = branch.Branch.open('.')
 
256
 
 
257
        branch_copy.lock_write()
 
258
        try:
 
259
            try:
 
260
                wt.lock_tree_write()
 
261
            except errors.LockError:
 
262
                # any error here means the locks are exclusive in some 
 
263
                # manner
 
264
                self.assertFalse(wt.is_locked())
 
265
                self.assertFalse(wt.branch.is_locked())
 
266
                return
 
267
            else:
 
268
                # no error - the branch allows read locks while writes
 
269
                # are taken, just pass.
 
270
                wt.unlock()
 
271
        finally:
 
272
            branch_copy.unlock()