13
13
# You should have received a copy of the GNU General Public License
14
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
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
"""Test locks across all branch implemenations"""
33
33
def get_instrumented_branch(self):
34
34
"""Get a Branch object which has been instrumented"""
35
# TODO: jam 20060630 It may be that not all formats have a
35
# TODO: jam 20060630 It may be that not all formats have a
36
36
# 'control_files' member. So we should fail gracefully if
37
# not there. But assuming it has them lets us test the exact
37
# not there. But assuming it has them lets us test the exact
38
38
# lock/unlock order.
40
40
b = LockWrapper(self.locks, self.get_branch(), 'b')
42
42
bcf = b.control_files
43
43
rcf = getattr(b.repository, 'control_files', None)
46
"This tests depends on being able to instrument "
47
"repository.control_files, but %r doesn't have control_files."
50
# Look out for branch types that reuse their control files
51
self.combined_control = bcf is rcf
53
b.control_files = LockWrapper(self.locks, b.control_files, 'bc')
54
b.repository.control_files = \
55
LockWrapper(self.locks, b.repository.control_files, 'rc')
45
self.combined_branch = False
47
# Look out for branch types that reuse their control files
48
self.combined_control = bcf is rcf
50
b.control_files = LockWrapper(self.locks, b.control_files, 'bc')
51
except AttributeError:
52
# RemoteBranch seems to trigger this.
53
raise TestSkipped("Could not instrument branch control files.")
54
if self.combined_control:
55
# instrument the repository control files too to ensure its worked
56
# with correctly. When they are not shared, we trust the repository
57
# API and only instrument the repository itself.
58
b.repository.control_files = \
59
LockWrapper(self.locks, b.repository.control_files, 'rc')
58
62
def test_01_lock_read(self):
70
74
self.assertFalse(b.is_locked())
71
75
self.assertFalse(b.repository.is_locked())
73
self.assertEqual([('b', 'lr', True),
77
if self.combined_control:
78
self.assertEqual([('b', 'lr', True),
88
self.assertEqual([('b', 'lr', True),
83
96
def test_02_lock_write(self):
84
97
# Test that locking occurs in the correct order
95
108
self.assertFalse(b.is_locked())
96
109
self.assertFalse(b.repository.is_locked())
98
self.assertEqual([('b', 'lw', True),
111
if self.combined_control:
112
self.assertEqual([('b', 'lw', True),
122
self.assertEqual([('b', 'lw', True),
108
130
def test_03_lock_fail_unlock_repo(self):
109
131
# Make sure branch.unlock() is called, even if there is a
124
146
self.assertFalse(b.is_locked())
125
147
self.assertTrue(b.repository.is_locked())
127
# We unlock the branch control files, even if
149
# We unlock the branch control files, even if
128
150
# we fail to unlock the repository
129
self.assertEqual([('b', 'lw', True),
151
if self.combined_control:
152
self.assertEqual([('b', 'lw', True),
161
self.assertEqual([('b', 'lw', True),
139
170
# For cleanup purposes, make sure we are unlocked
140
171
b.repository._other.unlock()
142
173
def test_04_lock_fail_unlock_control(self):
143
# Make sure repository.unlock() is called, if we fail to unlock self
174
# Make sure repository.unlock() is not called, if we fail to unlock
175
# self leaving ourselves still locked, so that attempts to recover
176
# don't encounter an unlocked repository.
144
177
b = self.get_instrumented_branch()
145
178
b.control_files.disable_unlock()
152
185
self.assertTrue(b.repository.is_locked())
153
186
self.assertRaises(TestPreventLocking, b.unlock)
154
187
self.assertTrue(b.is_locked())
155
if self.combined_control:
156
self.assertTrue(b.repository.is_locked())
158
self.assertFalse(b.repository.is_locked())
188
self.assertTrue(b.repository.is_locked())
160
# We unlock the repository even if
190
# We unlock the repository even if
161
191
# we fail to unlock the control files
162
self.assertEqual([('b', 'lw', True),
192
if self.combined_control:
193
self.assertEqual([('b', 'lw', True),
203
self.assertEqual([('b', 'lw', True),
173
211
# For cleanup purposes, make sure we are unlocked
183
221
self.assertFalse(b.repository.is_locked())
185
223
self.assertEqual([('b', 'lr', True),
189
227
def test_06_lock_write_fail_repo(self):
196
234
self.assertFalse(b.repository.is_locked())
198
236
self.assertEqual([('b', 'lw', True),
202
240
def test_07_lock_read_fail_control(self):
208
246
self.assertFalse(b.is_locked())
209
247
self.assertFalse(b.repository.is_locked())
211
self.assertEqual([('b', 'lr', True),
249
if self.combined_control:
250
self.assertEqual([('b', 'lr', True),
258
self.assertEqual([('b', 'lr', True),
219
264
def test_08_lock_write_fail_control(self):
220
265
# Test the repository is unlocked if we can't lock self
224
269
self.assertRaises(TestPreventLocking, b.lock_write)
225
270
self.assertFalse(b.is_locked())
226
271
self.assertFalse(b.repository.is_locked())
228
self.assertEqual([('b', 'lw', True),
272
if self.combined_control:
273
self.assertEqual([('b', 'lw', True),
281
self.assertEqual([('b', 'lw', True),
236
287
def test_lock_write_returns_None_refuses_token(self):
237
288
branch = self.make_branch('b')
382
433
# We should be unable to relock the repo.
383
434
self.assertRaises(errors.LockContention, branch.lock_write)
436
branch.lock_write(token)
437
branch.dont_leave_lock_in_place()
385
440
def test_dont_leave_lock_in_place(self):
386
441
branch = self.make_branch('b')
398
453
except NotImplementedError:
399
454
# This branch doesn't support this API.
401
branch.repository.leave_lock_in_place()
402
repo_token = branch.repository.lock_write()
403
branch.repository.unlock()
457
branch.repository.leave_lock_in_place()
458
except NotImplementedError:
459
# This repo doesn't support leaving locks around,
460
# assume it is essentially lock-free.
463
repo_token = branch.repository.lock_write()
464
branch.repository.unlock()
406
467
# Reacquire the lock (with a different branch object) by using the
408
469
new_branch = branch.bzrdir.open_branch()
409
# We have to explicitly lock the repository first.
410
new_branch.repository.lock_write(token=repo_token)
470
if repo_token is not None:
471
# We have to explicitly lock the repository first.
472
new_branch.repository.lock_write(token=repo_token)
411
473
new_branch.lock_write(token=token)
412
# Now we don't need our own repository lock anymore (the branch is
413
# holding it for us).
414
new_branch.repository.unlock()
474
if repo_token is not None:
475
# Now we don't need our own repository lock anymore (the branch is
476
# holding it for us).
477
new_branch.repository.unlock()
415
478
# Call dont_leave_lock_in_place, so that the lock will be released by
416
479
# this instance, even though the lock wasn't originally acquired by it.
417
480
new_branch.dont_leave_lock_in_place()
418
new_branch.repository.dont_leave_lock_in_place()
481
if repo_token is not None:
482
new_branch.repository.dont_leave_lock_in_place()
419
483
new_branch.unlock()
420
484
# Now the branch (and repository) is unlocked. Test this by locking it
421
485
# without tokens.
437
501
branch = branch.bzrdir.open_branch()
438
502
branch.lock_write()
440
# Now the branch.repository is locked, so we can't lock it with a new
441
# repository without a token.
504
# The branch should have asked the repository to lock.
505
self.assertTrue(branch.repository.is_write_locked())
506
# Does the repository type actually lock?
507
if not branch.repository.get_physical_lock_status():
508
# The test was successfully applied, so it was applicable.
510
# Now the branch.repository is physically locked, so we can't lock
511
# it with a new repository instance.
442
512
new_repo = branch.bzrdir.open_repository()
443
513
self.assertRaises(errors.LockContention, new_repo.lock_write)
444
514
# We can call lock_write on the original repository object though,
447
517
branch.repository.unlock()
521
def test_lock_and_unlock_leaves_repo_unlocked(self):
522
branch = self.make_branch('b')
525
self.assertRaises(errors.LockNotHeld, branch.repository.unlock)
added
removed
bzrlib/tests/branch_implementations/test_locking.py
