1
# Copyright (C) 2006 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for the (un)lock interfaces on all working tree implemenations."""
24
from bzrlib.tests.workingtree_implementations import TestCaseWithWorkingTree
27
class TestWorkingTreeLocking(TestCaseWithWorkingTree):
29
def test_trivial_lock_read_unlock(self):
30
"""Locking and unlocking should work trivially."""
31
wt = self.make_branch_and_tree('.')
33
self.assertFalse(wt.is_locked())
34
self.assertFalse(wt.branch.is_locked())
37
self.assertTrue(wt.is_locked())
38
self.assertTrue(wt.branch.is_locked())
41
self.assertFalse(wt.is_locked())
42
self.assertFalse(wt.branch.is_locked())
44
def test_trivial_lock_write_unlock(self):
45
"""Locking for write and unlocking should work trivially."""
46
wt = self.make_branch_and_tree('.')
48
self.assertFalse(wt.is_locked())
49
self.assertFalse(wt.branch.is_locked())
52
self.assertTrue(wt.is_locked())
53
self.assertTrue(wt.branch.is_locked())
56
self.assertFalse(wt.is_locked())
57
self.assertFalse(wt.branch.is_locked())
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('.')
63
self.assertFalse(wt.is_locked())
64
self.assertFalse(wt.branch.is_locked())
67
self.assertTrue(wt.is_locked())
68
self.assertTrue(wt.branch.is_locked())
71
self.assertFalse(wt.is_locked())
72
self.assertFalse(wt.branch.is_locked())
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('.')
78
self.assertFalse(wt.is_locked())
79
self.assertFalse(wt.branch.is_locked())
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.
91
self.assertFalse(wt.is_locked())
92
self.assertFalse(wt.branch.is_locked())
95
self.assertTrue(wt.is_locked())
96
self.assertTrue(wt.branch.is_locked())
99
self.assertFalse(wt.is_locked())
100
self.assertTrue(wt.branch.is_locked())
103
def _test_unlock_with_lock_method(self, methodname):
104
"""Create a tree and then test its unlocking behaviour.
106
:param methodname: The lock method to use to establish locks.
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
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.
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.
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())
143
def test_unlock_from_tree_write_lock_flushes(self):
144
self._test_unlock_with_lock_method("lock_tree_write")
146
def test_unlock_from_write_lock_flushes(self):
147
self._test_unlock_with_lock_method("lock_write")
149
def test_unlock_branch_failures(self):
150
"""If the branch unlock fails the tree must still unlock."""
151
# The public interface for WorkingTree requires a branch, but
152
# does not require that the working tree use the branch - its
153
# implementation specific how the WorkingTree, Branch, and Repository
155
# in order to test that implementations which *do* unlock via the branch
156
# do so correctly, we unlock the branch after locking the working tree.
157
# The next unlock on working tree should trigger a LockNotHeld exception
158
# from the branch object, which must be exposed to the caller. To meet
159
# our object model - where locking a tree locks its branch, and
160
# unlocking a branch does not unlock a working tree, *even* for
161
# all-in-one implementations like bzr 0.6, git, and hg, implementations
162
# must have some separate counter for each object, so our explicit
163
# unlock should trigger some error on all implementations, and
164
# requiring that to be LockNotHeld seems reasonable.
166
# we use this approach rather than decorating the Branch, because the
167
# public interface of WorkingTree does not permit altering the branch
168
# object - and we cannot tell which attribute might allow us to
169
# backdoor-in and change it reliably. For implementation specific tests
170
# we can do such skullduggery, but not for interface specific tests.
171
# And, its simpler :)
172
wt = self.make_branch_and_tree('.')
174
self.assertFalse(wt.is_locked())
175
self.assertFalse(wt.branch.is_locked())
177
self.assertTrue(wt.is_locked())
178
self.assertTrue(wt.branch.is_locked())
180
# manually unlock the branch, preparing a LockNotHeld error.
182
# the branch *may* still be locked here, if its an all-in-one
183
# implementation because there is a single lock object with three
184
# references on it, and unlocking the branch only drops this by two
185
self.assertRaises(errors.LockNotHeld, wt.unlock)
186
# but now, the tree must be unlocked
187
self.assertFalse(wt.is_locked())
188
# and the branch too.
189
self.assertFalse(wt.branch.is_locked())
191
def test_failing_to_lock_branch_does_not_lock(self):
192
"""If the branch cannot be locked, dont lock the tree."""
193
# Many implementations treat read-locks as non-blocking, but some
194
# treat them as blocking with writes.. Accordingly we test this by
195
# opening the branch twice, and locking the branch for write in the
196
# second instance. Our lock contract requires separate instances to
197
# mutually exclude if a lock is exclusive at all: If we get no error
198
# locking, the test still passes.
199
wt = self.make_branch_and_tree('.')
200
branch_copy = branch.Branch.open('.')
201
branch_copy.lock_write()
205
except errors.LockError:
206
# any error here means the locks are exclusive in some
208
self.assertFalse(wt.is_locked())
209
self.assertFalse(wt.branch.is_locked())
212
# no error - the branch allows read locks while writes
213
# are taken, just pass.
218
def test_failing_to_lock_write_branch_does_not_lock(self):
219
"""If the branch cannot be write locked, dont lock the tree."""
220
# all implementations of branch are required to treat write
221
# locks as blocking (compare to repositories which are not required
223
# Accordingly we test this by opening the branch twice, and locking the
224
# branch for write in the second instance. Our lock contract requires
225
# separate instances to mutually exclude.
226
wt = self.make_branch_and_tree('.')
227
branch_copy = branch.Branch.open('.')
228
branch_copy.lock_write()
231
orig_default = lockdir._DEFAULT_TIMEOUT_SECONDS
233
lockdir._DEFAULT_TIMEOUT_SECONDS = 1
234
self.assertRaises(errors.LockError, wt.lock_write)
236
lockdir._DEFAULT_TIMEOUT_SECONDS = orig_default
238
self.assertFalse(wt.is_locked())
239
self.assertFalse(wt.branch.is_locked())
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('.')
257
branch_copy.lock_write()
261
except errors.LockError:
262
# any error here means the locks are exclusive in some
264
self.assertFalse(wt.is_locked())
265
self.assertFalse(wt.branch.is_locked())
268
# no error - the branch allows read locks while writes
269
# are taken, just pass.
added
removed
bzrlib/tests/workingtree_implementations/test_locking.py
