← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/branch_implementations/test_locking.py

  • Committer: John Arbash Meinel
  • Date: 2007-04-28 15:04:17 UTC
  • mfrom: (2466 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2566.
  • Revision ID: john@arbash-meinel.com-20070428150417-trp3pi0pzd411pu4
[merge] bzr.dev 2466

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/branch_implementations/test_locking.py
18
18
 
19
19
from bzrlib import errors
20
20
from bzrlib.branch import BzrBranchFormat4
 
21
from bzrlib.bzrdir import RemoteBzrDirFormat
21
22
from bzrlib.tests import TestSkipped
22
23
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
23
24
from bzrlib.tests.lock_helpers import TestPreventLocking, LockWrapper
39
40
        b = LockWrapper(self.locks, self.get_branch(), 'b')
40
41
        b.repository = LockWrapper(self.locks, b.repository, 'r')
41
42
        bcf = b.control_files
42
 
        rcf = b.repository.control_files
 
43
        rcf = getattr(b.repository, 'control_files', None)
 
44
        if rcf is None:
 
45
            raise TestSkipped(
 
46
                "This tests depends on being able to instrument "
 
47
                "repository.control_files, but %r doesn't have control_files."
 
48
                % (b.repository,))
43
49
 
44
50
        # Look out for branch types that reuse their control files
45
51
        self.combined_control = bcf is rcf
227
233
                          ('rc', 'ul', True),
228
234
                         ], self.locks)
229
235
 
 
236
    def test_lock_write_returns_None_refuses_token(self):
 
237
        branch = self.make_branch('b')
 
238
        token = branch.lock_write()
 
239
        try:
 
240
            if token is not None:
 
241
                # This test does not apply, because this lockable supports
 
242
                # tokens.
 
243
                return
 
244
            self.assertRaises(errors.TokenLockingNotSupported,
 
245
                              branch.lock_write, token='token')
 
246
        finally:
 
247
            branch.unlock()
 
248
 
 
249
    def test_reentering_lock_write_raises_on_token_mismatch(self):
 
250
        branch = self.make_branch('b')
 
251
        token = branch.lock_write()
 
252
        try:
 
253
            if token is None:
 
254
                # This test does not apply, because this lockable refuses
 
255
                # tokens.
 
256
                return
 
257
            different_branch_token = token + 'xxx'
 
258
            # Re-using the same lockable instance with a different branch token
 
259
            # will raise TokenMismatch.
 
260
            self.assertRaises(errors.TokenMismatch,
 
261
                              branch.lock_write,
 
262
                              token=different_branch_token)
 
263
        finally:
 
264
            branch.unlock()
 
265
 
 
266
    def test_lock_write_with_nonmatching_token(self):
 
267
        branch = self.make_branch('b')
 
268
        token = branch.lock_write()
 
269
        try:
 
270
            if token is None:
 
271
                # This test does not apply, because this branch refuses
 
272
                # tokens.
 
273
                return
 
274
            different_branch_token = token + 'xxx'
 
275
 
 
276
            new_branch = branch.bzrdir.open_branch()
 
277
            # We only want to test the relocking abilities of branch, so use the
 
278
            # existing repository object which is already locked.
 
279
            new_branch.repository = branch.repository
 
280
            self.assertRaises(errors.TokenMismatch,
 
281
                              new_branch.lock_write,
 
282
                              token=different_branch_token)
 
283
        finally:
 
284
            branch.unlock()
 
285
 
 
286
 
 
287
    def test_lock_write_with_matching_token(self):
 
288
        """Test that a branch can be locked with a token, if it is already
 
289
        locked by that token."""
 
290
        branch = self.make_branch('b')
 
291
        token = branch.lock_write()
 
292
        try:
 
293
            if token is None:
 
294
                # This test does not apply, because this branch refuses tokens.
 
295
                return
 
296
            # The same instance will accept a second lock_write if the specified
 
297
            # token matches.
 
298
            branch.lock_write(token=token)
 
299
            branch.unlock()
 
300
            # Calling lock_write on a new instance for the same lockable will
 
301
            # also succeed.
 
302
            new_branch = branch.bzrdir.open_branch()
 
303
            # We only want to test the relocking abilities of branch, so use the
 
304
            # existing repository object which is already locked.
 
305
            new_branch.repository = branch.repository
 
306
            new_branch.lock_write(token=token)
 
307
            new_branch.unlock()
 
308
        finally:
 
309
            branch.unlock()
 
310
 
 
311
    def test_unlock_after_lock_write_with_token(self):
 
312
        # If lock_write did not physically acquire the lock (because it was
 
313
        # passed some tokens), then unlock should not physically release it.
 
314
        branch = self.make_branch('b')
 
315
        token = branch.lock_write()
 
316
        try:
 
317
            if token is None:
 
318
                # This test does not apply, because this lockable refuses
 
319
                # tokens.
 
320
                return
 
321
            new_branch = branch.bzrdir.open_branch()
 
322
            # We only want to test the relocking abilities of branch, so use the
 
323
            # existing repository object which is already locked.
 
324
            new_branch.repository = branch.repository
 
325
            new_branch.lock_write(token=token)
 
326
            new_branch.unlock()
 
327
            self.assertTrue(branch.get_physical_lock_status()) #XXX
 
328
        finally:
 
329
            branch.unlock()
 
330
 
 
331
    def test_lock_write_with_token_fails_when_unlocked(self):
 
332
        # First, lock and then unlock to get superficially valid tokens.  This
 
333
        # mimics a likely programming error, where a caller accidentally tries
 
334
        # to lock with a token that is no longer valid (because the original
 
335
        # lock was released).
 
336
        branch = self.make_branch('b')
 
337
        token = branch.lock_write()
 
338
        branch.unlock()
 
339
        if token is None:
 
340
            # This test does not apply, because this lockable refuses
 
341
            # tokens.
 
342
            return
 
343
 
 
344
        self.assertRaises(errors.TokenMismatch,
 
345
                          branch.lock_write, token=token)
 
346
 
 
347
    def test_lock_write_reenter_with_token(self):
 
348
        branch = self.make_branch('b')
 
349
        token = branch.lock_write()
 
350
        try:
 
351
            if token is None:
 
352
                # This test does not apply, because this lockable refuses
 
353
                # tokens.
 
354
                return
 
355
            # Relock with a token.
 
356
            branch.lock_write(token=token)
 
357
            branch.unlock()
 
358
        finally:
 
359
            branch.unlock()
 
360
        # The lock should be unlocked on disk.  Verify that with a new lock
 
361
        # instance.
 
362
        new_branch = branch.bzrdir.open_branch()
 
363
        # Calling lock_write now should work, rather than raise LockContention.
 
364
        new_branch.lock_write()
 
365
        new_branch.unlock()
 
366
 
 
367
    def test_leave_lock_in_place(self):
 
368
        branch = self.make_branch('b')
 
369
        # Lock the branch, then use leave_lock_in_place so that when we
 
370
        # unlock the branch the lock is still held on disk.
 
371
        token = branch.lock_write()
 
372
        try:
 
373
            if token is None:
 
374
                # This test does not apply, because this repository refuses lock
 
375
                # tokens.
 
376
                self.assertRaises(NotImplementedError,
 
377
                                  branch.leave_lock_in_place)
 
378
                return
 
379
            branch.leave_lock_in_place()
 
380
        finally:
 
381
            branch.unlock()
 
382
        # We should be unable to relock the repo.
 
383
        self.assertRaises(errors.LockContention, branch.lock_write)
 
384
 
 
385
    def test_dont_leave_lock_in_place(self):
 
386
        branch = self.make_branch('b')
 
387
        # Create a lock on disk.
 
388
        token = branch.lock_write()
 
389
        try:
 
390
            if token is None:
 
391
                # This test does not apply, because this branch refuses lock
 
392
                # tokens.
 
393
                self.assertRaises(NotImplementedError,
 
394
                                  branch.dont_leave_lock_in_place)
 
395
                return
 
396
            try:
 
397
                branch.leave_lock_in_place()
 
398
            except NotImplementedError:
 
399
                # This branch doesn't support this API.
 
400
                return
 
401
            branch.repository.leave_lock_in_place()
 
402
            repo_token = branch.repository.lock_write()
 
403
            branch.repository.unlock()
 
404
        finally:
 
405
            branch.unlock()
 
406
        # Reacquire the lock (with a different branch object) by using the
 
407
        # tokens.
 
408
        new_branch = branch.bzrdir.open_branch()
 
409
        # We have to explicitly lock the repository first.
 
410
        new_branch.repository.lock_write(token=repo_token)
 
411
        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()
 
415
        # Call dont_leave_lock_in_place, so that the lock will be released by
 
416
        # this instance, even though the lock wasn't originally acquired by it.
 
417
        new_branch.dont_leave_lock_in_place()
 
418
        new_branch.repository.dont_leave_lock_in_place()
 
419
        new_branch.unlock()
 
420
        # Now the branch (and repository) is unlocked.  Test this by locking it
 
421
        # without tokens.
 
422
        branch.lock_write()
 
423
        branch.unlock()
 
424
 
230
425
    def test_lock_read_then_unlock(self):
231
426
        # Calling lock_read then unlocking should work without errors.
232
427
        branch = self.make_branch('b')