32
40
b = LockWrapper(self.locks, self.get_branch(), 'b')
33
41
b.repository = LockWrapper(self.locks, b.repository, 'r')
34
42
bcf = b.control_files
35
rcf = b.repository.control_files
37
# Look out for branch types that reuse their control files
38
self.combined_control = bcf is rcf
40
b.control_files = LockWrapper(self.locks, b.control_files, 'bc')
41
b.repository.control_files = \
42
LockWrapper(self.locks, b.repository.control_files, 'rc')
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')
45
62
def test_01_lock_read(self):
211
271
self.assertRaises(TestPreventLocking, b.lock_write)
212
272
self.assertFalse(b.is_locked())
213
273
self.assertFalse(b.repository.is_locked())
215
self.assertEqual([('b', 'lw', True),
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
