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
"""Test locks across all branch implemenations"""
19
from bzrlib import errors
20
from bzrlib.branch import BzrBranchFormat4
21
from bzrlib.bzrdir import RemoteBzrDirFormat
22
from bzrlib.tests import TestSkipped
23
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
24
from bzrlib.tests.lock_helpers import TestPreventLocking, LockWrapper
27
class TestBranchLocking(TestCaseWithBranch):
30
TestCaseWithBranch.setUp(self)
31
self.reduceLockdirTimeout()
33
def get_instrumented_branch(self):
34
"""Get a Branch object which has been instrumented"""
35
# TODO: jam 20060630 It may be that not all formats have a
36
# 'control_files' member. So we should fail gracefully if
37
# not there. But assuming it has them lets us test the exact
40
b = LockWrapper(self.locks, self.get_branch(), 'b')
41
b.repository = LockWrapper(self.locks, b.repository, 'r')
43
rcf = getattr(b.repository, 'control_files', None)
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')
62
def test_01_lock_read(self):
63
# Test that locking occurs in the correct order
64
b = self.get_instrumented_branch()
66
self.assertFalse(b.is_locked())
67
self.assertFalse(b.repository.is_locked())
70
self.assertTrue(b.is_locked())
71
self.assertTrue(b.repository.is_locked())
74
self.assertFalse(b.is_locked())
75
self.assertFalse(b.repository.is_locked())
77
if self.combined_control:
78
self.assertEqual([('b', 'lr', True),
88
self.assertEqual([('b', 'lr', True),
96
def test_02_lock_write(self):
97
# Test that locking occurs in the correct order
98
b = self.get_instrumented_branch()
100
self.assertFalse(b.is_locked())
101
self.assertFalse(b.repository.is_locked())
104
self.assertTrue(b.is_locked())
105
self.assertTrue(b.repository.is_locked())
108
self.assertFalse(b.is_locked())
109
self.assertFalse(b.repository.is_locked())
111
if self.combined_control:
112
self.assertEqual([('b', 'lw', True),
122
self.assertEqual([('b', 'lw', True),
130
def test_03_lock_fail_unlock_repo(self):
131
# Make sure branch.unlock() is called, even if there is a
132
# failure while unlocking the repository.
133
b = self.get_instrumented_branch()
134
b.repository.disable_unlock()
136
self.assertFalse(b.is_locked())
137
self.assertFalse(b.repository.is_locked())
140
self.assertTrue(b.is_locked())
141
self.assertTrue(b.repository.is_locked())
142
self.assertRaises(TestPreventLocking, b.unlock)
143
if self.combined_control:
144
self.assertTrue(b.is_locked())
146
self.assertFalse(b.is_locked())
147
self.assertTrue(b.repository.is_locked())
149
# We unlock the branch control files, even if
150
# we fail to unlock the repository
151
if self.combined_control:
152
self.assertEqual([('b', 'lw', True),
161
self.assertEqual([('b', 'lw', True),
170
# For cleanup purposes, make sure we are unlocked
171
b.repository._other.unlock()
173
def test_04_lock_fail_unlock_control(self):
174
# Make sure repository.unlock() is called, if we fail to unlock self
175
b = self.get_instrumented_branch()
176
b.control_files.disable_unlock()
178
self.assertFalse(b.is_locked())
179
self.assertFalse(b.repository.is_locked())
182
self.assertTrue(b.is_locked())
183
self.assertTrue(b.repository.is_locked())
184
self.assertRaises(TestPreventLocking, b.unlock)
185
self.assertTrue(b.is_locked())
186
if self.combined_control:
187
self.assertTrue(b.repository.is_locked())
189
self.assertFalse(b.repository.is_locked())
191
# We unlock the repository even if
192
# we fail to unlock the control files
193
if self.combined_control:
194
self.assertEqual([('b', 'lw', True),
204
self.assertEqual([('b', 'lw', True),
213
# For cleanup purposes, make sure we are unlocked
214
b.control_files._other.unlock()
216
def test_05_lock_read_fail_repo(self):
217
# Test that the branch is not locked if it cannot lock the repository
218
b = self.get_instrumented_branch()
219
b.repository.disable_lock_read()
221
self.assertRaises(TestPreventLocking, b.lock_read)
222
self.assertFalse(b.is_locked())
223
self.assertFalse(b.repository.is_locked())
225
self.assertEqual([('b', 'lr', True),
229
def test_06_lock_write_fail_repo(self):
230
# Test that the branch is not locked if it cannot lock the repository
231
b = self.get_instrumented_branch()
232
b.repository.disable_lock_write()
234
self.assertRaises(TestPreventLocking, b.lock_write)
235
self.assertFalse(b.is_locked())
236
self.assertFalse(b.repository.is_locked())
238
self.assertEqual([('b', 'lw', True),
242
def test_07_lock_read_fail_control(self):
243
# Test the repository is unlocked if we can't lock self
244
b = self.get_instrumented_branch()
245
b.control_files.disable_lock_read()
247
self.assertRaises(TestPreventLocking, b.lock_read)
248
self.assertFalse(b.is_locked())
249
self.assertFalse(b.repository.is_locked())
251
if self.combined_control:
252
self.assertEqual([('b', 'lr', True),
260
self.assertEqual([('b', 'lr', True),
266
def test_08_lock_write_fail_control(self):
267
# Test the repository is unlocked if we can't lock self
268
b = self.get_instrumented_branch()
269
b.control_files.disable_lock_write()
271
self.assertRaises(TestPreventLocking, b.lock_write)
272
self.assertFalse(b.is_locked())
273
self.assertFalse(b.repository.is_locked())
274
if self.combined_control:
275
self.assertEqual([('b', 'lw', True),
283
self.assertEqual([('b', 'lw', True),
289
def test_lock_write_returns_None_refuses_token(self):
290
branch = self.make_branch('b')
291
token = branch.lock_write()
293
if token is not None:
294
# This test does not apply, because this lockable supports
297
self.assertRaises(errors.TokenLockingNotSupported,
298
branch.lock_write, token='token')
302
def test_reentering_lock_write_raises_on_token_mismatch(self):
303
branch = self.make_branch('b')
304
token = branch.lock_write()
307
# This test does not apply, because this lockable refuses
310
different_branch_token = token + 'xxx'
311
# Re-using the same lockable instance with a different branch token
312
# will raise TokenMismatch.
313
self.assertRaises(errors.TokenMismatch,
315
token=different_branch_token)
319
def test_lock_write_with_nonmatching_token(self):
320
branch = self.make_branch('b')
321
token = branch.lock_write()
324
# This test does not apply, because this branch refuses
327
different_branch_token = token + 'xxx'
329
new_branch = branch.bzrdir.open_branch()
330
# We only want to test the relocking abilities of branch, so use the
331
# existing repository object which is already locked.
332
new_branch.repository = branch.repository
333
self.assertRaises(errors.TokenMismatch,
334
new_branch.lock_write,
335
token=different_branch_token)
340
def test_lock_write_with_matching_token(self):
341
"""Test that a branch can be locked with a token, if it is already
342
locked by that token."""
343
branch = self.make_branch('b')
344
token = branch.lock_write()
347
# This test does not apply, because this branch refuses tokens.
349
# The same instance will accept a second lock_write if the specified
351
branch.lock_write(token=token)
353
# Calling lock_write on a new instance for the same lockable will
355
new_branch = branch.bzrdir.open_branch()
356
# We only want to test the relocking abilities of branch, so use the
357
# existing repository object which is already locked.
358
new_branch.repository = branch.repository
359
new_branch.lock_write(token=token)
364
def test_unlock_after_lock_write_with_token(self):
365
# If lock_write did not physically acquire the lock (because it was
366
# passed some tokens), then unlock should not physically release it.
367
branch = self.make_branch('b')
368
token = branch.lock_write()
371
# This test does not apply, because this lockable refuses
374
new_branch = branch.bzrdir.open_branch()
375
# We only want to test the relocking abilities of branch, so use the
376
# existing repository object which is already locked.
377
new_branch.repository = branch.repository
378
new_branch.lock_write(token=token)
380
self.assertTrue(branch.get_physical_lock_status()) #XXX
384
def test_lock_write_with_token_fails_when_unlocked(self):
385
# First, lock and then unlock to get superficially valid tokens. This
386
# mimics a likely programming error, where a caller accidentally tries
387
# to lock with a token that is no longer valid (because the original
388
# lock was released).
389
branch = self.make_branch('b')
390
token = branch.lock_write()
393
# This test does not apply, because this lockable refuses
397
self.assertRaises(errors.TokenMismatch,
398
branch.lock_write, token=token)
400
def test_lock_write_reenter_with_token(self):
401
branch = self.make_branch('b')
402
token = branch.lock_write()
405
# This test does not apply, because this lockable refuses
408
# Relock with a token.
409
branch.lock_write(token=token)
413
# The lock should be unlocked on disk. Verify that with a new lock
415
new_branch = branch.bzrdir.open_branch()
416
# Calling lock_write now should work, rather than raise LockContention.
417
new_branch.lock_write()
420
def test_leave_lock_in_place(self):
421
branch = self.make_branch('b')
422
# Lock the branch, then use leave_lock_in_place so that when we
423
# unlock the branch the lock is still held on disk.
424
token = branch.lock_write()
427
# This test does not apply, because this repository refuses lock
429
self.assertRaises(NotImplementedError,
430
branch.leave_lock_in_place)
432
branch.leave_lock_in_place()
435
# We should be unable to relock the repo.
436
self.assertRaises(errors.LockContention, branch.lock_write)
438
def test_dont_leave_lock_in_place(self):
439
branch = self.make_branch('b')
440
# Create a lock on disk.
441
token = branch.lock_write()
444
# This test does not apply, because this branch refuses lock
446
self.assertRaises(NotImplementedError,
447
branch.dont_leave_lock_in_place)
450
branch.leave_lock_in_place()
451
except NotImplementedError:
452
# This branch doesn't support this API.
455
branch.repository.leave_lock_in_place()
456
except NotImplementedError:
457
# This repo doesn't support leaving locks around,
458
# assume it is essentially lock-free.
461
repo_token = branch.repository.lock_write()
462
branch.repository.unlock()
465
# Reacquire the lock (with a different branch object) by using the
467
new_branch = branch.bzrdir.open_branch()
468
if repo_token is not None:
469
# We have to explicitly lock the repository first.
470
new_branch.repository.lock_write(token=repo_token)
471
new_branch.lock_write(token=token)
472
if repo_token is not None:
473
# Now we don't need our own repository lock anymore (the branch is
474
# holding it for us).
475
new_branch.repository.unlock()
476
# Call dont_leave_lock_in_place, so that the lock will be released by
477
# this instance, even though the lock wasn't originally acquired by it.
478
new_branch.dont_leave_lock_in_place()
479
if repo_token is not None:
480
new_branch.repository.dont_leave_lock_in_place()
482
# Now the branch (and repository) is unlocked. Test this by locking it
487
def test_lock_read_then_unlock(self):
488
# Calling lock_read then unlocking should work without errors.
489
branch = self.make_branch('b')
493
def test_lock_write_locks_repo_too(self):
494
if isinstance(self.branch_format, BzrBranchFormat4):
495
# Branch format 4 is combined with the repository, so this test
498
branch = self.make_branch('b')
499
branch = branch.bzrdir.open_branch()
502
# The branch should have asked the repository to lock.
503
self.assertTrue(branch.repository.is_write_locked())
504
# Does the repository type actually lock?
505
if not branch.repository.get_physical_lock_status():
506
# The test was successfully applied, so it was applicable.
508
# Now the branch.repository is physically locked, so we can't lock
509
# it with a new repository instance.
510
new_repo = branch.bzrdir.open_repository()
511
self.assertRaises(errors.LockContention, new_repo.lock_write)
512
# We can call lock_write on the original repository object though,
513
# because it is already locked.
514
branch.repository.lock_write()
515
branch.repository.unlock()
added
removed
bzrlib/tests/branch_implementations/test_locking.py
