1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
|
# Copyright (C) 2006 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Tests for the (un)lock interfaces on all working tree implemenations."""
from bzrlib import (
branch,
errors,
lockdir,
)
from bzrlib.tests.workingtree_implementations import TestCaseWithWorkingTree
class TestWorkingTreeLocking(TestCaseWithWorkingTree):
def test_trivial_lock_read_unlock(self):
"""Locking and unlocking should work trivially."""
wt = self.make_branch_and_tree('.')
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
wt.lock_read()
try:
self.assertTrue(wt.is_locked())
self.assertTrue(wt.branch.is_locked())
finally:
wt.unlock()
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
def test_trivial_lock_write_unlock(self):
"""Locking for write and unlocking should work trivially."""
wt = self.make_branch_and_tree('.')
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
wt.lock_write()
try:
self.assertTrue(wt.is_locked())
self.assertTrue(wt.branch.is_locked())
finally:
wt.unlock()
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
def test_trivial_lock_tree_write_unlock(self):
"""Locking for tree write is ok when the branch is not locked."""
wt = self.make_branch_and_tree('.')
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
wt.lock_tree_write()
try:
self.assertTrue(wt.is_locked())
self.assertTrue(wt.branch.is_locked())
finally:
wt.unlock()
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
def test_trivial_lock_tree_write_branch_read_locked(self):
"""It is ok to lock_tree_write when the branch is read locked."""
wt = self.make_branch_and_tree('.')
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
wt.branch.lock_read()
try:
wt.lock_tree_write()
except errors.ReadOnlyError:
# When ReadOnlyError is raised, it indicates that the
# workingtree shares its lock with the branch, which is what
# the git/hg/bzr0.6 formats do.
# in this case, no lock should have been taken - but the tree
# will have been locked because they share a lock. Unlocking
# just the branch should make everything match again correctly.
wt.branch.unlock()
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
return
try:
self.assertTrue(wt.is_locked())
self.assertTrue(wt.branch.is_locked())
finally:
wt.unlock()
self.assertFalse(wt.is_locked())
self.assertTrue(wt.branch.is_locked())
wt.branch.unlock()
def _test_unlock_with_lock_method(self, methodname):
"""Create a tree and then test its unlocking behaviour.
:param methodname: The lock method to use to establish locks.
"""
# when unlocking the last lock count from tree_write_lock,
# the tree should do a flush().
# we test that by changing the inventory using set_root_id
tree = self.make_branch_and_tree('tree')
# prepare for a series of changes that will modify the
# inventory
getattr(tree, methodname)()
# note that we dont have a try:finally here because of two reasons:
# firstly there will only be errors reported if the test fails, and
# when it fails thats ok as long as the test suite cleanup still works,
# which it will as the lock objects are released (thats where the
# warning comes from. Secondly, it is hard in this test to be
# sure that we've got the right interactions between try:finally
# and the lock/unlocks we are doing.
getattr(tree, methodname)()
# this should really do something within the public api
# e.g. mkdir('foo') but all the mutating methods at the
# moment trigger inventory writes and thus will not
# let us trigger a read-when-dirty situation.
old_root = tree.get_root_id()
tree.set_root_id('new-root')
# to detect that the inventory is written by unlock, we
# first check that it was not written yet.
reference_tree = tree.bzrdir.open_workingtree()
self.assertEqual(old_root, reference_tree.get_root_id())
# now unlock the second held lock, which should do nothing.
tree.unlock()
reference_tree = tree.bzrdir.open_workingtree()
self.assertEqual(old_root, reference_tree.get_root_id())
# unlocking the first lock we took will now flush.
tree.unlock()
# and check it was written using another reference tree
reference_tree = tree.bzrdir.open_workingtree()
self.assertEqual('new-root', reference_tree.get_root_id())
def test_unlock_from_tree_write_lock_flushes(self):
self._test_unlock_with_lock_method("lock_tree_write")
def test_unlock_from_write_lock_flushes(self):
self._test_unlock_with_lock_method("lock_write")
def test_unlock_branch_failures(self):
"""If the branch unlock fails the tree must still unlock."""
# The public interface for WorkingTree requires a branch, but
# does not require that the working tree use the branch - its
# implementation specific how the WorkingTree, Branch, and Repository
# hang together.
# in order to test that implementations which *do* unlock via the branch
# do so correctly, we unlock the branch after locking the working tree.
# The next unlock on working tree should trigger a LockNotHeld exception
# from the branch object, which must be exposed to the caller. To meet
# our object model - where locking a tree locks its branch, and
# unlocking a branch does not unlock a working tree, *even* for
# all-in-one implementations like bzr 0.6, git, and hg, implementations
# must have some separate counter for each object, so our explicit
# unlock should trigger some error on all implementations, and
# requiring that to be LockNotHeld seems reasonable.
#
# we use this approach rather than decorating the Branch, because the
# public interface of WorkingTree does not permit altering the branch
# object - and we cannot tell which attribute might allow us to
# backdoor-in and change it reliably. For implementation specific tests
# we can do such skullduggery, but not for interface specific tests.
# And, its simpler :)
wt = self.make_branch_and_tree('.')
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
wt.lock_write()
self.assertTrue(wt.is_locked())
self.assertTrue(wt.branch.is_locked())
# manually unlock the branch, preparing a LockNotHeld error.
wt.branch.unlock()
# the branch *may* still be locked here, if its an all-in-one
# implementation because there is a single lock object with three
# references on it, and unlocking the branch only drops this by two
self.assertRaises(errors.LockNotHeld, wt.unlock)
# but now, the tree must be unlocked
self.assertFalse(wt.is_locked())
# and the branch too.
self.assertFalse(wt.branch.is_locked())
def test_failing_to_lock_branch_does_not_lock(self):
"""If the branch cannot be locked, dont lock the tree."""
# Many implementations treat read-locks as non-blocking, but some
# treat them as blocking with writes.. Accordingly we test this by
# opening the branch twice, and locking the branch for write in the
# second instance. Our lock contract requires separate instances to
# mutually exclude if a lock is exclusive at all: If we get no error
# locking, the test still passes.
wt = self.make_branch_and_tree('.')
branch_copy = branch.Branch.open('.')
branch_copy.lock_write()
try:
try:
wt.lock_read()
except errors.LockError:
# any error here means the locks are exclusive in some
# manner
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
return
else:
# no error - the branch allows read locks while writes
# are taken, just pass.
wt.unlock()
finally:
branch_copy.unlock()
def test_failing_to_lock_write_branch_does_not_lock(self):
"""If the branch cannot be write locked, dont lock the tree."""
# all implementations of branch are required to treat write
# locks as blocking (compare to repositories which are not required
# to do so).
# Accordingly we test this by opening the branch twice, and locking the
# branch for write in the second instance. Our lock contract requires
# separate instances to mutually exclude.
wt = self.make_branch_and_tree('.')
branch_copy = branch.Branch.open('.')
branch_copy.lock_write()
try:
try:
self.assertRaises(errors.LockError, wt.lock_write)
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
finally:
if wt.is_locked():
wt.unlock()
finally:
branch_copy.unlock()
def test_failing_to_lock_tree_write_branch_does_not_lock(self):
"""If the branch cannot be read locked, dont lock the tree."""
# Many implementations treat read-locks as non-blocking, but some
# treat them as blocking with writes.. Accordingly we test this by
# opening the branch twice, and locking the branch for write in the
# second instance. Our lock contract requires separate instances to
# mutually exclude if a lock is exclusive at all: If we get no error
# locking, the test still passes.
wt = self.make_branch_and_tree('.')
branch_copy = branch.Branch.open('.')
branch_copy.lock_write()
try:
try:
wt.lock_tree_write()
except errors.LockError:
# any error here means the locks are exclusive in some
# manner
self.assertFalse(wt.is_locked())
self.assertFalse(wt.branch.is_locked())
return
else:
# no error - the branch allows read locks while writes
# are taken, just pass.
wt.unlock()
finally:
branch_copy.unlock()
|